Re: [PATCH] Guaranteed throughput smsc-independent

diff --git a/gw/smsc/smsc_smpp.c b/gw/smsc/smsc_smpp.c
index cf61122..60cd2df 100644
--- a/gw/smsc/smsc_smpp.c
+++ b/gw/smsc/smsc_smpp.c
@@ -79,6 +79,7 @@
 #include "dlr.h"
 #include "bearerbox.h"
 #include "meta_data.h"
+#include "load.h"
 
 #define SMPP_DEFAULT_CHARSET "UTF-8"
 
@@ -111,7 +112,7 @@
 #define SMPP_MAX_PENDING_SUBMITS    10
 #define SMPP_DEFAULT_VERSION        0x34
 #define SMPP_DEFAULT_PRIORITY       0
-#define SMPP_THROTTLING_SLEEP_TIME  15
+#define SMPP_THROTTLING_SLEEP_TIME  1
 #define SMPP_DEFAULT_CONNECTION_TIMEOUT  10 * SMPP_ENQUIRE_LINK_INTERVAL
 #define SMPP_DEFAULT_WAITACK        60
 #define SMPP_DEFAULT_SHUTDOWN_TIMEOUT 30
@@ -167,6 +168,7 @@ typedef struct {
     long connection_timeout;
     long wait_ack;
     int wait_ack_action;
+    Load *load;
     SMSCConn *conn;
 } SMPP;
 
@@ -180,7 +182,7 @@ struct smpp_msg {
 /*
  * create smpp_msg struct
  */
-static struct smpp_msg* smpp_msg_create(Msg *msg)
+static inline struct smpp_msg* smpp_msg_create(Msg *msg)
 {
     struct smpp_msg *result = gw_malloc(sizeof(struct smpp_msg));
 
@@ -195,7 +197,7 @@ static struct smpp_msg* smpp_msg_create(Msg *msg)
 /*
  * destroy smpp_msg struct. If destroy_msg flag is set, then message will be freed as well
  */
-static void smpp_msg_destroy(struct smpp_msg *msg, int destroy_msg)
+static inline void smpp_msg_destroy(struct smpp_msg *msg, int destroy_msg)
 {
     /* sanity check */
     if (msg == NULL)
@@ -263,6 +265,8 @@ static SMPP *smpp_create(SMSCConn *conn, Octstr *host, int transmit_port,
     smpp->bind_addr_ton = 0;
     smpp->bind_addr_npi = 0;
     smpp->use_ssl = 0;
+    smpp->load = load_create_real(0);
+    load_add_interval(smpp->load, 1);
 
     return smpp;
 }
@@ -284,6 +288,7 @@ static void smpp_destroy(SMPP *smpp)
         octstr_destroy(smpp->my_number);
         octstr_destroy(smpp->alt_charset);
         octstr_destroy(smpp->alt_addr_charset);
+        load_destroy(smpp->load);
         gw_free(smpp);
     }
 }
@@ -912,15 +917,15 @@ static SMPP_PDU *msg_to_pdu(SMPP *smpp, Msg *msg)
      * Note: we always send in UTC and just define "Time Difference" as 00 and
      *       direction '+'.
      */
-    validity = msg->sms.validity >= 0 ? msg->sms.validity : smpp->validityperiod;
-    if (validity >= 0) {
+    validity = msg->sms.validity != SMS_PARAM_UNDEFINED ? msg->sms.validity : smpp->validityperiod;
+    if (validity != SMS_PARAM_UNDEFINED) {
         struct tm tm = gw_gmtime(time(NULL) + validity * 60);
         pdu->u.submit_sm.validity_period = octstr_format("%02d%02d%02d%02d%02d%02d000+",
                 tm.tm_year % 100, tm.tm_mon + 1, tm.tm_mday,
                 tm.tm_hour, tm.tm_min, tm.tm_sec);
     }
 
-    if (msg->sms.deferred >= 0) {
+    if (msg->sms.deferred != SMS_PARAM_UNDEFINED) {
         struct tm tm = gw_gmtime(time(NULL) + msg->sms.deferred * 60);
         pdu->u.submit_sm.schedule_delivery_time = octstr_format("%02d%02d%02d%02d%02d%02d000+",
                 tm.tm_year % 100, tm.tm_mon + 1, tm.tm_mday,
@@ -976,6 +981,25 @@ static int send_enquire_link(SMPP *smpp, Connection *conn, long *last_sent)
     return ret;
 }
 
+static int send_gnack(SMPP *smpp, Connection *conn, long reason, unsigned long seq_num)
+{
+    SMPP_PDU *pdu;
+    Octstr *os;
+    int ret;
+
+    pdu = smpp_pdu_create(generic_nack, seq_num);
+    pdu->u.generic_nack.command_status = reason;
+    dump_pdu("Sending generic_nack:", smpp->conn->id, pdu);
+    os = smpp_pdu_pack(pdu);
+    if (os != NULL)
+        ret = conn_write(conn, os);
+    else
+        ret = -1;
+    octstr_destroy(os);
+    smpp_pdu_destroy(pdu);
+
+    return ret;
+}
 
 static int send_unbind(SMPP *smpp, Connection *conn)
 {
@@ -1021,16 +1045,20 @@ static int send_messages(SMPP *smpp, Connection *conn, long *pending_submits)
     Msg *msg;
     SMPP_PDU *pdu;
     Octstr *os;
-    double delay = 0;
 
     if (*pending_submits == -1)
         return 0;
 
-    if (smpp->conn->throughput > 0) {
-        delay = 1.0 / smpp->conn->throughput;
-    }
-
     while (*pending_submits < smpp->max_pending_submits) {
+        /* check our throughput */
+        if (smpp->conn->throughput > 0 && load_get(smpp->load, 0) >= smpp->conn->throughput) {
+            debug("bb.sms.smpp", 0, "SMPP[%s]: throughput limit exceeded (%.02f,%.02f)",
+                  octstr_get_cstr(smpp->conn->id), load_get(smpp->load, 0), smpp->conn->throughput);
+            break;
+        }
+        debug("bb.sms.smpp", 0, "SMPP[%s]: throughput (%.02f,%.02f)",
+              octstr_get_cstr(smpp->conn->id), load_get(smpp->load, 0), smpp->conn->throughput);
+
     	/* Get next message, quit if none to be sent */
     	msg = gw_prioqueue_remove(smpp->msgs_to_send);
         if (msg == NULL)
@@ -1050,12 +1078,9 @@ static int send_messages(SMPP *smpp, Connection *conn, long *pending_submits)
             smpp_pdu_destroy(pdu);
             octstr_destroy(os);
             ++(*pending_submits);
-            /*
-             * obey throughput speed limit, if any.
-             */
-            if (smpp->conn->throughput > 0)
-                gwthread_sleep(delay);
-        } else { /* write error occurs */
+            load_increase(smpp->load);
+        }
+        else { /* write error occurs */
             smpp_pdu_destroy(pdu);
             bb_smscconn_send_failed(smpp->conn, msg, SMSCCONN_FAILED_TEMPORARILY, NULL);
             return -1;
@@ -1613,15 +1638,19 @@ static int handle_pdu(SMPP *smpp, Connection *conn, SMPP_PDU *pdu,
             break;
 
         case bind_transmitter_resp:
-            if (pdu->u.bind_transmitter_resp.command_status != 0) {
-                error(0, "SMPP[%s]: SMSC rejected login to transmit, "
-		              "code 0x%08lx (%s).",
+            if (pdu->u.bind_transmitter_resp.command_status != 0 &&
+                pdu->u.bind_transmitter_resp.command_status != SMPP_ESME_RALYNBD) {
+                error(0, "SMPP[%s]: SMSC rejected login to transmit, code 0x%08lx (%s).",
                       octstr_get_cstr(smpp->conn->id),
                       pdu->u.bind_transmitter_resp.command_status,
-		      smpp_error_to_string(pdu->u.bind_transmitter_resp.command_status));
+                smpp_error_to_string(pdu->u.bind_transmitter_resp.command_status));
+                mutex_lock(smpp->conn->flow_mutex);
+                smpp->conn->status = SMSCCONN_DISCONNECTED;
+                mutex_unlock(smpp->conn->flow_mutex);
                 if (pdu->u.bind_transmitter_resp.command_status == SMPP_ESME_RINVSYSID ||
-                    pdu->u.bind_transmitter_resp.command_status == SMPP_ESME_RINVPASWD)
+                    pdu->u.bind_transmitter_resp.command_status == SMPP_ESME_RINVPASWD) {
                     smpp->quitting = 1;
+                }
             } else {
                 *pending_submits = 0;
                 mutex_lock(smpp->conn->flow_mutex);
@@ -1633,15 +1662,19 @@ static int handle_pdu(SMPP *smpp, Connection *conn, SMPP_PDU *pdu,
             break;
 
         case bind_transceiver_resp:
-            if (pdu->u.bind_transceiver_resp.command_status != 0) {
-                error(0, "SMPP[%s]: SMSC rejected login to transmit, "
-                      "code 0x%08lx (%s).",
+            if (pdu->u.bind_transceiver_resp.command_status != 0 &&
+	            pdu->u.bind_transceiver_resp.command_status != SMPP_ESME_RALYNBD) {
+                error(0, "SMPP[%s]: SMSC rejected login to transmit, code 0x%08lx (%s).",
                       octstr_get_cstr(smpp->conn->id),
                       pdu->u.bind_transceiver_resp.command_status,
-		      smpp_error_to_string(pdu->u.bind_transceiver_resp.command_status));
-                if (pdu->u.bind_transceiver_resp.command_status == SMPP_ESME_RINVSYSID ||
-                    pdu->u.bind_transceiver_resp.command_status == SMPP_ESME_RINVPASWD)
-                    smpp->quitting = 1;
+		         smpp_error_to_string(pdu->u.bind_transceiver_resp.command_status));
+                 mutex_lock(smpp->conn->flow_mutex);
+                 smpp->conn->status = SMSCCONN_DISCONNECTED;
+                 mutex_unlock(smpp->conn->flow_mutex);
+                 if (pdu->u.bind_transceiver_resp.command_status == SMPP_ESME_RINVSYSID ||
+                     pdu->u.bind_transceiver_resp.command_status == SMPP_ESME_RINVPASWD) {
+                     smpp->quitting = 1;
+                 }
             } else {
                 *pending_submits = 0;
                 mutex_lock(smpp->conn->flow_mutex);
@@ -1653,15 +1686,19 @@ static int handle_pdu(SMPP *smpp, Connection *conn, SMPP_PDU *pdu,
             break;
 
         case bind_receiver_resp:
-            if (pdu->u.bind_receiver_resp.command_status != 0) {
-                error(0, "SMPP[%s]: SMSC rejected login to receive, "
-                      "code 0x%08lx (%s).",
+            if (pdu->u.bind_receiver_resp.command_status != 0 &&
+                pdu->u.bind_receiver_resp.command_status != SMPP_ESME_RALYNBD) {
+                error(0, "SMPP[%s]: SMSC rejected login to receive, code 0x%08lx (%s).",
                       octstr_get_cstr(smpp->conn->id),
                       pdu->u.bind_receiver_resp.command_status,
-		      smpp_error_to_string(pdu->u.bind_receiver_resp.command_status));
-                if (pdu->u.bind_receiver_resp.command_status == SMPP_ESME_RINVSYSID ||
-                    pdu->u.bind_receiver_resp.command_status == SMPP_ESME_RINVPASWD)
-                    smpp->quitting = 1;
+                 smpp_error_to_string(pdu->u.bind_receiver_resp.command_status));
+                 mutex_lock(smpp->conn->flow_mutex);
+                 smpp->conn->status = SMSCCONN_DISCONNECTED;
+                 mutex_unlock(smpp->conn->flow_mutex);
+                 if (pdu->u.bind_receiver_resp.command_status == SMPP_ESME_RINVSYSID ||
+                     pdu->u.bind_receiver_resp.command_status == SMPP_ESME_RINVPASWD) {
+                     smpp->quitting = 1;
+                 }
             } else {
                 /* set only resceive status if no transmitt is bind */
                 mutex_lock(smpp->conn->flow_mutex);
@@ -1730,8 +1767,7 @@ static int handle_pdu(SMPP *smpp, Connection *conn, SMPP_PDU *pdu,
              * We received an unknown PDU type, therefore we will respond
              * with a generic_nack PDU, see SMPP v3.4 spec, section 3.3.
              */
-            resp = smpp_pdu_create(generic_nack, pdu->u.generic_nack.sequence_number);
-            resp->u.generic_nack.command_status = SMPP_ESME_RINVCMDID;
+            ret = send_gnack(smpp, conn, SMPP_ESME_RINVCMDID, pdu->u.generic_nack.sequence_number);
             break;
     }
 
@@ -1837,12 +1873,11 @@ static void io_thread(void *arg)
     int transmitter;
     Connection *conn;
     int ret;
-    long last_enquire_sent;
     long pending_submits;
     long len;
     SMPP_PDU *pdu;
     double timeout;
-    time_t last_response, last_cleanup;
+    time_t last_cleanup, last_enquire_sent, last_response, now;
 
     io_arg = arg;
     smpp = io_arg->smpp;
@@ -1852,6 +1887,8 @@ static void io_thread(void *arg)
     /* Make sure we log into our own log-file if defined */
     log_thread_to(smpp->conn->log_idx);
 
+#define IS_ACTIVE (smpp->conn->status == SMSCCONN_ACTIVE || smpp->conn->status == SMSCCONN_ACTIVE_RECV)
+
     conn = NULL;
     while (!smpp->quitting) {
         if (transmitter == 1)
@@ -1860,94 +1897,125 @@ static void io_thread(void *arg)
             conn = open_transceiver(smpp);
         else
             conn = open_receiver(smpp);
-
-        last_enquire_sent = last_cleanup = last_response = date_universal_now();
+        
         pending_submits = -1;
         len = 0;
-        smpp->throttling_err_time = 0;
-        for (;conn != NULL;) {
-            timeout = last_enquire_sent + smpp->enquire_link_interval
-                        - date_universal_now();
-
-            if (conn_wait(conn, timeout) == -1)
+        last_response = last_cleanup = last_enquire_sent = time(NULL);
+        while(conn != NULL) {
+            ret = read_pdu(smpp, conn, &len, &pdu);
+            if (ret == -1) { /* connection broken */
+                error(0, "SMPP[%s]: I/O error or other error. Re-connecting.",
+                      octstr_get_cstr(smpp->conn->id));
                 break;
-
-            /* unbind
-             * Read so long as unbind_resp received or timeout passed. Otherwise we have
-             * double delivered messages.
-             */
-            if (smpp->quitting) {
-                send_unbind(smpp, conn);
-                last_response = time(NULL);
-                while(conn_wait(conn, 1.00) != -1 &&
-                      difftime(time(NULL), last_response) < SMPP_DEFAULT_SHUTDOWN_TIMEOUT &&
-                      smpp->conn->status != SMSCCONN_DISCONNECTED) {
-                    if (read_pdu(smpp, conn, &len, &pdu) == 1) {
-                        dump_pdu("Got PDU:", smpp->conn->id, pdu);
-                        handle_pdu(smpp, conn, pdu, &pending_submits);
-                        smpp_pdu_destroy(pdu);
-                    }
+            } else if (ret == -2) {
+                /* wrong pdu length , send gnack */
+                len = 0;
+                if (send_gnack(smpp, conn, SMPP_ESME_RINVCMDLEN, 0) == -1) {
+                    error(0, "SMPP[%s]: I/O error or other error. Re-connecting.",
+                          octstr_get_cstr(smpp->conn->id));
+                    break;
                 }
-                debug("bb.sms.smpp", 0, "SMPP[%s]: %s: break and shutting down",
-                      octstr_get_cstr(smpp->conn->id), __func__);
-
-                break;
-            }
-
-            send_enquire_link(smpp, conn, &last_enquire_sent);
-
-            while ((ret = read_pdu(smpp, conn, &len, &pdu)) == 1) {
-                last_response = time(NULL);
+            } else if (ret == 1) { /* data available */
                 /* Deal with the PDU we just got */
                 dump_pdu("Got PDU:", smpp->conn->id, pdu);
-                handle_pdu(smpp, conn, pdu, &pending_submits);
+                ret = handle_pdu(smpp, conn, pdu, &pending_submits);
                 smpp_pdu_destroy(pdu);
-
+                if (ret == -1) {
+                    error(0, "SMPP[%s]: I/O error or other error. Re-connecting.",
+                          octstr_get_cstr(smpp->conn->id));
+                    break;
+                }
+                
+                /*
+                 * check if we are still connected
+                 * Note: Function handle_pdu will set status to SMSCCONN_DISCONNECTED
+                 * when unbind was received.
+                 */
+                if (smpp->conn->status == SMSCCONN_DISCONNECTED)
+                    break;
+                
                 /*
-                 * check if we are still connected.
-                 * Note: smsc can send unbind request, so we must check it.
+                 * If we are not bounded then no PDU may coming from SMSC.
+                 * It's just a workaround for buggy SMSC's whoes send enquire_link's
+                 * although link is not bounded. Means: we doesn't notice these and if link
+                 * keep to be not bounden we are reconnect after defined timeout elapsed.
                  */
-                if (smpp->conn->status != SMSCCONN_ACTIVE && smpp->conn->status != SMSCCONN_ACTIVE_RECV) {
-                     /* we are disconnected */
-                     ret = -1;
-                     break;
+                if (IS_ACTIVE) {
+                    /*
+                     * Store last response time.
+                     */
+                    time(&last_response);
                 }
-
-                /* Make sure we send enquire_link even if we read a lot */
-                send_enquire_link(smpp, conn, &last_enquire_sent);
-
-                /* Make sure we send even if we read a lot */
-                if (transmitter && difftime(time(NULL), smpp->throttling_err_time) > SMPP_THROTTLING_SLEEP_TIME) {
-                    smpp->throttling_err_time = 0;
-                    send_messages(smpp, conn, &pending_submits);
+            } else { /* no data available */
+                /* check last enquire_resp, if difftime > as idle_timeout
+                 * mark connection as broken.
+                 * We have some SMSC connections where connection seems to be OK, but
+                 * in reallity is broken, because no responses received.
+                 */
+                if (smpp->connection_timeout > 0 &&
+                    difftime(time(NULL), last_response) > smpp->connection_timeout) {
+                    /* connection seems to be broken */
+                    warning(0, "Got no responses within %ld sec., reconnecting...",
+                            (long) difftime(time(NULL), last_response));
+                    break;
                 }
+                
+                time(&now);
+                timeout = last_enquire_sent + smpp->enquire_link_interval - now;
+                if (!IS_ACTIVE && timeout <= 0)
+                    timeout = smpp->enquire_link_interval;
+                if (transmitter && gw_prioqueue_len(smpp->msgs_to_send) > 0 &&
+                    smpp->throttling_err_time > 0 && pending_submits < smpp->max_pending_submits) {
+                    time_t tr_timeout = smpp->throttling_err_time + SMPP_THROTTLING_SLEEP_TIME - now;
+                    timeout = timeout > tr_timeout ? tr_timeout : timeout;
+                } else if (transmitter && gw_prioqueue_len(smpp->msgs_to_send) > 0 && smpp->conn->throughput > 0 &&
+                           smpp->max_pending_submits > pending_submits) {
+                    double t = 1.0 / smpp->conn->throughput;
+                    timeout = t < timeout ? t : timeout;
+                }
+                /* sleep a while */
+                if (timeout > 0 && conn_wait(conn, timeout) == -1)
+                    break;
             }
-
-            if (ret == -1) {
-                error(0, "SMPP[%s]: I/O error or other error. Re-connecting.",
-                      octstr_get_cstr(smpp->conn->id));
-                break;
-            }
-
-            /* if no PDU was received and connection timeout was set and over the limit */
-            if (ret == 0 && smpp->connection_timeout > 0 &&
-                difftime(time(NULL), last_response) > smpp->connection_timeout) {
-                error(0, "SMPP[%s]: No responses from SMSC within %ld sec. Reconnecting.",
-                         octstr_get_cstr(smpp->conn->id), smpp->connection_timeout);
+            
+            /* send enquire link, only if connection is active */
+            if (IS_ACTIVE && send_enquire_link(smpp, conn, &last_enquire_sent) == -1)
                 break;
-            }
-
-
+            
             /* cleanup sent queue */
             if (transmitter && difftime(time(NULL), last_cleanup) > smpp->wait_ack) {
                 if (do_queue_cleanup(smpp, &pending_submits))
                     break; /* reconnect */
                 time(&last_cleanup);
             }
-
+            
+            /* make sure we send */
             if (transmitter && difftime(time(NULL), smpp->throttling_err_time) > SMPP_THROTTLING_SLEEP_TIME) {
                 smpp->throttling_err_time = 0;
-                send_messages(smpp, conn, &pending_submits);
+                if (send_messages(smpp, conn, &pending_submits) == -1)
+                    break;
+            }
+            
+            /* unbind
+             * Read so long as unbind_resp received or timeout passed. Otherwise we have
+             * double delivered messages.
+             */
+            if (smpp->quitting) {
+                if (!IS_ACTIVE || send_unbind(smpp, conn) == -1)
+                    break;
+                time(&last_response);
+                while(conn_wait(conn, 1.00) != -1 && IS_ACTIVE &&
+                      difftime(time(NULL), last_response) < SMPP_DEFAULT_SHUTDOWN_TIMEOUT) {
+                    if (read_pdu(smpp, conn, &len, &pdu) == 1) {
+                        dump_pdu("Got PDU:", smpp->conn->id, pdu);
+                        handle_pdu(smpp, conn, pdu, &pending_submits);
+                        smpp_pdu_destroy(pdu);
+                    }
+                }
+                debug("bb.sms.smpp", 0, "SMPP[%s]: %s: break and shutting down",
+                      octstr_get_cstr(smpp->conn->id), __PRETTY_FUNCTION__);
+                
+                break;
             }
         }
 
@@ -1955,14 +2023,19 @@ static void io_thread(void *arg)
             conn_destroy(conn);
             conn = NULL;
         }
-        /* set reconnect status */
-        mutex_lock(smpp->conn->flow_mutex);
-        smpp->conn->status = SMSCCONN_RECONNECTING;
-        mutex_unlock(smpp->conn->flow_mutex);
+        /* set reconnecting status first so that core don't put msgs into our queue */
+        if (!smpp->quitting) {
+            error(0, "SMPP[%s]: Couldn't connect to SMS center (retrying in %ld seconds).",
+                  octstr_get_cstr(smpp->conn->id), smpp->conn->reconnect_delay);
+            mutex_lock(smpp->conn->flow_mutex);
+            smpp->conn->status = SMSCCONN_RECONNECTING;
+            mutex_unlock(smpp->conn->flow_mutex);
+            gwthread_sleep(smpp->conn->reconnect_delay);
+        }
         /*
-         * Put all queued messages back into global queue,so if
-         * we have another link running then messages will be delivered
-         * quickly.
+         * put all queued messages back into global queue,so if
+         * we have another link running than messages will be delivered
+         * quickly
          */
         if (transmitter) {
             Msg *msg;
@@ -1978,7 +2051,7 @@ static void io_thread(void *arg)
             noresp = dict_keys(smpp->sent_msgs);
             while((key = gwlist_extract_first(noresp)) != NULL) {
                 smpp_msg = dict_remove(smpp->sent_msgs, key);
-                if (smpp_msg != NULL && smpp_msg->msg) {
+                if (smpp_msg != NULL) {
                     bb_smscconn_send_failed(smpp->conn, smpp_msg->msg, reason, NULL);
                     smpp_msg_destroy(smpp_msg, 0);
                 }
@@ -1986,18 +2059,31 @@ static void io_thread(void *arg)
             }
             gwlist_destroy(noresp, NULL);
         }
-        /*
-         * Reconnect if that was a connection problem.
-         */
-        if (!smpp->quitting) {
-            error(0, "SMPP[%s]: Couldn't connect to SMS center (retrying in %ld seconds).",
-                  octstr_get_cstr(smpp->conn->id), smpp->conn->reconnect_delay);
-            gwthread_sleep(smpp->conn->reconnect_delay);
-        }
     }
-    mutex_lock(smpp->conn->flow_mutex);
-    smpp->conn->status = SMSCCONN_DEAD;
-    mutex_unlock(smpp->conn->flow_mutex);
+    
+#undef IS_ACTIVE
+    
+    /*
+     * Shutdown sequence as follow:
+     *    1) if this is TX session so join receiver and free SMPP
+     *    2) if RX session available but no TX session so nothing to join then free SMPP
+     */
+    if (transmitter && smpp->receiver != -1) {
+        gwthread_wakeup(smpp->receiver);
+        gwthread_join(smpp->receiver);
+    }
+    if (transmitter || smpp->transmitter == -1) {
+        debug("bb.smpp", 0, "SMSCConn %s shut down.",
+              octstr_get_cstr(smpp->conn->name));
+        
+        mutex_lock(smpp->conn->flow_mutex);
+        smpp->conn->status = SMSCCONN_DEAD;
+        smpp->conn->data = NULL;
+        mutex_unlock(smpp->conn->flow_mutex);
+        
+        smpp_destroy(smpp);
+        bb_smscconn_killed();
+    }
 }
 
 
@@ -2032,30 +2118,32 @@ static int shutdown_cb(SMSCConn *conn, int finish_sending)
 {
     SMPP *smpp;
 
+    if (conn == NULL)
+        return -1;
+
     debug("bb.smpp", 0, "Shutting down SMSCConn %s (%s)",
           octstr_get_cstr(conn->name),
           finish_sending ? "slow" : "instant");
 
-    conn->why_killed = SMSCCONN_KILLED_SHUTDOWN;
+    mutex_lock(conn->flow_mutex);
 
-    /* XXX implement finish_sending */
+    conn->why_killed = SMSCCONN_KILLED_SHUTDOWN;
 
     smpp = conn->data;
-    smpp->quitting = 1;
-    if (smpp->transmitter != -1) {
-        gwthread_wakeup(smpp->transmitter);
-        gwthread_join(smpp->transmitter);
-    }
-    if (smpp->receiver != -1) {
-        gwthread_wakeup(smpp->receiver);
-        gwthread_join(smpp->receiver);
+    if (smpp == NULL) {
+        mutex_unlock(conn->flow_mutex);
+        return 0;
     }
-    smpp_destroy(smpp);
 
-    debug("bb.smpp", 0, "SMSCConn %s shut down.",
-          octstr_get_cstr(conn->name));
-    conn->status = SMSCCONN_DEAD;
-    bb_smscconn_killed();
+    smpp->quitting = 1;
+    if  (smpp->transmitter != -1)
+    	gwthread_wakeup(smpp->transmitter);
+
+    if (smpp->receiver != -1)
+    	gwthread_wakeup(smpp->receiver);
+
+    mutex_unlock(conn->flow_mutex);
+
     return 0;
 }