modules_k/rls: RLS NOTIFY requests now trickled out instead of in one big batch on timeout

- New modparam: notifier_poll_rate (default 10 per second) can be
  used to set how often the notifier process should check for work.
  Each time the notifier process does something it should send (if
  the hash works evenly) 1/(waitn_time * notifier_poll_rate) sets
  of RLS NOTIFY requests.
- RLS NOTIFY requests will still come out within waitn_time seconds of
  the back-end NOTIFY.
- Corrected the the default value of waitn_time.  It is now 5 seconds
  which matches the documentation and the RFC.
- Fixed a whitespace/indentation inconsistency and a typo in one of the
  diagnostic messages.

modules_k/rls/README

@@ -33,19 +33,20 @@ Anca-Maria Vamanu
               3.8. clean_period (int)
               3.9. rlpres_clean_period (int)
               3.10. waitn_time (int)
-              3.11. max_expires (int)
-              3.12. expires_offset (int)
-              3.13. hash_size (int)
-              3.14. xcap_root (str)
-              3.15. integrated_xcap_server (int)
-              3.16. to_presence_code (int)
-              3.17. rls_event (str)
-              3.18. outbound_proxy (str)
-              3.19. server_address (str)
-              3.20. max_notify_body_length (int)
-              3.21. fetch_rows (integer)
-              3.22. disable_remote_presence (integer)
-              3.23. max_backend_subs (integer)
+              3.11. notifier_poll_rate (int)
+              3.12. max_expires (int)
+              3.13. expires_offset (int)
+              3.14. hash_size (int)
+              3.15. xcap_root (str)
+              3.16. integrated_xcap_server (int)
+              3.17. to_presence_code (int)
+              3.18. rls_event (str)
+              3.19. outbound_proxy (str)
+              3.20. server_address (str)
+              3.21. max_notify_body_length (int)
+              3.22. fetch_rows (integer)
+              3.23. disable_remote_presence (integer)
+              3.24. max_backend_subs (integer)
 
         4. Functions
 
@@ -69,22 +70,23 @@ Anca-Maria Vamanu
    1.8. Set clean_period parameter
    1.9. Set rlpres_clean_period parameter
    1.10. Set waitn_time parameter
-   1.11. Set max_expires parameter
-   1.12. Set expires_offset parameter
-   1.13. Set hash_size parameter
+   1.11. Set notifier_poll_rate parameter
+   1.12. Set max_expires parameter
+   1.13. Set expires_offset parameter
    1.14. Set hash_size parameter
-   1.15. Set integrated_xcap_server parameter
-   1.16. Set to_presence_code parameter
-   1.17. Set rls_event parameter
-   1.18. Set outbound_proxy parameter
-   1.19. Set server_address parameter
-   1.20. Set max_notify_body_length parameter
-   1.21. Set fetch_rows parameter
-   1.22. Set disable_remote_presence parameter
-   1.23. Set max_backend_subs parameter
-   1.24. rls_handle_subscribe usage
-   1.25. rls_handle_notify usage
-   1.26. rls_update_subs usage
+   1.15. Set hash_size parameter
+   1.16. Set integrated_xcap_server parameter
+   1.17. Set to_presence_code parameter
+   1.18. Set rls_event parameter
+   1.19. Set outbound_proxy parameter
+   1.20. Set server_address parameter
+   1.21. Set max_notify_body_length parameter
+   1.22. Set fetch_rows parameter
+   1.23. Set disable_remote_presence parameter
+   1.24. Set max_backend_subs parameter
+   1.25. rls_handle_subscribe usage
+   1.26. rls_handle_notify usage
+   1.27. rls_update_subs usage
 
 Chapter 1. Admin Guide
 
@@ -108,19 +110,20 @@ Chapter 1. Admin Guide
         3.8. clean_period (int)
         3.9. rlpres_clean_period (int)
         3.10. waitn_time (int)
-        3.11. max_expires (int)
-        3.12. expires_offset (int)
-        3.13. hash_size (int)
-        3.14. xcap_root (str)
-        3.15. integrated_xcap_server (int)
-        3.16. to_presence_code (int)
-        3.17. rls_event (str)
-        3.18. outbound_proxy (str)
-        3.19. server_address (str)
-        3.20. max_notify_body_length (int)
-        3.21. fetch_rows (integer)
-        3.22. disable_remote_presence (integer)
-        3.23. max_backend_subs (integer)
+        3.11. notifier_poll_rate (int)
+        3.12. max_expires (int)
+        3.13. expires_offset (int)
+        3.14. hash_size (int)
+        3.15. xcap_root (str)
+        3.16. integrated_xcap_server (int)
+        3.17. to_presence_code (int)
+        3.18. rls_event (str)
+        3.19. outbound_proxy (str)
+        3.20. server_address (str)
+        3.21. max_notify_body_length (int)
+        3.22. fetch_rows (integer)
+        3.23. disable_remote_presence (integer)
+        3.24. max_backend_subs (integer)
 
    4. Functions
 
@@ -189,19 +192,20 @@ Chapter 1. Admin Guide
    3.8. clean_period (int)
    3.9. rlpres_clean_period (int)
    3.10. waitn_time (int)
-   3.11. max_expires (int)
-   3.12. expires_offset (int)
-   3.13. hash_size (int)
-   3.14. xcap_root (str)
-   3.15. integrated_xcap_server (int)
-   3.16. to_presence_code (int)
-   3.17. rls_event (str)
-   3.18. outbound_proxy (str)
-   3.19. server_address (str)
-   3.20. max_notify_body_length (int)
-   3.21. fetch_rows (integer)
-   3.22. disable_remote_presence (integer)
-   3.23. max_backend_subs (integer)
+   3.11. notifier_poll_rate (int)
+   3.12. max_expires (int)
+   3.13. expires_offset (int)
+   3.14. hash_size (int)
+   3.15. xcap_root (str)
+   3.16. integrated_xcap_server (int)
+   3.17. to_presence_code (int)
+   3.18. rls_event (str)
+   3.19. outbound_proxy (str)
+   3.20. server_address (str)
+   3.21. max_notify_body_length (int)
+   3.22. fetch_rows (integer)
+   3.23. disable_remote_presence (integer)
+   3.24. max_backend_subs (integer)
 
 3.1. db_url(str)
 
@@ -325,9 +329,10 @@ modparam("rls", "rlpres_clean_period", 100)
 
 3.10. waitn_time (int)
 
-   The timer period at which the server should attempt to send Notifies
-   with the updated presence state of the subscribed list or watcher
-   information.
+   The maximum time period that RLS NOTIFY requests will be buffered for.
+   The server will attempt to send NOTIFY requests with the updated
+   presence state of the subscribed list or watcher information within
+   this many seconds of a change occurring.
 
    Default value is “5”.
 
@@ -336,18 +341,31 @@ modparam("rls", "rlpres_clean_period", 100)
 modparam("rls", "waitn_time", 10)
 ...
 
-3.11. max_expires (int)
+3.11. notifier_poll_rate (int)
+
+   The number of times per second that the notifier process should check
+   for work. Approximately 1/(waitn_time * notifier_poll_rate) of the
+   pending updates will be sent each time the notifier runs.
+
+   Default value is “10”.
+
+   Example 1.11. Set notifier_poll_rate parameter
+...
+modparam("rls", "notifier_poll_rate", 20)
+...
+
+3.12. max_expires (int)
 
    The maximum accepted expires for a subscription to a list.
 
    Default value is “7200”.
 
-   Example 1.11. Set max_expires parameter
+   Example 1.12. Set max_expires parameter
 ...
 modparam("rls", "max_expires", 10800)
 ...
 
-3.12. expires_offset (int)
+3.13. expires_offset (int)
 
    This paramater only has an effect when the db_mode is DB_ONLY (mode 2).
    When expired subscribers are checked for deletion from the database,
@@ -360,12 +378,12 @@ modparam("rls", "max_expires", 10800)
 
    Default value is “0”.
 
-   Example 1.12. Set expires_offset parameter
+   Example 1.13. Set expires_offset parameter
 ...
 modparam("rls", "expires_offset", 0)
 ...
 
-3.13. hash_size (int)
+3.14. hash_size (int)
 
    The dimension of the hash table used to store subscription to a list.
    This parameter will be used as the power of 2 when computing table
@@ -373,35 +391,35 @@ modparam("rls", "expires_offset", 0)
 
    Default value is “9 (512)”.
 
-   Example 1.13. Set hash_size parameter
+   Example 1.14. Set hash_size parameter
 ...
 modparam("rls", "hash_size", 11)
 ...
 
-3.14. xcap_root (str)
+3.15. xcap_root (str)
 
    The address of the xcap server.
 
    Default value is “NULL”.
 
-   Example 1.14. Set hash_size parameter
+   Example 1.15. Set hash_size parameter
 ...
 modparam("rls", "xcap_root", "http://192.168.2.132/xcap-root:800")
 ...
 
-3.15. integrated_xcap_server (int)
+3.16. integrated_xcap_server (int)
 
    This parameter should be set if only integrated xcap servers are used
    to store resource lists.
 
    Default value is “0”.
 
-   Example 1.15. Set integrated_xcap_server parameter
+   Example 1.16. Set integrated_xcap_server parameter
 ...
 modparam("rls", "integrated_xcap_server", 1)
 ...
 
-3.16. to_presence_code (int)
+3.17. to_presence_code (int)
 
    The code to be returned by rls_handle_subscribe function if the
    processed Subscribe is not a resource list Subscribe. This code can be
@@ -411,12 +429,12 @@ modparam("rls", "integrated_xcap_server", 1)
 
    Default value is “0”.
 
-   Example 1.16. Set to_presence_code parameter
+   Example 1.17. Set to_presence_code parameter
 ...
 modparam("rls", "to_presence_code", 10)
 ...
 
-3.17. rls_event (str)
+3.18. rls_event (str)
 
    The default event that RLS handles is presence. If some other events
    should also be handled by RLS they should be added using this
@@ -424,57 +442,57 @@ modparam("rls", "to_presence_code", 10)
 
    Default value is “"presence"”.
 
-   Example 1.17. Set rls_event parameter
+   Example 1.18. Set rls_event parameter
 ...
 modparam("rls", "rls_event", "dialog;sla")
 ...
 
-3.18. outbound_proxy (str)
+3.19. outbound_proxy (str)
 
    The SIP address where to send RLS subscriptions (outbound proxy address
    as SIP URI).
 
    Default value is “NULL”.
 
-   Example 1.18. Set outbound_proxy parameter
+   Example 1.19. Set outbound_proxy parameter
 ...
 modparam("rls", "outbound_proxy", "sip:presence.kamailio.org")
 ...
 
-3.19. server_address (str)
+3.20. server_address (str)
 
    The address of the server that will be used as a contact in sent
    Subscribe requests and 200 OK replies for Subscribe requests for RLS.
    It is a mandatory parameter.
 
-   Example 1.19. Set server_address parameter
+   Example 1.20. Set server_address parameter
 ...
 modparam("rls", "server_address", "sip:[email protected]:5060")
 ...
 
-3.20. max_notify_body_length (int)
+3.21. max_notify_body_length (int)
 
    The maximum size that the body of a NOTIFY message may be. If set to 0
    (the default), no size limit is applied. Note that this refers only to
    the body, not the complete NOTIFY message.
 
-   Example 1.20. Set max_notify_body_length parameter
+   Example 1.21. Set max_notify_body_length parameter
 ...
 modparam("rls", "max_notify_body_length", 32000)
 ...
 
-3.21. fetch_rows (integer)
+3.22. fetch_rows (integer)
 
    Number of rows to be loaded in one step from database.
 
    Default value is 500.
 
-   Example 1.21. Set fetch_rows parameter
+   Example 1.22. Set fetch_rows parameter
 ...
 modparam("rls", "fetch_rows", 1000)
 ...
 
-3.22. disable_remote_presence (integer)
+3.23. disable_remote_presence (integer)
 
    When set to a non-zero value RLS will not perform back-end SUBSCRIBEs
    to non-local presentities. When people have large contact lists RLS
@@ -487,12 +505,12 @@ modparam("rls", "fetch_rows", 1000)
 
    Default value is 0
 
-   Example 1.22. Set disable_remote_presence parameter
+   Example 1.23. Set disable_remote_presence parameter
 ...
 modparam("rls", "disable_remote_presence", 1)
 ...
 
-3.23. max_backend_subs (integer)
+3.24. max_backend_subs (integer)
 
    When set to a non-zero value RLS will limit the number of back-end
    SUBSCRIBEs for each RLS SUBSCRIBE to this value. Leaving this at the
@@ -503,7 +521,7 @@ modparam("rls", "disable_remote_presence", 1)
 
    Default value is 0
 
-   Example 1.23. Set max_backend_subs parameter
+   Example 1.24. Set max_backend_subs parameter
 ...
 modparam("rls", "max_backend_subs", 30)
 ...
@@ -523,7 +541,7 @@ modparam("rls", "max_backend_subs", 30)
 
    This function can be used from REQUEST_ROUTE.
 
-   Example 1.24. rls_handle_subscribe usage
+   Example 1.25. rls_handle_subscribe usage
 ...
 For presence and rls on the same machine:
         modparam("rls", "to_presence_code", 10)
@@ -551,7 +569,7 @@ For rls only:
 
    This function can be used from REQUEST_ROUTE.
 
-   Example 1.25. rls_handle_notify usage
+   Example 1.26. rls_handle_notify usage
 ...
 if(method=="NOTIFY")
     rls_handle_notify();
@@ -570,7 +588,7 @@ if(method=="NOTIFY")
 
    This function can be used from ANY_ROUTE.
 
-   Example 1.26. rls_update_subs usage
+   Example 1.27. rls_update_subs usage
 ...
 Within event_route[xhttp:request]:
         case "PUT":

modules_k/rls/doc/rls_admin.xml

@@ -296,9 +296,11 @@ modparam("rls", "rlpres_clean_period", 100)
 	<section>
 		<title><varname>waitn_time</varname> (int)</title>
 		<para>
-	 The timer period at which the server should attempt to send
-	 Notifies with the updated presence state of the subscribed list
-	 or watcher information.
+		The maximum time period that RLS NOTIFY requests will
+		be buffered for.  The server will attempt to send
+		NOTIFY requests with the updated presence state of the
+		subscribed list or watcher information within this
+		many seconds of a change occurring.
 		</para>
 		<para>
 		<emphasis>Default value is <quote>5</quote>.
@@ -314,6 +316,28 @@ modparam("rls", "waitn_time", 10)
 		</example>
 	</section>
 
+	<section>
+		<title><varname>notifier_poll_rate</varname> (int)</title>
+		<para>
+		The number of times per second that the notifier process
+		should check for work.  Approximately
+		1/(waitn_time * notifier_poll_rate) of the pending updates
+		will be sent each time the notifier runs.
+		</para>
+		<para>
+		<emphasis>Default value is <quote>10</quote>.
+		</emphasis>
+		</para>
+		<example>
+		<title>Set <varname>notifier_poll_rate</varname> parameter</title>
+		<programlisting format="linespecific">
+...
+modparam("rls", "notifier_poll_rate", 20)
+...
+</programlisting>
+		</example>
+	</section>
+
 	<section>
 		<title><varname>max_expires</varname> (int)</title>
 		<para>

modules_k/rls/resource_notify.c

@@ -703,7 +703,9 @@ int rls_handle_notify(struct sip_msg* msg, char* c1, char* c2)
 	query_cols[n_query_cols]= &str_updated_col;
 	query_vals[n_query_cols].type = DB1_INT;
 	query_vals[n_query_cols].nul = 0;
-	query_vals[n_query_cols].val.int_val= UPDATED_TYPE; 
+	query_vals[n_query_cols].val.int_val=
+		core_hash(res_id, NULL,
+				(waitn_time * rls_notifier_poll_rate) - 1);
 	n_query_cols++;
 		
 	query_cols[n_query_cols]= &str_auth_state_col;
@@ -821,6 +823,7 @@ error:
 
 void timer_send_notify(unsigned int ticks,void *param)
 {
+	static int round = 0;
 	db_key_t query_cols[1], update_cols[1], result_cols[6];
 	db_val_t query_vals[1], update_vals[1];
 	int did_col, resource_uri_col, auth_state_col, reason_col,
@@ -831,7 +834,8 @@ void timer_send_notify(unsigned int ticks,void *param)
 	query_cols[0]= &str_updated_col;
 	query_vals[0].type = DB1_INT;
 	query_vals[0].nul = 0;
-	query_vals[0].val.int_val= UPDATED_TYPE; 
+	query_vals[0].val.int_val= round;
+	if (++round > (waitn_time * rls_notifier_poll_rate) - 1) round = 0;
 
 	result_cols[did_col= n_result_cols++]= &str_rlsubs_did_col;
 	result_cols[resource_uri_col= n_result_cols++]= &str_resource_uri_col;

modules_k/rls/rls.c

@@ -41,6 +41,7 @@
 #include "../../dprint.h"
 #include "../../error.h"
 #include "../../ut.h"
+#include "../../timer_proc.h"
 #include "../../hashes.h"
 #include "../../mem/mem.h"
 #include "../../mem/shm_mem.h"
@@ -73,7 +74,8 @@ db_func_t rls_xcap_dbf;
 /** modules variables */
 str rls_server_address = {0, 0};
 int rls_expires_offset=0;
-int waitn_time = 10;
+int waitn_time = 5;
+int rls_notifier_poll_rate = 10;
 str rlsubs_table = str_init("rls_watchers");
 str rlpres_table = str_init("rls_presentity");
 str rls_xcap_table = str_init("xcap");
@@ -222,6 +224,7 @@ static param_export_t params[]={
 	{ "rlpres_table",           STR_PARAM,   &rlpres_table.s                 },
 	{ "xcap_table",             STR_PARAM,   &rls_xcap_table.s               },
 	{ "waitn_time",             INT_PARAM,   &waitn_time                     },
+	{ "notifier_poll_rate",     INT_PARAM,   &rls_notifier_poll_rate         },
 	{ "clean_period",           INT_PARAM,   &clean_period                   },
 	{ "rlpres_clean_period",    INT_PARAM,   &rlpres_clean_period            },
 	{ "max_expires",            INT_PARAM,   &rls_max_expires                },
@@ -297,34 +300,34 @@ static int mod_init(void)
 		rls_outbound_proxy.len = strlen(rls_outbound_proxy.s);
 	/* extract port if any */
 	if(xcap_root)
-    {
-        sep= strchr(xcap_root, ':');
-        if(sep)
-        {
-            char* sep2= NULL;
-            sep2= strchr(sep+ 1, ':');
-            if(sep2)
-                sep= sep2;
-
-            str port_str;
-
-            port_str.s= sep+ 1;
-            port_str.len= strlen(xcap_root)- (port_str.s-xcap_root);
-
-            if(str2int(&port_str, &xcap_port)< 0)
-            {
-                LM_ERR("converting string to int [port]= %.*s\n",port_str.len,
-                        port_str.s);
-                return -1;
-            }
-            if(xcap_port< 0 || xcap_port> 65535)
-            {
-                LM_ERR("wrong xcap server port\n");
-                return -1;
-            }
-            *sep= '\0';
-        }
-    }
+	{
+		sep= strchr(xcap_root, ':');
+		if(sep)
+		{
+			char* sep2= NULL;
+			sep2= strchr(sep+ 1, ':');
+			if(sep2)
+				sep= sep2;
+
+			str port_str;
+
+			port_str.s= sep+ 1;
+			port_str.len= strlen(xcap_root)- (port_str.s-xcap_root);
+
+			if(str2int(&port_str, &xcap_port)< 0)
+			{
+				LM_ERR("converting string to int [port]= %.*s\n",
+					port_str.len, port_str.s);
+				return -1;
+			}
+			if(xcap_port< 0 || xcap_port> 65535)
+			{
+				LM_ERR("wrong xcap server port\n");
+				return -1;
+			}
+			*sep= '\0';
+		}
+	}
 
 	/* bind the SL API */
 	if (sl_load_api(&slb)!=0) {
@@ -535,6 +538,9 @@ static int mod_init(void)
 	if(waitn_time<= 0)
 		waitn_time= 5;
 
+	if(rls_notifier_poll_rate<= 0)
+		rls_notifier_poll_rate= 10;
+
 	/* bind libxml wrapper functions */
 
 	if((bind_libxml=(bind_libxml_t)find_export("bind_libxml_api", 1, 0))== NULL)
@@ -615,7 +621,8 @@ static int mod_init(void)
 			return -1;
 		}
 	}
-	register_timer(timer_send_notify,0, waitn_time);
+
+	register_basic_timers(1);
 
 	if (rlpres_clean_period < 0)
 		rlpres_clean_period = clean_period;
@@ -628,7 +635,7 @@ static int mod_init(void)
 
 	if ((rls_update_subs_lock = lock_alloc()) == NULL)
 	{
-		LM_ERR("Failed to alloc rls_updae_subs_lock\n");
+		LM_ERR("Failed to alloc rls_update_subs_lock\n");
 		return -1;
 	}
 	if (lock_init(rls_update_subs_lock) == NULL)
@@ -648,6 +655,17 @@ static int child_init(int rank)
 	if (rank==PROC_INIT || rank==PROC_TCP_MAIN)
 		return 0; /* don't call child_init for main process more than once */
 
+	if (rank==PROC_MAIN)
+	{
+		if (fork_basic_utimer(PROC_TIMER, "RLS NOTIFIER", 1,
+					timer_send_notify, NULL,
+					1000000/rls_notifier_poll_rate) < 0)
+		{
+			LM_ERR("Failed to start RLS NOTIFIER\n");
+			return -1;
+		}
+	}
+
 	LM_DBG("child [%d]  pid [%d]\n", rank, getpid());
 
 	if (rls_dbf.init==0)

modules_k/rls/rls.h

@@ -88,6 +88,7 @@ extern char* xcap_root;
 extern unsigned int xcap_port;
 extern str rls_server_address;
 extern int waitn_time;
+extern int rls_notifier_poll_rate;
 extern str rlsubs_table;
 extern str rlpres_table;
 extern str rls_xcap_table;