/*

 * libetp implementation

 *

 * Copyright (c) 2007,2008,2009,2010,2011,2012,2013,2015 Marc Alexander Lehmann <libetp@schmorp.de>

 * All rights reserved.

 *

 * Redistribution and use in source and binary forms, with or without modifica-

 * tion, are permitted provided that the following conditions are met:

 *

 *   1.  Redistributions of source code must retain the above copyright notice,

 *       this list of conditions and the following disclaimer.

 *

 *   2.  Redistributions in binary form must reproduce the above copyright

 *       notice, this list of conditions and the following disclaimer in the

 *       documentation and/or other materials provided with the distribution.

 *

 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED

 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MER-

 * CHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO

 * EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPE-

 * CIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,

 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;

 * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,

 * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTH-

 * ERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED

 * OF THE POSSIBILITY OF SUCH DAMAGE.

 *

 * Alternatively, the contents of this file may be used under the terms of

 * the GNU General Public License ("GPL") version 2 or any later version,

 * in which case the provisions of the GPL are applicable instead of

 * the above. If you wish to allow the use of your version of this file

 * only under the terms of the GPL and not to allow others to use your

 * version of this file under the BSD license, indicate your decision

 * by deleting the provisions above and replace them with the notice

 * and other provisions required by the GPL. If you do not delete the

 * provisions above, a recipient may use your version of this file under

 * either the BSD or the GPL.

 */

#ifndef ETP_API_DECL

# define ETP_API_DECL static

#endif

#ifndef ETP_PRI_MIN

# define ETP_PRI_MIN 0

# define ETP_PRI_MAX 0

#endif

#ifndef ETP_TYPE_QUIT

# define ETP_TYPE_QUIT 0

#endif

#ifndef ETP_TYPE_GROUP

# define ETP_TYPE_GROUP 1

#endif

#ifndef ETP_CB

typedef void (*ETP_CB) (void *);

# define ETP_CB ETP_CB

#endif

#ifndef ETP_WANT_POLL

# define ETP_WANT_POLL(user) user->want_poll_cb (user->userdata)

#endif

#ifndef ETP_DONE_POLL

# define ETP_DONE_POLL(user) user->done_poll_cb (user->userdata)

#endif

#define ETP_NUM_PRI (ETP_PRI_MAX - ETP_PRI_MIN + 1)

#define ETP_TICKS ((1000000 + 1023) >> 10)

enum {

  ETP_FLAG_GROUPADD = 0x04, /* some request was added to the group */

  ETP_FLAG_DELAYED  = 0x08, /* groiup request has been delayed */

};

/* calculate time difference in ~1/ETP_TICKS of a second */

ecb_inline int

etp_tvdiff (struct timeval *tv1, struct timeval *tv2)

{

  return  (tv2->tv_sec  - tv1->tv_sec ) * ETP_TICKS

       + ((tv2->tv_usec - tv1->tv_usec) >> 10);

}

struct etp_tmpbuf

{

  void *ptr;

  int len;

};

static void *

etp_tmpbuf_get (struct etp_tmpbuf *buf, int len)

{

  if (buf->len < len)

    {

      free (buf->ptr);

      buf->ptr = malloc (buf->len = len);

    }

  return buf->ptr;

}

/*

 * a somewhat faster data structure might be nice, but

 * with 8 priorities this actually needs <20 insns

 * per shift, the most expensive operation.

 */

typedef struct

{

  ETP_REQ *qs[ETP_NUM_PRI], *qe[ETP_NUM_PRI]; /* qstart, qend */

  int size;

} etp_reqq;

typedef struct etp_pool *etp_pool;

typedef struct etp_pool_user *etp_pool_user;

typedef struct etp_worker

{

  etp_pool pool;

  struct etp_tmpbuf tmpbuf;

#ifdef ETP_WORKER_COMMON

  ETP_WORKER_COMMON

#endif

} etp_worker;

struct etp_pool

{

   etp_reqq req_queue;

   unsigned int started, idle, wanted;

   unsigned int nreqs_run;    /* pool->lock */

   unsigned int max_idle;     /* maximum number of threads that can pool->idle indefinitely */

   unsigned int idle_timeout; /* number of seconds after which an pool->idle threads exit */

   xmutex_t lock;

   xcond_t  reqwait;

   xcond_t  wrkwait;

   int (*on_start_cb)(void *data);

   void *on_start_data;

   int (*on_stop_cb)(void *data);

   void *on_stop_data;

};

struct etp_pool_user

{

   etp_pool pool;

   void *userdata;

   etp_reqq res_queue;

   unsigned int max_poll_time;

   unsigned int max_poll_reqs;

   unsigned int nreqs;

   ETP_CB want_poll_cb;

   ETP_CB done_poll_cb;

   xmutex_t lock;

};

/* worker threads management */

static void ecb_cold

etp_worker_free (etp_worker *wrk)

{

  free (wrk->tmpbuf.ptr);

  free (wrk);

}

ETP_API_DECL unsigned int

etp_nreqs (etp_pool_user user)

{

  return user->nreqs;

}

ETP_API_DECL unsigned int

etp_npending (etp_pool_user user)

{

  unsigned int retval;

  if (WORDACCESS_UNSAFE) X_LOCK   (user->lock);

  retval = user->res_queue.size;

  if (WORDACCESS_UNSAFE) X_UNLOCK (user->lock);

  return retval;

}

ETP_API_DECL unsigned int

etp_nthreads (etp_pool pool)

{

  unsigned int retval;

  if (WORDACCESS_UNSAFE) X_LOCK   (pool->lock);

  retval = pool->started;

  if (WORDACCESS_UNSAFE) X_UNLOCK (pool->lock);

  return retval;

}

static void ecb_noinline ecb_cold

reqq_init (etp_reqq *q)

{

  int pri;

  for (pri = 0; pri < ETP_NUM_PRI; ++pri)

    q->qs[pri] = q->qe[pri] = 0;

  q->size = 0;

}

static int ecb_noinline

reqq_push (etp_reqq *q, ETP_REQ *req)

{

  int pri = req->pri;

  req->next = 0;

  if (q->qe[pri])

    {

      q->qe[pri]->next = req;

      q->qe[pri] = req;

    }

  else

    q->qe[pri] = q->qs[pri] = req;

  return q->size++;

}

static ETP_REQ * ecb_noinline

reqq_shift (etp_reqq *q)

{

  int pri;

  if (!q->size)

    return 0;

  --q->size;

  for (pri = ETP_NUM_PRI; pri--; )

    {

      ETP_REQ *req = q->qs[pri];

      if (req)

        {

          if (!(q->qs[pri] = (ETP_REQ *)req->next))

            q->qe[pri] = 0;

          return req;

        }

    }

  abort ();

}

ETP_API_DECL int ecb_cold

etp_init (etp_pool pool)

{

  X_MUTEX_CREATE (pool->lock);

  X_COND_CREATE  (pool->reqwait);

  X_COND_CREATE  (pool->wrkwait);

  reqq_init (&pool->req_queue);

  pool->started  = 0;

  pool->idle     = 0;

  pool->wanted   = 4;

  pool->nreqs_run  = 0;

  pool->max_idle = 4;      /* maximum number of threads that can pool->idle indefinitely */

  pool->idle_timeout = 10; /* number of seconds after which an pool->idle threads exit */

  return 0;

}

ETP_API_DECL int ecb_cold

etp_user_init (etp_pool_user user, void *userdata, ETP_CB want_poll, ETP_CB done_poll)

{

  user->pool = NULL;

  X_MUTEX_CREATE (user->lock);

  reqq_init (&user->res_queue);

  user->max_poll_time = 0;

  user->max_poll_reqs = 0;

  user->nreqs    = 0;

  user->userdata     = userdata;

  user->want_poll_cb = want_poll;

  user->done_poll_cb = done_poll;

  return 0;

}

static void ecb_noinline ecb_cold

etp_proc_init (void)

{

#if HAVE_PRCTL_SET_NAME

  /* provide a more sensible "thread name" */

  char name[16 + 1];

  const int namelen = sizeof (name) - 1;

  int len;

  prctl (PR_GET_NAME, (unsigned long)name, 0, 0, 0);

  name [namelen] = 0;

  len = strlen (name);

  strcpy (name + (len <= namelen - 4 ? len : namelen - 4), "/eio");

  prctl (PR_SET_NAME, (unsigned long)name, 0, 0, 0);

#endif

}

X_THREAD_PROC (etp_proc)

{

  ETP_REQ *req;

  struct timespec ts;

  etp_pool pool = (etp_pool)thr_arg;

  etp_worker self = {};

  self.pool = pool;

  etp_pool_user user; /* per request */

  etp_proc_init ();

  /* try to distribute timeouts somewhat evenly (nanosecond part) */

  ts.tv_nsec = (unsigned long)random() * (1000000000UL / RAND_MAX);

  X_LOCK (pool->lock);

  if (pool->on_start_cb)

    if (pool->on_start_cb(pool->on_start_data))

      goto error;

  for (;;)

    {

      for (;;)

        {

          req = reqq_shift (&pool->req_queue);

          if (ecb_expect_true (req))

            break;

          if (pool->started > pool->wanted) /* someone is shrinking the pool */

            goto quit;

          ++pool->idle;

          if (pool->idle <= pool->max_idle)

            {

              /* we are allowed to pool->idle, so do so without any timeout */

              X_COND_WAIT (pool->reqwait, pool->lock);

              --pool->idle;

            }

          else

            {

              ts.tv_sec = time (0) + pool->idle_timeout;

              if (X_COND_TIMEDWAIT (pool->reqwait, pool->lock, ts) != ETIMEDOUT)

                continue;

              --pool->idle;

              goto quit;

            }

        }

      ++pool->nreqs_run;

      X_UNLOCK (pool->lock);

      user = req->pool_user;

      ETP_EXECUTE (&self, req);

      X_LOCK (user->lock);

      if (!reqq_push (&user->res_queue, req))

        ETP_WANT_POLL (user);

      X_UNLOCK (user->lock);

      X_LOCK (pool->lock);

      --pool->nreqs_run;

    }

quit:

  assert(pool->started > 0);

  pool->started--;

  X_COND_BROADCAST (pool->wrkwait);

  X_UNLOCK (pool->lock);

  if (pool->on_stop_cb)

    pool->on_stop_cb(pool->on_stop_data);

  return 0;

error:

  assert(pool->started > 0);

  pool->started--;

  X_COND_BROADCAST (pool->wrkwait);

  X_UNLOCK (pool->lock);

  return 0;

}

static void ecb_cold

etp_start_thread (etp_pool pool)

{

  xthread_t tid;

  int threads;

  if (xthread_create (&tid, etp_proc, (void *)pool) != 0)

    return;

  X_LOCK (pool->lock);

  assert(pool->started > 0);

  threads = --pool->started;

  X_COND_BROADCAST (pool->wrkwait);

  X_UNLOCK (pool->lock);

  /* Assume if at least one thread managed to start the queue will drain

   * eventually. If not, tasks will never complete; the best we can do

   * is to die now.

   */

  if (threads == 0)

    {

      fputs("failed to start thread in ETP pool", stderr);

      abort();

    }

}

ETP_API_DECL int

etp_poll (etp_pool_user user)

{

  unsigned int maxreqs;

  unsigned int maxtime;

  struct timeval tv_start, tv_now;

  maxreqs = user->max_poll_reqs;

  maxtime = user->max_poll_time;

  if (maxtime)

    gettimeofday (&tv_start, 0);

  for (;;)

    {

      ETP_REQ *req;

      X_LOCK (user->lock);

      req = reqq_shift (&user->res_queue);

      if (ecb_expect_true (req))

        {

          if (ecb_expect_true (user->nreqs))

            --user->nreqs;

          if (!user->res_queue.size)

            ETP_DONE_POLL (user);

        }

      X_UNLOCK (user->lock);

      if (ecb_expect_false (!req))

        return 0;

      if (ecb_expect_false (req->type == ETP_TYPE_GROUP && req->size))

        {

          req->flags |= ETP_FLAG_DELAYED; /* mark request as delayed */

          continue;

        }

      else

        {

          int res = ETP_FINISH (req);

          if (ecb_expect_false (res))

            return res;

        }

      if (ecb_expect_false (maxreqs && !--maxreqs))

        break;

      if (maxtime)

        {

          gettimeofday (&tv_now, 0);

          if (etp_tvdiff (&tv_start, &tv_now) >= maxtime)

            break;

        }

    }

  errno = EAGAIN;

  return -1;

}

ETP_API_DECL void

etp_grp_cancel (etp_pool_user user, ETP_REQ *grp);

ETP_API_DECL void

etp_cancel (etp_pool_user user, ETP_REQ *req)

{

  req->cancelled = 1;

  etp_grp_cancel (user, req);

}

ETP_API_DECL void

etp_grp_cancel (etp_pool_user user, ETP_REQ *grp)

{

  for (grp = grp->grp_first; grp; grp = grp->grp_next)

    etp_cancel (user, grp);

}

ETP_API_DECL void

etp_submit (etp_pool_user user, ETP_REQ *req)

{

  req->pri -= ETP_PRI_MIN;

  if (ecb_expect_false (req->pri < ETP_PRI_MIN - ETP_PRI_MIN)) req->pri = ETP_PRI_MIN - ETP_PRI_MIN;

  if (ecb_expect_false (req->pri > ETP_PRI_MAX - ETP_PRI_MIN)) req->pri = ETP_PRI_MAX - ETP_PRI_MIN;

  user->nreqs++;

  if (ecb_expect_false (req->type == ETP_TYPE_GROUP))

    {

      X_LOCK (user->lock);

      if (!reqq_push (&user->res_queue, req))

        ETP_WANT_POLL (user);

      X_UNLOCK (user->lock);

    }

  else

    {

      etp_pool pool = user->pool;

      int need_thread = 0;

      X_LOCK (pool->lock);

      req->pool_user = user;

      reqq_push (&pool->req_queue, req);

      if (ecb_expect_false(pool->req_queue.size + pool->nreqs_run > pool->started &&

                           pool->started < pool->wanted))

        {

          /* arrange for a thread to start */

          need_thread = 1;

          pool->started++;

        }

      X_COND_SIGNAL (pool->reqwait);

      X_UNLOCK (pool->lock);

      if (ecb_expect_false(need_thread))

        etp_start_thread(pool);

    }

}

ETP_API_DECL void ecb_cold

etp_set_max_poll_time (etp_pool_user user, double seconds)

{

  user->max_poll_time = seconds * ETP_TICKS;

}

ETP_API_DECL void ecb_cold

etp_set_max_poll_reqs (etp_pool_user user, unsigned int maxreqs)

{

  user->max_poll_reqs = maxreqs;

}

ETP_API_DECL void ecb_cold

etp_set_thread_on_start(etp_pool pool, int (*on_start_cb)(void *), void *data)

{

  pool->on_start_cb = on_start_cb;

  pool->on_start_data = data;

}

ETP_API_DECL void ecb_cold

etp_set_thread_on_stop(etp_pool pool, int (*on_stop_cb)(void *), void *data)

{

  pool->on_stop_cb = on_stop_cb;

  pool->on_stop_data = data;

}

ETP_API_DECL void ecb_cold

etp_set_max_idle (etp_pool pool, unsigned int threads)

{

  X_LOCK   (pool->lock);

  pool->max_idle = threads;

  X_UNLOCK (pool->lock);

}

ETP_API_DECL void ecb_cold

etp_set_idle_timeout (etp_pool pool, unsigned int seconds)

{

  X_LOCK   (pool->lock);

  pool->idle_timeout = seconds;

  X_UNLOCK (pool->lock);

}

ETP_API_DECL void ecb_cold

etp_set_min_parallel (etp_pool pool, unsigned int threads)

{

  X_LOCK   (pool->lock);

  if (pool->wanted < threads)

    pool->wanted = threads;

  X_UNLOCK (pool->lock);

}

ETP_API_DECL int ecb_cold

etp_set_max_parallel (etp_pool pool, unsigned int threads)

{

  int retval;

  X_LOCK   (pool->lock);

  retval = pool->wanted;

  if (pool->wanted > threads)

    pool->wanted = threads;

  while (pool->started > pool->wanted)

    {

      X_COND_BROADCAST(pool->reqwait); /* wake idle threads */

      X_COND_WAIT(pool->wrkwait, pool->lock);

    }

  X_UNLOCK (pool->lock);

  return retval;

}