[ViewVC] Diff of: cvs/IO-AIO/AIO.xs

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.95 by root, Sun Nov 26 22:47:58 2006 UTC vs.
Revision 1.106 by root, Mon Sep 24 18:14:00 2007 UTC

-/* solaris */
+#include "xthread.h"
-#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 "autoconf/config.h"
-#include <pthread.h>
 #include <stddef.h>
 #include <stdlib.h>
 #include <errno.h>
-#include <sys/time.h>
-#include <sys/select.h>
 #include <sys/types.h>
 #include <sys/stat.h>
 #include <limits.h>
-#include <unistd.h>
 #include <fcntl.h>
-#include <signal.h>
 #include <sched.h>
+#ifdef _WIN32
+# define SIGIO 0
+  typedef Direntry_t X_DIRENT;
+#undef malloc
+#undef free
+// perl overrides all those nice win32 functions
+# undef open
+# undef read
+# undef write
+# undef send
+# undef recv
+# undef stat
+# undef fstat
+# define lstat stat
+# undef truncate
+# undef ftruncate
+# undef open
+# undef close
+# undef unlink
+# undef rmdir
+# undef rename
+# undef lseek
+# define chown(a,b,c)    (errno = ENOSYS, -1)
+# define fchown(a,b,c)   (errno = ENOSYS, -1)
+# define fchmod(a,b)     (errno = ENOSYS, -1)
+# define symlink(a,b)    (errno = ENOSYS, -1)
+# define readlink(a,b,c) (errno = ENOSYS, -1)
+# define mknod(a,b,c)    (errno = ENOSYS, -1)
+# define truncate(a,b)   (errno = ENOSYS, -1)
+# define ftruncate(fd,o) chsize ((fd), (o))
+# define fsync(fd)       _commit (fd)
+# define opendir(fd)     (errno = ENOSYS, 0)
+# define readdir(fd)     (errno = ENOSYS, -1)
+# define closedir(fd)    (errno = ENOSYS, -1)
+# define mkdir(a,b)      mkdir (a)
+#else
+# include "autoconf/config.h"
+# include <sys/time.h>
+# include <sys/select.h>
+# include <unistd.h>
+# include <utime.h>
+# include <signal.h>
+  typedef struct dirent X_DIRENT;
+#endif
 #if HAVE_SENDFILE
 # if __linux
 #  include <sys/sendfile.h>
 # elif __freebsd
 /* used for struct dirent, AIX doesn't provide it */
 #ifndef NAME_MAX
 # define NAME_MAX 4096
 #endif
-#ifndef PTHREAD_STACK_MIN
-/* care for broken platforms, e.g. windows */
-# define PTHREAD_STACK_MIN 16384
-#endif
-#if __ia64
-# define STACKSIZE 65536
-#elif __i386 || __x86_64 /* 16k is unreasonably high :( */
-# define STACKSIZE PTHREAD_STACK_MIN
-#else
-# define STACKSIZE 16384
-#endif
-/* 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
 /* buffer size for various temporary buffers */
 #define AIO_BUFSIZE 65536
+/* use NV for 32 bit perls as it allows larger offsets */
+#if IVSIZE >= 8
+# define SvVAL64 SvIV
+#else
+# define SvVAL64 SvNV
+#endif
 #define dBUF	 				\
   char *aio_buf;				\
-  LOCK (wrklock);				\
+  X_LOCK (wrklock);				\
   self->dbuf = aio_buf = malloc (AIO_BUFSIZE);	\
-  UNLOCK (wrklock);				\
+  X_UNLOCK (wrklock);				\
   if (!aio_buf)					\
     return -1;
 typedef SV SV8; /* byte-sv, used for argument-checking */
 enum {
   REQ_QUIT,
   REQ_OPEN, REQ_CLOSE,
-  REQ_READ, REQ_WRITE, REQ_READAHEAD,
+  REQ_READ, REQ_WRITE,
-  REQ_SENDFILE,
+  REQ_READAHEAD, REQ_SENDFILE,
   REQ_STAT, REQ_LSTAT, REQ_FSTAT,
+  REQ_TRUNCATE, REQ_FTRUNCATE,
+  REQ_UTIME, REQ_FUTIME,
+  REQ_CHMOD, REQ_FCHMOD,
+  REQ_CHOWN, REQ_FCHOWN,
   REQ_FSYNC, REQ_FDATASYNC,
-  REQ_UNLINK, REQ_RMDIR, REQ_RENAME,
+  REQ_UNLINK, REQ_RMDIR, REQ_MKDIR, REQ_RENAME,
   REQ_MKNOD, REQ_READDIR,
   REQ_LINK, REQ_SYMLINK, REQ_READLINK,
   REQ_GROUP, REQ_NOP,
   REQ_BUSY,
 };
 typedef struct aio_cb
 {
   struct aio_cb *volatile next;
-  SV *callback, *fh;
+  SV *callback;
   SV *sv1, *sv2;
   void *ptr1, *ptr2;
   off_t offs;
   size_t size;
   ssize_t result;
+  double nv1, nv2;
   STRLEN stroffset;
   int type;
-  int int1, int2;
+  int int1, int2, int3;
   int errorno;
   mode_t mode; /* open */
   unsigned char flags;
   unsigned char pri;
   struct aio_cb *grp, *grp_prev, *grp_next, *grp_first;
 } aio_cb;
 enum {
   FLAG_CANCELLED     = 0x01, /* request was cancelled */
-  FLAG_SV1_RO_OFF    = 0x40, /* data was set readonly */
+  FLAG_SV2_RO_OFF    = 0x40, /* data was set readonly */
   FLAG_PTR2_FREE     = 0x80, /* need to free(ptr2) */
 };
 typedef aio_cb *aio_req;
 typedef aio_cb *aio_req_ornot;
 {
   return  (tv2->tv_sec  - tv1->tv_sec ) * AIO_TICKS
        + ((tv2->tv_usec - tv1->tv_usec) >> 10);
 }
-static pthread_t main_tid;
+static thread_t main_tid;
 static int main_sig;
 static int block_sig_level;
 void block_sig ()
 {
 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;
+static mutex_t wrklock = X_MUTEX_INIT;
 typedef struct worker {
   /* locked by wrklock */
   struct worker *prev, *next;
-  pthread_t tid;
+  thread_t tid;
   /* locked by reslock, reqlock or wrklock */
   aio_req req; /* currently processed request */
   void *dbuf;
   DIR *dirp;
 }
 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_osf [2], respipe [2] = { -1, -1 };
-static pthread_mutex_t reslock = AIO_MUTEX_INIT;
+static mutex_t reslock = X_MUTEX_INIT;
-static pthread_mutex_t reqlock = AIO_MUTEX_INIT;
+static mutex_t reqlock = X_MUTEX_INIT;
-static pthread_cond_t  reqwait = PTHREAD_COND_INITIALIZER;
+static cond_t  reqwait = X_COND_INIT;
 #if WORDACCESS_UNSAFE
 static unsigned int get_nready ()
 {
   unsigned int retval;
-  LOCK   (reqlock);
+  X_LOCK   (reqlock);
   retval = nready;
-  UNLOCK (reqlock);
+  X_UNLOCK (reqlock);
   return retval;
 }
 static unsigned int get_npending ()
 {
   unsigned int retval;
-  LOCK   (reslock);
+  X_LOCK   (reslock);
   retval = npending;
-  UNLOCK (reslock);
+  X_UNLOCK (reslock);
   return retval;
 }
 static unsigned int get_nthreads ()
 {
   unsigned int retval;
-  LOCK   (wrklock);
+  X_LOCK   (wrklock);
   retval = started;
-  UNLOCK (wrklock);
+  X_UNLOCK (wrklock);
   return retval;
 }
 #else
   abort ();
 }
 static int poll_cb ();
 static int req_invoke (aio_req req);
-static void req_free (aio_req req);
+static void req_destroy (aio_req req);
 static void req_cancel (aio_req req);
 /* must be called at most once */
 static SV *req_sv (aio_req req, const char *klass)
 {
     {
       block_sig ();
       if (!req_invoke (grp))
         {
-          req_free (grp);
+          req_destroy (grp);
           unblock_sig ();
           croak (0);
         }
-      req_free (grp);
+      req_destroy (grp);
       unblock_sig ();
     }
 }
 static int req_invoke (aio_req req)
 {
   dSP;
-  if (req->flags & FLAG_SV1_RO_OFF)
+  if (req->flags & FLAG_SV2_RO_OFF)
-    SvREADONLY_off (req->sv1);
+    SvREADONLY_off (req->sv2);
   if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback))
     {
       ENTER;
       SAVETMPS;
             break;
           case REQ_READLINK:
             if (req->result > 0)
               {
-                SvCUR_set (req->sv1, req->result);
+                SvCUR_set (req->sv2, req->result);
-                *SvEND (req->sv1) = 0;
+                *SvEND (req->sv2) = 0;
-                PUSHs (req->sv1);
+                PUSHs (req->sv2);
               }
             break;
           case REQ_STAT:
           case REQ_LSTAT:
             PL_statcache   = *(Stat_t *)(req->ptr2);
             PUSHs (sv_2mortal (newSViv (req->result)));
             break;
           case REQ_READ:
-            SvCUR_set (req->sv1, req->stroffset + (req->result > 0 ? req->result : 0));
+            SvCUR_set (req->sv2, req->stroffset + (req->result > 0 ? req->result : 0));
-            *SvEND (req->sv1) = 0;
+            *SvEND (req->sv2) = 0;
             PUSHs (sv_2mortal (newSViv (req->result)));
             break;
           default:
             PUSHs (sv_2mortal (newSViv (req->result)));
     }
   return !SvTRUE (ERRSV);
 }
-static void req_free (aio_req req)
+static void req_destroy (aio_req req)
 {
   if (req->self)
     {
       sv_unmagic (req->self, PERL_MAGIC_ext);
       SvREFCNT_dec (req->self);
     }
-  SvREFCNT_dec (req->fh);
   SvREFCNT_dec (req->sv1);
   SvREFCNT_dec (req->sv2);
   SvREFCNT_dec (req->callback);
   if (req->flags & FLAG_PTR2_FREE)
   req->flags |= FLAG_CANCELLED;
   req_cancel_subs (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_respipe ()
+{
+  int old_readfd = respipe [0];
+  if (respipe [1] >= 0)
+    respipe_close (TO_SOCKET (respipe [1]));
+#ifdef _WIN32
+  if (PerlSock_socketpair (AF_UNIX, SOCK_STREAM, 0, respipe))
+#else
+  if (pipe (respipe))
+#endif
+    croak ("unable to initialize result pipe");
+  if (old_readfd >= 0)
+    {
+      if (dup2 (TO_SOCKET (respipe [0]), TO_SOCKET (old_readfd)) < 0)
+        croak ("unable to initialize result pipe(2)");
+      respipe_close (respipe [0]);
+      respipe [0] = old_readfd;
+    }
+#ifdef _WIN32
+  int arg = 1;
+  if (ioctlsocket (TO_SOCKET (respipe [0]), FIONBIO, &arg)
+      || ioctlsocket (TO_SOCKET (respipe [1]), FIONBIO, &arg))
+#else
+  if (fcntl (respipe [0], F_SETFL, O_NONBLOCK)
+      || fcntl (respipe [1], F_SETFL, O_NONBLOCK))
+#endif
+    croak ("unable to initialize result pipe(3)");
+  respipe_osf [0] = TO_SOCKET (respipe [0]);
+  respipe_osf [1] = TO_SOCKET (respipe [1]);
+}
+X_THREAD_PROC (aio_proc);
 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_setstacksize (&attr, STACKSIZE);
-  pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
-#ifdef PTHREAD_SCOPE_PROCESS
-  pthread_attr_setscope (&attr, PTHREAD_SCOPE_PROCESS);
-#endif
-  sigfillset (&fullsigset);
-  LOCK (wrklock);
+  X_LOCK (wrklock);
-  pthread_sigmask (SIG_SETMASK, &fullsigset, &oldsigset);
-  if (pthread_create (&wrk->tid, &attr, aio_proc, (void *)wrk) == 0)
+  if (thread_create (&wrk->tid, aio_proc, (void *)wrk))
     {
       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);
+  X_UNLOCK (wrklock);
 }
 static void maybe_start_thread ()
 {
   if (get_nthreads () >= wanted)
 {
   block_sig ();
   ++nreqs;
-  LOCK (reqlock);
+  X_LOCK (reqlock);
   ++nready;
   reqq_push (&req_queue, req);
-  pthread_cond_signal (&reqwait);
+  X_COND_SIGNAL (reqwait);
-  UNLOCK (reqlock);
+  X_UNLOCK (reqlock);
   unblock_sig ();
   maybe_start_thread ();
 }
   Newz (0, req, 1, aio_cb);
   req->type = REQ_QUIT;
   req->pri  = PRI_MAX + PRI_BIAS;
-  LOCK (reqlock);
+  X_LOCK (reqlock);
   reqq_push (&req_queue, req);
-  pthread_cond_signal (&reqwait);
+  X_COND_SIGNAL (reqwait);
-  UNLOCK (reqlock);
+  X_UNLOCK (reqlock);
-  LOCK (wrklock);
+  X_LOCK (wrklock);
   --started;
-  UNLOCK (wrklock);
+  X_UNLOCK (wrklock);
 }
 static void set_max_idle (int nthreads)
 {
-  if (WORDACCESS_UNSAFE) LOCK   (reqlock);
+  if (WORDACCESS_UNSAFE) X_LOCK   (reqlock);
   max_idle = nthreads <= 0 ? 1 : nthreads;
-  if (WORDACCESS_UNSAFE) UNLOCK (reqlock);
+  if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
 }
 static void min_parallel (int nthreads)
 {
   if (wanted < nthreads)
   fd_set rfd;
   while (nreqs)
     {
       int size;
-      if (WORDACCESS_UNSAFE) LOCK   (reslock);
+      if (WORDACCESS_UNSAFE) X_LOCK   (reslock);
       size = res_queue.size;
-      if (WORDACCESS_UNSAFE) UNLOCK (reslock);
+      if (WORDACCESS_UNSAFE) X_UNLOCK (reslock);
       if (size)
         return;
       maybe_start_thread ();
-      FD_ZERO(&rfd);
+      FD_ZERO (&rfd);
-      FD_SET(respipe [0], &rfd);
+      FD_SET (respipe [0], &rfd);
-      select (respipe [0] + 1, &rfd, 0, 0, 0);
+      PerlSock_select (respipe [0] + 1, &rfd, 0, 0, 0);
     }
 }
 static int poll_cb ()
 {
     {
       for (;;)
         {
           maybe_start_thread ();
-          LOCK (reslock);
+          X_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)
+                  while (respipe_read (respipe [0], buf, 4) == 4)
                     ;
                 }
             }
-          UNLOCK (reslock);
+          X_UNLOCK (reslock);
           if (!req)
             break;
           --nreqs;
             }
           else
             {
               if (!req_invoke (req))
                 {
-                  req_free (req);
+                  req_destroy (req);
                   unblock_sig ();
                   croak (0);
                 }
               count++;
             }
-          req_free (req);
+          req_destroy (req);
           if (maxreqs && !--maxreqs)
             break;
           if (max_poll_time)
       ++maxreqs;
     }
   unblock_sig ();
   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");
 }
 /*****************************************************************************/
 /* work around various missing functions */
 /*
  * make our pread/pwrite safe against themselves, but not against
  * normal read/write by using a mutex. slows down execution a lot,
  * but that's your problem, not mine.
  */
-static pthread_mutex_t preadwritelock = PTHREAD_MUTEX_INITIALIZER;
+static mutex_t preadwritelock = X_MUTEX_INIT;
 static ssize_t pread (int fd, void *buf, size_t count, off_t offset)
 {
   ssize_t res;
   off_t ooffset;
-  LOCK (preadwritelock);
+  X_LOCK (preadwritelock);
   ooffset = lseek (fd, 0, SEEK_CUR);
   lseek (fd, offset, SEEK_SET);
   res = read (fd, buf, count);
   lseek (fd, ooffset, SEEK_SET);
-  UNLOCK (preadwritelock);
+  X_UNLOCK (preadwritelock);
   return res;
 }
 static ssize_t pwrite (int fd, void *buf, size_t count, off_t offset)
 {
   ssize_t res;
   off_t ooffset;
-  LOCK (preadwritelock);
+  X_LOCK (preadwritelock);
   ooffset = lseek (fd, 0, SEEK_CUR);
   lseek (fd, offset, SEEK_SET);
   res = write (fd, buf, count);
   lseek (fd, offset, SEEK_SET);
-  UNLOCK (preadwritelock);
+  X_UNLOCK (preadwritelock);
   return res;
 }
+#endif
+#ifndef HAVE_FUTIMES
+# define utimes(path,times)  aio_utimes  (path, times)
+# define futimes(fd,times)   aio_futimes (fd, times)
+int aio_utimes (const char *filename, const struct timeval times[2])
+{
+  if (times)
+    {
+      struct utimbuf buf;
+      buf.actime  = times[0].tv_sec;
+      buf.modtime = times[1].tv_sec;
+      return utime (filename, &buf);
+    }
+  else
+    return utime (filename, 0);
+}
+int aio_futimes (int fd, const struct timeval tv[2])
+{
+  errno = ENOSYS;
+  return -1;
+}
 #endif
 #if !HAVE_FDATASYNC
 # define fdatasync fsync
 #endif
 #if !HAVE_READAHEAD
 # define readahead(fd,offset,count) aio_readahead (fd, offset, count, self)
 static ssize_t aio_readahead (int fd, off_t offset, size_t count, worker *self)
 {
+  size_t todo = count;
   dBUF;
-  while (count > 0)
+  while (todo > 0)
     {
-      size_t len = count < AIO_BUFSIZE ? count : AIO_BUFSIZE;
+      size_t len = todo < AIO_BUFSIZE ? todo : AIO_BUFSIZE;
       pread (fd, aio_buf, len, offset);
       offset += len;
-      count  -= len;
+      todo   -= len;
     }
   errno = 0;
+  return count;
 }
 #endif
 #if !HAVE_READDIR_R
 # define readdir_r aio_readdir_r
-static pthread_mutex_t readdirlock = PTHREAD_MUTEX_INITIALIZER;
+static mutex_t readdirlock = X_MUTEX_INIT;
-static int readdir_r (DIR *dirp, struct dirent *ent, struct dirent **res)
+static int readdir_r (DIR *dirp, X_DIRENT *ent, X_DIRENT **res)
 {
-  struct dirent *e;
+  X_DIRENT *e;
   int errorno;
-  LOCK (readdirlock);
+  X_LOCK (readdirlock);
   e = readdir (dirp);
   errorno = errno;
   if (e)
       strcpy (ent->d_name, e->d_name);
     }
   else
     *res = 0;
-  UNLOCK (readdirlock);
+  X_UNLOCK (readdirlock);
   errno = errorno;
   return e ? 0 : -1;
 }
 #endif
 static void scandir_ (aio_req req, worker *self)
 {
   DIR *dirp;
   union
   {
-    struct dirent d;
+    X_DIRENT d;
-    char b [offsetof (struct dirent, d_name) + NAME_MAX + 1];
+    char b [offsetof (X_DIRENT, d_name) + NAME_MAX + 1];
   } *u;
-  struct dirent *entp;
+  X_DIRENT *entp;
   char *name, *names;
   int memlen = 4096;
   int memofs = 0;
   int res = 0;
-  int errorno;
-  LOCK (wrklock);
+  X_LOCK (wrklock);
   self->dirp = dirp = opendir (req->ptr1);
   self->dbuf = u = malloc (sizeof (*u));
   req->flags |= FLAG_PTR2_FREE;
   req->ptr2 = names = malloc (memlen);
-  UNLOCK (wrklock);
+  X_UNLOCK (wrklock);
   if (dirp && u && names)
     for (;;)
       {
         errno = 0;
             res++;
             while (memofs + len > memlen)
               {
                 memlen *= 2;
-                LOCK (wrklock);
+                X_LOCK (wrklock);
                 req->ptr2 = names = realloc (names, memlen);
-                UNLOCK (wrklock);
+                X_UNLOCK (wrklock);
                 if (!names)
                   break;
               }
   req->result = res;
 }
 /*****************************************************************************/
-static void *aio_proc (void *thr_arg)
+X_THREAD_PROC (aio_proc)
 {
+    {//D
   aio_req req;
   struct timespec ts;
   worker *self = (worker *)thr_arg;
-  /* try to distribute timeouts somewhat evenly */
+  /* try to distribute timeouts somewhat randomly */
-  ts.tv_nsec = (((unsigned long)self + (unsigned long)ts.tv_sec) & 1023UL)
+  ts.tv_nsec = ((unsigned long)self & 1023UL) * (1000000000UL / 1024UL);
-               * (1000000000UL / 1024UL);
   for (;;)
     {
       ts.tv_sec  = time (0) + IDLE_TIMEOUT;
-      LOCK (reqlock);
+      X_LOCK (reqlock);
       for (;;)
         {
           self->req = req = reqq_shift (&req_queue);
           if (req)
             break;
           ++idle;
-          if (pthread_cond_timedwait (&reqwait, &reqlock, &ts)
+          if (X_COND_TIMEDWAIT (reqwait, reqlock, ts)
               == ETIMEDOUT)
             {
               if (idle > max_idle)
                 {
                   --idle;
-                  UNLOCK (reqlock);
+                  X_UNLOCK (reqlock);
-                  LOCK (wrklock);
+                  X_LOCK (wrklock);
                   --started;
-                  UNLOCK (wrklock);
+                  X_UNLOCK (wrklock);
                   goto quit;
                 }
               /* we are allowed to idle, so do so without any timeout */
-              pthread_cond_wait (&reqwait, &reqlock);
+              X_COND_WAIT (reqwait, reqlock);
               ts.tv_sec  = time (0) + IDLE_TIMEOUT;
             }
           --idle;
         }
       --nready;
-      UNLOCK (reqlock);
+      X_UNLOCK (reqlock);
       errno = 0; /* strictly unnecessary */
       if (!(req->flags & FLAG_CANCELLED))
         switch (req->type)
           {
-            case REQ_READ:      req->result = pread     (req->int1, req->ptr1, req->size, req->offs); break;
+            case REQ_READ:      req->result = req->offs >= 0
-            case REQ_WRITE:     req->result = pwrite    (req->int1, req->ptr1, req->size, req->offs); break;
+                                            ? pread     (req->int1, req->ptr1, req->size, req->offs)
+                                            : read      (req->int1, req->ptr1, req->size); break;
+            case REQ_WRITE:     req->result = req->offs >= 0
+                                            ? pwrite    (req->int1, req->ptr1, req->size, req->offs)
+                                            : write     (req->int1, req->ptr1, req->size); break;
             case REQ_READAHEAD: req->result = readahead (req->int1, req->offs, req->size); break;
             case REQ_SENDFILE:  req->result = sendfile_ (req->int1, req->int2, req->offs, req->size, self); break;
             case REQ_STAT:      req->result = stat      (req->ptr1, (Stat_t *)req->ptr2); break;
             case REQ_LSTAT:     req->result = lstat     (req->ptr1, (Stat_t *)req->ptr2); break;
             case REQ_FSTAT:     req->result = fstat     (req->int1, (Stat_t *)req->ptr2); break;
+            case REQ_CHOWN:     req->result = chown     (req->ptr1, req->int2, req->int3); break;
+            case REQ_FCHOWN:    req->result = fchown    (req->int1, req->int2, req->int3); break;
+            case REQ_CHMOD:     req->result = chmod     (req->ptr1, req->mode); break;
+            case REQ_FCHMOD:    req->result = fchmod    (req->int1, req->mode); break;
+            case REQ_TRUNCATE:  req->result = truncate  (req->ptr1, req->offs); break;
+            case REQ_FTRUNCATE: req->result = ftruncate (req->int1, req->offs); break;
             case REQ_OPEN:      req->result = open      (req->ptr1, req->int1, req->mode); break;
             case REQ_CLOSE:     req->result = close     (req->int1); break;
             case REQ_UNLINK:    req->result = unlink    (req->ptr1); break;
             case REQ_RMDIR:     req->result = rmdir     (req->ptr1); break;
+            case REQ_MKDIR:     req->result = mkdir     (req->ptr1, req->mode); break;
             case REQ_RENAME:    req->result = rename    (req->ptr2, req->ptr1); break;
             case REQ_LINK:      req->result = link      (req->ptr2, req->ptr1); break;
             case REQ_SYMLINK:   req->result = symlink   (req->ptr2, req->ptr1); break;
             case REQ_MKNOD:     req->result = mknod     (req->ptr2, req->mode, (dev_t)req->offs); break;
             case REQ_READLINK:  req->result = readlink  (req->ptr2, req->ptr1, NAME_MAX); break;
             case REQ_FDATASYNC: req->result = fdatasync (req->int1); break;
             case REQ_FSYNC:     req->result = fsync     (req->int1); break;
             case REQ_READDIR:   scandir_ (req, self); break;
             case REQ_BUSY:
+#ifdef _WIN32
+	      Sleep (req->nv1 * 1000.);
+#else
               {
                 struct timeval tv;
-                tv.tv_sec  = req->int1;
+                tv.tv_sec  = req->nv1;
-                tv.tv_usec = req->int2;
+                tv.tv_usec = (req->nv1 - tv.tv_sec) * 1000000.;
                 req->result = select (0, 0, 0, 0, &tv);
+              }
+#endif
+              break;
+            case REQ_UTIME:
+            case REQ_FUTIME:
+              {
+                struct timeval tv[2];
+                struct timeval *times;
+                if (req->nv1 != -1. || req->nv2 != -1.)
+                  {
+                    tv[0].tv_sec  = req->nv1;
+                    tv[0].tv_usec = (req->nv1 - tv[0].tv_sec) * 1000000.;
+                    tv[1].tv_sec  = req->nv2;
+                    tv[1].tv_usec = (req->nv2 - tv[1].tv_sec) * 1000000.;
+                    times = tv;
+                  }
+                else
+                  times = 0;
+                req->result = req->type == REQ_FUTIME
+                              ? futimes (req->int1, times)
+                              : utimes  (req->ptr1, times);
               }
             case REQ_GROUP:
             case REQ_NOP:
               break;
             case REQ_QUIT:
               goto quit;
             default:
-              req->result = ENOSYS;
+              req->result = -1;
               break;
           }
       req->errorno = errno;
-      LOCK (reslock);
+      X_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);
+          respipe_write (respipe_osf [1], (const void *)&respipe_osf, 1);
           /* optionally signal the main thread asynchronously */
           if (main_sig)
             pthread_kill (main_tid, main_sig);
         }
       self->req = 0;
       worker_clear (self);
-      UNLOCK (reslock);
+      X_UNLOCK (reslock);
     }
 quit:
-  LOCK (wrklock);
+  X_LOCK (wrklock);
   worker_free (self);
-  UNLOCK (wrklock);
+  X_UNLOCK (wrklock);
   return 0;
+    }//D
 }
 /*****************************************************************************/
 static void atfork_prepare (void)
 {
-  LOCK (wrklock);
+  X_LOCK (wrklock);
-  LOCK (reqlock);
+  X_LOCK (reqlock);
-  LOCK (reslock);
+  X_LOCK (reslock);
 #if !HAVE_PREADWRITE
-  LOCK (preadwritelock);
+  X_LOCK (preadwritelock);
 #endif
 #if !HAVE_READDIR_R
-  LOCK (readdirlock);
+  X_LOCK (readdirlock);
 #endif
 }
 static void atfork_parent (void)
 {
 #if !HAVE_READDIR_R
-  UNLOCK (readdirlock);
+  X_UNLOCK (readdirlock);
 #endif
 #if !HAVE_PREADWRITE
-  UNLOCK (preadwritelock);
+  X_UNLOCK (preadwritelock);
 #endif
-  UNLOCK (reslock);
+  X_UNLOCK (reslock);
-  UNLOCK (reqlock);
+  X_UNLOCK (reqlock);
-  UNLOCK (wrklock);
+  X_UNLOCK (wrklock);
 }
 static void atfork_child (void)
 {
   aio_req prv;
   while (prv = reqq_shift (&req_queue))
-    req_free (prv);
+    req_destroy (prv);
   while (prv = reqq_shift (&res_queue))
-    req_free (prv);
+    req_destroy (prv);
   while (wrk_first.next != &wrk_first)
     {
       worker *wrk = wrk_first.next;
       if (wrk->req)
-        req_free (wrk->req);
+        req_destroy (wrk->req);
       worker_clear (wrk);
       worker_free (wrk);
     }
   idle     = 0;
   nreqs    = 0;
   nready   = 0;
   npending = 0;
-  close (respipe [0]);
-  close (respipe [1]);
-  create_pipe ();
+  create_respipe ();
   atfork_parent ();
 }
 #define dREQ							\
         newCONSTSUB (stash, "EXDEV",    newSViv (EXDEV));
         newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY));
         newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY));
         newCONSTSUB (stash, "O_CREAT",  newSViv (O_CREAT));
         newCONSTSUB (stash, "O_TRUNC",  newSViv (O_TRUNC));
+#ifdef _WIN32
+        X_MUTEX_CHECK (wrklock);
+        X_MUTEX_CHECK (reslock);
+        X_MUTEX_CHECK (reqlock);
+        X_MUTEX_CHECK (reqwait);
+        X_MUTEX_CHECK (preadwritelock);
+        X_MUTEX_CHECK (readdirlock);
+	X_COND_CHECK  (reqwait);
+#else
         newCONSTSUB (stash, "S_IFIFO",  newSViv (S_IFIFO));
         newCONSTSUB (stash, "SIGIO",    newSViv (SIGIO));
+#endif
-	create_pipe ();
+        create_respipe ();
-        pthread_atfork (atfork_prepare, atfork_parent, atfork_child);
+        X_THREAD_ATFORK (atfork_prepare, atfork_parent, atfork_child);
 }
 void
 max_poll_reqs (int nreqs)
 	PROTOTYPE: $
         max_outstanding = maxreqs;
 	OUTPUT:
         RETVAL
 void
-aio_open (pathname,flags,mode,callback=&PL_sv_undef)
+aio_open (SV8 *pathname, int flags, int mode, SV *callback=&PL_sv_undef)
-	SV8 *	pathname
-        int	flags
-        int	mode
-        SV *	callback
 	PROTOTYPE: $$$;$
 	PPCODE:
 {
         dREQ;
         REQ_SEND;
 }
 void
-aio_close (fh,callback=&PL_sv_undef)
+aio_close (SV *fh, SV *callback=&PL_sv_undef)
-	SV *	fh
-        SV *	callback
 	PROTOTYPE: $;$
         ALIAS:
            aio_close     = REQ_CLOSE
            aio_fsync     = REQ_FSYNC
            aio_fdatasync = REQ_FDATASYNC
 	PPCODE:
 {
         dREQ;
         req->type = ix;
-        req->fh   = newSVsv (fh);
+        req->sv1  = newSVsv (fh);
         req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh)));
         REQ_SEND (req);
 }
 void
-aio_read (fh,offset,length,data,dataoffset,callback=&PL_sv_undef)
+aio_read (SV *fh, SV *offset, SV *length, SV8 *data, IV dataoffset, SV *callback=&PL_sv_undef)
-	SV *	fh
-        UV	offset
-        UV	length
-        SV8 *	data
-        UV	dataoffset
-        SV *	callback
         ALIAS:
            aio_read  = REQ_READ
            aio_write = REQ_WRITE
 	PROTOTYPE: $$$$$;$
         PPCODE:
 {
         STRLEN svlen;
         char *svptr = SvPVbyte (data, svlen);
+        UV len = SvUV (length);
         SvUPGRADE (data, SVt_PV);
         SvPOK_on (data);
         if (dataoffset < 0)
           dataoffset += svlen;
         if (dataoffset < 0 || dataoffset > svlen)
-          croak ("data offset outside of string");
+          croak ("dataoffset outside of data scalar");
         if (ix == REQ_WRITE)
           {
             /* write: check length and adjust. */
-            if (length < 0 || length + dataoffset > svlen)
+            if (!SvOK (length) || len + dataoffset > svlen)
-              length = svlen - dataoffset;
+              len = svlen - dataoffset;
           }
         else
           {
             /* read: grow scalar as necessary */
-            svptr = SvGROW (data, length + dataoffset);
+            svptr = SvGROW (data, len + dataoffset + 1);
           }
-        if (length < 0)
+        if (len < 0)
           croak ("length must not be negative");
         {
           dREQ;
           req->type = ix;
-          req->fh   = newSVsv (fh);
+          req->sv1  = newSVsv (fh);
           req->int1 = PerlIO_fileno (ix == REQ_READ ? IoIFP (sv_2io (fh))
                                                     : IoOFP (sv_2io (fh)));
-          req->offs = offset;
+          req->offs = SvOK (offset) ? SvVAL64 (offset) : -1;
-          req->size = length;
+          req->size = len;
-          req->sv1  = SvREFCNT_inc (data);
+          req->sv2  = SvREFCNT_inc (data);
           req->ptr1 = (char *)svptr + dataoffset;
           req->stroffset = dataoffset;
           if (!SvREADONLY (data))
             {
               SvREADONLY_on (data);
-              req->flags |= FLAG_SV1_RO_OFF;
+              req->flags |= FLAG_SV2_RO_OFF;
             }
           REQ_SEND;
         }
 }
 void
-aio_readlink (path,callback=&PL_sv_undef)
+aio_readlink (SV8 *path, SV *callback=&PL_sv_undef)
-	SV8 *	path
-        SV *	callback
 	PROTOTYPE: $$;$
         PPCODE:
 {
 	SV *data;
         dREQ;
         data = newSV (NAME_MAX);
         SvPOK_on (data);
         req->type = REQ_READLINK;
-        req->fh   = newSVsv (path);
+        req->sv1  = newSVsv (path);
-        req->ptr2 = SvPVbyte_nolen (req->fh);
+        req->ptr2 = SvPVbyte_nolen (req->sv1);
-        req->sv1  = data;
+        req->sv2  = data;
         req->ptr1 = SvPVbyte_nolen (data);
         REQ_SEND;
 }
 void
-aio_sendfile (out_fh,in_fh,in_offset,length,callback=&PL_sv_undef)
+aio_sendfile (SV *out_fh, SV *in_fh, SV *in_offset, UV length, SV *callback=&PL_sv_undef)
-        SV *	out_fh
-        SV *	in_fh
-        UV	in_offset
-        UV	length
-        SV *	callback
 	PROTOTYPE: $$$$;$
         PPCODE:
 {
 	dREQ;
         req->type = REQ_SENDFILE;
-        req->fh   = newSVsv (out_fh);
+        req->sv1  = newSVsv (out_fh);
         req->int1 = PerlIO_fileno (IoIFP (sv_2io (out_fh)));
         req->sv2  = newSVsv (in_fh);
         req->int2 = PerlIO_fileno (IoIFP (sv_2io (in_fh)));
-        req->offs = in_offset;
+        req->offs = SvVAL64 (in_offset);
         req->size = length;
         REQ_SEND;
 }
 void
-aio_readahead (fh,offset,length,callback=&PL_sv_undef)
+aio_readahead (SV *fh, SV *offset, IV length, SV *callback=&PL_sv_undef)
-        SV *	fh
-        UV	offset
-        IV	length
-        SV *	callback
 	PROTOTYPE: $$$;$
         PPCODE:
 {
 	dREQ;
         req->type = REQ_READAHEAD;
-        req->fh   = newSVsv (fh);
+        req->sv1  = newSVsv (fh);
         req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh)));
-        req->offs = offset;
+        req->offs = SvVAL64 (offset);
         req->size = length;
         REQ_SEND;
 }
 void
-aio_stat (fh_or_path,callback=&PL_sv_undef)
+aio_stat (SV8 *fh_or_path, SV *callback=&PL_sv_undef)
-        SV8 *		fh_or_path
-        SV *		callback
         ALIAS:
            aio_stat  = REQ_STAT
            aio_lstat = REQ_LSTAT
 	PPCODE:
 {
 	dREQ;
         req->ptr2 = malloc (sizeof (Stat_t));
         if (!req->ptr2)
           {
-            req_free (req);
+            req_destroy (req);
             croak ("out of memory during aio_stat statdata allocation");
           }
         req->flags |= FLAG_PTR2_FREE;
+        req->sv1 = newSVsv (fh_or_path);
         if (SvPOK (fh_or_path))
           {
             req->type = ix;
-            req->sv1  = newSVsv (fh_or_path);
             req->ptr1 = SvPVbyte_nolen (req->sv1);
           }
         else
           {
             req->type = REQ_FSTAT;
-            req->fh   = newSVsv (fh_or_path);
             req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
           }
         REQ_SEND;
 }
 void
+aio_utime (SV8 *fh_or_path, SV *atime, SV *mtime, SV *callback=&PL_sv_undef)
+	PPCODE:
+{
+	dREQ;
+        req->nv1 = SvOK (atime) ? SvNV (atime) : -1.;
+        req->nv2 = SvOK (mtime) ? SvNV (mtime) : -1.;
+        req->sv1 = newSVsv (fh_or_path);
+        if (SvPOK (fh_or_path))
+          {
+            req->type = REQ_UTIME;
+            req->ptr1 = SvPVbyte_nolen (req->sv1);
+          }
+        else
+          {
+            req->type = REQ_FUTIME;
+            req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
+          }
+        REQ_SEND;
+}
+void
+aio_truncate (SV8 *fh_or_path, SV *offset, SV *callback=&PL_sv_undef)
+	PPCODE:
+{
+	dREQ;
+        req->sv1  = newSVsv (fh_or_path);
+        req->offs = SvOK (offset) ? SvVAL64 (offset) : -1;
+        if (SvPOK (fh_or_path))
+          {
+            req->type = REQ_TRUNCATE;
+            req->ptr1 = SvPVbyte_nolen (req->sv1);
+          }
+        else
+          {
+            req->type = REQ_FTRUNCATE;
+            req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
+          }
+        REQ_SEND;
+}
+void
+aio_chmod (SV8 *fh_or_path, int mode, SV *callback=&PL_sv_undef)
+	PPCODE:
+{
+	dREQ;
+        req->mode = mode;
+        req->sv1  = newSVsv (fh_or_path);
+        if (SvPOK (fh_or_path))
+          {
+            req->type = REQ_CHMOD;
+            req->ptr1 = SvPVbyte_nolen (req->sv1);
+          }
+        else
+          {
+            req->type = REQ_FCHMOD;
+            req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
+          }
+        REQ_SEND;
+}
+void
+aio_chown (SV8 *fh_or_path, SV *uid, SV *gid, SV *callback=&PL_sv_undef)
+	PPCODE:
+{
+	dREQ;
+        req->int2 = SvOK (uid) ? SvIV (uid) : -1;
+        req->int3 = SvOK (gid) ? SvIV (gid) : -1;
+        req->sv1  = newSVsv (fh_or_path);
+        if (SvPOK (fh_or_path))
+          {
+            req->type = REQ_CHOWN;
+            req->ptr1 = SvPVbyte_nolen (req->sv1);
+          }
+        else
+          {
+            req->type = REQ_FCHOWN;
+            req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
+          }
+        REQ_SEND;
+}
+void
-aio_unlink (pathname,callback=&PL_sv_undef)
+aio_unlink (SV8 *pathname, SV *callback=&PL_sv_undef)
-	SV8 *	pathname
-	SV *	callback
         ALIAS:
            aio_unlink  = REQ_UNLINK
            aio_rmdir   = REQ_RMDIR
            aio_readdir = REQ_READDIR
 	PPCODE:
 	REQ_SEND;
 }
 void
+aio_mkdir (SV8 *pathname, int mode, SV *callback=&PL_sv_undef)
+	PPCODE:
+{
+	dREQ;
+        req->type = REQ_MKDIR;
+	req->sv1  = newSVsv (pathname);
+	req->ptr1 = SvPVbyte_nolen (req->sv1);
+        req->mode = mode;
+	REQ_SEND;
+}
+void
-aio_link (oldpath,newpath,callback=&PL_sv_undef)
+aio_link (SV8 *oldpath, SV8 *newpath, SV *callback=&PL_sv_undef)
-	SV8 *	oldpath
-	SV8 *	newpath
-	SV *	callback
         ALIAS:
            aio_link    = REQ_LINK
            aio_symlink = REQ_SYMLINK
            aio_rename  = REQ_RENAME
 	PPCODE:
 {
 	dREQ;
         req->type = ix;
-	req->fh   = newSVsv (oldpath);
+	req->sv2  = newSVsv (oldpath);
-	req->ptr2 = SvPVbyte_nolen (req->fh);
+	req->ptr2 = SvPVbyte_nolen (req->sv2);
 	req->sv1  = newSVsv (newpath);
 	req->ptr1 = SvPVbyte_nolen (req->sv1);
 	REQ_SEND;
 }
 void
-aio_mknod (pathname,mode,dev,callback=&PL_sv_undef)
+aio_mknod (SV8 *pathname, int mode, UV dev, SV *callback=&PL_sv_undef)
-	SV8 *	pathname
-	SV *	callback
-        UV	mode
-        UV	dev
 	PPCODE:
 {
 	dREQ;
         req->type = REQ_MKNOD;
 	REQ_SEND;
 }
 void
-aio_busy (delay,callback=&PL_sv_undef)
+aio_busy (double delay, SV *callback=&PL_sv_undef)
-	double	delay
-	SV *	callback
 	PPCODE:
 {
 	dREQ;
         req->type = REQ_BUSY;
-        req->int1 = delay < 0. ? 0 : delay;
+        req->nv1  = delay < 0. ? 0. : delay;
-        req->int2 = delay < 0. ? 0 : 1000. * (delay - req->int1);
 	REQ_SEND;
 }
 void
-aio_group (callback=&PL_sv_undef)
+aio_group (SV *callback=&PL_sv_undef)
-        SV *	callback
 	PROTOTYPE: ;$
         PPCODE:
 {
 	dREQ;
         req_send (req);
         XPUSHs (req_sv (req, AIO_GRP_KLASS));
 }
 void
-aio_nop (callback=&PL_sv_undef)
+aio_nop (SV *callback=&PL_sv_undef)
-	SV *	callback
 	PPCODE:
 {
 	dREQ;
         req->type = REQ_NOP;
         CODE:
 {
         if (block_sig_level)
           croak ("cannot call IO::AIO::setsig from within aio_block/callback");
-        LOCK (reslock);
+        X_LOCK (reslock);
         main_tid = pthread_self ();
         main_sig = signum;
-        UNLOCK (reslock);
+        X_UNLOCK (reslock);
         if (main_sig && npending)
 	  pthread_kill (main_tid, main_sig);
 }
 int
 nthreads()
 	PROTOTYPE:
 	CODE:
-        if (WORDACCESS_UNSAFE) LOCK   (wrklock);
+        if (WORDACCESS_UNSAFE) X_LOCK   (wrklock);
         RETVAL = started;
-        if (WORDACCESS_UNSAFE) UNLOCK (wrklock);
+        if (WORDACCESS_UNSAFE) X_UNLOCK (wrklock);
 	OUTPUT:
 	RETVAL
 PROTOTYPES: DISABLE

Diff Legend

-–
+Removed lines
-+
+Added lines
-<
+Changed lines
->
+Changed lines

Comparing IO-AIO/AIO.xs (file contents): Revision 1.95 by root, Sun Nov 26 22:47:58 2006 UTC vs. Revision 1.106 by root, Mon Sep 24 18:14:00 2007 UTC

Diff Legend

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.95 by root, Sun Nov 26 22:47:58 2006 UTC vs.
Revision 1.106 by root, Mon Sep 24 18:14:00 2007 UTC