main/stasis: Allow subscriptions to use a threadpool for message delivery

Prior to this patch, all Stasis subscriptions would receive a dedicated
thread for servicing published messages. In contrast, prior to r400178
(see review https://reviewboard.asterisk.org/r/2881/), the subscriptions
shared a thread pool. It was discovered during some initial work on Stasis
that, for a low subscription count with high message throughput, the
threadpool was not as performant as simply having a dedicated thread per
subscriber.

For situations where a subscriber receives a substantial number of messages
and is always present, the model of having a dedicated thread per subscriber
makes sense. While we still have plenty of subscriptions that would follow
this model, e.g., AMI, CDRs, CEL, etc., there are plenty that also fall into
the following two categories:
* Large number of subscriptions, specifically those tied to endpoints/peers.
* Low number of messages. Some subscriptions exist specifically to coordinate
  a single message - the subscription is created, a message is published, the
  delivery is synchronized, and the subscription is destroyed.
In both of the latter two cases, creating a dedicated thread is wasteful (and
in the case of a large number of peers/endpoints, harmful). In those cases,
having shared delivery threads is far more performant.

This patch adds the ability of a subscriber to Stasis to choose whether or not
their messages are dispatched on a dedicated thread or on a threadpool. The
threadpool is configurable through stasis.conf.

Review: https://reviewboard.asterisk.org/r/4193

ASTERISK-24533 #close
Reported by: xrobau
Tested by: xrobau
........

Merged revisions 428681 from http://svn.asterisk.org/svn/asterisk/branches/12


git-svn-id: https://origsvn.digium.com/svn/asterisk/branches/13@428687 65c4cc65-6c06-0410-ace0-fbb531ad65f3

6 files changed, 6 insertions, 6 deletions

diff --git a/channels/chan_dahdi.c b/channels/chan_dahdi.c
index 9ccec641f..8ef0cec30 100644
--- a/channels/chan_dahdi.c
+++ b/channels/chan_dahdi.c
@@ -12581,7 +12581,7 @@ static struct dahdi_pvt *mkintf(int channel, const struct dahdi_chan_conf *conf,
 
 			mailbox_specific_topic = ast_mwi_topic(tmp->mailbox);
 			if (mailbox_specific_topic) {
-				tmp->mwi_event_sub = stasis_subscribe(mailbox_specific_topic, mwi_event_cb, NULL);
+				tmp->mwi_event_sub = stasis_subscribe_pool(mailbox_specific_topic, mwi_event_cb, NULL);
 			}
 		}
 #ifdef HAVE_DAHDI_LINEREVERSE_VMWI
diff --git a/channels/chan_iax2.c b/channels/chan_iax2.c
index d093438c8..39861db0e 100644
--- a/channels/chan_iax2.c
+++ b/channels/chan_iax2.c
@@ -13096,7 +13096,7 @@ static struct iax2_peer *build_peer(const char *name, struct ast_variable *v, st
 
 		mailbox_specific_topic = ast_mwi_topic(peer->mailbox);
 		if (mailbox_specific_topic) {
-			peer->mwi_event_sub = stasis_subscribe(mailbox_specific_topic, mwi_event_cb, NULL);
+			peer->mwi_event_sub = stasis_subscribe_pool(mailbox_specific_topic, mwi_event_cb, NULL);
 		}
 	}
 
diff --git a/channels/chan_mgcp.c b/channels/chan_mgcp.c
index 53d7b42e5..08c4dc2e2 100644
--- a/channels/chan_mgcp.c
+++ b/channels/chan_mgcp.c
@@ -4237,7 +4237,7 @@ static struct mgcp_gateway *build_gateway(char *cat, struct ast_variable *v)
 
 					mailbox_specific_topic = ast_mwi_topic(e->mailbox);
 					if (mailbox_specific_topic) {
-						e->mwi_event_sub = stasis_subscribe(mailbox_specific_topic, mwi_event_cb, NULL);
+						e->mwi_event_sub = stasis_subscribe_pool(mailbox_specific_topic, mwi_event_cb, NULL);
 					}
 				}
 				snprintf(e->rqnt_ident, sizeof(e->rqnt_ident), "%08lx", (unsigned long)ast_random());
diff --git a/channels/chan_sip.c b/channels/chan_sip.c
index ea20f8d6d..ab03580ad 100644
--- a/channels/chan_sip.c
+++ b/channels/chan_sip.c
@@ -27245,7 +27245,7 @@ static void add_peer_mwi_subs(struct sip_peer *peer)
 			if (!peer_name) {
 				return;
 			}
-			mailbox->event_sub = stasis_subscribe(mailbox_specific_topic, mwi_event_cb, peer_name);
+			mailbox->event_sub = stasis_subscribe_pool(mailbox_specific_topic, mwi_event_cb, peer_name);
 		}
 	}
 }
diff --git a/channels/chan_skinny.c b/channels/chan_skinny.c
index 9c074ec6e..9269fbf6e 100644
--- a/channels/chan_skinny.c
+++ b/channels/chan_skinny.c
@@ -8295,7 +8295,7 @@ static struct skinny_line *config_line(const char *lname, struct ast_variable *v
 
 		mailbox_specific_topic = ast_mwi_topic(l->mailbox);
 		if (mailbox_specific_topic) {
-			l->mwi_event_sub = stasis_subscribe(mailbox_specific_topic, mwi_event_cb, l);
+			l->mwi_event_sub = stasis_subscribe_pool(mailbox_specific_topic, mwi_event_cb, l);
 		}
 	}
 
diff --git a/channels/sig_pri.c b/channels/sig_pri.c
index e9e17322f..a26b56611 100644
--- a/channels/sig_pri.c
+++ b/channels/sig_pri.c
@@ -9174,7 +9174,7 @@ int sig_pri_start_pri(struct sig_pri_span *pri)
 
 		mailbox_specific_topic = ast_mwi_topic(mbox_id);
 		if (mailbox_specific_topic) {
-			pri->mbox[i].sub = stasis_subscribe(mailbox_specific_topic, sig_pri_mwi_event_cb, pri);
+			pri->mbox[i].sub = stasis_subscribe_pool(mailbox_specific_topic, sig_pri_mwi_event_cb, pri);
 		}
 		if (!pri->mbox[i].sub) {
 			ast_log(LOG_ERROR, "%s span %d could not subscribe to MWI events for %s(%s).\n",