Linux-AIO/AIO.xs

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

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

#define STACKSIZE 1024 /* 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, *callback;
  void *dataptr;
  STRLEN dataoffset;
} aio_cb;

typedef aio_cb *aio_req;

static int started;
static int nreqs;
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;
  New (0, req, 1, aio_cb);
  req->type = REQ_QUIT;
  write (reqpipe[1], &req, sizeof (aio_req));
}

static void
read_write (pTHX_ int dowrite, int fd, off_t offset, size_t length,
            SV *data, STRLEN dataoffset, SV*callback)
{
  aio_req req;
  STRLEN svlen;
  char *svptr = SvPV (data, svlen);

  if (dataoffset < 0)
    dataoffset += svlen;

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

  if (dowrite)
    {
      /* 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");

  New (0, req, 1, aio_cb);

  if (!req)
    croak ("out of memory during aio_req allocation");

  req->type = dowrite ? REQ_WRITE : REQ_READ;
  req->fd = fd;
  req->offset = offset;
  req->length = length;
  req->data = SvREFCNT_inc (data);
  req->dataptr = svptr + dataoffset;
  req->callback = SvREFCNT_inc (callback);

  nreqs++;
  write (reqpipe[1], &req, sizeof (aio_req));
}

static int
poll_cb (pTHX)
{
  dSP;
  int count = 0;
  aio_req req;

  while (read (respipe[0], (void *)&req, sizeof (req)) == sizeof (req))
    {
      if (req->type == REQ_QUIT)
        {
          Safefree (req->thread);
          started--;
        }
      else
        {
          int errorno = errno;
          errno = req->errorno;

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

          PUSHMARK (SP);
          XPUSHs (sv_2mortal (newSViv (req->result)));
          PUTBACK;
          call_sv (req->callback, G_VOID);
          SPAGAIN;
          
          SvREFCNT_dec (req->data);
          SvREFCNT_dec (req->callback);

          errno = errorno;
          nreqs--;
          count++;
        }

      Safefree (req);
    }

  return count;
}

#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++)
    signal (sig, SIG_DFL);

  signal (SIGPIPE, 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");

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

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

void
max_parallel(nthreads)
        int     nthreads
        PROTOTYPE: $
        CODE:
        int cur = started;
        while (cur > nthreads)
          {          
            end_thread ();
            cur--;
          }

        poll_cb ();
        while (started > nthreads)
          {
            sched_yield ();
            poll_cb ();
          }

void
aio_read(fh,offset,length,data,dataoffset,callback)
        PerlIO *        fh
        unsigned long   offset
        long            length
        SV *            data
        STRLEN          dataoffset
        SV *            callback
        PROTOTYPE: $$$$$$
        ALIAS:
          aio_write = 1
        CODE:
        sv_upgrade (data, SVt_PV);
        read_write (aTHX_ ix, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);

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

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

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

Revision:	1.5
Committed:	Tue Aug 14 18:06:37 2001 UTC (22 years, 9 months ago) by root
Branch:	MAIN
Changes since 1.4:	+148 -11 lines
Log Message:	* empty log message *
#	Content
1	#include "EXTERN.h"
2	#include "perl.h"
3	#include "XSUB.h"
4
5	#include <sys/types.h>
6	#include <unistd.h>
7	#include <fcntl.h>
8	#include <sched.h>
9
10	#define STACKSIZE 1024 /* yeah */
11
12	#define REQ_QUIT 0
13	#define REQ_READ 1
14	#define REQ_WRITE 2
15
16	typedef struct {
17	char stack[STACKSIZE];
18	} aio_thread;
19
20	typedef struct {
21	int type;
22	aio_thread *thread;
23
24	/* read/write */
25	int fd;
26	off_t offset;
27	size_t length;
28	ssize_t result;
29	int errorno;
30
31	SV data, callback;
32	void *dataptr;
33	STRLEN dataoffset;
34	} aio_cb;
35
36	typedef aio_cb *aio_req;
37
38	static int started;
39	static int nreqs;
40	static int reqpipe[2], respipe[2];
41
42	static int aio_proc(void *arg);
43
44	static void
45	start_thread(void)
46	{
47	aio_thread *thr;
48
49	New (0, thr, 1, aio_thread);
50
51	if (clone (aio_proc,
52	&(thr->stack[STACKSIZE]),
53	CLONE_VM\|CLONE_FS\|CLONE_FILES\|CLONE_SIGHAND,
54	thr) >= 0)
55	started++;
56	else
57	Safefree (thr);
58	}
59
60	static void
61	end_thread(void)
62	{
63	aio_req req;
64	New (0, req, 1, aio_cb);
65	req->type = REQ_QUIT;
66	write (reqpipe[1], &req, sizeof (aio_req));
67	}
68
69	static void
70	read_write (pTHX_ int dowrite, int fd, off_t offset, size_t length,
71	SV data, STRLEN dataoffset, SVcallback)
72	{
73	aio_req req;
74	STRLEN svlen;
75	char *svptr = SvPV (data, svlen);
76
77	if (dataoffset < 0)
78	dataoffset += svlen;
79
80	if (dataoffset < 0 \|\| dataoffset > svlen)
81	croak ("data offset outside of string");
82
83	if (dowrite)
84	{
85	/* write: check length and adjust. */
86	if (length < 0 \|\| length + dataoffset > svlen)
87	length = svlen - dataoffset;
88	}
89	else
90	{
91	/* read: grow scalar as necessary */
92	svptr = SvGROW (data, length + dataoffset);
93	}
94
95	if (length < 0)
96	croak ("length must not be negative");
97
98	New (0, req, 1, aio_cb);
99
100	if (!req)
101	croak ("out of memory during aio_req allocation");
102
103	req->type = dowrite ? REQ_WRITE : REQ_READ;
104	req->fd = fd;
105	req->offset = offset;
106	req->length = length;
107	req->data = SvREFCNT_inc (data);
108	req->dataptr = svptr + dataoffset;
109	req->callback = SvREFCNT_inc (callback);
110
111	nreqs++;
112	write (reqpipe[1], &req, sizeof (aio_req));
113	}
114
115	static int
116	poll_cb (pTHX)
117	{
118	dSP;
119	int count = 0;
120	aio_req req;
121
122	while (read (respipe[0], (void *)&req, sizeof (req)) == sizeof (req))
123	{
124	if (req->type == REQ_QUIT)
125	{
126	Safefree (req->thread);
127	started--;
128	}
129	else
130	{
131	int errorno = errno;
132	errno = req->errorno;
133
134	if (req->type == REQ_READ)
135	SvCUR_set (req->data, req->result > 0
136	? req->dataoffset + req->result
137	: req->dataoffset);
138
139	PUSHMARK (SP);
140	XPUSHs (sv_2mortal (newSViv (req->result)));
141	PUTBACK;
142	call_sv (req->callback, G_VOID);
143	SPAGAIN;
144
145	SvREFCNT_dec (req->data);
146	SvREFCNT_dec (req->callback);
147
148	errno = errorno;
149	nreqs--;
150	count++;
151	}
152
153	Safefree (req);
154	}
155
156	return count;
157	}
158
159	#undef errno
160	#include <asm/unistd.h>
161
162	static int
163	aio_proc(void *thr_arg)
164	{
165	aio_thread *thr = thr_arg;
166	int sig;
167	int errno;
168	aio_req req;
169
170	/* we rely on gcc's ability to create closures. */
171	_syscall3(int,lseek,int,fd,off_t,offset,int,whence);
172	_syscall3(int,read,int,fd,char *,buf,off_t,count);
173	_syscall3(int,write,int,fd,char *,buf,off_t,count);
174
175	/* first get rid of any signals */
176	for (sig = 1; sig < _NSIG; sig++)
177	signal (sig, SIG_DFL);
178
179	signal (SIGPIPE, SIG_IGN);
180
181	/* then loop */
182	while (read (reqpipe[0], (void *)&req, sizeof (req)) == sizeof (req))
183	{
184	req->thread = thr;
185
186	if (req->type == REQ_READ \|\| req->type == REQ_WRITE)
187	{
188	errno = 0;
189
190	if (lseek (req->fd, req->offset, SEEK_SET) == req->offset)
191	{
192	if (req->type == REQ_READ)
193	req->result = read (req->fd, req->dataptr, req->length);
194	else
195	req->result = write(req->fd, req->dataptr, req->length);
196	}
197
198	req->errorno = errno;
199	}
200	else
201	{
202	write (respipe[1], (void *)&req, sizeof (req));
203	break;
204	}
205
206	write (respipe[1], (void *)&req, sizeof (req));
207	}
208
209	return 0;
210	}
211
212	MODULE = Linux::AIO PACKAGE = Linux::AIO
213
214	BOOT:
215	{
216	if (pipe (reqpipe) \|\| pipe (respipe))
217	croak ("unable to initialize request or result pipe");
218
219	if (fcntl (respipe[0], F_SETFL, O_NONBLOCK))
220	croak ("cannot set result pipe to nonblocking mode");
221	}
222
223	void
224	min_parallel(nthreads)
225	int nthreads
226	PROTOTYPE: $
227	CODE:
228	while (nthreads > started)
229	start_thread ();
230
231	void
232	max_parallel(nthreads)
233	int nthreads
234	PROTOTYPE: $
235	CODE:
236	int cur = started;
237	while (cur > nthreads)
238	{
239	end_thread ();
240	cur--;
241	}
242
243	poll_cb ();
244	while (started > nthreads)
245	{
246	sched_yield ();
247	poll_cb ();
248	}
249
250	void
251	aio_read(fh,offset,length,data,dataoffset,callback)
252	PerlIO * fh
253	unsigned long offset
254	long length
255	SV * data
256	STRLEN dataoffset
257	SV * callback
258	PROTOTYPE: $$$$$$
259	ALIAS:
260	aio_write = 1
261	CODE:
262	sv_upgrade (data, SVt_PV);
263	read_write (aTHX_ ix, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
264
265	int
266	poll_fileno()
267	PROTOTYPE:
268	CODE:
269	RETVAL = respipe[0];
270	OUTPUT:
271	RETVAL
272
273	int
274	poll_cb()
275	PROTOTYPE:
276	CODE:
277	RETVAL = poll_cb (aTHX);
278	OUTPUT:
279	RETVAL
280
281	int
282	nreqs()
283	PROTOTYPE:
284	CODE:
285	RETVAL = nreqs;
286	OUTPUT:
287	RETVAL
288