@@ -1,3 +1,17 @@

+Thu Feb 12 19:19:42 EET 2009 (edwin)

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

+ * clamd/, shared/: merge from clamd-proto branch:     handle multiple %v parameters     introduce poll_fds that also does the read into a buffer     introduce thrmgr_group* commands for keeping track of groups of

+     commands (multiscan, IDSESSION)     introduce 2 queues in thrmgr, multiscan queues get lower

+     priority. Commands are processed from both queues in a 4:1 ratio

+     to avoid starvation.  unify reply code: conn_reply*     unify scanner code into scan_callback that is called from

+     cli_ftw multiscan doesn't need stat() now if readdir() provides

+     dt_type redesign clamd main loop: there is now an accept thread, and

+     a recv()/dispatch thread, +MaxThreads worker (scanner) threads.      introduce limiting on commands when worker threads are

+     contended.  introduce IDSESSION, a replacement for the old

+     SESSION command, that allows asyncronous (scan) commands and scan

+     replies introduce INSTREAM that allows sending the data to scan on

+     same connection (vs STREAM that requires another port)     introduce zCOMMANDS

+

 Thu Feb 12 19:08:28 EET 2009 (edwin)

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

  * libclamav/libclamav.map, libclamav/others.h,

@@ -99,7 +99,7 @@ int main(int argc, char **argv)

 	time_t currtime;

 	const char *dbdir, *cfgfile;

 	char *pua_cats = NULL, *pt;

-	int ret, tcpsock = 0, localsock = 0, i;

+	int ret, tcpsock = 0, localsock = 0, i, min_port, max_port;

 	unsigned int sigs = 0;

 	int lsockets[2], nlsockets = 0;

 	unsigned int dboptions = 0;

@@ -287,6 +287,15 @@ int main(int argc, char **argv)

     else

 	logg("#Log file size limit disabled.\n");

+    min_port = optget(opts, "StreamMinPort")->numarg;

+    max_port = optget(opts, "StreamMaxPort")->numarg;

+    if (min_port < 1024 || min_port > max_port || max_port > 65535) {

+	logg("!Invalid StreaMinPort/StreamMaxPort: %d, %d\n", min_port, max_port);

+	logg_close();

+	optfree(opts);

+	return 1;

+    }

+

     if(!(engine = cl_engine_new())) {

 	logg("!Can't initialize antivirus engine\n");

 	logg_close();

@@ -488,7 +497,7 @@ int main(int argc, char **argv)

         foreground = 1;

-    ret = acceptloop_th(lsockets, nlsockets, engine, dboptions, opts);

+    ret = recvloop_th(lsockets, nlsockets, engine, dboptions, opts);

 #ifdef C_WINDOWS

     if(tcpsock)

@@ -40,7 +40,7 @@

 #include "clamuko.h"

 struct dazuko_access *acc;

-

+short int clamuko_scanning;

 static void clamuko_exit(int sig)

 {

@@ -21,7 +21,7 @@

 #ifndef __CLAMUKO_H

 #define __CLAMUKO_H

-short int clamuko_scanning;

+extern short int clamuko_scanning;

 void *clamukoth(void *arg);

 #endif

@@ -80,9 +80,6 @@

 #include "session.h"

 #include "others.h"

-#define ENV_FILE  "CLAM_VIRUSEVENT_FILENAME"

-#define ENV_VIRUS "CLAM_VIRUSEVENT_VIRUSNAME"

-

 #ifdef	C_WINDOWS

 void virusaction(const char *filename, const char *virname, const struct optstruct *opts)

 {

@@ -91,14 +88,19 @@ void virusaction(const char *filename, const char *virname, const struct optstru

 }

 #else

+

+#define VE_FILENAME  "CLAM_VIRUSEVENT_FILENAME"

+#define VE_VIRUSNAME "CLAM_VIRUSEVENT_VIRUSNAME"

+

 static pthread_mutex_t virusaction_lock = PTHREAD_MUTEX_INITIALIZER;

 void virusaction(const char *filename, const char *virname, const struct optstruct *opts)

 {

 	pid_t pid;

 	const struct optstruct *opt;

-	char *buffer, *pt, *cmd, *buffer_file, *buffer_vir;

-	size_t j;

+	char *buffer_file, *buffer_vir, *buffer_cmd;

+	const char *pt;

+	size_t i, j, v = 0, len;

 	char *env[4];

 	if(!(opt = optget(opts, "VirusEvent"))->enabled)

@@ -107,336 +109,487 @@ void virusaction(const char *filename, const char *virname, const struct optstru

 	env[0] = getenv("PATH");

 	j = env[0] ? 1 : 0;

 	/* Allocate env vars.. to be portable env vars should not be freed */

-	buffer_file = (char *) malloc(strlen(ENV_FILE) + strlen(filename) + 2);

+	buffer_file = (char *) malloc(strlen(VE_FILENAME) + strlen(filename) + 2);

 	if(buffer_file) {

-		sprintf(buffer_file, "%s=%s", ENV_FILE, filename);

+		sprintf(buffer_file, "%s=%s", VE_FILENAME, filename);

 		env[j++] = buffer_file;

 	}

-	buffer_vir = (char *) malloc(strlen(ENV_VIRUS) + strlen(virname) + 2);

+	buffer_vir = (char *) malloc(strlen(VE_VIRUSNAME) + strlen(virname) + 2);

 	if(buffer_vir) {

-		sprintf(buffer_vir, "%s=%s", ENV_VIRUS, virname);

+		sprintf(buffer_vir, "%s=%s", VE_VIRUSNAME, virname);

 		env[j++] = buffer_vir;

 	}

 	env[j++] = NULL;

-	cmd = strdup(opt->strarg);

-

-	if(cmd && (pt = strstr(cmd, "%v"))) {

-		buffer = (char *) malloc(strlen(cmd) + strlen(virname) + 10);

-		if(buffer) {

-			*pt = 0; pt += 2;

-			strcpy(buffer, cmd);

-			strcat(buffer, virname);

-			strcat(buffer, pt);

-			free(cmd);

-			cmd = strdup(buffer);

-			free(buffer);

-		}

+	pt = opt->strarg;

+	while((pt = strstr(pt, "%v"))) {

+	    pt += 2;

+	    v++;

 	}

-

-	if(!cmd) {

-		free(buffer_file);

-		free(buffer_vir);

-		return;

+	len = strlen(opt->strarg);

+	buffer_cmd = (char *) calloc(len + v * strlen(virname) + 1, sizeof(char));

+	if(!buffer_cmd) {

+	    free(buffer_file);

+	    free(buffer_vir);

+	    return;

 	}

+	for(i = 0, j = 0; i < len; i++) {

+	    if(i + 1 < len && opt->strarg[i] == '%' && opt->strarg[i + 1] == 'v') {

+		strcat(buffer_cmd, virname);

+		j += strlen(virname);

+		i++;

+	    } else {

+		buffer_cmd[j++] = opt->strarg[i];

+	    }

+	}

+

 	pthread_mutex_lock(&virusaction_lock);

 	/* We can only call async-signal-safe functions after fork(). */

 	pid = fork();

-

-	if ( pid == 0 ) {

-		/* child... */

-		/* WARNING: this is uninterruptable ! */

-		exit(execle("/bin/sh", "sh", "-c", cmd, NULL, env));

-	} else if (pid > 0) {

-		pthread_mutex_unlock(&virusaction_lock);

-		/* parent */

-		waitpid(pid, NULL, 0);

+	if(!pid) { /* child */

+	    exit(execle("/bin/sh", "sh", "-c", buffer_cmd, NULL, env));

+	} else if(pid > 0) { /* parent */

+	    pthread_mutex_unlock(&virusaction_lock);

+	    waitpid(pid, NULL, 0);

 	} else {

-		pthread_mutex_unlock(&virusaction_lock);

-		/* error.. */

-		logg("!VirusAction: fork failed.\n");

+	    logg("!VirusEvent: fork failed.\n");

 	}

-	free(cmd);

+	free(buffer_cmd);

 	free(buffer_file);

 	free(buffer_vir);

 }

 #endif /* C_WINDOWS */

-int poll_fds(int *fds, int nfds, int timeout_sec, int check_signals)

+/* Function: writen

+	Try hard to write the specified number of bytes

+*/

+int writen(int fd, void *buff, unsigned int count)

 {

 	int retval;

-	int i;

-#ifdef HAVE_POLL

-	struct pollfd poll_1[1];

-	struct pollfd *poll_data = poll_1;

-

-    if (nfds>1) {

-	poll_data = malloc(nfds*sizeof(*poll_data));

-	if(!poll_data) {

-	    logg("!poll_fds: Can't allocate memory for poll_data\n");

-	    return -1;

-	}

-    }

-

-    for (i=0; i<nfds; i++) {

-	poll_data[i].fd = fds[i];

-	poll_data[i].events = POLLIN;

-	poll_data[i].revents = 0;

-    }

-

-    if (timeout_sec > 0) {

-    	timeout_sec *= 1000;

-    }

-    while (1) {

-    	retval = poll(poll_data, nfds, timeout_sec);

-	if (retval == -1) {

-	    if (errno == EINTR && !check_signals) {

+	unsigned int todo;

+	unsigned char *current;

+ 

+    todo = count;

+    current = (unsigned char *) buff;

+ 

+    do {

+	retval = write(fd, current, todo);

+	if (retval < 0) {

+	    if (errno == EINTR) {

 		continue;

 	    }

-	    if (nfds>1)

-		free(poll_data);

 	    return -1;

 	}

-	if (nfds>1) {

-	    if (retval>0) {

-		for (i=0; i<nfds; i++) {

-		    if (poll_data[i].revents) {

-			retval = i+1;

-			break;

-		    }

-		}

-	    }

-	    free(poll_data);

-	}

-	return retval;

-    }

-

-#else

-	fd_set rfds;

-	struct timeval tv;

-	int maxfd = 0;

-

-    for (i=0; i<nfds; i++) {

-#ifndef	C_WINDOWS

-	if (fds[i] >= DEFAULT_FD_SETSIZE) {

-	    return -1;

-	}

-#endif

-	if (fds[i] > maxfd)

-	    maxfd = fds[i];

-    }

-

-    while (1) {

-	FD_ZERO(&rfds);

-	for (i=0; i<nfds; i++)

-	    FD_SET(fds[i], &rfds);

-	tv.tv_sec = timeout_sec;

-	tv.tv_usec = 0;

+	todo -= retval;

+	current += retval;

+    } while (todo > 0);

+ 

+    return count;

+}

-	retval = select(maxfd+1, &rfds, NULL, NULL,

-			(timeout_sec>0 ? &tv : NULL));

-	if (retval == -1) {

-	    if (errno == EINTR && !check_signals) {

-		continue;

-	    }

-	    return -1;

-	}

-	if ((nfds>1) && (retval>0)) {

-	    for (i=0; i<nfds; i++) {

-		if (FD_ISSET(fds[i],&rfds)) {

-		    retval = i+1;

-		    break;

-		}

-	    }

-	}

-	return retval;

+static int realloc_polldata(struct fd_data *data)

+{

+#ifdef HAVE_POLL

+    if (data->poll_data_nfds == data->nfds)

+	return 0;

+    if (data->poll_data)

+	free(data->poll_data);

+    data->poll_data = malloc(data->nfds*sizeof(*data->poll_data));

+    if (!data->poll_data) {

+	logg("!realloc_polldata: Memory allocation failed for poll_data\n");

+	return -1;

     }

+    data->poll_data_nfds = data->nfds;

 #endif

-

-    return -1;

+    return 0;

 }

 int poll_fd(int fd, int timeout_sec, int check_signals)

 {

-    return poll_fds(&fd, 1, timeout_sec, check_signals);

+    int ret;

+    struct fd_data fds;

+

+    memset(&fds, 0, sizeof(fds));

+    if (fds_add(&fds, fd, 1) == -1)

+	return -1;

+    ret = fds_poll_recv(&fds, timeout_sec, check_signals);

+    fds_free(&fds);

+    return ret;

+}

+

+static void cleanup_fds(struct fd_data *data)

+{

+    struct fd_buf *newbuf;

+    unsigned i,j;

+    for (i=0,j=0;i < data->nfds; i++) {

+	if (data->buf[i].fd < 0) {

+	    if (data->buf[i].buffer)

+		free(data->buf[i].buffer);

+	    continue;

+	}

+	if (i != j)

+	    data->buf[j] = data->buf[i];

+	j++;

+    }

+    for (i = j ; i < data->nfds; i++)

+	data->buf[i].fd = -1;

+    data->nfds = j;

+    /* Shrink buffer */

+    newbuf = realloc(data->buf, j*sizeof(*newbuf));

+    if (newbuf)

+	data->buf = newbuf;/* non-fatal if shrink fails */

 }

-int is_fd_connected(int fd)

+static int read_fd_data(struct fd_buf *buf)

 {

-#ifdef HAVE_POLL

-	struct pollfd poll_data[1];

-	int count;

+    ssize_t n;

-    poll_data[0].fd = fd;

-    poll_data[0].events = POLLIN;

-    poll_data[0].revents = 0;

+    buf->got_newdata=1;

+    if (!buf->buffer) /* listen-only socket */

+	return 1;

-    if ((count=poll(poll_data, 1, 0)) == -1) {

-    	if (errno == EINTR) {

-		return 1;

-	}

-	return 0;

-    }

-    if (count == 0) {

-    	return 1;

-    }

-    if (poll_data[0].revents & POLLHUP) {

-	return 0;

-    }

-    if ((poll_data[0].revents & POLLIN) && (ioctl(fd, FIONREAD, &count) == 0)) {

-    	if (count == 0) {

-		return 0;

-	}

-    }

-    return 1;

+    if (buf->off >= buf->bufsize)

+	return -1;

+   /* Read the pending packet, it may contain more than one command, but

+    * that is to the cmdparser to handle. 

+    * It will handle 1st command, and then move leftover to beginning of buffer

+    */

+#ifdef HAVE_FD_PASSING

+  {

+      struct msghdr msg;

+      struct cmsghdr *cmsg;

+      unsigned char buff[CMSG_SPACE(sizeof(int))];

+      struct iovec iov[1];

+

+      if (buf->recvfd != -1) {

+	  close(buf->recvfd);

+	  buf->recvfd = -1;

+      }

+      memset(&msg, 0, sizeof(msg));

+      iov[0].iov_base = buf->buffer + buf->off;

+      iov[0].iov_len = buf->bufsize - buf->off;

+      msg.msg_iov = iov;

+      msg.msg_iovlen = 1;

+      msg.msg_control = buff;

+      msg.msg_controllen = sizeof(buff);

+

+      n = recvmsg(buf->fd, &msg, 0);

+      if (n < 0)

+	  return -1;

+      if ((msg.msg_flags & MSG_TRUNC) || (msg.msg_flags & MSG_CTRUNC)) {

+	  logg("!control message truncated");

+	  return -1;

+      }

+      for (cmsg = CMSG_FIRSTHDR(&msg); cmsg != NULL;

+	   cmsg = CMSG_NXTHDR(&msg, cmsg)) {

+	  if (cmsg->cmsg_len == CMSG_LEN(sizeof(int)) &&

+	      cmsg->cmsg_level == SOL_SOCKET &&

+	      cmsg->cmsg_type == SCM_RIGHTS) {

+	      if (buf->recvfd != -1) {

+		  logg("^Unclaimed file descriptor received. closing\n");

+		  close(buf->recvfd);

+	      }

+	      buf->recvfd = *(int *)CMSG_DATA(cmsg);

+	      logg("*RECVMSG: got FD %d\n", buf->recvfd);

+	  }

+      }

+  }

 #else

-	fd_set rfds;

-	struct timeval tv;

-	char buff[1];

+   n = recv(buf->fd, buf->buffer + buf->off, buf->bufsize - buf->off,0);

+   if (n < 0)

+       return -1;

+#endif

+   buf->off += n;

+   return n;

+}

-#ifndef	C_WINDOWS

-    if (fd >= DEFAULT_FD_SETSIZE) {

-        return 1;

+static int buf_init(struct fd_buf *buf, int listen_only)

+{

+    buf->off = 0;

+    buf->got_newdata = 0;

+    buf->recvfd = -1;

+    buf->mode = MODE_COMMAND;

+    buf->id = 0;

+    buf->dumpfd = -1;

+    buf->chunksize = 0;

+    buf->quota = 0;

+    buf->dumpname = NULL;

+    buf->group = NULL;

+    buf->term = '\0';

+    if (!listen_only) {

+	if (!buf->buffer) {

+	    buf->bufsize = PATH_MAX+8;

+	    /* plus extra space for a \0 so we can make sure every command is \0

+	     * terminated */

+	    if (!(buf->buffer = malloc(buf->bufsize + 1))) {

+		logg("!add_fd: Memory allocation failed for command buffer\n");

+		return -1;

+	    }

+	}

+    } else {

+	if (buf->buffer)

+	    free(buf->buffer);

+	buf->bufsize = 0;

+	buf->buffer = NULL;

     }

-#endif

+    return 0;

+}

-    FD_ZERO(&rfds);

-    FD_SET(fd, &rfds);

-    tv.tv_sec = 0;

-    tv.tv_usec = 0;

-    if (select(fd+1, &rfds, NULL, NULL, &tv) <= 0) {

-	return 1;

+int fds_add(struct fd_data *data, int fd, int listen_only)

+{

+    struct fd_buf *buf;

+    unsigned n;

+    if (fd < 0) {

+	logg("!add_fd: invalid fd passed to add_fd\n");

+	return -1;

     }

-    if (FD_ISSET(fd, &rfds)) {

-	if (recv(fd, buff, 1, MSG_PEEK) != 1) {

+    /* we may already have this fd, if

+     * the old FD got closed, and the kernel reused the FD */

+    for (n = 0; n < data->nfds; n++)

+	if (data->buf[n].fd == fd) {

+	    /* clear stale data in buffer */

+	    if (buf_init(&data->buf[n], listen_only) < 0)

+		return -1;

 	    return 0;

 	}

+

+    n++;

+    buf = realloc(data->buf, n*sizeof(*buf));

+    if (!buf) {

+	logg("!add_fd: Memory allocation failed for fd_buf\n");

+	return -1;

     }

-    return 1;

-#endif

+    data->buf = buf;

+    data->nfds = n;

+    data->buf[n-1].buffer = NULL;

+    if (buf_init(&data->buf[n-1], listen_only) < 0)

+	return -1;

+    data->buf[n-1].fd = fd;

+    return 0;

 }

-/* Function: writen

-	Try hard to write the specified number of bytes

-*/

-int writen(int fd, void *buff, unsigned int count)

+void fds_remove(struct fd_data *data, int fd)

 {

-	int retval;

-	unsigned int todo;

-	unsigned char *current;

- 

-    todo = count;

-    current = (unsigned char *) buff;

- 

-    do {

-	retval = write(fd, current, todo);

-	if (retval < 0) {

-	    if (errno == EINTR) {

-		continue;

+    size_t i;

+    pthread_mutex_lock(&data->buf_mutex);

+    if (data->buf) {

+	for (i=0;i<data->nfds;i++) {

+	    if (data->buf[i].fd == fd) {

+		data->buf[i].fd = -1;

+		break;

 	    }

-	    return -1;

 	}

-	todo -= retval;

-	current += retval;

-    } while (todo > 0);

- 

-    return count;

+    }

+    pthread_mutex_unlock(&data->buf_mutex);

 }

-/* FD Support Submitted by Richard Lyons <frob-clamav*webcentral.com.au> */

-/*

-   This procedure does timed clamd command and delimited input processing.  

-   It is complex for several reasons:

-       2) Newline delimited commands are indicated by a command which is prefixed by an 'n' character.  

-          This character serves to indicate that the command will contain a newline which will cause

-          command data to be read until the command input buffer is full or a newline is encountered.

-          Once the delimiter is encountered, the data is returned without the prefixing 'n' byte.

-       3) Legacy clamd clients presented commands which may or may not have been delimited by a newline.

-          If a command happens to be delimted by a newline, then only that command (and its newline) is

-          read and passed back, otherwise, all data read (in a single read) which fits in the specified

-          buffer will be returned.

-*/

-int readsock(int sockfd, char *buf, size_t size, unsigned char delim, int timeout_sec, int force_delim, int read_command)

+#ifndef	C_WINDOWS

+#define	closesocket(s)	close(s)

+#endif

+#define BUFFSIZE 1024

+/* Wait till data is available to be read on any of the fds,

+ * read available data on all fds, and mark them as appropriate.

+ * One of the fds should be a pipe, used by the accept thread to wake us.

+ * timeout is specified in seconds, if check_signals is non-zero, then

+ * poll_recv_fds() will return upon receipt of a signal, even if no data

+ * is received on any of the sockets.

+ * Must be called with buf_mutex held.

+ */

+/* TODO: handle ReadTimeout */

+int fds_poll_recv(struct fd_data *data, int timeout, int check_signals)

 {

-	ssize_t n;

-	size_t boff = 0;

-	char *pdelim;

-	time_t starttime, timenow;

-

-    time(&starttime);

-    while(1) {

-	time(&timenow);

-	switch(poll_fd(sockfd, (timeout_sec && ((timeout_sec-(timenow-starttime)) > 0)) ? timeout_sec-(timenow-starttime) : 0, 0)) {

-	    case 0: /* timeout */

-		return -2;

-	    case -1:

-		if(errno == EINTR)

+    unsigned fdsok = data->nfds;

+    size_t i;

+    int retval;

+

+    /* we must have at least one fd, the control fd! */

+    if (!data->nfds)

+	return 0;

+    for (i=0;i < data->nfds;i++)

+	data->buf[i].got_newdata = 0;

+#ifdef HAVE_POLL

+    /* Use poll() if available, preferred because:

+     *  - can poll any number of FDs

+     *  - can notify of both data available / socket disconnected events

+     *  - when it says POLLIN it is guaranteed that a following recv() won't

+     *  block (select may say that data is available to read, but a following 

+     *  recv() may still block according to the manpage

+     */

+

+    if (realloc_polldata(data) == -1)

+	return -1;

+    if (timeout > 0) {

+	/* seconds to ms */

+	timeout *= 1000;

+    }

+    for (i=0;i < data->nfds;i++) {

+	data->poll_data[i].fd = data->buf[i].fd;

+	data->poll_data[i].events = POLLIN;

+	data->poll_data[i].revents = 0;

+    }

+    do {

+	int n = data->nfds;

+

+	pthread_mutex_unlock(&data->buf_mutex);

+	retval = poll(data->poll_data, n, timeout);

+	pthread_mutex_lock(&data->buf_mutex);

+

+	if (retval > 0) {

+	    fdsok = 0;

+	    /* nfds may change during poll, but not

+	     * poll_data_nfds */

+	    for (i=0;i < data->poll_data_nfds; i++) {

+		short revents;

+		if (data->buf[i].fd < 0)

 		    continue;

-		return -1;

+		if (data->buf[i].fd != data->poll_data[i].fd) {

+		    /* should never happen */

+		    logg("!poll_recv_fds FD mismatch\n");

+		    continue;

+		}

+		revents = data->poll_data[i].revents;

+		if (revents & (POLLIN|POLLHUP)) {

+		    logg("*POLL: POLLIN|POLLHUP on fd %d\n",data->poll_data[i].fd);

+		}

+		if (revents & POLLIN) {

+		    int ret = read_fd_data(&data->buf[i]);

+		    /* Data available to be read */

+		    if (ret == -1)

+			revents |= POLLERR;

+		    else if (!ret)

+			revents = POLLHUP;

+		}

+

+		if (revents & (POLLHUP | POLLERR | POLLNVAL)) {

+		    if (revents & (POLLHUP| POLLNVAL)) {

+			/* remote disconnected */

+			logg("^poll_recv_fds: Client disconnected (FD %d)\n",

+			     data->poll_data[i].fd);

+		    } else {

+			/* error on file descriptor */

+			logg("!poll_recv_fds: Error condition on fd %d\n",

+			     data->poll_data[i].fd);

+		    }

+		    data->buf[i].got_newdata = -1;

+		} else {

+		    fdsok++;

+		}

+	    }

 	}

-	break;

-    }

-    n = recv(sockfd, buf, size, MSG_PEEK);

-    if(n < 0)

-	return -1;

-    if(read_command) {

-	if((n >= 1) && (buf[0] == 'n')) { /* Newline delimited command */

-	    force_delim = 1;

-	    delim = '\n';

+    } while (retval == -1 && !check_signals && errno == EINTR);

+#else

+    fd_set rfds;

+    struct timeval tv;

+    int maxfd = -1;

+

+    for (i=0;i < data->nfds; i++) {

+	int fd = data->buf[i].fd;

+	if (fd >= FD_SETSIZE) {

+	    logg ("!poll_recv_fds: file descriptor is not valid for FD_SET\n");

+	    return -1;

 	}

+

+	maxfd = MAX(maxfd, fd);

     }

-    while(boff < size) {

-	if(force_delim) {

-	    pdelim = memchr(buf, delim, n+boff);

-	    if(pdelim) {

-		n = recv(sockfd, buf+boff, pdelim-buf+1-boff, 0);

-		break;

-	    } else {

-		n = recv(sockfd, buf+boff, n, 0);

-		if(n < 0)

-		    return -1;

-		if((boff+n) == size)

-		    break;

-		boff += n;

+

+    do {

+	FD_ZERO(&rfds);

+	for(i=0;i < data->nfds;i++) {

+	    int fd = data->buf[i].fd;

+	    if (fd >= 0)

+		FD_SET(fd, &rfds);

+	}

+	tv.tv_sec = timeout;

+	tv.tv_usec = 0;

+

+	pthread_mutex_unlock(&data->buf_mutex);

+	retval = select(maxfd+1, &rfds, NULL, NULL, timeout > 0 ? &tv : NULL);

+	pthread_mutex_lock(&data->buf_mutex);

+	if (retval > 0) {

+	    fdsok = data->nfds;

+	    for (i=0; i < data->nfds; i++) {

+		if (data->buf[i].fd < 0) {

+		    fdsok--;

+		    continue;

+		}

+		if (FD_ISSET(data->buf[i].fd, &rfds)) {

+		    int ret = read_fd_data(&data->buf[i]);

+		    if (ret == -1 || !ret) {

+			if (ret == -1)

+			    logg("!poll_recv_fds: Error condition on fd %d\n",

+				 data->buf[i].fd);

+			else

+			    logg("^poll_recv_fds: Client disconnected\n");

+			data->buf[i].got_newdata = -1;

+		    }

+		}

 	    }

-	} else {

-	    pdelim = memchr(buf, delim, n+boff);

-	    if(pdelim)

-		n = recv(sockfd, buf+boff, pdelim-buf+1-boff, 0);

-	    else

-		n = recv(sockfd, buf+boff, size-boff, 0);

-	    break;

 	}

-	while(1) {

-	    time(&timenow);

-	    switch(poll_fd(sockfd, ((timeout_sec-(timenow-starttime)) > 0) ? timeout_sec-(timenow-starttime) : 0, 0)) {

-		case 0: /* timeout */

-		    return -2;

-		case -1:

-		    if(errno == EINTR)

-			continue;

-		    return -1;

+	if (retval < 0 && errno == EBADF) {

+	    /* unlike poll(),  select() won't tell us which FD is bad, so

+	     * we have to check them one by one. */

+	    tv.tv_sec = 0;

+	    tv.tv_usec = 0;

+	    /* with tv == 0 it doesn't check for EBADF */

+	    FD_ZERO(&rfds);

+	    for (i=0; i< data->nfds; i++) {

+		if (data->buf[i].fd == -1)

+		    continue;

+		FD_SET(data->buf[i].fd, &rfds);

+		do {

+		    retval = select(data->buf[i].fd+1, &rfds, NULL, NULL, &tv);

+		} while (retval == -1 && errno == EINTR);

+		if (retval == -1) {

+		    data->buf[i].fd = -1;

+		} else {

+		    FD_CLR(data->buf[i].fd, &rfds);

+		}

 	    }

-	    break;

+	    retval = -1;

+	    errno = EINTR;

+	    continue;

 	}

-        n = recv(sockfd, buf+boff, size-boff, MSG_PEEK);

-	if(n < 0)

-	    return -1;

-	if(n == 0)

-	    break;

+    } while (retval == -1 && !check_signals && errno == EINTR);

+#endif

+

+    if (retval == -1 && errno != EINTR) {

+	char buff[BUFFSIZE + 1];

+#ifdef HAVE_STRERROR_R

+	strerror_r(errno, buff, BUFFSIZE);

+#else

+	buff[0] = '\0';

+#endif

+#ifdef HAVE_POLL

+	logg("!poll_recv_fds: poll failed: %s\n", buff);

+#else

+	logg("!poll_recv_fds: select failed: %s\n", buff);

+#endif

     }

-    if(n < 0)

-	return -1;

-    n += boff;

-    if(read_command) {

-	if((n >= 1) && (buf[0] == 'n')) { /* Need to strip leading 'n' from command to attain standard command */

-	    --n;

-	    memcpy(buf, buf+1, n);

-	    buf[n] = '\0';

+

+    /* Remove closed / error fds */

+    if (fdsok != data->poll_data_nfds)

+	cleanup_fds(data);

+    return retval;

+}

+

+void fds_free(struct fd_data *data)

+{

+    unsigned i;

+    pthread_mutex_lock(&data->buf_mutex);

+    for (i=0;i < data->nfds;i++) {

+	if (data->buf[i].buffer) {

+	    free(data->buf[i].buffer);

 	}

-	return !strncmp(buf, "FD", 2) ? -1 : n; /* an explicit FD command is invalid */

     }

-    return n;

+    if (data->buf)

+	free(data->buf);

+#ifdef HAVE_POLL

+    if (data->poll_data)

+	free(data->poll_data);

+#endif

+    data->buf = NULL;

+    data->nfds = 0;

+    pthread_mutex_unlock(&data->buf_mutex);

 }

@@ -25,13 +25,51 @@

 #include <stdlib.h>

 #include "shared/optparser.h"

+#include "thrmgr.h"

+#include "cltypes.h"

+

+enum mode {

+    MODE_COMMAND,

+    MODE_STREAM,

+    MODE_WAITREPLY,

+    MODE_WAITANCILL

+};

+

+struct fd_buf {

+    unsigned char *buffer;

+    size_t bufsize;

+    size_t off;