Message lumps are saved to shared memory when t_relay() is called

the first time, instead of saving them by t_newtran(). The advantage is
that the SIP msg modifications that are made after t_newtran() are saved
as well, and they are propagated to failure route.
t_save_lumps() function is introduced, it can be used to force saving
the lumps before t_relay(), check the doc for details.
Fixes SER-303

The change can be reverted by uncommenting #define POSTPONE_MSG_CLONING
in sip_msg.h.

modules/tm/doc/functions.xml

@@ -683,4 +683,55 @@ failure_route[0]{
 	    </programlisting>
 	</example>
     </section>
+
+    <section id="t_save_lumps">
+	<title>
+	    <function>t_save_lumps()</function>
+	</title>
+	<para>
+		Forces the modifications of the processed SIP message
+		to be saved in shared memory before t_relay() is called.
+		The new branches which are created in failure_route will
+		contain the same modifications, and any other modification
+		after t_save_lumps() will be lost.
+	</para>
+	<para>
+		Note that t_relay() automatically saves the modifications
+		when it is called the first time, there is no need for
+		t_save_lumps() unless message changes between t_save_lumps()
+		and t_relay() must not be propagated to failure_route.
+	</para>
+	<para>
+		The transaction must be created by t_newtran() before
+		calling t_save_lumps().
+	</para>
+	<example>
+	    <title><function>t_save_lumps()</function> usage</title>
+	    <programlisting>
+route {
+	...
+	t_newtran();
+	append_hf("hf1: my first header\r\n");
+	...
+	t_save_lumps();
+	append_hf("hf2: my second header\r\n");
+	...
+	t_on_failure("1");
+	t_relay();
+}
+
+failure_route[1] {
+	append_branch();
+	append_hf("hf3: my third header\r\n");
+	#
+	# This branch contains hf1 and hf3, but does
+	# not contain hf2 header.
+	# hf2 would be also present here without
+	# t_save_lumps().
+	...
+	t_relay();
+}
+	    </programlisting>
+	</example>
+    </section>
 </section>

modules/tm/h_table.c

@@ -309,9 +309,13 @@ struct cell*  build_cell( struct sip_msg* p_msg )
 			run_reqin_callbacks( new_cell, p_msg, p_msg->REQ_METHOD);
 
 	if (p_msg) {
+#ifndef POSTPONE_MSG_CLONING
+		/* it makes no sense to clean the lumps when they are not cloned (Miklos) */
+
 		/* clean possible previous added vias/clen header or else they would 
 		 * get propagated in the failure routes */
 		free_via_clen_lump(&p_msg->add_rm);
+#endif
 		new_cell->uas.request = sip_msg_cloner(p_msg,&sip_msg_len);
 		if (!new_cell->uas.request)
 			goto error;

modules/tm/sip_msg.c

@@ -49,6 +49,9 @@
  *  2006-04-20  via->comp is also translated (andrei)
  *  2006-10-16  HDR_{PROXY,WWW}_AUTHENTICATE_T cloned (andrei)
  *  2007-01-26  HDR_DATE_T, HDR_IDENTITY_T, HDR_IDENTITY_INFO_T added (gergo)
+ *  2007-09-05  A separate memory block is allocated for the lumps
+ *              in case of requests in order to allow cloning them
+ *              later than the SIP msg. (Miklos)
  */
 
 #include "defs.h"
@@ -63,6 +66,11 @@
 #include "../../ut.h"
 #include "../../parser/digest/digest.h"
 
+#ifdef POSTPONE_MSG_CLONING
+#include "../../atomic_ops.h"
+#include "fix_lumps.h"
+#endif
+
 
 /* rounds to the first 4 byte multiple on 32 bit archs
  * and to the first 8 byte multiple on 64 bit archs */
@@ -84,8 +92,91 @@
 			(_ptr)+=ROUND4((_old)->len);}\
 	}
 
+/* length of the data lump structures */
+#define LUMP_LIST_LEN(len, list) \
+do { \
+        struct lump* tmp, *chain; \
+	chain = (list); \
+	while (chain) \
+	{ \
+		(len) += lump_len(chain); \
+		tmp = chain->before; \
+		while ( tmp ) \
+		{ \
+			(len) += lump_len( tmp ); \
+			tmp = tmp->before; \
+		} \
+		tmp = chain->after; \
+		while ( tmp ) \
+		{ \
+			(len) += lump_len( tmp ); \
+			tmp = tmp->after; \
+		} \
+		chain = chain->next; \
+	} \
+} while(0);
+
+/* length of the reply lump structure */
+#define RPL_LUMP_LIST_LEN(len, list) \
+do { \
+	struct lump_rpl* rpl_lump; \
+	for(rpl_lump=(list);rpl_lump;rpl_lump=rpl_lump->next) \
+		(len)+=ROUND4(sizeof(struct lump_rpl))+ROUND4(rpl_lump->text.len); \
+} while(0);
 
+/* clones data lumps */
+#define CLONE_LUMP_LIST(anchor, list, _ptr) \
+do { \
+	struct lump* lump_tmp, *l; \
+	struct lump** lump_anchor2, **a; \
+	a = (anchor); \
+	l = (list); \
+	while (l) \
+	{ \
+		lump_clone( (*a) , l , (_ptr) ); \
+		/*before list*/ \
+		lump_tmp = l->before; \
+		lump_anchor2 = &((*a)->before); \
+		while ( lump_tmp ) \
+		{ \
+			lump_clone( (*lump_anchor2) , lump_tmp , (_ptr) ); \
+			lump_anchor2 = &((*lump_anchor2)->before); \
+			lump_tmp = lump_tmp->before; \
+		} \
+		/*after list*/ \
+		lump_tmp = l->after; \
+		lump_anchor2 = &((*a)->after); \
+		while ( lump_tmp ) \
+		{ \
+			lump_clone( (*lump_anchor2) , lump_tmp , (_ptr) ); \
+			lump_anchor2 = &((*lump_anchor2)->after); \
+			lump_tmp = lump_tmp->after; \
+		} \
+		a = &((*a)->next); \
+		l = l->next; \
+	} \
+} while(0)
 
+/* clones reply lumps */
+#define CLONE_RPL_LUMP_LIST(anchor, list, _ptr) \
+do { \
+	struct lump_rpl* rpl_lump; \
+	struct lump_rpl** rpl_lump_anchor; \
+	rpl_lump_anchor = (anchor); \
+	for(rpl_lump=(list);rpl_lump;rpl_lump=rpl_lump->next) \
+	{ \
+		*(rpl_lump_anchor)=(struct lump_rpl*)(_ptr); \
+		(_ptr)+=ROUND4(sizeof( struct lump_rpl )); \
+		(*rpl_lump_anchor)->flags = LUMP_RPL_SHMEM | \
+			(rpl_lump->flags&(~(LUMP_RPL_NODUP|LUMP_RPL_NOFREE))); \
+		(*rpl_lump_anchor)->text.len = rpl_lump->text.len; \
+		(*rpl_lump_anchor)->text.s=(_ptr); \
+		(_ptr)+=ROUND4(rpl_lump->text.len); \
+		memcpy((*rpl_lump_anchor)->text.s,rpl_lump->text.s,rpl_lump->text.len); \
+		(*rpl_lump_anchor)->next=0; \
+		rpl_lump_anchor = &((*rpl_lump_anchor)->next); \
+	} \
+} while (0)
 
 inline struct via_body* via_body_cloner( char* new_buf,
 					char *org_buf, struct via_body *param_org_via, char **p)
@@ -295,7 +386,6 @@ struct sip_msg*  sip_msg_cloner( struct sip_msg *org_msg, int *sip_msg_len )
 	struct via_param  *prm;
 	struct to_param   *to_prm,*new_to_prm;
 	struct sip_msg    *new_msg;
-	struct lump_rpl   *rpl_lump, **rpl_lump_anchor;
 	char              *p;
 
 	/*computing the length of entire sip_msg structure*/
@@ -401,36 +491,18 @@ struct sip_msg*  sip_msg_cloner( struct sip_msg *org_msg, int *sip_msg_len )
 		}/*switch*/
 	}/*for all headers*/
 
-	/* length of the data lump structures */
-#define LUMP_LIST_LEN(len, list) \
-do { \
-        struct lump* tmp, *chain; \
-	chain = (list); \
-	while (chain) \
-	{ \
-		(len) += lump_len(chain); \
-		tmp = chain->before; \
-		while ( tmp ) \
-		{ \
-			(len) += lump_len( tmp ); \
-			tmp = tmp->before; \
-		} \
-		tmp = chain->after; \
-		while ( tmp ) \
-		{ \
-			(len) += lump_len( tmp ); \
-			tmp = tmp->after; \
-		} \
-		chain = chain->next; \
-	} \
-} while(0);
-
-	LUMP_LIST_LEN(len, org_msg->add_rm);
-	LUMP_LIST_LEN(len, org_msg->body_lumps);
+#ifdef POSTPONE_MSG_CLONING
+	/* take care of the lumps only for replies if the msg cloning is postponed */
+	if (org_msg->first_line.type==SIP_REPLY) {
+#endif
+		/* calculate the length of the data and reply lump structures */
+		LUMP_LIST_LEN(len, org_msg->add_rm);
+		LUMP_LIST_LEN(len, org_msg->body_lumps);
+		RPL_LUMP_LIST_LEN(len, org_msg->reply_lump);
 
-	/*length of reply lump structures*/
-	for(rpl_lump=org_msg->reply_lump;rpl_lump;rpl_lump=rpl_lump->next)
-			len+=ROUND4(sizeof(struct lump_rpl))+ROUND4(rpl_lump->text.len);
+#ifdef POSTPONE_MSG_CLONING
+	}
+#endif
 
 	p=(char *)shm_malloc(len);
 	if (!p)
@@ -450,6 +522,7 @@ do { \
 	p += ROUND4(sizeof(struct sip_msg));
 	new_msg->add_rm = 0;
 	new_msg->body_lumps = 0;
+	new_msg->reply_lump = 0;
 	/* new_uri */
 	if (org_msg->new_uri.s && org_msg->new_uri.len)
 	{
@@ -816,57 +889,18 @@ do { \
 		new_msg->last_header = last_hdr;
 	}
 
-	/* cloning data lump */
-#define CLONE_LUMP_LIST(anchor, list) \
-do { \
-	struct lump* lump_tmp, *l; \
-	struct lump** lump_anchor2, **a; \
-	a = (anchor); \
-	l = (list); \
-	while (l) \
-	{ \
-		lump_clone( (*a) , l , p ); \
-		/*before list*/ \
-		lump_tmp = l->before; \
-		lump_anchor2 = &((*a)->before); \
-		while ( lump_tmp ) \
-		{ \
-			lump_clone( (*lump_anchor2) , lump_tmp , p ); \
-			lump_anchor2 = &((*lump_anchor2)->before); \
-			lump_tmp = lump_tmp->before; \
-		} \
-		/*after list*/ \
-		lump_tmp = l->after; \
-		lump_anchor2 = &((*a)->after); \
-		while ( lump_tmp ) \
-		{ \
-			lump_clone( (*lump_anchor2) , lump_tmp , p ); \
-			lump_anchor2 = &((*lump_anchor2)->after); \
-			lump_tmp = lump_tmp->after; \
-		} \
-		a = &((*a)->next); \
-		l = l->next; \
-	} \
-} while(0)
-
-	CLONE_LUMP_LIST(&(new_msg->add_rm), org_msg->add_rm);
-	CLONE_LUMP_LIST(&(new_msg->body_lumps), org_msg->body_lumps);
+#ifdef POSTPONE_MSG_CLONING
+	/* take care of the lumps only for replies if the msg cloning is postponed */
+	if (org_msg->first_line.type==SIP_REPLY) {
+#endif
+		/*cloning data and reply lump structures*/
+		CLONE_LUMP_LIST(&(new_msg->add_rm), org_msg->add_rm, p);
+		CLONE_LUMP_LIST(&(new_msg->body_lumps), org_msg->body_lumps, p);
+		CLONE_RPL_LUMP_LIST(&(new_msg->reply_lump), org_msg->reply_lump, p);
 
-	/*cloning reply lump structures*/
-	rpl_lump_anchor = &(new_msg->reply_lump);
-	for(rpl_lump=org_msg->reply_lump;rpl_lump;rpl_lump=rpl_lump->next)
-	{
-		*(rpl_lump_anchor)=(struct lump_rpl*)p;
-		p+=ROUND4(sizeof( struct lump_rpl ));
-		(*rpl_lump_anchor)->flags = LUMP_RPL_SHMEM |
-			(rpl_lump->flags&(~(LUMP_RPL_NODUP|LUMP_RPL_NOFREE)));
-		(*rpl_lump_anchor)->text.len = rpl_lump->text.len;
-		(*rpl_lump_anchor)->text.s=p;
-		p+=ROUND4(rpl_lump->text.len);
-		memcpy((*rpl_lump_anchor)->text.s,rpl_lump->text.s,rpl_lump->text.len);
-		(*rpl_lump_anchor)->next=0;
-		rpl_lump_anchor = &((*rpl_lump_anchor)->next);
+#ifdef POSTPONE_MSG_CLONING
 	}
+#endif
 
 	if (clone_authorized_hooks(new_msg, org_msg) < 0) {
 		shm_free(new_msg);
@@ -875,3 +909,93 @@ do { \
 
 	return new_msg;
 }
+
+#ifdef POSTPONE_MSG_CLONING
+/* indicates wheter we have already cloned the msg lumps or not */
+unsigned char lumps_are_cloned = 0;
+
+/* clones the data and reply lumps from pkg_msg to shm_msg
+ * A new memory block is allocated for the lumps which is linked
+ * to shm_msg
+ *
+ * Note: the new memory block is linked to shm_msg->add_rm if
+ * at least one data lump is set, else it is linked to shm_msg->body_lumps
+ * if at least one body lump is set, otherwise it is linked to
+ * shm_msg->reply_lump
+ */
+static int msg_lump_cloner( struct sip_msg *shm_msg, struct sip_msg *pkg_msg)
+{
+	unsigned int	len;
+	char		*p;
+	struct lump	*add_rm, *body_lumps;
+	struct lump_rpl	*reply_lump;
+
+	/* calculate the length of the lumps */
+	len = 0;
+	LUMP_LIST_LEN(len, pkg_msg->add_rm);
+	LUMP_LIST_LEN(len, pkg_msg->body_lumps);
+	RPL_LUMP_LIST_LEN(len, pkg_msg->reply_lump);
+
+	if (!len)
+		return 0; /* nothing to do */
+
+	p=(char *)shm_malloc(len);
+	if (!p)
+	{
+		LOG(L_ERR, "ERROR: msg_lump_cloner: cannot allocate memory\n" );
+		return -1;
+	}
+
+	/* clone the lumps
+	 *
+	 * Better not to modify the lump structures of shm_msg directly
+	 * because no lock is held while they are read. We have to prepare
+	 * the lumps in separate lists, and fush the cache
+	 * with membar_write() before linking the lists to shm_msg.
+	 */
+	add_rm = body_lumps = 0;
+	reply_lump = 0;
+
+	CLONE_LUMP_LIST(&add_rm, pkg_msg->add_rm, p);
+	CLONE_LUMP_LIST(&body_lumps, pkg_msg->body_lumps, p);
+	CLONE_RPL_LUMP_LIST(&reply_lump, pkg_msg->reply_lump, p);
+	membar_write();
+
+	shm_msg->add_rm = add_rm;
+	shm_msg->body_lumps = body_lumps;
+	shm_msg->reply_lump = reply_lump;
+
+	return 0;	
+}
+
+/* wrapper function for msg_lump_cloner() with some additional sanity checks */
+int save_msg_lumps( struct sip_msg *shm_msg, struct sip_msg *pkg_msg)
+{
+	/* make sure that we do not clone the lumps twice */
+	if (lumps_are_cloned) {
+		LOG(L_DBG, "DEBUG: save_msg_lumps: lumps have been already cloned\n" );
+		return 0;
+	}
+	/* sanity checks */
+	if (unlikely(!shm_msg || ((shm_msg->msg_flags & FL_SHM_CLONE)==0))) {
+		LOG(L_ERR, "ERROR: save_msg_lumps: BUG, there is no shmem-ized message"
+			" (shm_msg=%p)\n", shm_msg);
+		return -1;
+	}
+	if (unlikely(shm_msg->first_line.type!=SIP_REQUEST)) {
+		LOG(L_ERR, "ERROR: save_msg_lumps: BUG, the function should be called only for requests\n" );
+		return -1;
+	}
+
+	/* needless to clone the lumps for ACK, they will not be used again */
+	if (shm_msg->REQ_METHOD == METHOD_ACK)
+		return 0;
+
+	/* clean possible previous added vias/clen header or else they would 
+	 * get propagated in the failure routes */
+	free_via_clen_lump(&pkg_msg->add_rm);
+
+	lumps_are_cloned = 1;
+	return msg_lump_cloner(shm_msg, pkg_msg);
+}
+#endif /* POSTPONE_MSG_CLONING */

modules/tm/sip_msg.h

@@ -36,11 +36,61 @@
 #include "../../parser/msg_parser.h"
 #include "../../mem/shm_mem.h"
 
-#define  sip_msg_free(_p_msg) shm_free( (_p_msg ))
-#define  sip_msg_free_unsafe(_p_msg) shm_free_unsafe( (_p_msg) )
+/* Allow postponing the cloning of SIP msg:
+ * t_newtran() copies the requests to shm mem without the lumps,
+ * and t_forward_nonack() clones the lumps later when it is called
+ * the first time.
+ * Replies use only one memory block.
+ */
+#define POSTPONE_MSG_CLONING
+
+#ifdef POSTPONE_MSG_CLONING
+#include "../../atomic_ops.h" /* membar_depends() */
+#endif
+
+#ifdef POSTPONE_MSG_CLONING
+	/* msg is a reply: one memory block was allocated
+	 * msg is a request: two memory blocks were allocated:
+	 *	- one for the sip_msg struct
+	 *	- another one for the lumps which is linked to
+	 *		add_rm, body_lumps, or reply_lump. 
+	 */
+#define  _sip_msg_free(_free_func, _p_msg) \
+		do{ \
+			if (_p_msg->first_line.type==SIP_REPLY) { \
+				_free_func( (_p_msg) ); \
+			} else { \
+				membar_depends(); \
+				if ((_p_msg)->add_rm) \
+					_free_func((_p_msg)->add_rm); \
+				else if ((_p_msg)->body_lumps) \
+					_free_func((_p_msg)->body_lumps); \
+				else if ((_p_msg)->reply_lump) \
+					_free_func((_p_msg)->reply_lump); \
+									  \
+				_free_func( (_p_msg) ); \
+			} \
+		}while(0)
+
+#else /* POSTPONE_MSG_CLONING */
+
+	/* only one memory block was allocated */
+#define  _sip_msg_free(_free_func, _p_msg) \
+		_free_func( (_p_msg) )
+
+#endif /* POSTPONE_MSG_CLONING */
+
+#define  sip_msg_free(_p_msg) _sip_msg_free(shm_free, _p_msg)
+#define  sip_msg_free_unsafe(_p_msg) _sip_msg_free(shm_free_unsafe, _p_msg)
 
 
 struct sip_msg*  sip_msg_cloner( struct sip_msg *org_msg, int *sip_msg_len );
 
+#ifdef POSTPONE_MSG_CLONING
+extern unsigned char lumps_are_cloned;
+
+int save_msg_lumps( struct sip_msg *shm_msg, struct sip_msg *pkg_msg);
+#endif
+
 
 #endif

modules/tm/t_fwd.c

@@ -115,6 +115,9 @@
 #include "../../dst_blacklist.h"
 #endif
 #include "../../select_buf.h" /* reset_static_buffer() */
+#ifdef POSTPONE_MSG_CLONING
+#include "../../atomic_ops.h" /* membar_depends() */
+#endif
 
 /* cancel hop by hop */
 #define E2E_CANCEL_HOP_BY_HOP
@@ -177,6 +180,11 @@ static char *print_uac_request( struct cell *t, struct sip_msg *i_req,
 		uri_backed_up=1;
 	}
 
+#ifdef POSTPONE_MSG_CLONING
+	/* lumps can be set outside of the lock, make sure that we read
+	 * the up-to-date values */
+	membar_depends();
+#endif
 	add_rm_backup = i_req->add_rm;
 	body_lumps_backup = i_req->body_lumps;
 	i_req->add_rm = dup_lump_list(i_req->add_rm);
@@ -629,6 +637,11 @@ int e2e_cancel_branch( struct sip_msg *cancel_msg, struct cell *t_cancel,
 	/* print */
 	if (cfg_get(tm, tm_cfg, reparse_invite)) {
 		/* buffer is built localy from the INVITE which was sent out */
+#ifdef POSTPONE_MSG_CLONING
+		/* lumps can be set outside of the lock, make sure that we read
+		 * the up-to-date values */
+		membar_depends();
+#endif
 		if (cancel_msg->add_rm || cancel_msg->body_lumps) {
 			LOG(L_WARN, "WARNING: e2e_cancel_branch: CANCEL is built locally, "
 			"thus lumps are not applied to the message!\n");
@@ -1021,6 +1034,15 @@ int t_forward_nonack( struct cell *t, struct sip_msg* p_msg ,
 	   is in additional branches (which may be continuously refilled
 	*/
 	if (first_branch==0) {
+#ifdef POSTPONE_MSG_CLONING
+		/* update the shmem-ized msg with the lumps */
+		if ((rmode == MODE_REQUEST) &&
+			save_msg_lumps(t->uas.request, p_msg)) {
+				LOG(L_ERR, "ERROR: t_forward_nonack: "
+					"failed to save the message lumps\n");
+				return -1;
+			}
+#endif
 		try_new=1;
 		branch_ret=add_uac( t, p_msg, GET_RURI(p_msg), GET_NEXT_HOP(p_msg),
 							proxy, proto );

modules/tm/tm.c

@@ -209,6 +209,7 @@ static int t_any_replied(struct sip_msg* msg, char*, char*);
 static int t_is_canceled(struct sip_msg* msg, char*, char*);
 static int t_grep_status(struct sip_msg* msg, char*, char*);
 static int w_t_drop_replies(struct sip_msg* msg, char* foo, char* bar);
+static int w_t_save_lumps(struct sip_msg* msg, char* foo, char* bar);
 
 
 /* by default the fr timers avps are not set, so that the avps won't be
@@ -319,6 +320,8 @@ static cmd_export_t cmds[]={
 			REQUEST_ROUTE|ONREPLY_ROUTE|FAILURE_ROUTE|BRANCH_ROUTE },
 	{"t_drop_replies",    w_t_drop_replies,         0, 0,
 			FAILURE_ROUTE},
+	{"t_save_lumps",      w_t_save_lumps,           0, 0,
+			REQUEST_ROUTE},
 
 
 	/* not applicable from the script */
@@ -582,6 +585,10 @@ static int script_init( struct sip_msg *foo, void *bar)
 	/* set request mode so that multiple-mode actions know
 	 * how to behave */
 	rmode=MODE_REQUEST;
+
+#ifdef POSTPONE_MSG_CLONING
+	lumps_are_cloned = 0;
+#endif
 	return 1;
 }
 
@@ -1448,8 +1455,6 @@ int t_is_canceled(struct sip_msg* msg, char* foo, char* bar)
 	return ret;
 }
 
-
-
 /* script function, returns: 1 if any of the branches did timeout, -1 if not */
 int t_any_timeout(struct sip_msg* msg, char* foo, char* bar)
 {
@@ -1530,6 +1535,31 @@ static int w_t_drop_replies(struct sip_msg* msg, char* foo, char* bar)
 	return 1;
 }
 
+/* save the message lumps after t_newtran() but before t_relay() */
+static int w_t_save_lumps(struct sip_msg* msg, char* foo, char* bar)
+{
+#ifdef POSTPONE_MSG_CLONING
+	struct cell *t;
+
+	t=get_t();
+	if (!t || t==T_UNDEFINED) {
+		LOG(L_ERR, "ERROR: w_t_save_lumps: transaction has not been created yet\n");
+		return -1;
+	}
+
+	if (save_msg_lumps(t->uas.request, msg)) {
+		LOG(L_ERR, "ERROR: w_t_save_lumps: "
+			"failed to save the message lumps\n");
+		return -1;
+	}
+	return 1;
+#else
+	LOG(L_ERR, "ERROR: w_t_save_lumps: POSTPONE_MSG_CLONING is not defined,"
+			" thus, the functionality is not supported\n");
+	return -1;
+#endif
+}
+
 static rpc_export_t tm_rpc[] = {
 	{"tm.cancel", rpc_cancel,   rpc_cancel_doc,   0},
 	{"tm.reply",  rpc_reply,    rpc_reply_doc,    0},