IO-AIO/AIO.xs

#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 <sys/types.h>
#include <sys/stat.h>
#include <limits.h>
#include <unistd.h>
#include <fcntl.h>
#include <signal.h>
#include <sched.h>

#if HAVE_SENDFILE
# if __linux
#  include <sys/sendfile.h>
# elif __freebsd
#  include <sys/socket.h>
#  include <sys/uio.h>
# elif __hpux
#  include <sys/socket.h>
# elif __solaris /* not yet */
#  include <sys/sendfile.h>
# else
#  error sendfile support requested but not available
# endif
#endif

/* used for struct dirent, AIX doesn't provide it */
#ifndef NAME_MAX
# define NAME_MAX 4096
#endif

#if __ia64
# define STACKSIZE 65536
#else
# define STACKSIZE  8192
#endif

enum {
  REQ_QUIT,
  REQ_OPEN, REQ_CLOSE,
  REQ_READ, REQ_WRITE, REQ_READAHEAD,
  REQ_SENDFILE,
  REQ_STAT, REQ_LSTAT, REQ_FSTAT,
  REQ_FSYNC, REQ_FDATASYNC,
  REQ_UNLINK, REQ_RMDIR, REQ_RENAME,
  REQ_READDIR,
  REQ_LINK, REQ_SYMLINK,
};

typedef struct aio_cb {
  struct aio_cb *volatile next;

  int type;

  /* should receive a cleanup, with unions */
  int fd, fd2;
  off_t offset;
  size_t length;
  ssize_t result;
  mode_t mode; /* open */
  int errorno;
  SV *data, *callback;
  SV *fh, *fh2;
  void *dataptr, *data2ptr;
  STRLEN dataoffset;

  Stat_t *statdata;
} aio_cb;

typedef aio_cb *aio_req;

static int started, wanted;
static volatile int nreqs;
static int max_outstanding = 1<<30;
static int respipe [2];

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

static volatile aio_req reqs, reqe; /* queue start, queue end */
static volatile aio_req ress, rese; /* queue start, queue end */

static void free_req (aio_req req)
{
  if (req->data)
    SvREFCNT_dec (req->data);

  if (req->fh)
    SvREFCNT_dec (req->fh);

  if (req->fh2)
    SvREFCNT_dec (req->fh2);

  if (req->statdata)
    Safefree (req->statdata);

  if (req->callback)
    SvREFCNT_dec (req->callback);

  if (req->type == REQ_READDIR && req->result >= 0)
    free (req->data2ptr);

  Safefree (req);
}

static void
poll_wait ()
{
  if (nreqs && !ress)
    {
      fd_set rfd;
      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 do_croak = 0;
  aio_req req;

  for (;;)
    {
      pthread_mutex_lock (&reslock);
      req = ress;

      if (req)
        {
          ress = req->next;

          if (!ress)
            {
              /* read any signals sent by the worker threads */
              char buf [32];
              while (read (respipe [0], buf, 32) == 32)
                ;

              rese = 0;
            }
        }

      pthread_mutex_unlock (&reslock);

      if (!req)
        break;

      nreqs--;

      if (req->type == REQ_QUIT)
        started--;
      else
        {
          int errorno = errno;
          errno = req->errorno;

          if (req->type == REQ_READ)
            SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));

          if (req->data2ptr && (req->type == REQ_READ || req->type == REQ_WRITE))
            SvREADONLY_off (req->data);

          if (req->statdata)
            {
              PL_laststype   = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
              PL_laststatval = req->result;
              PL_statcache   = *(req->statdata);
            }

          ENTER;
          PUSHMARK (SP);

          if (req->type == REQ_READDIR)
            {
              SV *rv = &PL_sv_undef;

              if (req->result >= 0)
                {
                  char *buf = req->data2ptr;
                  AV *av = newAV ();

                  while (req->result)
                    {
                      SV *sv = newSVpv (buf, 0);

                      av_push (av, sv);
                      buf += SvCUR (sv) + 1;
                      req->result--;
                    }

                  rv = sv_2mortal (newRV_noinc ((SV *)av));
                }

              XPUSHs (rv);
            }
          else
            {
              XPUSHs (sv_2mortal (newSViv (req->result)));

              if (req->type == REQ_OPEN)
                {
                  /* convert fd to fh */
                  SV *fh;

                  PUTBACK;
                  call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL);
                  SPAGAIN;

                  fh = SvREFCNT_inc (POPs);

                  PUSHMARK (SP);
                  XPUSHs (sv_2mortal (fh));
                }
            }

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

              if (SvTRUE (ERRSV))
                {
                  free_req (req);
                  croak (0);
                }
            }

          LEAVE;

          errno = errorno;
          count++;
        }

      free_req (req);
    }

  return count;
}

static void *aio_proc(void *arg);

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

  pthread_attr_init (&attr);
  pthread_attr_setstacksize (&attr, STACKSIZE);
  pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);

  sigfillset (&fullsigset);
  sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);

  if (pthread_create (&tid, &attr, aio_proc, 0) == 0)
    started++;

  sigprocmask (SIG_SETMASK, &oldsigset, 0);
}

static void
send_req (aio_req req)
{
  while (started < wanted && nreqs >= started)
    start_thread ();

  nreqs++;

  pthread_mutex_lock (&reqlock);

  req->next = 0;

  if (reqe)
    {
      reqe->next = req;
      reqe = req;
    }
  else
    reqe = reqs = req;

  pthread_cond_signal (&reqwait);
  pthread_mutex_unlock (&reqlock);

  if (nreqs > max_outstanding)
    for (;;)
      {
        poll_cb ();

        if (nreqs <= max_outstanding)
          break;

        poll_wait ();
      }
}

static void
end_thread (void)
{
  aio_req req;
  Newz (0, req, 1, aio_cb);
  req->type = REQ_QUIT;

  send_req (req);
}

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

static void max_parallel (int nthreads)
{
  int cur = started;

  if (wanted > nthreads)
    wanted = nthreads;

  while (cur > wanted)
    {
      end_thread ();
      cur--;
    }

  while (started > wanted)
    {
      poll_wait ();
      poll_cb ();
    }
}

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 */

#if !HAVE_PREADWRITE
# define pread  aio_pread
# define pwrite aio_pwrite

/*
 * 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 ssize_t
pread (int fd, void *buf, size_t count, off_t offset)
{
  ssize_t res;
  off_t ooffset;

  pthread_mutex_lock (&preadwritelock);
  ooffset = lseek (fd, 0, SEEK_CUR);
  lseek (fd, offset, SEEK_SET);
  res = read (fd, buf, count);
  lseek (fd, ooffset, SEEK_SET);
  pthread_mutex_unlock (&preadwritelock);

  return res;
}

static ssize_t
pwrite (int fd, void *buf, size_t count, off_t offset)
{
  ssize_t res;
  off_t ooffset;

  pthread_mutex_lock (&preadwritelock);
  ooffset = lseek (fd, 0, SEEK_CUR);
  lseek (fd, offset, SEEK_SET);
  res = write (fd, buf, count);
  lseek (fd, offset, SEEK_SET);
  pthread_mutex_unlock (&preadwritelock);

  return res;
}
#endif

#if !HAVE_FDATASYNC
# define fdatasync fsync
#endif

#if !HAVE_READAHEAD
# define readahead aio_readahead

static ssize_t
readahead (int fd, off_t offset, size_t count)
{
  char readahead_buf[4096];

  while (count > 0)
    {
      size_t len = count < sizeof (readahead_buf) ? count : sizeof (readahead_buf);

      pread (fd, readahead_buf, len, offset);
      offset += len;
      count  -= len;
    }

  errno = 0;
}
#endif

#if !HAVE_READDIR_R
# define readdir_r aio_readdir_r

static pthread_mutex_t readdirlock = PTHREAD_MUTEX_INITIALIZER;
  
static int
readdir_r (DIR *dirp, struct dirent *ent, struct dirent **res)
{
  struct dirent *e;
  int errorno;

  pthread_mutex_lock (&readdirlock);

  e = readdir (dirp);
  errorno = errno;

  if (e)
    {
      *res = ent;
      strcpy (ent->d_name, e->d_name);
    }
  else
    *res = 0;

  pthread_mutex_unlock (&readdirlock);

  errno = errorno;
  return e ? 0 : -1;
}
#endif

/* sendfile always needs emulation */
static ssize_t
sendfile_ (int ofd, int ifd, off_t offset, size_t count)
{
  ssize_t res;

  if (!count)
    return 0;

#if HAVE_SENDFILE
# if __linux
  res = sendfile (ofd, ifd, &offset, count);

# elif __freebsd
  /*
   * Of course, the freebsd sendfile is a dire hack with no thoughts
   * wasted on making it similar to other I/O functions.
   */
  {
    off_t sbytes;
    res = sendfile (ifd, ofd, offset, count, 0, &sbytes, 0);

    if (res < 0 && sbytes)
      /* maybe only on EAGAIN only: as usual, the manpage leaves you guessing */
      res = sbytes;
  }

# elif __hpux
  res = sendfile (ofd, ifd, offset, count, 0, 0);

# elif __solaris
  {
    struct sendfilevec vec;
    size_t sbytes;

    vec.sfv_fd   = ifd;
    vec.sfv_flag = 0;
    vec.sfv_off  = offset;
    vec.sfv_len  = count;

    res = sendfilev (ofd, &vec, 1, &sbytes);

    if (res < 0 && sbytes)
      res = sbytes;
  }

# endif
#else
  res = -1;
  errno = ENOSYS;
#endif

  if (res <  0
      && (errno == ENOSYS || errno == EINVAL || errno == ENOTSOCK
#if __solaris
          || errno == EAFNOSUPPORT || errno == EPROTOTYPE
#endif
         )
      )
    {
      /* emulate sendfile. this is a major pain in the ass */
      char buf[4096];
      res = 0;

      while (count)
        {
          ssize_t cnt;
          
          cnt = pread (ifd, buf, count > 4096 ? 4096 : count, offset);

          if (cnt <= 0)
            {
              if (cnt && !res) res = -1;
              break;
            }

          cnt = write (ofd, buf, cnt);

          if (cnt <= 0)
            {
              if (cnt && !res) res = -1;
              break;
            }

          offset += cnt;
          res    += cnt;
          count  -= cnt;
        }
    }

  return res;
}

/* read a full directory */
static int
scandir_ (const char *path, void **namesp)
{
  DIR *dirp = opendir (path);
  union
  {    
    struct dirent d;
    char b [offsetof (struct dirent, d_name) + NAME_MAX + 1];
  } u;
  struct dirent *entp;
  char *name, *names;
  int memlen = 4096;
  int memofs = 0;
  int res = 0;
  int errorno;

  if (!dirp)
    return -1;

  names = malloc (memlen);

  for (;;)
    {
      errno = 0, readdir_r (dirp, &u.d, &entp);

      if (!entp)
        break;

      name = entp->d_name;

      if (name [0] != '.' || (name [1] && (name [1] != '.' || name [2])))
        {
          int len = strlen (name) + 1;

          res++;

          while (memofs + len > memlen)
            {
              memlen *= 2;
              names = realloc (names, memlen);
              if (!names)
                break;
            }

          memcpy (names + memofs, name, len);
          memofs += len;
        }
    }

  errorno = errno;
  closedir (dirp);

  if (errorno)
    {
      free (names);
      errno = errorno;
      res = -1;
    }

  *namesp = (void *)names;
  return res;
}

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

static void *
aio_proc (void *thr_arg)
{
  aio_req req;
  int type;

  do
    {
      pthread_mutex_lock (&reqlock);

      for (;;)
        {
          req = reqs;

          if (reqs)
            {
              reqs = reqs->next;
              if (!reqs) reqe = 0;
            }

          if (req)
            break;

          pthread_cond_wait (&reqwait, &reqlock);
        }

      pthread_mutex_unlock (&reqlock);
     
      errno = 0; /* strictly unnecessary */

      type = req->type;

      switch (type)
        {
          case REQ_READ:      req->result = pread     (req->fd, req->dataptr, req->length, req->offset); break;
          case REQ_WRITE:     req->result = pwrite    (req->fd, req->dataptr, req->length, req->offset); break;

          case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
          case REQ_SENDFILE:  req->result = sendfile_ (req->fd, req->fd2, req->offset, req->length); break;

          case REQ_STAT:      req->result = stat      (req->dataptr, req->statdata); break;
          case REQ_LSTAT:     req->result = lstat     (req->dataptr, req->statdata); break;
          case REQ_FSTAT:     req->result = fstat     (req->fd     , req->statdata); break;

          case REQ_OPEN:      req->result = open      (req->dataptr, req->fd, req->mode); break;
          case REQ_CLOSE:     req->result = close     (req->fd); break;
          case REQ_UNLINK:    req->result = unlink    (req->dataptr); break;
          case REQ_RMDIR:     req->result = rmdir     (req->dataptr); break;
          case REQ_RENAME:    req->result = rename    (req->data2ptr, req->dataptr); break;
          case REQ_LINK:      req->result = link      (req->data2ptr, req->dataptr); break;
          case REQ_SYMLINK:   req->result = symlink   (req->data2ptr, req->dataptr); break;

          case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
          case REQ_FSYNC:     req->result = fsync     (req->fd); break;
          case REQ_READDIR:   req->result = scandir_  (req->dataptr, &req->data2ptr); break;

          case REQ_QUIT:
            break;

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

      req->errorno = errno;

      pthread_mutex_lock (&reslock);

      req->next = 0;

      if (rese)
        {
          rese->next = req;
          rese = req;
        }
      else
        {
          rese = ress = req;

          /* write a dummy byte to the pipe so fh becomes ready */
          write (respipe [1], &respipe, 1);
        }

      pthread_mutex_unlock (&reslock);
    }
  while (type != REQ_QUIT);

  return 0;
}

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

static void atfork_prepare (void)
{
  pthread_mutex_lock (&reqlock);
  pthread_mutex_lock (&reslock);
#if !HAVE_PREADWRITE
  pthread_mutex_lock (&preadwritelock);
#endif
#if !HAVE_READDIR_R
  pthread_mutex_lock (&readdirlock);
#endif
}

static void atfork_parent (void)
{
#if !HAVE_READDIR_R
  pthread_mutex_unlock (&readdirlock);
#endif
#if !HAVE_PREADWRITE
  pthread_mutex_unlock (&preadwritelock);
#endif
  pthread_mutex_unlock (&reslock);
  pthread_mutex_unlock (&reqlock);
}

static void atfork_child (void)
{
  aio_req prv;

  started = 0;

  while (reqs)
    {
      prv = reqs;
      reqs = prv->next;
      free_req (prv);
    }

  reqs = reqe = 0;
      
  while (ress)
    {
      prv = ress;
      ress = prv->next;
      free_req (prv);
    }
      
  ress = rese = 0;

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

  atfork_parent ();
}

#define dREQ                                                    \
  aio_req req;                                                  \
                                                                \
  if (SvOK (callback) && !SvROK (callback))                     \
    croak ("clalback 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);
        
MODULE = IO::AIO                PACKAGE = IO::AIO

PROTOTYPES: ENABLE

BOOT:
{
        create_pipe ();
        pthread_atfork (atfork_prepare, atfork_parent, atfork_child);
}

void
min_parallel (nthreads)
        int     nthreads
        PROTOTYPE: $

void
max_parallel (nthreads)
        int     nthreads
        PROTOTYPE: $

int
max_outstanding (nreqs)
        int nreqs
        PROTOTYPE: $
        CODE:
        RETVAL = max_outstanding;
        max_outstanding = nreqs;

void
aio_open (pathname,flags,mode,callback=&PL_sv_undef)
        SV *    pathname
        int     flags
        int     mode
        SV *    callback
        PROTOTYPE: $$$;$
        CODE:
{
        dREQ;

        req->type = REQ_OPEN;
        req->data = newSVsv (pathname);
        req->dataptr = SvPVbyte_nolen (req->data);
        req->fd = flags;
        req->mode = mode;

        send_req (req);
}

void
aio_close (fh,callback=&PL_sv_undef)
        SV *    fh
        SV *    callback
        PROTOTYPE: $;$
        ALIAS:
           aio_close     = REQ_CLOSE
           aio_fsync     = REQ_FSYNC
           aio_fdatasync = REQ_FDATASYNC
        CODE:
{
        dREQ;

        req->type = ix;
        req->fh = newSVsv (fh);
        req->fd = PerlIO_fileno (IoIFP (sv_2io (fh)));

        send_req (req);
}

void
aio_read (fh,offset,length,data,dataoffset,callback=&PL_sv_undef)
        SV *    fh
        UV      offset
        UV      length
        SV *    data
        UV      dataoffset
        SV *    callback
        ALIAS:
           aio_read  = REQ_READ
           aio_write = REQ_WRITE
        PROTOTYPE: $$$$$;$
        CODE:
{
        aio_req req;
        STRLEN svlen;
        char *svptr = SvPVbyte (data, svlen);

        SvUPGRADE (data, SVt_PV);
        SvPOK_on (data);

        if (dataoffset < 0)
          dataoffset += svlen;

        if (dataoffset < 0 || dataoffset > svlen)
          croak ("data offset outside of string");

        if (ix == REQ_WRITE)
          {
            /* write: check length and adjust. */
            if (length < 0 || length + dataoffset > svlen)
              length = svlen - dataoffset;
          }
        else
          {
            /* read: grow scalar as necessary */
            svptr = SvGROW (data, length + dataoffset);
          }

        if (length < 0)
          croak ("length must not be negative");

        {
          dREQ;

          req->type = ix;
          req->fh = newSVsv (fh);
          req->fd = PerlIO_fileno (ix == REQ_READ ? IoIFP (sv_2io (fh))
                                                  : IoOFP (sv_2io (fh)));
          req->offset = offset;
          req->length = length;
          req->data = SvREFCNT_inc (data);
          req->dataptr = (char *)svptr + dataoffset;

          if (!SvREADONLY (data))
            {
              SvREADONLY_on (data);
              req->data2ptr = (void *)data;
            }

          send_req (req);
        }
}

void
aio_sendfile (out_fh,in_fh,in_offset,length,callback=&PL_sv_undef)
        SV *    out_fh
        SV *    in_fh
        UV      in_offset
        UV      length
        SV *    callback
        PROTOTYPE: $$$$;$
        CODE:
{
        dREQ;

        req->type = REQ_SENDFILE;
        req->fh = newSVsv (out_fh);
        req->fd = PerlIO_fileno (IoIFP (sv_2io (out_fh)));
        req->fh2 = newSVsv (in_fh);
        req->fd2 = PerlIO_fileno (IoIFP (sv_2io (in_fh)));
        req->offset = in_offset;
        req->length = length;

        send_req (req);
}

void
aio_readahead (fh,offset,length,callback=&PL_sv_undef)
        SV *    fh
        UV      offset
        IV      length
        SV *    callback
        PROTOTYPE: $$$;$
        CODE:
{
        dREQ;

        req->type = REQ_READAHEAD;
        req->fh = newSVsv (fh);
        req->fd = PerlIO_fileno (IoIFP (sv_2io (fh)));
        req->offset = offset;
        req->length = length;

        send_req (req);
}

void
aio_stat (fh_or_path,callback=&PL_sv_undef)
        SV *            fh_or_path
        SV *            callback
        ALIAS:
           aio_stat  = REQ_STAT
           aio_lstat = REQ_LSTAT
        CODE:
{
        dREQ;

        New (0, req->statdata, 1, Stat_t);
        if (!req->statdata)
          {
            free_req (req);
            croak ("out of memory during aio_req->statdata allocation");
          }

        if (SvPOK (fh_or_path))
          {
            req->type = ix;
            req->data = newSVsv (fh_or_path);
            req->dataptr = SvPVbyte_nolen (req->data);
          }
        else
          {
            req->type = REQ_FSTAT;
            req->fh = newSVsv (fh_or_path);
            req->fd = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
          }

        send_req (req);
}

void
aio_unlink (pathname,callback=&PL_sv_undef)
        SV * pathname
        SV * callback
        ALIAS:
           aio_unlink  = REQ_UNLINK
           aio_rmdir   = REQ_RMDIR
           aio_readdir = REQ_READDIR
        CODE:
{
        dREQ;
        
        req->type = ix;
        req->data = newSVsv (pathname);
        req->dataptr = SvPVbyte_nolen (req->data);
        
        send_req (req);
}

void
aio_link (oldpath,newpath,callback=&PL_sv_undef)
        SV * oldpath
        SV * newpath
        SV * callback
        ALIAS:
           aio_link    = REQ_LINK
           aio_symlink = REQ_SYMLINK
           aio_rename  = REQ_RENAME
        CODE:
{
        dREQ;
        
        req->type = ix;
        req->fh = newSVsv (oldpath);
        req->data2ptr = SvPVbyte_nolen (req->fh);
        req->data = newSVsv (newpath);
        req->dataptr = SvPVbyte_nolen (req->data);
        
        send_req (req);
}

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

void
poll()
        PROTOTYPE:
        CODE:
        if (nreqs)
          {
            poll_wait ();
            poll_cb ();
          }

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:
        if (nreqs)
          poll_wait ();

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

Revision:	1.40
Committed:	Sat Jun 24 16:27:02 2006 UTC (17 years, 10 months ago) by root
Branch:	MAIN
Changes since 1.39:	+24 -32 lines
Log Message:	* empty log message *
#	Content
1	#define _REENTRANT 1
2	#include <errno.h>
3
4	#include "EXTERN.h"
5	#include "perl.h"
6	#include "XSUB.h"
7
8	#include "autoconf/config.h"
9
10	#include <pthread.h>
11
12	#include <stddef.h>
13	#include <sys/types.h>
14	#include <sys/stat.h>
15	#include <limits.h>
16	#include <unistd.h>
17	#include <fcntl.h>
18	#include <signal.h>
19	#include <sched.h>
20
21	#if HAVE_SENDFILE
22	# if __linux
23	# include <sys/sendfile.h>
24	# elif __freebsd
25	# include <sys/socket.h>
26	# include <sys/uio.h>
27	# elif __hpux
28	# include <sys/socket.h>
29	# elif __solaris /* not yet */
30	# include <sys/sendfile.h>
31	# else
32	# error sendfile support requested but not available
33	# endif
34	#endif
35
36	/* used for struct dirent, AIX doesn't provide it */
37	#ifndef NAME_MAX
38	# define NAME_MAX 4096
39	#endif
40
41	#if __ia64
42	# define STACKSIZE 65536
43	#else
44	# define STACKSIZE 8192
45	#endif
46
47	enum {
48	REQ_QUIT,
49	REQ_OPEN, REQ_CLOSE,
50	REQ_READ, REQ_WRITE, REQ_READAHEAD,
51	REQ_SENDFILE,
52	REQ_STAT, REQ_LSTAT, REQ_FSTAT,
53	REQ_FSYNC, REQ_FDATASYNC,
54	REQ_UNLINK, REQ_RMDIR, REQ_RENAME,
55	REQ_READDIR,
56	REQ_LINK, REQ_SYMLINK,
57	};
58
59	typedef struct aio_cb {
60	struct aio_cb *volatile next;
61
62	int type;
63
64	/* should receive a cleanup, with unions */
65	int fd, fd2;
66	off_t offset;
67	size_t length;
68	ssize_t result;
69	mode_t mode; /* open */
70	int errorno;
71	SV data, callback;
72	SV fh, fh2;
73	void dataptr, data2ptr;
74	STRLEN dataoffset;
75
76	Stat_t *statdata;
77	} aio_cb;
78
79	typedef aio_cb *aio_req;
80
81	static int started, wanted;
82	static volatile int nreqs;
83	static int max_outstanding = 1<<30;
84	static int respipe [2];
85
86	static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER;
87	static pthread_mutex_t reqlock = PTHREAD_MUTEX_INITIALIZER;
88	static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
89
90	static volatile aio_req reqs, reqe; /* queue start, queue end */
91	static volatile aio_req ress, rese; /* queue start, queue end */
92
93	static void free_req (aio_req req)
94	{
95	if (req->data)
96	SvREFCNT_dec (req->data);
97
98	if (req->fh)
99	SvREFCNT_dec (req->fh);
100
101	if (req->fh2)
102	SvREFCNT_dec (req->fh2);
103
104	if (req->statdata)
105	Safefree (req->statdata);
106
107	if (req->callback)
108	SvREFCNT_dec (req->callback);
109
110	if (req->type == REQ_READDIR && req->result >= 0)
111	free (req->data2ptr);
112
113	Safefree (req);
114	}
115
116	static void
117	poll_wait ()
118	{
119	if (nreqs && !ress)
120	{
121	fd_set rfd;
122	FD_ZERO(&rfd);
123	FD_SET(respipe [0], &rfd);
124
125	select (respipe [0] + 1, &rfd, 0, 0, 0);
126	}
127	}
128
129	static int
130	poll_cb ()
131	{
132	dSP;
133	int count = 0;
134	int do_croak = 0;
135	aio_req req;
136
137	for (;;)
138	{
139	pthread_mutex_lock (&reslock);
140	req = ress;
141
142	if (req)
143	{
144	ress = req->next;
145
146	if (!ress)
147	{
148	/* read any signals sent by the worker threads */
149	char buf [32];
150	while (read (respipe [0], buf, 32) == 32)
151	;
152
153	rese = 0;
154	}
155	}
156
157	pthread_mutex_unlock (&reslock);
158
159	if (!req)
160	break;
161
162	nreqs--;
163
164	if (req->type == REQ_QUIT)
165	started--;
166	else
167	{
168	int errorno = errno;
169	errno = req->errorno;
170
171	if (req->type == REQ_READ)
172	SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
173
174	if (req->data2ptr && (req->type == REQ_READ \|\| req->type == REQ_WRITE))
175	SvREADONLY_off (req->data);
176
177	if (req->statdata)
178	{
179	PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
180	PL_laststatval = req->result;
181	PL_statcache = *(req->statdata);
182	}
183
184	ENTER;
185	PUSHMARK (SP);
186
187	if (req->type == REQ_READDIR)
188	{
189	SV *rv = &PL_sv_undef;
190
191	if (req->result >= 0)
192	{
193	char *buf = req->data2ptr;
194	AV *av = newAV ();
195
196	while (req->result)
197	{
198	SV *sv = newSVpv (buf, 0);
199
200	av_push (av, sv);
201	buf += SvCUR (sv) + 1;
202	req->result--;
203	}
204
205	rv = sv_2mortal (newRV_noinc ((SV *)av));
206	}
207
208	XPUSHs (rv);
209	}
210	else
211	{
212	XPUSHs (sv_2mortal (newSViv (req->result)));
213
214	if (req->type == REQ_OPEN)
215	{
216	/* convert fd to fh */
217	SV *fh;
218
219	PUTBACK;
220	call_pv ("IO::AIO::_fd2fh", G_SCALAR \| G_EVAL);
221	SPAGAIN;
222
223	fh = SvREFCNT_inc (POPs);
224
225	PUSHMARK (SP);
226	XPUSHs (sv_2mortal (fh));
227	}
228	}
229
230	if (SvOK (req->callback))
231	{
232	PUTBACK;
233	call_sv (req->callback, G_VOID \| G_EVAL);
234	SPAGAIN;
235
236	if (SvTRUE (ERRSV))
237	{
238	free_req (req);
239	croak (0);
240	}
241	}
242
243	LEAVE;
244
245	errno = errorno;
246	count++;
247	}
248
249	free_req (req);
250	}
251
252	return count;
253	}
254
255	static void aio_proc(void arg);
256
257	static void
258	start_thread (void)
259	{
260	sigset_t fullsigset, oldsigset;
261	pthread_t tid;
262	pthread_attr_t attr;
263
264	pthread_attr_init (&attr);
265	pthread_attr_setstacksize (&attr, STACKSIZE);
266	pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
267
268	sigfillset (&fullsigset);
269	sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);
270
271	if (pthread_create (&tid, &attr, aio_proc, 0) == 0)
272	started++;
273
274	sigprocmask (SIG_SETMASK, &oldsigset, 0);
275	}
276
277	static void
278	send_req (aio_req req)
279	{
280	while (started < wanted && nreqs >= started)
281	start_thread ();
282
283	nreqs++;
284
285	pthread_mutex_lock (&reqlock);
286
287	req->next = 0;
288
289	if (reqe)
290	{
291	reqe->next = req;
292	reqe = req;
293	}
294	else
295	reqe = reqs = req;
296
297	pthread_cond_signal (&reqwait);
298	pthread_mutex_unlock (&reqlock);
299
300	if (nreqs > max_outstanding)
301	for (;;)
302	{
303	poll_cb ();
304
305	if (nreqs <= max_outstanding)
306	break;
307
308	poll_wait ();
309	}
310	}
311
312	static void
313	end_thread (void)
314	{
315	aio_req req;
316	Newz (0, req, 1, aio_cb);
317	req->type = REQ_QUIT;
318
319	send_req (req);
320	}
321
322	static void min_parallel (int nthreads)
323	{
324	if (wanted < nthreads)
325	wanted = nthreads;
326	}
327
328	static void max_parallel (int nthreads)
329	{
330	int cur = started;
331
332	if (wanted > nthreads)
333	wanted = nthreads;
334
335	while (cur > wanted)
336	{
337	end_thread ();
338	cur--;
339	}
340
341	while (started > wanted)
342	{
343	poll_wait ();
344	poll_cb ();
345	}
346	}
347
348	static void create_pipe ()
349	{
350	if (pipe (respipe))
351	croak ("unable to initialize result pipe");
352
353	if (fcntl (respipe [0], F_SETFL, O_NONBLOCK))
354	croak ("cannot set result pipe to nonblocking mode");
355
356	if (fcntl (respipe [1], F_SETFL, O_NONBLOCK))
357	croak ("cannot set result pipe to nonblocking mode");
358	}
359
360	/*****************************************************************************/
361	/* work around various missing functions */
362
363	#if !HAVE_PREADWRITE
364	# define pread aio_pread
365	# define pwrite aio_pwrite
366
367	/*
368	* make our pread/pwrite safe against themselves, but not against
369	* normal read/write by using a mutex. slows down execution a lot,
370	* but that's your problem, not mine.
371	*/
372	static pthread_mutex_t preadwritelock = PTHREAD_MUTEX_INITIALIZER;
373
374	static ssize_t
375	pread (int fd, void *buf, size_t count, off_t offset)
376	{
377	ssize_t res;
378	off_t ooffset;
379
380	pthread_mutex_lock (&preadwritelock);
381	ooffset = lseek (fd, 0, SEEK_CUR);
382	lseek (fd, offset, SEEK_SET);
383	res = read (fd, buf, count);
384	lseek (fd, ooffset, SEEK_SET);
385	pthread_mutex_unlock (&preadwritelock);
386
387	return res;
388	}
389
390	static ssize_t
391	pwrite (int fd, void *buf, size_t count, off_t offset)
392	{
393	ssize_t res;
394	off_t ooffset;
395
396	pthread_mutex_lock (&preadwritelock);
397	ooffset = lseek (fd, 0, SEEK_CUR);
398	lseek (fd, offset, SEEK_SET);
399	res = write (fd, buf, count);
400	lseek (fd, offset, SEEK_SET);
401	pthread_mutex_unlock (&preadwritelock);
402
403	return res;
404	}
405	#endif
406
407	#if !HAVE_FDATASYNC
408	# define fdatasync fsync
409	#endif
410
411	#if !HAVE_READAHEAD
412	# define readahead aio_readahead
413
414	static ssize_t
415	readahead (int fd, off_t offset, size_t count)
416	{
417	char readahead_buf[4096];
418
419	while (count > 0)
420	{
421	size_t len = count < sizeof (readahead_buf) ? count : sizeof (readahead_buf);
422
423	pread (fd, readahead_buf, len, offset);
424	offset += len;
425	count -= len;
426	}
427
428	errno = 0;
429	}
430	#endif
431
432	#if !HAVE_READDIR_R
433	# define readdir_r aio_readdir_r
434
435	static pthread_mutex_t readdirlock = PTHREAD_MUTEX_INITIALIZER;
436
437	static int
438	readdir_r (DIR dirp, struct dirent ent, struct dirent **res)
439	{
440	struct dirent *e;
441	int errorno;
442
443	pthread_mutex_lock (&readdirlock);
444
445	e = readdir (dirp);
446	errorno = errno;
447
448	if (e)
449	{
450	*res = ent;
451	strcpy (ent->d_name, e->d_name);
452	}
453	else
454	*res = 0;
455
456	pthread_mutex_unlock (&readdirlock);
457
458	errno = errorno;
459	return e ? 0 : -1;
460	}
461	#endif
462
463	/* sendfile always needs emulation */
464	static ssize_t
465	sendfile_ (int ofd, int ifd, off_t offset, size_t count)
466	{
467	ssize_t res;
468
469	if (!count)
470	return 0;
471
472	#if HAVE_SENDFILE
473	# if __linux
474	res = sendfile (ofd, ifd, &offset, count);
475
476	# elif __freebsd
477	/*
478	* Of course, the freebsd sendfile is a dire hack with no thoughts
479	* wasted on making it similar to other I/O functions.
480	*/
481	{
482	off_t sbytes;
483	res = sendfile (ifd, ofd, offset, count, 0, &sbytes, 0);
484
485	if (res < 0 && sbytes)
486	/* maybe only on EAGAIN only: as usual, the manpage leaves you guessing */
487	res = sbytes;
488	}
489
490	# elif __hpux
491	res = sendfile (ofd, ifd, offset, count, 0, 0);
492
493	# elif __solaris
494	{
495	struct sendfilevec vec;
496	size_t sbytes;
497
498	vec.sfv_fd = ifd;
499	vec.sfv_flag = 0;
500	vec.sfv_off = offset;
501	vec.sfv_len = count;
502
503	res = sendfilev (ofd, &vec, 1, &sbytes);
504
505	if (res < 0 && sbytes)
506	res = sbytes;
507	}
508
509	# endif
510	#else
511	res = -1;
512	errno = ENOSYS;
513	#endif
514
515	if (res < 0
516	&& (errno == ENOSYS \|\| errno == EINVAL \|\| errno == ENOTSOCK
517	#if __solaris
518	\|\| errno == EAFNOSUPPORT \|\| errno == EPROTOTYPE
519	#endif
520	)
521	)
522	{
523	/* emulate sendfile. this is a major pain in the ass */
524	char buf[4096];
525	res = 0;
526
527	while (count)
528	{
529	ssize_t cnt;
530
531	cnt = pread (ifd, buf, count > 4096 ? 4096 : count, offset);
532
533	if (cnt <= 0)
534	{
535	if (cnt && !res) res = -1;
536	break;
537	}
538
539	cnt = write (ofd, buf, cnt);
540
541	if (cnt <= 0)
542	{
543	if (cnt && !res) res = -1;
544	break;
545	}
546
547	offset += cnt;
548	res += cnt;
549	count -= cnt;
550	}
551	}
552
553	return res;
554	}
555
556	/* read a full directory */
557	static int
558	scandir_ (const char path, void *namesp)
559	{
560	DIR *dirp = opendir (path);
561	union
562	{
563	struct dirent d;
564	char b [offsetof (struct dirent, d_name) + NAME_MAX + 1];
565	} u;
566	struct dirent *entp;
567	char name, names;
568	int memlen = 4096;
569	int memofs = 0;
570	int res = 0;
571	int errorno;
572
573	if (!dirp)
574	return -1;
575
576	names = malloc (memlen);
577
578	for (;;)
579	{
580	errno = 0, readdir_r (dirp, &u.d, &entp);
581
582	if (!entp)
583	break;
584
585	name = entp->d_name;
586
587	if (name [0] != '.' \|\| (name [1] && (name [1] != '.' \|\| name [2])))
588	{
589	int len = strlen (name) + 1;
590
591	res++;
592
593	while (memofs + len > memlen)
594	{
595	memlen *= 2;
596	names = realloc (names, memlen);
597	if (!names)
598	break;
599	}
600
601	memcpy (names + memofs, name, len);
602	memofs += len;
603	}
604	}
605
606	errorno = errno;
607	closedir (dirp);
608
609	if (errorno)
610	{
611	free (names);
612	errno = errorno;
613	res = -1;
614	}
615
616	namesp = (void )names;
617	return res;
618	}
619
620	/*****************************************************************************/
621
622	static void *
623	aio_proc (void *thr_arg)
624	{
625	aio_req req;
626	int type;
627
628	do
629	{
630	pthread_mutex_lock (&reqlock);
631
632	for (;;)
633	{
634	req = reqs;
635
636	if (reqs)
637	{
638	reqs = reqs->next;
639	if (!reqs) reqe = 0;
640	}
641
642	if (req)
643	break;
644
645	pthread_cond_wait (&reqwait, &reqlock);
646	}
647
648	pthread_mutex_unlock (&reqlock);
649
650	errno = 0; /* strictly unnecessary */
651
652	type = req->type;
653
654	switch (type)
655	{
656	case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
657	case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
658
659	case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
660	case REQ_SENDFILE: req->result = sendfile_ (req->fd, req->fd2, req->offset, req->length); break;
661
662	case REQ_STAT: req->result = stat (req->dataptr, req->statdata); break;
663	case REQ_LSTAT: req->result = lstat (req->dataptr, req->statdata); break;
664	case REQ_FSTAT: req->result = fstat (req->fd , req->statdata); break;
665
666	case REQ_OPEN: req->result = open (req->dataptr, req->fd, req->mode); break;
667	case REQ_CLOSE: req->result = close (req->fd); break;
668	case REQ_UNLINK: req->result = unlink (req->dataptr); break;
669	case REQ_RMDIR: req->result = rmdir (req->dataptr); break;
670	case REQ_RENAME: req->result = rename (req->data2ptr, req->dataptr); break;
671	case REQ_LINK: req->result = link (req->data2ptr, req->dataptr); break;
672	case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break;
673
674	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
675	case REQ_FSYNC: req->result = fsync (req->fd); break;
676	case REQ_READDIR: req->result = scandir_ (req->dataptr, &req->data2ptr); break;
677
678	case REQ_QUIT:
679	break;
680
681	default:
682	req->result = ENOSYS;
683	break;
684	}
685
686	req->errorno = errno;
687
688	pthread_mutex_lock (&reslock);
689
690	req->next = 0;
691
692	if (rese)
693	{
694	rese->next = req;
695	rese = req;
696	}
697	else
698	{
699	rese = ress = req;
700
701	/* write a dummy byte to the pipe so fh becomes ready */
702	write (respipe [1], &respipe, 1);
703	}
704
705	pthread_mutex_unlock (&reslock);
706	}
707	while (type != REQ_QUIT);
708
709	return 0;
710	}
711
712	/*****************************************************************************/
713
714	static void atfork_prepare (void)
715	{
716	pthread_mutex_lock (&reqlock);
717	pthread_mutex_lock (&reslock);
718	#if !HAVE_PREADWRITE
719	pthread_mutex_lock (&preadwritelock);
720	#endif
721	#if !HAVE_READDIR_R
722	pthread_mutex_lock (&readdirlock);
723	#endif
724	}
725
726	static void atfork_parent (void)
727	{
728	#if !HAVE_READDIR_R
729	pthread_mutex_unlock (&readdirlock);
730	#endif
731	#if !HAVE_PREADWRITE
732	pthread_mutex_unlock (&preadwritelock);
733	#endif
734	pthread_mutex_unlock (&reslock);
735	pthread_mutex_unlock (&reqlock);
736	}
737
738	static void atfork_child (void)
739	{
740	aio_req prv;
741
742	started = 0;
743
744	while (reqs)
745	{
746	prv = reqs;
747	reqs = prv->next;
748	free_req (prv);
749	}
750
751	reqs = reqe = 0;
752
753	while (ress)
754	{
755	prv = ress;
756	ress = prv->next;
757	free_req (prv);
758	}
759
760	ress = rese = 0;
761
762	close (respipe [0]);
763	close (respipe [1]);
764	create_pipe ();
765
766	atfork_parent ();
767	}
768
769	#define dREQ \
770	aio_req req; \
771	\
772	if (SvOK (callback) && !SvROK (callback)) \
773	croak ("clalback must be undef or of reference type"); \
774	\
775	Newz (0, req, 1, aio_cb); \
776	if (!req) \
777	croak ("out of memory during aio_req allocation"); \
778	\
779	req->callback = newSVsv (callback);
780
781	MODULE = IO::AIO PACKAGE = IO::AIO
782
783	PROTOTYPES: ENABLE
784
785	BOOT:
786	{
787	create_pipe ();
788	pthread_atfork (atfork_prepare, atfork_parent, atfork_child);
789	}
790
791	void
792	min_parallel (nthreads)
793	int nthreads
794	PROTOTYPE: $
795
796	void
797	max_parallel (nthreads)
798	int nthreads
799	PROTOTYPE: $
800
801	int
802	max_outstanding (nreqs)
803	int nreqs
804	PROTOTYPE: $
805	CODE:
806	RETVAL = max_outstanding;
807	max_outstanding = nreqs;
808
809	void
810	aio_open (pathname,flags,mode,callback=&PL_sv_undef)
811	SV * pathname
812	int flags
813	int mode
814	SV * callback
815	PROTOTYPE: $$$;$
816	CODE:
817	{
818	dREQ;
819
820	req->type = REQ_OPEN;
821	req->data = newSVsv (pathname);
822	req->dataptr = SvPVbyte_nolen (req->data);
823	req->fd = flags;
824	req->mode = mode;
825
826	send_req (req);
827	}
828
829	void
830	aio_close (fh,callback=&PL_sv_undef)
831	SV * fh
832	SV * callback
833	PROTOTYPE: $;$
834	ALIAS:
835	aio_close = REQ_CLOSE
836	aio_fsync = REQ_FSYNC
837	aio_fdatasync = REQ_FDATASYNC
838	CODE:
839	{
840	dREQ;
841
842	req->type = ix;
843	req->fh = newSVsv (fh);
844	req->fd = PerlIO_fileno (IoIFP (sv_2io (fh)));
845
846	send_req (req);
847	}
848
849	void
850	aio_read (fh,offset,length,data,dataoffset,callback=&PL_sv_undef)
851	SV * fh
852	UV offset
853	UV length
854	SV * data
855	UV dataoffset
856	SV * callback
857	ALIAS:
858	aio_read = REQ_READ
859	aio_write = REQ_WRITE
860	PROTOTYPE: $$$$$;$
861	CODE:
862	{
863	aio_req req;
864	STRLEN svlen;
865	char *svptr = SvPVbyte (data, svlen);
866
867	SvUPGRADE (data, SVt_PV);
868	SvPOK_on (data);
869
870	if (dataoffset < 0)
871	dataoffset += svlen;
872
873	if (dataoffset < 0 \|\| dataoffset > svlen)
874	croak ("data offset outside of string");
875
876	if (ix == REQ_WRITE)
877	{
878	/* write: check length and adjust. */
879	if (length < 0 \|\| length + dataoffset > svlen)
880	length = svlen - dataoffset;
881	}
882	else
883	{
884	/* read: grow scalar as necessary */
885	svptr = SvGROW (data, length + dataoffset);
886	}
887
888	if (length < 0)
889	croak ("length must not be negative");
890
891	{
892	dREQ;
893
894	req->type = ix;
895	req->fh = newSVsv (fh);
896	req->fd = PerlIO_fileno (ix == REQ_READ ? IoIFP (sv_2io (fh))
897	: IoOFP (sv_2io (fh)));
898	req->offset = offset;
899	req->length = length;
900	req->data = SvREFCNT_inc (data);
901	req->dataptr = (char *)svptr + dataoffset;
902
903	if (!SvREADONLY (data))
904	{
905	SvREADONLY_on (data);
906	req->data2ptr = (void *)data;
907	}
908
909	send_req (req);
910	}
911	}
912
913	void
914	aio_sendfile (out_fh,in_fh,in_offset,length,callback=&PL_sv_undef)
915	SV * out_fh
916	SV * in_fh
917	UV in_offset
918	UV length
919	SV * callback
920	PROTOTYPE: $$$$;$
921	CODE:
922	{
923	dREQ;
924
925	req->type = REQ_SENDFILE;
926	req->fh = newSVsv (out_fh);
927	req->fd = PerlIO_fileno (IoIFP (sv_2io (out_fh)));
928	req->fh2 = newSVsv (in_fh);
929	req->fd2 = PerlIO_fileno (IoIFP (sv_2io (in_fh)));
930	req->offset = in_offset;
931	req->length = length;
932
933	send_req (req);
934	}
935
936	void
937	aio_readahead (fh,offset,length,callback=&PL_sv_undef)
938	SV * fh
939	UV offset
940	IV length
941	SV * callback
942	PROTOTYPE: $$$;$
943	CODE:
944	{
945	dREQ;
946
947	req->type = REQ_READAHEAD;
948	req->fh = newSVsv (fh);
949	req->fd = PerlIO_fileno (IoIFP (sv_2io (fh)));
950	req->offset = offset;
951	req->length = length;
952
953	send_req (req);
954	}
955
956	void
957	aio_stat (fh_or_path,callback=&PL_sv_undef)
958	SV * fh_or_path
959	SV * callback
960	ALIAS:
961	aio_stat = REQ_STAT
962	aio_lstat = REQ_LSTAT
963	CODE:
964	{
965	dREQ;
966
967	New (0, req->statdata, 1, Stat_t);
968	if (!req->statdata)
969	{
970	free_req (req);
971	croak ("out of memory during aio_req->statdata allocation");
972	}
973
974	if (SvPOK (fh_or_path))
975	{
976	req->type = ix;
977	req->data = newSVsv (fh_or_path);
978	req->dataptr = SvPVbyte_nolen (req->data);
979	}
980	else
981	{
982	req->type = REQ_FSTAT;
983	req->fh = newSVsv (fh_or_path);
984	req->fd = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
985	}
986
987	send_req (req);
988	}
989
990	void
991	aio_unlink (pathname,callback=&PL_sv_undef)
992	SV * pathname
993	SV * callback
994	ALIAS:
995	aio_unlink = REQ_UNLINK
996	aio_rmdir = REQ_RMDIR
997	aio_readdir = REQ_READDIR
998	CODE:
999	{
1000	dREQ;
1001
1002	req->type = ix;
1003	req->data = newSVsv (pathname);
1004	req->dataptr = SvPVbyte_nolen (req->data);
1005
1006	send_req (req);
1007	}
1008
1009	void
1010	aio_link (oldpath,newpath,callback=&PL_sv_undef)
1011	SV * oldpath
1012	SV * newpath
1013	SV * callback
1014	ALIAS:
1015	aio_link = REQ_LINK
1016	aio_symlink = REQ_SYMLINK
1017	aio_rename = REQ_RENAME
1018	CODE:
1019	{
1020	dREQ;
1021
1022	req->type = ix;
1023	req->fh = newSVsv (oldpath);
1024	req->data2ptr = SvPVbyte_nolen (req->fh);
1025	req->data = newSVsv (newpath);
1026	req->dataptr = SvPVbyte_nolen (req->data);
1027
1028	send_req (req);
1029	}
1030
1031	void
1032	flush ()
1033	PROTOTYPE:
1034	CODE:
1035	while (nreqs)
1036	{
1037	poll_wait ();
1038	poll_cb ();
1039	}
1040
1041	void
1042	poll()
1043	PROTOTYPE:
1044	CODE:
1045	if (nreqs)
1046	{
1047	poll_wait ();
1048	poll_cb ();
1049	}
1050
1051	int
1052	poll_fileno()
1053	PROTOTYPE:
1054	CODE:
1055	RETVAL = respipe [0];
1056	OUTPUT:
1057	RETVAL
1058
1059	int
1060	poll_cb(...)
1061	PROTOTYPE:
1062	CODE:
1063	RETVAL = poll_cb ();
1064	OUTPUT:
1065	RETVAL
1066
1067	void
1068	poll_wait()
1069	PROTOTYPE:
1070	CODE:
1071	if (nreqs)
1072	poll_wait ();
1073
1074	int
1075	nreqs()
1076	PROTOTYPE:
1077	CODE:
1078	RETVAL = nreqs;
1079	OUTPUT:
1080	RETVAL
1081