Linux-AIO/AIO.xs

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

#include <sys/types.h>
#include <unistd.h>
#include <sched.h>

#define STACKSIZE 128 /* yeah */

#define REQ_QUIT  0
#define REQ_READ  1
#define REQ_WRITE 2

typedef struct {
  char stack[STACKSIZE];
} aio_thread;

typedef struct {
  int type;
  aio_thread *thread;

/* read/write */
  int fd;
  off_t offset;
  size_t length;
  ssize_t result;
  int errorno;

  SV *data;
  void *dataptr;
  STRLEN dataoffset;
} aio_cb;

typedef aio_cb *aio_req;

static int started;
static int reqpipe[2], respipe[2];

static int aio_proc(void *arg);

static void
start_thread(void)
{
  aio_thread *thr;

  New (0, thr, 1, aio_thread);

  if (clone (aio_proc,
             &(thr->stack[STACKSIZE]),
             CLONE_VM|CLONE_FS|CLONE_FILES|CLONE_SIGHAND,
             thr) >= 0)
    started++;
  else
    Safefree (thr);
}

static void
end_thread(void)
{
  aio_req req = 0;
  write (reqpipe[1], &req, sizeof (aio_req));
  started--;
}

static void
set_errno(int errorno)
{
  errno = errorno;
}

#undef errno
#include <asm/unistd.h>

static int
aio_proc(void *thr_arg)
{
  aio_thread *thr = thr_arg;
  int sig;
  int errno;
  aio_req req;

  /* we rely on gcc's ability to create closures. */
  _syscall3(int,lseek,int,fd,off_t,offset,int,whence);
  _syscall3(int,read,int,fd,char *,buf,off_t,count);
  _syscall3(int,write,int,fd,char *,buf,off_t,count);

  /* first get rid of any signals */
  for (sig = 1; sig < _NSIG; sig++)
    if (sig != SIGTERM)
      signal (sig, SIG_IGN);
 
  /* then loop */
  while (read (reqpipe[0], (void *)&req, sizeof (req)) == sizeof (req))
    {
      req->thread = thr;

      if (req->type == REQ_READ || req->type == REQ_WRITE)
        {
          errno = 0;

          if (lseek (req->fd, req->offset, SEEK_SET) == req->offset)
            {
              if (req->type == REQ_READ)
                req->result = read (req->fd, req->dataptr, req->length);
              else
                req->result = write(req->fd, req->dataptr, req->length);
            }

          req->errorno = errno;
        }
      else
        {
          write (respipe[1], (void *)&req, sizeof (req));
          break;
        }

      write (respipe[1], (void *)&req, sizeof (req));
    }

  return 0;
}

MODULE = Linux::AIO                PACKAGE = Linux::AIO

BOOT:
{
        if (pipe (reqpipe) || pipe (respipe))
          croak ("unable to initialize request or result pipe");
}

void
min_parallel(nthreads)
        int     nthreads
        CODE:
        while (nthreads > started)
          start_thread ();

void
max_parallel(nthreads)
        int     nthreads
        CODE:
        while (started > nthreads)
          end_thread ();

void
read(fh,offset,length,data,dataoffset,callback)
        CODE:
Revision:	1.2
Committed:	Tue Aug 14 04:32:19 2001 UTC (22 years, 9 months ago) by root
Branch:	MAIN
Changes since 1.1:	+75 -6 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	#include "EXTERN.h"
2			#include "perl.h"
3			#include "XSUB.h"
4
5			#include <sys/types.h>
6			#include <unistd.h>
7			#include <sched.h>
8
9			#define STACKSIZE 128 /* yeah */
10
11	root	1.2	#define REQ_QUIT 0
12	root	1.1	#define REQ_READ 1
13			#define REQ_WRITE 2
14
15			typedef struct {
16			char stack[STACKSIZE];
17			} aio_thread;
18
19			typedef struct {
20			int type;
21			aio_thread *thread;
22
23			/* read/write */
24			int fd;
25			off_t offset;
26			size_t length;
27	root	1.2	ssize_t result;
28	root	1.1	int errorno;
29
30			SV *data;
31			void *dataptr;
32	root	1.2	STRLEN dataoffset;
33	root	1.1	} aio_cb;
34
35			typedef aio_cb *aio_req;
36
37			static int started;
38			static int reqpipe[2], respipe[2];
39
40	root	1.2	static int aio_proc(void *arg);
41
42	root	1.1	static void
43			start_thread(void)
44			{
45	root	1.2	aio_thread *thr;
46
47			New (0, thr, 1, aio_thread);
48
49			if (clone (aio_proc,
50			&(thr->stack[STACKSIZE]),
51			CLONE_VM\|CLONE_FS\|CLONE_FILES\|CLONE_SIGHAND,
52			thr) >= 0)
53			started++;
54			else
55			Safefree (thr);
56	root	1.1	}
57
58			static void
59			end_thread(void)
60			{
61			aio_req req = 0;
62			write (reqpipe[1], &req, sizeof (aio_req));
63	root	1.2	started--;
64			}
65
66			static void
67			set_errno(int errorno)
68			{
69			errno = errorno;
70			}
71
72			#undef errno
73			#include <asm/unistd.h>
74
75			static int
76			aio_proc(void *thr_arg)
77			{
78			aio_thread *thr = thr_arg;
79			int sig;
80			int errno;
81			aio_req req;
82
83			/* we rely on gcc's ability to create closures. */
84			_syscall3(int,lseek,int,fd,off_t,offset,int,whence);
85			_syscall3(int,read,int,fd,char *,buf,off_t,count);
86			_syscall3(int,write,int,fd,char *,buf,off_t,count);
87
88			/* first get rid of any signals */
89			for (sig = 1; sig < _NSIG; sig++)
90			if (sig != SIGTERM)
91			signal (sig, SIG_IGN);
92
93			/* then loop */
94			while (read (reqpipe[0], (void *)&req, sizeof (req)) == sizeof (req))
95			{
96			req->thread = thr;
97
98			if (req->type == REQ_READ \|\| req->type == REQ_WRITE)
99			{
100			errno = 0;
101
102			if (lseek (req->fd, req->offset, SEEK_SET) == req->offset)
103			{
104			if (req->type == REQ_READ)
105			req->result = read (req->fd, req->dataptr, req->length);
106			else
107			req->result = write(req->fd, req->dataptr, req->length);
108			}
109
110			req->errorno = errno;
111			}
112			else
113			{
114			write (respipe[1], (void *)&req, sizeof (req));
115			break;
116			}
117
118			write (respipe[1], (void *)&req, sizeof (req));
119			}
120
121			return 0;
122	root	1.1	}
123
124			MODULE = Linux::AIO PACKAGE = Linux::AIO
125
126			BOOT:
127			{
128			if (pipe (reqpipe) \|\| pipe (respipe))
129			croak ("unable to initialize request or result pipe");
130			}
131
132			void
133			min_parallel(nthreads)
134			int nthreads
135			CODE:
136			while (nthreads > started)
137			start_thread ();
138
139			void
140			max_parallel(nthreads)
141			int nthreads
142			CODE:
143			while (started > nthreads)
144			end_thread ();
145
146			void
147			read(fh,offset,length,data,dataoffset,callback)
148			CODE: