Implement async callback for Android mediacodec

pjlib/build/Makefile

@@ -31,10 +31,10 @@ export _LDFLAGS := $(CC_LDFLAGS) $(OS_LDFLAGS) $(M_LDFLAGS) $(HOST_LDFLAGS) \
 #
 export PJLIB_SRCDIR = ../src/pj
 export PJLIB_OBJS += $(OS_OBJS) $(M_OBJS) $(CC_OBJS) $(HOST_OBJS) \
-	activesock.o array.o config.o ctype.o errno.o except.o fifobuf.o \
-	guid.o hash.o ip_helper_generic.o list.o lock.o log.o os_time_common.o \
-	os_info.o pool.o pool_buf.o pool_caching.o pool_dbg.o rand.o \
-	rbtree.o sock_common.o sock_qos_common.o \
+	activesock.o array.o atomic_queue.o config.o ctype.o errno.o except.o \
+	fifobuf.o guid.o hash.o ip_helper_generic.o list.o lock.o log.o \
+	os_time_common.o os_info.o pool.o pool_buf.o pool_caching.o pool_dbg.o \
+	rand.o rbtree.o sock_common.o sock_qos_common.o \
 	ssl_sock_common.o ssl_sock_ossl.o ssl_sock_gtls.o ssl_sock_dump.o \
 	ssl_sock_darwin.o string.o timer.o types.o unittest.o
 export PJLIB_CFLAGS += $(_CFLAGS)

pjlib/include/pj/atomic_queue.h

@@ -0,0 +1,97 @@
+/*
+ * Copyright (C) 2024 Teluu Inc. (http://www.teluu.com)
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+#ifndef __PJ_ATOMIC_QUEUE_H__
+#define __PJ_ATOMIC_QUEUE_H__
+
+/**
+ * @file atomic_queue.h
+ * @brief Atomic Queue operations
+ * @{
+ *
+ * Atomic queue for a single consumer and producer.
+ * This cyclic queue employs a ring buffer for storage, maintaining read/write
+ * pointers without locks. It’s designed for one producer and one consumer,
+ * as having multiple producers/consumers could lead to conflicts with the
+ * read/write pointers.
+ * The producer uses #pj_atomic_queue_put() to add an item to the back
+ * of the queue, while the consumer uses #pj_atomic_queue_get()
+ * to retrieve an item from the front.
+ */
+
+#include <pj/types.h>
+
+PJ_BEGIN_DECL
+
+/**
+ * Create a new Atomic Queue for single consumer and single producer case.
+ *
+ * @param pool          The pool to allocate the atomic queue structure.
+ * @param max_item_cnt  The maximum number of items that can be stored.
+ * @param item_size     The size of each item.
+ * @param name          The name of the queue.
+ * @param atomic_queue  Pointer to hold the newly created Atomic Queue.
+ *
+ * @return              PJ_SUCCESS on success.
+ */
+PJ_DECL(pj_status_t) pj_atomic_queue_create(pj_pool_t *pool,
+                                            unsigned max_item_cnt,
+                                            unsigned item_size,
+                                            const char *name,
+                                            pj_atomic_queue_t **atomic_queue);
+
+/**
+ * Destroy the Atomic Queue.
+ *
+ * @param atomic_queue  The Atomic Queue to be destroyed.
+ *
+ * @return              PJ_SUCCESS if success.
+ */
+PJ_DECL(pj_status_t) pj_atomic_queue_destroy(pj_atomic_queue_t *atomic_queue);
+
+/**
+ * Put an item to the back of the queue. If the queue is almost full
+ * (the write pointer is right before the read pointer) the producer will
+ * forcefully discard the oldest item in the head of the queue by incrementing
+ * the read pointer.
+ *
+ * @param atomic_queue  The Atomic Queue.
+ * @param item          The pointer to the data to store.
+ *
+ * @return              PJ_SUCCESS if success.
+ */
+PJ_DECL(pj_status_t) pj_atomic_queue_put(pj_atomic_queue_t *atomic_queue,
+                                         void *item);
+
+/**
+ * Get an item from the head of the queue.
+ *
+ * @param atomic_queue  The Atomic Queue.
+ * @param item          The pointer to data to get the data.
+ *
+ * @return              PJ_SUCCESS if success.
+ */
+PJ_DECL(pj_status_t) pj_atomic_queue_get(pj_atomic_queue_t *atomic_queue,
+                                         void *item);
+
+/**
+ * @}
+ */
+
+PJ_END_DECL
+
+#endif

pjlib/include/pj/types.h

@@ -242,7 +242,12 @@ typedef struct pj_atomic_t pj_atomic_t;
  * Value type of an atomic variable.
  */
 typedef PJ_ATOMIC_VALUE_TYPE pj_atomic_value_t;
-
+
+/**
+ * Opaque data type for atomic queue.
+ */
+typedef struct pj_atomic_queue_t pj_atomic_queue_t;
+
 /* ************************************************************************* */
 
 /** Thread handle. */

pjlib/include/pjlib.h

@@ -30,6 +30,7 @@
 #include <pj/argparse.h>
 #include <pj/array.h>
 #include <pj/assert.h>
+#include <pj/atomic_queue.h>
 #include <pj/ctype.h>
 #include <pj/errno.h>
 #include <pj/except.h>

pjlib/src/pj/atomic_queue.cpp

@@ -0,0 +1,185 @@
+/*
+ * Copyright (C) 2024 Teluu Inc. (http://www.teluu.com)
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+#include <pj/assert.h>
+#include <pj/atomic_queue.h>
+#include <pj/errno.h>
+#include <pj/pool.h>
+#include <pj/string.h>
+#include <atomic>
+
+#if 0
+#   define TRACE_(arg) PJ_LOG(4,arg)
+#else
+#   define TRACE_(expr)
+#endif
+class AtomicQueue {
+public:
+    /**
+     * Constructor
+     */
+    AtomicQueue(unsigned maxItemCnt, unsigned itemSize, const char* name = ""):
+                maxItemCnt_(maxItemCnt),
+                itemSize_(itemSize),
+                ptrWrite(0),
+                ptrRead(0),
+                buffer(NULL),
+                name_(name)
+    {
+        buffer = new char[maxItemCnt_ * itemSize_];
+
+        /* Surpress warning when debugging log is disabled */
+        PJ_UNUSED_ARG(name_);
+
+        TRACE_((name_, "Created AtomicQueue: maxItemCnt=%d itemSize=%d",
+                maxItemCnt_, itemSize_));
+    }
+
+    /**
+     * Destructor
+     */
+    ~AtomicQueue()
+    {
+        delete [] buffer;
+    }
+
+    /**
+     * Get a item from the head of the queue
+     */
+    bool get(void* item)
+    {
+        if (ptrRead == ptrWrite)
+            return false;
+
+        unsigned cur_ptr = ptrRead;
+        void *p = &buffer[cur_ptr * itemSize_];
+        pj_memcpy(item, p, itemSize_);
+        inc_ptrRead_if_not_yet(cur_ptr);
+
+        TRACE_((name_, "GET: ptrRead=%d ptrWrite=%d\n",
+               ptrRead.load(), ptrWrite.load()));
+
+        return true;
+    }
+
+    /**
+     * Put a item to the back of the queue
+     */
+    void put(void* item)
+    {
+        unsigned cur_ptr = ptrWrite;
+        void *p = &buffer[cur_ptr * itemSize_];
+        pj_memcpy(p, item, itemSize_);
+        unsigned next_ptr = inc_ptrWrite(cur_ptr);
+
+        /* Increment read pointer if next write is overlapping
+         * (next_ptr == read ptr)
+         */
+        unsigned next_read_ptr = (next_ptr == maxItemCnt_-1)? 0 : (next_ptr+1);
+        ptrRead.compare_exchange_strong(next_ptr, next_read_ptr);
+
+        TRACE_((name_, "PUT: ptrRead=%d ptrWrite=%d\n",
+               ptrRead.load(), ptrWrite.load()));
+    }
+
+private:
+    unsigned maxItemCnt_;
+    unsigned itemSize_;
+    std::atomic<unsigned> ptrWrite;
+    std::atomic<unsigned> ptrRead;
+    char *buffer;
+    const char *name_;
+
+    /* Increment read pointer, only if producer not incemented it already.
+     * Producer may increment the read pointer if the write pointer is
+     * right before the read pointer (buffer almost full).
+     */
+    bool inc_ptrRead_if_not_yet(unsigned old_ptr)
+    {
+        unsigned new_ptr = (old_ptr == maxItemCnt_-1)? 0 : (old_ptr+1);
+        return ptrRead.compare_exchange_strong(old_ptr, new_ptr);
+    }
+
+    /* Increment write pointer */
+    unsigned inc_ptrWrite(unsigned old_ptr)
+    {
+        unsigned new_ptr = (old_ptr == maxItemCnt_-1)? 0 : (old_ptr+1);
+        if (ptrWrite.compare_exchange_strong(old_ptr, new_ptr))
+            return new_ptr;
+
+        /* Should never happen */
+        pj_assert(!"There is more than one producer!");
+        return old_ptr;
+    }
+
+    AtomicQueue() {}
+};
+
+struct pj_atomic_queue_t
+{
+    AtomicQueue    *aQ;
+};
+
+PJ_DEF(pj_status_t) pj_atomic_queue_create(pj_pool_t *pool,
+                                           unsigned max_item_cnt,
+                                           unsigned item_size,
+                                           const char *name,
+                                           pj_atomic_queue_t **atomic_queue)
+{
+    pj_atomic_queue_t *aqueue;
+
+    PJ_ASSERT_RETURN(pool, PJ_EINVAL);
+    aqueue = PJ_POOL_ZALLOC_T(pool, pj_atomic_queue_t);
+    aqueue->aQ = new AtomicQueue(max_item_cnt, item_size, name);
+    *atomic_queue = aqueue;
+    return PJ_SUCCESS;
+}
+
+PJ_DEF(pj_status_t) pj_atomic_queue_destroy(pj_atomic_queue_t *atomic_queue)
+{
+    PJ_ASSERT_RETURN(atomic_queue && atomic_queue->aQ, PJ_EINVAL);
+    delete atomic_queue->aQ;
+    atomic_queue->aQ = NULL;
+    return PJ_SUCCESS;
+}
+
+/**
+ * For producer, there is write pointer 'ptrWrite' that will be incremented
+ * every time a item is queued to the back of the queue.
+ */
+PJ_DEF(pj_status_t) pj_atomic_queue_put(pj_atomic_queue_t *atomic_queue,
+                                        void *item)
+{
+    PJ_ASSERT_RETURN(atomic_queue && atomic_queue->aQ && item, PJ_EINVAL);
+    atomic_queue->aQ->put(item);
+    return PJ_SUCCESS;
+}
+
+/**
+ * For consumer, there is read pointer 'ptrRead' that will be incremented
+ * every time a item is fetched from the head of the queue, only if the
+ * pointer is not modified by producer (in case of queue full).
+ */
+PJ_DEF(pj_status_t) pj_atomic_queue_get(pj_atomic_queue_t *atomic_queue,
+                                        void *item)
+{
+    PJ_ASSERT_RETURN(atomic_queue && atomic_queue->aQ && item, PJ_EINVAL);
+    if (atomic_queue->aQ->get(item))
+        return PJ_SUCCESS;
+    else
+        return PJ_ENOTFOUND;
+}