[core] improve accuracy and make configurable the reactor timeout

The timeout accuracy for the async script operations depends on the value of the I/O reactor timeout (how often the reactor can check for timeout events).
In order to cope with 1 second async timeout, we need a much lower reactor timeout -> the reactor timeout was moved from seconds to miliseconds
Also this reactor timeout can not be configurable via script param, see the new `worker_reactor_timeout` core parameter. This takes values between 100 and 1000 ms.

Author: bogdan-iancu

cfg.lex

@@ -248,6 +248,7 @@ MEMLOG		"memlog"|"mem_log"
 MEMDUMP		"memdump"|"mem_dump"
 SHM_MEMLOG_SIZE			"shm_memlog_size"
 EXECMSGTHRESHOLD		"execmsgthreshold"|"exec_msg_threshold"
+WORKER_REACTOR_TIMEOUT	"workerreactortimeout"|"worker_reactor_timeout"
 EXECDNSTHRESHOLD		"execdnsthreshold"|"exec_dns_threshold"
 TCPTHRESHOLD			"tcpthreshold"|"tcp_threshold"
 EVENT_SHM_THRESHOLD		"event_shm_threshold"
@@ -482,6 +483,7 @@ SPACE		[ ]
 <INITIAL>{MEMDUMP}	{ count(); yylval.strval=yytext; return MEMDUMP; }
 <INITIAL>{SHM_MEMLOG_SIZE}	{ count(); yylval.strval=yytext; return SHM_MEMLOG_SIZE; }
 <INITIAL>{EXECMSGTHRESHOLD}	{ count(); yylval.strval=yytext; return EXECMSGTHRESHOLD; }
+<INITIAL>{WORKER_REACTOR_TIMEOUT}	{ count(); yylval.strval=yytext; return WORKER_REACTOR_TIMEOUT; }
 <INITIAL>{EXECDNSTHRESHOLD}	{ count(); yylval.strval=yytext; return EXECDNSTHRESHOLD; }
 <INITIAL>{TCPTHRESHOLD}	{ count(); yylval.strval=yytext; return TCPTHRESHOLD; }
 <INITIAL>{EVENT_SHM_THRESHOLD}	{ count(); yylval.strval=yytext; return EVENT_SHM_THRESHOLD; }

cfg.y

@@ -111,6 +111,7 @@
 #include "config.h"
 #include "mem/rpm_mem.h"
 #include "poll_types.h"
+#include "reactor_defs.h"
 
 #ifdef SHM_EXTRA_STATS
 #include "mem/module_info.h"
@@ -354,6 +355,7 @@ extern int cfg_parse_only_routes;
 %token MEMDUMP
 %token SHM_MEMLOG_SIZE
 %token EXECMSGTHRESHOLD
+%token WORKER_REACTOR_TIMEOUT
 %token EXECDNSTHRESHOLD
 %token TCPTHRESHOLD
 %token EVENT_SHM_THRESHOLD
@@ -1259,6 +1261,8 @@ assign_stm: LOGLEVEL EQUAL snumber { IFOR();
 		| MEMDUMP EQUAL error { yyerror("int value expected"); }
 		| EXECMSGTHRESHOLD EQUAL NUMBER {  IFOR();execmsgthreshold=$3; }
 		| EXECMSGTHRESHOLD EQUAL error { yyerror("int value expected"); }
+		| WORKER_REACTOR_TIMEOUT EQUAL NUMBER { IFOR(); worker_reactor_timeout=$3; }
+		| WORKER_REACTOR_TIMEOUT EQUAL error { yyerror("int value expected"); }
 		| EXECDNSTHRESHOLD EQUAL NUMBER { IFOR(); execdnsthreshold=$3; }
 		| EXECDNSTHRESHOLD EQUAL error { yyerror("int value expected"); }
 		| TCPTHRESHOLD EQUAL NUMBER { IFOR(); tcpthreshold=$3; }

docs/manual/Script-CoreParameters.md

@@ -1505,6 +1505,21 @@ wdir = "/usr/local/opensips"
 wdir = /usr/opensips_wd
 ```
 
+### worker_reactor_timeout
+
+The timeout, in milliseconds, used by the internal I/O reactor in OpenSIPS worker processes. This is relevant for the timeout of the async operations. If you use low timeout values in your script async operations or you want these timeouts to be really accurate, you should lower this reactor timeout (so it can check more often for the async timeout jobs). If your aync timouts are high and precision is not a key concern, you can keep the the default of 1 second setting.
+
+By `worker` processes we do understand processes handling SIP traffic or SIP operations - shortly this impacts the UDP, TCP and TIMER related processes.
+
+Default value is `1000` milliseconds. Values lower than `100` are rounded up to `100`; values higher than `1000` are rounded down to `1000`.
+
+Example of usage:
+```opensips
+
+    worker_reactor_timeout = 500
+
+```
+
 ### xlog_buf_size
 
 Size of the buffer used to print a single line through the selected **OpenSIPS** logging facility. If the buffer is too small, an overflow error will be printed and the line will be skipped. The default value is `4096` bytes.

io_wait.h

@@ -114,7 +114,7 @@ struct fd_map {
 	                       * internal usage */
 	int app_flags;        /* flags to be used by upper layer apps, not by
 	                       * the reactor */
-	unsigned int timeout;
+	utime_t timeout;
 };
 
 
@@ -219,7 +219,7 @@ static inline struct fd_map* hash_fd_map(	io_wait_h* h,
 						fd_type type,
 						void* data,
 						int flags,
-						unsigned int timeout,
+						utime_t timeout,
 						int *already)
 {
 	if (h->fd_hash[fd].fd <= 0) {
@@ -367,7 +367,7 @@ inline static int io_watch_add(	io_wait_h* h, // lgtm [cpp/use-of-goto]
 								fd_type type,
 								void* data,
 								int prio,
-								unsigned int timeout,
+								utime_t timeout,
 								int flags)
 {
 
@@ -474,8 +474,9 @@ inline static int io_watch_add(	io_wait_h* h, // lgtm [cpp/use-of-goto]
 		return 0;
 	}
 
+	/* convert timeout from seconds to miliseconds*/
 	if (timeout)
-		timeout+=get_ticks();
+		timeout = timeout*1000000 + get_uticks();
 
 	if ((e=hash_fd_map(h, fd, type, data,flags, timeout, &already))==0){
 		LM_ERR("[%s] failed to hash the fd %d\n",h->name, fd);

io_wait_loop.h

@@ -62,7 +62,7 @@ static int handle_io(struct fd_map* fm, int idx,int event_type) {
 /*! \brief io_wait_loop_x style function
  * wait for io using poll()
  * \param h io_wait handle
- * \param t timeout in s
+ * \param t timeout in ms
  * \param repeat if !=0 handle_io will be called until it returns <=0
  * \return number of IO events handled on success (can be 0), -1 on error
  */
@@ -74,7 +74,7 @@ inline static int io_wait_loop_poll(io_wait_h* h, int t, int repeat)
 	unsigned int curr_time;
 
 again:
-		ret=n=poll(h->fd_array, h->fd_no, t*1000);
+		ret=n=poll(h->fd_array, h->fd_no, t);
 		if (n==-1){
 			if (errno==EINTR) goto again; /* signal, ignore it */
 			else{
@@ -136,8 +136,8 @@ inline static int io_wait_loop_select(io_wait_h* h, int t, int repeat)
 
 again:
 		sel_set=h->master_set;
-		timeout.tv_sec=t;
-		timeout.tv_usec=0;
+		timeout.tv_sec=t/1000;
+		timeout.tv_usec=(t%1000)*1000;
 		ret=n=select(h->max_fd_select+1, &sel_set, 0, 0, &timeout);
 		if (n<0){
 			if (errno==EINTR) goto again; /* just a signal */
@@ -172,25 +172,25 @@ inline static int io_wait_loop_epoll(io_wait_h* h, int t, int repeat)
 	struct fd_map *e;
 	struct epoll_event ep_event;
 	int fd;
-	unsigned int curr_time;
+	utime_t curr_time;
 
 again:
-		ret=n=epoll_wait(h->epfd, h->ep_array, h->fd_no, t*1000);
+		ret=n=epoll_wait(h->epfd, h->ep_array, h->fd_no, t);
 		if (n==-1){
 			if (errno == EINTR) {
 				goto again; /* signal, ignore it */
 			} else if (h->fd_no == 0) {
-				sleep(t);
+				poll(NULL, 0, t);
 				return 0;
 			} else {
 				LM_ERR("[%s] epoll_wait(%d, %p, %d, %d): %s [%d]\n",
-					h->name,h->epfd, h->ep_array, h->fd_no, t*1000,
+					h->name,h->epfd, h->ep_array, h->fd_no, t,
 					strerror(errno), errno);
 				goto error;
 			}
 		}
 
-		curr_time = get_ticks();
+		curr_time = get_uticks();
 
 		for (r=0; r<n; r++) {
 #if 0
@@ -327,8 +327,8 @@ inline static int io_wait_loop_kqueue(io_wait_h* h, int t, int repeat)
 	struct fd_map *e;
 	unsigned int curr_time;
 
-	tspec.tv_sec=t;
-	tspec.tv_nsec=0;
+	tspec.tv_sec=t/1000;
+	tspec.tv_nsec=(t%1000)*1000000;
 again:
 		ret=n=kevent(h->kq_fd, h->kq_changes, h->kq_nchanges,  h->kq_array,
 					h->fd_no, &tspec);
@@ -397,8 +397,8 @@ inline static int io_wait_loop_sigio_rt(io_wait_h* h, int t)
 	struct fd_map* fm;
 
 	ret=1; /* 1 event per call normally */
-	ts.tv_sec=t;
-	ts.tv_nsec=0;
+	ts.tv_sec=t/1000;
+	ts.tv_nsec=(t%1000)*1000000;
 	if (!sigismember(&h->sset, h->signo) || !sigismember(&h->sset, SIGIO)){
 		LM_CRIT("[%s] the signal mask is not properly set!\n",h->name);
 		goto error;
@@ -505,7 +505,7 @@ inline static int io_wait_loop_devpoll(io_wait_h* h, int t, int repeat)
 	struct fd_map *e;
 	unsigned int curr_time;
 
-		dpoll.dp_timeout=t*1000;
+		dpoll.dp_timeout=t;
 		dpoll.dp_nfds=h->fd_no;
 		dpoll.dp_fds=h->dp_changes;
 again:

modules/event_kafka/kafka_producer.c

@@ -45,7 +45,7 @@
 #define F_KAFKA_JOB -3    /* new job from an worker process */
 #define F_KAFKA_EVENT -4  /* new event in librdkafka main event queue */
 
-#define KAFKA_REACTOR_TIMEOUT 1
+#define KAFKA_REACTOR_TIMEOUT 1000 /* ms */
 
 void dr_msg_cb(rd_kafka_t *rk, const rd_kafka_message_t *rkmessage, void *opaque);
 

modules/event_stream/stream_send.c

@@ -37,7 +37,7 @@
 
 #define JSONRPC_REQ_NEW		0
 #define JSONRPC_REQ_SENT	1
-#define STREAM_REACTOR_TIMEOUT  1 /* sec */
+#define STREAM_REACTOR_TIMEOUT  1000 /* ms */
 #define STREAM_MAX_PENDING_READS 4
 #define IS_ERR(_err) (errno == _err)
 #define STREAM_ADDR(con) \

modules/freeswitch/fs_proc.c

@@ -37,7 +37,7 @@
 #include "fs_api.h"
 #include "fs_ipc.h"
 
-#define FS_REACTOR_TIMEOUT  1 /* sec */
+#define FS_REACTOR_TIMEOUT  1000 /* ms */
 #define FS_STATS_EVENT_NAME "HEARTBEAT"
 
 extern struct list_head *fs_sockets;

net/net_tcp_proc.c

@@ -275,7 +275,7 @@ int tcp_worker_proc_reactor_init( int unix_sock)
 void tcp_worker_proc_loop(void)
 {
 	/* main loop */
-	reactor_main_loop(TCP_CHILD_SELECT_TIMEOUT, error,);
+	reactor_main_loop( worker_reactor_timeout, error,);
 	LM_CRIT("exiting...");
 	exit(-1);
 error:

net/net_udp.c

@@ -37,8 +37,6 @@
 #include "net_udp.h"
 
 
-#define UDP_SELECT_TIMEOUT  1
-
 /* if the UDP network layer is used or not by some protos */
 static int udp_disabled = 1;
 
@@ -442,7 +440,7 @@ static int fork_dynamic_udp_process(void *si_filter)
 		/* the child proc is done read&write) dealing with the status pipe */
 		clean_read_pipeend();
 
-		reactor_main_loop(UDP_SELECT_TIMEOUT, error, );
+		reactor_main_loop( worker_reactor_timeout, error, );
 		destroy_worker_reactor();
 error:
 		report_failure_status();
@@ -567,7 +565,7 @@ int udp_start_processes(int *chd_rank, int *startup_done)
 					 * \see receive_msg
 					 * \see main_loop
 					 */
-					reactor_main_loop(UDP_SELECT_TIMEOUT, error, );
+					reactor_main_loop( worker_reactor_timeout, error, );
 					destroy_worker_reactor();
 					exit(-1);
 				} else {