modules/cdp: new mod param: workerq_length_threshold_percentage

This is the threshold of the length of the worker queue as a percentage of
the maximum queue size - when exceeded a warning is written to the log
file. Nice to check if worker queue is growing.

modules/cdp/doc/cdp_admin.xml

@@ -126,6 +126,24 @@ modparam("cdp", "latency_threshold", 1000)
 </programlisting>
       </example>
     </section>
+    <section>
+      <title>workerq_length_threshold_percentage (int)</title>
+
+      <para>The threshold of the length of the worker queue as a percentage of 
+	  the maximum queue size - when exceeded a warning is written to the log
+	  file. 0  means disabled</para>
+
+      <para><emphasis> Default value is <quote>0</quote>. </emphasis></para>
+
+      <example>
+        <title>Set <varname>workerq_length_threshold_percentage</varname> parameter</title>
+
+        <programlisting format="linespecific">...
+modparam("cdp", "workerq_length_threshold_percentage", 25)
+...
+	</programlisting>
+      </example>
+    </section>
   </section>
 
   <section>

modules/cdp/mod.c

@@ -62,6 +62,7 @@ char* config_file="DiameterPeer.xml"; 	/**< default DiameterPeer configuration f
 unsigned int latency_threshold = 500;			/**< default threshold for Diameter calls (ms) */
 unsigned int *latency_threshold_p = &latency_threshold;
 unsigned int workerq_latency_threshold = 100;	/**< default threshold for putting a task into worker queue (ms) */
+unsigned int workerq_length_threshold_percentage = 0;	/**< default threshold for worker queue length, percentage of max queue length - by default disabled */
 
 extern dp_config *config; 				/**< DiameterPeer configuration structure */
 
@@ -165,6 +166,7 @@ static param_export_t cdp_params[] = {
 	{ "config_file",				PARAM_STRING,	&config_file}, 				/**< configuration filename */
 	{ "latency_threshold", 			PARAM_INT, 		&latency_threshold},		/**<threshold above which we will log*/
 	{ "workerq_latency_threshold", 	PARAM_INT, 		&workerq_latency_threshold},/**<time threshold putting job into queue*/
+	{ "workerq_length_threshold_percentage", 	PARAM_INT, 		&workerq_length_threshold_percentage},/**<queue length threshold - percentage of max queue length*/
 	{ 0, 0, 0 }
 };
 
@@ -215,7 +217,7 @@ static int cdp_init( void )
 		LM_ERR("failed to register stat\n");
 		return -1;
 	}
-
+	
 	if (register_module_stats( exports.name, mod_stats)!=0 ) {
 		LM_ERR("failed to register core statistics\n");
 		return -1;

modules/cdp/worker.c

@@ -70,6 +70,7 @@ task_queue_t *tasks; /**< queue of tasks */
 cdp_cb_list_t *callbacks; /**< list of callbacks for message processing */
 
 extern unsigned int workerq_latency_threshold; /**<max delay for putting task into worker queue */
+extern unsigned int workerq_length_threshold_percentage;	/**< default threshold for worker queue length, percentage of max queue length */
 /**
  * Initializes the worker structures, like the task queue.
  */
@@ -207,6 +208,8 @@ void cb_remove(cdp_cb_t *cb) {
 int put_task(peer *p, AAAMessage *msg) {
 
     struct timeval start, stop;
+    int num_tasks, length_percentage;
+
     long elapsed_useconds=0, elapsed_seconds=0, elapsed_millis=0;
     lock_get(tasks->lock);
 
@@ -246,6 +249,13 @@ int put_task(peer *p, AAAMessage *msg) {
         LM_WARN("Error releasing tasks->empty semaphore > %s!\n", strerror(errno));
     lock_release(tasks->lock);
 
+    if(workerq_length_threshold_percentage > 0) {
+	num_tasks = tasks->end - tasks->start;
+	length_percentage = num_tasks/tasks->max*100;
+	if(length_percentage > workerq_length_threshold_percentage) {
+	    LM_WARN("Queue length has exceeded length threshold percentage [%i] and is length [%i]", length_percentage, num_tasks);
+	}
+    }
     //int num_tasks = tasks->end - tasks->start;
     //LM_ERR("Added task to task queue.  Queue length [%i]", num_tasks);