Refactor SSL socket send path and improve test coverage

.github/workflows/ci-linux.yml

@@ -545,6 +545,57 @@ jobs:
         name: ${{ runner.os }}-${{ runner.arch }}-${{ github.job }}-${{ github.run_id }}
         path: artifacts
 
+  ioq-no-fast-track:
+    needs: [detect-changes]
+    if: github.event_name == 'push' || needs.detect-changes.outputs.pjlib_deps == 'true'
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - name: select+ossl
+            config_site: |
+              #define PJ_TODO(x)
+              #define PJ_IOQUEUE_FAST_TRACK 0
+              #define PJ_IOQUEUE_IMP PJ_IOQUEUE_IMP_SELECT
+            configure_args: ""
+            install_deps: ""
+          - name: epoll+gtls
+            config_site: |
+              #define PJ_TODO(x)
+              #define PJ_IOQUEUE_FAST_TRACK 0
+            configure_args: "--with-gnutls=/usr/"
+            install_deps: "sudo apt-get update && sudo apt-get install -y libgnutls28-dev"
+    name: ioq-no-fast-track / ${{ matrix.name }}
+    steps:
+    - uses: actions/checkout@v2
+    - name: install cirunner
+      run: |
+        git clone --depth 1 https://github.com/pjsip/cirunner.git
+        cirunner/installlinux.sh
+    - name: install dependencies
+      if: matrix.install_deps != ''
+      run: ${{ matrix.install_deps }}
+    - name: config site
+      run: |
+        cat > pjlib/include/pj/config_site.h << 'SITE_EOF'
+        ${{ matrix.config_site }}
+        SITE_EOF
+    - name: configure
+      run: CFLAGS="-g -fsanitize=address" LDFLAGS="-rdynamic -fsanitize=address" ./configure ${{ matrix.configure_args }}
+    - name: make
+      run: $MAKE_FAST
+    - name: pjlib ioqueue, activesock, ssl_sock test
+      run: |
+        export LSAN_OPTIONS="suppressions=$GITHUB_WORKSPACE/tests/sanitizers/lsan.supp"
+        cd pjlib/build && ../bin/pjlib-test-`make -s -C ../.. infotarget` $CI_MODE $CI_ARGS udp_ioqueue_test tcp_ioqueue_test udp_ioqueue_unreg_test activesock_test ssl_sock_test ssl_sock_stress_test
+    - name: upload artifacts on failure
+      if: ${{ failure() }}
+      uses: actions/upload-artifact@v4
+      with:
+        name: ${{ runner.os }}-${{ runner.arch }}-${{ github.job }}-${{ matrix.name }}-${{ github.run_id }}
+        path: artifacts
+
   cmake-build:
     runs-on: ubuntu-latest
     name: CMake / ${{ matrix.build_type }}

pjlib/CMakeLists.txt

@@ -729,6 +729,7 @@ if(BUILD_TESTING)
     src/pjlib-test/sock.c
     src/pjlib-test/sock_perf.c
     src/pjlib-test/ssl_sock.c
+    src/pjlib-test/ssl_sock_stress.c
     src/pjlib-test/string.c
     src/pjlib-test/test.c
     src/pjlib-test/thread.c

pjlib/build/Makefile

@@ -51,7 +51,7 @@ export TEST_OBJS += activesock.o atomic.o atomic_slist.o \
 		    fifobuf.o file.o hash_test.o ioq_perf.o ioq_udp.o \
 		    ioq_stress_test.o ioq_unreg.o ioq_tcp.o ioq_iocp_unreg_test.o \
 		    list.o mutex.o os.o pool.o pool_perf.o rand.o rbtree.o \
-		    select.o sleep.o sock.o sock_perf.o ssl_sock.o \
+		    select.o sleep.o sock.o sock_perf.o ssl_sock.o ssl_sock_stress.o \
 		    string.o test.o thread.o timer.o timestamp.o \
 		    udp_echo_srv_sync.o udp_echo_srv_ioqueue.o \
 		    unittest_test.o util.o

pjlib/build/pjlib_test.vcxproj

@@ -789,6 +789,7 @@
     <ClCompile Include="..\src\pjlib-test\sock.c" />
     <ClCompile Include="..\src\pjlib-test\sock_perf.c" />
     <ClCompile Include="..\src\pjlib-test\ssl_sock.c" />
+    <ClCompile Include="..\src\pjlib-test\ssl_sock_stress.c" />
     <ClCompile Include="..\src\pjlib-test\atomic_slist.c" />
     <ClCompile Include="..\src\pjlib-test\string.c" />
     <ClCompile Include="..\src\pjlib-test\test.c" />

pjlib/build/pjlib_test.vcxproj.filters

@@ -93,6 +93,9 @@
     <ClCompile Include="..\src\pjlib-test\ssl_sock.c">
       <Filter>Source Files</Filter>
     </ClCompile>
+    <ClCompile Include="..\src\pjlib-test\ssl_sock_stress.c">
+      <Filter>Source Files</Filter>
+    </ClCompile>
     <ClCompile Include="..\src\pjlib-test\string.c">
       <Filter>Source Files</Filter>
     </ClCompile>

pjlib/src/pj/ssl_sock_apple.m

@@ -1135,12 +1135,12 @@ static pj_status_t network_setup_connection(pj_ssl_sock_t *ssock,
     pj_status_t status;
 
     /* Initialize input circular buffer */
-    status = circ_init(ssock->pool->factory, &ssock->circ_buf_input, 8192);
+    status = circ_init(ssock->pool->factory, &ssock->ssl_read_buf, 8192);
     if (status != PJ_SUCCESS)
         return status;
 
     /* Initialize output circular buffer */
-    status = circ_init(ssock->pool->factory, &ssock->circ_buf_output, 8192);
+    status = circ_init(ssock->pool->factory, &ssock->ssl_write_buf, 8192);
     if (status != PJ_SUCCESS)
         return status;
 
@@ -1525,8 +1525,8 @@ static void ssl_destroy(pj_ssl_sock_t *ssock)
     }
 
     /* Destroy circular buffers */
-    circ_deinit(&ssock->circ_buf_input);
-    circ_deinit(&ssock->circ_buf_output);
+    circ_deinit(&ssock->ssl_read_buf);
+    circ_deinit(&ssock->ssl_write_buf);
 
     PJ_LOG(4, (THIS_FILE, "SSL %p destroyed", ssock));
 }
@@ -1535,9 +1535,9 @@ static void ssl_destroy(pj_ssl_sock_t *ssock)
 /* Reset socket state. */
 static void ssl_reset_sock_state(pj_ssl_sock_t *ssock)
 {
-    pj_lock_acquire(ssock->circ_buf_output_mutex);
+    pj_lock_acquire(ssock->ssl_write_buf_mutex);
     ssock->ssl_state = SSL_STATE_NULL;
-    pj_lock_release(ssock->circ_buf_output_mutex);
+    pj_lock_release(ssock->ssl_write_buf_mutex);
 
 #if SSL_DEBUG
     PJ_LOG(3, (THIS_FILE, "SSL reset sock state %p", ssock));
@@ -2170,20 +2170,20 @@ static pj_status_t ssl_read(pj_ssl_sock_t *ssock, void *data, int *size)
 {
     pj_size_t circ_buf_size, read_size;
 
-    pj_lock_acquire(ssock->circ_buf_input_mutex);
+    pj_lock_acquire(ssock->ssl_read_buf_mutex);
 
-    if (circ_empty(&ssock->circ_buf_input)) {
-        pj_lock_release(ssock->circ_buf_input_mutex);
+    if (circ_empty(&ssock->ssl_read_buf)) {
+        pj_lock_release(ssock->ssl_read_buf_mutex);
         *size = 0;
         return PJ_SUCCESS;
     }
 
-    circ_buf_size = circ_size(&ssock->circ_buf_input);
+    circ_buf_size = circ_size(&ssock->ssl_read_buf);
     read_size = PJ_MIN(circ_buf_size, (pj_size_t)*size);
 
-    circ_read(&ssock->circ_buf_input, data, read_size);
+    circ_read(&ssock->ssl_read_buf, data, read_size);
 
-    pj_lock_release(ssock->circ_buf_input_mutex);
+    pj_lock_release(ssock->ssl_read_buf_mutex);
 
     *size = read_size;
 
@@ -2192,14 +2192,14 @@ static pj_status_t ssl_read(pj_ssl_sock_t *ssock, void *data, int *size)
 
 /*
  * Write the plain data to buffer. It will be encrypted later during
- * sending.
+ * sending. Caller must hold ssock->write_mutex.
  */
 static pj_status_t ssl_write(pj_ssl_sock_t *ssock, const void *data,
                              pj_ssize_t size, int *nwritten)
 {
     pj_status_t status;
 
-    status = circ_write(&ssock->circ_buf_output, data, size);
+    status = circ_write(&ssock->ssl_write_buf, data, size);
     *nwritten = (status == PJ_SUCCESS)? (int)size: 0;
 
     return status;

pjlib/src/pj/ssl_sock_darwin.c

@@ -109,7 +109,7 @@ static OSStatus SocketWrite(SSLConnectionRef connection,
     pj_size_t len = *dataLength;
 
     pj_lock_acquire(ssock->write_mutex);
-    if (circ_write(&ssock->circ_buf_output, data, len) != PJ_SUCCESS) {
+    if (circ_write(&ssock->ssl_write_buf, data, len) != PJ_SUCCESS) {
         pj_lock_release(ssock->write_mutex);
         *dataLength = 0;
         return errSSLInternal;
@@ -128,22 +128,22 @@ static OSStatus SocketRead(SSLConnectionRef connection,
     pj_ssl_sock_t *ssock = (pj_ssl_sock_t *)connection;
     pj_size_t len = *dataLength;
 
-    pj_lock_acquire(ssock->circ_buf_input_mutex);
+    pj_lock_acquire(ssock->ssl_read_buf_mutex);
 
-    if (circ_empty(&ssock->circ_buf_input)) {
-        pj_lock_release(ssock->circ_buf_input_mutex);
+    if (circ_empty(&ssock->ssl_read_buf)) {
+        pj_lock_release(ssock->ssl_read_buf_mutex);
 
         /* Data buffers not yet filled */
         *dataLength = 0;
         return errSSLWouldBlock;
     }
 
-    pj_size_t circ_buf_size = circ_size(&ssock->circ_buf_input);
+    pj_size_t circ_buf_size = circ_size(&ssock->ssl_read_buf);
     pj_size_t read_size = PJ_MIN(circ_buf_size, len);
 
-    circ_read(&ssock->circ_buf_input, data, read_size);
+    circ_read(&ssock->ssl_read_buf, data, read_size);
 
-    pj_lock_release(ssock->circ_buf_input_mutex);
+    pj_lock_release(ssock->ssl_read_buf_mutex);
 
     *dataLength = read_size;
 
@@ -377,12 +377,12 @@ static pj_status_t ssl_create(pj_ssl_sock_t *ssock)
     pj_status_t status;
 
    /* Initialize input circular buffer */
-    status = circ_init(ssock->pool->factory, &ssock->circ_buf_input, 8192);
+    status = circ_init(ssock->pool->factory, &ssock->ssl_read_buf, 8192);
     if (status != PJ_SUCCESS)
         return status;
 
     /* Initialize output circular buffer */
-    status = circ_init(ssock->pool->factory, &ssock->circ_buf_output, 8192);
+    status = circ_init(ssock->pool->factory, &ssock->ssl_write_buf, 8192);
     if (status != PJ_SUCCESS)
         return status;
 
@@ -511,17 +511,17 @@ static void ssl_destroy(pj_ssl_sock_t *ssock)
     }
 
     /* Destroy circular buffers */
-    circ_deinit(&ssock->circ_buf_input);
-    circ_deinit(&ssock->circ_buf_output);
+    circ_deinit(&ssock->ssl_read_buf);
+    circ_deinit(&ssock->ssl_write_buf);
 }
 
 
 /* Reset socket state. */
 static void ssl_reset_sock_state(pj_ssl_sock_t *ssock)
 {
-    pj_lock_acquire(ssock->circ_buf_output_mutex);
+    pj_lock_acquire(ssock->ssl_write_buf_mutex);
     ssock->ssl_state = SSL_STATE_NULL;
-    pj_lock_release(ssock->circ_buf_output_mutex);
+    pj_lock_release(ssock->ssl_write_buf_mutex);
 
     ssl_close_sockets(ssock);
 }
@@ -1366,11 +1366,6 @@ static pj_status_t ssl_do_handshake(pj_ssl_sock_t *ssock)
     }
     pj_lock_release(ssock->write_mutex);
 
-    status = flush_circ_buf_output(ssock, &ssock->handshake_op_key, 0, 0);
-    if (status != PJ_SUCCESS && status != PJ_EPENDING) {
-        return status;
-    }
-
     if (ret == noErr) {
         /* Handshake has been completed */
         ssock->ssl_state = SSL_STATE_ESTABLISHED;
@@ -1402,6 +1397,7 @@ static pj_status_t ssl_read(pj_ssl_sock_t *ssock, void *data, int *size)
 /*
  * Write the plain data to Darwin SSL, it will be encrypted by SSLWrite()
  * and call SocketWrite.
+ * Caller must hold ssock->write_mutex.
  */
 static pj_status_t ssl_write(pj_ssl_sock_t *ssock, const void *data,
                              pj_ssize_t size, int *nwritten)