Changeset 563 for pjproject/trunk/pjsip/src/pjsip/sip_transport_tcp.c

Timestamp:

Jun 28, 2006 4:46:49 PM (17 years ago)

Author:

bennylp

Message:

Major improvements in PJSIP to support TCP. The changes fall into these categories: (1) the TCP transport implementation itself (*.[hc]), (2) bug-fix in SIP transaction when using reliable transports, (3) support for TCP transport in PJSUA-LIB/PJSUA, and (4) changes in PJSIP-TEST to support TCP testing.

File:

• pjproject/trunk/pjsip/src/pjsip/sip_transport_tcp.c (modified) (45 diffs)

pjproject/trunk/pjsip/src/pjsip/sip_transport_tcp.c

@@ -21 +21 @@
 #include <pjsip/sip_errno.h>
 #include <pj/compat/socket.h>
+#include <pj/addr_resolv.h>
 #include <pj/assert.h>
 #include <pj/ioqueue.h>
@@ -34 +35 @@
 #define MAX_ASYNC_CNT   16
 #define POOL_LIS_INIT   4000
-#define POOL_LIS_INC    4000
+#define POOL_LIS_INC    4001
 #define POOL_TP_INIT    4000
-#define POOL_TP_INC     4000
+#define POOL_TP_INC     4002
 
 
@@ -43 +44 @@
 
 
+/*
+ * This structure is "descendant" of pj_ioqueue_op_key_t, and it is used to
+ * track pending/asynchronous accept() operation. TCP transport may have
+ * more than one pending accept() operations, depending on the value of
+ * async_cnt.
+ */
 struct pending_accept
 {
@@ -53 +60 @@
 };
 
-struct pending_connect
-{
-    pj_ioqueue_op_key_t      op_key;
-    struct tcp_transport    *transport;
-};
-
-
+
+/*
+ * This is the TCP listener, which is a "descendant" of pjsip_tpfactory (the
+ * SIP transport factory).
+ */
 struct tcp_listener
 {
     pjsip_tpfactory          factory;
-    char                     name[PJ_MAX_OBJ_NAME];
-    pj_bool_t                active;
+    char                     obj_name[PJ_MAX_OBJ_NAME];
+    pj_bool_t                is_registered;
     pjsip_endpoint          *endpt;
     pjsip_tpmgr             *tpmgr;
@@ -74 +79 @@
 
 
-struct pending_tdata
-{
-    PJ_DECL_LIST_MEMBER(struct pending_tdata);
+/*
+ * This structure is used to keep delayed transmit operation in a list.
+ * A delayed transmission occurs when application sends tx_data when
+ * the TCP connect/establishment is still in progress. These delayed
+ * transmission will be "flushed" once the socket is connected (either
+ * successfully or with errors).
+ */
+struct delayed_tdata
+{
+    PJ_DECL_LIST_MEMBER(struct delayed_tdata);
     pjsip_tx_data_op_key    *tdata_op_key;
 };
 
 
+/*
+ * This structure describes the TCP transport, and it's descendant of
+ * pjsip_transport.
+ */
 struct tcp_transport
 {
     pjsip_transport          base;
+    pj_bool_t                is_server;
     struct tcp_listener     *listener;
     pj_bool_t                is_registered;
     pj_bool_t                is_closing;
+    pj_status_t              close_reason;
     pj_sock_t                sock;
     pj_ioqueue_key_t        *key;
     pj_bool_t                has_pending_connect;
-    struct pending_connect   connect_op;
 
 
@@ -100 +117 @@
 
     /* Pending transmission list. */
-    struct pending_tdata     tx_list;
+    struct delayed_tdata     delayed_list;
 };
 
 
-/*
- * This callback is called when #pj_ioqueue_accept completes.
- */
+/****************************************************************************
+ * PROTOTYPES
+ */
+
+/* This callback is called when pending accept() operation completes. */
 static void on_accept_complete( pj_ioqueue_key_t *key, 
                                 pj_ioqueue_op_key_t *op_key, 
@@ -112 +131 @@
                                 pj_status_t status);
 
-static pj_status_t lis_destroy(struct tcp_listener *listener);
+/* This callback is called by transport manager to destroy listener */
+static pj_status_t lis_destroy(pjsip_tpfactory *factory);
+
+/* This callback is called by transport manager to create transport */
 static pj_status_t lis_create_transport(pjsip_tpfactory *factory,
                                         pjsip_tpmgr *mgr,
@@ -120 +142 @@
                                         pjsip_transport **transport);
 
-
-static pj_status_t create_tcp_transport(struct tcp_listener *listener,
-                                        pj_sock_t sock,
-                                        const pj_sockaddr_in *local,
-                                        const pj_sockaddr_in *remote,
-                                        struct tcp_transport **p_tcp);
+/* Common function to create and initialize transport */
+static pj_status_t tcp_create(struct tcp_listener *listener,
+                              pj_sock_t sock, pj_bool_t is_server,
+                              const pj_sockaddr_in *local,
+                              const pj_sockaddr_in *remote,
+                              struct tcp_transport **p_tcp);
 
 
@@ -139 +161 @@
 
 
+static void sockaddr_to_host_port( pj_pool_t *pool,
+                                   pjsip_host_port *host_port,
+                                   const pj_sockaddr_in *addr )
+{
+    host_port->host.ptr = pj_pool_alloc(pool, 48);
+    host_port->host.slen = pj_ansi_sprintf( host_port->host.ptr, "%s", 
+                                            pj_inet_ntoa(addr->sin_addr));
+    host_port->port = pj_ntohs(addr->sin_port);
+}
+
+
+
+/****************************************************************************
+ * The TCP listener/transport factory.
+ */
+
+/*
+ * This is the public API to create, initialize, register, and start the
+ * TCP listener.
+ */
 PJ_DEF(pj_status_t) pjsip_tcp_transport_start( pjsip_endpoint *endpt,
                                                const pj_sockaddr_in *local,
-                                               unsigned async_cnt)
+                                               unsigned async_cnt,
+                                               pjsip_tpfactory **p_factory)
 {
     pj_pool_t *pool;
     struct tcp_listener *listener;
     pj_ioqueue_callback listener_cb;
+    pj_sockaddr_in *listener_addr;
+    int addr_len;
     unsigned i;
     pj_status_t status;
 
     /* Sanity check */
-    PJ_ASSERT_RETURN(endpt && local && async_cnt, PJ_EINVAL);
+    PJ_ASSERT_RETURN(endpt && async_cnt, PJ_EINVAL);
 
 
@@ -159 +204 @@
 
     listener = pj_pool_zalloc(pool, sizeof(struct tcp_listener));
-    pj_ansi_sprintf(listener->name, "tcp:%d", (int)pj_ntohs(local->sin_port));
     listener->factory.pool = pool;
     listener->factory.type = PJSIP_TRANSPORT_TCP;
-    pj_ansi_strcpy(listener->factory.type_name, "tcp");
+    listener->factory.type_name = "tcp";
     listener->factory.flag = 
         pjsip_transport_get_flag_from_type(PJSIP_TRANSPORT_TCP);
     listener->sock = PJ_INVALID_SOCKET;
+
+    pj_ansi_strcpy(listener->obj_name, "tcp");
 
     status = pj_lock_create_recursive_mutex(pool, "tcplis", 
@@ -178 +224 @@
         goto on_error;
 
-    pj_memcpy(&listener->factory.local_addr, local, sizeof(pj_sockaddr_in));
-    status = pj_sock_bind(listener->sock, local, sizeof(*local));
+    listener_addr = (pj_sockaddr_in*)&listener->factory.local_addr;
+    if (local) {
+        pj_memcpy(listener_addr, local, sizeof(pj_sockaddr_in));
+    } else {
+        pj_sockaddr_in_init(listener_addr, NULL, 0);
+    }
+
+    status = pj_sock_bind(listener->sock, listener_addr, 
+                          sizeof(pj_sockaddr_in));
     if (status != PJ_SUCCESS)
         goto on_error;
+
+    /* Retrieve the bound address */
+    addr_len = sizeof(pj_sockaddr_in);
+    status = pj_sock_getsockname(listener->sock, listener_addr, &addr_len);
+    if (status != PJ_SUCCESS)
+        goto on_error;
+
+    /* If the address returns 0.0.0.0, use the first interface address
+     * as the transport's address.
+     */
+    if (listener_addr->sin_addr.s_addr == 0) {
+        const pj_str_t *hostname;
+        struct pj_hostent he;
+
+        hostname = pj_gethostname();
+        status = pj_gethostbyname(hostname, &he);
+        if (status != PJ_SUCCESS)
+            goto on_error;
+
+        listener_addr->sin_addr = *(pj_in_addr*)he.h_addr;
+    }
+
+    pj_ansi_sprintf(listener->obj_name, "tcp:%d", 
+                     (int)pj_ntohs(listener_addr->sin_port));
+
+    /* Save the address name */
+    sockaddr_to_host_port(listener->factory.pool, 
+                          &listener->factory.addr_name, listener_addr);
+
+    /* Start listening to the address */
+    status = pj_sock_listen(listener->sock, PJSIP_TCP_TRANSPORT_BACKLOG);
+    if (status != PJ_SUCCESS)
+        goto on_error;
+
 
     /* Register socket to ioqeuue */
@@ -192 +279 @@
         goto on_error;
 
-    /* Start pending accept() operation */
+    /* Register to transport manager */
+    listener->endpt = endpt;
+    listener->tpmgr = pjsip_endpt_get_tpmgr(endpt);
+    listener->factory.create_transport = lis_create_transport;
+    listener->factory.destroy = lis_destroy;
+    listener->is_registered = PJ_TRUE;
+    status = pjsip_tpmgr_register_tpfactory(listener->tpmgr,
+                                            &listener->factory);
+    if (status != PJ_SUCCESS) {
+        listener->is_registered = PJ_FALSE;
+        goto on_error;
+    }
+
+
+    /* Start pending accept() operations */
     if (async_cnt > MAX_ASYNC_CNT) async_cnt = MAX_ASYNC_CNT;
     listener->async_cnt = async_cnt;
@@ -201 +302 @@
         listener->accept_op[i].listener = listener;
 
-        status = pj_ioqueue_accept(listener->key, 
-                                   &listener->accept_op[i].op_key,
-                                   &listener->accept_op[i].new_sock,
-                                   &listener->accept_op[i].local_addr,
-                                   &listener->accept_op[i].remote_addr,
-                                   &listener->accept_op[i].addr_len);
-        if (status != PJ_SUCCESS && status != PJ_EPENDING)
-            goto on_error;
-    }
-
-    /* Register to transport manager */
-    listener->endpt = endpt;
-    listener->tpmgr = pjsip_endpt_get_tpmgr(endpt);
-    listener->factory.create_transport = lis_create_transport;
-    status = pjsip_tpmgr_register_tpfactory(listener->tpmgr,
-                                            &listener->factory);
-    if (status != PJ_SUCCESS)
-        goto on_error;
-
-    /* Done! */
-    listener->active = PJ_TRUE;
-
-    PJ_LOG(4,(listener->name, 
-             "SIP TCP transport listening for incoming connections at %s:%d",
-             pj_inet_ntoa(local->sin_addr), (int)pj_ntohs(local->sin_port)));
+        on_accept_complete(listener->key, &listener->accept_op[i].op_key,
+                           listener->sock, PJ_EPENDING);
+    }
+
+    PJ_LOG(4,(listener->obj_name, 
+             "SIP TCP listener ready for incoming connections at %s:%d",
+             pj_inet_ntoa(listener_addr->sin_addr), 
+             (int)pj_ntohs(listener_addr->sin_port)));
+
+    /* Return the pointer to user */
+    if (p_factory) *p_factory = &listener->factory;
 
     return PJ_SUCCESS;
 
 on_error:
-    lis_destroy(listener);
+    lis_destroy(&listener->factory);
     return status;
 }
 
 
-static pj_status_t lis_destroy(struct tcp_listener *listener)
-{
-    if (listener->active) {
+/* This callback is called by transport manager to destroy listener */
+static pj_status_t lis_destroy(pjsip_tpfactory *factory)
+{
+    struct tcp_listener *listener = (struct tcp_listener *)factory;
+
+    if (listener->is_registered) {
         pjsip_tpmgr_unregister_tpfactory(listener->tpmgr, &listener->factory);
-        listener->active = PJ_FALSE;
+        listener->is_registered = PJ_FALSE;
     }
 
@@ -259 +349 @@
 
     if (listener->factory.pool) {
-        PJ_LOG(4,(listener->name,  "SIP TCP transport destroyed"));
-        pj_pool_release(listener->factory.pool);
+        pj_pool_t *pool = listener->factory.pool;
+
+        PJ_LOG(4,(listener->obj_name,  "SIP TCP listener destroyed"));
+
         listener->factory.pool = NULL;
+        pj_pool_release(pool);
     }
 
@@ -289 +382 @@
 static pj_status_t tcp_shutdown(pjsip_transport *transport);
 
-/* Called by transport manager to destroy */
-static pj_status_t tcp_destroy(pjsip_transport *transport);
+/* Called by transport manager to destroy transport */
+static pj_status_t tcp_destroy_transport(pjsip_transport *transport);
+
+/* Utility to destroy transport */
+static pj_status_t tcp_destroy(pjsip_transport *transport,
+                               pj_status_t reason);
 
 /* Callback from ioqueue on incoming packet */
@@ -307 +404 @@
 
 
-static void sockaddr_to_host_port( pj_pool_t *pool,
-                                   pjsip_host_port *host_port,
-                                   const pj_sockaddr_in *addr )
-{
-    host_port->host.ptr = pj_pool_alloc(pool, 48);
-    host_port->host.slen = pj_ansi_sprintf( host_port->host.ptr, "%s", 
-                                            pj_inet_ntoa(addr->sin_addr));
-    host_port->port = pj_ntohs(addr->sin_port);
-}
-
-
 /*
- * Utilities to create TCP transport.
- */
-static pj_status_t create_tcp_transport(struct tcp_listener *listener,
-                                        pj_sock_t sock,
-                                        const pj_sockaddr_in *local,
-                                        const pj_sockaddr_in *remote,
-                                        struct tcp_transport **p_tcp)
+ * Common function to create TCP transport, called when pending accept() and
+ * pending connect() complete.
+ */
+static pj_status_t tcp_create( struct tcp_listener *listener,
+                               pj_sock_t sock, pj_bool_t is_server,
+                               const pj_sockaddr_in *local,
+                               const pj_sockaddr_in *remote,
+                               struct tcp_transport **p_tcp)
 {
     struct tcp_transport *tcp;
@@ -333 +420 @@
     pj_status_t status;
     
-    pool = pjsip_endpt_create_pool(listener->endpt, "tcp", 
+
+    PJ_ASSERT_RETURN(sock != PJ_INVALID_SOCKET, PJ_EINVAL);
+
+
+    pool = pjsip_endpt_create_pool(listener->endpt, "tcp",
                                    POOL_TP_INIT, POOL_TP_INC);
+    PJ_ASSERT_RETURN(pool != NULL, PJ_ENOMEM);
     
+
     /*
      * Create and initialize basic transport structure.
@@ -341 +434 @@
     tcp = pj_pool_zalloc(pool, sizeof(*tcp));
     tcp->sock = sock;
+    tcp->is_server = is_server;
     tcp->listener = listener;
-    pj_list_init(&tcp->tx_list);
-
-
-    pj_ansi_snprintf(tcp->base.obj_name, PJ_MAX_OBJ_NAME, "tcp%p", tcp);
+    pj_list_init(&tcp->delayed_list);
     tcp->base.pool = pool;
 
+    pj_ansi_snprintf(tcp->base.obj_name, PJ_MAX_OBJ_NAME, 
+                     (is_server ? "tcps%p" :"tcpc%p"), tcp);
+
     status = pj_atomic_create(pool, 0, &tcp->base.ref_cnt);
-    if (status != PJ_SUCCESS)
+    if (status != PJ_SUCCESS) {
         goto on_error;
+    }
 
     status = pj_lock_create_recursive_mutex(pool, "tcp", &tcp->base.lock);
-    if (status != PJ_SUCCESS)
+    if (status != PJ_SUCCESS) {
         goto on_error;
+    }
 
     tcp->base.key.type = PJSIP_TRANSPORT_TCP;
@@ -375 +471 @@
     tcp->base.send_msg = &tcp_send_msg;
     tcp->base.do_shutdown = &tcp_shutdown;
-    tcp->base.destroy = &tcp_destroy;
+    tcp->base.destroy = &tcp_destroy_transport;
 
 
@@ -387 +483 @@
     status = pj_ioqueue_register_sock(pool, ioqueue, sock, 
                                       tcp, &tcp_callback, &tcp->key);
-    if (status != PJ_SUCCESS)
+    if (status != PJ_SUCCESS) {
         goto on_error;
+    }
 
     /* Register transport to transport manager */
     status = pjsip_transport_register(listener->tpmgr, &tcp->base);
-    if (status != PJ_SUCCESS)
+    if (status != PJ_SUCCESS) {
         goto on_error;
+    }
 
     tcp->is_registered = PJ_TRUE;
@@ -400 +498 @@
     *p_tcp = tcp;
 
+    PJ_LOG(4,(tcp->base.obj_name, "TCP %s transport created",
+              (tcp->is_server ? "server" : "client")));
+
+    return PJ_SUCCESS;
+
 on_error:
-    tcp_destroy(&tcp->base);
+    tcp_destroy(&tcp->base, status);
     return status;
 }
 
 
-/* Flush all pending send operations */
-static tcp_flush_pending_tx(struct tcp_transport *tcp)
+/* Flush all delayed transmision once the socket is connected. */
+static void tcp_flush_pending_tx(struct tcp_transport *tcp)
 {
     pj_lock_acquire(tcp->base.lock);
-    while (!pj_list_empty(&tcp->tx_list)) {
-        struct pending_tdata *pending_tx;
+    while (!pj_list_empty(&tcp->delayed_list)) {
+        struct delayed_tdata *pending_tx;
         pjsip_tx_data *tdata;
         pj_ioqueue_op_key_t *op_key;
@@ -417 +520 @@
         pj_status_t status;
 
-        pending_tx = tcp->tx_list.next;
+        pending_tx = tcp->delayed_list.next;
         pj_list_erase(pending_tx);
 
@@ -437 +540 @@
 
 
+/* Called by transport manager to destroy transport */
+static pj_status_t tcp_destroy_transport(pjsip_transport *transport)
+{
+    struct tcp_transport *tcp = (struct tcp_transport*)transport;
+
+    /* Transport would have been unregistered by now since this callback
+     * is called by transport manager.
+     */
+    tcp->is_registered = PJ_FALSE;
+
+    return tcp_destroy(transport, tcp->close_reason);
+}
+
 
 /* Destroy TCP transport */
-static pj_status_t tcp_destroy(pjsip_transport *transport)
+static pj_status_t tcp_destroy(pjsip_transport *transport, 
+                               pj_status_t reason)
 {
     struct tcp_transport *tcp = (struct tcp_transport*)transport;
 
-    /* Cancel all pending transmits */
-    while (!pj_list_empty(&tcp->tx_list)) {
-        struct pending_tdata *pending_tx;
+    if (tcp->close_reason == 0)
+        tcp->close_reason = reason;
+
+    if (tcp->is_registered) {
+        tcp->is_registered = PJ_FALSE;
+        pjsip_transport_destroy(transport);
+
+        /* pjsip_transport_destroy will recursively call this function
+         * again.
+         */
+        return PJ_SUCCESS;
+    }
+
+    /* Mark transport as closing */
+    tcp->is_closing = PJ_TRUE;
+
+    /* Cancel all delayed transmits */
+    while (!pj_list_empty(&tcp->delayed_list)) {
+        struct delayed_tdata *pending_tx;
         pj_ioqueue_op_key_t *op_key;
 
-        pending_tx = tcp->tx_list.next;
+        pending_tx = tcp->delayed_list.next;
         pj_list_erase(pending_tx);
 
         op_key = (pj_ioqueue_op_key_t*)pending_tx->tdata_op_key;
 
-        on_write_complete(tcp->key, op_key, 
-                          -PJ_RETURN_OS_ERROR(OSERR_ENOTCONN));
-    }
-
-    if (tcp->is_registered) {
-        pjsip_transport_destroy(transport);
-        tcp->is_registered = PJ_FALSE;
+        on_write_complete(tcp->key, op_key, -reason);
     }
 
@@ -470 +597 @@
         pj_ioqueue_unregister(tcp->key);
         tcp->key = NULL;
+        tcp->sock = PJ_INVALID_SOCKET;
+    }
+
+    if (tcp->sock != PJ_INVALID_SOCKET) {
+        pj_sock_close(tcp->sock);
+        tcp->sock = PJ_INVALID_SOCKET;
     }
 
@@ -483 +616 @@
 
     if (tcp->base.pool) {
-        PJ_LOG(4,(tcp->base.obj_name, "TCP transport destroyed"));
-        pj_pool_release(tcp->base.pool);
+        pj_pool_t *pool;
+
+        if (reason != PJ_SUCCESS) {
+            char errmsg[PJ_ERR_MSG_SIZE];
+
+            pj_strerror(reason, errmsg, sizeof(errmsg));
+            PJ_LOG(4,(tcp->base.obj_name, 
+                      "TCP transport destroyed with reason %d: %s", 
+                      reason, errmsg));
+
+        } else {
+
+            PJ_LOG(4,(tcp->base.obj_name, 
+                      "TCP transport destroyed normally"));
+
+        }
+
+        pool = tcp->base.pool;
         tcp->base.pool = NULL;
+        pj_pool_release(pool);
     }
 
@@ -494 +644 @@
 /*
  * This utility function creates receive data buffers and start
- * asynchronous recv() operations from the socket.
+ * asynchronous recv() operations from the socket. It is called after
+ * accept() or connect() operation complete.
  */
 static pj_status_t tcp_start_read(struct tcp_transport *tcp)
@@ -532 +683 @@
                              tcp->rdata.pkt_info.packet, &size,
                              PJ_IOQUEUE_ALWAYS_ASYNC);
-    if (status != PJ_SUCCESS) {
+    if (status != PJ_SUCCESS && status != PJ_EPENDING) {
         tcp_perror(tcp->base.obj_name, "ioqueue recv() error", status);
         return status;
@@ -594 +745 @@
 
     /* Create the transport descriptor */
-    status = create_tcp_transport(listener, sock, &local_addr, 
-                                  (pj_sockaddr_in*)rem_addr, &tcp);
+    status = tcp_create(listener, sock, PJ_FALSE, &local_addr, 
+                        (pj_sockaddr_in*)rem_addr, &tcp);
     if (status != PJ_SUCCESS)
         return status;
-        
+
+
     /* Start asynchronous connect() operation */
     tcp->has_pending_connect = PJ_TRUE;
-    pj_ioqueue_op_key_init(&tcp->connect_op.op_key, 
-                           sizeof(tcp->connect_op.op_key));
-    tcp->connect_op.transport = tcp;
     status = pj_ioqueue_connect(tcp->key, rem_addr, sizeof(pj_sockaddr_in));
-    if (status != PJ_SUCCESS) {
-        tcp_destroy(&tcp->base);
+    if (status == PJ_SUCCESS) {
+        tcp->has_pending_connect = PJ_FALSE;
+    } else if (status != PJ_EPENDING) {
+        tcp_destroy(&tcp->base, status);
         return status;
     }
@@ -630 +781 @@
     }
 
+    if (tcp->has_pending_connect) {
+        PJ_LOG(4,(tcp->base.obj_name, 
+                  "TCP transport %.*s:%d is connecting to %.*s:%d...",
+                  (int)tcp->base.local_name.host.slen,
+                  tcp->base.local_name.host.ptr,
+                  tcp->base.local_name.port,
+                  (int)tcp->base.remote_name.host.slen,
+                  tcp->base.remote_name.host.ptr,
+                  tcp->base.remote_name.port));
+    }
+
     /* Done */
     *p_transport = &tcp->base;
@@ -654 +816 @@
     accept_op = (struct pending_accept*) op_key;
 
+    /*
+     * Loop while there is immediate connection or when there is error.
+     */
     do {
-        if (status != PJ_SUCCESS) {
-            tcp_perror(listener->name, "Error in accept()", status);
-
+        if (status == PJ_EPENDING) {
+            /*
+             * This can only happen when this function is called during
+             * initialization to kick off asynchronous accept().
+             */
+
+        } else if (status != PJ_SUCCESS) {
+
+            /*
+             * Error in accept().
+             */
+            tcp_perror(listener->obj_name, "Error in accept()", status);
+
+            /*
+             * Prevent endless accept() error loop by limiting the
+             * number of consecutive errors. Once the number of errors
+             * is equal to maximum, we treat this as permanent error, and
+             * we stop the accept() operation.
+             */
             ++err_cnt;
-            if (err_cnt >= 5) {
-                PJ_LOG(1, (listener->name, 
+            if (err_cnt >= 10) {
+                PJ_LOG(1, (listener->obj_name, 
                            "Too many errors, listener stopping"));
             }
 
-            goto start_next_accept;
-        }
-
-        status = create_tcp_transport( listener, sock, 
-                                       &accept_op->local_addr, 
-                                       &accept_op->remote_addr, &tcp);
-        if (status == PJ_SUCCESS) {
-            status = tcp_start_read(tcp);
-            if (status != PJ_SUCCESS) {
-                PJ_LOG(3,(tcp->base.obj_name, "New transport cancelled"));
-                tcp_destroy(&tcp->base);
+        } else {
+
+            if (sock == PJ_INVALID_SOCKET) {
+                sock = accept_op->new_sock;
+                PJ_LOG(4,(listener->obj_name, 
+                          "Warning: ioqueue reports -1 in on_accept_complete()"
+                          " sock argument"));
+            }
+
+            PJ_LOG(4,(listener->obj_name, 
+                      "TCP listener %.*s:%d: got incoming TCP connection "
+                      "from %s:%d, sock=%d",
+                      (int)listener->factory.addr_name.host.slen,
+                      listener->factory.addr_name.host.ptr,
+                      listener->factory.addr_name.port,
+                      pj_inet_ntoa(accept_op->remote_addr.sin_addr),
+                      pj_ntohs(accept_op->remote_addr.sin_port),
+                      sock));
+
+            /* 
+             * Incoming connections!
+             * Create TCP transport for the new socket.
+             */
+            status = tcp_create( listener, sock, PJ_TRUE,
+                                 &accept_op->local_addr, 
+                                 &accept_op->remote_addr, &tcp);
+            if (status == PJ_SUCCESS) {
+                status = tcp_start_read(tcp);
+                if (status != PJ_SUCCESS) {
+                    PJ_LOG(3,(tcp->base.obj_name, "New transport cancelled"));
+                    tcp_destroy(&tcp->base, status);
+                }
             }
         }
 
-start_next_accept:
+        /*
+         * Start the next asynchronous accept() operation.
+         */
+        accept_op->addr_len = sizeof(pj_sockaddr_in);
+        accept_op->new_sock = PJ_INVALID_SOCKET;
 
         status = pj_ioqueue_accept(listener->key, 
@@ -687 +893 @@
                                    &accept_op->addr_len);
 
+        /*
+         * Loop while we have immediate connection or when there is error.
+         */
+
     } while (status != PJ_EPENDING);
 }
 
 
-/* Callback from ioqueue when packet is sent */
+/* 
+ * Callback from ioqueue when packet is sent.
+ */
 static void on_write_complete(pj_ioqueue_key_t *key, 
                               pj_ioqueue_op_key_t *op_key, 
                               pj_ssize_t bytes_sent)
 {
-    struct tcp_transport *tp = pj_ioqueue_get_user_data(key);
+    struct tcp_transport *tcp = pj_ioqueue_get_user_data(key);
     pjsip_tx_data_op_key *tdata_op_key = (pjsip_tx_data_op_key*)op_key;
 
     tdata_op_key->tdata = NULL;
 
+    /* Check for error/closure */
+    if (bytes_sent <= 0) {
+        pj_status_t status;
+
+        PJ_LOG(5,(tcp->base.obj_name, "TCP send() error, sent=%d", 
+                  bytes_sent));
+
+        status = (bytes_sent == 0) ? PJ_RETURN_OS_ERROR(OSERR_ENOTCONN) :
+                                     -bytes_sent;
+        if (tcp->close_reason==PJ_SUCCESS) tcp->close_reason = status;
+        pjsip_transport_shutdown(&tcp->base);
+    }
+
     if (tdata_op_key->callback) {
-        tdata_op_key->callback(&tp->base, tdata_op_key->token, bytes_sent);
-    }
-}
-
-
-/* This callback is called by transport manager to send SIP message */
+        /*
+         * Notify sip_transport.c that packet has been sent.
+         */
+        tdata_op_key->callback(&tcp->base, tdata_op_key->token, bytes_sent);
+    }
+}
+
+
+/* 
+ * This callback is called by transport manager to send SIP message 
+ */
 static pj_status_t tcp_send_msg(pjsip_transport *transport, 
                                 pjsip_tx_data *tdata,
@@ -719 +949 @@
     struct tcp_transport *tcp = (struct tcp_transport*)transport;
     pj_ssize_t size;
-    pj_status_t status;
+    pj_bool_t delayed = PJ_FALSE;
+    pj_status_t status = PJ_SUCCESS;
 
     /* Sanity check */
@@ -738 +969 @@
 
     /* If asynchronous connect() has not completed yet, just put the
-     * transmit data in the pending transmission list.
+     * transmit data in the pending transmission list since we can not
+     * use the socket yet.
      */
-    pj_lock_acquire(tcp->base.lock);
-
     if (tcp->has_pending_connect) {
-        struct pending_tdata *pending_tdata;
-
-        /* Pust to list */
-        pending_tdata = pj_pool_alloc(tdata->pool, sizeof(*pending_tdata));
-        pending_tdata->tdata_op_key = &tdata->op_key;
-
-        pj_list_push_back(&tcp->tx_list, pending_tdata);
-        status = PJ_EPENDING;
-
-    } else {
-        /* send to ioqueue! */
+
+        /*
+         * Looks like connect() is still in progress. Check again (this time
+         * with holding the lock) to be sure.
+         */
+        pj_lock_acquire(tcp->base.lock);
+
+        if (tcp->has_pending_connect) {
+            struct delayed_tdata *delayed_tdata;
+
+            /*
+             * connect() is still in progress. Put the transmit data to
+             * the delayed list.
+             */
+            delayed_tdata = pj_pool_alloc(tdata->pool, 
+                                          sizeof(*delayed_tdata));
+            delayed_tdata->tdata_op_key = &tdata->op_key;
+
+            pj_list_push_back(&tcp->delayed_list, delayed_tdata);
+            status = PJ_EPENDING;
+
+            /* Prevent pj_ioqueue_send() to be called below */
+            delayed = PJ_TRUE;
+        }
+
+        pj_lock_release(tcp->base.lock);
+    } 
+    
+    if (!delayed) {
+        /*
+         * Transport is ready to go. Send the packet to ioqueue to be
+         * sent asynchronously.
+         */
         size = tdata->buf.cur - tdata->buf.start;
         status = pj_ioqueue_send(tcp->key, 
@@ -759 +1011 @@
                                  tdata->buf.start, &size, 0);
 
-        if (status != PJ_EPENDING)
+        if (status != PJ_EPENDING) {
+            /* Not pending (could be immediate success or error) */
             tdata->op_key.tdata = NULL;
-    }
-
-    pj_lock_release(tcp->base.lock);
+
+            /* Shutdown transport on closure/errors */
+            if (size <= 0) {
+
+                PJ_LOG(5,(tcp->base.obj_name, "TCP send() error, sent=%d", 
+                          size));
+
+                if (status == PJ_SUCCESS) 
+                    status = PJ_RETURN_OS_ERROR(OSERR_ENOTCONN);
+                if (tcp->close_reason==PJ_SUCCESS) tcp->close_reason = status;
+                pjsip_transport_shutdown(&tcp->base);
+            }
+        }
+    }
 
     return status;
@@ -769 +1033 @@
 
 
-/* This callback is called by transport manager to shutdown transport */
+/* 
+ * This callback is called by transport manager to shutdown transport.
+ * This normally is only used by UDP transport.
+ */
 static pj_status_t tcp_shutdown(pjsip_transport *transport)
 {
@@ -780 +1047 @@
 
 
-/* Callback from ioqueue on incoming packet */
+/* 
+ * Callback from ioqueue that an incoming data is received from the socket.
+ */
 static void on_read_complete(pj_ioqueue_key_t *key, 
                              pj_ioqueue_op_key_t *op_key, 
@@ -788 +1057 @@
     pjsip_rx_data_op_key *rdata_op_key = (pjsip_rx_data_op_key*) op_key;
     pjsip_rx_data *rdata = rdata_op_key->rdata;
-    struct tcp_transport *tp = (struct tcp_transport*)rdata->tp_info.transport;
+    struct tcp_transport *tcp = 
+        (struct tcp_transport*)rdata->tp_info.transport;
     int i;
     pj_status_t status;
 
     /* Don't do anything if transport is closing. */
-    if (tp->is_closing) {
-        tp->is_closing++;
+    if (tcp->is_closing) {
+        tcp->is_closing++;
         return;
     }
@@ -807 +1077 @@
         pj_uint32_t flags;
 
-        /* Report the packet to transport manager. */
+        /* Houston, we have packet! Report the packet to transport manager
+         * to be parsed.
+         */
         if (bytes_read > 0) {
             pj_size_t size_eaten;
@@ -816 +1088 @@
             pj_gettimeofday(&rdata->pkt_info.timestamp);
 
+            /* Report to transport manager.
+             * The transport manager will tell us how many bytes of the packet
+             * have been processed (as valid SIP message).
+             */
             size_eaten = 
                 pjsip_tpmgr_receive_packet(rdata->tp_info.transport->tpmgr, 
@@ -834 +1110 @@
 
             /* Transport is closed */
-            PJ_LOG(4,(tp->base.obj_name, "tcp connection closed"));
-            tcp_destroy(&tp->base);
+            PJ_LOG(4,(tcp->base.obj_name, "TCP connection closed"));
+            
+            /* We can not destroy the transport since high level objects may
+             * still keep reference to this transport. So we can only 
+             * instruct transport manager to gracefully start the shutdown
+             * procedure for this transport.
+             */
+            if (tcp->close_reason==PJ_SUCCESS) 
+                tcp->close_reason = PJ_RETURN_OS_ERROR(OSERR_ENOTCONN);
+            pjsip_transport_shutdown(&tcp->base);
+
             return;
 
-        } else if (bytes_read < 0)  {
+        //} else if (bytes_read < 0)  {
+        } else if (-bytes_read != PJ_STATUS_FROM_OS(OSERR_EWOULDBLOCK) &&
+                   -bytes_read != PJ_STATUS_FROM_OS(OSERR_EINPROGRESS) && 
+                   -bytes_read != PJ_STATUS_FROM_OS(OSERR_ECONNRESET)) 
+        {
 
             /* Report error to endpoint. */
             PJSIP_ENDPT_LOG_ERROR((rdata->tp_info.transport->endpt,
                                    rdata->tp_info.transport->obj_name,
-                                   -bytes_read, "tcp recv() error"));
-
-            /* Transport error, close transport */
-            tcp_destroy(&tp->base);
+                                   -bytes_read, "TCP recv() error"));
+
+            /* We can not destroy the transport since high level objects may
+             * still keep reference to this transport. So we can only 
+             * instruct transport manager to gracefully start the shutdown
+             * procedure for this transport.
+             */
+            if (tcp->close_reason==PJ_SUCCESS) tcp->close_reason = -bytes_read;
+            pjsip_transport_shutdown(&tcp->base);
+
             return;
         }
 
         if (i >= MAX_IMMEDIATE_PACKET) {
-            /* Force ioqueue_recv() to return PJ_EPENDING */
+            /* Receive quota reached. Force ioqueue_recv() to 
+             * return PJ_EPENDING 
+             */
             flags = PJ_IOQUEUE_ALWAYS_ASYNC;
         } else {
@@ -868 +1165 @@
 
         if (status == PJ_SUCCESS) {
+
             /* Continue loop. */
             pj_assert(i < MAX_IMMEDIATE_PACKET);
@@ -880 +1178 @@
                                    status, "tcp recv() error"));
 
-            /* Transport error, close transport */
-            tcp_destroy(&tp->base);
+            /* We can not destroy the transport since high level objects may
+             * still keep reference to this transport. So we can only 
+             * instruct transport manager to gracefully start the shutdown
+             * procedure for this transport.
+             */
+            if (tcp->close_reason==PJ_SUCCESS) tcp->close_reason = status;
+            pjsip_transport_shutdown(&tcp->base);
+
             return;
         }
@@ -888 +1192 @@
 
 
-/* Callback from ioqueue when connect completes */
+/* 
+ * Callback from ioqueue when asynchronous connect() operation completes.
+ */
 static void on_connect_complete(pj_ioqueue_key_t *key, 
                                 pj_status_t status)
 {
-    struct pending_connect *connect_op = (struct pending_connect *)key;
-    struct tcp_transport *tcp = connect_op->transport;
+    struct tcp_transport *tcp;
     pj_sockaddr_in addr;
     int addrlen;
 
+    tcp = pj_ioqueue_get_user_data(key);
+
+    PJ_LOG(4,(tcp->base.obj_name, 
+              "TCP transport %.*s:%d is connected to %.*s:%d",
+              (int)tcp->base.local_name.host.slen,
+              tcp->base.local_name.host.ptr,
+              tcp->base.local_name.port,
+              (int)tcp->base.remote_name.host.slen,
+              tcp->base.remote_name.host.ptr,
+              tcp->base.remote_name.port));
+
     /* Mark that pending connect() operation has completed. */
     tcp->has_pending_connect = PJ_FALSE;
@@ -902 +1218 @@
     /* Check connect() status */
     if (status != PJ_SUCCESS) {
+
         tcp_perror(tcp->base.obj_name, "TCP connect() error", status);
-        tcp_destroy(&tcp->base);
+
+        /* We can not destroy the transport since high level objects may
+         * still keep reference to this transport. So we can only 
+         * instruct transport manager to gracefully start the shutdown
+         * procedure for this transport.
+         */
+        if (tcp->close_reason==PJ_SUCCESS) tcp->close_reason = status;
+        pjsip_transport_shutdown(&tcp->base);
         return;
     }
@@ -926 +1250 @@
     status = tcp_start_read(tcp);
     if (status != PJ_SUCCESS) {
-        tcp_destroy(&tcp->base);
+        /* We can not destroy the transport since high level objects may
+         * still keep reference to this transport. So we can only 
+         * instruct transport manager to gracefully start the shutdown
+         * procedure for this transport.
+         */
+        if (tcp->close_reason==PJ_SUCCESS) tcp->close_reason = status;
+        pjsip_transport_shutdown(&tcp->base);
         return;
     }