[PATCH] boxc connections

Index: gw/bb_boxc.c
===================================================================
RCS file: /home/cvs/gateway/gw/bb_boxc.c,v
retrieving revision 1.69
diff -a -u -r1.69 bb_boxc.c
--- gw/bb_boxc.c	21 Aug 2003 17:06:45 -0000	1.69
+++ gw/bb_boxc.c	25 Aug 2003 18:49:42 -0000
@@ -43,9 +43,9 @@
 static List	*smsbox_list = NULL;
 
 /* dictionaries for holding the smsbox routing information */
-static Dict *smsbox_by_id = NULL; 
-static Dict *smsbox_by_smsc = NULL; 
-static Dict *smsbox_by_receiver = NULL; 
+static Dict *smsbox_by_id = NULL;
+static Dict *smsbox_by_smsc = NULL;
+static Dict *smsbox_by_receiver = NULL;
 
 static long	smsbox_port;
 static int smsbox_port_ssl = 0;
@@ -57,7 +57,7 @@
 
 
 static long	boxid = 0;
-extern Mutex *boxid_mutex; 
+extern Mutex *boxid_mutex;
 
 
 typedef struct _boxc {
@@ -65,17 +65,20 @@
     int		is_wap;
     long      	id;
     int		load;
-    time_t	connect_time;
+    time_t	        connect_time;
     Octstr    	*client_ip;
-    List      	*incoming;
-    List      	*retry;   	/* If sending fails */
+    List        	*incoming;
+    List      	        *retry;   	/* If sending fails */
     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*/
+    Octstr          *boxc_id; /* identifies the connected smsbox instance */
 } Boxc;
 
 
+/* forward declaration */
+static void sms_to_smsboxes(void *arg);
+
+
 /*-------------------------------------------------
  *  receiver thingies
  */
@@ -88,6 +91,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)
@@ -225,33 +229,28 @@
             /* if this is an identification message from an smsbox instance */
             else if (msg_type(msg) == admin && msg->admin.command == cmd_identify) {
                 
-                /* 
+                /*
                  * any smsbox sends this command even if boxc_id is NULL,
                  * 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;
-
-                    /* add this identified smsbox to the dictionary */       
-                    dict_put(smsbox_by_id, msg->admin.boxc_id, conn);
+                    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 */
+                    /* 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));
                 }
-                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 +293,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
@@ -346,6 +338,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);
 }
 
@@ -362,17 +358,15 @@
     boxc->is_wap = 0;
     boxc->load = 0;
     boxc->conn = conn_wrap_fd(fd, ssl);
-    mutex_lock(boxid_mutex); 
+    mutex_lock(boxid_mutex);
     boxc->id = boxid++;
     mutex_unlock(boxid_mutex);
     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;
-}    
+}
 
 static void boxc_destroy(Boxc *boxc)
 {
@@ -385,9 +379,8 @@
 	    conn_destroy(boxc->conn);
     octstr_destroy(boxc->client_ip);
     octstr_destroy(boxc->boxc_id);
-    mutex_destroy(boxc->boxc_id_mutex);
     gw_free(boxc);
-}    
+}
 
 
 
@@ -444,7 +437,7 @@
     int fd;
     Boxc *newconn;
     long sender;
-    
+
     list_add_producer(flow_threads);
     fd = (int)arg;
     newconn = accept_boxc(fd, smsbox_port_ssl);
@@ -452,11 +445,10 @@
 	    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) {
@@ -464,22 +456,24 @@
 	          octstr_get_cstr(newconn->client_ip));
 	    goto cleanup;
     }
+    list_append(smsbox_list, newconn);
+
     list_add_producer(newconn->outgoing);
     boxc_receiver(newconn);
     list_remove_producer(newconn->outgoing);
 
-    list_remove_producer(newconn->incoming);
-    gwthread_join(sender);
-
-cleanup:    
+    /* remove us from smsbox routing list */
+    list_delete_equal(smsbox_list, newconn);
     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);
+
+    list_remove_producer(newconn->incoming);
+    gwthread_join(sender);
+
+cleanup:
+    gw_assert(list_len(newconn->incoming) == 0);
+    list_destroy(newconn->incoming, NULL);
     boxc_destroy(newconn);
 
     list_remove_producer(flow_threads);
@@ -540,7 +534,7 @@
 	    list_remove_producer(newlist);
 
     newconn->alive = 0;
-    
+
     gwthread_join(sender);
 
 cleanup:
@@ -627,7 +621,7 @@
 	    }
 	    conn = best;
 	    conn->load++;	/* simulate new client until we get new values */
-	
+
 	    ap = gw_malloc(sizeof(AddrPar));
 	    ap->address = octstr_duplicate(msg->wdp_datagram.source_address);
 	    ap->port = msg->wdp_datagram.source_port;
@@ -707,22 +701,22 @@
 }
 
 
-
-
-
-
-static void wait_for_connections(int fd, void (*function) (void *arg), 
+static void wait_for_connections(int fd, void (*function) (void *arg),
     	    	    	    	 List *waited)
 {
     fd_set rf;
     struct timeval tv;
     int ret;
-    
+
+    gw_assert(function != NULL);
+
     while(bb_status != BB_DEAD) {
 
 	/* XXX: if we are being shutdowned, as long as there is
 	 * messages in incoming list allow new connections, but when
 	 * list is empty, exit
+         * XXX: if no boxes connected and msgs in the queue we will wait for ever!!!
+         *          timeout must here!!! (alex)
 	 */
 	    if (bb_status == BB_SHUTDOWN) {
 	        ret = list_wait_until_nonempty(waited);
@@ -732,7 +726,7 @@
 	    FD_ZERO(&rf);
 	    tv.tv_sec = 1;
 	    tv.tv_usec = 0;
-	
+
 	    if (bb_status != BB_SUSPENDED)
 	        FD_SET(fd, &rf);
 
@@ -758,8 +752,8 @@
     list_add_producer(flow_threads);
     gwthread_wakeup(MAIN_THREAD_ID);
     port = (int)arg;
-    
-    fd = make_server_socket(port, NULL); 
+
+    fd = make_server_socket(port, NULL);
     	/* XXX add interface_name if required */
 
     if (fd < 0) {
@@ -781,13 +775,14 @@
      * is completely over
      */
 
-    /* XXX KLUDGE fix when list_wait_until_empty() exists */
-    while(list_wait_until_nonempty(smsbox_list)!= -1)
-	    sleep(1);
+    list_wait_until_empty(smsbox_list);
 
     /* close listen socket */
     close(fd);
 
+    while(list_consume(smsbox_list))
+        ;
+
     list_destroy(smsbox_list, NULL);
     smsbox_list = NULL;
 
@@ -798,7 +793,7 @@
     smsbox_by_smsc = NULL;
     dict_destroy(smsbox_by_receiver);
     smsbox_by_receiver = NULL;
-    
+
     list_remove_producer(flow_threads);
 }
 
@@ -810,7 +805,7 @@
     list_add_producer(flow_threads);
     gwthread_wakeup(MAIN_THREAD_ID);
     port = (int)arg;
-    
+
     fd = make_server_socket(port, NULL);
     	/* XXX add interface_name if required */
 
@@ -827,21 +822,19 @@
 
     /* 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);
+
+    list_wait_until_empty(wapbox_list);
 
     /* wait for wdp_to_wapboxes to exit */
     while(list_consume(wapbox_list)!=NULL)
 	;
-    
+
     /* close listen socket */
-    close(fd);   
- 
+    close(fd);
+
     list_destroy(wapbox_list, NULL);
     wapbox_list = NULL;
-    
+
     list_remove_producer(flow_threads);
 }
 
@@ -858,7 +851,7 @@
 
     boxc_id = smsc_ids = shortcuts = NULL;
 
-    list = cfg_get_multi_group(cfg, octstr_imm("smsbox-route")); 
+    list = cfg_get_multi_group(cfg, octstr_imm("smsbox-route"));
  
     /* loop multi-group "smsbox-route" */
     while (list && (grp = list_extract_first(list)) != NULL) { 
@@ -949,6 +942,9 @@
 
     smsbox_running = 1;
     
+    if (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");
 
@@ -1001,7 +997,7 @@
 
     if (gwthread_create(wdp_to_wapboxes, NULL) == -1)
  	    panic(0, "Failed to start a new thread for wapbox routing");
- 
+
     if (gwthread_create(wapboxc_run, (void *)wapbox_port) == -1)
 	    panic(0, "Failed to start a new thread for wapbox connections");
 
@@ -1169,8 +1165,9 @@
  */
 void 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,7 +1181,7 @@
     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
@@ -1204,14 +1201,77 @@
         bc = r ? dict_get(smsbox_by_id, r) : (s ? dict_get(smsbox_by_id, s) : NULL);
     }
 
-    /* 
+    /*
      * 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
      */
-    if (bc == NULL)
-        list_produce(incoming_sms, msg);
-    else
+    if (bc != NULL) {
         list_produce(bc->incoming, msg);
+        return; /* we are done */
+    }
+
+    /* no named boxes found so try round robin */
+    if (list_len(smsbox_list) == 0) {
+        list_produce(incoming_sms, msg);
+        return;
+    }
+
+    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);
+
+    for(i = 0; i < list_len(smsbox_list); i++) {
+	bc = list_get(smsbox_list, (i+b) % len);
+	if ((bc != NULL && best != NULL && bc->load < best->load) ||
+             (bc != NULL && best == NULL)) {
+	    best = bc;
+        }
+    }
+
+    list_unlock(smsbox_list);
+
+    if (best == NULL) {
+	warning(0, "smsbox_list empty!");
+        list_produce(incoming_sms, msg);
+	return;
+    }
+
+    best->load++;
+    list_produce(best->incoming, msg);
 }
 
+static void sms_to_smsboxes(void *arg)
+{
+    Msg *msg = NULL;
+
+    list_add_producer(flow_threads);
+    list_add_producer(smsbox_list);
+
+    while(bb_status != BB_DEAD) {
+        list_consume(suspended);
+
+        msg = list_consume(incoming_sms);
+        if (msg == NULL)
+            break;
+
+        gw_assert(msg_type(msg) == sms);
+
+        if (list_len(smsbox_list) < 1) { /* wait for new connections */
+            list_produce(incoming_sms, msg);
+            list_wait_until_empty(smsbox_list);
+        }
+        else {
+            route_incoming_to_boxc(msg);
+        }
+    }
+
+    list_remove_producer(smsbox_list);
+    list_remove_producer(flow_threads);
+}
Index: gwlib/list.c
===================================================================
RCS file: /home/cvs/gateway/gwlib/list.c,v
retrieving revision 1.35
diff -a -u -r1.35 list.c
--- gwlib/list.c	14 Jul 2003 15:58:34 -0000	1.35
+++ gwlib/list.c	25 Aug 2003 18:49:42 -0000
@@ -317,6 +317,29 @@
 }
 
 
+int list_wait_until_empty(List *list)
+{
+    int ret;
+
+    gw_assert(list != NULL);
+
+    lock(list);
+    while (list->len > 0 && list->num_producers > 0) {
+        list->single_operation_lock->owner = -1;
+        pthread_cond_wait(&list->nonempty,
+                          &list->single_operation_lock->mutex);
+        list->single_operation_lock->owner = gwthread_self();
+    }
+    if (list->len == 0)
+        ret = 1;
+    else
+        ret = -1;
+    unlock(list);
+
+    return ret;
+}
+
+
 void list_add_producer(List *list)
 {
     lock(list);
Index: gwlib/list.h
===================================================================
RCS file: /home/cvs/gateway/gwlib/list.h,v
retrieving revision 1.19
diff -a -u -r1.19 list.h
--- gwlib/list.h	16 Jan 2002 21:24:15 -0000	1.19
+++ gwlib/list.h	25 Aug 2003 18:49:42 -0000
@@ -186,6 +186,11 @@
 
 
 /*
+ * Sleep until the list is empty. 
+ */
+int list_wait_until_empty(List *list);
+
+/*
  * Register a new producer to the list.
  */
 void list_add_producer(List *list);
Index: test/test_boxc.c
===================================================================
RCS file: /home/cvs/gateway/test/test_boxc.c,v
retrieving revision 1.2
diff -a -u -r1.2 test_boxc.c
--- test/test_boxc.c	22 Aug 2003 08:19:16 -0000	1.2
+++ test/test_boxc.c	25 Aug 2003 18:49:42 -0000
@@ -40,7 +40,7 @@
         /* identify ourself to bearerbox */
         msg = msg_create(admin);
         msg->admin.command = cmd_identify;
-        msg->admin.boxc_id = octstr_create("test-smsbox");
+        msg->admin.boxc_id = octstr_format("test-smsbox-%d", getpid());
         write_to_bearerbox(msg);
 
         /* do something, like passing MT messages */