[mpp]: add basic mpp thread codec flow

git-svn-id: https://10.10.10.66:8443/svn/MediaProcessPlatform/trunk/mpp@158 6e48237b-75ef-9749-8fc9-41990f28c85a

inc/mpp_packet.h

@@ -28,12 +28,33 @@ extern "C" {
 
 /*
  * MppPacket interface
+ *
+ * mpp_packet_init = mpp_packet_new + mpp_packet_set_data + mpp_packet_set_size
  */
+MPP_RET mpp_packet_new(MppPacket *packet);
 MPP_RET mpp_packet_init(MppPacket *packet, void *data, size_t size);
 MPP_RET mpp_packet_deinit(MppPacket packet);
 
-MPP_RET mpp_packet_set_pts(MppPacket packet, RK_S64 pts);
+
-MPP_RET mpp_packet_set_dts(MppPacket packet, RK_S64 dts);
+void    mpp_packet_set_data(MppPacket packet, void *data);
+void*   mpp_packet_get_data(const MppPacket packet);
+void    mpp_packet_set_size(MppPacket packet, size_t size);
+size_t  mpp_packet_get_size(const MppPacket packet);
+
+void    mpp_packet_set_offset(MppPacket packet, size_t offset);
+size_t  mpp_packet_get_offset(const MppPacket packet);
+void    mpp_packet_set_length(MppPacket packet, size_t length);
+size_t  mpp_packet_get_length(const MppPacket packet);
+
+void    mpp_packet_set_pts(MppPacket packet, RK_S64 pts);
+RK_S64  mpp_packet_get_pts(const MppPacket packet);
+void    mpp_packet_set_dts(MppPacket packet, RK_S64 dts);
+RK_S64  mpp_packet_get_dts(const MppPacket packet);
+
+void    mpp_packet_set_flag(MppPacket packet, RK_U32 flag);
+RK_U32  mpp_packet_get_flag(const MppPacket packet);
+
+
 MPP_RET mpp_packet_set_eos(MppPacket packet);
 MPP_RET mpp_packet_set_extra_data(MppPacket packet);
 

inc/rk_mpi.h

@@ -17,11 +17,7 @@
 #ifndef __RK_MPI_H__
 #define __RK_MPI_H__
 
-#include "rk_type.h"
-
-#include "mpp_err.h"
 #include "mpp_packet.h"
-#include "mpp_buffer.h"
 #include "mpp_frame.h"
 
 typedef enum {

mpp/CMakeLists.txt

@@ -25,6 +25,7 @@ add_library(mpp STATIC
     mpp_buffer.cpp
     mpp_packet.cpp
     mpp_frame.cpp
+    mpp.cpp
     mpi_impl.cpp
     mpi.cpp
     )

mpp/mpi.cpp

@@ -23,6 +23,7 @@
 #include "mpp_log.h"
 #include "mpp_mem.h"
 #include "mpi_impl.h"
+#include "mpp.h"
 
 static MPP_RET mpi_config(MppCtx ctx, MppEncConfig cfg)
 {
@@ -58,42 +59,70 @@ static MPP_RET mpi_encode(MppCtx ctx, MppFrame frame, MppPacket *packet)
 
 static MPP_RET mpi_decode_put_packet(MppCtx ctx, MppPacket packet)
 {
-    (void)ctx;
-    (void)packet;
     MPI_FUNCTION_ENTER();
+    MpiImpl *p = (MpiImpl *)ctx;
+
+    if (NULL == p || p->check != p || NULL == p->ctx || NULL == packet) {
+        mpp_err("mpi_decode_put_packet found invalid context input ctx %p packet %p\n",
+                ctx, packet);
+        return MPP_ERR_UNKNOW;
+    }
+
+    MPP_RET ret = p->ctx->put_packet(packet);
 
     MPI_FUNCTION_LEAVE();
-    return MPP_OK;
+    return ret;
 }
 
 static MPP_RET mpi_decode_get_frame(MppCtx ctx, MppFrame *frame)
 {
-    (void)ctx;
-    (void)frame;
     MPI_FUNCTION_ENTER();
+    MpiImpl *p = (MpiImpl *)ctx;
+
+    if (NULL == p || p->check != p || NULL == p->ctx || NULL == frame) {
+        mpp_err("mpi_decode_get_frame found invalid context input ctx %p frame %p\n",
+                ctx, frame);
+        return MPP_ERR_UNKNOW;
+    }
+
+    MPP_RET ret = p->ctx->get_frame(frame);
 
     MPI_FUNCTION_LEAVE();
-    return MPP_OK;
+    return ret;
 }
 
 static MPP_RET mpi_encode_put_frame(MppCtx ctx, MppFrame frame)
 {
-    (void)ctx;
-    (void)frame;
     MPI_FUNCTION_ENTER();
+    MpiImpl *p = (MpiImpl *)ctx;
+
+    if (NULL == p || p->check != p || NULL == p->ctx || NULL == frame) {
+        mpp_err("mpi_encode_put_frame found invalid context input ctx %p frame %p\n",
+                ctx, frame);
+        return MPP_ERR_UNKNOW;
+    }
+
+    MPP_RET ret = p->ctx->put_frame(frame);
 
     MPI_FUNCTION_LEAVE();
-    return MPP_OK;
+    return ret;
 }
 
 static MPP_RET mpi_encode_get_packet(MppCtx ctx, MppPacket *packet)
 {
-    (void)ctx;
-    (void)packet;
     MPI_FUNCTION_ENTER();
+    MpiImpl *p = (MpiImpl *)ctx;
+
+    if (NULL == p || p->check != p || NULL == p->ctx || NULL == packet) {
+        mpp_err("mpi_encode_get_packet found invalid context input ctx %p packet %p\n",
+                ctx, packet);
+        return MPP_ERR_UNKNOW;
+    }
+
+    MPP_RET ret = p->ctx->get_packet(packet);
 
     MPI_FUNCTION_LEAVE();
-    return MPP_OK;
+    return ret;
 }
 
 static MPP_RET mpi_flush(MppCtx ctx)
@@ -144,6 +173,9 @@ MPP_RET mpp_init(MppCtx *ctx, MppApi **mpi, MppCtxType type, MppCodingType codin
         return MPP_ERR_NULL_PTR;
     }
 
+    *ctx = NULL;
+    *mpi = NULL;
+
     p = mpp_malloc(MpiImpl, 1);
     if (NULL == p) {
         mpp_err("mpp_init failed to allocate context\n");
@@ -151,6 +183,14 @@ MPP_RET mpp_init(MppCtx *ctx, MppApi **mpi, MppCtxType type, MppCodingType codin
     }
 
     memset(p, 0, sizeof(*p));
+
+    p->ctx = new Mpp(type);
+    if (NULL == p->ctx) {
+        mpp_free(p);
+        mpp_err("mpp_init failed to new Mpp\n");
+        return MPP_ERR_MALLOC;
+    }
+
     p->api      = &mpp_api;
     p->check    = p;
     p->type     = type;
@@ -181,6 +221,9 @@ MPP_RET mpp_deinit(MppCtx ctx)
         return MPP_ERR_UNKNOW;
     }
 
+    if (p->ctx)
+        delete p->ctx;
+
     if (p)
         mpp_free(p);
 

mpp/mpi_impl.h

@@ -17,8 +17,7 @@
 #ifndef __MPP_IMPL_H__
 #define __MPP_IMPL_H__
 
-#include "rk_type.h"
+#include "mpp.h"
-#include "rk_mpi.h"
 #include "vpu_api.h"
 
 #define MPI_DBG_FUNCTION            (0x00000001)
@@ -36,11 +35,11 @@ typedef struct MpiImpl_t MpiImpl;
 
 struct MpiImpl_t {
     MpiImpl         *check;
-    MppCtxType    type;
+    MppCtxType      type;
     MppCodingType   coding;
 
     MppApi          *api;
-    void            *ctx;
+    Mpp             *ctx;
 };
 
 #ifdef __cplusplus

mpp/mpp.cpp

@@ -0,0 +1,163 @@
+/*
+ * Copyright 2010 Rockchip Electronics S.LSI Co. LTD
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#define  MODULE_TAG "mpp"
+
+#include "mpp_log.h"
+#include "mpp_mem.h"
+#include "mpp_time.h"
+
+#include "mpp.h"
+#include "mpp_frame_impl.h"
+#include "mpp_buffer.h"
+#include "mpp_packet.h"
+#include "mpp_packet_impl.h"
+
+static void *thread_dec(void *data)
+{
+    Mpp *mpp = (Mpp*)data;
+    mpp_list *packets = mpp->packets;
+    mpp_list *frames  = mpp->frames;
+    MppPacketImpl packet;
+    MppFrame frame;
+
+    while (mpp->thread_running) {
+        if (packets->list_size()) {
+            /*
+             * packet will be destroyed outside, here just copy the content
+             */
+            packets->del_at_head(&packet, sizeof(packet));
+
+            /*
+             * generate a new frame, copy the pointer to list
+             */
+            mpp_frame_init(&frame);
+            frames->add_at_tail(&frame, sizeof(frame));
+        }
+    }
+
+    return NULL;
+}
+
+static void *thread_enc(void *data)
+{
+    Mpp *mpp = (Mpp*)data;
+    mpp_list *packets = mpp->packets;
+    mpp_list *frames  = mpp->frames;
+    MppFrameImpl frame;
+    MppPacket packet;
+    size_t size = SZ_1M;
+    char *buf = mpp_malloc(char, size);
+
+    while (mpp->thread_running) {
+        if (frames->list_size()) {
+            frames->del_at_head(&frame, sizeof(frame));
+
+            mpp_packet_init(&packet, buf, size);
+            packets->add_at_tail(&packet, sizeof(packet));
+        }
+    }
+    mpp_free(buf);
+    return NULL;
+}
+
+Mpp::Mpp(MppCtxType type)
+    : packets(NULL),
+      frames(NULL),
+      thread_running(0),
+      thread_reset(0),
+      status(0)
+{
+    switch (type) {
+    case MPP_CTX_DEC : {
+        thread_start(thread_dec);
+        packets = new mpp_list((node_destructor)NULL);
+        frames  = new mpp_list((node_destructor)mpp_frame_deinit);
+    } break;
+    case MPP_CTX_ENC : {
+        thread_start(thread_enc);
+        frames  = new mpp_list((node_destructor)NULL);
+        packets = new mpp_list((node_destructor)mpp_packet_deinit);
+    } break;
+    default : {
+        mpp_err("Mpp error type %d\n", type);
+    } break;
+    }
+    if (thread_running) {
+    }
+}
+
+Mpp::~Mpp ()
+{
+    if (thread_running)
+        thread_stop();
+    if (packets)
+        delete packets;
+    if (frames)
+        delete frames;
+}
+
+void Mpp::thread_start(MppThread func)
+{
+    if (!thread_running) {
+    	pthread_attr_t attr;
+    	pthread_attr_init(&attr);
+    	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
+        if (pthread_create(&thread, &attr, func, (void*)this))
+            status = MPP_ERR_FATAL_THREAD;
+        else
+            thread_running = 1;
+    	pthread_attr_destroy(&attr);
+    }
+}
+
+void Mpp::thread_stop()
+{
+    thread_running = 0;
+    void *dummy;
+    pthread_join(thread, &dummy);
+}
+
+MPP_RET Mpp::put_packet(MppPacket packet)
+{
+    // TODO: packet data need preprocess or can be write to hardware buffer
+    return (MPP_RET)packets->add_at_tail(packet, sizeof(MppPacketImpl));
+}
+
+MPP_RET Mpp::get_frame(MppFrame *frame)
+{
+    MPP_RET ret = MPP_NOK;
+    if (frames->list_size()) {
+        ret = (MPP_RET)frames->del_at_tail(frame, sizeof(frame));
+    }
+    return ret;
+}
+
+MPP_RET Mpp::put_frame(MppFrame frame)
+{
+    MPP_RET ret = (MPP_RET)frames->add_at_tail(frame, sizeof(MppFrameImpl));
+    return ret;
+}
+
+MPP_RET Mpp::get_packet(MppPacket *packet)
+{
+    MPP_RET ret = MPP_NOK;
+    if (packets->list_size()) {
+        ret = (MPP_RET)packets->del_at_tail(packet, sizeof(packet));
+    }
+    return ret;
+}
+

mpp/mpp.h

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2010 Rockchip Electronics S.LSI Co. LTD
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef __MPP_H__
+#define __MPP_H__
+
+#include "rk_mpi.h"
+#include "mpp_list.h"
+#include "mpp_thread.h"
+
+typedef void *(*MppThread)(void *);
+
+class Mpp {
+public:
+    Mpp(MppCtxType type);
+    ~Mpp();
+
+    mpp_list        *packets;
+    mpp_list        *frames;
+
+    pthread_t       thread;
+	RK_S32          thread_running;
+	RK_S32          thread_reset;
+
+    RK_U32          status;
+
+    MPP_RET put_packet(MppPacket packet);
+    MPP_RET get_frame(MppFrame *frame);
+
+    MPP_RET put_frame(MppFrame frame);
+    MPP_RET get_packet(MppPacket *packet);
+
+private:
+    void thread_start(MppThread func);
+    void thread_stop();
+
+    Mpp();
+    Mpp(const Mpp &);
+    Mpp &operator=(const Mpp &);
+};
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*__MPP_H__*/

mpp/mpp_packet.cpp

@@ -21,6 +21,14 @@
 #include "mpp_packet.h"
 #include "mpp_packet_impl.h"
 
+MPP_PACKET_ACCESSORS(MppPacketImpl, void*,  data)
+MPP_PACKET_ACCESSORS(MppPacketImpl, size_t, size)
+MPP_PACKET_ACCESSORS(MppPacketImpl, size_t, offset)
+MPP_PACKET_ACCESSORS(MppPacketImpl, size_t, length)
+MPP_PACKET_ACCESSORS(MppPacketImpl, RK_S64, pts)
+MPP_PACKET_ACCESSORS(MppPacketImpl, RK_S64, dts)
+MPP_PACKET_ACCESSORS(MppPacketImpl, RK_U32, flag)
+
 MPP_RET mpp_packet_init(MppPacket *packet, void *data, size_t size)
 {
     if (NULL == packet || NULL == data || 0 == size) {
@@ -29,7 +37,7 @@ MPP_RET mpp_packet_init(MppPacket *packet, void *data, size_t size)
         return MPP_ERR_NULL_PTR;
     }
 
-    mpp_packet_impl *p = mpp_calloc(mpp_packet_impl, 1);
+    MppPacketImpl *p = mpp_calloc(MppPacketImpl, 1);
     if (NULL == p) {
         mpp_err("mpp_packet_init malloc failed\n");
         return MPP_ERR_NULL_PTR;
@@ -43,27 +51,14 @@ MPP_RET mpp_packet_init(MppPacket *packet, void *data, size_t size)
     return MPP_OK;
 }
 
-MPP_RET mpp_packet_set_pts(MppPacket packet, RK_S64 pts)
+MPP_RET mpp_packet_deinit(MppPacket packet)
 {
     if (NULL == packet) {
-        mpp_err("mpp_packet_set_pts found NULL input\n");
+        mpp_err("mpp_packet_deinit found NULL input\n");
         return MPP_ERR_NULL_PTR;
     }
 
-    mpp_packet_impl *p = (mpp_packet_impl *)packet;
+    mpp_free((MppPacketImpl *)packet);
-    p->pts = pts;
-    return MPP_OK;
-}
-
-MPP_RET mpp_packet_set_dts(MppPacket packet, RK_S64 dts)
-{
-    if (NULL == packet) {
-        mpp_err("mpp_packet_set_dts found NULL input\n");
-        return MPP_ERR_NULL_PTR;
-    }
-
-    mpp_packet_impl *p = (mpp_packet_impl *)packet;
-    p->dts = dts;
     return MPP_OK;
 }
 
@@ -74,7 +69,7 @@ MPP_RET mpp_packet_set_eos(MppPacket packet)
         return MPP_ERR_NULL_PTR;
     }
 
-    mpp_packet_impl *p = (mpp_packet_impl *)packet;
+    MppPacketImpl *p = (MppPacketImpl *)packet;
     p->flag |= MPP_PACKET_FLAG_EOS;
     return MPP_OK;
 }
@@ -86,19 +81,8 @@ MPP_RET mpp_packet_set_extra_data(MppPacket packet)
         return MPP_ERR_NULL_PTR;
     }
 
-    mpp_packet_impl *p = (mpp_packet_impl *)packet;
+    MppPacketImpl *p = (MppPacketImpl *)packet;
     p->flag |= MPP_PACKET_FLAG_EXTRA_DATA;
     return MPP_OK;
 }
 
-MPP_RET mpp_packet_deinit(MppPacket packet)
-{
-    if (NULL == packet) {
-        mpp_err("mpp_packet_deinit found NULL input\n");
-        return MPP_ERR_NULL_PTR;
-    }
-
-    mpp_free((mpp_packet_impl *)packet);
-    return MPP_OK;
-}
-

mpp/mpp_packet_impl.h

@@ -42,6 +42,13 @@ typedef struct {
     RK_S64  dts;
 
     RK_U32  flag;
-} mpp_packet_impl;
+} MppPacketImpl;
+
+/*
+ * object access function macro
+ */
+#define MPP_PACKET_ACCESSORS(impl, type, field) \
+    type mpp_packet_get_##field(const MppPacket *s) { return ((impl*)s)->field; } \
+    void mpp_packet_set_##field(MppPacket *s, type v) { ((impl*)s)->field = v; }
 
 #endif /*__MPP_IMPL_H__*/

test/mpi_test.c

@@ -172,9 +172,55 @@ int main()
         goto MPP_TEST_FAILED;
     }
 
-    mpi->encode(ctx, enc_in, &enc_out);
+
-    mpi->encode_put_frame(ctx, enc_in);
+
-    mpi->encode_get_packet(ctx, &dec_out);
+    // interface with both input and output
+    for (i = 0; i < MPI_ENC_LOOP_COUNT; i++) {
+        mpp_frame_init(&enc_in);
+
+        mpi->encode(ctx, enc_in, &enc_out);
+        if (MPP_OK != ret) {
+            goto MPP_TEST_FAILED;
+        }
+
+        if (enc_out) {
+            mpp_packet_deinit(enc_out);
+            enc_out = NULL;
+        }
+
+        mpp_frame_deinit(enc_in);
+        enc_in = NULL;
+    }
+
+    // interface with input and output separated
+    for (i = 0; i < MPI_ENC_LOOP_COUNT; i++) {
+        mpp_frame_init(&enc_in);
+
+        // TODO: read stream data to buf
+
+        mpi->encode_put_frame(ctx, enc_in);
+        if (MPP_OK != ret) {
+            goto MPP_TEST_FAILED;
+        }
+
+        mpp_frame_deinit(enc_in);
+        enc_in = NULL;
+    }
+
+    for (i = 0; i < MPI_ENC_LOOP_COUNT; i++) {
+        mpi->encode_get_packet(ctx, &enc_out);
+        if (MPP_OK != ret) {
+            goto MPP_TEST_FAILED;
+        }
+
+        if (enc_out) {
+            if (enc_out) {
+                mpp_packet_deinit(enc_out);
+                enc_out = NULL;
+            }
+        }
+    }
+
 
     ret = mpi->flush(ctx);
     if (MPP_OK != ret) {
@@ -187,6 +233,11 @@ int main()
         dec_in = NULL;
     }
 
+    if (enc_in) {
+        mpp_frame_deinit(enc_in);
+        enc_in = NULL;
+    }
+
     mpp_deinit(ctx);
     free(buf);
 
@@ -200,6 +251,11 @@ MPP_TEST_FAILED:
         dec_in = NULL;
     }
 
+    if (enc_in) {
+        mpp_frame_deinit(enc_in);
+        enc_in = NULL;
+    }
+
     if (ctx)
         mpp_deinit(ctx);
     if (buf)