BDB/BDB.xs

/* solaris */
#define _POSIX_PTHREAD_SEMANTICS 1

#if __linux && !defined(_GNU_SOURCE)
# define _GNU_SOURCE
#endif

/* just in case */
#define _REENTRANT 1

#include <errno.h>

#include "EXTERN.h"
#include "perl.h"
#include "XSUB.h"

#include <pthread.h>

#include <stddef.h>
#include <stdlib.h>
#include <errno.h>
#include <sys/time.h>
#include <sys/types.h>
#include <limits.h>
#include <unistd.h>
#include <fcntl.h>

#include <db.h>

/* number of seconds after which idle threads exit */
#define IDLE_TIMEOUT 10

/* wether word reads are potentially non-atomic.
 * this is conservatice, likely most arches this runs
 * on have atomic word read/writes.
 */
#ifndef WORDACCESS_UNSAFE
# if __i386 || __x86_64
#  define WORDACCESS_UNSAFE 0
# else
#  define WORDACCESS_UNSAFE 1
# endif
#endif

typedef SV SV8; /* byte-sv, used for argument-checking */

enum
{
  REQ_QUIT,
  REQ_ENV_OPEN, REQ_ENV_CLOSE,
  REQ_DB_OPEN, REQ_DB_CLOSE,
};

typedef struct aio_cb
{
  struct aio_cb *volatile next;
  SV *callback;
  int type, pri, errorno;

  DB_ENV *env;
  DB *db;
  DB_TXN *txn;
  DBC *cursor;
  int int1, int2;
  U32 uint1, uint2;
  char *buf1, *buf2;
} aio_cb;

typedef aio_cb *aio_req;

enum {
  PRI_MIN     = -4,
  PRI_MAX     =  4,

  DEFAULT_PRI = 0,
  PRI_BIAS    = -PRI_MIN,
  NUM_PRI     = PRI_MAX + PRI_BIAS + 1,
};

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

static unsigned int max_poll_time = 0;
static unsigned int max_poll_reqs = 0;

/* calculcate time difference in ~1/AIO_TICKS of a second */
static int tvdiff (struct timeval *tv1, struct timeval *tv2)
{
  return  (tv2->tv_sec  - tv1->tv_sec ) * AIO_TICKS
       + ((tv2->tv_usec - tv1->tv_usec) >> 10);
}

static int next_pri = DEFAULT_PRI + PRI_BIAS;

static unsigned int started, idle, wanted;

#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP)
# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP
#else
# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER
#endif

#define LOCK(mutex)   pthread_mutex_lock   (&(mutex))
#define UNLOCK(mutex) pthread_mutex_unlock (&(mutex))

/* worker threads management */
static pthread_mutex_t wrklock = AIO_MUTEX_INIT;

typedef struct worker {
  /* locked by wrklock */
  struct worker *prev, *next;

  pthread_t tid;

  /* locked by reslock, reqlock or wrklock */
  aio_req req; /* currently processed request */
  void *dbuf;
  DIR *dirp;
} worker;

static worker wrk_first = { &wrk_first, &wrk_first, 0 };

static void worker_clear (worker *wrk)
{
}

static void worker_free (worker *wrk)
{
  wrk->next->prev = wrk->prev;
  wrk->prev->next = wrk->next;

  free (wrk);
}

static volatile unsigned int nreqs, nready, npending;
static volatile unsigned int max_idle = 4;
static volatile unsigned int max_outstanding = 0xffffffff;
static int respipe [2];

static pthread_mutex_t reslock = AIO_MUTEX_INIT;
static pthread_mutex_t reqlock = AIO_MUTEX_INIT;
static pthread_cond_t  reqwait = PTHREAD_COND_INITIALIZER;

#if WORDACCESS_UNSAFE

static unsigned int get_nready ()
{
  unsigned int retval;

  LOCK   (reqlock);
  retval = nready;
  UNLOCK (reqlock);

  return retval;
}

static unsigned int get_npending ()
{
  unsigned int retval;

  LOCK   (reslock);
  retval = npending;
  UNLOCK (reslock);

  return retval;
}

static unsigned int get_nthreads ()
{
  unsigned int retval;

  LOCK   (wrklock);
  retval = started;
  UNLOCK (wrklock);

  return retval;
}

#else

# define get_nready()   nready
# define get_npending() npending
# define get_nthreads() started

#endif

/*
 * 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 {
  aio_req qs[NUM_PRI], qe[NUM_PRI]; /* qstart, qend */
  int size;
} reqq;

static reqq req_queue;
static reqq res_queue;

int reqq_push (reqq *q, aio_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++;
}

aio_req reqq_shift (reqq *q)
{
  int pri;

  if (!q->size)
    return 0;

  --q->size;

  for (pri = NUM_PRI; pri--; )
    {
      aio_req req = q->qs[pri];

      if (req)
        {
          if (!(q->qs[pri] = req->next))
            q->qe[pri] = 0;

          return req;
        }
    }

  abort ();
}

static int poll_cb ();
static void req_free (aio_req req);
static void req_cancel (aio_req req);

static int req_invoke (aio_req req)
{
  dSP;

  if (SvOK (req->callback))
    {
      ENTER;
      SAVETMPS;
      PUSHMARK (SP);
      EXTEND (SP, 1);

      switch (req->type)
        {
        }

      PUTBACK;
      call_sv (req->callback, G_VOID | G_EVAL);
      SPAGAIN;

      FREETMPS;
      LEAVE;
    }

  return !SvTRUE (ERRSV);
}

static void req_free (aio_req req)
{
  free (req->buf1);
  free (req->buf2);
  Safefree (req);
}

static void *aio_proc(void *arg);

static void start_thread (void)
{
  sigset_t fullsigset, oldsigset;
  pthread_attr_t attr;

  worker *wrk = calloc (1, sizeof (worker));

  if (!wrk)
    croak ("unable to allocate worker thread data");

  pthread_attr_init (&attr);
  pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
#ifdef PTHREAD_SCOPE_PROCESS
  pthread_attr_setscope (&attr, PTHREAD_SCOPE_PROCESS);
#endif

  sigfillset (&fullsigset);

  LOCK (wrklock);
  pthread_sigmask (SIG_SETMASK, &fullsigset, &oldsigset);

  if (pthread_create (&wrk->tid, &attr, aio_proc, (void *)wrk) == 0)
    {
      wrk->prev = &wrk_first;
      wrk->next = wrk_first.next;
      wrk_first.next->prev = wrk;
      wrk_first.next = wrk;
      ++started;
    }
  else
    free (wrk);

  pthread_sigmask (SIG_SETMASK, &oldsigset, 0);
  UNLOCK (wrklock);
}

static void maybe_start_thread ()
{
  if (get_nthreads () >= wanted)
    return;
  
  /* todo: maybe use idle here, but might be less exact */
  if (0 <= (int)get_nthreads () + (int)get_npending () - (int)nreqs)
    return;

  start_thread ();
}

static void req_send (aio_req req)
{
  ++nreqs;

  LOCK (reqlock);
  ++nready;
  reqq_push (&req_queue, req);
  pthread_cond_signal (&reqwait);
  UNLOCK (reqlock);

  maybe_start_thread ();
}

static void end_thread (void)
{
  aio_req req;

  Newz (0, req, 1, aio_cb);

  req->type = REQ_QUIT;
  req->pri  = PRI_MAX + PRI_BIAS;

  LOCK (reqlock);
  reqq_push (&req_queue, req);
  pthread_cond_signal (&reqwait);
  UNLOCK (reqlock);

  LOCK (wrklock);
  --started;
  UNLOCK (wrklock);
}

static void set_max_idle (int nthreads)
{
  if (WORDACCESS_UNSAFE) LOCK   (reqlock);
  max_idle = nthreads <= 0 ? 1 : nthreads;
  if (WORDACCESS_UNSAFE) UNLOCK (reqlock);
}

static void min_parallel (int nthreads)
{
  if (wanted < nthreads)
    wanted = nthreads;
}

static void max_parallel (int nthreads)
{
  if (wanted > nthreads)
    wanted = nthreads;

  while (started > wanted)
    end_thread ();
}

static void poll_wait ()
{
  fd_set rfd;

  while (nreqs)
    {
      int size;
      if (WORDACCESS_UNSAFE) LOCK   (reslock);
      size = res_queue.size;
      if (WORDACCESS_UNSAFE) UNLOCK (reslock);

      if (size)
        return;

      maybe_start_thread ();

      FD_ZERO(&rfd);
      FD_SET(respipe [0], &rfd);

      select (respipe [0] + 1, &rfd, 0, 0, 0);
    }
}

static int poll_cb ()
{
  dSP;
  int count = 0;
  int maxreqs = max_poll_reqs;
  int do_croak = 0;
  struct timeval tv_start, tv_now;
  aio_req req;

  if (max_poll_time)
    gettimeofday (&tv_start, 0);

  for (;;)
    {
      for (;;)
        {
          maybe_start_thread ();

          LOCK (reslock);
          req = reqq_shift (&res_queue);

          if (req)
            {
              --npending;

              if (!res_queue.size)
                {
                  /* read any signals sent by the worker threads */
                  char buf [4];
                  while (read (respipe [0], buf, 4) == 4)
                    ;
                }
            }

          UNLOCK (reslock);

          if (!req)
            break;

          --nreqs;

          if (!req_invoke (req))
            {
              req_free (req);
              croak (0);
            }

          count++;

          req_free (req);

          if (maxreqs && !--maxreqs)
            break;

          if (max_poll_time)
            {
              gettimeofday (&tv_now, 0);

              if (tvdiff (&tv_start, &tv_now) >= max_poll_time)
                break;
            }
        }

      if (nreqs <= max_outstanding)
        break;

      poll_wait ();

      ++maxreqs;
    }

  return count;
}

static void create_pipe ()
{
  if (pipe (respipe))
    croak ("unable to initialize result pipe");

  if (fcntl (respipe [0], F_SETFL, O_NONBLOCK))
    croak ("cannot set result pipe to nonblocking mode");

  if (fcntl (respipe [1], F_SETFL, O_NONBLOCK))
    croak ("cannot set result pipe to nonblocking mode");
}

/*****************************************************************************/

static void *aio_proc (void *thr_arg)
{
  aio_req req;
  struct timespec ts;
  worker *self = (worker *)thr_arg;

  /* try to distribute timeouts somewhat evenly */
  ts.tv_nsec = (((unsigned long)self + (unsigned long)ts.tv_sec) & 1023UL)
               * (1000000000UL / 1024UL);

  for (;;)
    {
      ts.tv_sec  = time (0) + IDLE_TIMEOUT;

      LOCK (reqlock);

      for (;;)
        {
          self->req = req = reqq_shift (&req_queue);

          if (req)
            break;

          ++idle;

          if (pthread_cond_timedwait (&reqwait, &reqlock, &ts)
              == ETIMEDOUT)
            {
              if (idle > max_idle)
                {
                  --idle;
                  UNLOCK (reqlock);
                  LOCK (wrklock);
                  --started;
                  UNLOCK (wrklock);
                  goto quit;
                }

              /* we are allowed to idle, so do so without any timeout */
              pthread_cond_wait (&reqwait, &reqlock);
              ts.tv_sec  = time (0) + IDLE_TIMEOUT;
            }

          --idle;
        }

      --nready;

      UNLOCK (reqlock);
     
      errno = 0; /* strictly unnecessary */

      switch (req->type)
        {
          case REQ_QUIT:
            goto quit;

          default:
            //req->result = ENOSYS;
            break;
        }

      //req->errorno = errno;

      LOCK (reslock);

      ++npending;

      if (!reqq_push (&res_queue, req))
        /* write a dummy byte to the pipe so fh becomes ready */
        write (respipe [1], &respipe, 1);

      self->req = 0;
      worker_clear (self);

      UNLOCK (reslock);
    }

quit:
  LOCK (wrklock);
  worker_free (self);
  UNLOCK (wrklock);

  return 0;
}

/*****************************************************************************/

static void atfork_prepare (void)
{
  LOCK (wrklock);
  LOCK (reqlock);
  LOCK (reslock);
}

static void atfork_parent (void)
{
  UNLOCK (reslock);
  UNLOCK (reqlock);
  UNLOCK (wrklock);
}

static void atfork_child (void)
{
  aio_req prv;

  while (prv = reqq_shift (&req_queue))
    req_free (prv);

  while (prv = reqq_shift (&res_queue))
    req_free (prv);

  while (wrk_first.next != &wrk_first)
    {
      worker *wrk = wrk_first.next;

      if (wrk->req)
        req_free (wrk->req);

      worker_clear (wrk);
      worker_free (wrk);
    }

  started  = 0;
  idle     = 0;
  nreqs    = 0;
  nready   = 0;
  npending = 0;

  close (respipe [0]);
  close (respipe [1]);
  create_pipe ();

  atfork_parent ();
}

#define dREQ(reqtype)                                           \
  aio_req req;                                                  \
  int req_pri = next_pri;                                       \
  next_pri = DEFAULT_PRI + PRI_BIAS;                            \
                                                                \
  if (SvOK (callback) && !SvROK (callback))                     \
    croak ("callback must be undef or of reference type");      \
                                                                \
  Newz (0, req, 1, aio_cb);                                     \
  if (!req)                                                     \
    croak ("out of memory during aio_req allocation");          \
                                                                \
  req->callback = newSVsv (callback);                           \
  req->type = (reqtype);                                        \
  req->pri = req_pri

#define REQ_SEND                                                \
  req_send (req)

#define SvPTR(var, arg, type, class)                            \
  if (!SvOK (arg))                                              \
    (var) = 0;                                                  \
  else if (sv_derived_from ((arg), # class))                    \
    {                                                           \
      IV tmp = SvIV ((SV*) SvRV (arg));                         \
      (var) = INT2PTR (type, tmp);                              \
    }                                                           \
  else                                                          \
    Perl_croak (# var " is not of type " # type)
        
MODULE = BDB                PACKAGE = BDB

PROTOTYPES: ENABLE

BOOT:
{
        HV *stash = gv_stashpv ("BDB", 1);

        static const struct {
          const char *name;
          IV iv;
        } *civ, const_iv[] = {
#define const_iv(name) { # name, (IV)DB_ ## name },
          const_iv (RPCCLIENT)
          const_iv (INIT_CDB)
          const_iv (INIT_LOCK)
          const_iv (INIT_LOG)
          const_iv (INIT_MPOOL)
          const_iv (INIT_REP)
          const_iv (INIT_TXN)
          const_iv (RECOVER)
          const_iv (INIT_TXN)
          const_iv (RECOVER_FATAL)
          const_iv (CREATE)
          const_iv (USE_ENVIRON)
          const_iv (USE_ENVIRON_ROOT)
          const_iv (LOCKDOWN)
          const_iv (PRIVATE)
          const_iv (REGISTER)
          const_iv (SYSTEM_MEM)
          const_iv (AUTO_COMMIT)
          const_iv (CDB_ALLDB)
          const_iv (DIRECT_DB)
          const_iv (DIRECT_LOG)
          const_iv (DSYNC_DB)
          const_iv (DSYNC_LOG)
          const_iv (LOG_AUTOREMOVE)
          const_iv (LOG_INMEMORY)
          const_iv (NOLOCKING)
          const_iv (MULTIVERSION)
          const_iv (NOMMAP)
          const_iv (NOPANIC)
          const_iv (OVERWRITE)
          const_iv (PANIC_ENVIRONMENT)
          const_iv (REGION_INIT)
          const_iv (TIME_NOTGRANTED)
          const_iv (TXN_NOSYNC)
          const_iv (TXN_SNAPSHOT)
          const_iv (TXN_WRITE_NOSYNC)
          const_iv (YIELDCPU)
          const_iv (ENCRYPT_AES)
          const_iv (XA_CREATE)
          const_iv (BTREE)
          const_iv (HASH)
          const_iv (QUEUE)
          const_iv (RECNO)
          const_iv (UNKNOWN)
          const_iv (EXCL)
          const_iv (READ_UNCOMMITTED)
          const_iv (TRUNCATE)
          const_iv (NOSYNC)
          const_iv (CHKSUM)
          const_iv (ENCRYPT)
          const_iv (TXN_NOT_DURABLE)
          const_iv (DUP)
          const_iv (DUPSORT)
          const_iv (RECNUM)
          const_iv (RENUMBER)
          const_iv (REVSPLITOFF)
          const_iv (INORDER)
          const_iv (CONSUME)
          const_iv (CONSUME_WAIT)
          const_iv (SNAPSHOT)
          const_iv (JOIN_ITEM)
          const_iv (RMW)

          const_iv (NOTFOUND)
          const_iv (KEYEMPTY)
          const_iv (LOCK_DEADLOCK)
          const_iv (LOCK_NOTGRANTED)
          const_iv (RUNRECOVERY)
        };

        for (civ = const_iv + sizeof (const_iv) / sizeof (const_iv [0]); civ-- > const_iv; )
          newCONSTSUB (stash, (char *)civ->name, newSViv (civ->iv));

        create_pipe ();
        pthread_atfork (atfork_prepare, atfork_parent, atfork_child);
}

void
max_poll_reqs (int nreqs)
        PROTOTYPE: $
        CODE:
        max_poll_reqs = nreqs;

void
max_poll_time (double nseconds)
        PROTOTYPE: $
        CODE:
        max_poll_time = nseconds * AIO_TICKS;

void
min_parallel (int nthreads)
        PROTOTYPE: $

void
max_parallel (int nthreads)
        PROTOTYPE: $

void
max_idle (int nthreads)
        PROTOTYPE: $
        CODE:
        set_max_idle (nthreads);

int
max_outstanding (int maxreqs)
        PROTOTYPE: $
        CODE:
        RETVAL = max_outstanding;
        max_outstanding = maxreqs;
        OUTPUT:
        RETVAL

int
bdbreq_pri (int pri = 0)
        PROTOTYPE: ;$
        CODE:
        RETVAL = next_pri - PRI_BIAS;
        if (items > 0)
          {
            if (pri < PRI_MIN) pri = PRI_MIN;
            if (pri > PRI_MAX) pri = PRI_MAX;
            next_pri = pri + PRI_BIAS;
          }
        OUTPUT:
        RETVAL

void
bdbreq_nice (int nice = 0)
        CODE:
        nice = next_pri - nice;
        if (nice < PRI_MIN) nice = PRI_MIN;
        if (nice > PRI_MAX) nice = PRI_MAX;
        next_pri = nice + PRI_BIAS;

void
flush ()
        PROTOTYPE:
        CODE:
        while (nreqs)
          {
            poll_wait ();
            poll_cb ();
          }

int
poll()
        PROTOTYPE:
        CODE:
        poll_wait ();
        RETVAL = poll_cb ();
        OUTPUT:
        RETVAL

int
poll_fileno()
        PROTOTYPE:
        CODE:
        RETVAL = respipe [0];
        OUTPUT:
        RETVAL

int
poll_cb(...)
        PROTOTYPE:
        CODE:
        RETVAL = poll_cb ();
        OUTPUT:
        RETVAL

void
poll_wait()
        PROTOTYPE:
        CODE:
        poll_wait ();

int
nreqs()
        PROTOTYPE:
        CODE:
        RETVAL = nreqs;
        OUTPUT:
        RETVAL

int
nready()
        PROTOTYPE:
        CODE:
        RETVAL = get_nready ();
        OUTPUT:
        RETVAL

int
npending()
        PROTOTYPE:
        CODE:
        RETVAL = get_npending ();
        OUTPUT:
        RETVAL

int
nthreads()
        PROTOTYPE:
        CODE:
        if (WORDACCESS_UNSAFE) LOCK   (wrklock);
        RETVAL = started;
        if (WORDACCESS_UNSAFE) UNLOCK (wrklock);
        OUTPUT:
        RETVAL

DB_ENV *
bdb_env_create (U32 env_flags = 0)
        CODE:
{
        int err = db_env_create (&RETVAL, env_flags);
        if (err)
          croak ("db_env_create: %s", db_strerror (err));
}

void
bdb_env_open (DB_ENV *env, char *db_home, U32 open_flags, int mode, SV *callback = 0)
        CODE:
{
        dREQ (REQ_ENV_OPEN);
        req->env   = env;
        req->uint1 = open_flags;
        req->int1  = mode;
        req->buf1  = strdup (db_home);
        REQ_SEND;
}

void
bdb_env_close (DB_ENV *env, U32 flags = 0, SV *callback = 0)
        CODE:
{
        dREQ (REQ_ENV_CLOSE);
        req->env   = env;
        req->uint1 = flags;
        REQ_SEND;
}

DB *
bdb_db_create (DB_ENV *env = 0, U32 flags = 0)
        CODE:
{
        int err = db_create (&RETVAL, env, flags);
        if (err)
          croak ("db_env_create: %s", db_strerror (err));
}

void
bdb_db_open (DB *db, DB_TXN *txnid, const char *file, const char *database, int type, U32 flags, int mode, SV *callback = 0)
        CODE:
{
        dREQ (REQ_DB_OPEN);
        req->db    = db;
        req->txn   = txnid;
        req->buf1  = strdup (file);
        req->buf2  = strdup (database);
        req->int1  = type;
        req->uint1 = flags;
        req->int2  = mode;
        REQ_SEND;
}

void
bdb_db_close (DB *db, U32 flags = 0, SV *callback = 0)
        CODE:
{
        dREQ (REQ_DB_CLOSE);
        req->db    = db;
        req->uint1 = flags;
        REQ_SEND;
}


MODULE = BDB            PACKAGE = BDB::Env

int set_cachesize (DB_ENV *env, U32 gbytes, U32 bytes, int ncache = 0)

int set_flags (DB_ENV *env, U32 flags, int onoff)

int set_encrypt (DB_ENV *env, const char *password, U32 flags)

MODULE = BDB            PACKAGE = BDB::Db

int set_cachesize (DB *db, U32 gbytes, U32 bytes, int ncache = 0)

int set_flags (DB *env, U32 flags, int onoff)

int set_encrypt (DB *db, const char *password, U32 flags)

int set_lorder (DB *db, int lorder)


int set_bt_minkey (DB *db, U32 minkey)

int set_re_delim(DB *db, int delim);

int set_re_pad (DB *db, int re_pad)

int set_re_source (DB *db, char *source)

int set_re_len (DB *db, U32 re_len)

int set_h_ffactor (DB *db, U32 h_ffactor)

int set_h_nelem (DB *db, U32 h_nelem)

int set_q_extentsize (DB *db, U32 extentsize)


Revision:	1.2
Committed:	Mon Feb 5 20:21:38 2007 UTC (17 years, 3 months ago) by root
Branch:	MAIN
Changes since 1.1:	+215 -20 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	/* solaris */
2			#define _POSIX_PTHREAD_SEMANTICS 1
3
4			#if __linux && !defined(_GNU_SOURCE)
5			# define _GNU_SOURCE
6			#endif
7
8			/* just in case */
9			#define _REENTRANT 1
10
11			#include <errno.h>
12
13			#include "EXTERN.h"
14			#include "perl.h"
15			#include "XSUB.h"
16
17			#include <pthread.h>
18
19			#include <stddef.h>
20			#include <stdlib.h>
21			#include <errno.h>
22			#include <sys/time.h>
23			#include <sys/types.h>
24			#include <limits.h>
25			#include <unistd.h>
26			#include <fcntl.h>
27	root	1.2
28			#include <db.h>
29	root	1.1
30			/* number of seconds after which idle threads exit */
31			#define IDLE_TIMEOUT 10
32
33			/* wether word reads are potentially non-atomic.
34			* this is conservatice, likely most arches this runs
35			* on have atomic word read/writes.
36			*/
37			#ifndef WORDACCESS_UNSAFE
38			# if __i386 \|\| __x86_64
39			# define WORDACCESS_UNSAFE 0
40			# else
41			# define WORDACCESS_UNSAFE 1
42			# endif
43			#endif
44
45			typedef SV SV8; /* byte-sv, used for argument-checking */
46
47	root	1.2	enum
48			{
49	root	1.1	REQ_QUIT,
50	root	1.2	REQ_ENV_OPEN, REQ_ENV_CLOSE,
51			REQ_DB_OPEN, REQ_DB_CLOSE,
52	root	1.1	};
53
54			typedef struct aio_cb
55			{
56	root	1.2	struct aio_cb *volatile next;
57			SV *callback;
58			int type, pri, errorno;
59
60			DB_ENV *env;
61			DB *db;
62			DB_TXN *txn;
63			DBC *cursor;
64			int int1, int2;
65			U32 uint1, uint2;
66			char buf1, buf2;
67	root	1.1	} aio_cb;
68
69			typedef aio_cb *aio_req;
70
71			enum {
72			PRI_MIN = -4,
73			PRI_MAX = 4,
74
75			DEFAULT_PRI = 0,
76			PRI_BIAS = -PRI_MIN,
77			NUM_PRI = PRI_MAX + PRI_BIAS + 1,
78			};
79
80			#define AIO_TICKS ((1000000 + 1023) >> 10)
81
82			static unsigned int max_poll_time = 0;
83			static unsigned int max_poll_reqs = 0;
84
85			/* calculcate time difference in ~1/AIO_TICKS of a second */
86			static int tvdiff (struct timeval tv1, struct timeval tv2)
87			{
88			return (tv2->tv_sec - tv1->tv_sec ) * AIO_TICKS
89			+ ((tv2->tv_usec - tv1->tv_usec) >> 10);
90			}
91
92			static int next_pri = DEFAULT_PRI + PRI_BIAS;
93
94			static unsigned int started, idle, wanted;
95
96			#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP)
97			# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP
98			#else
99			# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER
100			#endif
101
102			#define LOCK(mutex) pthread_mutex_lock (&(mutex))
103			#define UNLOCK(mutex) pthread_mutex_unlock (&(mutex))
104
105			/* worker threads management */
106			static pthread_mutex_t wrklock = AIO_MUTEX_INIT;
107
108			typedef struct worker {
109			/* locked by wrklock */
110			struct worker prev, next;
111
112			pthread_t tid;
113
114			/* locked by reslock, reqlock or wrklock */
115			aio_req req; /* currently processed request */
116			void *dbuf;
117			DIR *dirp;
118			} worker;
119
120			static worker wrk_first = { &wrk_first, &wrk_first, 0 };
121
122			static void worker_clear (worker *wrk)
123			{
124			}
125
126			static void worker_free (worker *wrk)
127			{
128			wrk->next->prev = wrk->prev;
129			wrk->prev->next = wrk->next;
130
131			free (wrk);
132			}
133
134			static volatile unsigned int nreqs, nready, npending;
135			static volatile unsigned int max_idle = 4;
136			static volatile unsigned int max_outstanding = 0xffffffff;
137			static int respipe [2];
138
139			static pthread_mutex_t reslock = AIO_MUTEX_INIT;
140			static pthread_mutex_t reqlock = AIO_MUTEX_INIT;
141			static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
142
143			#if WORDACCESS_UNSAFE
144
145			static unsigned int get_nready ()
146			{
147			unsigned int retval;
148
149			LOCK (reqlock);
150			retval = nready;
151			UNLOCK (reqlock);
152
153			return retval;
154			}
155
156			static unsigned int get_npending ()
157			{
158			unsigned int retval;
159
160			LOCK (reslock);
161			retval = npending;
162			UNLOCK (reslock);
163
164			return retval;
165			}
166
167			static unsigned int get_nthreads ()
168			{
169			unsigned int retval;
170
171			LOCK (wrklock);
172			retval = started;
173			UNLOCK (wrklock);
174
175			return retval;
176			}
177
178			#else
179
180			# define get_nready() nready
181			# define get_npending() npending
182			# define get_nthreads() started
183
184			#endif
185
186			/*
187			* a somewhat faster data structure might be nice, but
188			* with 8 priorities this actually needs <20 insns
189			* per shift, the most expensive operation.
190			*/
191			typedef struct {
192			aio_req qs[NUM_PRI], qe[NUM_PRI]; /* qstart, qend */
193			int size;
194			} reqq;
195
196			static reqq req_queue;
197			static reqq res_queue;
198
199			int reqq_push (reqq *q, aio_req req)
200			{
201			int pri = req->pri;
202			req->next = 0;
203
204			if (q->qe[pri])
205			{
206			q->qe[pri]->next = req;
207			q->qe[pri] = req;
208			}
209			else
210			q->qe[pri] = q->qs[pri] = req;
211
212			return q->size++;
213			}
214
215			aio_req reqq_shift (reqq *q)
216			{
217			int pri;
218
219			if (!q->size)
220			return 0;
221
222			--q->size;
223
224			for (pri = NUM_PRI; pri--; )
225			{
226			aio_req req = q->qs[pri];
227
228			if (req)
229			{
230			if (!(q->qs[pri] = req->next))
231			q->qe[pri] = 0;
232
233			return req;
234			}
235			}
236
237			abort ();
238			}
239
240			static int poll_cb ();
241			static void req_free (aio_req req);
242			static void req_cancel (aio_req req);
243
244			static int req_invoke (aio_req req)
245			{
246			dSP;
247
248			if (SvOK (req->callback))
249			{
250			ENTER;
251			SAVETMPS;
252			PUSHMARK (SP);
253			EXTEND (SP, 1);
254
255			switch (req->type)
256			{
257			}
258
259			PUTBACK;
260			call_sv (req->callback, G_VOID \| G_EVAL);
261			SPAGAIN;
262
263			FREETMPS;
264			LEAVE;
265			}
266
267			return !SvTRUE (ERRSV);
268			}
269
270			static void req_free (aio_req req)
271			{
272	root	1.2	free (req->buf1);
273			free (req->buf2);
274	root	1.1	Safefree (req);
275			}
276
277			static void aio_proc(void arg);
278
279			static void start_thread (void)
280			{
281			sigset_t fullsigset, oldsigset;
282			pthread_attr_t attr;
283
284			worker *wrk = calloc (1, sizeof (worker));
285
286			if (!wrk)
287			croak ("unable to allocate worker thread data");
288
289			pthread_attr_init (&attr);
290			pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
291			#ifdef PTHREAD_SCOPE_PROCESS
292			pthread_attr_setscope (&attr, PTHREAD_SCOPE_PROCESS);
293			#endif
294
295			sigfillset (&fullsigset);
296
297			LOCK (wrklock);
298			pthread_sigmask (SIG_SETMASK, &fullsigset, &oldsigset);
299
300			if (pthread_create (&wrk->tid, &attr, aio_proc, (void *)wrk) == 0)
301			{
302			wrk->prev = &wrk_first;
303			wrk->next = wrk_first.next;
304			wrk_first.next->prev = wrk;
305			wrk_first.next = wrk;
306			++started;
307			}
308			else
309			free (wrk);
310
311			pthread_sigmask (SIG_SETMASK, &oldsigset, 0);
312			UNLOCK (wrklock);
313			}
314
315			static void maybe_start_thread ()
316			{
317			if (get_nthreads () >= wanted)
318			return;
319
320			/* todo: maybe use idle here, but might be less exact */
321			if (0 <= (int)get_nthreads () + (int)get_npending () - (int)nreqs)
322			return;
323
324			start_thread ();
325			}
326
327			static void req_send (aio_req req)
328			{
329			++nreqs;
330
331			LOCK (reqlock);
332			++nready;
333			reqq_push (&req_queue, req);
334			pthread_cond_signal (&reqwait);
335			UNLOCK (reqlock);
336
337			maybe_start_thread ();
338			}
339
340			static void end_thread (void)
341			{
342			aio_req req;
343
344			Newz (0, req, 1, aio_cb);
345
346			req->type = REQ_QUIT;
347			req->pri = PRI_MAX + PRI_BIAS;
348
349			LOCK (reqlock);
350			reqq_push (&req_queue, req);
351			pthread_cond_signal (&reqwait);
352			UNLOCK (reqlock);
353
354			LOCK (wrklock);
355			--started;
356			UNLOCK (wrklock);
357			}
358
359			static void set_max_idle (int nthreads)
360			{
361			if (WORDACCESS_UNSAFE) LOCK (reqlock);
362			max_idle = nthreads <= 0 ? 1 : nthreads;
363			if (WORDACCESS_UNSAFE) UNLOCK (reqlock);
364			}
365
366			static void min_parallel (int nthreads)
367			{
368			if (wanted < nthreads)
369			wanted = nthreads;
370			}
371
372			static void max_parallel (int nthreads)
373			{
374			if (wanted > nthreads)
375			wanted = nthreads;
376
377			while (started > wanted)
378			end_thread ();
379			}
380
381			static void poll_wait ()
382			{
383			fd_set rfd;
384
385			while (nreqs)
386			{
387			int size;
388			if (WORDACCESS_UNSAFE) LOCK (reslock);
389			size = res_queue.size;
390			if (WORDACCESS_UNSAFE) UNLOCK (reslock);
391
392			if (size)
393			return;
394
395			maybe_start_thread ();
396
397			FD_ZERO(&rfd);
398			FD_SET(respipe [0], &rfd);
399
400			select (respipe [0] + 1, &rfd, 0, 0, 0);
401			}
402			}
403
404			static int poll_cb ()
405			{
406			dSP;
407			int count = 0;
408			int maxreqs = max_poll_reqs;
409			int do_croak = 0;
410			struct timeval tv_start, tv_now;
411			aio_req req;
412
413			if (max_poll_time)
414			gettimeofday (&tv_start, 0);
415
416			for (;;)
417			{
418			for (;;)
419			{
420			maybe_start_thread ();
421
422			LOCK (reslock);
423			req = reqq_shift (&res_queue);
424
425			if (req)
426			{
427			--npending;
428
429			if (!res_queue.size)
430			{
431			/* read any signals sent by the worker threads */
432			char buf [4];
433			while (read (respipe [0], buf, 4) == 4)
434			;
435			}
436			}
437
438			UNLOCK (reslock);
439
440			if (!req)
441			break;
442
443			--nreqs;
444
445			if (!req_invoke (req))
446			{
447			req_free (req);
448			croak (0);
449			}
450
451			count++;
452
453			req_free (req);
454
455			if (maxreqs && !--maxreqs)
456			break;
457
458			if (max_poll_time)
459			{
460			gettimeofday (&tv_now, 0);
461
462			if (tvdiff (&tv_start, &tv_now) >= max_poll_time)
463			break;
464			}
465			}
466
467			if (nreqs <= max_outstanding)
468			break;
469
470			poll_wait ();
471
472			++maxreqs;
473			}
474
475			return count;
476			}
477
478			static void create_pipe ()
479			{
480			if (pipe (respipe))
481			croak ("unable to initialize result pipe");
482
483			if (fcntl (respipe [0], F_SETFL, O_NONBLOCK))
484			croak ("cannot set result pipe to nonblocking mode");
485
486			if (fcntl (respipe [1], F_SETFL, O_NONBLOCK))
487			croak ("cannot set result pipe to nonblocking mode");
488			}
489
490			/*****************************************************************************/
491
492			static void aio_proc (void thr_arg)
493			{
494			aio_req req;
495			struct timespec ts;
496			worker self = (worker )thr_arg;
497
498			/* try to distribute timeouts somewhat evenly */
499			ts.tv_nsec = (((unsigned long)self + (unsigned long)ts.tv_sec) & 1023UL)
500			* (1000000000UL / 1024UL);
501
502			for (;;)
503			{
504			ts.tv_sec = time (0) + IDLE_TIMEOUT;
505
506			LOCK (reqlock);
507
508			for (;;)
509			{
510			self->req = req = reqq_shift (&req_queue);
511
512			if (req)
513			break;
514
515			++idle;
516
517			if (pthread_cond_timedwait (&reqwait, &reqlock, &ts)
518			== ETIMEDOUT)
519			{
520			if (idle > max_idle)
521			{
522			--idle;
523			UNLOCK (reqlock);
524			LOCK (wrklock);
525			--started;
526			UNLOCK (wrklock);
527			goto quit;
528			}
529
530			/* we are allowed to idle, so do so without any timeout */
531			pthread_cond_wait (&reqwait, &reqlock);
532			ts.tv_sec = time (0) + IDLE_TIMEOUT;
533			}
534
535			--idle;
536			}
537
538			--nready;
539
540			UNLOCK (reqlock);
541
542			errno = 0; /* strictly unnecessary */
543
544			switch (req->type)
545			{
546			case REQ_QUIT:
547			goto quit;
548
549			default:
550			//req->result = ENOSYS;
551			break;
552			}
553
554			//req->errorno = errno;
555
556			LOCK (reslock);
557
558			++npending;
559
560			if (!reqq_push (&res_queue, req))
561			/* write a dummy byte to the pipe so fh becomes ready */
562			write (respipe [1], &respipe, 1);
563
564			self->req = 0;
565			worker_clear (self);
566
567			UNLOCK (reslock);
568			}
569
570			quit:
571			LOCK (wrklock);
572			worker_free (self);
573			UNLOCK (wrklock);
574
575			return 0;
576			}
577
578			/*****************************************************************************/
579
580			static void atfork_prepare (void)
581			{
582			LOCK (wrklock);
583			LOCK (reqlock);
584			LOCK (reslock);
585			}
586
587			static void atfork_parent (void)
588			{
589			UNLOCK (reslock);
590			UNLOCK (reqlock);
591			UNLOCK (wrklock);
592			}
593
594			static void atfork_child (void)
595			{
596			aio_req prv;
597
598			while (prv = reqq_shift (&req_queue))
599			req_free (prv);
600
601			while (prv = reqq_shift (&res_queue))
602			req_free (prv);
603
604			while (wrk_first.next != &wrk_first)
605			{
606			worker *wrk = wrk_first.next;
607
608			if (wrk->req)
609			req_free (wrk->req);
610
611			worker_clear (wrk);
612			worker_free (wrk);
613			}
614
615			started = 0;
616			idle = 0;
617			nreqs = 0;
618			nready = 0;
619			npending = 0;
620
621			close (respipe [0]);
622			close (respipe [1]);
623			create_pipe ();
624
625			atfork_parent ();
626			}
627
628	root	1.2	#define dREQ(reqtype) \
629	root	1.1	aio_req req; \
630			int req_pri = next_pri; \
631			next_pri = DEFAULT_PRI + PRI_BIAS; \
632			\
633			if (SvOK (callback) && !SvROK (callback)) \
634			croak ("callback must be undef or of reference type"); \
635			\
636	root	1.2	Newz (0, req, 1, aio_cb); \
637	root	1.1	if (!req) \
638			croak ("out of memory during aio_req allocation"); \
639			\
640			req->callback = newSVsv (callback); \
641	root	1.2	req->type = (reqtype); \
642	root	1.1	req->pri = req_pri
643
644			#define REQ_SEND \
645	root	1.2	req_send (req)
646
647			#define SvPTR(var, arg, type, class) \
648			if (!SvOK (arg)) \
649			(var) = 0; \
650			else if (sv_derived_from ((arg), # class)) \
651			{ \
652			IV tmp = SvIV ((SV*) SvRV (arg)); \
653			(var) = INT2PTR (type, tmp); \
654			} \
655			else \
656			Perl_croak (# var " is not of type " # type)
657	root	1.1
658	root	1.2	MODULE = BDB PACKAGE = BDB
659	root	1.1
660			PROTOTYPES: ENABLE
661
662			BOOT:
663			{
664	root	1.2	HV *stash = gv_stashpv ("BDB", 1);
665
666			static const struct {
667			const char *name;
668			IV iv;
669			} *civ, const_iv[] = {
670			#define const_iv(name) { # name, (IV)DB_ ## name },
671			const_iv (RPCCLIENT)
672			const_iv (INIT_CDB)
673			const_iv (INIT_LOCK)
674			const_iv (INIT_LOG)
675			const_iv (INIT_MPOOL)
676			const_iv (INIT_REP)
677			const_iv (INIT_TXN)
678			const_iv (RECOVER)
679			const_iv (INIT_TXN)
680			const_iv (RECOVER_FATAL)
681			const_iv (CREATE)
682			const_iv (USE_ENVIRON)
683			const_iv (USE_ENVIRON_ROOT)
684			const_iv (LOCKDOWN)
685			const_iv (PRIVATE)
686			const_iv (REGISTER)
687			const_iv (SYSTEM_MEM)
688			const_iv (AUTO_COMMIT)
689			const_iv (CDB_ALLDB)
690			const_iv (DIRECT_DB)
691			const_iv (DIRECT_LOG)
692			const_iv (DSYNC_DB)
693			const_iv (DSYNC_LOG)
694			const_iv (LOG_AUTOREMOVE)
695			const_iv (LOG_INMEMORY)
696			const_iv (NOLOCKING)
697			const_iv (MULTIVERSION)
698			const_iv (NOMMAP)
699			const_iv (NOPANIC)
700			const_iv (OVERWRITE)
701			const_iv (PANIC_ENVIRONMENT)
702			const_iv (REGION_INIT)
703			const_iv (TIME_NOTGRANTED)
704			const_iv (TXN_NOSYNC)
705			const_iv (TXN_SNAPSHOT)
706			const_iv (TXN_WRITE_NOSYNC)
707			const_iv (YIELDCPU)
708			const_iv (ENCRYPT_AES)
709			const_iv (XA_CREATE)
710			const_iv (BTREE)
711			const_iv (HASH)
712			const_iv (QUEUE)
713			const_iv (RECNO)
714			const_iv (UNKNOWN)
715			const_iv (EXCL)
716			const_iv (READ_UNCOMMITTED)
717			const_iv (TRUNCATE)
718			const_iv (NOSYNC)
719			const_iv (CHKSUM)
720			const_iv (ENCRYPT)
721			const_iv (TXN_NOT_DURABLE)
722			const_iv (DUP)
723			const_iv (DUPSORT)
724			const_iv (RECNUM)
725			const_iv (RENUMBER)
726			const_iv (REVSPLITOFF)
727			const_iv (INORDER)
728			const_iv (CONSUME)
729			const_iv (CONSUME_WAIT)
730			const_iv (SNAPSHOT)
731			const_iv (JOIN_ITEM)
732			const_iv (RMW)
733
734			const_iv (NOTFOUND)
735			const_iv (KEYEMPTY)
736			const_iv (LOCK_DEADLOCK)
737			const_iv (LOCK_NOTGRANTED)
738			const_iv (RUNRECOVERY)
739			};
740
741			for (civ = const_iv + sizeof (const_iv) / sizeof (const_iv [0]); civ-- > const_iv; )
742			newCONSTSUB (stash, (char *)civ->name, newSViv (civ->iv));
743	root	1.1
744			create_pipe ();
745			pthread_atfork (atfork_prepare, atfork_parent, atfork_child);
746			}
747
748			void
749			max_poll_reqs (int nreqs)
750			PROTOTYPE: $
751			CODE:
752			max_poll_reqs = nreqs;
753
754			void
755			max_poll_time (double nseconds)
756			PROTOTYPE: $
757			CODE:
758			max_poll_time = nseconds * AIO_TICKS;
759
760			void
761			min_parallel (int nthreads)
762			PROTOTYPE: $
763
764			void
765			max_parallel (int nthreads)
766			PROTOTYPE: $
767
768			void
769			max_idle (int nthreads)
770			PROTOTYPE: $
771			CODE:
772			set_max_idle (nthreads);
773
774			int
775			max_outstanding (int maxreqs)
776			PROTOTYPE: $
777			CODE:
778			RETVAL = max_outstanding;
779			max_outstanding = maxreqs;
780			OUTPUT:
781			RETVAL
782
783			int
784			bdbreq_pri (int pri = 0)
785			PROTOTYPE: ;$
786			CODE:
787			RETVAL = next_pri - PRI_BIAS;
788			if (items > 0)
789			{
790			if (pri < PRI_MIN) pri = PRI_MIN;
791			if (pri > PRI_MAX) pri = PRI_MAX;
792			next_pri = pri + PRI_BIAS;
793			}
794			OUTPUT:
795			RETVAL
796
797			void
798			bdbreq_nice (int nice = 0)
799			CODE:
800			nice = next_pri - nice;
801			if (nice < PRI_MIN) nice = PRI_MIN;
802			if (nice > PRI_MAX) nice = PRI_MAX;
803			next_pri = nice + PRI_BIAS;
804
805			void
806			flush ()
807			PROTOTYPE:
808			CODE:
809			while (nreqs)
810			{
811			poll_wait ();
812			poll_cb ();
813			}
814
815			int
816			poll()
817			PROTOTYPE:
818			CODE:
819			poll_wait ();
820			RETVAL = poll_cb ();
821			OUTPUT:
822			RETVAL
823
824			int
825			poll_fileno()
826			PROTOTYPE:
827			CODE:
828			RETVAL = respipe [0];
829			OUTPUT:
830			RETVAL
831
832			int
833			poll_cb(...)
834			PROTOTYPE:
835			CODE:
836			RETVAL = poll_cb ();
837			OUTPUT:
838			RETVAL
839
840			void
841			poll_wait()
842			PROTOTYPE:
843			CODE:
844			poll_wait ();
845
846			int
847			nreqs()
848			PROTOTYPE:
849			CODE:
850			RETVAL = nreqs;
851			OUTPUT:
852			RETVAL
853
854			int
855			nready()
856			PROTOTYPE:
857			CODE:
858			RETVAL = get_nready ();
859			OUTPUT:
860			RETVAL
861
862			int
863			npending()
864			PROTOTYPE:
865			CODE:
866			RETVAL = get_npending ();
867			OUTPUT:
868			RETVAL
869
870			int
871			nthreads()
872			PROTOTYPE:
873			CODE:
874			if (WORDACCESS_UNSAFE) LOCK (wrklock);
875			RETVAL = started;
876			if (WORDACCESS_UNSAFE) UNLOCK (wrklock);
877			OUTPUT:
878			RETVAL
879
880	root	1.2	DB_ENV *
881			bdb_env_create (U32 env_flags = 0)
882			CODE:
883			{
884			int err = db_env_create (&RETVAL, env_flags);
885			if (err)
886			croak ("db_env_create: %s", db_strerror (err));
887			}
888
889			void
890			bdb_env_open (DB_ENV env, char db_home, U32 open_flags, int mode, SV *callback = 0)
891			CODE:
892			{
893			dREQ (REQ_ENV_OPEN);
894			req->env = env;
895			req->uint1 = open_flags;
896			req->int1 = mode;
897			req->buf1 = strdup (db_home);
898			REQ_SEND;
899			}
900
901			void
902			bdb_env_close (DB_ENV env, U32 flags = 0, SV callback = 0)
903			CODE:
904			{
905			dREQ (REQ_ENV_CLOSE);
906			req->env = env;
907			req->uint1 = flags;
908			REQ_SEND;
909			}
910
911			DB *
912			bdb_db_create (DB_ENV *env = 0, U32 flags = 0)
913			CODE:
914			{
915			int err = db_create (&RETVAL, env, flags);
916			if (err)
917			croak ("db_env_create: %s", db_strerror (err));
918			}
919
920			void
921			bdb_db_open (DB db, DB_TXN txnid, const char file, const char database, int type, U32 flags, int mode, SV *callback = 0)
922			CODE:
923			{
924			dREQ (REQ_DB_OPEN);
925			req->db = db;
926			req->txn = txnid;
927			req->buf1 = strdup (file);
928			req->buf2 = strdup (database);
929			req->int1 = type;
930			req->uint1 = flags;
931			req->int2 = mode;
932			REQ_SEND;
933			}
934
935			void
936			bdb_db_close (DB db, U32 flags = 0, SV callback = 0)
937			CODE:
938			{
939			dREQ (REQ_DB_CLOSE);
940			req->db = db;
941			req->uint1 = flags;
942			REQ_SEND;
943			}
944
945
946			MODULE = BDB PACKAGE = BDB::Env
947
948			int set_cachesize (DB_ENV *env, U32 gbytes, U32 bytes, int ncache = 0)
949
950			int set_flags (DB_ENV *env, U32 flags, int onoff)
951
952			int set_encrypt (DB_ENV env, const char password, U32 flags)
953
954			MODULE = BDB PACKAGE = BDB::Db
955
956			int set_cachesize (DB *db, U32 gbytes, U32 bytes, int ncache = 0)
957
958			int set_flags (DB *env, U32 flags, int onoff)
959
960			int set_encrypt (DB db, const char password, U32 flags)
961
962			int set_lorder (DB *db, int lorder)
963
964
965			int set_bt_minkey (DB *db, U32 minkey)
966
967			int set_re_delim(DB *db, int delim);
968
969			int set_re_pad (DB *db, int re_pad)
970
971			int set_re_source (DB db, char source)
972
973			int set_re_len (DB *db, U32 re_len)
974
975			int set_h_ffactor (DB *db, U32 h_ffactor)
976
977			int set_h_nelem (DB *db, U32 h_nelem)
978
979			int set_q_extentsize (DB *db, U32 extentsize)
980
981	root	1.1