kazoo - fix timer interval

amqp timer and publish proc take all cpu
new param amqp_timmer_process_interval in MS
and removed the timeout on select from tcp workers

modules/kazoo/kazoo.c

@@ -70,6 +70,8 @@ struct timeval kz_sock_tv = (struct timeval){0,100000};
 struct timeval kz_amqp_tv = (struct timeval){0,100000};
 struct timeval kz_qtimeout_tv = (struct timeval){2,0};
 struct timeval kz_ack_tv = (struct timeval){0,100000};
+struct timeval kz_timer_tv = (struct timeval){0,200000};
+int kz_timer_ms = 200;
 
 
 str dbk_consumer_event_key = str_init("Event-Category");
@@ -148,6 +150,7 @@ static param_export_t params[] = {
   //  {"mwi_expires", INT_PARAM, &dbk_mwi_expires},
     {"amqp_connection", STR_PARAM|USE_FUNC_PARAM,(void*)kz_amqp_add_connection},
     {"amqp_max_channels", INT_PARAM, &dbk_channels},
+    {"amqp_timmer_process_interval", INT_PARAM, &kz_timer_ms},
     {"amqp_consumer_ack_timeout_micro", INT_PARAM, &kz_ack_tv.tv_usec},
     {"amqp_consumer_ack_timeout_sec", INT_PARAM, &kz_ack_tv.tv_sec},
     {"amqp_interprocess_timeout_micro", INT_PARAM, &kz_sock_tv.tv_usec},
@@ -235,6 +238,11 @@ static int mod_init(void) {
     if(!kz_amqp_init()) {
    		return -1;
     }
+
+    if(kz_timer_ms > 0) {
+    	kz_timer_tv.tv_usec = (kz_timer_ms  1000) * 1000;
+    	kz_timer_tv.tv_sec = kz_timer_ms / 1000;
+    }
     
     if(dbk_pua_mode == 1) {
 		kz_db_url.len = kz_db_url.s ? strlen(kz_db_url.s) : 0;

modules/kazoo/kz_amqp.c

@@ -35,6 +35,7 @@ extern struct timeval kz_sock_tv;
 extern struct timeval kz_amqp_tv;
 extern struct timeval kz_qtimeout_tv;
 extern struct timeval kz_ack_tv;
+extern struct timeval kz_timer_tv;
 
 extern int dbk_internal_loop_count;
 extern int dbk_consumer_loop_count;
@@ -1909,6 +1910,7 @@ void kz_amqp_timeout_proc(int child_no)
 	int i;
     while(1) {
 		struct timeval now;
+		usleep(kz_timer_tv.tv_usec);
 		for(i=0; i < dbk_channels; i++) {
 			gettimeofday(&now, NULL);
 			if(channels[i].state == KZ_AMQP_CALLING
@@ -1972,7 +1974,7 @@ void kz_amqp_publisher_proc(int child_no)
     	while(OK) {
 			FD_ZERO(&fdset);
 			FD_SET(data_pipe, &fdset);
-			selret = select(FD_SETSIZE, &fdset, NULL, NULL, &kz_sock_tv);
+			selret = select(FD_SETSIZE, &fdset, NULL, NULL, NULL);
 			if (selret < 0) {
 				LM_ERR("select() failed: %s\n", strerror(errno));
 				continue;