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 2048 /* 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 = (char *)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->dataoffset
                                  + req->result > 0 ? req->result : 0);

          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
        UV              offset
        STRLEN          length
        SV *            data
        STRLEN          dataoffset
        SV *            callback
        PROTOTYPE: $$$$$$
        ALIAS:
          aio_write = 1
        CODE:
        SvUPGRADE (data, SVt_PV);
        SvPOK_on (data);
        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.6
Committed:	Tue Aug 14 23:25:39 2001 UTC (22 years, 9 months ago) by root
Branch:	MAIN
Changes since 1.5:	+9 -9 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 2048 /* 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 = (char *)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->dataoffset
136	+ req->result > 0 ? req->result : 0);
137
138	PUSHMARK (SP);
139	XPUSHs (sv_2mortal (newSViv (req->result)));
140	PUTBACK;
141	call_sv (req->callback, G_VOID);
142	SPAGAIN;
143
144	SvREFCNT_dec (req->data);
145	SvREFCNT_dec (req->callback);
146
147	errno = errorno;
148	nreqs--;
149	count++;
150	}
151
152	Safefree (req);
153	}
154
155	return count;
156	}
157
158	#undef errno
159	#include <asm/unistd.h>
160
161	static int
162	aio_proc(void *thr_arg)
163	{
164	aio_thread *thr = thr_arg;
165	int sig;
166	int errno;
167	aio_req req;
168
169	/* we rely on gcc's ability to create closures. */
170	_syscall3(int,lseek,int,fd,off_t,offset,int,whence);
171	_syscall3(int,read,int,fd,char *,buf,off_t,count);
172	_syscall3(int,write,int,fd,char *,buf,off_t,count);
173
174	/* first get rid of any signals */
175	for (sig = 1; sig < _NSIG; sig++)
176	signal (sig, SIG_DFL);
177
178	signal (SIGPIPE, SIG_IGN);
179
180	/* then loop */
181	while (read (reqpipe[0], (void *)&req, sizeof (req)) == sizeof (req))
182	{
183	req->thread = thr;
184
185	if (req->type == REQ_READ \|\| req->type == REQ_WRITE)
186	{
187	errno = 0;
188
189	if (lseek (req->fd, req->offset, SEEK_SET) == req->offset)
190	{
191	if (req->type == REQ_READ)
192	req->result = read (req->fd, req->dataptr, req->length);
193	else
194	req->result = write(req->fd, req->dataptr, req->length);
195	}
196
197	req->errorno = errno;
198	}
199	else
200	{
201	write (respipe[1], (void *)&req, sizeof (req));
202	break;
203	}
204
205	write (respipe[1], (void *)&req, sizeof (req));
206	}
207
208	return 0;
209	}
210
211	MODULE = Linux::AIO PACKAGE = Linux::AIO
212
213	BOOT:
214	{
215	if (pipe (reqpipe) \|\| pipe (respipe))
216	croak ("unable to initialize request or result pipe");
217
218	if (fcntl (respipe[0], F_SETFL, O_NONBLOCK))
219	croak ("cannot set result pipe to nonblocking mode");
220	}
221
222	void
223	min_parallel(nthreads)
224	int nthreads
225	PROTOTYPE: $
226	CODE:
227	while (nthreads > started)
228	start_thread ();
229
230	void
231	max_parallel(nthreads)
232	int nthreads
233	PROTOTYPE: $
234	CODE:
235	int cur = started;
236	while (cur > nthreads)
237	{
238	end_thread ();
239	cur--;
240	}
241
242	poll_cb ();
243	while (started > nthreads)
244	{
245	sched_yield ();
246	poll_cb ();
247	}
248
249	void
250	aio_read(fh,offset,length,data,dataoffset,callback)
251	PerlIO * fh
252	UV offset
253	STRLEN length
254	SV * data
255	STRLEN dataoffset
256	SV * callback
257	PROTOTYPE: $$$$$$
258	ALIAS:
259	aio_write = 1
260	CODE:
261	SvUPGRADE (data, SVt_PV);
262	SvPOK_on (data);
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