Index: pjproject/main/pjlib/src/pjlib-test/ioq_perf.c
===================================================================
--- pjproject/main/pjlib/src/pjlib-test/ioq_perf.c	(revision 6)
+++ pjproject/main/pjlib/src/pjlib-test/ioq_perf.c	(revision 11)
@@ -35,14 +35,17 @@
 typedef struct test_item
 {
-    pj_sock_t       server_fd, 
-                    client_fd;
-    pj_ioqueue_t    *ioqueue;
-    pj_ioqueue_key_t *server_key,
-                   *client_key;
-    pj_size_t       buffer_size;
-    char           *outgoing_buffer;
-    char           *incoming_buffer;
-    pj_size_t       bytes_sent, 
-                    bytes_recv;
+    pj_sock_t            server_fd, 
+                         client_fd;
+    pj_ioqueue_t        *ioqueue;
+    pj_ioqueue_key_t    *server_key,
+                        *client_key;
+    pj_ioqueue_op_key_t  recv_op,
+                         send_op;
+    int                  has_pending_send;
+    pj_size_t            buffer_size;
+    char                *outgoing_buffer;
+    char                *incoming_buffer;
+    pj_size_t            bytes_sent, 
+                         bytes_recv;
 } test_item;
 
@@ -50,54 +53,77 @@
  * Increment item->bytes_recv and ready to read the next data.
  */
-static void on_read_complete(pj_ioqueue_key_t *key, pj_ssize_t bytes_read)
+static void on_read_complete(pj_ioqueue_key_t *key, 
+                             pj_ioqueue_op_key_t *op_key,
+                             pj_ssize_t bytes_read)
 {
     test_item *item = pj_ioqueue_get_user_data(key);
     pj_status_t rc;
+    int data_is_available = 1;
 
     //TRACE_((THIS_FILE, "     read complete, bytes_read=%d", bytes_read));
 
-    if (thread_quit_flag)
-        return;
-
-    if (bytes_read < 0) {
-        pj_status_t rc = -bytes_read;
-        char errmsg[128];
-
-	if (rc != last_error) {
-	    last_error = rc;
-	    pj_strerror(rc, errmsg, sizeof(errmsg));
-	    PJ_LOG(3,(THIS_FILE, "...error: read error, bytes_read=%d (%s)", 
-		      bytes_read, errmsg));
-	    PJ_LOG(3,(THIS_FILE, 
-		      ".....additional info: total read=%u, total written=%u",
-		      item->bytes_recv, item->bytes_sent));
-	} else {
-	    last_error_counter++;
-	}
-        bytes_read = 0;
-
-    } else if (bytes_read == 0) {
-        PJ_LOG(3,(THIS_FILE, "...socket has closed!"));
-    }
-
-    item->bytes_recv += bytes_read;
+    do {
+        if (thread_quit_flag)
+            return;
+
+        if (bytes_read < 0) {
+            pj_status_t rc = -bytes_read;
+            char errmsg[128];
+
+	    if (rc != last_error) {
+	        last_error = rc;
+	        pj_strerror(rc, errmsg, sizeof(errmsg));
+	        PJ_LOG(3,(THIS_FILE, "...error: read error, bytes_read=%d (%s)", 
+		          bytes_read, errmsg));
+	        PJ_LOG(3,(THIS_FILE, 
+		          ".....additional info: total read=%u, total written=%u",
+		          item->bytes_recv, item->bytes_sent));
+	    } else {
+	        last_error_counter++;
+	    }
+            bytes_read = 0;
+
+        } else if (bytes_read == 0) {
+            PJ_LOG(3,(THIS_FILE, "...socket has closed!"));
+        }
+
+        item->bytes_recv += bytes_read;
     
-    /* To assure that the test quits, even if main thread
-     * doesn't have time to run.
-     */
-    if (item->bytes_recv > item->buffer_size * 10000) 
-	thread_quit_flag = 1;
-
-    rc = pj_ioqueue_recv( item->ioqueue, item->server_key,
-                          item->incoming_buffer, item->buffer_size, 0 );
-
-    if (rc != PJ_SUCCESS && rc != PJ_EPENDING) {
-	if (rc != last_error) {
-	    last_error = rc;
-	    app_perror("...error: read error", rc);
-	} else {
-	    last_error_counter++;
-	}
-    }
+        /* To assure that the test quits, even if main thread
+         * doesn't have time to run.
+         */
+        if (item->bytes_recv > item->buffer_size * 10000) 
+	    thread_quit_flag = 1;
+
+        bytes_read = item->buffer_size;
+        rc = pj_ioqueue_recv( key, op_key,
+                              item->incoming_buffer, &bytes_read, 0 );
+
+        if (rc == PJ_SUCCESS) {
+            data_is_available = 1;
+        } else if (rc == PJ_EPENDING) {
+            data_is_available = 0;
+        } else {
+            data_is_available = 0;
+	    if (rc != last_error) {
+	        last_error = rc;
+	        app_perror("...error: read error", rc);
+	    } else {
+	        last_error_counter++;
+	    }
+        }
+
+        if (!item->has_pending_send) {
+            pj_ssize_t sent = item->buffer_size;
+            rc = pj_ioqueue_send(item->client_key, &item->send_op,
+                                 item->outgoing_buffer, &sent, 0);
+            if (rc != PJ_SUCCESS && rc != PJ_EPENDING) {
+                app_perror("...error: write error", rc);
+            }
+
+            item->has_pending_send = (rc==PJ_EPENDING);
+        }
+
+    } while (data_is_available);
 }
 
@@ -105,5 +131,7 @@
  * Increment item->bytes_sent and write the next data.
  */
-static void on_write_complete(pj_ioqueue_key_t *key, pj_ssize_t bytes_sent)
+static void on_write_complete(pj_ioqueue_key_t *key, 
+                              pj_ioqueue_op_key_t *op_key,
+                              pj_ssize_t bytes_sent)
 {
     test_item *item = pj_ioqueue_get_user_data(key);
@@ -114,4 +142,5 @@
         return;
 
+    item->has_pending_send = 0;
     item->bytes_sent += bytes_sent;
 
@@ -123,9 +152,12 @@
         pj_status_t rc;
 
-        rc = pj_ioqueue_write(item->ioqueue, item->client_key, 
-                              item->outgoing_buffer, item->buffer_size);
+        bytes_sent = item->buffer_size;
+        rc = pj_ioqueue_send( item->client_key, op_key,
+                              item->outgoing_buffer, &bytes_sent, 0);
         if (rc != PJ_SUCCESS && rc != PJ_EPENDING) {
             app_perror("...error: write error", rc);
         }
+
+        item->has_pending_send = (rc==PJ_EPENDING);
     }
 }
@@ -192,5 +224,5 @@
 
     TRACE_((THIS_FILE, "     creating ioqueue.."));
-    rc = pj_ioqueue_create(pool, sockpair_cnt*2, thread_cnt, &ioqueue);
+    rc = pj_ioqueue_create(pool, sockpair_cnt*2, &ioqueue);
     if (rc != PJ_SUCCESS) {
         app_perror("...error: unable to create ioqueue", rc);
@@ -200,4 +232,5 @@
     /* Initialize each producer-consumer pair. */
     for (i=0; i<sockpair_cnt; ++i) {
+        pj_ssize_t bytes;
 
         items[i].ioqueue = ioqueue;
@@ -243,8 +276,9 @@
         /* Start reading. */
 	TRACE_((THIS_FILE, "      pj_ioqueue_recv.."));
-        rc = pj_ioqueue_recv(ioqueue, items[i].server_key,
-                             items[i].incoming_buffer, items[i].buffer_size,
+        bytes = items[i].buffer_size;
+        rc = pj_ioqueue_recv(items[i].server_key, &items[i].recv_op,
+                             items[i].incoming_buffer, &bytes,
 			     0);
-        if (rc != PJ_SUCCESS && rc != PJ_EPENDING) {
+        if (rc != PJ_EPENDING) {
             app_perror("...error: pj_ioqueue_recv", rc);
             return -73;
@@ -253,6 +287,7 @@
         /* Start writing. */
 	TRACE_((THIS_FILE, "      pj_ioqueue_write.."));
-        rc = pj_ioqueue_write(ioqueue, items[i].client_key,
-                              items[i].outgoing_buffer, items[i].buffer_size);
+        bytes = items[i].buffer_size;
+        rc = pj_ioqueue_send(items[i].client_key, &items[i].recv_op,
+                             items[i].outgoing_buffer, &bytes, 0);
         if (rc != PJ_SUCCESS && rc != PJ_EPENDING) {
             app_perror("...error: pj_ioqueue_write", rc);
@@ -260,4 +295,5 @@
         }
 
+        items[i].has_pending_send = (rc==PJ_EPENDING);
     }
 
@@ -325,6 +361,6 @@
     TRACE_((THIS_FILE, "     closing all sockets.."));
     for (i=0; i<sockpair_cnt; ++i) {
-        pj_ioqueue_unregister(ioqueue, items[i].server_key);
-        pj_ioqueue_unregister(ioqueue, items[i].client_key);
+        pj_ioqueue_unregister(items[i].server_key);
+        pj_ioqueue_unregister(items[i].client_key);
         pj_sock_close(items[i].server_fd);
         pj_sock_close(items[i].client_fd);
