--- BDB/BDB.xs 2007/07/08 09:16:19 1.12 +++ BDB/BDB.xs 2007/07/08 11:12:12 1.13 @@ -6,19 +6,31 @@ #include "perl.h" #include "XSUB.h" +// perl stupidly defines these as macros, breaking +// lots and lots of code. +#undef open +#undef close +#undef abort +#undef malloc +#undef free +#undef send + #include #include #include -#include #include #include -#include #include +#ifndef _WIN32 +# include +# include +#endif + #include -#if DB_VERSION_MAJOR < 4 || (DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR < 5) -# error you need Berkeley DB 4.5 or newer installed +#if DB_VERSION_MAJOR < 4 || (DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR < 4) +# error you need Berkeley DB 4.4 or newer installed #endif /* number of seconds after which idle threads exit */ @@ -35,13 +47,13 @@ static SV *prepare_cb; -static inline char * +static char * strdup_ornull (const char *s) { return s ? strdup (s) : 0; } -static inline void +static void sv_to_dbt (DBT *dbt, SV *sv) { STRLEN len; @@ -53,7 +65,7 @@ dbt->flags = DB_DBT_REALLOC; } -static inline void +static void dbt_to_sv (SV *sv, DBT *dbt) { if (sv) @@ -159,7 +171,7 @@ 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 int respipe [2], respipe_osf [2]; static mutex_t reslock = X_MUTEX_INIT; static mutex_t reqlock = X_MUTEX_INIT; @@ -340,7 +352,32 @@ Safefree (req); } -static void *aio_proc (void *arg); +#ifdef USE_SOCKETS_AS_HANDLES +# define TO_SOCKET(x) (win32_get_osfhandle (x)) +#else +# define TO_SOCKET(x) (x) +#endif + +static void +create_pipe (int fd[2]) +{ +#ifdef _WIN32 + int arg = 1; + if (PerlSock_socketpair (AF_UNIX, SOCK_STREAM, 0, fd) + || ioctlsocket (TO_SOCKET (fd [0]), FIONBIO, &arg) + || ioctlsocket (TO_SOCKET (fd [1]), FIONBIO, &arg)) +#else + if (pipe (fd) + || fcntl (fd [0], F_SETFL, O_NONBLOCK) + || fcntl (fd [1], F_SETFL, O_NONBLOCK)) +#endif + croak ("unable to initialize result pipe"); + + respipe_osf [0] = TO_SOCKET (respipe [0]); + respipe_osf [1] = TO_SOCKET (respipe [1]); +} + +X_THREAD_PROC (bdb_proc); static void start_thread (void) { @@ -350,7 +387,7 @@ croak ("unable to allocate worker thread data"); X_LOCK (wrklock); - if (thread_create (&wrk->tid, aio_proc, (void *)wrk)) + if (thread_create (&wrk->tid, bdb_proc, (void *)wrk)) { wrk->prev = &wrk_first; wrk->next = wrk_first.next; @@ -383,10 +420,12 @@ // synthesize callback if none given if (!SvOK (req->callback)) { + int count; + dSP; PUSHMARK (SP); PUTBACK; - int count = call_sv (prepare_cb, G_ARRAY); + count = call_sv (prepare_cb, G_ARRAY); SPAGAIN; if (count != 2) @@ -474,10 +513,10 @@ maybe_start_thread (); - FD_ZERO(&rfd); - FD_SET(respipe [0], &rfd); + FD_ZERO (&rfd); + FD_SET (respipe [0], &rfd); - select (respipe [0] + 1, &rfd, 0, 0, 0); + PerlSock_select (respipe [0] + 1, &rfd, 0, 0, 0); } } @@ -510,7 +549,7 @@ { /* read any signals sent by the worker threads */ char buf [4]; - while (read (respipe [0], buf, 4) == 4) + while (respipe_read (respipe [0], buf, 4) == 4) ; } } @@ -557,15 +596,14 @@ /*****************************************************************************/ -static void *aio_proc (void *thr_arg) +X_THREAD_PROC (bdb_proc) { 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); + ts.tv_nsec = ((unsigned long)self & 1023UL) * (1000000000UL / 1024UL); for (;;) { @@ -735,7 +773,7 @@ if (!reqq_push (&res_queue, req)) /* write a dummy byte to the pipe so fh becomes ready */ - write (respipe [1], &respipe, 1); + respipe_write (respipe_osf [1], (const void *)&respipe_osf, 1); self->req = 0; worker_clear (self); @@ -794,11 +832,10 @@ nready = 0; npending = 0; - close (respipe [0]); - close (respipe [1]); + respipe_close (respipe [0]); + respipe_close (respipe [1]); - if (!create_pipe (respipe)) - croak ("unable to initialize result pipe"); + create_pipe (respipe); atfork_parent (); } @@ -887,7 +924,6 @@ const_iv (LOG_AUTOREMOVE) const_iv (LOG_INMEMORY) const_iv (NOLOCKING) - const_iv (MULTIVERSION) const_iv (NOMMAP) const_iv (NOPANIC) const_iv (OVERWRITE) @@ -895,7 +931,6 @@ const_iv (REGION_INIT) const_iv (TIME_NOTGRANTED) const_iv (TXN_NOSYNC) - const_iv (TXN_SNAPSHOT) const_iv (TXN_WRITE_NOSYNC) const_iv (WRITECURSOR) const_iv (YIELDCPU) @@ -948,7 +983,6 @@ const_iv (NOOVERWRITE) const_iv (TXN_NOWAIT) - const_iv (TXN_SNAPSHOT) const_iv (TXN_SYNC) const_iv (SET_LOCK_TIMEOUT) @@ -986,13 +1020,16 @@ const_iv (SEQ_DEC) const_iv (SEQ_INC) const_iv (SEQ_WRAP) +#if DB_VERSION_MINOR >= 5 + const_iv (MULTIVERSION) + const_iv (TXN_SNAPSHOT) +#endif }; for (civ = const_iv + sizeof (const_iv) / sizeof (const_iv [0]); civ-- > const_iv; ) newCONSTSUB (stash, (char *)civ->name, newSViv (civ->iv)); - if (!create_pipe (respipe)) - croak ("unable to initialize result pipe"); + create_pipe (respipe); X_THREAD_ATFORK (atfork_prepare, atfork_parent, atfork_child); #ifdef _WIN32 @@ -1159,9 +1196,10 @@ db_env_open (DB_ENV *env, octetstring db_home, U32 open_flags, int mode, SV *callback = &PL_sv_undef) CODE: { + dREQ (REQ_ENV_OPEN); + env->set_thread_count (env, get_nthreads ()); - dREQ (REQ_ENV_OPEN); req->env = env; req->uint1 = open_flags | DB_THREAD; req->int1 = mode;