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,81 @@
+/*
+ * 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 __ATOMIC_QUEUE_H__
+#define __ATOMIC_QUEUE_H__
+
+#include <atomic>
+
+/* Atomic queue (ring buffer) for single consumer & single producer.
+ *
+ * Producer invokes 'put(item)' to put an item to the back of the queue and
+ * consumer invokes 'get(item)' to get an item from the head of the queue.
+ *
+ * For producer, there is write pointer 'ptrWrite' that will be incremented
+ * every time a item is queued 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 read pointer.
+ *
+ * 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).
+ */
+class pj_atomic_queue {
+public:
+    /**
+     * Constructor
+     */
+    pj_atomic_queue(unsigned max_item_cnt_, unsigned item_size_,
+                    const char* name_);
+
+    /**
+     * Destructor
+     */
+    ~pj_atomic_queue();
+
+    /**
+     * Get a item from the head of the queue
+     */
+    bool get(void* item);
+
+    /**
+     * Put a item to the back of the queue
+     */
+    void put(void* item);
+
+private:
+    unsigned max_item_cnt_;
+    unsigned item_size_;
+    std::atomic<unsigned> ptr_write;
+    std::atomic<unsigned> ptr_read;
+    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_ptr_read_if_not_yet(unsigned old_ptr);
+
+    /* Increment write pointer */
+    unsigned inc_ptr_write(unsigned old_ptr);
+
+    pj_atomic_queue() {}
+};
+
+#endif

pjlib/src/pj/atomic_queue.cpp

@@ -0,0 +1,99 @@
+/*
+ * 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/string.h>
+
+#if defined(PJ_ANDROID) && PJ_ANDROID != 0
+
+//#include <android/log.h>
+
+pj_atomic_queue::pj_atomic_queue(unsigned max_item_cnt, unsigned item_size,
+                                 const char* name = "") :
+                                 max_item_cnt_(max_item_cnt),
+                                 item_size_(item_size),
+                                 ptr_write(0),
+                                 ptr_read(0),
+                                 buffer(NULL),
+                                 name_(name)
+{
+    buffer = new char[max_item_cnt_ * item_size_];
+
+    /* Surpress warning when debugging log is disabled */
+    PJ_UNUSED_ARG(name_);
+
+    // __android_log_print(ANDROID_LOG_INFO, name,
+    //                 "Created AtomicQueue: max_item_cnt=%d item_size=%d\n",
+    //                 max_item_cnt_, item_size_);
+}
+
+pj_atomic_queue::~pj_atomic_queue() {
+    delete [] buffer;
+}
+
+/* Get a item from the head of the queue */
+bool pj_atomic_queue::get(void* item) {
+    if (ptr_read == ptr_write)
+        return false;
+
+    unsigned cur_ptr = ptr_read;
+    void *p = &buffer[cur_ptr * item_size_];
+    pj_memcpy(item, p, item_size_);
+    inc_ptr_read_if_not_yet(cur_ptr);
+
+    // __android_log_print(ANDROID_LOG_INFO, name,
+    //                    "GET: ptr_read=%d ptr_write=%d\n",
+    //                    ptr_read.load(), ptr_write.load());
+    return true;
+}
+
+/* Put a item to the back of the queue */
+void pj_atomic_queue::put(void* item) {
+    unsigned cur_ptr = ptr_write;
+    void *p = &buffer[cur_ptr * item_size_];
+    pj_memcpy(p, item, item_size_);
+    unsigned next_ptr = inc_ptr_write(cur_ptr);
+
+    /* Increment read pointer if next write is overlapping
+     * (next_ptr == read ptr)
+     */
+    unsigned next_read_ptr = (next_ptr == max_item_cnt_-1)? 0 : (next_ptr+1);
+    ptr_read.compare_exchange_strong(next_ptr, next_read_ptr);
+
+    // __android_log_print(ANDROID_LOG_INFO, name_,
+    //                    "PUT: ptr_read=%d ptr_write=%d\n",
+    //                    ptr_read.load(), ptr_write.load());
+}
+
+bool pj_atomic_queue::inc_ptr_read_if_not_yet(unsigned old_ptr) {
+    unsigned new_ptr = (old_ptr == max_item_cnt_-1)? 0 : (old_ptr+1);
+    return ptr_read.compare_exchange_strong(old_ptr, new_ptr);
+}
+
+/* Increment write pointer */
+unsigned pj_atomic_queue::inc_ptr_write(unsigned old_ptr) {
+    unsigned new_ptr = (old_ptr == max_item_cnt_-1)? 0 : (old_ptr+1);
+    if (ptr_write.compare_exchange_strong(old_ptr, new_ptr))
+        return new_ptr;
+
+    /* Should never happen */
+    pj_assert(!"There is more than one producer!");
+    return old_ptr;
+}
+
+#endif

pjmedia/src/pjmedia-audiodev/oboe_dev.cpp

@@ -21,6 +21,7 @@
 
 #include <pjmedia-audiodev/audiodev_imp.h>
 #include <pj/assert.h>
+#include <pj/atomic_queue.h>
 #include <pj/lock.h>
 #include <pj/log.h>
 #include <pj/os.h>
@@ -38,9 +39,6 @@
 #include <android/log.h>
 #include <oboe/Oboe.h>
 
-#include <atomic>
-
-
 /* Device info */
 typedef struct aud_dev_info
 {
@@ -525,106 +523,6 @@ static pj_status_t oboe_default_param(pjmedia_aud_dev_factory *ff,
     return PJ_SUCCESS;
 }
 
-
-/* Atomic queue (ring buffer) for single consumer & single producer.
- *
- * Producer invokes 'put(frame)' to put a frame to the back of the queue and
- * consumer invokes 'get(frame)' to get a frame from the head of the queue.
- *
- * For producer, there is write pointer 'ptrWrite' that will be incremented
- * every time a frame is queued 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 frame in the head of the
- * queue by incrementing read pointer.
- *
- * For consumer, there is read pointer 'ptrRead' that will be incremented
- * every time a frame is fetched from the head of the queue, only if the
- * pointer is not modified by producer (in case of queue full).
- */
-class AtomicQueue {
-public:
-
-    AtomicQueue(unsigned max_frame_cnt, unsigned frame_size,
-                const char* name_= "") :
-        maxFrameCnt(max_frame_cnt), frameSize(frame_size),
-        ptrWrite(0), ptrRead(0),
-        buffer(NULL), name(name_)
-    {
-        buffer = new char[maxFrameCnt * frameSize];
-
-        /* Surpress warning when debugging log is disabled */
-        PJ_UNUSED_ARG(name);
-    }
-
-    ~AtomicQueue() {
-        delete [] buffer;
-    }
-
-    /* Get a frame from the head of the queue */
-    bool get(void* frame) {
-        if (ptrRead == ptrWrite)
-            return false;
-
-        unsigned cur_ptr = ptrRead;
-        void *p = &buffer[cur_ptr * frameSize];
-        pj_memcpy(frame, p, frameSize);
-        inc_ptr_read_if_not_yet(cur_ptr);
-
-        //__android_log_print(ANDROID_LOG_INFO, name,
-        //                    "GET: ptrRead=%d ptrWrite=%d\n",
-        //                    ptrRead.load(), ptrWrite.load());
-        return true;
-    }
-
-    /* Put a frame to the back of the queue */
-    void put(void* frame) {
-        unsigned cur_ptr = ptrWrite;
-        void *p = &buffer[cur_ptr * frameSize];
-        pj_memcpy(p, frame, frameSize);
-        unsigned next_ptr = inc_ptr_write(cur_ptr);
-
-        /* Increment read pointer if next write is overlapping (next_ptr == read ptr) */
-        unsigned next_read_ptr = (next_ptr == maxFrameCnt-1)? 0 : (next_ptr+1);
-        ptrRead.compare_exchange_strong(next_ptr, next_read_ptr);
-
-        //__android_log_print(ANDROID_LOG_INFO, name,
-        //                    "PUT: ptrRead=%d ptrWrite=%d\n",
-        //                    ptrRead.load(), ptrWrite.load());
-    }
-
-private:
-
-    unsigned maxFrameCnt;
-    unsigned frameSize;
-    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_ptr_read_if_not_yet(unsigned old_ptr) {
-        unsigned new_ptr = (old_ptr == maxFrameCnt-1)? 0 : (old_ptr+1);
-        return ptrRead.compare_exchange_strong(old_ptr, new_ptr);
-    }
-
-    /* Increment write pointer */
-    unsigned inc_ptr_write(unsigned old_ptr) {
-        unsigned new_ptr = (old_ptr == maxFrameCnt-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() {}
-};
-
-
 /* Interface to Oboe */
 class MyOboeEngine : oboe::AudioStreamDataCallback,
                      oboe::AudioStreamErrorCallback
@@ -720,8 +618,9 @@ class MyOboeEngine : oboe::AudioStreamDataCallback,
                   "Oboe stream %s queue size=%d frames (latency=%d ms)",
                   dir_st, queue_size, latency));
 
-        queue = new AtomicQueue(queue_size, stream->param.samples_per_frame*2,
-                                dir_st);
+        queue = new pj_atomic_queue(queue_size,
+                                    stream->param.samples_per_frame*2,
+                                    dir_st);
 
         /* Create semaphore */
         if (sem_init(&sem, 0, 0) != 0) {
@@ -1031,7 +930,7 @@ class MyOboeEngine : oboe::AudioStreamDataCallback,
     pj_thread_t                 *thread;
     volatile pj_bool_t           thread_quit;
     sem_t                        sem;
-    AtomicQueue                 *queue;
+    pj_atomic_queue             *queue;
     pj_timestamp                 ts;
     bool                         err_thread_registered;
     pj_thread_desc               err_thread_desc;