Hi List,
attached you can find v3 of boxc patch. For more details for what it should be
good, see Message-Id: <[EMAIL PROTECTED]>.
Comments and votes please...
--
Best regards / Mit besten Gr��en aus D�sseldorf
Dipl.-Ing.
Alexander Malysh
___________________________________________
Centrium GmbH
Vogelsanger Weg 80
40470 D�sseldorf
Fon: +49 (0211) 74 84 51 80
Fax: +49 (0211) 277 49 109
email: [EMAIL PROTECTED]
web: www.centrium.de
msn: [EMAIL PROTECTED]
icq: 98063111
___________________________________________
Please avoid sending me Word or PowerPoint attachments.
See http://www.fsf.org/philosophy/no-word-attachments.html
Index: gw/bb_boxc.c
===================================================================
RCS file: /home/cvs/gateway/gw/bb_boxc.c,v
retrieving revision 1.70
diff -a -u -r1.70 bb_boxc.c
--- gw/bb_boxc.c 12 Oct 2003 21:46:09 -0000 1.70
+++ gw/bb_boxc.c 16 Oct 2003 16:28:25 -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,15 @@
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*/
+ /* used to mark connection usable or still waiting for ident. msg */
+ volatile int routable;
} Boxc;
+/* forward declaration */
+static void sms_to_smsboxes(void *arg);
+
+
/*-------------------------------------------------
* receiver thingies
*/
@@ -88,6 +96,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)
@@ -110,7 +119,7 @@
return NULL;
}
}
-
+
if (pack == NULL)
return NULL;
@@ -140,6 +149,7 @@
gw_assert(mack != NULL);
mack->ack.id = msg->sms.id;
mack->ack.time = msg->sms.time;
+
store_save(msg);
rc = smsc2_rout(msg);
@@ -150,8 +160,9 @@
case 0:
mack->ack.nack = ack_buffered;
break;
- case -1:
- warning(0, "Message rejected by bearerbox, no router!");
+ case -1: /* no router at all */
+ case -2: /* outgoing queue too long */
+ warning(0, "Message rejected by bearerbox, %s!", (rc==-1) ? "no router": "queue full");
/*
* first create nack for store-file, in order to delete
* message from store-file.
@@ -160,12 +171,12 @@
gw_assert(mack_store != NULL);
mack_store->ack.id = msg->sms.id;
mack_store->ack.time = msg->sms.time;
- mack_store->ack.nack = ack_failed;
+ mack_store->ack.nack = (rc==-1?ack_failed:ack_failed_tmp);
store_save(mack_store);
msg_destroy(mack_store);
/* create failed nack */
- mack->ack.nack = ack_failed;
+ mack->ack.nack = (rc==-1?ack_failed:ack_failed_tmp);
/* destroy original message */
msg_destroy(msg);
@@ -180,7 +191,7 @@
static void boxc_receiver(void *arg)
{
Boxc *conn = arg;
- Msg *msg;
+ Msg *msg, *mack;
/* remove messages from socket until it is closed */
while (bb_status != BB_DEAD && conn->alive) {
@@ -194,15 +205,33 @@
break;
}
+ /* we don't accept new messages in shutdown phase */
+ if ((bb_status == BB_SHUTDOWN || bb_status == BB_DEAD) && msg_type(msg) == sms) {
+ mack = msg_create(ack);
+ mack->ack.id = msg->sms.id;
+ mack->ack.time = msg->sms.time;
+ mack->ack.nack = ack_failed_tmp;
+ msg_destroy(msg);
+ list_produce(conn->incoming, mack);
+ continue;
+ }
+
if (msg_type(msg) == sms && conn->is_wap == 0) {
debug("bb.boxc", 0, "boxc_receiver: sms received");
/* deliver message to queue */
deliver_sms_to_queue(msg, conn);
+ if (conn->routable == 0) {
+ conn->routable = 1;
+ /* wakeup the dequeue thread */
+ gwthread_wakeup(sms_dequeue_thread);
+ }
} else if (msg_type(msg) == wdp_datagram && conn->is_wap) {
debug("bb.boxc", 0, "boxc_receiver: got wdp from wapbox");
-
+
+ /* XXX we should block these in SHUTDOWN phase too, but
+ we need ack/nack msgs implemented first. */
list_produce(conn->outgoing, msg);
} else if (msg_type(msg) == sms && conn->is_wap) {
@@ -211,6 +240,11 @@
/* should be a WAP push message, so tried it the same way */
deliver_sms_to_queue(msg, conn);
+ if (conn->routable == 0) {
+ conn->routable = 1;
+ /* wakeup the dequeue thread */
+ gwthread_wakeup(sms_dequeue_thread);
+ }
} else {
if (msg_type(msg) == heartbeat) {
if (msg->heartbeat.load != conn->load)
@@ -224,41 +258,40 @@
}
/* 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);
+
+ conn->routable = 1;
+ /* wakeup the dequeue thread */
+ gwthread_wakeup(sms_dequeue_thread);
}
else
warning(0, "boxc_receiver: unknown msg received from <%s>, "
"ignored", octstr_get_cstr(conn->client_ip));
msg_destroy(msg);
}
- }
+ }
}
@@ -271,17 +304,24 @@
Octstr *pack;
pack = msg_pack(pmsg);
+
+ if (pack == NULL)
+ return -1;
+
if (boxconn->boxc_id != NULL)
debug("bb.boxc", 0, "send_msg: sending msg to boxc: <%s>",
octstr_get_cstr(boxconn->boxc_id));
else
debug("bb.boxc", 0, "send_msg: sending msg to box: <%s>",
octstr_get_cstr(boxconn->client_ip));
+
if (conn_write_withlen(boxconn->conn, pack) == -1) {
error(0, "Couldn't write Msg to box <%s>, disconnecting",
octstr_get_cstr(boxconn->client_ip));
- return -1;
+ octstr_destroy(pack);
+ return -1;
}
+
octstr_destroy(pack);
return 0;
}
@@ -294,58 +334,62 @@
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
- * doing the potentially blocking list_consume()s */
- conn_flush(conn->conn);
-
- 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.
+ * Make sure there's no data left in the outgoing connection before
+ * doing the potentially blocking list_consume()s
*/
- if ((msg = list_consume(conn->incoming)) == NULL) {
+ conn_flush(conn->conn);
- /* tell sms/wapbox to die */
- msg = msg_create(admin);
- msg->admin.command = restart ? cmd_restart : cmd_shutdown;
- send_msg(conn, msg);
- msg_destroy(msg);
- break;
- }
- if (msg_type(msg) == heartbeat) {
- debug("bb.boxc", 0, "boxc_sender: catch an heartbeat - we are alive");
- msg_destroy(msg);
- continue;
- }
- if (!conn->alive) {
- /* we got message here */
- list_produce(conn->retry, msg);
- break;
- }
- if (send_msg(conn, msg) == -1) {
- /* if we fail to send, return msg to the list it came from
- * before dying off */
- list_produce(conn->retry, msg);
- break;
- }
- msg_destroy(msg);
- debug("bb.boxc", 0, "boxc_sender: sent message to <%s>",
- octstr_get_cstr(conn->client_ip));
+ list_consume(suspended); /* block here if suspended */
+
+ if ((msg = list_consume(conn->incoming)) == NULL) {
+ /* tell sms/wapbox to die */
+ msg = msg_create(admin);
+ msg->admin.command = restart ? cmd_restart : cmd_shutdown;
+ send_msg(conn, msg);
+ msg_destroy(msg);
+ break;
+ }
+ if (msg_type(msg) == heartbeat) {
+ debug("bb.boxc", 0, "boxc_sender: catch an heartbeat - we are alive");
+ msg_destroy(msg);
+ continue;
+ }
+ if (!conn->alive || send_msg(conn, msg) == -1) {
+ /* we got message here */
+ if (msg_type(msg) != ack) /* XXX can't route ack's */
+ list_produce(conn->retry, msg);
+ break;
+ }
+ msg_destroy(msg);
+ debug("bb.boxc", 0, "boxc_sender: sent message to <%s>",
+ octstr_get_cstr(conn->client_ip));
}
/* the client closes the connection, after that die in receiver */
/* conn->alive = 0; */
+ /* set conn to unroutable */
+ list_lock(smsbox_list); /* to be sure no router are there */
+ conn->routable = 0;
+ list_unlock(smsbox_list);
+
+ /* clear our send queue */
+ while((msg = list_extract_first(conn->incoming)) != NULL) {
+ /*
+ * XXX: we can't route ack's at the moment, so just drop these
+ * from the queue. In future, we should have boxc_id for ack's
+ * so we can properly route these when smsbox will connect again.
+ */
+ if (msg_type(msg) == ack) {
+ warning(0, "bb_boxc: not sent ACK found, drop them.");
+ msg_destroy(msg);
+ }
+ else
+ list_produce(incoming_sms, msg);
+ }
+
list_remove_producer(flow_threads);
}
@@ -357,37 +401,35 @@
static Boxc *boxc_create(int fd, Octstr *ip, int ssl)
{
Boxc *boxc;
-
+
boxc = gw_malloc(sizeof(Boxc));
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;
+ boxc->routable = 0;
return boxc;
-}
+}
static void boxc_destroy(Boxc *boxc)
{
if (boxc == NULL)
return;
-
+
/* do nothing to the lists, as they are only references */
if (boxc->conn)
conn_destroy(boxc->conn);
octstr_destroy(boxc->client_ip);
octstr_destroy(boxc->boxc_id);
- mutex_destroy(boxc->boxc_id_mutex);
gw_free(boxc);
-}
+}
@@ -416,7 +458,7 @@
return NULL;
}
newconn = boxc_create(newfd, ip, ssl);
-
+
/*
* check if the SSL handshake was successfull, otherwise
* this is no valid box connection any more
@@ -430,7 +472,7 @@
info(0, "Client connected from <%s> using SSL", octstr_get_cstr(ip));
else
info(0, "Client connected from <%s>", octstr_get_cstr(ip));
-
+
/* XXX TODO: do the hand-shake, baby, yeah-yeah! */
@@ -444,7 +486,7 @@
int fd;
Boxc *newconn;
long sender;
-
+
list_add_producer(flow_threads);
fd = (int)arg;
newconn = accept_boxc(fd, smsbox_port_ssl);
@@ -452,11 +494,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,26 +505,50 @@
octstr_get_cstr(newconn->client_ip));
goto cleanup;
}
+ /*
+ * We register newconn in the smsbox_list here but mark newconn as routable
+ * after identification or first message received from smsbox. So we can avoid
+ * a race condition for routable smsboxes (otherwise between startup and
+ * registration we will forward some messages to smsbox).
+ */
+ list_lock(smsbox_list);
+ list_append(smsbox_list, newconn);
+ list_unlock(smsbox_list);
+
list_add_producer(newconn->outgoing);
boxc_receiver(newconn);
list_remove_producer(newconn->outgoing);
- /* remove producer only if that not a global incoming list */
- if (incoming_sms != newconn->incoming)
+ /* 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);
+cleanup:
+ {
+ long i;
+ for (i=0; i < list_len(newconn->incoming); i++)
+ msg_dump(list_get(newconn->incoming, i), 0);
}
- list_delete_equal(smsbox_list, newconn);
+ gw_assert(list_len(newconn->incoming) == 0);
+ list_destroy(newconn->incoming, NULL);
boxc_destroy(newconn);
+ /* wakeup the dequeueing thread */
+ gwthread_wakeup(sms_dequeue_thread);
+
list_remove_producer(flow_threads);
}
@@ -709,39 +774,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;
@@ -776,6 +839,8 @@
wait_for_connections(fd, run_smsbox, incoming_sms);
+ list_remove_producer(smsbox_list);
+
/* continue avalanche */
list_remove_producer(outgoing_sms);
@@ -783,13 +848,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;
@@ -830,9 +897,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)
@@ -892,8 +958,9 @@
dict_put(smsbox_by_smsc, item, boxc_id);
}
list_destroy(items, octstr_destroy_item);
+ octstr_destroy(smsc_ids);
}
-
+
if (shortcuts) {
items = octstr_split(shortcuts, octstr_imm(";"));
for (i = 0; i < list_len(items); i++) {
@@ -906,8 +973,11 @@
dict_put(smsbox_by_receiver, item, boxc_id);
}
list_destroy(items, octstr_destroy_item);
+ octstr_destroy(shortcuts);
}
- }
+ }
+
+ list_destroy(list, NULL);
}
@@ -948,9 +1018,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");
@@ -1094,10 +1168,12 @@
if (status_type == BBSTATUS_XML)
octstr_format_append(tmp, "<box>\n\t\t<type>smsbox</type>\n"
"\t\t<id>%s</id>\n\t\t<IP>%s</IP>\n"
+ "\t\t<queue>%ld</queue>\n"
"\t\t<status>on-line %ldd %ldh %ldm %lds</status>\n"
"\t\t<ssl>%s</ssl>\n\t</box>",
(bi->boxc_id ? octstr_get_cstr(bi->boxc_id) : ""),
octstr_get_cstr(bi->client_ip),
+ list_len(bi->incoming),
t/3600/24, t/3600%24, t/60%60, t%60,
#ifdef HAVE_LIBSSL
conn_get_ssl(bi->conn) != NULL ? "yes" : "no"
@@ -1106,9 +1182,9 @@
#endif
);
else
- octstr_format_append(tmp, "%ssmsbox:%s, IP %s (on-line %ldd %ldh %ldm %lds) %s %s",
+ octstr_format_append(tmp, "%ssmsbox:%s, IP %s (%ld queued), (on-line %ldd %ldh %ldm %lds) %s %s",
ws, (bi->boxc_id ? octstr_get_cstr(bi->boxc_id) : "(none)"),
- octstr_get_cstr(bi->client_ip),
+ octstr_get_cstr(bi->client_ip), list_len(bi->incoming),
t/3600/24, t/3600%24, t/60%60, t%60,
#ifdef HAVE_LIBSSL
conn_get_ssl(bi->conn) != NULL ? "using SSL" : "",
@@ -1169,10 +1245,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);
@@ -1186,14 +1263,20 @@
if (msg->sms.boxc_id != NULL) {
bc = dict_get(smsbox_by_id, msg->sms.boxc_id);
- if (bc == 0) {
- /*
- * something is wrong, this was the smsbox connection we used
+ 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;
+ }
}
/*
@@ -1206,14 +1289,109 @@
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);
+ if (bc->boxc_id != NULL || bc->routable == 0)
+ 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);
+}
