sendmail/libmilter/worker.c

/*
 *  Copyright (c) 2003-2004, 2007, 2009-2012 Proofpoint, Inc. and its suppliers.
 *  All rights reserved.
 *
 * By using this file, you agree to the terms and conditions set
 * forth in the LICENSE file which can be found at the top level of
 * the sendmail distribution.
 *
 * Contributed by Jose Marcio Martins da Cruz - Ecole des Mines de Paris
 *   Jose-Marcio.Martins@ensmp.fr
 */

#include <sm/gen.h>
SM_RCSID("@(#)$Id: worker.c,v 8.25 2013-11-22 20:51:37 ca Exp $")

#include "libmilter.h"

#if _FFR_WORKERS_POOL

typedef struct taskmgr_S taskmgr_T;

#define TM_SIGNATURE        0x23021957

struct taskmgr_S
{
    long        tm_signature; /* has the controller been initialized */
    sthread_t   tm_tid; /* thread id of controller */
    smfi_hd_T   tm_ctx_head; /* head of the linked list of contexts */

    int     tm_nb_workers;  /* number of workers in the pool */
    int     tm_nb_idle; /* number of workers waiting */

    int     tm_p[2];    /* poll control pipe */

    smutex_t    tm_w_mutex; /* linked list access mutex */
    scond_t     tm_w_cond;  /* */
};

static taskmgr_T     Tskmgr = {0};

#define WRK_CTX_HEAD    Tskmgr.tm_ctx_head

#define RD_PIPE (Tskmgr.tm_p[0])
#define WR_PIPE (Tskmgr.tm_p[1])

#define PIPE_SEND_SIGNAL()                      \
    do                              \
    {                               \
        char evt = 0x5a;                    \
        int fd = WR_PIPE;                   \
        if (write(fd, &evt, sizeof(evt)) != sizeof(evt))    \
            smi_log(SMI_LOG_ERR,                \
                "Error writing to event pipe: %s",  \
                sm_errstring(errno));           \
    } while (0)

#ifndef USE_PIPE_WAKE_POLL
# define USE_PIPE_WAKE_POLL 1
#endif /* USE_PIPE_WAKE_POLL */

/* poll check periodicity (default 10000 - 10 s) */
#define POLL_TIMEOUT   10000

/* worker conditional wait timeout (default 10 s) */
#define COND_TIMEOUT     10

/* functions */
static int mi_close_session __P((SMFICTX_PTR));

static void *mi_worker __P((void *));
static void *mi_pool_controller __P((void *));

static int mi_list_add_ctx __P((SMFICTX_PTR));
static int mi_list_del_ctx __P((SMFICTX_PTR));

/*
**  periodicity of cleaning up old sessions (timedout)
**  sessions list will be checked to find old inactive
**  sessions each DT_CHECK_OLD_SESSIONS sec
*/

#define DT_CHECK_OLD_SESSIONS   600

#ifndef OLD_SESSION_TIMEOUT
# define OLD_SESSION_TIMEOUT      ctx->ctx_timeout
#endif /* OLD_SESSION_TIMEOUT */

/* session states - with respect to the pool of workers */
#define WKST_INIT       0   /* initial state */
#define WKST_READY_TO_RUN   1   /* command ready do be read */
#define WKST_RUNNING        2   /* session running on a worker */
#define WKST_READY_TO_WAIT  3   /* session just finished by a worker */
#define WKST_WAITING        4   /* waiting for new command */
#define WKST_CLOSING        5   /* session finished */

#ifndef MIN_WORKERS
# define MIN_WORKERS    2  /* minimum number of threads to keep around */
#endif

#define MIN_IDLE    1  /* minimum number of idle threads */


/*
**  Macros for threads and mutex management
*/

#define TASKMGR_LOCK()                          \
    do                              \
    {                               \
        if (!smutex_lock(&Tskmgr.tm_w_mutex))           \
            smi_log(SMI_LOG_ERR, "TASKMGR_LOCK error"); \
    } while (0)

#define TASKMGR_UNLOCK()                        \
    do                              \
    {                               \
        if (!smutex_unlock(&Tskmgr.tm_w_mutex))         \
            smi_log(SMI_LOG_ERR, "TASKMGR_UNLOCK error");   \
    } while (0)

#define TASKMGR_COND_WAIT()                     \
    scond_timedwait(&Tskmgr.tm_w_cond, &Tskmgr.tm_w_mutex, COND_TIMEOUT)

#define TASKMGR_COND_SIGNAL()                       \
    do                              \
    {                               \
        if (scond_signal(&Tskmgr.tm_w_cond) != 0)       \
            smi_log(SMI_LOG_ERR, "TASKMGR_COND_SIGNAL error"); \
    } while (0)

#define LAUNCH_WORKER(ctx)                      \
    do                              \
    {                               \
        int r;                          \
        sthread_t tid;                      \
                                    \
        if ((r = thread_create(&tid, mi_worker, ctx)) != 0) \
            smi_log(SMI_LOG_ERR, "LAUNCH_WORKER error: %s",\
                sm_errstring(r));           \
    } while (0)

#if POOL_DEBUG
# define POOL_LEV_DPRINTF(lev, x)                   \
    do                              \
    {                               \
        if ((lev) < ctx->ctx_dbg)               \
            sm_dprintf x;                   \
    } while (0)
#else /* POOL_DEBUG */
# define POOL_LEV_DPRINTF(lev, x)
#endif /* POOL_DEBUG */

/*
**  MI_START_SESSION -- Start a session in the pool of workers
**
**  Parameters:
**      ctx -- context structure
**
**  Returns:
**      MI_SUCCESS/MI_FAILURE
*/

int
mi_start_session(ctx)
    SMFICTX_PTR ctx;
{
    static long id = 0;

    /* this can happen if the milter is shutting down */
    if (Tskmgr.tm_signature != TM_SIGNATURE)
        return MI_FAILURE;
    SM_ASSERT(ctx != NULL);
    POOL_LEV_DPRINTF(4, ("PIPE r=[%d] w=[%d]", RD_PIPE, WR_PIPE));
    TASKMGR_LOCK();

    if (mi_list_add_ctx(ctx) != MI_SUCCESS)
    {
        TASKMGR_UNLOCK();
        return MI_FAILURE;
    }

    ctx->ctx_sid = id++;

    /* if there is an idle worker, signal it, otherwise start new worker */
    if (Tskmgr.tm_nb_idle > 0)
    {
        ctx->ctx_wstate = WKST_READY_TO_RUN;
        TASKMGR_COND_SIGNAL();
    }
    else
    {
        ctx->ctx_wstate = WKST_RUNNING;
        LAUNCH_WORKER(ctx);
    }
    TASKMGR_UNLOCK();
    return MI_SUCCESS;
}

/*
**  MI_CLOSE_SESSION -- Close a session and clean up data structures
**
**  Parameters:
**      ctx -- context structure
**
**  Returns:
**      MI_SUCCESS/MI_FAILURE
*/

static int
mi_close_session(ctx)
    SMFICTX_PTR ctx;
{
    SM_ASSERT(ctx != NULL);

    (void) mi_list_del_ctx(ctx);
    mi_clr_ctx(ctx);

    return MI_SUCCESS;
}

/*
**  NONBLOCKING -- set nonblocking mode for a file descriptor.
**
**  Parameters:
**      fd -- file descriptor
**      name -- name for (error) logging
**
**  Returns:
**      MI_SUCCESS/MI_FAILURE
*/

static int
nonblocking(int fd, const char *name)
{
    int r;

    errno = 0;
    r = fcntl(fd, F_GETFL, 0);
    if (r == -1)
    {
        smi_log(SMI_LOG_ERR, "fcntl(%s, F_GETFL)=%s",
            name, sm_errstring(errno));
        return MI_FAILURE;
    }
    errno = 0;
    r = fcntl(fd, F_SETFL, r | O_NONBLOCK);
    if (r == -1)
    {
        smi_log(SMI_LOG_ERR, "fcntl(%s, F_SETFL, O_NONBLOCK)=%s",
            name, sm_errstring(errno));
        return MI_FAILURE;
    }
    return MI_SUCCESS;
}

/*
**  MI_POOL_CONTROLER_INIT -- Launch the worker pool controller
**      Must be called before starting sessions.
**
**  Parameters:
**      none
**
**  Returns:
**      MI_SUCCESS/MI_FAILURE
*/

int
mi_pool_controller_init()
{
    sthread_t tid;
    int r, i;

    if (Tskmgr.tm_signature == TM_SIGNATURE)
        return MI_SUCCESS;

    SM_TAILQ_INIT(&WRK_CTX_HEAD);
    Tskmgr.tm_tid = (sthread_t) -1;
    Tskmgr.tm_nb_workers = 0;
    Tskmgr.tm_nb_idle = 0;

    if (pipe(Tskmgr.tm_p) != 0)
    {
        smi_log(SMI_LOG_ERR, "can't create event pipe: %s",
            sm_errstring(errno));
        return MI_FAILURE;
    }
    r = nonblocking(WR_PIPE, "WR_PIPE");
    if (r != MI_SUCCESS)
        return r;
    r = nonblocking(RD_PIPE, "RD_PIPE");
    if (r != MI_SUCCESS)
        return r;

    (void) smutex_init(&Tskmgr.tm_w_mutex);
    (void) scond_init(&Tskmgr.tm_w_cond);

    /* Launch the pool controller */
    if ((r = thread_create(&tid, mi_pool_controller, (void *) NULL)) != 0)
    {
        smi_log(SMI_LOG_ERR, "can't create controller thread: %s",
            sm_errstring(r));
        return MI_FAILURE;
    }
    Tskmgr.tm_tid = tid;
    Tskmgr.tm_signature = TM_SIGNATURE;

    /* Create the pool of workers */
    for (i = 0; i < MIN_WORKERS; i++)
    {
        if ((r = thread_create(&tid, mi_worker, (void *) NULL)) != 0)
        {
            smi_log(SMI_LOG_ERR, "can't create workers crew: %s",
                sm_errstring(r));
            return MI_FAILURE;
        }
    }

    return MI_SUCCESS;
}

/*
**  MI_POOL_CONTROLLER -- manage the pool of workers
**  This thread must be running when listener begins
**  starting sessions
**
**  Parameters:
**      arg -- unused
**
**  Returns:
**      NULL
**
**  Control flow:
**      for (;;)
**          Look for timed out sessions
**          Select sessions to wait for sendmail command
**          Poll set of file descriptors
**          if timeout
**              continue
**          For each file descriptor ready
**              launch new thread if no worker available
**              else
**              signal waiting worker
*/

/* Poll structure array (pollfd) size step */
#define PFD_STEP    256

#define WAIT_FD(i)  (pfd[i].fd)
#define WAITFN      "POLL"

static void *
mi_pool_controller(arg)
    void *arg;
{
    struct pollfd *pfd = NULL;
    int dim_pfd = 0;
    bool rebuild_set = true;
    int pcnt = 0; /* error count for poll() failures */
    time_t lastcheck;

    Tskmgr.tm_tid = sthread_get_id();
    if (pthread_detach(Tskmgr.tm_tid) != 0)
    {
        smi_log(SMI_LOG_ERR, "Failed to detach pool controller thread");
        return NULL;
    }

    pfd = (struct pollfd *) malloc(PFD_STEP * sizeof(struct pollfd));
    if (pfd == NULL)
    {
        smi_log(SMI_LOG_ERR, "Failed to malloc pollfd array: %s",
            sm_errstring(errno));
        return NULL;
    }
    dim_pfd = PFD_STEP;

    lastcheck = time(NULL);
    for (;;)
    {
        SMFICTX_PTR ctx;
        int nfd, r, i;
        time_t now;

        POOL_LEV_DPRINTF(4, ("Let's %s again...", WAITFN));

        if (mi_stop() != MILTER_CONT)
            break;

        TASKMGR_LOCK();

        now = time(NULL);

        /* check for timed out sessions? */
        if (lastcheck + DT_CHECK_OLD_SESSIONS < now)
        {
            ctx = SM_TAILQ_FIRST(&WRK_CTX_HEAD);
            while (ctx != SM_TAILQ_END(&WRK_CTX_HEAD))
            {
                SMFICTX_PTR ctx_nxt;

                ctx_nxt = SM_TAILQ_NEXT(ctx, ctx_link);
                if (ctx->ctx_wstate == WKST_WAITING)
                {
                    if (ctx->ctx_wait == 0)
                        ctx->ctx_wait = now;
                    else if (ctx->ctx_wait + OLD_SESSION_TIMEOUT
                         < now)
                    {
                        /* if session timed out, close it */
                        sfsistat (*fi_close) __P((SMFICTX *));

                        POOL_LEV_DPRINTF(4,
                            ("Closing old connection: sd=%d id=%d",
                            ctx->ctx_sd,
                            ctx->ctx_sid));

                        if ((fi_close = ctx->ctx_smfi->xxfi_close) != NULL)
                            (void) (*fi_close)(ctx);

                        mi_close_session(ctx);
                    }
                }
                ctx = ctx_nxt;
            }
            lastcheck = now;
        }

        if (rebuild_set)
        {
            /*
            **  Initialize poll set.
            **  Insert into the poll set the file descriptors of
            **  all sessions waiting for a command from sendmail.
            */

            nfd = 0;

            /* begin with worker pipe */
            pfd[nfd].fd = RD_PIPE;
            pfd[nfd].events = MI_POLL_RD_FLAGS;
            pfd[nfd].revents = 0;
            nfd++;

            SM_TAILQ_FOREACH(ctx, &WRK_CTX_HEAD, ctx_link)
            {
                /*
                **  update ctx_wait - start of wait moment -
                **  for timeout
                */

                if (ctx->ctx_wstate == WKST_READY_TO_WAIT)
                    ctx->ctx_wait = now;

                /* add the session to the pollfd array? */
                if ((ctx->ctx_wstate == WKST_READY_TO_WAIT) ||
                    (ctx->ctx_wstate == WKST_WAITING))
                {
                    /*
                    **  Resize the pollfd array if it
                    **  isn't large enough.
                    */

                    if (nfd >= dim_pfd)
                    {
                        struct pollfd *tpfd;
                        size_t new;

                        new = (dim_pfd + PFD_STEP) *
                            sizeof(*tpfd);
                        tpfd = (struct pollfd *)
                            realloc(pfd, new);
                        if (tpfd != NULL)
                        {
                            pfd = tpfd;
                            dim_pfd += PFD_STEP;
                        }
                        else
                        {
                            smi_log(SMI_LOG_ERR,
                                "Failed to realloc pollfd array:%s",
                                sm_errstring(errno));
                        }
                    }

                    /* add the session to pollfd array */
                    if (nfd < dim_pfd)
                    {
                        ctx->ctx_wstate = WKST_WAITING;
                        pfd[nfd].fd = ctx->ctx_sd;
                        pfd[nfd].events = MI_POLL_RD_FLAGS;
                        pfd[nfd].revents = 0;
                        nfd++;
                    }
                }
            }
            rebuild_set = false;
        }

        TASKMGR_UNLOCK();

        /* Everything is ready, let's wait for an event */
        r = poll(pfd, nfd, POLL_TIMEOUT);

        POOL_LEV_DPRINTF(4, ("%s returned: at epoch %d value %d",
            WAITFN, now, nfd));

        /* timeout */
        if (r == 0)
            continue;

        rebuild_set = true;

        /* error */
        if (r < 0)
        {
            if (errno == EINTR)
                continue;
            pcnt++;
            smi_log(SMI_LOG_ERR,
                "%s() failed (%s), %s",
                WAITFN, sm_errstring(errno),
                pcnt >= MAX_FAILS_S ? "abort" : "try again");

            if (pcnt >= MAX_FAILS_S)
                goto err;
            continue;
        }
        pcnt = 0;

        /* something happened */
        for (i = 0; i < nfd; i++)
        {
            if (pfd[i].revents == 0)
                continue;

            POOL_LEV_DPRINTF(4, ("%s event on pfd[%d/%d]=%d ",
                WAITFN, i, nfd,
            WAIT_FD(i)));

            /* has a worker signaled an end of task? */
            if (WAIT_FD(i) == RD_PIPE)
            {
                char evts[256];
                ssize_t r;

                POOL_LEV_DPRINTF(4,
                    ("PIPE WILL READ evt = %08X %08X",
                    pfd[i].events, pfd[i].revents));

                r = 1;
                while ((pfd[i].revents & MI_POLL_RD_FLAGS) != 0
                    && r != -1)
                {
                    r = read(RD_PIPE, evts, sizeof(evts));
                }

                POOL_LEV_DPRINTF(4,
                    ("PIPE DONE READ i=[%d] fd=[%d] r=[%d] evt=[%d]",
                    i, RD_PIPE, (int) r, evts[0]));

                if ((pfd[i].revents & ~MI_POLL_RD_FLAGS) != 0)
                {
                    /* Exception handling */
                }
                continue;
            }

            /*
            **  Not the pipe for workers waking us,
            **  so must be something on an MTA connection.
            */

            TASKMGR_LOCK();
            SM_TAILQ_FOREACH(ctx, &WRK_CTX_HEAD, ctx_link)
            {
                if (ctx->ctx_wstate != WKST_WAITING)
                    continue;

                POOL_LEV_DPRINTF(4,
                    ("Checking context sd=%d - fd=%d ",
                    ctx->ctx_sd , WAIT_FD(i)));

                if (ctx->ctx_sd == pfd[i].fd)
                {

                    POOL_LEV_DPRINTF(4,
                        ("TASK: found %d for fd[%d]=%d",
                        ctx->ctx_sid, i, WAIT_FD(i)));

                    if (Tskmgr.tm_nb_idle > 0)
                    {
                        ctx->ctx_wstate = WKST_READY_TO_RUN;
                        TASKMGR_COND_SIGNAL();
                    }
                    else
                    {
                        ctx->ctx_wstate = WKST_RUNNING;
                        LAUNCH_WORKER(ctx);
                    }
                    break;
                }
            }
            TASKMGR_UNLOCK();

            POOL_LEV_DPRINTF(4,
                ("TASK %s FOUND - Checking PIPE for fd[%d]",
                ctx != NULL ? "" : "NOT", WAIT_FD(i)));
        }
    }

  err:
    if (pfd != NULL)
        free(pfd);

    Tskmgr.tm_signature = 0;
#if 0
    /*
    **  Do not clean up ctx -- it can cause double-free()s.
    **  The program is shutting down anyway, so it's not worth the trouble.
    **  There is a more complex solution that prevents race conditions
    **  while accessing ctx, but that's maybe for a later version.
    */

    for (;;)
    {
        SMFICTX_PTR ctx;

        ctx = SM_TAILQ_FIRST(&WRK_CTX_HEAD);
        if (ctx == NULL)
            break;
        mi_close_session(ctx);
    }
#endif

    (void) smutex_destroy(&Tskmgr.tm_w_mutex);
    (void) scond_destroy(&Tskmgr.tm_w_cond);

    return NULL;
}

/*
**  Look for a task ready to run.
**  Value of ctx is NULL or a pointer to a task ready to run.
*/

#define GET_TASK_READY_TO_RUN()                 \
    SM_TAILQ_FOREACH(ctx, &WRK_CTX_HEAD, ctx_link)      \
    {                           \
        if (ctx->ctx_wstate == WKST_READY_TO_RUN)   \
        {                       \
            ctx->ctx_wstate = WKST_RUNNING;     \
            break;                  \
        }                       \
    }

/*
**  MI_WORKER -- worker thread
**  executes tasks distributed by the mi_pool_controller
**  or by mi_start_session
**
**  Parameters:
**      arg -- pointer to context structure
**
**  Returns:
**      NULL pointer
*/

static void *
mi_worker(arg)
    void *arg;
{
    SMFICTX_PTR ctx;
    bool done;
    sthread_t t_id;
    int r;

    ctx = (SMFICTX_PTR) arg;
    done = false;
    if (ctx != NULL)
        ctx->ctx_wstate = WKST_RUNNING;

    t_id = sthread_get_id();
    if (pthread_detach(t_id) != 0)
    {
        smi_log(SMI_LOG_ERR, "Failed to detach worker thread");
        if (ctx != NULL)
            ctx->ctx_wstate = WKST_READY_TO_RUN;
        return NULL;
    }

    TASKMGR_LOCK();
    Tskmgr.tm_nb_workers++;
    TASKMGR_UNLOCK();

    while (!done)
    {
        if (mi_stop() != MILTER_CONT)
            break;

        /* let's handle next task... */
        if (ctx != NULL)
        {
            int res;

            POOL_LEV_DPRINTF(4,
                ("worker %d: new task -> let's handle it",
                t_id));
            res = mi_engine(ctx);
            POOL_LEV_DPRINTF(4,
                ("worker %d: mi_engine returned %d", t_id, res));

            TASKMGR_LOCK();
            if (res != MI_CONTINUE)
            {
                ctx->ctx_wstate = WKST_CLOSING;

                /*
                **  Delete context from linked list of
                **  sessions and close session.
                */

                mi_close_session(ctx);
            }
            else
            {
                ctx->ctx_wstate = WKST_READY_TO_WAIT;

                POOL_LEV_DPRINTF(4,
                    ("writing to event pipe..."));

                /*
                **  Signal task controller to add new session
                **  to poll set.
                */

                PIPE_SEND_SIGNAL();
            }
            TASKMGR_UNLOCK();
            ctx = NULL;

        }

        /* check if there is any task waiting to be served */
        TASKMGR_LOCK();

        GET_TASK_READY_TO_RUN();

        /* Got a task? */
        if (ctx != NULL)
        {
            TASKMGR_UNLOCK();
            continue;
        }

        /*
        **  if not, let's check if there is enough idle workers
        **  if yes: quit
        */

        if (Tskmgr.tm_nb_workers > MIN_WORKERS &&
            Tskmgr.tm_nb_idle > MIN_IDLE)
            done = true;

        POOL_LEV_DPRINTF(4, ("worker %d: checking ... %d %d", t_id,
            Tskmgr.tm_nb_workers, Tskmgr.tm_nb_idle + 1));

        if (done)
        {
            POOL_LEV_DPRINTF(4, ("worker %d: quitting... ", t_id));
            Tskmgr.tm_nb_workers--;
            TASKMGR_UNLOCK();
            continue;
        }

        /*
        **  if no task ready to run, wait for another one
        */

        Tskmgr.tm_nb_idle++;
        TASKMGR_COND_WAIT();
        Tskmgr.tm_nb_idle--;

        /* look for a task */
        GET_TASK_READY_TO_RUN();

        TASKMGR_UNLOCK();
    }
    return NULL;
}

/*
**  MI_LIST_ADD_CTX -- add new session to linked list
**
**  Parameters:
**      ctx -- context structure
**
**  Returns:
**      MI_FAILURE/MI_SUCCESS
*/

static int
mi_list_add_ctx(ctx)
    SMFICTX_PTR ctx;
{
    SM_ASSERT(ctx != NULL);
    SM_TAILQ_INSERT_TAIL(&WRK_CTX_HEAD, ctx, ctx_link);
    return MI_SUCCESS;
}

/*
**  MI_LIST_DEL_CTX -- remove session from linked list when finished
**
**  Parameters:
**      ctx -- context structure
**
**  Returns:
**      MI_FAILURE/MI_SUCCESS
*/

static int
mi_list_del_ctx(ctx)
    SMFICTX_PTR ctx;
{
    SM_ASSERT(ctx != NULL);
    if (SM_TAILQ_EMPTY(&WRK_CTX_HEAD))
        return MI_FAILURE;

    SM_TAILQ_REMOVE(&WRK_CTX_HEAD, ctx, ctx_link);
    return MI_SUCCESS;
}
#endif /* _FFR_WORKERS_POOL */