@@ -1,3 +1,10 @@

+Wed Jun 22 16:43:43 CEST 2005 (tk)

+----------------------------------

+  * clamd: limit the internal queue of requests to MaxConnectionQueueLength

+	   and add new option MaxConnectionQueueTime (maximum number of seconds

+	   a connection can be queued and unprocessed without being aborted).

+	   Patch by Mark Pizzolato <clamav-devel*subscriptions.pizzolato.net>

+

 Sun Jun 19 13:11:20 BST 2005 (njh)

 ----------------------------------

   * libclamav/mbox.c:	Changed debug message when no text part is found in

@@ -126,7 +133,7 @@ Sat May 21 23:06:03 BST 2005 (njh)

 Sat May 21 02:17:08 CEST 2005 (tk)

 ----------------------------------

   * Remove the use of tmpfile(). Patch by Mark Pizzolato

-    <clamav-devel@subscriptions.pizzolato.net>, full list of changes:

+    <clamav-devel*subscriptions.pizzolato.net>, full list of changes:

   * libclamav/others.c

     - in cli_gentempname call cli_rndnum with 256 instead of 255 to get the

@@ -445,8 +445,12 @@ int scanstream(int odesc, unsigned long int *scanned, const struct cl_node *root

 	    break;

     }

-    lseek(tmpd, 0, SEEK_SET);

-    ret = cl_scandesc(tmpd, &virname, scanned, root, limits, options);

+    if(retval == 1) {

+	lseek(tmpd, 0, SEEK_SET);

+	ret = cl_scandesc(tmpd, &virname, scanned, root, limits, options);

+    } else {

+    	ret = -1;

+    }

     close(tmpd);

     if(!cfgopt(copt, "LeaveTemporaryFiles")->enabled)

 	unlink(tmpname);

@@ -460,8 +464,10 @@ int scanstream(int odesc, unsigned long int *scanned, const struct cl_node *root

 	logg("stream %d: %s FOUND\n", port, virname);

 	virusaction("stream", virname, copt);

     } else if(ret != CL_CLEAN) {

-	mdprintf(odesc, "stream: %s ERROR\n", cl_strerror(ret));

-	logg("stream %d: %s ERROR\n", port, cl_strerror(ret));

+    	if(retval == 1) {

+	    mdprintf(odesc, "stream: %s ERROR\n", cl_strerror(ret));

+	    logg("stream %d: %s ERROR\n", port, cl_strerror(ret));

+	}

     } else {

 	mdprintf(odesc, "stream: OK\n");

         if(logok)

@@ -45,12 +45,14 @@

 #define FALSE (0)

 #define TRUE (1)

-int progexit = 0;

-pthread_mutex_t exit_mutex;

-int reload = 0;

-time_t reloaded_time = 0;

-pthread_mutex_t reload_mutex;

-int sighup = 0;

+volatile int progexit = 0;

+static pthread_mutex_t exit_mutex = PTHREAD_MUTEX_INITIALIZER;

+volatile int reload = 0;

+volatile time_t reloaded_time = 0;

+static pthread_mutex_t reload_mutex = PTHREAD_MUTEX_INITIALIZER;

+volatile int sighup = 0;

+static pthread_mutex_t session_mutex = PTHREAD_MUTEX_INITIALIZER;

+volatile int session_count = 0;

 typedef struct client_conn_tag {

     int sd;

@@ -58,6 +60,7 @@ typedef struct client_conn_tag {

     const struct cfgstruct *copt;

     struct cl_node *root;

     time_t root_timestamp;

+    time_t queue_time;

     const struct cl_limits *limits;

     pid_t mainpid;

 } client_conn_t;

@@ -74,6 +77,24 @@ void scanner_thread(void *arg)

     sigfillset(&sigset);

     pthread_sigmask(SIG_SETMASK, &sigset, NULL);

+    timeout = cfgopt(conn->copt, "MaxConnectionQueueTime")->numarg;

+    if(timeout && (conn->queue_time+timeout < time(NULL))) {

+	logg("^Aborting Stale connection request\n");

+	close(conn->sd);

+	cl_free(conn->root);

+	free(conn);

+	return;

+    }

+    pthread_mutex_lock(&exit_mutex);

+    if(progexit) {

+	pthread_mutex_unlock(&exit_mutex);

+	logg("^Dropping connection request to exit\n");

+	close(conn->sd);

+	cl_free(conn->root);

+	free(conn);

+	return;

+    }

+    pthread_mutex_unlock(&exit_mutex);

     timeout = cfgopt(conn->copt, "ReadTimeout")->numarg;

     if(!timeout)

     	timeout = -1;

@@ -99,28 +120,52 @@ void scanner_thread(void *arg)

 		break;

 	    case COMMAND_SESSION:

-		session = TRUE;

+		if(!session) {

+		    pthread_mutex_lock(&session_mutex);

+		    session_count++;

+		    if(cfgopt(conn->copt, "MaxThreads")->numarg <= session_count) {

+			logg("^Active sessions(%d) consuming all available threads(%d)\n", session_count, cfgopt(conn->copt, "MaxThreads")->numarg);

+		    }

+		    pthread_mutex_unlock(&session_mutex);

+		    session = TRUE;

+		}

 		timeout = 5;

 		break;

 	    case COMMAND_END:

-		session = FALSE;

+		if(session) {

+		    pthread_mutex_lock(&session_mutex);

+		    session_count--;

+		    pthread_mutex_unlock(&session_mutex);

+		    session = FALSE;

+		}

 		break;

 	}

 	if (session) {

 	    pthread_mutex_lock(&exit_mutex);

 	    if(progexit) {

 		session = FALSE;

+		pthread_mutex_lock(&session_mutex);

+		--session_count;

+		pthread_mutex_unlock(&session_mutex);

 	    }

 	    pthread_mutex_unlock(&exit_mutex);

 	    pthread_mutex_lock(&reload_mutex);

 	    if (conn->root_timestamp != reloaded_time) {

 		session = FALSE;

+		pthread_mutex_lock(&session_mutex);

+		--session_count;

+		pthread_mutex_unlock(&session_mutex);

 	    }

 	    pthread_mutex_unlock(&reload_mutex);

 	}

     } while (session);

+    if(session) {

+	pthread_mutex_lock(&session_mutex);

+	--session_count;

+	pthread_mutex_unlock(&session_mutex);

+    }

     close(conn->sd);

     cl_free(conn->root);

     free(conn);

@@ -215,7 +260,7 @@ static struct cl_node *reload_db(struct cl_node *root, const struct cfgstruct *c

 int acceptloop_th(int socketd, struct cl_node *root, const struct cfgstruct *copt)

 {

-	int new_sd, max_threads;

+	int new_sd, max_threads, max_queue_size;

 	unsigned int options = 0;

 	threadpool_t *thr_pool;

 	struct sigaction sigact;

@@ -261,6 +306,8 @@ int acceptloop_th(int socketd, struct cl_node *root, const struct cfgstruct *cop

     logg("*Listening daemon: PID: %d\n", getpid());

     max_threads = cfgopt(copt, "MaxThreads")->numarg;

+    max_queue_size = cfgopt(copt, "MaxConnectionQueueLength")->numarg;

+

     if(cfgopt(copt, "ScanArchive")->enabled || cfgopt(copt, "ClamukoScanArchive")->enabled) {

 	/* set up limits */

@@ -430,21 +477,21 @@ int acceptloop_th(int socketd, struct cl_node *root, const struct cfgstruct *cop

     idletimeout = cfgopt(copt, "IdleTimeout")->numarg;

-    if((thr_pool=thrmgr_new(max_threads, idletimeout, scanner_thread)) == NULL) {

+    if((thr_pool=thrmgr_new(max_threads, max_queue_size, idletimeout, scanner_thread)) == NULL) {

 	logg("!thrmgr_new failed\n");

 	exit(-1);

     }

     time(&start_time);

-    for(;;) {				

+    for(;;) {

 	new_sd = accept(socketd, NULL, NULL);

 	if((new_sd == -1) && (errno != EINTR)) {

 	    /* very bad - need to exit or restart */

 #ifdef HAVE_STRERROR_R

 	    logg("!accept() failed: %s\n", strerror_r(errno, buff, BUFFSIZE));

 #else

-	    logg("!accept() failed\n");

+	    logg("!accept() failed: %d\n", errno);

 #endif

 	    continue;

 	}

@@ -464,6 +511,7 @@ int acceptloop_th(int socketd, struct cl_node *root, const struct cfgstruct *cop

 		client_conn->copt = copt;

 		client_conn->root = cl_dup(root);

 		client_conn->root_timestamp = reloaded_time;

+		time(&client_conn->queue_time);

 		client_conn->limits = &limits;

 		client_conn->mainpid = mainpid;

 		if (!thrmgr_dispatch(thr_pool, client_conn)) {

@@ -501,7 +549,7 @@ int acceptloop_th(int socketd, struct cl_node *root, const struct cfgstruct *cop

 	    root = reload_db(root, copt, FALSE);

 	    pthread_mutex_lock(&reload_mutex);

 	    reload = 0;

-	    time(&reloaded_time);

+	    time((time_t *)&reloaded_time);

 	    pthread_mutex_unlock(&reload_mutex);

 #ifdef CLAMUKO

 	    if(cfgopt(copt, "ClamukoScanOnAccess")->enabled) {

@@ -43,7 +43,7 @@

 #include "output.h"

 #include "memory.h"

-pthread_mutex_t ctime_mutex = PTHREAD_MUTEX_INITIALIZER;

+static pthread_mutex_t ctime_mutex = PTHREAD_MUTEX_INITIALIZER;

 int command(int desc, const struct cl_node *root, const struct cl_limits *limits, int options, const struct cfgstruct *copt, int timeout)

 {

@@ -40,14 +40,17 @@ work_queue_t *work_queue_new()

 	return work_q;

 }

-void work_queue_add(work_queue_t *work_q, void *data)

+int work_queue_add(work_queue_t *work_q, void *data)

 {

 	work_item_t *work_item;

 	if (!work_q) {

-		return;

+		return FALSE;

 	}

 	work_item = (work_item_t *) mmalloc(sizeof(work_item_t));

+	if (!work_item) {

+		return FALSE;

+	}

 	work_item->next = NULL;

 	work_item->data = data;

 	gettimeofday(&(work_item->time_queued), NULL);

@@ -60,7 +63,7 @@ void work_queue_add(work_queue_t *work_q, void *data)

 		work_q->tail = work_item;

 		work_q->item_count++;

 	}

-	return;

+	return TRUE;

 }

 void *work_queue_pop(work_queue_t *work_q)

@@ -112,12 +115,13 @@ void thrmgr_destroy(threadpool_t *threadpool)

 	pthread_mutex_destroy(&(threadpool->pool_mutex));

 	pthread_cond_destroy(&(threadpool->pool_cond));

+	pthread_cond_destroy(&(threadpool->pool_notfull_cond));

 	pthread_attr_destroy(&(threadpool->pool_attr));

 	free(threadpool);

 	return;

 }

-threadpool_t *thrmgr_new(int max_threads, int idle_timeout, void (*handler)(void *))

+threadpool_t *thrmgr_new(int max_threads, int max_dispatch_queue, int idle_timeout, void (*handler)(void *))

 {

 	threadpool_t *threadpool;

@@ -133,6 +137,8 @@ threadpool_t *thrmgr_new(int max_threads, int idle_timeout, void (*handler)(void

 		return NULL;

 	}	

 	threadpool->thr_max = max_threads;

+	threadpool->thr_max_queue = max_dispatch_queue;

+	threadpool->thr_queued = 0;

 	threadpool->thr_alive = 0;

 	threadpool->thr_idle = 0;

 	threadpool->idle_timeout = idle_timeout;

@@ -143,8 +149,15 @@ threadpool_t *thrmgr_new(int max_threads, int idle_timeout, void (*handler)(void

 		free(threadpool);

 		return NULL;

 	}

+	if (pthread_cond_init(&(threadpool->pool_notfull_cond), NULL) != 0) {

+		pthread_cond_destroy(&(threadpool->pool_cond));

+		free(threadpool);

+		return NULL;

+	}

 	if (pthread_attr_init(&(threadpool->pool_attr)) != 0) {

+		pthread_cond_destroy(&(threadpool->pool_cond));

+		pthread_cond_destroy(&(threadpool->pool_notfull_cond));

 		free(threadpool);

 		return NULL;

 	}

@@ -162,7 +175,7 @@ void *thrmgr_worker(void *arg)

 {

 	threadpool_t *threadpool = (threadpool_t *) arg;

 	void *job_data;

-	int retval, must_exit = FALSE;

+	int retval;

 	struct timespec timeout;

 	/* loop looking for work */

@@ -181,15 +194,20 @@ void *thrmgr_worker(void *arg)

 			retval = pthread_cond_timedwait(&(threadpool->pool_cond),

 				&(threadpool->pool_mutex), &timeout);

 			if (retval == ETIMEDOUT) {

-				must_exit = TRUE;

 				break;

 			}

 		}

 		threadpool->thr_idle--;

-		if (threadpool->state == POOL_EXIT) {

-			must_exit = TRUE;

+

+		if (job_data) {

+			if ((threadpool->thr_max_queue > 0) &&

+			    (threadpool->thr_queued == threadpool->thr_max_queue)) {

+				logg("*Thread Queue: Resuming...\n");

+				/* signal that queue no longer full */

+				pthread_cond_signal(&threadpool->pool_notfull_cond);

+			}

+			threadpool->thr_queued--;

 		}

-		

 		if (pthread_mutex_unlock(&(threadpool->pool_mutex)) != 0) {

 			/* Fatal error */

 			logg("!Fatal: mutex unlock failed\n");

@@ -197,7 +215,7 @@ void *thrmgr_worker(void *arg)

 		}

 		if (job_data) {

 			threadpool->handler(job_data);

-		} else if (must_exit) {

+		} else {

 			break;

 		}

 	}

@@ -229,33 +247,81 @@ int thrmgr_dispatch(threadpool_t *threadpool, void *user_data)

 	/* Lock the threadpool */

 	if (pthread_mutex_lock(&(threadpool->pool_mutex)) != 0) {

-		logg("!Mutex lock failed\n");

+		logg("!Mutex lock failed: %d\n", errno);

 		return FALSE;

 	}

 	if (threadpool->state != POOL_VALID) {

 		if (pthread_mutex_unlock(&(threadpool->pool_mutex)) != 0) {

-			logg("!Mutex unlock failed\n");

+			logg("!Mutex unlock failed (!POOL_VALID): %d\n", errno);

 			return FALSE;

 		}

 		return FALSE;

 	}

-	work_queue_add(threadpool->queue, user_data);

+	if (work_queue_add(threadpool->queue, user_data) == FALSE) {

+		if (pthread_mutex_unlock(&(threadpool->pool_mutex)) != 0) {

+			logg("!Mutex unlock failed (work_queue_add): %d\n", errno);

+			return FALSE;

+		}

+		return FALSE;

+	}

+	threadpool->thr_queued++;

+	while ((threadpool->thr_max_queue > 0) && 

+		(threadpool->thr_queued >= threadpool->thr_max_queue) &&

+		(threadpool->state == POOL_VALID)) {

+		logg("*Thread Queue: Full, waiting...\n");

+		if (pthread_cond_wait(&threadpool->pool_notfull_cond, &threadpool->pool_mutex) != 0) {

+			logg("!pthread_cond_wait failed: %d\n", errno);

+			pthread_mutex_unlock(&threadpool->pool_mutex);

+			return FALSE;

+		}

+	}

 	if ((threadpool->thr_idle == 0) &&

-			(threadpool->thr_alive < threadpool->thr_max)) {

+	    (threadpool->thr_alive < threadpool->thr_max) &&

+	    (threadpool->state == POOL_VALID)) {

 		/* Start a new thread */

 		if (pthread_create(&thr_id, &(threadpool->pool_attr),

 				thrmgr_worker, threadpool) != 0) {

-			logg("!pthread_create failed\n");

+			logg("!pthread_create failed: %d\n", errno);

 		} else {

 			threadpool->thr_alive++;

 		}

 	}

-	pthread_cond_signal(&(threadpool->pool_cond));

+	switch(threadpool->thr_idle) {

+		case 0:

+			break;

+		case 1:

+			if (pthread_cond_signal(&(threadpool->pool_cond)) != 0) {

+				logg("!pthread_cond_signal failed: %d\n", errno);

+				if (pthread_mutex_unlock(&(threadpool->pool_mutex)) != 0) {

+					logg("!Mutex unlock failed (after pthread_cond_signal failure): %d\n", errno);

+				}

+				return FALSE;

+			}

+			break;

+		default:

+			if(threadpool->thr_queued > 1) {

+				if (pthread_cond_broadcast(&(threadpool->pool_cond)) != 0) {

+					logg("!pthread_cond_broadcast failed: %d\n", errno);

+					if (pthread_mutex_unlock(&(threadpool->pool_mutex)) != 0) {

+						logg("!Mutex unlock failed (after pthread_cond_broadcast failure): %d\n", errno);

+					}

+					return FALSE;

+				}

+			} else {

+				if (pthread_cond_signal(&(threadpool->pool_cond)) != 0) {

+					logg("!pthread_cond_signal failed: %d\n", errno);

+					if (pthread_mutex_unlock(&(threadpool->pool_mutex)) != 0) {

+						logg("!Mutex unlock failed (after pthread_cond_signal failure): %d\n", errno);

+					}

+					return FALSE;

+				}

+			}

+	}

 	if (pthread_mutex_unlock(&(threadpool->pool_mutex)) != 0) {

-		logg("!Mutex unlock failed\n");

+		logg("!Mutex unlock failed (before complete): %d\n", errno);

 		return FALSE;

 	}

 	return TRUE;

@@ -43,12 +43,15 @@ typedef enum {

 typedef struct threadpool_tag {

 	pthread_mutex_t pool_mutex;

 	pthread_cond_t pool_cond;

+	pthread_cond_t pool_notfull_cond;

 	pthread_attr_t pool_attr;

-	pool_state_t state;

-	int thr_max;

-	int thr_alive;

-	int thr_idle;

+	volatile pool_state_t state;

+	volatile int thr_max;

+	volatile int thr_max_queue;

+	volatile int thr_queued;

+	volatile int thr_alive;

+	volatile int thr_idle;

 	int idle_timeout;

 	void (*handler)(void *);

@@ -56,7 +59,7 @@ typedef struct threadpool_tag {

 	work_queue_t *queue;

 } threadpool_t;

-threadpool_t *thrmgr_new(int max_threads, int idle_timeout, void (*handler)(void *));

+threadpool_t *thrmgr_new(int max_threads, int max_dispatch_queue, int idle_timeout, void (*handler)(void *));

 void thrmgr_destroy(threadpool_t *threadpool);

 int thrmgr_dispatch(threadpool_t *threadpool, void *user_data);

@@ -104,6 +104,11 @@ Maximum length the queue of pending connections may grow to.

 .br 

 Default: 15

 .TP 

+\fBMaxConnectionQueueTime NUMBER\fR

+Maximum number of seconds a connection can be queued and unprocessed without being aborted when it finally comes time to process it. The value of zero disables the limit.

+.br 

+Default: 20

+.TP 

 \fBMaxThreads NUMBER\fR

 Maximal number of threads running at the same time.

 .br 

@@ -90,6 +90,12 @@ LocalSocket /tmp/clamd

 # Default: 15

 #MaxConnectionQueueLength 30

+# Maximum number of seconds a connection can be queued and unprocessed without

+# being aborted when it finally comes time to process it. The value of zero

+# disables the limit.

+# Default: 20

+#MaxConnectionQueueTime 30

+

 # Clamd uses FTP-like protocol to receive data from remote clients.

 # If you are using clamav-milter to balance load between remote clamd daemons

 # on firewall servers you may need to tune the options below.

@@ -45,7 +45,7 @@ extern short cli_debug_flag;

 #ifdef CL_THREAD_SAFE

 #  include <pthread.h>

-pthread_mutex_t cli_ref_mutex = PTHREAD_MUTEX_INITIALIZER;

+static pthread_mutex_t cli_ref_mutex = PTHREAD_MUTEX_INITIALIZER;

 #endif

 int cli_scanbuff(const char *buffer, unsigned int length, const char **virname, const struct cl_node *root, unsigned short ftype)

@@ -42,7 +42,7 @@

 #ifdef CL_THREAD_SAFE

 #  include <pthread.h>

-pthread_mutex_t cli_gentempname_mutex = PTHREAD_MUTEX_INITIALIZER;

+static pthread_mutex_t cli_gentempname_mutex = PTHREAD_MUTEX_INITIALIZER;

 #endif

 #if defined(HAVE_READDIR_R_3) || defined(HAVE_READDIR_R_2)

@@ -72,6 +72,7 @@ struct cfgstruct *getcfg(const char *cfgfile, int verbose)

 	    {"TCPSocket", OPT_NUM, -1, NULL, 0},

 	    {"LocalSocket", OPT_STR, -1, NULL, 0},

 	    {"MaxConnectionQueueLength", OPT_NUM, 15, NULL, 0},

+	    {"MaxConnectionQueueTime", OPT_NUM, 20, NULL, 0},

 	    {"StreamMaxLength", OPT_COMPSIZE, 10485760, NULL, 0},

 	    {"StreamMinPort", OPT_NUM, 1024, NULL, 0},

 	    {"StreamMaxPort", OPT_NUM, 2048, NULL, 0},