Re: outgoing queue limit in bearbox
Nick Mahilani <[email protected]>
| Newsgroups | gmane.comp.mobile.kannel.devel |
|---|---|
| Message-ID | <CAFtE-a31BE9yjVSsBEm_=qFumEhKfEnWxMpRYxO5hYS20VODEQ@mail.gmail.com> |
Thanks Amin ! While this approach works it is not very scalable if we have more operators in the future. As an alternative, I have defined a configurable in kannel config to bypass retries using the outgoing_sms queue. When turned on, this option disables queuing any messages to the 'outgoing_sms' queue and sends back a DLR NACK for any messages that fail for any reason(QFULL or submit_sm_resp with error from SMSC). This essentially moves the retry logic from kannel to the application using kannel and gives the application more control of the retry logic. This also avoids one saturated bind from affecting other binds. I wanted to get some feedback and if you see any potential problems with this approach. I have attached a diff for this change. Please review. -Nick Hi, I am using kannel with multiple SMSCs in my org and we often see that high load on one of the binds impacts the traffic on the other binds. After looking at the code, I see that the outgoing_sms queue is used as a global retry queue and if high load is directed to one of the binds, this fills up the outgoing_sms queue and causes 503 for the other sms traffic going through other binds.. As a potential fix, I have defined a configurable in kannel config to bypass retries using the outgoing_sms queue. When turned on, this option disables queuing any messages to the 'outgoing_sms' queue and sends back a DLR NACK for any messages that fail due to QFULL conditions or other cases that would otherwise enqueue to the 'outgoing_sms' queue. This essentially moves the retry logic from kannel to the application using kannel.. I wanted to get some feedback on if you see any potential problems with this approach. I can forward a diff if someone could review the diff and provide feedback. thanks, Nick On Wed, Nov 21, 2012 at 11:01 PM, Amin Mukhaimer < [email protected]> wrote: > Yes of course, just make different config files for each and run an > instance for each, like**** > > ** ** > > bearerbox operator1_config**** > > smsbox operator1_config**** > > ** ** > > bearerbox operator2_config**** > > smsbox operator2_config**** > > ** ** > > make sure they have different ports/spool directory/database tables/ log > files… and good luck**** > > ** ** > > Amin**** > > ** ** > > *From:* Nick Mahilani [mailto:[email protected]] > *Sent:* Wednesday, November 21, 2012 7:00 PM > *To:* [email protected] > *Cc:* [email protected] > *Subject:* RE: outgoing queue limit in bearbox**** > > ** ** > > Thanks Amin for your suggestion...**** > > Can we run multiple kannel instances on the same box?**** > > > Amin Mukhaimer <[email protected]> wrote:**** > > I had a somewhat similar problem is that it tends to get slow when sending > high amounts of SMSs to specific SMSC, and SMSs sent to other SMSCs are > delayed for a while, so I decided to run multiple Kannels, one for each > operator, hopping that would increase overall performance.**** > > **** > > I don’t know if that helps, but good luck.**** > > **** > > *From:* [email protected] [mailto:[email protected]<[email protected]>] > *On Behalf Of *Nick Mahilani > *Sent:* Wednesday, November 21, 2012 1:16 AM > *To:* [email protected] > *Subject:* outgoing queue limit in bearbox**** > > **** > > Hi,**** > > I have an application which is using Kannel to send outgoing sms via > multiple SMSC's. However, there is an insane volume of messages that needs > to go through one of the SMSC which is impacting the other binds due to the > global outgoing queue limit check in the code.**** > > **** > > if (max_outgoing_sms_qlength <http://doxygen.kannel.org/d5/d27/bb__smscconn_8c.html#a8> > 0 && !resend <http://doxygen.kannel.org/d6/d9f/wtp__init_8h.html#a33a30> &&**** > > 01091 queue_length > gwlist_len <http://doxygen.kannel.org/da/d23/list_8h.html#a6>(smsc_list) * max_outgoing_sms_qlength) {**** > > 01092 gw_rwlock_unlock <http://doxygen.kannel.org/d2/dbd/gw-rwlock_8h.html#a4>(&smsc_list_lock);**** > > 01093 debug <http://doxygen.kannel.org/d7/d7f/log_8h.html#a14>("bb.sms", 0, "sum(#queues) limit");**** > > 01094 return SMSCCONN_FAILED_QFULL;**** > > 01095 }**** > > I am very new to the kannel codebase so wanted to get some input on how > complicated is the fix to isolate the queue limits to each bind/SMSC. I > would like the other binds to not get affected by high volume of traffic on > one of the binds. So if one smsc queue is backed up, it does not impact the > messages sent to other SMSC's.**** > > Also, what is recommended outgoing queue limit value in terms of messages > per second being sent through Kannel?**** > > **** > > thanks,**** > > Nick**** >
disable_outgoing_sms_retry.diff
(application/octet-stream, 11.9 KB)
diff --git a/gw/bb_smscconn.c b/gw/bb_smscconn.c
index d9da883..4a968ee 100644
--- a/gw/bb_smscconn.c
+++ b/gw/bb_smscconn.c
@@ -119,6 +119,7 @@ extern long max_outgoing_sms_qlength;
/* incoming sms queue control */
extern long max_incoming_sms_qlength;
+extern int sms_retries_disabled;
/* configuration filename */
extern Octstr *cfg_filename;
@@ -197,8 +198,31 @@ void bb_smscconn_killed(void)
gwlist_remove_producer(flow_threads);
}
+/* Helper function to send dlr nack on SMSC connection */
+void send_dlr_nack(SMSCConn *conn, Msg *sms, Octstr *reply)
+{
+ /* write NACK to store file */
+ store_save_ack(sms, ack_failed);
+ if (sms == NULL) return;
+ if (conn) counter_increase(conn->failed);
+ /* generate relay confirmancy message */
+ if (DLR_IS_SMSC_FAIL(sms->sms.dlr_mask) ||
+ DLR_IS_FAIL(sms->sms.dlr_mask)) {
+ Msg *dlrmsg;
+
+ if (reply == NULL)
+ reply = octstr_create("");
+
+ octstr_insert_data(reply, 0, "NACK/", 5);
+ dlrmsg = create_dlr_from_msg((conn ? (conn->id?conn->id:conn->name) : NULL), sms,
+ reply, DLR_SMSC_FAIL);
+ if (dlrmsg != NULL) {
+ bb_smscconn_receive(conn, dlrmsg);
+ }
+ }
+}
-static void handle_split(SMSCConn *conn, Msg *msg, long reason)
+static void handle_split(SMSCConn *conn, Msg *msg, long reason, Octstr *reply)
{
struct split_parts *split = msg->sms.split_parts;
@@ -209,6 +233,16 @@ static void handle_split(SMSCConn *conn, Msg *msg, long reason)
*/
switch(reason) {
case SMSCCONN_FAILED_TEMPORARILY:
+ if (sms_retries_disabled) {
+ if (reply == NULL) {
+ reply = octstr_create("/Retries disabled");
+ } else {
+ octstr_append_cstr(reply, "/Retries disabled");
+ }
+ bb_alog_sms(conn, msg, "DISCARDED SMS");
+ send_dlr_nack(conn, msg, reply);
+ break;
+ }
/*
* Check if SMSC link alive and if so increase resend_try and set resend_time.
* If SMSC link is not active don't increase resend_try and don't set resend_time
@@ -271,7 +305,7 @@ static void handle_split(SMSCConn *conn, Msg *msg, long reason)
void bb_smscconn_sent(SMSCConn *conn, Msg *sms, Octstr *reply)
{
if (sms->sms.split_parts != NULL) {
- handle_split(conn, sms, SMSCCONN_SUCCESS);
+ handle_split(conn, sms, SMSCCONN_SUCCESS, reply);
octstr_destroy(reply);
return;
}
@@ -310,17 +344,24 @@ void bb_smscconn_sent(SMSCConn *conn, Msg *sms, Octstr *reply)
octstr_destroy(reply);
}
-
void bb_smscconn_send_failed(SMSCConn *conn, Msg *sms, int reason, Octstr *reply)
{
if (sms->sms.split_parts != NULL) {
- handle_split(conn, sms, reason);
+ handle_split(conn, sms, reason, reply);
octstr_destroy(reply);
return;
}
-
+
+ Octstr *reply_copy = (reply) ? octstr_duplicate(reply) : octstr_create("");
+
switch (reason) {
case SMSCCONN_FAILED_TEMPORARILY:
+ debug("bb.sms", 0, "SMSCCONN_FAILED_TEMPORARILY: sms_retries_disabled: %d", sms_retries_disabled);
+ if (sms_retries_disabled) {
+ octstr_append_cstr(reply_copy, "/Retries disabled");
+ bb_smscconn_send_failed(conn, sms, SMSCCONN_FAILED_DISCARDED, reply_copy);
+ break;
+ }
/*
* Check if SMSC link alive and if so increase resend_try and set resend_time.
* If SMSC link is not active don't increase resend_try and don't set resend_time
@@ -333,50 +374,36 @@ void bb_smscconn_send_failed(SMSCConn *conn, Msg *sms, int reason, Octstr *reply
*/
if (sms_resend_retry >= 0 && sms->sms.resend_try >= sms_resend_retry) {
warning(0, "Maximum retries for message exceeded, discarding it!");
- bb_smscconn_send_failed(NULL, sms, SMSCCONN_FAILED_DISCARDED,
+ bb_smscconn_send_failed(conn, sms, SMSCCONN_FAILED_DISCARDED,
octstr_create("Retries Exceeded"));
break;
}
sms->sms.resend_try = (sms->sms.resend_try > 0 ? sms->sms.resend_try + 1 : 1);
time(&sms->sms.resend_time);
}
- gwlist_produce(outgoing_sms, sms);
- break;
+ gwlist_produce(outgoing_sms, sms);
+ break;
case SMSCCONN_FAILED_SHUTDOWN:
+ debug("bb.sms", 0, "SMSCCONN_FAILED_SHUTDOWN: sms_retries_disabled: %d", sms_retries_disabled);
+ if (sms_retries_disabled) {
+ octstr_append_cstr(reply_copy, "/Retries disabled");
+ bb_smscconn_send_failed(conn, sms, SMSCCONN_FAILED_DISCARDED, reply_copy);
+ break;
+ }
gwlist_produce(outgoing_sms, sms);
break;
default:
- /* write NACK to store file */
- store_save_ack(sms, ack_failed);
-
- if (conn) counter_increase(conn->failed);
- if (reason == SMSCCONN_FAILED_DISCARDED)
- bb_alog_sms(conn, sms, "DISCARDED SMS");
- else
- bb_alog_sms(conn, sms, "FAILED Send SMS");
-
- /* generate relay confirmancy message */
- if (DLR_IS_SMSC_FAIL(sms->sms.dlr_mask) ||
- DLR_IS_FAIL(sms->sms.dlr_mask)) {
- Msg *dlrmsg;
-
- if (reply == NULL)
- reply = octstr_create("");
-
- octstr_insert_data(reply, 0, "NACK/", 5);
- dlrmsg = create_dlr_from_msg((conn ? (conn->id?conn->id:conn->name) : NULL), sms,
- reply, DLR_SMSC_FAIL);
- if (dlrmsg != NULL) {
- bb_smscconn_receive(conn, dlrmsg);
- }
- }
-
+ if (reason == SMSCCONN_FAILED_DISCARDED)
+ bb_alog_sms(conn, sms, "DISCARDED SMS");
+ else
+ bb_alog_sms(conn, sms, "FAILED Send SMS");
+ send_dlr_nack(conn, sms, reply);
+ octstr_destroy(reply_copy);
msg_destroy(sms);
break;
}
-
octstr_destroy(reply);
}
@@ -528,7 +555,6 @@ long bb_smscconn_receive(SMSCConn *conn, Msg *sms)
}
msg_destroy(sms);
-
return SMSCCONN_SUCCESS;
}
@@ -573,7 +599,7 @@ static void sms_router(void *arg)
while(bb_status != BB_SHUTDOWN && bb_status != BB_DEAD) {
- if (newmsg == startmsg) {
+ if (newmsg == startmsg) {
if (ret == SMSCCONN_QUEUED || ret == SMSCCONN_FAILED_QFULL) {
/* sleep: sms_resend_frequency / 2 , so we reduce amount of msgs to send */
double sleep_time = (sms_resend_frequency / 2 > 1 ? sms_resend_frequency / 2 : sms_resend_frequency);
@@ -598,8 +624,16 @@ static void sms_router(void *arg)
continue;
}
- debug("bb.sms", 0, "sms_router: handling message (%p vs %p)",
- msg, startmsg);
+ if (sms_retries_disabled) {
+ /* we should not have any messages in the outgoing_sms queue if sms_retries_disabled is set */
+ error(0, "unexpected message in outgoing_sms queue when sms_retries_disabled set");
+ newmsg = startmsg = NULL;
+ msg_destroy(msg);
+ continue;
+ }
+
+ debug("bb.sms", 0, "sms_router: handling message (%p vs %p) outgoing_sms len: %ld",
+ msg, startmsg, gwlist_len(outgoing_sms));
/* handle delayed msgs */
if (msg->sms.resend_try > 0 && difftime(time(NULL), msg->sms.resend_time) < sms_resend_frequency &&
@@ -620,6 +654,7 @@ static void sms_router(void *arg)
debug("bb.sms", 0, "Routing failed, re-queued.");
break;
case SMSCCONN_FAILED_DISCARDED:
+ debug("bb.sms", 0, "discard message (len: %ld)", gwlist_len(outgoing_sms));
msg_destroy(msg);
newmsg = startmsg = NULL;
break;
@@ -1280,7 +1315,8 @@ long smsc2_rout(Msg *msg, int resend)
if (max_outgoing_sms_qlength > 0 && !resend &&
queue_length > gwlist_len(smsc_list) * max_outgoing_sms_qlength) {
gw_rwlock_unlock(&smsc_list_lock);
- debug("bb.sms", 0, "sum(#queues) limit");
+ debug("bb.sms", 0, "sum(#queues) limit, outgoing_sms len: %ld, queue_length: %ld, max: %ld",
+ gwlist_len(outgoing_sms), queue_length, gwlist_len(smsc_list) * max_outgoing_sms_qlength);
return SMSCCONN_FAILED_QFULL;
}
} else {
@@ -1297,8 +1333,11 @@ long smsc2_rout(Msg *msg, int resend)
else if (bad_found) {
gw_rwlock_unlock(&smsc_list_lock);
if (max_outgoing_sms_qlength < 0 || gwlist_len(outgoing_sms) < max_outgoing_sms_qlength) {
- gwlist_produce(outgoing_sms, msg);
- return SMSCCONN_QUEUED;
+ if (!sms_retries_disabled) {
+ debug("bb.sms", 0, "smsc2_rout: bad_found: adding to outgoing_sms queue(len: %ld)", gwlist_len(outgoing_sms));
+ gwlist_produce(outgoing_sms, msg);
+ return SMSCCONN_QUEUED;
+ }
}
debug("bb.sms", 0, "bad_found queue full");
return SMSCCONN_FAILED_QFULL; /* queue full */
@@ -1320,8 +1359,10 @@ long smsc2_rout(Msg *msg, int resend)
gw_rwlock_unlock(&smsc_list_lock);
/* check the status of sending operation */
- if (ret == -1)
+ if (ret == -1) {
+ debug("bb.sms", 0, "resending message");
return smsc2_rout(msg, resend); /* re-try */
+ }
msg_destroy(msg);
return SMSCCONN_SUCCESS;
diff --git a/gw/bearerbox.c b/gw/bearerbox.c
index a624da9..99e1704 100644
--- a/gw/bearerbox.c
+++ b/gw/bearerbox.c
@@ -97,6 +97,11 @@ Counter *outgoing_wdp_counter;
/* incoming/outgoing sms queue control */
long max_incoming_sms_qlength;
long max_outgoing_sms_qlength;
+/* Flag to control use of global outgoing_sms queue for retries inside Kannel
+ * 0 - enqueue all messages that failed and should be retried to outgoing_sms
+ * 1 - do not enqueue failed messages to outgoing-sms queue and generate a DLR NACK back to sms sender
+ */
+int sms_retries_disabled;
Load *outgoing_sms_load;
@@ -519,6 +524,12 @@ static Cfg *init_bearerbox(Cfg *cfg)
if (max_outgoing_sms_qlength < 0)
max_outgoing_sms_qlength = DEFAULT_OUTGOING_SMS_QLENGTH;
+
+ if (cfg_get_bool(&sms_retries_disabled, grp,
+ octstr_imm("sms-retries-disabled")) == -1) {
+ sms_retries_disabled = DEFAULT_SMS_RETRIES_DISABLED;
+ }
+ warning(0, "sms_retries_disabled set to %d", sms_retries_disabled);
if (cfg_get_integer(&value, grp, octstr_imm("http-timeout")) == 0)
http_set_client_timeout(value);
diff --git a/gw/bearerbox.h b/gw/bearerbox.h
index 26f56b7..5d9c766 100644
--- a/gw/bearerbox.h
+++ b/gw/bearerbox.h
@@ -67,6 +67,8 @@
/* Default outgoing queue length */
#define DEFAULT_OUTGOING_SMS_QLENGTH 1000000
+/* Don't disable sms retries by default */
+#define DEFAULT_SMS_RETRIES_DISABLED 0
/* general bearerbox state */
diff --git a/gw/smsc/smsc_smpp.c b/gw/smsc/smsc_smpp.c
index 8966ab7..38b66fe 100644
--- a/gw/smsc/smsc_smpp.c
+++ b/gw/smsc/smsc_smpp.c
@@ -2060,8 +2060,11 @@ static void io_thread(void *arg)
long reason = (smpp->quitting?SMSCCONN_FAILED_SHUTDOWN:SMSCCONN_FAILED_TEMPORARILY);
- while((msg = gw_prioqueue_remove(smpp->msgs_to_send)) != NULL)
+ while((msg = gw_prioqueue_remove(smpp->msgs_to_send)) != NULL) {
+ debug("bb.sms.smpp", 0, "SMPP[%s]: %s: send failed message",
+ octstr_get_cstr(smpp->conn->id), __PRETTY_FUNCTION__);
bb_smscconn_send_failed(smpp->conn, msg, reason, NULL);
+ }
noresp = dict_keys(smpp->sent_msgs);
while((key = gwlist_extract_first(noresp)) != NULL) {
diff --git a/gwlib/cfg.def b/gwlib/cfg.def
index 0b48d9e..6f7fa35 100644
--- a/gwlib/cfg.def
+++ b/gwlib/cfg.def
@@ -125,6 +125,7 @@ SINGLE_GROUP(core,
OCTSTR(maximum-queue-length)
OCTSTR(sms-incoming-queue-limit)
OCTSTR(sms-outgoing-queue-limit)
+ OCTSTR(sms-retries-disabled)
OCTSTR(sms-resend-freq)
OCTSTR(sms-resend-retry)
OCTSTR(sms-combine-concatenated-mo)