Changeset 25:67ca08d5bc48 in freeDiameter

Timestamp:

Oct 26, 2009, 4:00:49 PM (15 years ago)

Author:

Sebastien Decugis <sdecugis@nict.go.jp>

Branch:

default

Phase:

public

Message:

Completed connection context files

Files:

• extensions/dbg_monitor/monitor.c (modified) (2 diffs)
• freeDiameter/CMakeLists.txt (modified) (2 diffs)
• freeDiameter/cnxctx.c (modified) (13 diffs)
• freeDiameter/cnxctx.h (added)
• freeDiameter/fD.h (modified) (3 diffs)
• freeDiameter/main.c (modified) (2 diffs)
• freeDiameter/p_expiry.c (modified) (3 diffs)
• freeDiameter/p_psm.c (modified) (5 diffs)
• freeDiameter/sctp.c (modified) (4 diffs)
• freeDiameter/sctps.c (added)
• freeDiameter/server.c (modified) (4 diffs)
• freeDiameter/tcp.c (modified) (1 diff)
• include/freeDiameter/freeDiameter.h (modified) (2 diffs)
• include/freeDiameter/libfreeDiameter.h (modified) (3 diffs)
• libfreeDiameter/fifo.c (modified) (3 diffs)
• libfreeDiameter/lists.c (modified) (1 diff)

extensions/dbg_monitor/monitor.c

@@ -49 +49 @@
 static void got_sig(int signal)
 {
-        CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_DUMP_DICT, NULL), /* continue */);
-        CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_DUMP_CONFIG, NULL), /* continue */);
-        CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_DUMP_EXT, NULL), /* continue */);
+        CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_DUMP_DICT, 0, NULL), /* continue */);
+        CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_DUMP_CONFIG, 0, NULL), /* continue */);
+        CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_DUMP_EXT, 0, NULL), /* continue */);
 }
 /* Thread to display periodical debug information */
@@ -77 +77 @@
                 #endif /* DEBUG */
                 TRACE_DEBUG(NONE, "Monitor information");
-                CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_DUMP_QUEUES, NULL), /* continue */);
-                CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_DUMP_PEERS, NULL), /* continue */);
+                CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_DUMP_QUEUES, 0, NULL), /* continue */);
+                CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_DUMP_SERV, 0, NULL), /* continue */);
+                CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_DUMP_PEERS, 0, NULL), /* continue */);
                 pthread_testcancel();
         }

freeDiameter/CMakeLists.txt

@@ -10 +10 @@
 SET(FD_COMMON_SRC
         fD.h
+        cnxctx.h
         config.c
         cnxctx.c
@@ -26 +27 @@
 
 IF(NOT DISABLE_SCTP)
-        SET(FD_COMMON_SRC ${FD_COMMON_SRC} sctp.c)
+        SET(FD_COMMON_SRC ${FD_COMMON_SRC} sctp.c sctps.c)
 ENDIF(NOT DISABLE_SCTP)
 

freeDiameter/cnxctx.c

@@ -35 +35 @@
 
 #include "fD.h"
+#include "cnxctx.h"
+
+/* The maximum size of Diameter message we accept to receive (<= 2^24) to avoid too big mallocs in case of trashed headers */
+#ifndef DIAMETER_MSG_SIZE_MAX
+#define DIAMETER_MSG_SIZE_MAX   65535   /* in bytes */
+#endif /* DIAMETER_MSG_SIZE_MAX */
 
 /* Connections contexts (cnxctx) in freeDiameter are wrappers around the sockets and TLS operations .
@@ -57 +63 @@
  *
  * 3) Usage
- *    - fd_cnx_receive, fd_cnx_send : exchange messages on this connection (send is synchronous, receive is not).
+ *    - fd_cnx_receive, fd_cnx_send : exchange messages on this connection (send is synchronous, receive is not, but blocking).
  *    - fd_cnx_recv_setaltfifo : when a message is received, the event is sent to an external fifo list. fd_cnx_receive does not work when the alt_fifo is set.
  *    - fd_cnx_getid : retrieve a descriptive string for the connection (for debug)
@@ -68 +74 @@
  */
 
-/* The connection context structure */
-struct cnxctx {
-        char            cc_id[60];      /* The name of this connection */
-        char            cc_remid[60];   /* Id of remote peer */
-
-        int             cc_socket;      /* The socket object of the connection -- <=0 if no socket is created */
-
-        int             cc_proto;       /* IPPROTO_TCP or IPPROTO_SCTP */
-        int             cc_tls;         /* Is TLS already started ? */
-
-        struct fifo *   cc_events;      /* Events occuring on the connection */
-        pthread_t       cc_mgr;         /* manager thread for the connection */
-        struct fifo *   cc_incoming;    /* FIFO queue of messages received on the connection */
-        struct fifo *   cc_alt;         /* alternate fifo to send FDEVP_CNX_MSG_RECV events to. */
-
-        /* If cc_proto == SCTP */
-        struct  {
-                int             str_out;/* Out streams */
-                int             str_in; /* In streams */
-                int             pairs;  /* max number of pairs ( = min(in, out)) */
-                int             next;   /* # of stream the next message will be sent to */
-        }               cc_sctp_para;
-
-        /* If cc_tls == true */
-        struct {
-                int                              mode;          /* GNUTLS_CLIENT / GNUTLS_SERVER */
-                gnutls_session_t                 session;       /* Session object (stream #0 in case of SCTP) */
-        }               cc_tls_para;
-
-        /* If both conditions */
-        struct {
-                gnutls_session_t                *res_sessions;  /* Sessions of other pairs of streams, resumed from the first */
-                /* Buffers, threads, ... */
-        }               cc_sctp_tls_para;
-};
-
+
+/*******************************************/
+/*     Creation of a connection object     */
+/*******************************************/
 
 /* Initialize a context structure */
@@ -116 +90 @@
 
         if (full) {
-                CHECK_FCT_DO( fd_fifo_new ( &conn->cc_events ), return NULL );
                 CHECK_FCT_DO( fd_fifo_new ( &conn->cc_incoming ), return NULL );
         }
@@ -352 +325 @@
 }
 
-/* Same for SCTP, accepts a list of remote addresses to connect to (see sctp_connectx) */
+/* Same for SCTP, accepts a list of remote addresses to connect to (see sctp_connectx for how they are used) */
 struct cnxctx * fd_cnx_cli_connect_sctp(int no_ip6, uint16_t port, struct fd_list * list)
 {
@@ -382 +355 @@
                 cnx->cc_sctp_para.pairs = cnx->cc_sctp_para.str_in;
         
+        if (TRACE_BOOL(INFO)) {
+                fd_log_debug("Connection established to server '");
+                sSA_DUMP_NODE_SERV( &primary, NI_NUMERICSERV);
+                fd_log_debug("' (SCTP:%d, %d/%d streams).\n", sock, cnx->cc_sctp_para.str_in, cnx->cc_sctp_para.str_out);
+        }
+        
         /* Generate the names for the object */
         {
@@ -403 +382 @@
         }
         
-        if (TRACE_BOOL(INFO)) {
-                fd_log_debug("Connection established to server '");
-                sSA_DUMP_NODE_SERV( &primary, NI_NUMERICSERV);
-                fd_log_debug("' (SCTP:%d).\n", sock);
-        }
-        
         return cnx;
 
@@ -424 +397 @@
 }
 
+/* Get the list of endpoints (IP addresses) of the local and remote peers on this connection */
+int fd_cnx_getendpoints(struct cnxctx * conn, struct fd_list * local, struct fd_list * remote)
+{
+        TRACE_ENTRY("%p %p %p", conn, local, remote);
+        CHECK_PARAMS(conn);
+        
+        if (local) {
+                /* Retrieve the local endpoint(s) of the connection */
+                switch (conn->cc_proto) {
+                        case IPPROTO_TCP: {
+                                sSS ss;
+                                socklen_t sl;
+                                CHECK_FCT(fd_tcp_get_local_ep(conn->cc_socket, &ss, &sl));
+                                CHECK_FCT(fd_ep_add_merge( local, (sSA *)&ss, sl, EP_FL_LL | EP_FL_PRIMARY));
+                        }
+                        break;
+
+                        #ifndef DISABLE_SCTP
+                        case IPPROTO_SCTP: {
+                                CHECK_FCT(fd_sctp_get_local_ep(conn->cc_socket, local));
+                        }
+                        break;
+                        #endif /* DISABLE_SCTP */
+
+                        default:
+                                CHECK_PARAMS(0);
+                }
+        }
+        
+        if (remote) {
+                /* Check we have a full connection object, not a listening socket (with no remote) */
+                CHECK_PARAMS( conn->cc_incoming );
+                
+                /* Retrieve the peer endpoint(s) of the connection */
+                switch (conn->cc_proto) {
+                        case IPPROTO_TCP: {
+                                sSS ss;
+                                socklen_t sl;
+                                CHECK_FCT(fd_tcp_get_remote_ep(conn->cc_socket, &ss, &sl));
+                                CHECK_FCT(fd_ep_add_merge( remote, (sSA *)&ss, sl, EP_FL_LL | EP_FL_PRIMARY ));
+                        }
+                        break;
+
+                        #ifndef DISABLE_SCTP
+                        case IPPROTO_SCTP: {
+                                CHECK_FCT(fd_sctp_get_remote_ep(conn->cc_socket, remote));
+                        }
+                        break;
+                        #endif /* DISABLE_SCTP */
+
+                        default:
+                                CHECK_PARAMS(0);
+                }
+        }
+
+        return 0;
+}
+
+
+/* Get a string describing the remote peer address (ip address or fqdn) */
+char * fd_cnx_getremoteid(struct cnxctx * conn)
+{
+        CHECK_PARAMS_DO( conn, return "" );
+        return conn->cc_remid;
+}
+
+
+/**************************************/
+/*     Use of a connection object     */
+/**************************************/
+
+/* Receiver thread (TCP & noTLS) : incoming message is directly saved into the target queue */
+static void * rcvthr_notls_tcp(void * arg)
+{
+        struct cnxctx * conn = arg;
+        
+        TRACE_ENTRY("%p", arg);
+        
+        CHECK_PARAMS_DO(conn && (conn->cc_socket > 0), goto out);
+        ASSERT( conn->cc_proto == IPPROTO_TCP );
+        ASSERT( conn->cc_tls == 0 );
+        ASSERT( Target_Queue(conn) );
+        
+        /* Receive from a TCP connection: we have to rebuild the message boundaries */
+        do {
+                uint8_t header[4];
+                uint8_t * newmsg;
+                size_t  length;
+                ssize_t ret = 0;
+                size_t  received = 0;
+
+                do {
+                        ret = recv(conn->cc_socket, &header[received], sizeof(header) - received, 0);
+                        if (ret <= 0) {
+                                CHECK_SYS_DO(ret, /* continue */);
+                                goto error; /* Stop the thread, the recipient of the event will cleanup */
+                        }
+
+                        received += ret;
+                } while (received < sizeof(header));
+
+                length = ((size_t)header[1] << 16) + ((size_t)header[2] << 8) + (size_t)header[3];
+
+                /* Check the received word is a valid begining of a Diameter message */
+                if ((header[0] != DIAMETER_VERSION)     /* defined in <libfreeDiameter.h> */
+                   || (length > DIAMETER_MSG_SIZE_MAX)) { /* to avoid too big mallocs */
+                        /* The message is suspect */
+                        TRACE_DEBUG(INFO, "Received suspect header [ver: %d, size: %g], assume disconnection", (int)header[0], length);
+                        goto error; /* Stop the thread, the recipient of the event will cleanup */
+                }
+
+                /* Ok, now we can really receive the data */
+                CHECK_MALLOC_DO(  newmsg = malloc( length ), goto error );
+                memcpy(newmsg, header, sizeof(header));
+
+                while (received < length) {
+                        pthread_cleanup_push(free, newmsg); /* In case we are canceled, clean the partialy built buffer */
+                        ret = recv(conn->cc_socket, newmsg + received, length - received, 0);
+                        pthread_cleanup_pop(0);
+
+                        if (ret <= 0) {
+                                CHECK_SYS_DO(ret, /* continue */);
+                                free(newmsg);
+                                goto error; /* Stop the thread, the recipient of the event will cleanup */
+                        }
+                        received += ret;
+                }
+                
+                /* We have received a complete message, send it */
+                CHECK_FCT_DO( fd_event_send( Target_Queue(conn), FDEVP_CNX_MSG_RECV, length, newmsg), /* continue or destroy everything? */);
+                
+        } while (conn->cc_loop);
+        
+out:
+        TRACE_DEBUG(FULL, "Thread terminated"); 
+        return NULL;
+error:
+        CHECK_FCT_DO( fd_event_send( Target_Queue(conn), FDEVP_CNX_ERROR, 0, NULL), /* continue or destroy everything? */);
+        goto out;
+}
+
+#ifndef DISABLE_SCTP
+/* Receiver thread (SCTP & noTLS) : incoming message is directly saved into cc_incoming, no need to care for the stream ID */
+static void * rcvthr_notls_sctp(void * arg)
+{
+        struct cnxctx * conn = arg;
+        uint8_t * buf;
+        size_t    bufsz;
+        int       event;
+        
+        TRACE_ENTRY("%p", arg);
+        
+        CHECK_PARAMS_DO(conn && (conn->cc_socket > 0), goto out);
+        ASSERT( conn->cc_proto == IPPROTO_SCTP );
+        ASSERT( conn->cc_tls == 0 );
+        ASSERT( Target_Queue(conn) );
+        
+        do {
+                CHECK_FCT_DO( fd_sctp_recvmeta(conn->cc_socket, NULL, &buf, &bufsz, &event), goto error );
+                if (event == FDEVP_CNX_ERROR) {
+                        goto error;
+                }
+                
+                CHECK_FCT_DO( fd_event_send( Target_Queue(conn), event, bufsz, buf), goto error );
+                
+        } while (conn->cc_loop);
+        
+out:
+        TRACE_DEBUG(FULL, "Thread terminated"); 
+        return NULL;
+error:
+        CHECK_FCT_DO( fd_event_send( Target_Queue(conn), FDEVP_CNX_ERROR, 0, NULL), /* continue or destroy everything? */);
+        goto out;
+}
+#endif /* DISABLE_SCTP */
+
+/* Returns 0 on error, received data size otherwise (always >= 0) */
+static ssize_t fd_tls_recv_handle_error(struct cnxctx * conn, gnutls_session_t session, void * data, size_t sz)
+{
+        ssize_t ret;
+again:  
+        CHECK_GNUTLS_DO( ret = gnutls_record_recv(session, data, sz),
+                {
+                        switch (ret) {
+                                case GNUTLS_E_REHANDSHAKE: 
+                                        CHECK_GNUTLS_DO( ret = gnutls_handshake(session),
+                                                {
+                                                        if (TRACE_BOOL(INFO)) {
+                                                                fd_log_debug("TLS re-handshake failed on socket %d (%s) : %s\n", conn->cc_socket, conn->cc_id, gnutls_strerror(ret));
+                                                        }
+                                                        ret = 0;
+                                                        goto end;
+                                                } );
+
+                                case GNUTLS_E_AGAIN:
+                                case GNUTLS_E_INTERRUPTED:
+                                        goto again;
+
+                                default:
+                                        TRACE_DEBUG(INFO, "This TLS error is not handled, assume unrecoverable error");
+                                        ret = 0;
+                        }
+                } );
+end:    
+        return ret;
+}
+
+/* The function that receives TLS data and re-builds a Diameter message -- it exits only on error or cancelation */
+int fd_tls_rcvthr_core(struct cnxctx * conn, gnutls_session_t session)
+{
+        /* No guaranty that GnuTLS preserves the message boundaries, so we re-build it as in TCP */
+        do {
+                uint8_t header[4];
+                uint8_t * newmsg;
+                size_t  length;
+                ssize_t ret = 0;
+                size_t  received = 0;
+
+                do {
+                        ret = fd_tls_recv_handle_error(conn, conn->cc_tls_para.session, &header[received], sizeof(header) - received);
+                        if (ret == 0) {
+                                /* The connection is closed */
+                                goto out;
+                        }
+                        received += ret;
+                } while (received < sizeof(header));
+
+                length = ((size_t)header[1] << 16) + ((size_t)header[2] << 8) + (size_t)header[3];
+
+                /* Check the received word is a valid beginning of a Diameter message */
+                if ((header[0] != DIAMETER_VERSION)     /* defined in <libfreeDiameter.h> */
+                   || (length > DIAMETER_MSG_SIZE_MAX)) { /* to avoid too big mallocs */
+                        /* The message is suspect */
+                        TRACE_DEBUG(INFO, "Received suspect header [ver: %d, size: %g], assume disconnection", (int)header[0], length);
+                        goto out;
+                }
+
+                /* Ok, now we can really receive the data */
+                CHECK_MALLOC(  newmsg = malloc( length ) );
+                memcpy(newmsg, header, sizeof(header));
+
+                while (received < length) {
+                        pthread_cleanup_push(free, newmsg); /* In case we are canceled, clean the partialy built buffer */
+                        ret = fd_tls_recv_handle_error(conn, conn->cc_tls_para.session, newmsg + received, length - received);
+                        pthread_cleanup_pop(0);
+
+                        if (ret == 0) {
+                                free(newmsg);
+                                goto out; /* Stop the thread, the recipient of the event will cleanup */
+                        }
+                        received += ret;
+                }
+                
+                /* We have received a complete message, send it */
+                CHECK_FCT_DO( fd_event_send( Target_Queue(conn), FDEVP_CNX_MSG_RECV, length, newmsg), /* continue or destroy everything? */);
+                
+        } while (1);
+out:
+        return ENOTCONN;
+}
+
+/* Receiver thread (TLS & 1 stream SCTP or TCP) : gnutls directly handles the socket, save records into the target queue */
+static void * rcvthr_tls_single(void * arg)
+{
+        struct cnxctx * conn = arg;
+        
+        TRACE_ENTRY("%p", arg);
+        
+        CHECK_PARAMS_DO(conn && (conn->cc_socket > 0), goto error);
+        ASSERT( conn->cc_tls == 1 );
+        ASSERT( Target_Queue(conn) );
+        
+        CHECK_FCT_DO(fd_tls_rcvthr_core(conn, conn->cc_tls_para.session), /* continue */);
+error:
+        CHECK_FCT_DO( fd_event_send( Target_Queue(conn), FDEVP_CNX_ERROR, 0, NULL), /* continue or destroy everything? */);
+        TRACE_DEBUG(FULL, "Thread terminated"); 
+        return NULL;
+}
+
 /* Start receving messages in clear (no TLS) on the connection */
-int fd_cnx_start_clear(struct cnxctx * conn)
-{
-
-        TODO("...");
-        return ENOTSUP;
+int fd_cnx_start_clear(struct cnxctx * conn, int loop)
+{
+        TRACE_ENTRY("%p %i", conn, loop);
+        
+        CHECK_PARAMS( conn && Target_Queue(conn) && (!conn->cc_tls) && (!conn->cc_loop));
+        
+        /* Save the loop request */
+        conn->cc_loop = loop;
+        
+        /* Release resources in case of a previous call was already made */
+        CHECK_FCT_DO( fd_thr_term(&conn->cc_rcvthr), /* continue */);
+        
+        switch (conn->cc_proto) {
+                case IPPROTO_TCP:
+                        /* Start the tcp_notls thread */
+                        CHECK_POSIX( pthread_create( &conn->cc_rcvthr, NULL, rcvthr_notls_tcp, conn ) );
+                        break;
+#ifndef DISABLE_SCTP
+                case IPPROTO_SCTP:
+                        /* Start the tcp_notls thread */
+                        CHECK_POSIX( pthread_create( &conn->cc_rcvthr, NULL, rcvthr_notls_sctp, conn ) );
+                        break;
+#endif /* DISABLE_SCTP */
+                default:
+                        TRACE_DEBUG(INFO, "Unknown protocol: %d", conn->cc_proto);
+                        return ENOTSUP;
+        }
+                        
+        return 0;
+}
+
+/* Prepare a gnutls session object for handshake */
+int fd_tls_prepare(gnutls_session_t * session, int mode, char * priority)
+{
+        /* Create the master session context */
+        CHECK_GNUTLS_DO( gnutls_init (session, mode), return ENOMEM );
+
+        /* Set the algorithm suite */
+        if (priority) {
+                const char * errorpos;
+                CHECK_GNUTLS_DO( gnutls_priority_set_direct( *session, priority, &errorpos ), 
+                        { TRACE_DEBUG(INFO, "Error in priority string '%s' at position: '%s'\n", priority, errorpos); return EINVAL; } );
+        } else {
+                CHECK_GNUTLS_DO( gnutls_priority_set( *session, fd_g_config->cnf_sec_data.prio_cache ), return EINVAL );
+        }
+
+        /* Set the credentials of this side of the connection */
+        CHECK_GNUTLS_DO( gnutls_credentials_set (*session, GNUTLS_CRD_CERTIFICATE, fd_g_config->cnf_sec_data.credentials), return EINVAL );
+
+        /* Request the remote credentials as well */
+        if (mode == GNUTLS_SERVER) {
+                gnutls_certificate_server_set_request (*session, GNUTLS_CERT_REQUIRE);
+        }
+        
+        return 0;
 }
 
@@ -436 +738 @@
 {
         TRACE_ENTRY( "%p %d", conn, mode);
-        CHECK_PARAMS( conn && ( (mode == GNUTLS_CLIENT) || (mode == GNUTLS_SERVER) ) );
+        CHECK_PARAMS( conn && (!conn->cc_tls) && ( (mode == GNUTLS_CLIENT) || (mode == GNUTLS_SERVER) ) && (!conn->cc_loop) );
 
         /* Save the mode */
         conn->cc_tls_para.mode = mode;
-
-        /* Create the master session context */
-        CHECK_GNUTLS_DO( gnutls_init (&conn->cc_tls_para.session, mode), return ENOMEM );
-
-        /* Set the algorithm suite */
-        TODO("Use overwrite priority if non NULL");
-        CHECK_GNUTLS_DO( gnutls_priority_set( conn->cc_tls_para.session, fd_g_config->cnf_sec_data.prio_cache ), return EINVAL );
-
-        /* Set the credentials of this side of the connection */
-        CHECK_GNUTLS_DO( gnutls_credentials_set (conn->cc_tls_para.session, GNUTLS_CRD_CERTIFICATE, fd_g_config->cnf_sec_data.credentials), return EINVAL );
-
-        /* Request the remote credentials as well */
-        if (mode == GNUTLS_SERVER) {
-                gnutls_certificate_server_set_request (conn->cc_tls_para.session, GNUTLS_CERT_REQUIRE);
-        }
-
-        /* Set the socket info in the session */
-        gnutls_transport_set_ptr (conn->cc_tls_para.session, (gnutls_transport_ptr_t) conn->cc_socket);
+        
+        /* Cancel receiving thread if any -- it should already be terminated anyway, we just release the resources */
+        CHECK_FCT_DO( fd_thr_term(&conn->cc_rcvthr), /* continue */);
+        
+        /* Once TLS handshake is done, we don't stop after the first message */
+        conn->cc_loop = 1;
+        
+        /* Prepare the master session credentials and priority */
+        CHECK_FCT( fd_tls_prepare(&conn->cc_tls_para.session, mode, priority) );
 
         /* Special case: multi-stream TLS is not natively managed in GNU TLS, we use a wrapper library */
-        if ((conn->cc_proto == IPPROTO_SCTP) && (conn->cc_sctp_para.pairs > 0)) {
-#ifndef DISABLE_SCTP
-                TODO("Initialize the SCTP TLS wrapper");
-                TODO("Set the lowat, push and pull functions");
+        if (conn->cc_sctp_para.pairs > 1) {
+#ifdef DISABLE_SCTP
+                ASSERT(0);
+                CHECK_FCT( ENOTSUP );
 #else /* DISABLE_SCTP */
-                ASSERT(0);
+                /* Initialize the wrapper, start the demux thread */
+                CHECK_FCT( fd_sctps_init(conn) );
 #endif /* DISABLE_SCTP */
+        } else {
+                /* Set the socket info in the session */
+                gnutls_transport_set_ptr (conn->cc_tls_para.session, (gnutls_transport_ptr_t) conn->cc_socket);
         }
 
@@ -500 +797 @@
         }
 
-        /* Other sessions in case of multi-stream SCTP are resumed from the master */
-        if ((conn->cc_proto == IPPROTO_SCTP) && (conn->cc_sctp_para.pairs > 0)) {
+        /* Multi-stream TLS: handshake other streams as well */
+        if (conn->cc_sctp_para.pairs > 1) {
 #ifndef DISABLE_SCTP
-                TODO("Init and resume all additional sessions from the master one.");
+                /* Resume all additional sessions from the master one. */
+                CHECK_FCT(fd_sctps_handshake_others(conn, priority));
+                
+                /* Start decrypting the messages from all threads and queuing them in target queue */
+                CHECK_FCT(fd_sctps_startthreads(conn));
 #endif /* DISABLE_SCTP */
-        }
-
-        TODO("Start the connection state machine thread");
+        } else {
+                /* Start decrypting the data */
+                CHECK_POSIX( pthread_create( &conn->cc_rcvthr, NULL, rcvthr_tls_single, conn ) );
+        }
 
         return 0;
@@ -515 +817 @@
 int fd_cnx_getcred(struct cnxctx * conn, const gnutls_datum_t **cert_list, unsigned int *cert_list_size)
 {
-
-        TODO("...");
-        return ENOTSUP;
-}
-
-/* Get the list of endpoints (IP addresses) of the local and remote peers on this connection */
-int fd_cnx_getendpoints(struct cnxctx * conn, struct fd_list * local, struct fd_list * remote)
-{
-        TRACE_ENTRY("%p %p %p", conn, local, remote);
-        CHECK_PARAMS(conn);
-        
-        if (local) {
-                /* Retrieve the local endpoint(s) of the connection */
-                switch (conn->cc_proto) {
-                        case IPPROTO_TCP: {
-                                sSS ss;
-                                socklen_t sl;
-                                CHECK_FCT(fd_tcp_get_local_ep(conn->cc_socket, &ss, &sl));
-                                CHECK_FCT(fd_ep_add_merge( local, (sSA *)&ss, sl, EP_FL_LL | EP_FL_PRIMARY));
-                        }
+        TRACE_ENTRY("%p %p %p", conn, cert_list, cert_list_size);
+        CHECK_PARAMS( conn && (conn->cc_tls) && cert_list && cert_list_size );
+        
+        /* This function only works for X.509 certificates. */
+        CHECK_PARAMS( gnutls_certificate_type_get (conn->cc_tls_para.session) == GNUTLS_CRT_X509 );
+        
+        *cert_list = gnutls_certificate_get_peers (conn->cc_tls_para.session, cert_list_size);
+        if (*cert_list == NULL) {
+                TRACE_DEBUG(INFO, "No certificate was provided by remote peer / an error occurred.");
+                return EINVAL;
+        }
+
+        TRACE_DEBUG( FULL, "Remote peer provided %d certificates.\n", *cert_list_size);
+        
+        return 0;
+}
+
+/* Receive next message. if timeout is not NULL, wait only until timeout. This function only pulls from a queue, mgr thread is filling that queue aynchrounously. */
+int fd_cnx_receive(struct cnxctx * conn, struct timespec * timeout, unsigned char **buf, size_t * len)
+{
+        int    ev;
+        size_t ev_sz;
+        void * ev_data;
+        
+        TRACE_ENTRY("%p %p %p %p", conn, timeout, buf, len);
+        CHECK_PARAMS(conn && (conn->cc_socket > 0) && buf && len);
+        CHECK_PARAMS(conn->cc_rcvthr != (pthread_t)NULL);
+        CHECK_PARAMS(conn->cc_alt == NULL);
+
+        /* Now, pull the first event */
+get_next:
+        if (timeout) {
+                CHECK_FCT( fd_event_timedget(conn->cc_incoming, timeout, FDEVP_PSM_TIMEOUT, &ev, &ev_sz, &ev_data) );
+        } else {
+                CHECK_FCT( fd_event_get(conn->cc_incoming, &ev, &ev_sz, &ev_data) );
+        }
+        
+        switch (ev) {
+                case FDEVP_CNX_MSG_RECV:
+                        /* We got one */
+                        *len = ev_sz;
+                        *buf = ev_data;
+                        return 0;
+                        
+                case FDEVP_PSM_TIMEOUT:
+                        TRACE_DEBUG(FULL, "Timeout event received");
+                        return ETIMEDOUT;
+                        
+                case FDEVP_CNX_EP_CHANGE:
+                        /* We ignore this event */
+                        goto get_next;
+                        
+                case FDEVP_CNX_ERROR:
+                        TRACE_DEBUG(FULL, "Received ERROR event on the connection");
+                        return ENOTCONN;
+        }
+        
+        TRACE_DEBUG(INFO, "Received unexpected event %d (%s)", ev, fd_pev_str(ev));
+        return EINVAL;
+}
+
+/* Set an alternate FIFO list to send FDEVP_CNX_* events to */
+int fd_cnx_recv_setaltfifo(struct cnxctx * conn, struct fifo * alt_fifo)
+{
+        TRACE_ENTRY( "%p %p", conn, alt_fifo );
+        CHECK_PARAMS( conn && alt_fifo && conn->cc_incoming );
+        
+        /* The magic function does it all */
+        CHECK_FCT( fd_fifo_move( &conn->cc_incoming, alt_fifo, &conn->cc_alt ) );
+        
+        return 0;
+}
+
+/* Send function when no multi-stream is involved, or sending on stream #0 (send() always use stream 0)*/
+static int send_simple(struct cnxctx * conn, unsigned char * buf, size_t len)
+{
+        ssize_t ret;
+        size_t sent = 0;
+        TRACE_ENTRY("%p %p %g", conn, buf, len);
+        do {
+                if (conn->cc_tls) {
+                        CHECK_GNUTLS_DO( ret = gnutls_record_send (conn->cc_tls_para.session, buf + sent, len - sent), return ENOTCONN );
+                } else {
+                        CHECK_SYS( ret = send(conn->cc_socket, buf + sent, len - sent, 0) ); /* better to replace with sendmsg for atomic sending? */
+                }
+                sent += ret;
+        } while ( sent < len );
+        return 0;
+}
+
+/* Send a message -- this is synchronous -- and we assume it's never called by several threads at the same time, so we don't protect. */
+int fd_cnx_send(struct cnxctx * conn, unsigned char * buf, size_t len)
+{
+        TRACE_ENTRY("%p %p %g", conn, buf, len);
+        
+        CHECK_PARAMS(conn && (conn->cc_socket > 0) && buf && len);
+
+        TRACE_DEBUG(FULL, "Sending %gb %sdata on connection %s", len, conn->cc_tls ? "TLS-protected ":"", conn->cc_id);
+        
+        switch (conn->cc_proto) {
+                case IPPROTO_TCP:
+                        CHECK_FCT( send_simple(conn, buf, len) );
                         break;
-
-                        #ifndef DISABLE_SCTP
-                        case IPPROTO_SCTP: {
-                                CHECK_FCT(fd_sctp_get_local_ep(conn->cc_socket, local));
-                        }
-                        break;
-                        #endif /* DISABLE_SCTP */
-
-                        default:
-                                CHECK_PARAMS(0);
-                }
-        }
-        
-        if (remote) {
-                /* Check we have a full connection object, not a listening socket (with no remote) */
-                CHECK_PARAMS( conn->cc_events );
-                
-                /* Retrieve the peer endpoint(s) of the connection */
-                switch (conn->cc_proto) {
-                        case IPPROTO_TCP: {
-                                sSS ss;
-                                socklen_t sl;
-                                CHECK_FCT(fd_tcp_get_remote_ep(conn->cc_socket, &ss, &sl));
-                                CHECK_FCT(fd_ep_add_merge( remote, (sSA *)&ss, sl, EP_FL_LL | EP_FL_PRIMARY ));
-                        }
-                        break;
-
-                        #ifndef DISABLE_SCTP
-                        case IPPROTO_SCTP: {
-                                CHECK_FCT(fd_sctp_get_remote_ep(conn->cc_socket, remote));
-                        }
-                        break;
-                        #endif /* DISABLE_SCTP */
-
-                        default:
-                                CHECK_PARAMS(0);
-                }
-        }
-
+                
+#ifndef DISABLE_SCTP
+                case IPPROTO_SCTP: {
+                        int multistr = 0;
+                        
+                        if ((conn->cc_sctp_para.str_out > 1) && ((! conn->cc_tls) || (conn->cc_sctp_para.pairs > 1)))  {
+                                /* Update the id of the stream we will send this message on */
+                                conn->cc_sctp_para.next += 1;
+                                conn->cc_sctp_para.next %= (conn->cc_tls ? conn->cc_sctp_para.pairs : conn->cc_sctp_para.str_out);
+                                multistr = 1;
+                        }
+                        
+                        if ((!multistr) || (conn->cc_sctp_para.next == 0)) {
+                                CHECK_FCT( send_simple(conn, buf, len) );
+                        } else {
+                                if (!conn->cc_tls) {
+                                        CHECK_FCT( fd_sctp_sendstr(conn->cc_socket, conn->cc_sctp_para.next, buf, len) );
+                                } else {
+                                        /* push the record to the appropriate session */
+                                        ssize_t ret;
+                                        size_t sent = 0;
+                                        ASSERT(conn->cc_sctps_data.array != NULL);
+                                        do {
+                                                CHECK_GNUTLS_DO( ret = gnutls_record_send (conn->cc_sctps_data.array[conn->cc_sctp_para.next - 1].session, buf + sent, len - sent), { TODO("Handle error (re-handshake, etc.."); return ENOTCONN; } );
+                                                sent += ret;
+                                        } while ( sent < len );
+                                }
+                        }
+                }
+                break;
+#endif /* DISABLE_SCTP */
+        
+                default:
+                        TRACE_DEBUG(INFO, "Unknwon protocol: %d", conn->cc_proto);
+                        return ENOTSUP; /* or EINVAL... */
+        }
+        
         return 0;
 }
 
 
-/* Get a string describing the remote peer address (ip address or fqdn) */
-char * fd_cnx_getremoteid(struct cnxctx * conn)
-{
-        CHECK_PARAMS_DO( conn, return "" );
-        return conn->cc_remid;
-}
-
-
-/* Receive next message. if timeout is not NULL, wait only until timeout */
-int fd_cnx_receive(struct cnxctx * conn, struct timespec * timeout, unsigned char **buf, size_t * len)
-{
-
-        TODO("...");
-        return ENOTSUP;
-}
-
-/* Set / reset alternate FIFO list to send FDEVP_CNX_MSG_RECV to when message is received */
-int fd_cnx_recv_setaltfifo(struct cnxctx * conn, struct fifo * alt_fifo)
-{
-        TRACE_ENTRY( "%p %p", conn, alt_fifo );
-        CHECK_PARAMS( conn );
-        
-        /* Let's cross fingers that there is no race condition here... */
-        conn->cc_alt = alt_fifo;
-        
-        return 0;
-}
-
-/* Send a message */
-int fd_cnx_send(struct cnxctx * conn, unsigned char * buf, size_t len)
-{
-
-        TODO("...");
-        return ENOTSUP;
-}
-
+/**************************************/
+/*     Destruction of connection      */
+/**************************************/
 
 /* Destroy a conn structure, and shutdown the socket */
@@ -623 +972 @@
         CHECK_PARAMS_DO(conn, return);
 
-        TODO("End TLS session(s) if started");
-        
-        TODO("Stop manager thread if running");
+        /* In case of TLS, stop receiver thread, then close properly the gnutls session */
+        if ((conn->cc_tls) && (conn->cc_sctp_para.pairs > 1)) {
+#ifndef DISABLE_SCTP
+                /* Multi-stream TLS: Stop all decipher threads, but not the demux thread */
+                fd_sctps_stopthreads(conn);
+#endif /* DISABLE_SCTP */
+        } else {
+                /* Stop the decoding thread */
+                CHECK_FCT_DO( fd_thr_term(&conn->cc_rcvthr), /* continue */ );
+        }
+        
+        /* Terminate properly the TLS session(s) */
+        if (conn->cc_tls) {
+                /* Master session */
+                CHECK_GNUTLS_DO( gnutls_bye(conn->cc_tls_para.session, GNUTLS_SHUT_RDWR), /* Continue */ );
+                gnutls_deinit(conn->cc_tls_para.session);
+                
+#ifndef DISABLE_SCTP
+                if (conn->cc_sctp_para.pairs > 1) {
+                        /* Multi-stream TLS: destroy the wrapper and stop the demux thread */
+                        fd_sctps_destroy(conn);
+                }
+#endif /* DISABLE_SCTP */
+                
+        }
         
         /* Shut the connection down */
@@ -632 +1003 @@
         }
         
-        TODO("Empty FIFO queues");
-        
-        /* Destroy FIFO lists */
-        if (conn->cc_events)
-                CHECK_FCT_DO( fd_fifo_del ( &conn->cc_events ), /* continue */ );
-        if (conn->cc_incoming)
-                CHECK_FCT_DO( fd_fifo_del ( &conn->cc_incoming ), /* continue */ );
+        /* Empty and destroy FIFO list */
+        if (conn->cc_incoming) {
+                fd_event_destroy( &conn->cc_incoming, free );
+        }
         
         /* Free the object */

freeDiameter/fD.h

@@ -152 +152 @@
 enum {
         /* Dump all info about this peer in the debug log */
-         FDEVP_DUMP_ALL = 2000
+         FDEVP_DUMP_ALL = 1500
         
         /* request to terminate this peer : disconnect, requeue all messages */
         ,FDEVP_TERMINATE
         
-        /* A connection object has received a message -- stored in event->data */
+        /* A connection object has received a message. */
         ,FDEVP_CNX_MSG_RECV
+                         
+        /* A connection object has encountered an error (disconnected). */
+        ,FDEVP_CNX_ERROR
+        
+        /* Endpoints of a connection have been changed (multihomed SCTP). */
+        ,FDEVP_CNX_EP_CHANGE
         
         /* A message was received in the peer */
@@ -210 +216 @@
 struct cnxctx * fd_cnx_cli_connect_sctp(int no_ip6, uint16_t port, struct fd_list * list);
 char * fd_cnx_getid(struct cnxctx * conn);
-int fd_cnx_start_clear(struct cnxctx * conn);
+int fd_cnx_start_clear(struct cnxctx * conn, int loop);
 int fd_cnx_handshake(struct cnxctx * conn, int mode, char * priority);
 int fd_cnx_getcred(struct cnxctx * conn, const gnutls_datum_t **cert_list, unsigned int *cert_list_size);
@@ -220 +226 @@
 void fd_cnx_destroy(struct cnxctx * conn);
 
-/* TCP */
-int fd_tcp_create_bind_server( int * sock, sSA * sa, socklen_t salen );
-int fd_tcp_listen( int sock );
-int fd_tcp_client( int *sock, sSA * sa, socklen_t salen );
-int fd_tcp_get_local_ep(int sock, sSS * ss, socklen_t *sl);
-int fd_tcp_get_remote_ep(int sock, sSS * ss, socklen_t *sl);
-
-/* SCTP */
-#ifndef DISABLE_SCTP
-int fd_sctp_create_bind_server( int * sock, struct fd_list * list, uint16_t port );
-int fd_sctp_listen( int sock );
-int fd_sctp_client( int *sock, int no_ip6, uint16_t port, struct fd_list * list );
-int fd_sctp_get_local_ep(int sock, struct fd_list * list);
-int fd_sctp_get_remote_ep(int sock, struct fd_list * list);
-int fd_sctp_get_str_info( int sock, int *in, int *out, sSS *primary );
-
-#endif /* DISABLE_SCTP */
-
-
 
 #endif /* _FD_H */

freeDiameter/main.c

@@ -117 +117 @@
         TRACE_DEBUG(INFO, FD_PROJECT_BINARY " daemon initialized.");
         while (1) {
-                int code;
-                CHECK_FCT_DO(  fd_event_get(fd_g_config->cnf_main_ev, &code, NULL),  break  );
+                int code; size_t sz; void * data;
+                CHECK_FCT_DO(  fd_event_get(fd_g_config->cnf_main_ev, &code, &sz, &data),  break  );
                 switch (code) {
                         case FDEV_DUMP_DICT:
@@ -339 +339 @@
         
         TRACE_DEBUG(INFO, "Received signal %s (%d), exiting", SIGNALSTR(sig), sig);
-        CHECK_FCT_DO( fd_event_send(fd_g_config->cnf_main_ev, FDEV_TERMINATE, NULL), exit(2) );
+        CHECK_FCT_DO( fd_event_send(fd_g_config->cnf_main_ev, FDEV_TERMINATE, 0, NULL), exit(2) );
         return NULL;
 }

freeDiameter/p_expiry.c

@@ -87 +87 @@
         TRACE_DEBUG(INFO, "An error occurred in peers module! GC thread is terminating...");
         ASSERT(0);
-        CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_TERMINATE, NULL), );
+        CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_TERMINATE, 0, NULL), );
         return NULL;
 }
@@ -132 +132 @@
                 /* Now, the first peer in the list is expired; signal it */
                 fd_list_unlink( &first->p_expiry );
-                CHECK_FCT_DO( fd_event_send(first->p_events, FDEVP_TERMINATE, NULL), goto error );
+                CHECK_FCT_DO( fd_event_send(first->p_events, FDEVP_TERMINATE, 0, NULL), goto error );
                 
         } while (1);
@@ -140 +140 @@
         TRACE_DEBUG(INFO, "An error occurred in peers module! Expiry thread is terminating...");
         ASSERT(0);
-        CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_TERMINATE, NULL), );
+        CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_TERMINATE, 0, NULL), );
         return NULL;
 }

freeDiameter/p_psm.c

@@ -58 +58 @@
                 case_str(FDEVP_TERMINATE);
                 case_str(FDEVP_CNX_MSG_RECV);
+                case_str(FDEVP_CNX_ERROR);
+                case_str(FDEVP_CNX_EP_CHANGE);
                 case_str(FDEVP_MSG_INCOMING);
                 case_str(FDEVP_PSM_TIMEOUT);
@@ -126 +128 @@
 }
 
-/* Wait for the next event in the PSM, or timeout */
-static int psm_ev_timedget(struct fd_peer * peer, int *code, void ** data)
-{
-        struct fd_event * ev;
-        int ret = 0;
-        
-        TRACE_ENTRY("%p %p %p", peer, code, data);
-        
-        ret = fd_fifo_timedget(peer->p_events, &ev, &peer->p_psm_timer);
-        if (ret == ETIMEDOUT) {
-                *code = FDEVP_PSM_TIMEOUT;
-                *data = NULL;
-        } else {
-                CHECK_FCT( ret );
-                *code = ev->code;
-                *data = ev->data;
-                free(ev);
-        }
-        
-        return 0;
-}
-
 /* The state machine thread (controler) */
 static void * p_psm_th( void * arg )
@@ -154 +134 @@
         int created_started = started;
         int event;
+        size_t ev_sz;
         void * ev_data;
         
@@ -182 +163 @@
 psm_loop:
         /* Get next event */
-        CHECK_FCT_DO( psm_ev_timedget(peer, &event, &ev_data), goto psm_end );
-        TRACE_DEBUG(FULL, "'%s'\t<-- '%s'\t(%p)\t'%s'",
+        CHECK_FCT_DO( fd_event_timedget(peer->p_events, &peer->p_psm_timer, FDEVP_PSM_TIMEOUT, &event, &ev_sz, &ev_data), goto psm_end );
+        TRACE_DEBUG(FULL, "'%s'\t<-- '%s'\t(%p,%g)\t'%s'",
                         STATE_STR(peer->p_hdr.info.pi_state),
-                        fd_pev_str(event), ev_data,
+                        fd_pev_str(event), ev_data, ev_sz,
                         peer->p_hdr.info.pi_diamid);
 
@@ -272 +253 @@
         
         if (peer->p_hdr.info.pi_state != STATE_ZOMBIE) {
-                CHECK_FCT( fd_event_send(peer->p_events, FDEVP_TERMINATE, NULL) );
+                CHECK_FCT( fd_event_send(peer->p_events, FDEVP_TERMINATE, 0, NULL) );
         } else {
                 TRACE_DEBUG(FULL, "Peer '%s' was already terminated", peer->p_hdr.info.pi_diamid);

freeDiameter/sctp.c

@@ -35 +35 @@
 
 #include "fD.h"
+#include "cnxctx.h"
+
 #include <netinet/sctp.h>
 #include <sys/uio.h>
+
+/* Size of buffer to receive ancilliary data. May need to be enlarged if more sockopt are set... */
+#ifndef CMSG_BUF_LEN
+#define CMSG_BUF_LEN    1024
+#endif /* CMSG_BUF_LEN */
 
 /* Pre-binding socket options -- # streams read in config */
@@ -744 +751 @@
 
 /* Retrieve streams information from a connected association -- optionaly provide the primary address */
-int fd_sctp_get_str_info( int sock, int *in, int *out, sSS *primary )
+int fd_sctp_get_str_info( int sock, uint16_t *in, uint16_t *out, sSS *primary )
 {
         struct sctp_status status;
@@ -776 +783 @@
         #endif /* DEBUG_SCTP */
         
-        *in = (int)status.sstat_instrms;
-        *out = (int)status.sstat_outstrms;
+        *in = status.sstat_instrms;
+        *out = status.sstat_outstrms;
         
         if (primary)
@@ -893 +900 @@
 }
 
+/* Send a buffer over a specified stream */
+int fd_sctp_sendstr(int sock, uint16_t strid, uint8_t * buf, size_t len)
+{
+        struct msghdr mhdr;
+        struct iovec  iov;
+        struct {
+                struct cmsghdr          hdr;
+                struct sctp_sndrcvinfo  sndrcv;
+        } anci;
+        ssize_t ret;
+        
+        TRACE_ENTRY("%d %hu %p %g", sock, strid, buf, len);
+        
+        memset(&mhdr, 0, sizeof(mhdr));
+        memset(&iov,  0, sizeof(iov));
+        memset(&anci, 0, sizeof(anci));
+        
+        /* IO Vector: message data */
+        iov.iov_base = buf;
+        iov.iov_len  = len;
+        
+        /* Anciliary data: specify SCTP stream */
+        anci.hdr.cmsg_len   = sizeof(anci);
+        anci.hdr.cmsg_level = IPPROTO_SCTP;
+        anci.hdr.cmsg_type  = SCTP_SNDRCV;
+        anci.sndrcv.sinfo_stream = strid;
+        /* note : we could store other data also, for example in .sinfo_ppid for remote peer or in .sinfo_context for errors. */
+        
+        /* We don't use mhdr.msg_name here; it could be used to specify an address different from the primary */
+        
+        mhdr.msg_iov    = &iov;
+        mhdr.msg_iovlen = 1;
+        
+        mhdr.msg_control    = &anci;
+        mhdr.msg_controllen = sizeof(anci);
+        
+        #ifdef DEBUG_SCTP
+        TRACE_DEBUG(FULL, "Sending %db data on stream %hu of socket %d", len, strid, sock);
+        #endif /* DEBUG_SCTP */
+        
+        CHECK_SYS( ret = sendmsg(sock, &mhdr, 0) );
+        ASSERT( ret == len ); /* There should not be partial delivery with sendmsg... */
+        
+        return 0;
+}
+
+/* Receive the next data from the socket, or next notification */
+int fd_sctp_recvmeta(int sock, uint16_t * strid, uint8_t ** buf, size_t * len, int *event)
+{
+        ssize_t                  ret = 0;
+        struct msghdr            mhdr;
+        char                     ancidata[ CMSG_BUF_LEN ];
+        struct iovec             iov;
+        uint8_t                 *data = NULL;
+        size_t                   bufsz = 0, datasize = 0;
+        size_t                   mempagesz = sysconf(_SC_PAGESIZE); /* We alloc buffer by memory pages for efficiency */
+        
+        TRACE_ENTRY("%d %p %p %p %p", sock, strid, buf, len, event);
+        CHECK_PARAMS( (sock > 0) && buf && len && event );
+        
+        /* Cleanup out parameters */
+        *buf = NULL;
+        *len = 0;
+        *event = 0;
+        
+        /* Prepare header for receiving message */
+        memset(&mhdr, 0, sizeof(mhdr));
+        mhdr.msg_iov    = &iov;
+        mhdr.msg_iovlen = 1;
+        mhdr.msg_control    = &ancidata;
+        mhdr.msg_controllen = sizeof(ancidata);
+        
+        /* We will loop while all data is not received. */
+incomplete:
+        if (datasize == bufsz) {
+                /* The buffer is full, enlarge it */
+                bufsz += mempagesz;
+                CHECK_MALLOC( data = realloc(data, bufsz) );
+        }
+        /* the new data will be received following the preceding */
+        memset(&iov,  0, sizeof(iov));
+        iov.iov_base = data + datasize ;
+        iov.iov_len  = bufsz - datasize;
+
+        /* Receive data from the socket */
+        pthread_cleanup_push(free, data);
+        ret = recvmsg(sock, &mhdr, 0);
+        pthread_cleanup_pop(0);
+        
+        /* Handle errors */
+        if (ret <= 0) { /* Socket is closed, or an error occurred */
+                CHECK_SYS_DO(ret, /* to log in case of error */);
+                free(data);
+                *event = FDEVP_CNX_ERROR;
+                return 0;
+        }
+        
+        /* Update the size of data we received */
+        datasize += ret;
+
+        /* SCTP provides an indication when we received a full record; loop if it is not the case */
+        if ( ! (mhdr.msg_flags & MSG_EOR) ) {
+                goto incomplete;
+        }
+        
+        /* Handle the case where the data received is a notification */
+        if (mhdr.msg_flags & MSG_NOTIFICATION) {
+                union sctp_notification * notif = (union sctp_notification *) data;
+                
+                switch (notif->sn_header.sn_type) {
+                        
+                        case SCTP_ASSOC_CHANGE:
+                                #ifdef DEBUG_SCTP
+                                TRACE_DEBUG(FULL, "Received SCTP_ASSOC_CHANGE notification");
+                                TRACE_DEBUG(FULL, "    state : %hu", notif->sn_assoc_change.sac_state);
+                                TRACE_DEBUG(FULL, "    error : %hu", notif->sn_assoc_change.sac_error);
+                                TRACE_DEBUG(FULL, "    instr : %hu", notif->sn_assoc_change.sac_inbound_streams);
+                                TRACE_DEBUG(FULL, "   outstr : %hu", notif->sn_assoc_change.sac_outbound_streams);
+                                #endif /* DEBUG_SCTP */
+                                
+                                *event = FDEVP_CNX_EP_CHANGE;
+                                break;
+        
+                        case SCTP_PEER_ADDR_CHANGE:
+                                #ifdef DEBUG_SCTP
+                                TRACE_DEBUG(FULL, "Received SCTP_PEER_ADDR_CHANGE notification");
+                                TRACE_DEBUG_sSA(FULL, "    intf_change : ", &(notif->sn_paddr_change.spc_aaddr), NI_NUMERICHOST | NI_NUMERICSERV, "" );
+                                TRACE_DEBUG(FULL, "          state : %d", notif->sn_paddr_change.spc_state);
+                                TRACE_DEBUG(FULL, "          error : %d", notif->sn_paddr_change.spc_error);
+                                #endif /* DEBUG_SCTP */
+                                
+                                *event = FDEVP_CNX_EP_CHANGE;
+                                break;
+        
+                        case SCTP_SEND_FAILED:
+                                #ifdef DEBUG_SCTP
+                                TRACE_DEBUG(FULL, "Received SCTP_SEND_FAILED notification");
+                                TRACE_DEBUG(FULL, "    len : %hu", notif->sn_send_failed.ssf_length);
+                                TRACE_DEBUG(FULL, "    err : %d",  notif->sn_send_failed.ssf_error);
+                                #endif /* DEBUG_SCTP */
+                                
+                                *event = FDEVP_CNX_ERROR;
+                                break;
+                        
+                        case SCTP_REMOTE_ERROR:
+                                #ifdef DEBUG_SCTP
+                                TRACE_DEBUG(FULL, "Received SCTP_REMOTE_ERROR notification");
+                                TRACE_DEBUG(FULL, "    err : %hu", ntohs(notif->sn_remote_error.sre_error));
+                                TRACE_DEBUG(FULL, "    len : %hu", ntohs(notif->sn_remote_error.sre_length));
+                                #endif /* DEBUG_SCTP */
+                                
+                                *event = FDEVP_CNX_ERROR;
+                                break;
+        
+                        case SCTP_SHUTDOWN_EVENT:
+                                #ifdef DEBUG_SCTP
+                                TRACE_DEBUG(FULL, "Received SCTP_SHUTDOWN_EVENT notification");
+                                #endif /* DEBUG_SCTP */
+                                
+                                *event = FDEVP_CNX_ERROR;
+                                break;
+                        
+                        default:        
+                                TRACE_DEBUG(FULL, "Received unknown notification %d, assume error", notif->sn_header.sn_type);
+                                *event = FDEVP_CNX_ERROR;
+                }
+                
+                free(data);
+                return 0;
+        }
+        
+        /* From this point, we have received a message */
+        *event = FDEVP_CNX_MSG_RECV;
+        *buf = data;
+        *len = datasize;
+        
+        if (strid) {
+                struct cmsghdr          *hdr;
+                struct sctp_sndrcvinfo  *sndrcv;
+                
+                /* Handle the anciliary data */
+                for (hdr = CMSG_FIRSTHDR(&mhdr); hdr; hdr = CMSG_NXTHDR(&mhdr, hdr)) {
+
+                        /* We deal only with anciliary data at SCTP level */
+                        if (hdr->cmsg_level != IPPROTO_SCTP) {
+                                TRACE_DEBUG(FULL, "Received some anciliary data at level %d, skipped", hdr->cmsg_level);
+                                continue;
+                        }
+                        
+                        /* Also only interested in SCTP_SNDRCV message for the moment */
+                        if (hdr->cmsg_type != SCTP_SNDRCV) {
+                                TRACE_DEBUG(FULL, "Anciliary block IPPROTO_SCTP / %d, skipped", hdr->cmsg_type);
+                                continue;
+                        }
+                        
+                        sndrcv = (struct sctp_sndrcvinfo *) CMSG_DATA(hdr);
+                        #ifdef DEBUG_SCTP
+                        TRACE_DEBUG(FULL, "Anciliary block IPPROTO_SCTP / SCTP_SNDRCV");
+                        TRACE_DEBUG(FULL, "    sinfo_stream    : %hu", sndrcv->sinfo_stream);
+                        TRACE_DEBUG(FULL, "    sinfo_ssn       : %hu", sndrcv->sinfo_ssn);
+                        TRACE_DEBUG(FULL, "    sinfo_flags     : %hu", sndrcv->sinfo_flags);
+                        /* TRACE_DEBUG(FULL, "    sinfo_pr_policy : %hu", sndrcv->sinfo_pr_policy); */
+                        TRACE_DEBUG(FULL, "    sinfo_ppid      : %u" , sndrcv->sinfo_ppid);
+                        TRACE_DEBUG(FULL, "    sinfo_context   : %u" , sndrcv->sinfo_context);
+                        /* TRACE_DEBUG(FULL, "    sinfo_pr_value  : %u" , sndrcv->sinfo_pr_value); */
+                        TRACE_DEBUG(FULL, "    sinfo_tsn       : %u" , sndrcv->sinfo_tsn);
+                        TRACE_DEBUG(FULL, "    sinfo_cumtsn    : %u" , sndrcv->sinfo_cumtsn);
+                        #endif /* DEBUG_SCTP */
+
+                        *strid = sndrcv->sinfo_stream;
+                }
+        }
+        
+        return 0;
+}

freeDiameter/server.c

@@ -120 +120 @@
                 }
         } else {
-                CHECK_FCT_DO( fd_cnx_start_clear(c->conn), goto cleanup );
+                CHECK_FCT_DO( fd_cnx_start_clear(c->conn, 0), goto cleanup );
         }
         
@@ -135 +135 @@
         TODO("Send event to the peer");
         
+        TODO("(later) handshake or start_clear(.., 1)");
         /* The end */
 cleanup:
@@ -152 +153 @@
         
 fatal_error:    /* This has effect to terminate the daemon */
-        CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_TERMINATE, NULL), );
+        CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_TERMINATE, 0, NULL), );
         return NULL;
 }
@@ -198 +199 @@
         /* Send error signal to the daemon */
         TRACE_DEBUG(INFO, "An error occurred in server module! Thread is terminating...");
-        CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_TERMINATE, NULL), );
+        CHECK_FCT_DO(fd_event_send(fd_g_config->cnf_main_ev, FDEV_TERMINATE, 0, NULL), );
 
         return NULL;

freeDiameter/tcp.c

@@ -35 +35 @@
 
 #include "fD.h"
+#include "cnxctx.h"
+
 #include <netinet/tcp.h>
 #include <netinet/ip6.h>

include/freeDiameter/freeDiameter.h

@@ -152 +152 @@
 struct fd_event {
         int      code; /* codespace depends on the queue */
+        size_t   size;
         void    *data;
 };
 
-static __inline__ int fd_event_send(struct fifo *queue, int code, void * data)
-{
-        struct fd_event * ev;
-        CHECK_MALLOC( ev = malloc(sizeof(struct fd_event)) );
-        ev->code = code;
-        ev->data = data;
-        CHECK_FCT( fd_fifo_post(queue, &ev) );
-        return 0;
-}
-static __inline__ int fd_event_get(struct fifo *queue, int *code, void ** data)
-{
-        struct fd_event * ev;
-        CHECK_FCT( fd_fifo_get(queue, &ev) );
-        if (code)
-                *code = ev->code;
-        if (data)
-                *data = ev->data;
-        free(ev);
-        return 0;
-}
-
-/* Events codespace for fd_g_config->cnf_main_ev */
+/* Daemon's codespace: 1000->1999 */
 enum {
-         FDEV_TERMINATE = 1000  /* request to terminate */
+         FDEV_TERMINATE = 1000  /* request to terminate */
         ,FDEV_DUMP_DICT         /* Dump the content of the dictionary */
         ,FDEV_DUMP_EXT          /* Dump state of extensions */
@@ -186 +166 @@
         ,FDEV_DUMP_PEERS        /* Dump the list of peers */
 };
+
+static __inline__ int fd_event_send(struct fifo *queue, int code, size_t datasz, void * data)
+{
+        struct fd_event * ev;
+        CHECK_MALLOC( ev = malloc(sizeof(struct fd_event)) );
+        ev->code = code;
+        ev->size = datasz;
+        ev->data = data;
+        CHECK_FCT( fd_fifo_post(queue, &ev) );
+        return 0;
+}
+static __inline__ int fd_event_get(struct fifo *queue, int *code, size_t *datasz, void ** data)
+{
+        struct fd_event * ev;
+        CHECK_FCT( fd_fifo_get(queue, &ev) );
+        if (code)
+                *code = ev->code;
+        if (datasz)
+                *datasz = ev->size;
+        if (data)
+                *data = ev->data;
+        free(ev);
+        return 0;
+}
+static __inline__ int fd_event_timedget(struct fifo *queue, struct timespec * timeout, int timeoutcode, int *code, size_t *datasz, void ** data)
+{
+        struct fd_event * ev;
+        int ret = 0;
+        ret = fd_fifo_timedget(queue, &ev, timeout);
+        if (ret == ETIMEDOUT) {
+                if (code)
+                        *code = timeoutcode;
+                if (datasz)
+                        *datasz = 0;
+                if (data)
+                        *data = NULL;
+        } else {
+                CHECK_FCT( ret );
+                if (code)
+                        *code = ev->code;
+                if (datasz)
+                        *datasz = ev->size;
+                if (data)
+                        *data = ev->data;
+                free(ev);
+        }
+        return 0;
+}
+static __inline__ void fd_event_destroy(struct fifo **queue, void (*free_cb)(void * data))
+{
+        struct fd_event * ev;
+        /* Purge all events, and free the associated data if any */
+        while (fd_fifo_tryget( *queue, &ev ) == 0) {
+                (*free_cb)(ev->data);
+                free(ev);
+        }
+        CHECK_FCT_DO( fd_fifo_del(queue), /* continue */ );
+        return ;
+}  
 const char * fd_ev_str(int event); /* defined in freeDiameter/main.c */
 

include/freeDiameter/libfreeDiameter.h

@@ -427 +427 @@
 
 
+/* Trace a binary buffer content */
+#define TRACE_DEBUG_BUFFER(level, prefix, buf, bufsz, suffix ) {                                                                \
+        if ( TRACE_BOOL(level) ) {                                                                                              \
+                int __i;                                                                                                        \
+                size_t __sz = (size_t)(bufsz);                                                                                  \
+                uint8_t * __buf = (uint8_t *)(buf);                                                                             \
+                char * __thn = ((char *)pthread_getspecific(fd_log_thname) ?: "unnamed");                                       \
+                fd_log_debug("\t | tid:%-20s\t%s\tin %s@%s:%d\n"                                                                \
+                          "\t%s|%*s" prefix ,                                                                                   \
+                                        __thn, fd_log_time(NULL, __buf, sizeof(__buf)), __PRETTY_FUNCTION__, __FILE__, __LINE__,\
+                                        (level < FULL)?"@":" ",level, "");                                                      \
+                for (__i = 0; __i < __sz; __i++) {                                                                              \
+                        fd_log_debug("%02.2hhx", __buf[__i]);                                                                   \
+                }                                                                                                               \
+                fd_log_debug(suffix "\n");                                                                                      \
+        }                                                                                                                       \
+}
+
+
+
 /*============================================================*/
 /*                          THREADS                           */
@@ -508 +528 @@
 void fd_list_insert_after  ( struct fd_list * ref, struct fd_list * item );
 void fd_list_insert_before ( struct fd_list * ref, struct fd_list * item );
+
+/* Move a list at the end of another */
+void fd_list_move_end(struct fd_list * ref, struct fd_list * senti);
 
 /* Insert an item in an ordered list -- ordering function provided. If duplicate object found, EEXIST and it is returned in ref_duplicate */
@@ -2385 +2408 @@
 
 /*
+ * FUNCTION:    fd_fifo_move
+ *
+ * PARAMETERS:
+ *  old         : Location of a FIFO that is to be emptied and deleted.
+ *  new         : A FIFO that will receive the old data.
+ *  loc_update  : if non NULL, a place to store the pointer to new FIFO atomically with the move.
+ *
+ * DESCRIPTION: 
+ *  Delete a queue and move its content to another one atomically.
+ *
+ * RETURN VALUE:
+ *  0           : The queue has been destroyed successfully.
+ *  EINVAL      : A parameter is invalid.
+ */
+int fd_fifo_move ( struct fifo ** old, struct fifo * new, struct fifo ** loc_update );
+
+/*
  * FUNCTION:    fd_fifo_length
  *

libfreeDiameter/fifo.c

@@ -150 +150 @@
         CHECK_POSIX(  pthread_mutex_lock( &q->mtx )  );
         
-        /* Ok, now invalidate the queue */
-        q->eyec = 0xdead;
-        
         if ((q->count != 0) || (q->data != NULL)) {
                 TRACE_DEBUG(INFO, "The queue cannot be destroyed (%d, %p)", q->count, q->data);
@@ -158 +155 @@
                 return EINVAL;
         }
+        
+        /* Ok, now invalidate the queue */
+        q->eyec = 0xdead;
         
         while (q->thrs) {
@@ -179 +179 @@
         free(q);
         *queue = NULL;
+        
+        return 0;
+}
+
+/* Move the content of old into new, and update loc_update atomically */
+int fd_fifo_move ( struct fifo ** old, struct fifo * new, struct fifo ** loc_update )
+{
+        struct fifo * q;
+        int loops = 0;
+        
+        TRACE_ENTRY("%p %p %p", old, new, loc_update);
+        CHECK_PARAMS( old && CHECK_FIFO( *old ) && CHECK_FIFO( new ));
+        
+        q = *old;
+        CHECK_PARAMS( ! q->data );
+        if (new->high) {
+                TODO("Implement support for thresholds in fd_fifo_move...");
+        }
+        
+        /* Update loc_update */
+        *old = NULL;
+        if (loc_update)
+                *loc_update = new;
+        
+        /* Lock the queues */
+        CHECK_POSIX(  pthread_mutex_lock( &q->mtx )  );
+        CHECK_POSIX(  pthread_mutex_lock( &new->mtx )  );
+        
+        /* Any waiting thread on the old queue returns an error */
+        q->eyec = 0xdead;
+        while (q->thrs) {
+                CHECK_POSIX(  pthread_cond_signal(&q->cond)  );
+                CHECK_POSIX(  pthread_mutex_unlock( &q->mtx ));
+                pthread_yield();
+                CHECK_POSIX(  pthread_mutex_lock( &q->mtx )  );
+                ASSERT( ++loops < 10 ); /* detect infinite loops */
+        }
+        
+        /* Move all data from old to new */
+        fd_list_move_end( &new->list, &q->list );
+        if (q->count && (!new->count)) {
+                CHECK_POSIX(  pthread_cond_signal(&new->cond)  );
+        }
+        new->count += q->count;
+        
+        /* Destroy old */
+        CHECK_POSIX(  pthread_mutex_unlock( &q->mtx )  );
+        CHECK_POSIX(  pthread_cond_destroy( &q->cond )  );
+        CHECK_POSIX(  pthread_mutex_destroy( &q->mtx )  );
+        free(q);
+        
+        /* Unlock new, we're done */
+        CHECK_POSIX(  pthread_mutex_unlock( &new->mtx )  );
         
         return 0;

libfreeDiameter/lists.c

@@ -70 +70 @@
         ASSERT(ref->head != item);
         list_insert_after(ref, item);
+}
+
+/* Move all elements of list senti at the end of list ref */
+void fd_list_move_end(struct fd_list * ref, struct fd_list * senti)
+{
+        ASSERT(ref->head == ref);
+        ASSERT(senti->head == senti);
+        
+        if (senti->next == senti)
+                return;
+        
+        senti->next->prev = ref->prev;
+        ref->prev->next   = senti->next;
+        senti->prev->next = ref;
+        ref->prev         = senti->prev;
+        senti->prev = senti;
+        senti->next = senti;
+        
 }