worker.c revision 266692 
1/*
2 *  Copyright (c) 2003-2004, 2007, 2009-2012 Proofpoint, Inc. and its suppliers.
3 *	All rights reserved.
4 *
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.
8 *
9 * Contributed by Jose Marcio Martins da Cruz - Ecole des Mines de Paris
10 *   Jose-Marcio.Martins@ensmp.fr
11 */
12
13#include <sm/gen.h>
14SM_RCSID("@(#)$Id: worker.c,v 8.25 2013-11-22 20:51:37 ca Exp $")
15
16#include "libmilter.h"
17
18#if _FFR_WORKERS_POOL
19
20typedef struct taskmgr_S taskmgr_T;
21
22#define TM_SIGNATURE		0x23021957
23
24struct taskmgr_S
25{
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 */
29
30	int		tm_nb_workers;	/* number of workers in the pool */
31	int		tm_nb_idle;	/* number of workers waiting */
32
33	int		tm_p[2];	/* poll control pipe */
34
35	smutex_t	tm_w_mutex;	/* linked list access mutex */
36	scond_t		tm_w_cond;	/* */
37};
38
39static taskmgr_T     Tskmgr = {0};
40
41#define WRK_CTX_HEAD	Tskmgr.tm_ctx_head
42
43#define RD_PIPE	(Tskmgr.tm_p[0])
44#define WR_PIPE	(Tskmgr.tm_p[1])
45
46#define PIPE_SEND_SIGNAL()						\
47	do								\
48	{								\
49		char evt = 0x5a;					\
50		int fd = WR_PIPE;					\
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));			\
55	} while (0)
56
57#ifndef USE_PIPE_WAKE_POLL
58# define USE_PIPE_WAKE_POLL 1
59#endif /* USE_PIPE_WAKE_POLL */
60
61/* poll check periodicity (default 10000 - 10 s) */
62#define POLL_TIMEOUT   10000
63
64/* worker conditional wait timeout (default 10 s) */
65#define COND_TIMEOUT     10
66
67/* functions */
68static int mi_close_session __P((SMFICTX_PTR));
69
70static void *mi_worker __P((void *));
71static void *mi_pool_controller __P((void *));
72
73static int mi_list_add_ctx __P((SMFICTX_PTR));
74static int mi_list_del_ctx __P((SMFICTX_PTR));
75
76/*
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
80*/
81
82#define DT_CHECK_OLD_SESSIONS   600
83
84#ifndef OLD_SESSION_TIMEOUT
85# define OLD_SESSION_TIMEOUT      ctx->ctx_timeout
86#endif /* OLD_SESSION_TIMEOUT */
87
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 */
95
96#ifndef MIN_WORKERS
97# define MIN_WORKERS	2  /* minimum number of threads to keep around */
98#endif
99
100#define MIN_IDLE	1  /* minimum number of idle threads */
101
102
103/*
104**  Macros for threads and mutex management
105*/
106
107#define TASKMGR_LOCK()							\
108	do								\
109	{								\
110		if (!smutex_lock(&Tskmgr.tm_w_mutex))			\
111			smi_log(SMI_LOG_ERR, "TASKMGR_LOCK error");	\
112	} while (0)
113
114#define TASKMGR_UNLOCK()						\
115	do								\
116	{								\
117		if (!smutex_unlock(&Tskmgr.tm_w_mutex))			\
118			smi_log(SMI_LOG_ERR, "TASKMGR_UNLOCK error");	\
119	} while (0)
120
121#define	TASKMGR_COND_WAIT()						\
122	scond_timedwait(&Tskmgr.tm_w_cond, &Tskmgr.tm_w_mutex, COND_TIMEOUT)
123
124#define	TASKMGR_COND_SIGNAL()						\
125	do								\
126	{								\
127		if (scond_signal(&Tskmgr.tm_w_cond) != 0)		\
128			smi_log(SMI_LOG_ERR, "TASKMGR_COND_SIGNAL error"); \
129	} while (0)
130
131#define LAUNCH_WORKER(ctx)						\
132	do								\
133	{								\
134		int r;							\
135		sthread_t tid;						\
136									\
137		if ((r = thread_create(&tid, mi_worker, ctx)) != 0)	\
138			smi_log(SMI_LOG_ERR, "LAUNCH_WORKER error: %s",\
139				sm_errstring(r));			\
140	} while (0)
141
142#if POOL_DEBUG
143# define POOL_LEV_DPRINTF(lev, x)					\
144	do								\
145	{								\
146		if ((lev) < ctx->ctx_dbg)				\
147			sm_dprintf x;					\
148	} while (0)
149#else /* POOL_DEBUG */
150# define POOL_LEV_DPRINTF(lev, x)
151#endif /* POOL_DEBUG */
152
153/*
154**  MI_START_SESSION -- Start a session in the pool of workers
155**
156**	Parameters:
157**		ctx -- context structure
158**
159**	Returns:
160**		MI_SUCCESS/MI_FAILURE
161*/
162
163int
164mi_start_session(ctx)
165	SMFICTX_PTR ctx;
166{
167	static long id = 0;
168
169	/* this can happen if the milter is shutting down */
170	if (Tskmgr.tm_signature != TM_SIGNATURE)
171		return MI_FAILURE;
172	SM_ASSERT(ctx != NULL);
173	POOL_LEV_DPRINTF(4, ("PIPE r=[%d] w=[%d]", RD_PIPE, WR_PIPE));
174	TASKMGR_LOCK();
175
176	if (mi_list_add_ctx(ctx) != MI_SUCCESS)
177	{
178		TASKMGR_UNLOCK();
179		return MI_FAILURE;
180	}
181
182	ctx->ctx_sid = id++;
183
184	/* if there is an idle worker, signal it, otherwise start new worker */
185	if (Tskmgr.tm_nb_idle > 0)
186	{
187		ctx->ctx_wstate = WKST_READY_TO_RUN;
188		TASKMGR_COND_SIGNAL();
189	}
190	else
191	{
192		ctx->ctx_wstate = WKST_RUNNING;
193		LAUNCH_WORKER(ctx);
194	}
195	TASKMGR_UNLOCK();
196	return MI_SUCCESS;
197}
198
199/*
200**  MI_CLOSE_SESSION -- Close a session and clean up data structures
201**
202**	Parameters:
203**		ctx -- context structure
204**
205**	Returns:
206**		MI_SUCCESS/MI_FAILURE
207*/
208
209static int
210mi_close_session(ctx)
211	SMFICTX_PTR ctx;
212{
213	SM_ASSERT(ctx != NULL);
214
215	(void) mi_list_del_ctx(ctx);
216	mi_clr_ctx(ctx);
217
218	return MI_SUCCESS;
219}
220
221/*
222**  NONBLOCKING -- set nonblocking mode for a file descriptor.
223**
224**	Parameters:
225**		fd -- file descriptor
226**		name -- name for (error) logging
227**
228**	Returns:
229**		MI_SUCCESS/MI_FAILURE
230*/
231
232static int
233nonblocking(int fd, const char *name)
234{
235	int r;
236
237	errno = 0;
238	r = fcntl(fd, F_GETFL, 0);
239	if (r == -1)
240	{
241		smi_log(SMI_LOG_ERR, "fcntl(%s, F_GETFL)=%s",
242			name, sm_errstring(errno));
243		return MI_FAILURE;
244	}
245	errno = 0;
246	r = fcntl(fd, F_SETFL, r | O_NONBLOCK);
247	if (r == -1)
248	{
249		smi_log(SMI_LOG_ERR, "fcntl(%s, F_SETFL, O_NONBLOCK)=%s",
250			name, sm_errstring(errno));
251		return MI_FAILURE;
252	}
253	return MI_SUCCESS;
254}
255
256/*
257**  MI_POOL_CONTROLER_INIT -- Launch the worker pool controller
258**		Must be called before starting sessions.
259**
260**	Parameters:
261**		none
262**
263**	Returns:
264**		MI_SUCCESS/MI_FAILURE
265*/
266
267int
268mi_pool_controller_init()
269{
270	sthread_t tid;
271	int r, i;
272
273	if (Tskmgr.tm_signature == TM_SIGNATURE)
274		return MI_SUCCESS;
275
276	SM_TAILQ_INIT(&WRK_CTX_HEAD);
277	Tskmgr.tm_tid = (sthread_t) -1;
278	Tskmgr.tm_nb_workers = 0;
279	Tskmgr.tm_nb_idle = 0;
280
281	if (pipe(Tskmgr.tm_p) != 0)
282	{
283		smi_log(SMI_LOG_ERR, "can't create event pipe: %s",
284			sm_errstring(errno));
285		return MI_FAILURE;
286	}
287	r = nonblocking(WR_PIPE, "WR_PIPE");
288	if (r != MI_SUCCESS)
289		return r;
290	r = nonblocking(RD_PIPE, "RD_PIPE");
291	if (r != MI_SUCCESS)
292		return r;
293
294	(void) smutex_init(&Tskmgr.tm_w_mutex);
295	(void) scond_init(&Tskmgr.tm_w_cond);
296
297	/* Launch the pool controller */
298	if ((r = thread_create(&tid, mi_pool_controller, (void *) NULL)) != 0)
299	{
300		smi_log(SMI_LOG_ERR, "can't create controller thread: %s",
301			sm_errstring(r));
302		return MI_FAILURE;
303	}
304	Tskmgr.tm_tid = tid;
305	Tskmgr.tm_signature = TM_SIGNATURE;
306
307	/* Create the pool of workers */
308	for (i = 0; i < MIN_WORKERS; i++)
309	{
310		if ((r = thread_create(&tid, mi_worker, (void *) NULL)) != 0)
311		{
312			smi_log(SMI_LOG_ERR, "can't create workers crew: %s",
313				sm_errstring(r));
314			return MI_FAILURE;
315		}
316	}
317
318	return MI_SUCCESS;
319}
320
321/*
322**  MI_POOL_CONTROLLER -- manage the pool of workers
323**	This thread must be running when listener begins
324**	starting sessions
325**
326**	Parameters:
327**		arg -- unused
328**
329**	Returns:
330**		NULL
331**
332**	Control flow:
333**		for (;;)
334**			Look for timed out sessions
335**			Select sessions to wait for sendmail command
336**			Poll set of file descriptors
337**			if timeout
338**				continue
339**			For each file descriptor ready
340**				launch new thread if no worker available
341**				else
342**				signal waiting worker
343*/
344
345/* Poll structure array (pollfd) size step */
346#define PFD_STEP	256
347
348#define WAIT_FD(i)	(pfd[i].fd)
349#define WAITFN		"POLL"
350
351static void *
352mi_pool_controller(arg)
353	void *arg;
354{
355	struct pollfd *pfd = NULL;
356	int dim_pfd = 0;
357	bool rebuild_set = true;
358	int pcnt = 0; /* error count for poll() failures */
359	time_t lastcheck;
360
361	Tskmgr.tm_tid = sthread_get_id();
362	if (pthread_detach(Tskmgr.tm_tid) != 0)
363	{
364		smi_log(SMI_LOG_ERR, "Failed to detach pool controller thread");
365		return NULL;
366	}
367
368	pfd = (struct pollfd *) malloc(PFD_STEP * sizeof(struct pollfd));
369	if (pfd == NULL)
370	{
371		smi_log(SMI_LOG_ERR, "Failed to malloc pollfd array: %s",
372			sm_errstring(errno));
373		return NULL;
374	}
375	dim_pfd = PFD_STEP;
376
377	lastcheck = time(NULL);
378	for (;;)
379	{
380		SMFICTX_PTR ctx;
381		int nfd, r, i;
382		time_t now;
383
384		POOL_LEV_DPRINTF(4, ("Let's %s again...", WAITFN));
385
386		if (mi_stop() != MILTER_CONT)
387			break;
388
389		TASKMGR_LOCK();
390
391		now = time(NULL);
392
393		/* check for timed out sessions? */
394		if (lastcheck + DT_CHECK_OLD_SESSIONS < now)
395		{
396			ctx = SM_TAILQ_FIRST(&WRK_CTX_HEAD);
397			while (ctx != SM_TAILQ_END(&WRK_CTX_HEAD))
398			{
399				SMFICTX_PTR ctx_nxt;
400
401				ctx_nxt = SM_TAILQ_NEXT(ctx, ctx_link);
402				if (ctx->ctx_wstate == WKST_WAITING)
403				{
404					if (ctx->ctx_wait == 0)
405						ctx->ctx_wait = now;
406					else if (ctx->ctx_wait + OLD_SESSION_TIMEOUT
407						 < now)
408					{
409						/* if session timed out, close it */
410						sfsistat (*fi_close) __P((SMFICTX *));
411
412						POOL_LEV_DPRINTF(4,
413							("Closing old connection: sd=%d id=%d",
414							ctx->ctx_sd,
415							ctx->ctx_sid));
416
417						if ((fi_close = ctx->ctx_smfi->xxfi_close) != NULL)
418							(void) (*fi_close)(ctx);
419
420						mi_close_session(ctx);
421					}
422				}
423				ctx = ctx_nxt;
424			}
425			lastcheck = now;
426		}
427
428		if (rebuild_set)
429		{
430			/*
431			**  Initialize poll set.
432			**  Insert into the poll set the file descriptors of
433			**  all sessions waiting for a command from sendmail.
434			*/
435
436			nfd = 0;
437
438			/* begin with worker pipe */
439			pfd[nfd].fd = RD_PIPE;
440			pfd[nfd].events = MI_POLL_RD_FLAGS;
441			pfd[nfd].revents = 0;
442			nfd++;
443
444			SM_TAILQ_FOREACH(ctx, &WRK_CTX_HEAD, ctx_link)
445			{
446				/*
447				**  update ctx_wait - start of wait moment -
448				**  for timeout
449				*/
450
451				if (ctx->ctx_wstate == WKST_READY_TO_WAIT)
452					ctx->ctx_wait = now;
453
454				/* add the session to the pollfd array? */
455				if ((ctx->ctx_wstate == WKST_READY_TO_WAIT) ||
456				    (ctx->ctx_wstate == WKST_WAITING))
457				{
458					/*
459					**  Resize the pollfd array if it
460					**  isn't large enough.
461					*/
462
463					if (nfd >= dim_pfd)
464					{
465						struct pollfd *tpfd;
466						size_t new;
467
468						new = (dim_pfd + PFD_STEP) *
469							sizeof(*tpfd);
470						tpfd = (struct pollfd *)
471							realloc(pfd, new);
472						if (tpfd != NULL)
473						{
474							pfd = tpfd;
475							dim_pfd += PFD_STEP;
476						}
477						else
478						{
479							smi_log(SMI_LOG_ERR,
480								"Failed to realloc pollfd array:%s",
481								sm_errstring(errno));
482						}
483					}
484
485					/* add the session to pollfd array */
486					if (nfd < dim_pfd)
487					{
488						ctx->ctx_wstate = WKST_WAITING;
489						pfd[nfd].fd = ctx->ctx_sd;
490						pfd[nfd].events = MI_POLL_RD_FLAGS;
491						pfd[nfd].revents = 0;
492						nfd++;
493					}
494				}
495			}
496			rebuild_set = false;
497		}
498
499		TASKMGR_UNLOCK();
500
501		/* Everything is ready, let's wait for an event */
502		r = poll(pfd, nfd, POLL_TIMEOUT);
503
504		POOL_LEV_DPRINTF(4, ("%s returned: at epoch %d value %d",
505			WAITFN, now, nfd));
506
507		/* timeout */
508		if (r == 0)
509			continue;
510
511		rebuild_set = true;
512
513		/* error */
514		if (r < 0)
515		{
516			if (errno == EINTR)
517				continue;
518			pcnt++;
519			smi_log(SMI_LOG_ERR,
520				"%s() failed (%s), %s",
521				WAITFN, sm_errstring(errno),
522				pcnt >= MAX_FAILS_S ? "abort" : "try again");
523
524			if (pcnt >= MAX_FAILS_S)
525				goto err;
526			continue;
527		}
528		pcnt = 0;
529
530		/* something happened */
531		for (i = 0; i < nfd; i++)
532		{
533			if (pfd[i].revents == 0)
534				continue;
535
536			POOL_LEV_DPRINTF(4, ("%s event on pfd[%d/%d]=%d ",
537				WAITFN, i, nfd,
538			WAIT_FD(i)));
539
540			/* has a worker signaled an end of task? */
541			if (WAIT_FD(i) == RD_PIPE)
542			{
543				char evts[256];
544				ssize_t r;
545
546				POOL_LEV_DPRINTF(4,
547					("PIPE WILL READ evt = %08X %08X",
548					pfd[i].events, pfd[i].revents));
549
550				r = 1;
551				while ((pfd[i].revents & MI_POLL_RD_FLAGS) != 0
552					&& r != -1)
553				{
554					r = read(RD_PIPE, evts, sizeof(evts));
555				}
556
557				POOL_LEV_DPRINTF(4,
558					("PIPE DONE READ i=[%d] fd=[%d] r=[%d] evt=[%d]",
559					i, RD_PIPE, (int) r, evts[0]));
560
561				if ((pfd[i].revents & ~MI_POLL_RD_FLAGS) != 0)
562				{
563					/* Exception handling */
564				}
565				continue;
566			}
567
568			/*
569			**  Not the pipe for workers waking us,
570			**  so must be something on an MTA connection.
571			*/
572
573			TASKMGR_LOCK();
574			SM_TAILQ_FOREACH(ctx, &WRK_CTX_HEAD, ctx_link)
575			{
576				if (ctx->ctx_wstate != WKST_WAITING)
577					continue;
578
579				POOL_LEV_DPRINTF(4,
580					("Checking context sd=%d - fd=%d ",
581					ctx->ctx_sd , WAIT_FD(i)));
582
583				if (ctx->ctx_sd == pfd[i].fd)
584				{
585
586					POOL_LEV_DPRINTF(4,
587						("TASK: found %d for fd[%d]=%d",
588						ctx->ctx_sid, i, WAIT_FD(i)));
589
590					if (Tskmgr.tm_nb_idle > 0)
591					{
592						ctx->ctx_wstate = WKST_READY_TO_RUN;
593						TASKMGR_COND_SIGNAL();
594					}
595					else
596					{
597						ctx->ctx_wstate = WKST_RUNNING;
598						LAUNCH_WORKER(ctx);
599					}
600					break;
601				}
602			}
603			TASKMGR_UNLOCK();
604
605			POOL_LEV_DPRINTF(4,
606				("TASK %s FOUND - Checking PIPE for fd[%d]",
607				ctx != NULL ? "" : "NOT", WAIT_FD(i)));
608		}
609	}
610
611  err:
612	if (pfd != NULL)
613		free(pfd);
614
615	Tskmgr.tm_signature = 0;
616#if 0
617	/*
618	**  Do not clean up ctx -- it can cause double-free()s.
619	**  The program is shutting down anyway, so it's not worth the trouble.
620	**  There is a more complex solution that prevents race conditions
621	**  while accessing ctx, but that's maybe for a later version.
622	*/
623
624	for (;;)
625	{
626		SMFICTX_PTR ctx;
627
628		ctx = SM_TAILQ_FIRST(&WRK_CTX_HEAD);
629		if (ctx == NULL)
630			break;
631		mi_close_session(ctx);
632	}
633#endif
634
635	(void) smutex_destroy(&Tskmgr.tm_w_mutex);
636	(void) scond_destroy(&Tskmgr.tm_w_cond);
637
638	return NULL;
639}
640
641/*
642**  Look for a task ready to run.
643**  Value of ctx is NULL or a pointer to a task ready to run.
644*/
645
646#define GET_TASK_READY_TO_RUN()					\
647	SM_TAILQ_FOREACH(ctx, &WRK_CTX_HEAD, ctx_link)		\
648	{							\
649		if (ctx->ctx_wstate == WKST_READY_TO_RUN)	\
650		{						\
651			ctx->ctx_wstate = WKST_RUNNING;		\
652			break;					\
653		}						\
654	}
655
656/*
657**  MI_WORKER -- worker thread
658**	executes tasks distributed by the mi_pool_controller
659**	or by mi_start_session
660**
661**	Parameters:
662**		arg -- pointer to context structure
663**
664**	Returns:
665**		NULL pointer
666*/
667
668static void *
669mi_worker(arg)
670	void *arg;
671{
672	SMFICTX_PTR ctx;
673	bool done;
674	sthread_t t_id;
675	int r;
676
677	ctx = (SMFICTX_PTR) arg;
678	done = false;
679	if (ctx != NULL)
680		ctx->ctx_wstate = WKST_RUNNING;
681
682	t_id = sthread_get_id();
683	if (pthread_detach(t_id) != 0)
684	{
685		smi_log(SMI_LOG_ERR, "Failed to detach worker thread");
686		if (ctx != NULL)
687			ctx->ctx_wstate = WKST_READY_TO_RUN;
688		return NULL;
689	}
690
691	TASKMGR_LOCK();
692	Tskmgr.tm_nb_workers++;
693	TASKMGR_UNLOCK();
694
695	while (!done)
696	{
697		if (mi_stop() != MILTER_CONT)
698			break;
699
700		/* let's handle next task... */
701		if (ctx != NULL)
702		{
703			int res;
704
705			POOL_LEV_DPRINTF(4,
706				("worker %d: new task -> let's handle it",
707				t_id));
708			res = mi_engine(ctx);
709			POOL_LEV_DPRINTF(4,
710				("worker %d: mi_engine returned %d", t_id, res));
711
712			TASKMGR_LOCK();
713			if (res != MI_CONTINUE)
714			{
715				ctx->ctx_wstate = WKST_CLOSING;
716
717				/*
718				**  Delete context from linked list of
719				**  sessions and close session.
720				*/
721
722				mi_close_session(ctx);
723			}
724			else
725			{
726				ctx->ctx_wstate = WKST_READY_TO_WAIT;
727
728				POOL_LEV_DPRINTF(4,
729					("writing to event pipe..."));
730
731				/*
732				**  Signal task controller to add new session
733				**  to poll set.
734				*/
735
736				PIPE_SEND_SIGNAL();
737			}
738			TASKMGR_UNLOCK();
739			ctx = NULL;
740
741		}
742
743		/* check if there is any task waiting to be served */
744		TASKMGR_LOCK();
745
746		GET_TASK_READY_TO_RUN();
747
748		/* Got a task? */
749		if (ctx != NULL)
750		{
751			TASKMGR_UNLOCK();
752			continue;
753		}
754
755		/*
756		**  if not, let's check if there is enough idle workers
757		**	if yes: quit
758		*/
759
760		if (Tskmgr.tm_nb_workers > MIN_WORKERS &&
761		    Tskmgr.tm_nb_idle > MIN_IDLE)
762			done = true;
763
764		POOL_LEV_DPRINTF(4, ("worker %d: checking ... %d %d", t_id,
765			Tskmgr.tm_nb_workers, Tskmgr.tm_nb_idle + 1));
766
767		if (done)
768		{
769			POOL_LEV_DPRINTF(4, ("worker %d: quitting... ", t_id));
770			Tskmgr.tm_nb_workers--;
771			TASKMGR_UNLOCK();
772			continue;
773		}
774
775		/*
776		**  if no task ready to run, wait for another one
777		*/
778
779		Tskmgr.tm_nb_idle++;
780		TASKMGR_COND_WAIT();
781		Tskmgr.tm_nb_idle--;
782
783		/* look for a task */
784		GET_TASK_READY_TO_RUN();
785
786		TASKMGR_UNLOCK();
787	}
788	return NULL;
789}
790
791/*
792**  MI_LIST_ADD_CTX -- add new session to linked list
793**
794**	Parameters:
795**		ctx -- context structure
796**
797**	Returns:
798**		MI_FAILURE/MI_SUCCESS
799*/
800
801static int
802mi_list_add_ctx(ctx)
803	SMFICTX_PTR ctx;
804{
805	SM_ASSERT(ctx != NULL);
806	SM_TAILQ_INSERT_TAIL(&WRK_CTX_HEAD, ctx, ctx_link);
807	return MI_SUCCESS;
808}
809
810/*
811**  MI_LIST_DEL_CTX -- remove session from linked list when finished
812**
813**	Parameters:
814**		ctx -- context structure
815**
816**	Returns:
817**		MI_FAILURE/MI_SUCCESS
818*/
819
820static int
821mi_list_del_ctx(ctx)
822	SMFICTX_PTR ctx;
823{
824	SM_ASSERT(ctx != NULL);
825	if (SM_TAILQ_EMPTY(&WRK_CTX_HEAD))
826		return MI_FAILURE;
827
828	SM_TAILQ_REMOVE(&WRK_CTX_HEAD, ctx, ctx_link);
829	return MI_SUCCESS;
830}
831#endif /* _FFR_WORKERS_POOL */
832