core: enable reuse of tcp sockets

- credits to Marco Sinibaldi for the original implementation

src/Makefile.defs

@@ -680,7 +680,7 @@ C_DEFS= $(extra_defs) \
 	 #-DSTATS \
 	 #-DNO_LOG \
 	 #-DPROFILING \
-	 #-DNO_SIG_DEBUG
+	 #-DNO_SIG_DEBUG 
 
 #PROFILE=  -pg	# set this if you want profiling
 				# you may also want to set -DPROFILING

src/core/cfg.lex

@@ -395,6 +395,7 @@ TCP_OPT_KEEPCNT		"tcp_keepcnt"
 TCP_OPT_CRLF_PING	"tcp_crlf_ping"
 TCP_OPT_ACCEPT_NO_CL	"tcp_accept_no_cl"
 TCP_CLONE_RCVBUF	"tcp_clone_rcvbuf"
+TCP_REUSE_PORT		"tcp_reuse_port"
 DISABLE_TLS		"disable_tls"|"tls_disable"
 ENABLE_TLS		"enable_tls"|"tls_enable"
 TLSLOG			"tlslog"|"tls_log"
@@ -837,6 +838,7 @@ IMPORTFILE      "import_file"
 									return TCP_OPT_ACCEPT_NO_CL; }
 <INITIAL>{TCP_CLONE_RCVBUF}		{ count(); yylval.strval=yytext;
 									return TCP_CLONE_RCVBUF; }
+<INITIAL>{TCP_REUSE_PORT}	{ count(); yylval.strval=yytext; return TCP_REUSE_PORT; }
 <INITIAL>{DISABLE_TLS}	{ count(); yylval.strval=yytext; return DISABLE_TLS; }
 <INITIAL>{ENABLE_TLS}	{ count(); yylval.strval=yytext; return ENABLE_TLS; }
 <INITIAL>{TLSLOG}		{ count(); yylval.strval=yytext; return TLS_PORT_NO; }

src/core/cfg.y

@@ -430,6 +430,7 @@ extern char *default_routename;
 %token TCP_OPT_CRLF_PING
 %token TCP_OPT_ACCEPT_NO_CL
 %token TCP_CLONE_RCVBUF
+%token TCP_REUSE_PORT
 %token DISABLE_TLS
 %token ENABLE_TLS
 %token TLSLOG
@@ -1204,6 +1205,18 @@ assign_stm:
 		#endif
 	}
 	| TCP_CLONE_RCVBUF EQUAL error { yyerror("number expected"); }
+	| TCP_REUSE_PORT EQUAL NUMBER {
+		#ifdef USE_TCP
+		#ifdef SO_REUSEPORT
+			tcp_default_cfg.reuse_port=$3;
+		#else
+			warn("support for SO_REUSEPORT not compiled in");
+		#endif
+		#else
+			warn("tcp support not compiled in");
+		#endif
+	}
+	| TCP_REUSE_PORT EQUAL error { yyerror("boolean value expected"); }
 	| DISABLE_TLS EQUAL NUMBER {
 		#ifdef USE_TLS
 			tls_disable=$3;

src/core/forward.h

@@ -151,8 +151,22 @@ static inline int msg_send_buffer(struct dest_info* dst, char* buf, int len,
 	) && sr_event_enabled(SREV_TCP_WS_FRAME_OUT))) {
 		if (unlikely(dst->send_flags.f & SND_F_FORCE_SOCKET
 				&& dst->send_sock)) {
+			
 			local_addr = dst->send_sock->su;
+#ifdef SO_REUSEPORT
+			if (cfg_get(tcp, tcp_cfg, reuse_port)) {
+                        	LM_DBG("sending to: %s, force_socket=%d, send_sock=%p\n",
+                                        	su2a(&dst->to,sizeof(struct sockaddr_in)),
+                                        	(dst->send_flags.f & SND_F_FORCE_SOCKET),
+                                        	dst->send_sock);
+
+                        	su_setport(&local_addr, dst->send_sock->port_no);
+			}
+			else
+				su_setport(&local_addr, 0); /* any local port will do */
+#else
 			su_setport(&local_addr, 0); /* any local port will do */
+#endif
 			from = &local_addr;
 		}
 
@@ -212,7 +226,20 @@ static inline int msg_send_buffer(struct dest_info* dst, char* buf, int len,
 			if (unlikely((dst->send_flags.f & SND_F_FORCE_SOCKET) &&
 						dst->send_sock)) {
 				local_addr=dst->send_sock->su;
+#ifdef SO_REUSEPORT
+				if (cfg_get(tcp, tcp_cfg, reuse_port)) {
+                        		LM_DBG("sending to: %s, force_socket=%d, send_sock=%p\n",
+                                        		su2a(&dst->to,sizeof(struct sockaddr_in)),
+                                        		(dst->send_flags.f & SND_F_FORCE_SOCKET),
+                                        		dst->send_sock);
+
+                        		su_setport(&local_addr, dst->send_sock->port_no);
+				}
+				else
+					su_setport(&local_addr, 0); /* any local port will do */
+#else
 				su_setport(&local_addr, 0); /* any local port will do */
+#endif
 				from=&local_addr;
 			}
 			if (unlikely(tcp_send(dst, from, outb.s, outb.len)<0)){
@@ -232,7 +259,20 @@ static inline int msg_send_buffer(struct dest_info* dst, char* buf, int len,
 			if (unlikely((dst->send_flags.f & SND_F_FORCE_SOCKET) &&
 						dst->send_sock)) {
 				local_addr=dst->send_sock->su;
+#ifdef SO_REUSEPORT
+				if (cfg_get(tcp, tcp_cfg, reuse_port)) {
+                        		LM_DBG("sending to: %s, force_socket=%d, send_sock=%p\n",
+                                        		su2a(&dst->to,sizeof(struct sockaddr_in)),
+                                        		(dst->send_flags.f & SND_F_FORCE_SOCKET),
+                                        		dst->send_sock);
+
+                        		su_setport(&local_addr, dst->send_sock->port_no);
+				}
+				else
+					su_setport(&local_addr, 0); /* any local port will do */
+#else
 				su_setport(&local_addr, 0); /* any local port will do */
+#endif
 				from=&local_addr;
 			}
 			if (unlikely(tcp_send(dst, from, outb.s, outb.len)<0)){

src/core/tcp_main.c

@@ -313,6 +313,16 @@ static int init_sock_opt(int s, int af)
 		/* continue, not critical */
 	}
 #endif /* !TCP_DONT_REUSEADDR */
+
+#ifdef SO_REUSEPORT
+	if ((optval=cfg_get(tcp, tcp_cfg, reuse_port))) {
+		if (setsockopt(s, SOL_SOCKET, SO_REUSEPORT,
+				(void*)&optval, sizeof(optval))==-1) {
+			LM_ERR("setsockopt %s\n", strerror(errno));
+		}
+	}
+#endif
+
 #ifdef HAVE_TCP_SYNCNT
 	if ((optval=cfg_get(tcp, tcp_cfg, syncnt))){
 		if (setsockopt(s, IPPROTO_TCP, TCP_SYNCNT, &optval,
@@ -2747,6 +2757,16 @@ int tcp_init(struct socket_info* sock_info)
 		goto error;
 	}
 #endif
+
+#ifdef SO_REUSEPORT
+	if ((optval=cfg_get(tcp, tcp_cfg, reuse_port))) {
+		if (setsockopt(sock_info->socket, SOL_SOCKET, SO_REUSEPORT,
+				(void*)&optval, sizeof(optval))==-1) {
+			LM_ERR("setsockopt %s\n", strerror(errno));
+		}
+	}
+#endif
+
 	/* tos */
 	optval = tos;
 	if(sock_info->address.af==AF_INET){

src/core/tcp_options.c

@@ -106,6 +106,8 @@ static cfg_def_t tcp_cfg_def[] = {
 		"flags for the def. aliases for a new conn. (FORCE_ADD:1, REPLACE:2 "},
 	{ "accept_no_cl",   CFG_VAR_INT | CFG_ATOMIC,   0,        1,  0,         0,
 		"accept TCP messages without Content-Length "},
+	{ "reuse_port",   CFG_VAR_INT | CFG_ATOMIC,   0,        1,  0,         0,
+		"reuse TCP ports "},
 	/* internal and/or "fixed" versions of some vars
 	   (not supposed to be writeable, read will provide only debugging value*/
 	{ "rd_buf_size", CFG_VAR_INT | CFG_ATOMIC,    512,    16777216,  0,         0,
@@ -161,6 +163,7 @@ void init_tcp_options()
 	tcp_default_cfg.new_conn_alias_flags=TCP_ALIAS_REPLACE;
 	tcp_default_cfg.rd_buf_size=DEFAULT_TCP_BUF_SIZE;
 	tcp_default_cfg.wq_blk_size=DEFAULT_TCP_WBUF_SIZE;
+	tcp_default_cfg.reuse_port=0;
 }
 
 

src/core/tcp_options.h

@@ -137,6 +137,7 @@ struct cfg_group_tcp{
 	int alias_flags;
 	int new_conn_alias_flags;
 	int accept_no_cl;  /* on/off - accept messages without content-length */
+	int reuse_port;  /* enable SO_REUSEPORT */
 
 	/* internal, "fixed" vars */
 	unsigned int rd_buf_size; /* read buffer size (should be > max. datagram)*/

src/modules/tm/uac.c

@@ -476,6 +476,14 @@ static inline int t_uac_prepare(uac_req_t *uac_r,
 	request->dst = dst;
 	request->flags |= nhtype;
 
+#ifdef SO_REUSEPORT
+	if (cfg_get(tcp, tcp_cfg, reuse_port) && 
+			uac_r->ssock!=NULL && uac_r->ssock->len>0 &&
+			request->dst.send_sock->proto == PROTO_TCP) {
+		request->dst.send_flags.f |= SND_F_FORCE_SOCKET;
+	}
+#endif
+
 	if (!is_ack) {
 #ifdef TM_DEL_UNREF
 		INIT_REF(new_cell, 1); /* ref'ed only from the hash */