ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/Linux-AIO/AIO.xs
Revision: 1.15
Committed: Sat May 18 21:48:36 2002 UTC (22 years ago) by root
Branch: MAIN
Changes since 1.14: +5 -4 lines
Log Message:
*** empty log message ***

File Contents

# 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 root 1.13 #include <sys/stat.h>
7 root 1.1 #include <unistd.h>
8 root 1.5 #include <fcntl.h>
9 root 1.8 #include <signal.h>
10 root 1.1 #include <sched.h>
11    
12 root 1.13 typedef void *InputStream; /* hack, but 5.6.1 is simply toooo old ;) */
13     typedef void *OutputStream; /* hack, but 5.6.1 is simply toooo old ;) */
14 root 1.12 typedef void *InOutStream; /* hack, but 5.6.1 is simply toooo old ;) */
15    
16 root 1.8 #define STACKSIZE 1024 /* yeah */
17 root 1.1
18 root 1.13 enum { REQ_QUIT, REQ_OPEN, REQ_CLOSE, REQ_READ, REQ_WRITE, REQ_STAT, REQ_LSTAT, REQ_FSTAT};
19 root 1.1
20     typedef struct {
21     char stack[STACKSIZE];
22     } aio_thread;
23    
24     typedef struct {
25     int type;
26     aio_thread *thread;
27    
28     int fd;
29     off_t offset;
30     size_t length;
31 root 1.2 ssize_t result;
32 root 1.8 mode_t mode; /* open */
33 root 1.1 int errorno;
34 root 1.5 SV *data, *callback;
35 root 1.1 void *dataptr;
36 root 1.2 STRLEN dataoffset;
37 root 1.13
38     struct stat64 *statdata;
39 root 1.1 } aio_cb;
40    
41     typedef aio_cb *aio_req;
42    
43     static int started;
44 root 1.5 static int nreqs;
45 root 1.1 static int reqpipe[2], respipe[2];
46    
47 root 1.2 static int aio_proc(void *arg);
48    
49 root 1.1 static void
50     start_thread(void)
51     {
52 root 1.2 aio_thread *thr;
53    
54     New (0, thr, 1, aio_thread);
55    
56     if (clone (aio_proc,
57     &(thr->stack[STACKSIZE]),
58     CLONE_VM|CLONE_FS|CLONE_FILES|CLONE_SIGHAND,
59     thr) >= 0)
60     started++;
61     else
62     Safefree (thr);
63 root 1.1 }
64    
65     static void
66     end_thread(void)
67     {
68 root 1.5 aio_req req;
69     New (0, req, 1, aio_cb);
70     req->type = REQ_QUIT;
71 root 1.1 write (reqpipe[1], &req, sizeof (aio_req));
72 root 1.2 }
73    
74     static void
75 root 1.8 send_req (aio_req req)
76     {
77     nreqs++;
78     write (reqpipe[1], &req, sizeof (aio_req));
79     }
80    
81     static void
82 root 1.5 read_write (pTHX_ int dowrite, int fd, off_t offset, size_t length,
83     SV *data, STRLEN dataoffset, SV*callback)
84 root 1.2 {
85 root 1.5 aio_req req;
86     STRLEN svlen;
87     char *svptr = SvPV (data, svlen);
88    
89 root 1.13 SvUPGRADE (data, SVt_PV);
90     SvPOK_on (data);
91    
92 root 1.5 if (dataoffset < 0)
93     dataoffset += svlen;
94    
95     if (dataoffset < 0 || dataoffset > svlen)
96     croak ("data offset outside of string");
97    
98     if (dowrite)
99     {
100     /* write: check length and adjust. */
101     if (length < 0 || length + dataoffset > svlen)
102     length = svlen - dataoffset;
103     }
104     else
105     {
106     /* read: grow scalar as necessary */
107     svptr = SvGROW (data, length + dataoffset);
108     }
109    
110     if (length < 0)
111     croak ("length must not be negative");
112    
113 root 1.14 Newz (0, req, 1, aio_cb);
114 root 1.5
115     if (!req)
116     croak ("out of memory during aio_req allocation");
117    
118     req->type = dowrite ? REQ_WRITE : REQ_READ;
119     req->fd = fd;
120     req->offset = offset;
121     req->length = length;
122     req->data = SvREFCNT_inc (data);
123 root 1.6 req->dataptr = (char *)svptr + dataoffset;
124 root 1.5 req->callback = SvREFCNT_inc (callback);
125    
126 root 1.8 send_req (req);
127 root 1.5 }
128    
129     static int
130     poll_cb (pTHX)
131     {
132     dSP;
133     int count = 0;
134     aio_req req;
135    
136     while (read (respipe[0], (void *)&req, sizeof (req)) == sizeof (req))
137     {
138     if (req->type == REQ_QUIT)
139     {
140     Safefree (req->thread);
141     started--;
142     }
143     else
144     {
145     int errorno = errno;
146     errno = req->errorno;
147    
148     if (req->type == REQ_READ)
149 root 1.6 SvCUR_set (req->data, req->dataoffset
150     + req->result > 0 ? req->result : 0);
151 root 1.5
152 root 1.14 if (req->data)
153     SvREFCNT_dec (req->data);
154    
155 root 1.13 if (req->type == REQ_STAT || req->type == REQ_LSTAT || req->type == REQ_FSTAT)
156     {
157     PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
158     PL_laststatval = req->result;
159     PL_statcache.st_dev = req->statdata->st_dev;
160     PL_statcache.st_ino = req->statdata->st_ino;
161     PL_statcache.st_mode = req->statdata->st_mode;
162     PL_statcache.st_nlink = req->statdata->st_nlink;
163     PL_statcache.st_uid = req->statdata->st_uid;
164     PL_statcache.st_gid = req->statdata->st_gid;
165     PL_statcache.st_rdev = req->statdata->st_rdev;
166     PL_statcache.st_size = req->statdata->st_size;
167     PL_statcache.st_atime = req->statdata->st_atime;
168     PL_statcache.st_mtime = req->statdata->st_mtime;
169     PL_statcache.st_ctime = req->statdata->st_ctime;
170     PL_statcache.st_blksize = req->statdata->st_blksize;
171     PL_statcache.st_blocks = req->statdata->st_blocks;
172    
173     Safefree (req->statdata);
174     }
175    
176 root 1.5 PUSHMARK (SP);
177     XPUSHs (sv_2mortal (newSViv (req->result)));
178     PUTBACK;
179     call_sv (req->callback, G_VOID);
180     SPAGAIN;
181    
182 root 1.14 if (req->callback)
183     SvREFCNT_dec (req->callback);
184 root 1.5
185     errno = errorno;
186     nreqs--;
187     count++;
188     }
189    
190     Safefree (req);
191     }
192    
193     return count;
194 root 1.2 }
195    
196 root 1.8 static sigset_t fullsigset;
197    
198 root 1.2 #undef errno
199     #include <asm/unistd.h>
200    
201     static int
202     aio_proc(void *thr_arg)
203     {
204     aio_thread *thr = thr_arg;
205 root 1.9 aio_req req;
206 root 1.2 int errno;
207    
208 root 1.15 /* this is very much x86 and kernel-specific :(:(:( */
209 root 1.11 /* we rely on gcc's ability to create closures. */
210 root 1.13 _syscall3(int,read,int,fd,char *,buf,size_t,count)
211     _syscall3(int,write,int,fd,char *,buf,size_t,count)
212    
213 root 1.11 _syscall3(int,open,char *,pathname,int,flags,mode_t,mode)
214     _syscall1(int,close,int,fd)
215    
216 root 1.15 _syscall5(int,pread,int,fd,char *,buf,size_t,count,unsigned int,offset_lo,unsigned int,offset_hi)
217     _syscall5(int,pwrite,int,fd,char *,buf,size_t,count,unsigned int,offset_lo,unsigned int,offset_hi)
218 root 1.13
219     _syscall2(int,stat64, const char *, filename, struct stat64 *, buf)
220     _syscall2(int,lstat64, const char *, filename, struct stat64 *, buf)
221     _syscall2(int,fstat64, int, fd, struct stat64 *, buf)
222    
223 root 1.8 sigprocmask (SIG_SETMASK, &fullsigset, 0);
224 root 1.2
225     /* then loop */
226     while (read (reqpipe[0], (void *)&req, sizeof (req)) == sizeof (req))
227     {
228     req->thread = thr;
229 root 1.14 errno = 0; /* strictly unnecessary */
230 root 1.2
231 root 1.13 if (req->type == REQ_READ)
232 root 1.15 req->result = pread (req->fd, req->dataptr, req->length, req->offset & 0xffffffff, req->offset >> 32);
233 root 1.13 else if (req->type == REQ_WRITE)
234 root 1.15 req->result = pwrite(req->fd, req->dataptr, req->length, req->offset & 0xffffffff, req->offset >> 32);
235 root 1.8 else if (req->type == REQ_OPEN)
236 root 1.13 req->result = open (req->dataptr, req->fd, req->mode);
237 root 1.10 else if (req->type == REQ_CLOSE)
238 root 1.13 req->result = close (req->fd);
239     else if (req->type == REQ_STAT)
240     req->result = stat64 (req->dataptr, req->statdata);
241     else if (req->type == REQ_LSTAT)
242     req->result = lstat64 (req->dataptr, req->statdata);
243     else if (req->type == REQ_FSTAT)
244     req->result = fstat64 (req->fd, req->statdata);
245 root 1.2 else
246     {
247     write (respipe[1], (void *)&req, sizeof (req));
248     break;
249     }
250    
251 root 1.8 req->errorno = errno;
252 root 1.2 write (respipe[1], (void *)&req, sizeof (req));
253     }
254    
255     return 0;
256 root 1.1 }
257    
258     MODULE = Linux::AIO PACKAGE = Linux::AIO
259    
260     BOOT:
261     {
262 root 1.8 sigfillset (&fullsigset);
263 root 1.9 sigdelset (&fullsigset, SIGTERM);
264     sigdelset (&fullsigset, SIGQUIT);
265     sigdelset (&fullsigset, SIGABRT);
266     sigdelset (&fullsigset, SIGINT);
267 root 1.8
268 root 1.1 if (pipe (reqpipe) || pipe (respipe))
269     croak ("unable to initialize request or result pipe");
270 root 1.5
271     if (fcntl (respipe[0], F_SETFL, O_NONBLOCK))
272     croak ("cannot set result pipe to nonblocking mode");
273 root 1.1 }
274    
275     void
276     min_parallel(nthreads)
277     int nthreads
278 root 1.5 PROTOTYPE: $
279 root 1.1 CODE:
280     while (nthreads > started)
281     start_thread ();
282    
283     void
284     max_parallel(nthreads)
285     int nthreads
286 root 1.5 PROTOTYPE: $
287 root 1.1 CODE:
288 root 1.5 int cur = started;
289     while (cur > nthreads)
290     {
291     end_thread ();
292     cur--;
293     }
294    
295 root 1.1 while (started > nthreads)
296 root 1.5 {
297 root 1.14 fd_set rfd;
298     FD_ZERO(&rfd);
299     FD_SET(respipe[0], &rfd);
300    
301     select (respipe[0] + 1, &rfd, 0, 0, 0);
302 root 1.5 poll_cb ();
303     }
304 root 1.1
305     void
306 root 1.8 aio_open(pathname,flags,mode,callback)
307 root 1.13 SV * pathname
308 root 1.8 int flags
309     int mode
310     SV * callback
311     PROTOTYPE: $$$$
312     CODE:
313     aio_req req;
314    
315 root 1.14 Newz (0, req, 1, aio_cb);
316 root 1.8
317     if (!req)
318     croak ("out of memory during aio_req allocation");
319    
320     req->type = REQ_OPEN;
321 root 1.14 req->data = newSVsv (pathname);
322     req->dataptr = SvPV_nolen (req->data);
323 root 1.8 req->fd = flags;
324     req->mode = mode;
325 root 1.10 req->callback = SvREFCNT_inc (callback);
326    
327     send_req (req);
328    
329     void
330     aio_close(fh,callback)
331 root 1.13 InputStream fh
332 root 1.10 SV * callback
333 root 1.14 PROTOTYPE: $$
334 root 1.10 CODE:
335     aio_req req;
336    
337 root 1.14 Newz (0, req, 1, aio_cb);
338 root 1.10
339     if (!req)
340     croak ("out of memory during aio_req allocation");
341    
342     req->type = REQ_CLOSE;
343     req->fd = PerlIO_fileno (fh);
344 root 1.13 req->callback = SvREFCNT_inc (callback);
345    
346     send_req (req);
347    
348     void
349     aio_read(fh,offset,length,data,dataoffset,callback)
350     InputStream fh
351     UV offset
352     IV length
353     SV * data
354     IV dataoffset
355     SV * callback
356     PROTOTYPE: $$$$$$
357     CODE:
358     read_write (aTHX_ 0, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
359    
360     void
361     aio_write(fh,offset,length,data,dataoffset,callback)
362     OutputStream fh
363     UV offset
364     IV length
365     SV * data
366     IV dataoffset
367     SV * callback
368     PROTOTYPE: $$$$$$
369     CODE:
370     read_write (aTHX_ 1, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
371    
372     void
373     aio_stat(fh_or_path,callback)
374     SV * fh_or_path
375     SV * callback
376     PROTOTYPE: $$
377     ALIAS:
378     aio_lstat = 1
379     CODE:
380     aio_req req;
381    
382 root 1.14 Newz (0, req, 1, aio_cb);
383 root 1.13
384     if (!req)
385     croak ("out of memory during aio_req allocation");
386    
387     New (0, req->statdata, 1, struct stat64);
388    
389     if (!req->statdata)
390     croak ("out of memory during aio_req->statdata allocation");
391    
392     if (SvPOK (fh_or_path))
393     {
394     req->type = ix ? REQ_LSTAT : REQ_STAT;
395 root 1.14 req->data = newSVsv (fh_or_path);
396     req->dataptr = SvPV_nolen (req->data);
397 root 1.13 }
398     else
399     {
400     req->type = REQ_FSTAT;
401     req->fd = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
402     }
403    
404 root 1.8 req->callback = SvREFCNT_inc (callback);
405    
406     send_req (req);
407 root 1.5
408     int
409     poll_fileno()
410     PROTOTYPE:
411     CODE:
412     RETVAL = respipe[0];
413     OUTPUT:
414     RETVAL
415    
416     int
417 root 1.6 poll_cb(...)
418 root 1.5 PROTOTYPE:
419     CODE:
420     RETVAL = poll_cb (aTHX);
421     OUTPUT:
422     RETVAL
423    
424     int
425     nreqs()
426     PROTOTYPE:
427     CODE:
428     RETVAL = nreqs;
429     OUTPUT:
430     RETVAL
431