[mpp]: Fix block input error

1. put_packet support block mode now.
2. The EosTask should be insure to be block mode.
3. NOTE: When set to block mode use put_packet and get_frame in single
thread may cause deadlock for the info change or eos may be blocked and
the put_packet can not return.

Change-Id: I8cf73f833e5d29c3b330c4b56fbd6be8d7432e60
Signed-off-by: Herman Chen <herman.chen@rock-chips.com>

mpp/base/mpp_task_impl.cpp

@@ -171,31 +171,33 @@ MPP_RET _mpp_port_poll(const char *caller, MppPort port, MppPollType timeout)
          * negtive  - block
          * positive - timeout value
          */
-        if (timeout != MPP_POLL_NON_BLOCK) {
+        if (timeout) {
             mpp_assert(curr->cond);
             Condition *cond = curr->cond;
 
-            if (timeout == MPP_POLL_BLOCK) {
+            if (timeout < 0) {
                 mpp_task_dbg_flow("mpp %p %s from %s poll %s port block wait start\n",
                                   queue->mpp, queue->name, caller,
                                   port_type_str[port_impl->type]);
-                cond->wait(queue->lock);
+
+                ret = (MPP_RET)cond->wait(queue->lock);
             } else {
                 mpp_task_dbg_flow("mpp %p %s from %s poll %s port %d timeout wait start\n",
                                   queue->mpp, queue->name, caller,
                                   port_type_str[port_impl->type], timeout);
-                cond->timedwait(queue->lock, timeout);
+                ret = (MPP_RET)cond->timedwait(queue->lock, timeout);
             }
 
             if (curr->count) {
                 mpp_assert(!list_empty(&curr->list));
                 ret = (MPP_RET)curr->count;
-            }
+            } else if (ret > 0)
+                ret = MPP_NOK;
         }
 
         mpp_task_dbg_flow("mpp %p %s from %s poll %s port timeout %d ret %d\n",
                           queue->mpp, queue->name, caller,
-                          port_type_str[port_impl->type], ret);
+                          port_type_str[port_impl->type], timeout, ret);
     }
 RET:
     mpp_task_dbg_func("leave\n");

mpp/inc/mpp.h

@@ -164,6 +164,7 @@ public:
     MppPollType     mOutputTimeout;
 
     MppTask         mInputTask;
+    MppTask         mEosTask;
 
     MppDec          mDec;
     MppEnc          mEnc;
@@ -180,7 +181,6 @@ private:
     RK_U32          mMultiFrame;
 
     RK_U32          mStatus;
-    RK_S32          mDecResTaskCnt;
 
     /* decoder paramter before init */
     MppDecCfgSet    mDecInitcfg;

mpp/mpp.cpp

@@ -80,6 +80,7 @@ Mpp::Mpp()
       mInputTimeout(MPP_POLL_BUTT),
       mOutputTimeout(MPP_POLL_BUTT),
       mInputTask(NULL),
+      mEosTask(NULL),
       mDec(NULL),
       mEnc(NULL),
       mEncVersion(0),
@@ -88,7 +89,6 @@ Mpp::Mpp()
       mInitDone(0),
       mMultiFrame(0),
       mStatus(0),
-      mDecResTaskCnt(1),
       mExtraPacket(NULL),
       mDump(NULL)
 {
@@ -291,9 +291,9 @@ MPP_RET Mpp::put_packet(MppPacket packet)
         return MPP_ERR_INIT;
 
     MPP_RET ret = MPP_NOK;
-    RK_U32 eos = mpp_packet_get_eos(packet);
     MppPollType timeout = mInputTimeout;
     MppTask task_dequeue = NULL;
+    RK_U32 pkt_copy = 0;
 
     if (mExtraPacket) {
         MppPacket extra = mExtraPacket;
@@ -302,20 +302,35 @@ MPP_RET Mpp::put_packet(MppPacket packet)
         put_packet(extra);
     }
 
-    /* handle eos packet on non-block mode */
+    if (!mEosTask) {
+        /* handle eos packet on block mode */
+        ret = poll(MPP_PORT_INPUT, MPP_POLL_BLOCK);
+        if (ret < 0)
+            goto RET;
+
+        dequeue(MPP_PORT_INPUT, &mEosTask);
+        if (NULL == mEosTask) {
+            mpp_err_f("fail to reserve eos task\n", ret);
+            ret = MPP_NOK;
+            goto RET;
+        }
+    }
+
+    if (mpp_packet_get_eos(packet)) {
+        mpp_assert(mEosTask);
+        task_dequeue = mEosTask;
+        mEosTask = NULL;
+    }
+
     /* Use reserved task to send eos packet */
-    if (mInputTask) {
+    if (mInputTask && !task_dequeue) {
         task_dequeue = mInputTask;
         mInputTask = NULL;
     }
 
     if (NULL == task_dequeue) {
         ret = poll(MPP_PORT_INPUT, timeout);
-        if (ret < 0)
+        if (ret < 0) {
-            goto RET;
-
-        /* non-eos packet should reserve one task for eos case */
-        if (!eos && ret <= mDecResTaskCnt) {
             ret = MPP_ERR_BUFFER_FULL;
             goto RET;
         }
@@ -335,6 +350,7 @@ MPP_RET Mpp::put_packet(MppPacket packet)
 
         mpp_packet_copy_init(&pkt_in, packet);
         mpp_packet_set_length(packet, 0);
+        pkt_copy = 1;
         packet = pkt_in;
         ret = MPP_OK;
     } else {
@@ -361,15 +377,15 @@ MPP_RET Mpp::put_packet(MppPacket packet)
 
     mPacketPutCount++;
 
-    if (timeout)
+    if (timeout && !pkt_copy)
         ret = poll(MPP_PORT_INPUT, timeout);
 
 RET:
     /* wait enqueued task finished */
     if (NULL == mInputTask) {
-        MPP_RET cnt = poll(MPP_PORT_INPUT, mInputTimeout);
+        MPP_RET cnt = poll(MPP_PORT_INPUT, MPP_POLL_NON_BLOCK);
         /* reserve one task for eos block mode */
-        if (cnt > mDecResTaskCnt) {
+        if (cnt >= 0) {
             dequeue(MPP_PORT_INPUT, &mInputTask);
             mpp_assert(mInputTask);
         }