Changeset 36 for pjproject/main/pjlib/src/pj/ioqueue_common_abs.c

Timestamp:

Nov 9, 2005 3:37:19 PM (18 years ago)

Author:

bennylp

Message:

Rework pjlib++

File:

• pjproject/main/pjlib/src/pj/ioqueue_common_abs.c (modified) (15 diffs)

pjproject/main/pjlib/src/pj/ioqueue_common_abs.c

@@ -229 +229 @@
         if (h->fd_type == PJ_SOCK_DGRAM) {
             pj_list_erase(write_op);
+            write_op->op = 0;
+
             if (pj_list_empty(&h->write_list))
                 ioqueue_remove_from_set(ioqueue, h->fd, WRITEABLE_EVENT);
@@ -269 +271 @@
                 /* Write completion of the whole stream. */
                 pj_list_erase(write_op);
+                write_op->op = 0;
 
                 /* Clear operation if there's no more data to send. */
@@ -315 +318 @@
         accept_op = h->accept_list.next;
         pj_list_erase(accept_op);
+        accept_op->op = 0;
 
         /* Clear bit in fdset if there is no more pending accept */
@@ -348 +352 @@
         read_op = h->read_list.next;
         pj_list_erase(read_op);
+        read_op->op = 0;
 
         /* Clear fdset if there is no pending read. */
@@ -477 +482 @@
     PJ_CHECK_STACK();
 
+    read_op = (struct read_operation*)op_key;
+    read_op->op = 0;
+
     /* Try to see if there's data immediately available. 
      */
@@ -497 +505 @@
      * Must schedule asynchronous operation to the ioqueue.
      */
-    read_op = (struct read_operation*)op_key;
-
     read_op->op = PJ_IOQUEUE_OP_RECV;
     read_op->buf = buffer;
@@ -531 +537 @@
     PJ_ASSERT_RETURN(key && op_key && buffer && length, PJ_EINVAL);
     PJ_CHECK_STACK();
+
+    read_op = (struct read_operation*)op_key;
+    read_op->op = 0;
 
     /* Try to see if there's data immediately available. 
@@ -553 +562 @@
      * Must schedule asynchronous operation to the ioqueue.
      */
-    read_op = (struct read_operation*)op_key;
-
     read_op->op = PJ_IOQUEUE_OP_RECV_FROM;
     read_op->buf = buffer;
@@ -587 +594 @@
     PJ_ASSERT_RETURN(key && op_key && data && length, PJ_EINVAL);
     PJ_CHECK_STACK();
+
+    write_op = (struct write_operation*)op_key;
+    write_op->op = 0;
 
     /* Fast track:
@@ -625 +635 @@
      * Schedule asynchronous send.
      */
-    write_op = (struct write_operation*)op_key;
     write_op->op = PJ_IOQUEUE_OP_SEND;
     write_op->buf = (void*)data;
@@ -660 +669 @@
     PJ_ASSERT_RETURN(key && op_key && data && length, PJ_EINVAL);
     PJ_CHECK_STACK();
+
+    write_op = (struct write_operation*)op_key;
+    write_op->op = 0;
 
     /* Fast track:
@@ -703 +715 @@
      * Schedule asynchronous send.
      */
-    write_op = (struct write_operation*)op_key;
     write_op->op = PJ_IOQUEUE_OP_SEND_TO;
     write_op->buf = (void*)data;
@@ -736 +747 @@
     /* check parameters. All must be specified! */
     PJ_ASSERT_RETURN(key && op_key && new_sock, PJ_EINVAL);
+
+    accept_op = (struct accept_operation*)op_key;
+    accept_op->op = 0;
 
     /* Fast track:
@@ -768 +782 @@
      * connection available.
      */
-    accept_op = (struct accept_operation*)op_key;
-
     accept_op->op = PJ_IOQUEUE_OP_ACCEPT;
     accept_op->accept_fd = new_sock;
@@ -822 +834 @@
 #endif  /* PJ_HAS_TCP */
 
+/*
+ * pj_ioqueue_is_pending()
+ */
+PJ_DEF(pj_bool_t) pj_ioqueue_is_pending( pj_ioqueue_key_t *key,
+                                         pj_ioqueue_op_key_t *op_key )
+{
+    struct generic_operation *op_rec;
+
+    PJ_UNUSED_ARG(key);
+
+    op_rec = (struct generic_operation*)op_key;
+    return op_rec->op != 0;
+}
+
+
+/*
+ * pj_ioqueue_post_completion()
+ */
+PJ_DEF(pj_status_t) pj_ioqueue_post_completion( pj_ioqueue_key_t *key,
+                                                pj_ioqueue_op_key_t *op_key,
+                                                pj_ssize_t bytes_status )
+{
+    struct generic_operation *op_rec;
+
+    /*
+     * Find the operation key in all pending operation list to
+     * really make sure that it's still there; then call the callback.
+     */
+    pj_mutex_lock(key->mutex);
+
+    /* Find the operation in the pending read list. */
+    op_rec = (struct generic_operation*)key->read_list.next;
+    while (op_rec != (void*)&key->read_list) {
+        if (op_rec == (void*)op_key) {
+            pj_list_erase(op_rec);
+            op_rec->op = 0;
+            pj_mutex_unlock(key->mutex);
+
+            (*key->cb.on_read_complete)(key, op_key, bytes_status);
+            return PJ_SUCCESS;
+        }
+        op_rec = op_rec->next;
+    }
+
+    /* Find the operation in the pending write list. */
+    op_rec = (struct generic_operation*)key->write_list.next;
+    while (op_rec != (void*)&key->write_list) {
+        if (op_rec == (void*)op_key) {
+            pj_list_erase(op_rec);
+            op_rec->op = 0;
+            pj_mutex_unlock(key->mutex);
+
+            (*key->cb.on_write_complete)(key, op_key, bytes_status);
+            return PJ_SUCCESS;
+        }
+        op_rec = op_rec->next;
+    }
+
+    /* Find the operation in the pending accept list. */
+    op_rec = (struct generic_operation*)key->accept_list.next;
+    while (op_rec != (void*)&key->accept_list) {
+        if (op_rec == (void*)op_key) {
+            pj_list_erase(op_rec);
+            op_rec->op = 0;
+            pj_mutex_unlock(key->mutex);
+
+            (*key->cb.on_accept_complete)(key, op_key, 
+                                          PJ_INVALID_SOCKET,
+                                          bytes_status);
+            return PJ_SUCCESS;
+        }
+        op_rec = op_rec->next;
+    }
+
+    pj_mutex_unlock(key->mutex);
+    
+    return PJ_EINVALIDOP;
+}
+