2 * Copyright (c) 2003-2004, 2007, 2009-2011 Sendmail, Inc. and its suppliers.
5 * By using this file, you agree to the terms and conditions set
6 * forth in the LICENSE file which can be found at the top level of
7 * the sendmail distribution.
9 * Contributed by Jose Marcio Martins da Cruz - Ecole des Mines de Paris
10 * Jose-Marcio.Martins@ensmp.fr
14 SM_RCSID("@(#)$Id: worker.c,v 8.19 2011/02/14 23:33:48 ca Exp $")
16 #include "libmilter.h"
20 typedef struct taskmgr_S taskmgr_T;
22 #define TM_SIGNATURE 0x23021957
26 long tm_signature; /* has the controller been initialized */
27 sthread_t tm_tid; /* thread id of controller */
28 smfi_hd_T tm_ctx_head; /* head of the linked list of contexts */
30 int tm_nb_workers; /* number of workers in the pool */
31 int tm_nb_idle; /* number of workers waiting */
33 int tm_p[2]; /* poll control pipe */
35 smutex_t tm_w_mutex; /* linked list access mutex */
36 scond_t tm_w_cond; /* */
39 static taskmgr_T Tskmgr = {0};
41 #define WRK_CTX_HEAD Tskmgr.tm_ctx_head
43 #define RD_PIPE (Tskmgr.tm_p[0])
44 #define WR_PIPE (Tskmgr.tm_p[1])
46 #define PIPE_SEND_SIGNAL() \
51 if (write(fd, &evt, sizeof(evt)) != sizeof(evt)) \
52 smi_log(SMI_LOG_ERR, \
53 "Error writing to event pipe: %s", \
54 sm_errstring(errno)); \
57 #ifndef USE_PIPE_WAKE_POLL
58 # define USE_PIPE_WAKE_POLL 1
59 #endif /* USE_PIPE_WAKE_POLL */
61 /* poll check periodicity (default 10000 - 10 s) */
62 #define POLL_TIMEOUT 10000
64 /* worker conditional wait timeout (default 10 s) */
65 #define COND_TIMEOUT 10
68 static int mi_close_session __P((SMFICTX_PTR));
70 static void *mi_worker __P((void *));
71 static void *mi_pool_controller __P((void *));
73 static int mi_list_add_ctx __P((SMFICTX_PTR));
74 static int mi_list_del_ctx __P((SMFICTX_PTR));
77 ** periodicity of cleaning up old sessions (timedout)
78 ** sessions list will be checked to find old inactive
79 ** sessions each DT_CHECK_OLD_SESSIONS sec
82 #define DT_CHECK_OLD_SESSIONS 600
84 #ifndef OLD_SESSION_TIMEOUT
85 # define OLD_SESSION_TIMEOUT ctx->ctx_timeout
86 #endif /* OLD_SESSION_TIMEOUT */
88 /* session states - with respect to the pool of workers */
89 #define WKST_INIT 0 /* initial state */
90 #define WKST_READY_TO_RUN 1 /* command ready do be read */
91 #define WKST_RUNNING 2 /* session running on a worker */
92 #define WKST_READY_TO_WAIT 3 /* session just finished by a worker */
93 #define WKST_WAITING 4 /* waiting for new command */
94 #define WKST_CLOSING 5 /* session finished */
97 # define MIN_WORKERS 2 /* minimum number of threads to keep around */
100 #define MIN_IDLE 1 /* minimum number of idle threads */
104 ** Macros for threads and mutex management
107 #define TASKMGR_LOCK() \
110 if (!smutex_lock(&Tskmgr.tm_w_mutex)) \
111 smi_log(SMI_LOG_ERR, "TASKMGR_LOCK error"); \
114 #define TASKMGR_UNLOCK() \
117 if (!smutex_unlock(&Tskmgr.tm_w_mutex)) \
118 smi_log(SMI_LOG_ERR, "TASKMGR_UNLOCK error"); \
121 #define TASKMGR_COND_WAIT() \
122 scond_timedwait(&Tskmgr.tm_w_cond, &Tskmgr.tm_w_mutex, COND_TIMEOUT)
124 #define TASKMGR_COND_SIGNAL() \
127 if (scond_signal(&Tskmgr.tm_w_cond) != 0) \
128 smi_log(SMI_LOG_ERR, "TASKMGR_COND_SIGNAL error"); \
131 #define LAUNCH_WORKER(ctx) \
137 if ((r = thread_create(&tid, mi_worker, ctx)) != 0) \
138 smi_log(SMI_LOG_ERR, "LAUNCH_WORKER error: %s",\
143 # define POOL_LEV_DPRINTF(lev, x) \
145 if ((lev) < ctx->ctx_dbg) \
148 #else /* POOL_DEBUG */
149 # define POOL_LEV_DPRINTF(lev, x)
150 #endif /* POOL_DEBUG */
153 ** MI_START_SESSION -- Start a session in the pool of workers
156 ** ctx -- context structure
159 ** MI_SUCCESS/MI_FAILURE
163 mi_start_session(ctx)
168 /* this can happen if the milter is shutting down */
169 if (Tskmgr.tm_signature != TM_SIGNATURE)
171 SM_ASSERT(ctx != NULL);
172 POOL_LEV_DPRINTF(4, ("PIPE r=[%d] w=[%d]", RD_PIPE, WR_PIPE));
175 if (mi_list_add_ctx(ctx) != MI_SUCCESS)
183 /* if there is an idle worker, signal it, otherwise start new worker */
184 if (Tskmgr.tm_nb_idle > 0)
186 ctx->ctx_wstate = WKST_READY_TO_RUN;
187 TASKMGR_COND_SIGNAL();
191 ctx->ctx_wstate = WKST_RUNNING;
199 ** MI_CLOSE_SESSION -- Close a session and clean up data structures
202 ** ctx -- context structure
205 ** MI_SUCCESS/MI_FAILURE
209 mi_close_session(ctx)
212 SM_ASSERT(ctx != NULL);
214 (void) mi_list_del_ctx(ctx);
221 ** NONBLOCKING -- set nonblocking mode for a file descriptor.
224 ** fd -- file descriptor
225 ** name -- name for (error) logging
228 ** MI_SUCCESS/MI_FAILURE
232 nonblocking(int fd, const char *name)
237 r = fcntl(fd, F_GETFL, 0);
240 smi_log(SMI_LOG_ERR, "fcntl(%s, F_GETFL)=%s",
241 name, sm_errstring(errno));
245 r = fcntl(fd, F_SETFL, r | O_NONBLOCK);
248 smi_log(SMI_LOG_ERR, "fcntl(%s, F_SETFL, O_NONBLOCK)=%s",
249 name, sm_errstring(errno));
256 ** MI_POOL_CONTROLER_INIT -- Launch the worker pool controller
257 ** Must be called before starting sessions.
263 ** MI_SUCCESS/MI_FAILURE
267 mi_pool_controller_init()
272 if (Tskmgr.tm_signature == TM_SIGNATURE)
275 SM_TAILQ_INIT(&WRK_CTX_HEAD);
276 Tskmgr.tm_tid = (sthread_t) -1;
277 Tskmgr.tm_nb_workers = 0;
278 Tskmgr.tm_nb_idle = 0;
280 if (pipe(Tskmgr.tm_p) != 0)
282 smi_log(SMI_LOG_ERR, "can't create event pipe: %s",
283 sm_errstring(errno));
286 r = nonblocking(WR_PIPE, "WR_PIPE");
289 r = nonblocking(RD_PIPE, "RD_PIPE");
293 (void) smutex_init(&Tskmgr.tm_w_mutex);
294 (void) scond_init(&Tskmgr.tm_w_cond);
296 /* Launch the pool controller */
297 if ((r = thread_create(&tid, mi_pool_controller, (void *) NULL)) != 0)
299 smi_log(SMI_LOG_ERR, "can't create controller thread: %s",
304 Tskmgr.tm_signature = TM_SIGNATURE;
306 /* Create the pool of workers */
307 for (i = 0; i < MIN_WORKERS; i++)
309 if ((r = thread_create(&tid, mi_worker, (void *) NULL)) != 0)
311 smi_log(SMI_LOG_ERR, "can't create workers crew: %s",
321 ** MI_POOL_CONTROLLER -- manage the pool of workers
322 ** This thread must be running when listener begins
333 ** Look for timed out sessions
334 ** Select sessions to wait for sendmail command
335 ** Poll set of file descriptors
338 ** For each file descriptor ready
339 ** launch new thread if no worker available
341 ** signal waiting worker
344 /* Poll structure array (pollfd) size step */
347 #define WAIT_FD(i) (pfd[i].fd)
348 #define WAITFN "POLL"
351 mi_pool_controller(arg)
354 struct pollfd *pfd = NULL;
356 bool rebuild_set = true;
357 int pcnt = 0; /* error count for poll() failures */
360 Tskmgr.tm_tid = sthread_get_id();
361 if (pthread_detach(Tskmgr.tm_tid) != 0)
363 smi_log(SMI_LOG_ERR, "Failed to detach pool controller thread");
367 pfd = (struct pollfd *) malloc(PFD_STEP * sizeof(struct pollfd));
370 smi_log(SMI_LOG_ERR, "Failed to malloc pollfd array: %s",
371 sm_errstring(errno));
376 lastcheck = time(NULL);
383 POOL_LEV_DPRINTF(4, ("Let's %s again...", WAITFN));
385 if (mi_stop() != MILTER_CONT)
392 /* check for timed out sessions? */
393 if (lastcheck + DT_CHECK_OLD_SESSIONS < now)
395 ctx = SM_TAILQ_FIRST(&WRK_CTX_HEAD);
396 while (ctx != SM_TAILQ_END(&WRK_CTX_HEAD))
400 ctx_nxt = SM_TAILQ_NEXT(ctx, ctx_link);
401 if (ctx->ctx_wstate == WKST_WAITING)
403 if (ctx->ctx_wait == 0)
405 else if (ctx->ctx_wait + OLD_SESSION_TIMEOUT
408 /* if session timed out, close it */
409 sfsistat (*fi_close) __P((SMFICTX *));
412 ("Closing old connection: sd=%d id=%d",
416 if ((fi_close = ctx->ctx_smfi->xxfi_close) != NULL)
417 (void) (*fi_close)(ctx);
419 mi_close_session(ctx);
430 ** Initialize poll set.
431 ** Insert into the poll set the file descriptors of
432 ** all sessions waiting for a command from sendmail.
437 /* begin with worker pipe */
438 pfd[nfd].fd = RD_PIPE;
439 pfd[nfd].events = MI_POLL_RD_FLAGS;
440 pfd[nfd].revents = 0;
443 SM_TAILQ_FOREACH(ctx, &WRK_CTX_HEAD, ctx_link)
446 ** update ctx_wait - start of wait moment -
450 if (ctx->ctx_wstate == WKST_READY_TO_WAIT)
453 /* add the session to the pollfd array? */
454 if ((ctx->ctx_wstate == WKST_READY_TO_WAIT) ||
455 (ctx->ctx_wstate == WKST_WAITING))
458 ** Resize the pollfd array if it
459 ** isn't large enough.
467 new = (dim_pfd + PFD_STEP) *
469 tpfd = (struct pollfd *)
479 "Failed to realloc pollfd array:%s",
480 sm_errstring(errno));
484 /* add the session to pollfd array */
487 ctx->ctx_wstate = WKST_WAITING;
488 pfd[nfd].fd = ctx->ctx_sd;
489 pfd[nfd].events = MI_POLL_RD_FLAGS;
490 pfd[nfd].revents = 0;
500 /* Everything is ready, let's wait for an event */
501 rfd = poll(pfd, nfd, POLL_TIMEOUT);
503 POOL_LEV_DPRINTF(4, ("%s returned: at epoch %d value %d",
519 "%s() failed (%s), %s",
520 WAITFN, sm_errstring(errno),
521 pcnt >= MAX_FAILS_S ? "abort" : "try again");
523 if (pcnt >= MAX_FAILS_S)
528 /* something happened */
529 for (i = 0; i < nfd; i++)
531 if (pfd[i].revents == 0)
534 POOL_LEV_DPRINTF(4, ("%s event on pfd[%d/%d]=%d ",
538 /* has a worker signaled an end of task ? */
539 if (WAIT_FD(i) == RD_PIPE)
545 ("PIPE WILL READ evt = %08X %08X",
546 pfd[i].events, pfd[i].revents));
549 while ((pfd[i].revents & MI_POLL_RD_FLAGS) != 0
552 r = read(RD_PIPE, evts, sizeof(evts));
556 ("PIPE DONE READ i=[%d] fd=[%d] r=[%d] evt=[%d]",
557 i, RD_PIPE, (int) r, evts[0]));
559 if ((pfd[i].revents & ~MI_POLL_RD_FLAGS) != 0)
561 /* Exception handling */
566 /* no ! sendmail wants to send a command */
567 SM_TAILQ_FOREACH(ctx, &WRK_CTX_HEAD, ctx_link)
569 if (ctx->ctx_wstate != WKST_WAITING)
573 ("Checking context sd=%d - fd=%d ",
574 ctx->ctx_sd , WAIT_FD(i)));
576 if (ctx->ctx_sd == pfd[i].fd)
581 ("TASK: found %d for fd[%d]=%d",
582 ctx->ctx_sid, i, WAIT_FD(i)));
584 if (Tskmgr.tm_nb_idle > 0)
586 ctx->ctx_wstate = WKST_READY_TO_RUN;
587 TASKMGR_COND_SIGNAL();
591 ctx->ctx_wstate = WKST_RUNNING;
600 ("TASK %s FOUND - Checking PIPE for fd[%d]",
601 ctx != NULL ? "" : "NOT", WAIT_FD(i)));
609 Tskmgr.tm_signature = 0;
614 ctx = SM_TAILQ_FIRST(&WRK_CTX_HEAD);
617 mi_close_session(ctx);
620 (void) smutex_destroy(&Tskmgr.tm_w_mutex);
621 (void) scond_destroy(&Tskmgr.tm_w_cond);
627 ** Look for a task ready to run.
628 ** Value of ctx is NULL or a pointer to a task ready to run.
631 #define GET_TASK_READY_TO_RUN() \
632 SM_TAILQ_FOREACH(ctx, &WRK_CTX_HEAD, ctx_link) \
634 if (ctx->ctx_wstate == WKST_READY_TO_RUN) \
636 ctx->ctx_wstate = WKST_RUNNING; \
642 ** MI_WORKER -- worker thread
643 ** executes tasks distributed by the mi_pool_controller
644 ** or by mi_start_session
647 ** arg -- pointer to context structure
662 ctx = (SMFICTX_PTR) arg;
665 ctx->ctx_wstate = WKST_RUNNING;
667 t_id = sthread_get_id();
668 if (pthread_detach(t_id) != 0)
670 smi_log(SMI_LOG_ERR, "Failed to detach worker thread");
672 ctx->ctx_wstate = WKST_READY_TO_RUN;
677 Tskmgr.tm_nb_workers++;
682 if (mi_stop() != MILTER_CONT)
685 /* let's handle next task... */
691 ("worker %d: new task -> let's handle it",
693 res = mi_engine(ctx);
695 ("worker %d: mi_engine returned %d", t_id, res));
698 if (res != MI_CONTINUE)
700 ctx->ctx_wstate = WKST_CLOSING;
703 ** Delete context from linked list of
704 ** sessions and close session.
707 mi_close_session(ctx);
711 ctx->ctx_wstate = WKST_READY_TO_WAIT;
714 ("writing to event pipe..."));
717 ** Signal task controller to add new session
728 /* check if there is any task waiting to be served */
731 GET_TASK_READY_TO_RUN();
741 ** if not, let's check if there is enough idle workers
745 if (Tskmgr.tm_nb_workers > MIN_WORKERS &&
746 Tskmgr.tm_nb_idle > MIN_IDLE)
749 POOL_LEV_DPRINTF(4, ("worker %d: checking ... %d %d", t_id,
750 Tskmgr.tm_nb_workers, Tskmgr.tm_nb_idle + 1));
754 POOL_LEV_DPRINTF(4, ("worker %d: quitting... ", t_id));
755 Tskmgr.tm_nb_workers--;
761 ** if no task ready to run, wait for another one
768 /* look for a task */
769 GET_TASK_READY_TO_RUN();
777 ** MI_LIST_ADD_CTX -- add new session to linked list
780 ** ctx -- context structure
783 ** MI_FAILURE/MI_SUCCESS
790 SM_ASSERT(ctx != NULL);
791 SM_TAILQ_INSERT_TAIL(&WRK_CTX_HEAD, ctx, ctx_link);
796 ** MI_LIST_DEL_CTX -- remove session from linked list when finished
799 ** ctx -- context structure
802 ** MI_FAILURE/MI_SUCCESS
809 SM_ASSERT(ctx != NULL);
810 if (SM_TAILQ_EMPTY(&WRK_CTX_HEAD))
813 SM_TAILQ_REMOVE(&WRK_CTX_HEAD, ctx, ctx_link);
816 #endif /* _FFR_WORKERS_POOL */