tcp: config option for the async write block size

 - the block size used for the async writes can now be configured
   both from ser.cfg (tcp_wq_blk_size) and at runtime. This value
   has only a little performance impact and only when writes are
   delayed.  Small values are safer (big values on proxies that
   open thousands of connections over slow links would eat up a
   lot of memory). For now it's main use is debugging.

NEWS

@@ -236,6 +236,20 @@ core:
                between the short name and long name in cache as CNAME record
 
 new config variables:
+  tcp_rd_buf_size = buffer size used for tcp reads.
+                    A high buffer size increases performance on server with few
+                    connections and lot of traffic on them, but also increases
+                     memory consumption (so for lots of connection is better 
+                    to use a low value). Note also that this value limits the
+                    maximum datagram size that can be received over tcp.
+                    Default: 4096, can be changed at runtime.
+  tcp_wq_blk_size = block size used for tcp async writes. It should be big
+                    enough to hold a few datagrams. If it's smaller then a
+                    datagram (in fact a tcp write()) size, it will be rounded
+                    up. It has no influenced on the number of datagrams 
+                    queued (for that see tcp_conn_wq_max or tcp_wq_max).
+                    It has mostly debugging and testing value (can be ignored).
+                    Default: 2100 (~ 2 INVITEs), can be changed at runtime.
   tcp_no_connect = yes/no - disable connects, ser will only accept new 
                      connections, it will never try to open new ones.
                      Default: no, can be changed at runtime.

cfg.lex

@@ -305,6 +305,8 @@ TCP_OPT_FD_CACHE	"tcp_fd_cache"
 TCP_OPT_BUF_WRITE	"tcp_buf_write"|"tcp_async"
 TCP_OPT_CONN_WQ_MAX	"tcp_conn_wq_max"
 TCP_OPT_WQ_MAX		"tcp_wq_max"
+TCP_OPT_RD_BUF		"tcp_rd_buf_size"
+TCP_OPT_WQ_BLK		"tcp_wq_blk_size"
 TCP_OPT_DEFER_ACCEPT "tcp_defer_accept"
 TCP_OPT_DELAYED_ACK	"tcp_delayed_ack"
 TCP_OPT_SYNCNT		"tcp_syncnt"
@@ -605,6 +607,10 @@ EAT_ABLE	[\ \t\b\r]
 									return TCP_OPT_CONN_WQ_MAX; }
 <INITIAL>{TCP_OPT_WQ_MAX}	{ count(); yylval.strval=yytext;
 									return TCP_OPT_WQ_MAX; }
+<INITIAL>{TCP_OPT_RD_BUF}	{ count(); yylval.strval=yytext;
+									return TCP_OPT_RD_BUF; }
+<INITIAL>{TCP_OPT_WQ_BLK}	{ count(); yylval.strval=yytext;
+									return TCP_OPT_WQ_BLK; }
 <INITIAL>{TCP_OPT_BUF_WRITE}	{ count(); yylval.strval=yytext;
 									return TCP_OPT_BUF_WRITE; }
 <INITIAL>{TCP_OPT_DEFER_ACCEPT}	{ count(); yylval.strval=yytext;

cfg.y

@@ -359,6 +359,8 @@ static void free_socket_id_lst(struct socket_id* i);
 %token TCP_OPT_BUF_WRITE
 %token TCP_OPT_CONN_WQ_MAX
 %token TCP_OPT_WQ_MAX
+%token TCP_OPT_RD_BUF
+%token TCP_OPT_WQ_BLK
 %token TCP_OPT_DEFER_ACCEPT
 %token TCP_OPT_DELAYED_ACK
 %token TCP_OPT_SYNCNT
@@ -907,7 +909,23 @@ assign_stm:
 			warn("tcp support not compiled in");
 		#endif
 	}
-	| TCP_OPT_WQ_MAX error { yyerror("boolean value expected"); }
+	| TCP_OPT_WQ_MAX error { yyerror("number expected"); }
+	| TCP_OPT_RD_BUF EQUAL NUMBER {
+		#ifdef USE_TCP
+			tcp_default_cfg.rd_buf_size=$3;
+		#else
+			warn("tcp support not compiled in");
+		#endif
+	}
+	| TCP_OPT_RD_BUF error { yyerror("number expected"); }
+	| TCP_OPT_WQ_BLK EQUAL NUMBER {
+		#ifdef USE_TCP
+			tcp_default_cfg.wq_blk_size=$3;
+		#else
+			warn("tcp support not compiled in");
+		#endif
+	}
+	| TCP_OPT_WQ_BLK error { yyerror("number expected"); }
 	| TCP_OPT_DEFER_ACCEPT EQUAL NUMBER {
 		#ifdef USE_TCP
 			tcp_default_cfg.defer_accept=$3;

tcp_init.h

@@ -39,6 +39,10 @@
 										  time, timeout */
 #define DEFAULT_TCP_MAX_CONNECTIONS 2048 /* maximum connections */
 
+#define DEFAULT_TCP_BUF_SIZE	4096  /* buffer size used for reads */
+
+#define DEFAULT_TCP_WBUF_SIZE	2100 /*  after debugging switch to 4-16k */
+
 struct tcp_child{
 	pid_t pid;
 	int proc_no; /* ser proc_no, for debugging */

tcp_main.c

@@ -208,7 +208,6 @@
 #define TCPCONN_WAIT_TIMEOUT 1 /* 1 tick */
 
 #ifdef TCP_ASYNC
-#define TCP_WBUF_SIZE	1024 /* FIXME: after debugging switch to 16-32k */
 static unsigned int* tcp_total_wq=0;
 #endif
 
@@ -642,7 +641,7 @@ inline static int _wbufq_add(struct  tcp_connection* c, char* data,
 	}
 	
 	if (unlikely(q->last==0)){
-		wb_size=MAX_unsigned(TCP_WBUF_SIZE, size);
+		wb_size=MAX_unsigned(cfg_get(tcp, tcp_cfg, wq_blk_size), size);
 		wb=shm_malloc(sizeof(*wb)+wb_size-1);
 		if (unlikely(wb==0))
 			goto error;
@@ -663,7 +662,7 @@ inline static int _wbufq_add(struct  tcp_connection* c, char* data,
 	while(size){
 		last_free=wb->b_size-q->last_used;
 		if (last_free==0){
-			wb_size=MAX_unsigned(TCP_WBUF_SIZE, size);
+			wb_size=MAX_unsigned(cfg_get(tcp, tcp_cfg, wq_blk_size), size);
 			wb=shm_malloc(sizeof(*wb)+wb_size-1);
 			if (unlikely(wb==0))
 				goto error;
@@ -926,13 +925,15 @@ struct tcp_connection* tcpconn_new(int sock, union sockaddr_union* su,
 									int state)
 {
 	struct tcp_connection *c;
+	int rd_b_size;
 	
-	c=(struct tcp_connection*)shm_malloc(sizeof(struct tcp_connection));
+	rd_b_size=cfg_get(tcp, tcp_cfg, rd_buf_size);
+	c=shm_malloc(sizeof(struct tcp_connection) + rd_b_size);
 	if (c==0){
 		LOG(L_ERR, "ERROR: tcpconn_new: mem. allocation failure\n");
 		goto error;
 	}
-	memset(c, 0, sizeof(struct tcp_connection)); /* zero init */
+	memset(c, 0, sizeof(struct tcp_connection)); /* zero init (skip rd buf)*/
 	c->s=sock;
 	c->fd=-1; /* not initialized */
 	if (lock_init(&c->write_lock)==0){
@@ -956,7 +957,7 @@ struct tcp_connection* tcpconn_new(int sock, union sockaddr_union* su,
 	}
 	print_ip("tcpconn_new: new tcp connection: ", &c->rcv.src_ip, "\n");
 	DBG(     "tcpconn_new: on port %d, type %d\n", c->rcv.src_port, type);
-	init_tcp_req(&c->req);
+	init_tcp_req(&c->req, (char*)c+sizeof(struct tcp_connection), rd_b_size);
 	c->id=(*connection_id)++;
 	c->rcv.proto_reserved1=0; /* this will be filled before receive_message*/
 	c->rcv.proto_reserved2=0;

tcp_options.c

@@ -133,6 +133,10 @@ static cfg_def_t tcp_cfg_def[] = {
 	{ "wq_timeout_ticks",   CFG_VAR_INT | CFG_READONLY, 0,
 									MAX_TCP_CON_LIFETIME,         0,         0,
 		"internal send_timeout value in ticks, used in async. mode"},
+	{ "rd_buf_size", CFG_VAR_INT | CFG_ATOMIC,    512,    65536,  0,         0,
+		"internal read buffer size (should be > max. expected datagram)"},
+	{ "wq_blk_size", CFG_VAR_INT | CFG_ATOMIC,    1,    65535,  0,         0,
+		"internal async write block size (debugging use only for now)"},
 	{0, 0, 0, 0, 0, 0, 0}
 };
 
@@ -175,6 +179,8 @@ void init_tcp_options()
 	tcp_default_cfg.alias_flags=TCP_ALIAS_FORCE_ADD;
 	/* flags used for adding the default aliases of a new tcp connection */
 	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;
 }
 
 
@@ -261,6 +267,39 @@ static int fix_max_conns(void* cfg_h, str* name, void** val)
 
 
 
+/** fix *val according to the cfg entry "name".
+ * (*val must be integer)
+ * 1. check if *val is between name min..max and if not change it to
+ *    the corresp. value
+ * 2. call fixup callback if defined in the cfg
+ * @return 0 on success
+ */
+static int tcp_cfg_def_fix(char* name, int* val)
+{
+	cfg_def_t* c;
+	str s;
+	
+	for (c=&tcp_cfg_def[0]; c->name; c++){
+		if (strcmp(name, c->name)==0){
+			/* found */
+			if ((c->type & CFG_VAR_INT)  && (c->min || c->max)){
+				if (*val < c->min) *val=c->min;
+				else if (*val > c->max) *val=c->max;
+				if (c->on_change_cb){
+					s.s=c->name;
+					s.len=strlen(s.s);
+					return c->on_change_cb(&tcp_default_cfg, &s, (void*)val);
+				}
+			}
+			return 0;
+		}
+	}
+	WARN("tcp config option \"%s\" not found\n", name);
+	return -1; /* not found */
+}
+
+
+
 /* checks & warns if some tcp_option cannot be enabled */
 void tcp_options_check()
 {
@@ -324,6 +363,9 @@ void tcp_options_check()
 	tcp_default_cfg.tcp_wq_timeout=S_TO_TICKS(tcp_default_cfg.send_timeout_s);
 #endif /* TCP_ASYNC */
 	tcp_default_cfg.max_connections=tcp_max_connections;
+	tcp_cfg_def_fix("rd_buf_size", (int*)&tcp_default_cfg.rd_buf_size);
+	tcp_cfg_def_fix("wq_blk_size", (int*)&tcp_default_cfg.wq_blk_size);
+	
 }
 
 

tcp_options.h

@@ -140,6 +140,8 @@ struct cfg_group_tcp{
 	int new_conn_alias_flags;
 	/* internal, "fixed" vars */
 	unsigned int tcp_wq_timeout; /* in ticks, timeout for queued writes */
+	unsigned int rd_buf_size; /* read buffer size (should be > max. datagram)*/
+	unsigned int wq_blk_size; /* async write block size (debugging use) */
 };
 
 extern struct cfg_group_tcp tcp_default_cfg;