[PATCH] boxc connections (#2)

Index: gw/bb_boxc.c
===================================================================
RCS file: /home/cvs/gateway/gw/bb_boxc.c,v
retrieving revision 1.69
diff -a -u -b -r1.69 bb_boxc.c
--- gw/bb_boxc.c	21 Aug 2003 17:06:45 -0000	1.69
+++ gw/bb_boxc.c	31 Aug 2003 19:35:21 -0000
@@ -59,6 +59,9 @@
 static long	boxid = 0;
 extern Mutex *boxid_mutex; 
 
+/* sms_to_smsboxes thread-id */
+static long sms_dequeue_thread;
+
 
 typedef struct _boxc {
     Connection	*conn;
@@ -72,10 +75,13 @@
     List       	*outgoing;
     volatile sig_atomic_t alive;
     Octstr *boxc_id; /* identifies the connected smsbox instance */
-    Mutex *boxc_id_mutex; /* stops boxc_sender until smsbox identification*/
 } Boxc;
 
 
+/* forward declaration */
+static void sms_to_smsboxes(void *arg);
+
+
 /*-------------------------------------------------
  *  receiver thingies
  */
@@ -88,6 +94,7 @@
 
     pack = NULL;
     while (bb_status != BB_DEAD && boxconn->alive) {
+            /* XXX: if box doesn't send (just keep conn open) we block here while shutdown */
 	    pack = conn_read_withlen(boxconn->conn);
 	    gw_claim_area(pack);
 	    if (pack != NULL)
@@ -230,28 +237,26 @@
                  * but we will only consider real identified boxes
                  */
                 if (msg->admin.boxc_id != NULL) {
-                    List *newlist;
 
                     /* and add the boxc_id into conn for boxc_status() output */
-                    if (conn->boxc_id == NULL)
-                        conn->boxc_id = octstr_duplicate(msg->admin.boxc_id);
-                    /* 
-                     * re-link the incoming queue for this connection to 
-                     * an own independent queue
-                     */
-                    newlist = list_create();
-                    list_add_producer(newlist);
-                    conn->incoming = newlist;
-                    conn->retry = newlist;
+                    if (conn->boxc_id != NULL) {
+                        dict_remove(smsbox_by_id, msg->admin.boxc_id);
+                        octstr_destroy(conn->boxc_id);
+                    }
+
+                    conn->boxc_id = msg->admin.boxc_id;
+                    msg->admin.boxc_id = NULL;
 
                     /* add this identified smsbox to the dictionary */       
-                    dict_put(smsbox_by_id, msg->admin.boxc_id, conn);
+                    /* XXX check for equal boxc_id in Dict, otherwise we overwrite it */
+                    dict_put(smsbox_by_id, conn->boxc_id, conn);
                     debug("bb.boxc", 0, "boxc_receiver: got boxc_id <%s> from <%s>",
-                          octstr_get_cstr(msg->admin.boxc_id),
+                          octstr_get_cstr(conn->boxc_id),
                           octstr_get_cstr(conn->client_ip));
+
+                    /* wakeup the dequeue thread */
+                    gwthread_wakeup(sms_dequeue_thread);
                 }
-                debug("bb.boxc", 0, "boxc_receiver: unlocking sender");
-                mutex_unlock(conn->boxc_id_mutex);
             }
             else
                 warning(0, "boxc_receiver: unknown msg received from <%s>, "
@@ -294,13 +299,6 @@
 
     list_add_producer(flow_threads);
 
-    /* wait for smsbox identification */
-    if (bb_status != BB_DEAD && conn->alive && conn->is_wap == 0) {
-        mutex_lock(conn->boxc_id_mutex);
-        debug("bb.boxc", 0, "boxc_sender: sender unlocked");
-        mutex_unlock(conn->boxc_id_mutex);
-    }
-
     while (bb_status != BB_DEAD && conn->alive) {
 
         /* Make sure there's no data left in the outgoing connection before
@@ -309,11 +307,6 @@
 
 	    list_consume(suspended);	/* block here if suspended */
 
-        /*
-         * XXX This list_comsume() makes us block until *all* producers of 
-         * it have been removed, which is sort of a problem, because this
-         * thread keeps running, until there are no more clients connected.
-         */
 	    if ((msg = list_consume(conn->incoming)) == NULL) {
 
 	    /* tell sms/wapbox to die */
@@ -346,6 +339,10 @@
     /* the client closes the connection, after that die in receiver */
     /* conn->alive = 0; */
 
+    /* clear our send queue */
+    while((msg = list_extract_first(conn->incoming)) != NULL)
+        list_produce(incoming_sms, msg);
+
     list_remove_producer(flow_threads);
 }
 
@@ -368,8 +365,6 @@
     boxc->client_ip = ip;
     boxc->alive = 1;
     boxc->connect_time = time(NULL);
-    boxc->boxc_id_mutex = mutex_create();
-    mutex_lock(boxc->boxc_id_mutex);
     boxc->boxc_id = NULL;
     return boxc;
 }    
@@ -385,7 +380,6 @@
 	    conn_destroy(boxc->conn);
     octstr_destroy(boxc->client_ip);
     octstr_destroy(boxc->boxc_id);
-    mutex_destroy(boxc->boxc_id_mutex);
     gw_free(boxc);
 }    
 
@@ -452,36 +446,52 @@
 	    list_remove_producer(flow_threads);
 	    return;
     }
-    newconn->incoming = incoming_sms;
+    newconn->incoming = list_create();
+    list_add_producer(newconn->incoming);
     newconn->retry = incoming_sms;
     newconn->outgoing = outgoing_sms;
     
-    list_append(smsbox_list, newconn);
-
     sender = gwthread_create(boxc_sender, newconn);
     if (sender == -1) {
 	    error(0, "Failed to start a new thread, disconnecting client <%s>",
 	          octstr_get_cstr(newconn->client_ip));
 	    goto cleanup;
     }
+    list_lock(smsbox_list);
+    list_append(smsbox_list, newconn);
+    list_unlock(smsbox_list);
+
+    /* wakeup the dequeueing thread */
+    gwthread_wakeup(sms_dequeue_thread);
+
     list_add_producer(newconn->outgoing);
     boxc_receiver(newconn);
     list_remove_producer(newconn->outgoing);
 
+    /* remove us from smsbox routing list */
+    list_lock(smsbox_list);
+    list_delete_equal(smsbox_list, newconn);
+    list_unlock(smsbox_list);
+    if (newconn->boxc_id) {
+        dict_remove(smsbox_by_id, newconn->boxc_id);
+    }
+
+    /*
+     * check if we in the shutdown phase and sms dequeueing thread
+     *   has removed the producer already
+     */
+    if (list_producer_count(newconn->incoming) > 0)
     list_remove_producer(newconn->incoming);
     gwthread_join(sender);
 
 cleanup:    
-    if (newconn->boxc_id) {
-        dict_remove(smsbox_by_id, newconn->boxc_id);
-        while (list_producer_count(newconn->incoming) > 0)
-            list_remove_producer(newconn->incoming);
         gw_assert(list_len(newconn->incoming) == 0);
         list_destroy(newconn->incoming, NULL);
-    }
-    list_delete_equal(smsbox_list, newconn);
     boxc_destroy(newconn);
 
+    /* wakeup the dequeueing thread */
+    gwthread_wakeup(sms_dequeue_thread);
+
     list_remove_producer(flow_threads);
 }
 
@@ -707,39 +717,37 @@
 }
 
 
-
-
-
-
 static void wait_for_connections(int fd, void (*function) (void *arg), 
     	    	    	    	 List *waited)
 {
-    fd_set rf;
-    struct timeval tv;
     int ret;
+    int timeout = 10; /* 10 sec. */
+
+    gw_assert(function != NULL);
     
     while(bb_status != BB_DEAD) {
 
-	/* XXX: if we are being shutdowned, as long as there is
+	/* if we are being shutdowned, as long as there is
 	 * messages in incoming list allow new connections, but when
-	 * list is empty, exit
+	 * list is empty, exit.
+         * Note: We have timeout (defined above) for which we allow new connections.
+         *           Otherwise we wait here for ever!
 	 */
 	    if (bb_status == BB_SHUTDOWN) {
 	        ret = list_wait_until_nonempty(waited);
-	        if (ret == -1) break;
+	        if (ret == -1 || !timeout)
+                    break;
+                else
+                    timeout--;
 	    }
 
-	    FD_ZERO(&rf);
-	    tv.tv_sec = 1;
-	    tv.tv_usec = 0;
-	
-	    if (bb_status != BB_SUSPENDED)
-	        FD_SET(fd, &rf);
+            /* block here if suspended */
+            list_consume(suspended);
 
-	    ret = select(FD_SETSIZE, &rf, NULL, NULL, &tv);
+            ret = gwthread_pollfd(fd, POLLIN, 1.0);
 	    if (ret > 0) {
 	        gwthread_create(function, (void *)fd);
-	        sleep(1);
+	        gwthread_sleep(1.0);
 	    } else if (ret < 0) {
 	        if(errno==EINTR) continue;
 	        if(errno==EAGAIN) continue;
@@ -774,6 +782,8 @@
 
     wait_for_connections(fd, run_smsbox, incoming_sms);
 
+    list_remove_producer(smsbox_list);
+
     /* continue avalanche */
     list_remove_producer(outgoing_sms);
 
@@ -781,13 +791,15 @@
      * is completely over
      */
 
-    /* XXX KLUDGE fix when list_wait_until_empty() exists */
-    while(list_wait_until_nonempty(smsbox_list)!= -1)
-	    sleep(1);
+    while(list_wait_until_nonempty(smsbox_list) == 1)
+        gwthread_sleep(1.0);
 
     /* close listen socket */
     close(fd);
 
+    gwthread_wakeup(sms_dequeue_thread);
+    gwthread_join(sms_dequeue_thread);
+
     list_destroy(smsbox_list, NULL);
     smsbox_list = NULL;
 
@@ -828,9 +840,8 @@
     /* wait for all connections to die and then remove list
      */
     
-    /* XXX KLUDGE fix when list_wait_until_empty() exists */
-    while(list_wait_until_nonempty(wapbox_list)== 1)
-	      sleep(1);
+    while(list_wait_until_nonempty(wapbox_list) == 1)
+        gwthread_sleep(1.0);
 
     /* wait for wdp_to_wapboxes to exit */
     while(list_consume(wapbox_list)!=NULL)
@@ -946,9 +957,13 @@
     init_smsbox_routes(cfg);
 
     list_add_producer(outgoing_sms);
+    list_add_producer(smsbox_list);
 
     smsbox_running = 1;
     
+    if ((sms_dequeue_thread = gwthread_create(sms_to_smsboxes, NULL)) == -1)
+ 	    panic(0, "Failed to start a new thread for smsbox routing");
+
     if (gwthread_create(smsboxc_run, (void *)smsbox_port) == -1)
 	    panic(0, "Failed to start a new thread for smsbox connections");
 
@@ -1167,10 +1182,11 @@
  * optimized, because every single MO message passes this function and we 
  * have to ensure that no unncessary overhead is done.
  */
-void route_incoming_to_boxc(Msg *msg)
+int route_incoming_to_boxc(Msg *msg)
 {
-    Boxc *bc = NULL;
+    Boxc *bc = NULL, *best = NULL;
     Octstr *s, *r;
+    long len, b, i;
 
     s = r = NULL;
     gw_assert(msg_type(msg) == sms);
@@ -1184,13 +1200,19 @@
     if (msg->sms.boxc_id != NULL) {
         
         bc = dict_get(smsbox_by_id, msg->sms.boxc_id);
-        if (bc == 0) {
+        if (bc == NULL) {
             /* 
              * something is wrong, this was the smsbox connection we used 
              * for sending, so it seems this smsbox is gone
              */
-            error(0,"Could not route message to smsbox id <%s>, smsbox is gone!",
+            warning(0,"Could not route message to smsbox id <%s>, smsbox is gone!",
                   octstr_get_cstr(msg->sms.boxc_id));
+            /*
+             * put msg into global incoming queue and wait until smsbox
+             * with this boxc_id connected agin
+            */
+            list_produce(incoming_sms, msg);
+            return -1;
         } 
     }
 
@@ -1204,14 +1226,107 @@
         bc = r ? dict_get(smsbox_by_id, r) : (s ? dict_get(smsbox_by_id, s) : NULL);
     }
 
+    /* check if we found our routing */
+    if (bc != NULL) {
+        list_produce(bc->incoming, msg);
+        return 0; /* we are done */
+    }
+    else if (s != NULL || r != NULL) {
+        /*
+         * we have routing defined, but no smsbox connected at the moment.
+         * put msg into global incoming queue and wait until smsbox with
+         * such boxc_id connected.
+         */
+         list_produce(incoming_sms, msg);
+         return -1;
+    }
+
+
     /* 
      * ok, none of the routing things applied previously, so route it to
-     * a random smsbox via the shared incoming_sms queue, otherwise to the
-     * smsc specific incoming queue
+     * a random smsbox.
      */
-    if (bc == NULL)
+    list_lock(smsbox_list);
+
+    /* take random smsbox from list, and then check all smsboxes
+     * and select the one with lowest load level - if tied, the first
+     * one
+     */
+    len = list_len(smsbox_list);
+    b = gw_rand() % len;
+    best = list_get(smsbox_list, b);
+    best = (best->boxc_id == NULL ? best : NULL);
+
+    for(i = 0; i < list_len(smsbox_list); i++) {
+	bc = list_get(smsbox_list, (i+b) % len);
+        bc = (bc->boxc_id == NULL ? bc : NULL);
+	if ((bc != NULL && best != NULL && bc->load < best->load) ||
+             (bc != NULL && best == NULL)) {
+	    best = bc;
+        }
+    }
+
+    if (best != NULL) {
+        best->load++;
+        list_produce(best->incoming, msg);
+    }
+
+    list_unlock(smsbox_list);
+
+    if (best == NULL) {
+	warning(0, "smsbox_list empty!");
         list_produce(incoming_sms, msg);
-    else
-        list_produce(bc->incoming, msg);
+	return -1;
+    }
+
+    return 0;
 }
 
+static void sms_to_smsboxes(void *arg)
+{
+    Msg *newmsg, *startmsg, *msg;
+    long i, len;
+    int ret = -1;
+    Boxc *boxc;
+
+    list_add_producer(flow_threads);
+
+    newmsg = startmsg = msg = NULL;
+
+    while(bb_status != BB_DEAD) {
+
+        if (newmsg == startmsg) {
+            /* check if we are in shutdown phase */
+            if (list_producer_count(smsbox_list) == 0)
+                break;
+
+            if (ret == -1)
+                gwthread_sleep(-1.0);
+            startmsg = list_consume(incoming_sms);
+            msg = startmsg;
+            newmsg = NULL;
+        }
+        else {
+            newmsg = list_consume(incoming_sms);
+            msg = newmsg;
+        }
+
+        if (msg == NULL)
+            break;
+
+        gw_assert(msg_type(msg) == sms);
+
+        if ((ret = route_incoming_to_boxc(msg)) == 0)
+            startmsg = newmsg = NULL;
+    }
+
+    list_lock(smsbox_list);
+    len = list_len(smsbox_list);
+    for (i=0; i < len; i++) {
+        boxc = list_get(smsbox_list, i);
+        list_remove_producer(boxc->incoming);
+    }
+    list_unlock(smsbox_list);
+
+    list_remove_producer(flow_threads);
+}