open-dsi
/
clamav
mirror of https://github.com/Cisco-Talos/clamav
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

recvloop. This doesn't compile yet.

Browse Source 
But perhaps it is more portable to use a single select for accept/recv, then
do accept on listen sockets, and recv on open sockets.
This  should be more portable than select on a pipe.
We can also do the operation with one mutex less.

git-svn-id: file:///var/lib/svn/clamav-devel/branches/clamd-proto@4598 77e5149b-7576-45b1-b177-96237e5ba77b
0.95
Török Edvin 17 years ago
parent 3a14515741
commit d2417fbf93
4 changed files with 203 additions and 94 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 8
      ChangeLog
  2. 10
      clamd/others.c
  3. 2
      clamd/others.h
  4. 277
      clamd/server-th.c

8
ChangeLog
Unescape View File

@ -1,3 +1,11 @@
Thu Jan 8 23:57:00 EET 2009 (edwin)
------------------------------------
* clamd/others.c, clamd/others.h, clamd/server-th.c: recvloop. This
doesn't compile yet. But perhaps it is more portable to use a
single select for accept/recv, then do accept on listen sockets, and
recv on open sockets. This should be more portable than select on
a pipe. We can also do the operation with one mutex less.
Thu Jan 8 22:48:43 EET 2009 (edwin)
------------------------------------
* clamd/server-th.c: accept thread continued

10
clamd/others.c
Unescape View File

@ -644,10 +644,16 @@ int fds_poll_recv(struct fd_data *data, int timeout, int check_signals)
#endif
if (retval == -1) {
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\n");
logg("!poll_recv_fds: poll failed: %s\n", buff);
#else
logg("!poll_recv_fds: select failed\n");
logg("!poll_recv_fds: select failed: %s\n", buff);
#endif
}

2
clamd/others.h
Unescape View File

@ -51,7 +51,7 @@ int writen(int fd, void *buff, unsigned int count);
int readsock(int sockfd, char *buf, size_t size, unsigned char delim, int timeout_sec, int force_delim, int read_command);
int fds_add(struct fd_data *data, int fd, int listen_only);
int fds_poll_recv(struct fd_data *data, int timeout, int check_signals)
int fds_poll_recv(struct fd_data *data, int timeout, int check_signals);
void fds_free(struct fd_data *data);
#endif

277
clamd/server-th.c
Unescape View File

@ -78,13 +78,12 @@ int sighup = 0;
static struct cl_stat *dbstat = NULL;
typedef struct client_conn_tag {
const char *cmd;
int sd;
unsigned int options;
const struct optstruct *opts;
struct cl_engine *engine;
time_t engine_timestamp;
int *socketds;
int nsockets;
} client_conn_t;
static void scanner_thread(void *arg)
@ -281,6 +280,151 @@ static struct cl_engine *reload_db(struct cl_engine *engine, unsigned int dbopti
return engine;
}
struct rcvloop {
threadpool_t *thr_pool;
struct fd_data *fds;
};
static const char *get_cmd(struct fd_buf *buf)
{
unsigned char *pos;
if (!buf->off)
return NULL;
switch (buf->buffer[0]) {
/* commands terminated by delimiters */
case 'n':
case 'z':
pos = memchr(buf->buffer, buf->buffer[0], buf->off);
if (!pos)
/* we don't have another full command yet */
return NULL;
*pos = '\0';
return buf->buffer + 1;
default:
/* one packet = one command */
return buf->buffer;
}
}
void* recvloop_th(void *arg)
{
struct rcvloop *rcvloop = (struct rcvloop*)arg;
struct fd_data *fds = rcvloop->fds;
pthread_mutex_lock(fds->buf_mutex);
for (;;) {
size_t i;
int status = fds_poll_recv(fds, -1, 1);
if (status == -1) {
/* very bad, already logged */
continue;
}
/* Loop over all sockets, checking which have a new command
* available */
for (i=0;i < fds->nfds; i++) {
client_conn_t *client_conn;
struct fd_buf *buf = &fds->buf[i];
const char *cmd;
if (buf->fd == -1 || !buf->got_newdata ||
!has_fullcmd(buf))
continue;
/* Parse commands */
while ((cmd = get_cmd(buf) != NULL)) {
client_conn = (client_conn_t *) malloc(sizeof(struct client_conn_tag));
if(client_conn) {
client_conn->sd = fds->buf[i].fd;
client_conn->fds = fds;
client_conn->cmd = cmd;
client_conn->options = options;
client_conn->opts = opts;
if(cl_engine_addref(engine)) {
closesocket(client_conn->sd);
free(client_conn);
logg("!cl_engine_addref() failed\n");
pthread_mutex_lock(&exit_mutex);
progexit = 1;
pthread_mutex_unlock(&exit_mutex);
} else {
client_conn->engine = engine;
client_conn->engine_timestamp = reloaded_time;
client_conn->socketds = socketds;
client_conn->nsockets = nsockets;
if(!thrmgr_dispatch(thr_pool, client_conn)) {
closesocket(client_conn->sd);
free(client_conn);
logg("!thread dispatch failed\n");
}
}
} else {
logg("!Can't allocate memory for client_conn\n");
closesocket(new_sd);
if(optget(opts, "ExitOnOOM")->enabled) {
pthread_mutex_lock(&exit_mutex);
progexit = 1;
pthread_mutex_unlock(&exit_mutex);
}
}
}
/* SIGHUP */
if (sighup) {
logg("SIGHUP caught: re-opening log file.\n");
logg_close();
sighup = 0;
if(!logg_file && (opt = optget(opts, "LogFile"))->enabled)
logg_file = opt->strarg;
}
/* SelfCheck */
if(selfchk) {
time(&current_time);
if((current_time - start_time) > (time_t)selfchk) {
if(reload_db(engine, dboptions, opts, TRUE, &ret)) {
pthread_mutex_lock(&reload_mutex);
reload = 1;
pthread_mutex_unlock(&reload_mutex);
}
time(&start_time);
}
}
/* DB reload */
pthread_mutex_lock(&reload_mutex);
if(reload) {
pthread_mutex_unlock(&reload_mutex);
engine = reload_db(engine, dboptions, opts, FALSE, &ret);
if(ret) {
logg("Terminating because of a fatal error.\n");
if(new_sd >= 0)
closesocket(new_sd);
break;
}
pthread_mutex_lock(&reload_mutex);
reload = 0;
time(&reloaded_time);
pthread_mutex_unlock(&reload_mutex);
#ifdef CLAMUKO
if(optget(opts, "ClamukoScanOnAccess")->enabled && tharg) {
logg("Stopping and restarting Clamuko.\n");
pthread_kill(clamuko_pid, SIGUSR1);
pthread_join(clamuko_pid, NULL);
tharg->engine = engine;
pthread_create(&clamuko_pid, &clamuko_attr, clamukoth, tharg);
}
#endif
} else {
pthread_mutex_unlock(&reload_mutex);
}
}
}
pthread_mutex_unlock(fds->buf_mutex);
return NULL
}
int acceptloop_th(int *socketds, int nsockets, struct cl_engine *engine, unsigned int dboptions, const struct optstruct *opts)
{
int max_threads, i, ret = 0;
@ -305,6 +449,9 @@ int acceptloop_th(int *socketds, int nsockets, struct cl_engine *engine, unsigne
uint32_t val32;
uint64_t val64;
struct fd_data fds;
pthread_t rcvth;
struct rcvloop rcvloop;
int notifypip[2]; /* notify recvloop that it has a new fd */
#ifdef CLAMUKO
pthread_t clamuko_pid;
@ -600,6 +747,18 @@ int acceptloop_th(int *socketds, int nsockets, struct cl_engine *engine, unsigne
idletimeout = optget(opts, "IdleTimeout")->numarg;
memset(&fds, 0, sizeof(fds));
if (pipe(notifypip, 0) == -1) {
logg("!pipe failed\n");
cl_engine_free(engine);
return 1;
}
if (add_fd_to_poll(&fds, notifypip[0]) == -1) {
logg("!add_fd_to_poll failed failed\n");
cl_engine_free(engine);
return 1;
}
for (i=0;i < nsockets;i++)
if (add_fd_to_poll(&fds, socketds[i], 1) == -1) {
logg("!add_fd_to_poll failed failed\n");
@ -607,15 +766,31 @@ int acceptloop_th(int *socketds, int nsockets, struct cl_engine *engine, unsigne
return 1;
}
if((thr_pool=thrmgr_new(max_threads, idletimeout, scanner_thread)) == NULL) {
if ((thr_pool=thrmgr_new(max_threads, idletimeout, scanner_thread)) == NULL) {
logg("!thrmgr_new failed\n");
exit(-1);
}
rcvloop.thr_pool = &thr_pool;
rcvloop.fds = &fds;
if (pthread_create(&rcvth, NULL, recvloop_th, &rcvloop)) {
logg("!pthread_create failed\n");
exit(-1);
}
if (pthread_detach(&rcvth)) {
logg("!pthread_detach failed\n");
exit(-1);
}
time(&start_time);
pthread_mutex_lock(&fds.buf_mutex);
for(;;) {
int new_sd = fds_poll_recv(&fds, -1, 1);
/* Block waiting for connection on any of the sockets,
* doesn't wake on signals, that is what recvloop does! */
int new_sd = fds_poll_recv(&fds, -1, 0);
/* TODO: what about sockets that get rm-ed? */
if (!fds.nfds) {
/* no more sockets to poll, all gave an error */
@ -628,48 +803,18 @@ int acceptloop_th(int *socketds, int nsockets, struct cl_engine *engine, unsigne
pthread_mutex_lock(&exit_mutex);
if (prog_exit) {
pthread_mutex_unlock(&exit_mutex);
#ifdef C_WINDOWS
closesocket(new_sd);
#else
if(new_sd >= 0)
close(new_sd);
#endif
closesocket(new_sd);
break;
}
pthread_mutex_unlock(&exit_mutex);
if (new_sd >= 0) {
client_conn = (client_conn_t *) malloc(sizeof(struct client_conn_tag));
if(client_conn) {
client_conn->sd = new_sd;
client_conn->options = options;
client_conn->opts = opts;
if(cl_engine_addref(engine)) {
closesocket(client_conn->sd);
free(client_conn);
logg("!cl_engine_addref() failed\n");
pthread_mutex_lock(&exit_mutex);
progexit = 1;
pthread_mutex_unlock(&exit_mutex);
} else {
client_conn->engine = engine;
client_conn->engine_timestamp = reloaded_time;
client_conn->socketds = socketds;
client_conn->nsockets = nsockets;
if(!thrmgr_dispatch(thr_pool, client_conn)) {
closesocket(client_conn->sd);
free(client_conn);
logg("!thread dispatch failed\n");
}
}
} else {
logg("!Can't allocate memory for client_conn\n");
closesocket(new_sd);
if(optget(opts, "ExitOnOOM")->enabled) {
pthread_mutex_lock(&exit_mutex);
progexit = 1;
pthread_mutex_unlock(&exit_mutex);
}
char dummy = 'x';
fds_add(&fds, new_sd, 0);
if (write(notifypip[1], &dummy, 1) == -1) {
/* failed to notify */
logg("!acceptloop_th: failed to notify recvloop\n");
}
}
}
@ -691,58 +836,8 @@ int acceptloop_th(int *socketds, int nsockets, struct cl_engine *engine, unsigne
continue;
}
if (sighup) {
logg("SIGHUP caught: re-opening log file.\n");
logg_close();
sighup = 0;
if(!logg_file && (opt = optget(opts, "LogFile"))->enabled)
logg_file = opt->strarg;
}
if(selfchk) {
time(&current_time);
if((current_time - start_time) > (time_t)selfchk) {
if(reload_db(engine, dboptions, opts, TRUE, &ret)) {
pthread_mutex_lock(&reload_mutex);
reload = 1;
pthread_mutex_unlock(&reload_mutex);
}
time(&start_time);
}
}
pthread_mutex_lock(&reload_mutex);
if(reload) {
pthread_mutex_unlock(&reload_mutex);
engine = reload_db(engine, dboptions, opts, FALSE, &ret);
if(ret) {
logg("Terminating because of a fatal error.\n");
#ifdef C_WINDOWS
closesocket(new_sd);
#else
if(new_sd >= 0)
close(new_sd);
#endif
break;
}
pthread_mutex_lock(&reload_mutex);
reload = 0;
time(&reloaded_time);
pthread_mutex_unlock(&reload_mutex);
#ifdef CLAMUKO
if(optget(opts, "ClamukoScanOnAccess")->enabled && tharg) {
logg("Stopping and restarting Clamuko.\n");
pthread_kill(clamuko_pid, SIGUSR1);
pthread_join(clamuko_pid, NULL);
tharg->engine = engine;
pthread_create(&clamuko_pid, &clamuko_attr, clamukoth, tharg);
}
#endif
} else {
pthread_mutex_unlock(&reload_mutex);
}
}
pthread_mutex_unlock(&fds.buf_mutex);
fds_free(&fds);
/* Destroy the thread manager.

Write Preview
Loading…
Cancel
Save