[tm] migrated to dynamic allocated branches per transaction

This allow a really high number of branches per transaction (like the default is 256, but can be increased). The branches are dynamically allocated, on demand, so some transactions which are not subjet to forking (typically non-INVITE) may have the minimum of 4 branches, while the INVITE transaction, depending on the need can scale up to 256

Author: bogdan-iancu

modules/tm/async.c

@@ -131,7 +131,7 @@ int t_resume_async_request(int fd, void*param, int was_timeout)
 	backup_list = set_avp_list( &t->user_avps );
 	/* set default send address to the saved value */
 	backup_si = bind_address;
-	bind_address = t->uac[0].request.dst.send_sock;
+	bind_address = TM_BRANCH( t, 0).request.dst.send_sock;
 
 	async_status = ASYNC_DONE; /* assume default status as done */
 	/* call the resume function in order to read and handle data */
@@ -260,7 +260,7 @@ int t_resume_async_reply(int fd, void*param, int was_timeout)
 	set_t( t );
 
 	msg_status=ctx->reply->REPLY_STATUS;
-	reply_uac=&t->uac[branch];
+	reply_uac=&TM_BRANCH( t, branch);
 	LM_DBG("org. status uas=%d, uac[%d]=%d local=%d is_invite=%d)\n",
 		t->uas.status, branch, reply_uac->last_received,
 		is_local(t), is_invite(t));

modules/tm/h_table.c

@@ -120,6 +120,7 @@ void free_cell( struct cell* dead_cell )
 	struct sip_msg *rpl;
 	struct totag_elem *tt, *foo;
 	struct proxy_l *p;
+	struct ua_client *uac;
 
 	if ( has_tran_tmcbs( dead_cell, TMCB_TRANS_DELETED) )
 		run_trans_callbacks( TMCB_TRANS_DELETED, dead_cell, 0, 0, 0);
@@ -142,17 +143,18 @@ void free_cell( struct cell* dead_cell )
 	/* UA Clients */
 	for ( i =0 ; i<dead_cell->nr_of_outgoings;  i++ )
 	{
+		uac = & TM_BRANCH( dead_cell, i);
 		/* retransmission buffer */
-		if ( (b=dead_cell->uac[i].request.buffer.s) )
+		if ( (b=uac->request.buffer.s) )
 			shm_free_bulk( b );
-		b=dead_cell->uac[i].local_cancel.buffer.s;
+		b=uac->local_cancel.buffer.s;
 		if (b!=0 && b!=BUSY_BUFFER)
 			shm_free_bulk( b );
-		rpl=dead_cell->uac[i].reply;
+		rpl=uac->reply;
 		if (rpl && rpl!=FAKED_REPLY && rpl->msg_flags&FL_SHM_CLONE) {
 			free_cloned_msg_unsafe( rpl );
 		}
-		if ( (p=dead_cell->uac[i].proxy)!=NULL ) {
+		if ( (p=uac->proxy)!=NULL ) {
 			if ( p->host.h_addr_list )
 				shm_free_bulk( p->host.h_addr_list );
 			if ( p->dn ) {
@@ -163,13 +165,16 @@ void free_cell( struct cell* dead_cell )
 			shm_free_bulk(p);
 		}
 
-		_clean_branch(dead_cell->uac[i],
-					shm_free_bulk, destroy_avp_list_bulk);
+		_clean_branch( *uac, shm_free_bulk, destroy_avp_list_bulk);
 
-		if (dead_cell->uac[i].on_reply)
-			shm_free_bulk(dead_cell->uac[i].on_reply);
+		if (uac->on_reply)
+			shm_free_bulk(uac->on_reply);
 	}
 
+	/* destroy the UAC arrays now */
+	for (i=0 ; i<TM_BRANCH_CHUNKS_NO && dead_cell->uac[i] ; i++ )
+		shm_free_bulk(dead_cell->uac[i]);
+
 	/* collected to tags */
 	tt=dead_cell->fwded_totags;
 	while(tt) {
@@ -239,7 +244,7 @@ static inline void init_branches(struct cell *t, unsigned int set)
 	unsigned int i;
 
 	for (i=0; i<MAX_BRANCHES; i++)
-		init_branch(&t->uac[i], i, set, t);
+		init_branch( &TM_BRANCH(t,i), i, set, t);
 }
 
 

modules/tm/h_table.h

@@ -51,6 +51,22 @@ struct cell;
 struct timer;
 struct retr_buf;
 
+
+/* be sure the MAX devides to CHUNK_SIZE, otherwise the behavor may be
+ * bogus (meaning crashes) */
+/* IMPORTANT - TM_BRANCH_MAX must be a multiple of 32, to align perfectly
+ * with the size of the bitmask (which is array of uint32) 
+ * Even more, this multiple of 32 guarantees the devision with 4, the 
+ * TM_BRANCH_CHUNK_SIZE (so, if you plan to change the chunk size, pick
+ * something that devides 32, like 16) */
+#define TM_BRANCH_MAX_FACTOR 8
+#define TM_BRANCH_MAX        (32*TM_BRANCH_MAX_FACTOR)
+#define TM_BRANCH_CHUNK_SIZE 4
+#define TM_BRANCH_CHUNKS_NO  (TM_BRANCH_MAX/TM_BRANCH_CHUNK_SIZE)
+
+#define TM_BRANCH( _t, _idx) \
+	( (_t)->uac[(_idx)/TM_BRANCH_CHUNK_SIZE][(_idx)%TM_BRANCH_CHUNK_SIZE] )
+
 #include "../../lib/container.h"
 #include "../../mem/shm_mem.h"
 #include "lock.h"
@@ -216,7 +232,6 @@ struct totag_elem {
 #define T_UAC_IS_PHONY        (1<<2)
 
 
-
 /* transaction context */
 
 typedef struct cell
@@ -274,7 +289,9 @@ typedef struct cell
 	/* UA Server */
 	struct ua_server  uas;
 	/* UA Clients */
-	struct ua_client  uac[ MAX_BRANCHES ];
+	/* each UAC pointer is a pointer to a
+	 * TM_BRANCH_CHUNK_SIZE array of UACs */
+	struct ua_client*  uac[TM_BRANCH_CHUNKS_NO];
 
 	/* protection against concurrent reply processing */
 	ser_lock_t   reply_mutex;

modules/tm/t_cancel.c

@@ -68,7 +68,7 @@ void cancel_uacs( struct cell *t, branch_bm_t cancel_bm )
 	for( i=0 ; i<t->nr_of_outgoings ; i++ )
 		if ( BRANCH_BM_TST_IDX( cancel_bm, i) ) {
 			/* any reply actually received on this branch */
-			if (t->uac[i].last_received!=0) {
+			if (TM_BRANCH(t,i).last_received!=0) {
 				/* send a cancel out */
 				cancel_branch(t, i);
 			} else {
@@ -77,7 +77,7 @@ void cancel_uacs( struct cell *t, branch_bm_t cancel_bm )
 				 * one out IF we receive later a reply on this branch, so let's
 				 * flag it for catching (and cancelling) such delaied replies
 				 */
-				t->uac[i].flags |= T_UAC_TO_CANCEL_FLAG;
+				TM_BRANCH(t,i).flags |= T_UAC_TO_CANCEL_FLAG;
 			}
 		}
 }
@@ -90,8 +90,8 @@ void cancel_branch( struct cell *t, int branch )
 	struct retr_buf *crb, *irb;
 	struct usr_avp **backup_list;
 
-	crb=&t->uac[branch].local_cancel;
-	irb=&t->uac[branch].request;
+	crb=&TM_BRANCH(t,branch).local_cancel;
+	irb=&TM_BRANCH(t,branch).request;
 
 #	ifdef EXTRA_DEBUG
 	if (crb->buffer.s!=0 && crb->buffer.s!=BUSY_BUFFER) {
@@ -124,10 +124,10 @@ void cancel_branch( struct cell *t, int branch )
 	}
 
 	LM_DBG("sending cancel...\n");
-	if (t->uac[branch].br_flags & tcp_no_new_conn_bflag)
+	if (TM_BRANCH(t,branch).br_flags & tcp_no_new_conn_bflag)
 		tcp_no_new_conn = 1;
 	backup_list = set_avp_list( &t->user_avps );
-	set_bavp_list(&t->uac[branch].user_avps);
+	set_bavp_list(&TM_BRANCH(t,branch).user_avps);
 	if (SEND_BUFFER( crb )==0) {
 		if ( has_tran_tmcbs( t, TMCB_MSG_SENT_OUT) ) {
 			set_extra_tmcb_params( &crb->buffer, &crb->dst);

modules/tm/t_cancel.h

@@ -55,15 +55,17 @@ char *build_cancel(struct cell *Trans,unsigned int branch,
 
 inline static short should_cancel_branch( struct cell *t, int b )
 {
+	struct ua_client* uac = & TM_BRANCH( t, b);
+
 	/* cancel only if provisional received and no one else
 	   attempted to cancel yet; also skip PHONY branches
 	   (they have no signaling, so no canceling */
-	if ( t->uac[b].local_cancel.buffer.s==NULL &&
-	(t->uac[b].flags & T_UAC_IS_PHONY)==0 ) {
-		if ( t->uac[b].last_received<200 ) {
+	if ( uac->local_cancel.buffer.s==NULL &&
+	(uac->flags & T_UAC_IS_PHONY)==0 ) {
+		if ( uac->last_received<200 ) {
 			/* we'll cancel -- label it so that no one else
 			(e.g. another 200 branch) will try to do the same */
-			t->uac[b].local_cancel.buffer.s=BUSY_BUFFER;
+			uac->local_cancel.buffer.s=BUSY_BUFFER;
 			return 1;
 		}
 	}