ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/IO-AIO/AIO.xs
Revision: 1.152
Committed: Tue Jul 14 00:32:27 2009 UTC (14 years, 10 months ago) by root
Branch: MAIN
Changes since 1.151: +3 -14 lines
Log Message:
*** empty log message ***

File Contents

# User Rev Content
1 root 1.115 #include "libeio/xthread.h"
2 root 1.63
3 root 1.19 #include <errno.h>
4    
5 root 1.15 #include "EXTERN.h"
6 root 1.1 #include "perl.h"
7     #include "XSUB.h"
8    
9 root 1.151 #include "schmorp.h"
10    
11 root 1.37 #include <stddef.h>
12 root 1.94 #include <stdlib.h>
13 root 1.41 #include <errno.h>
14 root 1.1 #include <sys/types.h>
15     #include <sys/stat.h>
16 root 1.37 #include <limits.h>
17 root 1.1 #include <fcntl.h>
18     #include <sched.h>
19 root 1.103
20 root 1.118 /* perl namespace pollution */
21     #undef VERSION
22    
23 root 1.103 #ifdef _WIN32
24    
25 root 1.115 # define EIO_STRUCT_DIRENT Direntry_t
26     # undef malloc
27     # undef free
28 root 1.103
29     // perl overrides all those nice win32 functions
30     # undef open
31     # undef read
32     # undef write
33 root 1.104 # undef send
34     # undef recv
35 root 1.103 # undef stat
36     # undef fstat
37     # define lstat stat
38     # undef truncate
39     # undef ftruncate
40     # undef open
41     # undef close
42     # undef unlink
43     # undef rmdir
44     # undef rename
45     # undef lseek
46    
47     # define chown(a,b,c) (errno = ENOSYS, -1)
48     # define fchown(a,b,c) (errno = ENOSYS, -1)
49     # define fchmod(a,b) (errno = ENOSYS, -1)
50     # define symlink(a,b) (errno = ENOSYS, -1)
51     # define readlink(a,b,c) (errno = ENOSYS, -1)
52     # define mknod(a,b,c) (errno = ENOSYS, -1)
53     # define truncate(a,b) (errno = ENOSYS, -1)
54     # define ftruncate(fd,o) chsize ((fd), (o))
55     # define fsync(fd) _commit (fd)
56     # define opendir(fd) (errno = ENOSYS, 0)
57     # define readdir(fd) (errno = ENOSYS, -1)
58     # define closedir(fd) (errno = ENOSYS, -1)
59     # define mkdir(a,b) mkdir (a)
60    
61     #else
62    
63     # include <sys/time.h>
64     # include <sys/select.h>
65     # include <unistd.h>
66     # include <utime.h>
67     # include <signal.h>
68 root 1.115 # define EIO_STRUCT_DIRENT struct dirent
69 root 1.103
70     #endif
71 root 1.1
72 root 1.125 /* perl stupidly overrides readdir and maybe others */
73     /* with thread-unsafe versions, imagine that :( */
74     #undef readdir
75     #undef opendir
76     #undef closedir
77    
78 root 1.115 #define EIO_STRUCT_STAT Stat_t
79 root 1.65
80 root 1.101 /* use NV for 32 bit perls as it allows larger offsets */
81     #if IVSIZE >= 8
82 root 1.148 # define VAL64 IV
83 root 1.101 # define SvVAL64 SvIV
84 root 1.148 # define newSVval64 newSViv
85 root 1.101 #else
86 root 1.148 # define VAL64 NV
87 root 1.101 # define SvVAL64 SvNV
88 root 1.148 # define newSVval64 newSVnv
89 root 1.101 #endif
90    
91 root 1.127 /*****************************************************************************/
92    
93     #if __GNUC__ >= 3
94     # define expect(expr,value) __builtin_expect ((expr),(value))
95     #else
96     # define expect(expr,value) (expr)
97     #endif
98    
99     #define expect_false(expr) expect ((expr) != 0, 0)
100     #define expect_true(expr) expect ((expr) != 0, 1)
101    
102     /*****************************************************************************/
103    
104 root 1.107 static HV *stash;
105 root 1.90 typedef SV SV8; /* byte-sv, used for argument-checking */
106 root 1.148 typedef int aio_rfd; /* read file desriptor */
107     typedef int aio_wfd; /* write file descriptor */
108 root 1.90
109 root 1.44 #define AIO_REQ_KLASS "IO::AIO::REQ"
110     #define AIO_GRP_KLASS "IO::AIO::GRP"
111 root 1.43
112 root 1.121 #define EIO_REQ_MEMBERS \
113 root 1.115 SV *callback; \
114     SV *sv1, *sv2; \
115     STRLEN stroffset; \
116 root 1.121 SV *self;
117 root 1.115
118 root 1.122 #define EIO_NO_WRAPPERS 1
119    
120 root 1.115 #include "libeio/eio.h"
121    
122 root 1.148 #ifndef POSIX_FADV_NORMAL
123     # define POSIX_FADV_NORMAL 0
124 root 1.149 # define NO_FADVISE 1
125 root 1.148 #endif
126    
127     #ifndef POSIX_FADV_SEQUENTIAL
128     # define POSIX_FADV_SEQUENTIAL 0
129     #endif
130    
131     #ifndef POSIX_FADV_RANDOM
132     # define POSIX_FADV_RANDOM 0
133     #endif
134    
135     #ifndef POSIX_FADV_NOREUSE
136     # define POSIX_FADV_NOREUSE 0
137     #endif
138    
139     #ifndef POSIX_FADV_WILLNEED
140     # define POSIX_FADV_WILLNEED 0
141     #endif
142    
143     #ifndef POSIX_FADV_DONTNEED
144     # define POSIX_FADV_DONTNEED 0
145     #endif
146    
147 root 1.115 static int req_invoke (eio_req *req);
148     #define EIO_FINISH(req) req_invoke (req)
149     static void req_destroy (eio_req *grp);
150     #define EIO_DESTROY(req) req_destroy (req)
151 root 1.1
152 root 1.58 enum {
153 root 1.115 FLAG_SV2_RO_OFF = 0x40, /* data was set readonly */
154 root 1.58 };
155    
156 root 1.115 #include "libeio/eio.c"
157 root 1.1
158 root 1.115 typedef eio_req *aio_req;
159     typedef eio_req *aio_req_ornot;
160 root 1.60
161 root 1.117 static SV *on_next_submit;
162 root 1.121 static int next_pri = EIO_PRI_DEFAULT;
163 root 1.117 static int max_outstanding;
164 root 1.94
165 root 1.106 static int respipe_osf [2], respipe [2] = { -1, -1 };
166 root 1.80
167 root 1.115 static void req_destroy (aio_req req);
168     static void req_cancel (aio_req req);
169 root 1.80
170 root 1.115 static void want_poll (void)
171 root 1.80 {
172 root 1.115 /* write a dummy byte to the pipe so fh becomes ready */
173     respipe_write (respipe_osf [1], (const void *)&respipe_osf, 1);
174 root 1.67 }
175    
176 root 1.115 static void done_poll (void)
177 root 1.67 {
178 root 1.115 /* read any signals sent by the worker threads */
179     char buf [4];
180     while (respipe_read (respipe [0], buf, 4) == 4)
181     ;
182 root 1.67 }
183 root 1.1
184 root 1.43 /* must be called at most once */
185 root 1.44 static SV *req_sv (aio_req req, const char *klass)
186 root 1.43 {
187 root 1.49 if (!req->self)
188     {
189     req->self = (SV *)newHV ();
190     sv_magic (req->self, 0, PERL_MAGIC_ext, (char *)req, 0);
191     }
192 root 1.43
193 root 1.45 return sv_2mortal (sv_bless (newRV_inc (req->self), gv_stashpv (klass, 1)));
194 root 1.43 }
195    
196 root 1.45 static aio_req SvAIO_REQ (SV *sv)
197 root 1.26 {
198 root 1.53 MAGIC *mg;
199    
200 root 1.45 if (!sv_derived_from (sv, AIO_REQ_KLASS) || !SvROK (sv))
201     croak ("object of class " AIO_REQ_KLASS " expected");
202 root 1.43
203 root 1.53 mg = mg_find (SvRV (sv), PERL_MAGIC_ext);
204 root 1.43
205     return mg ? (aio_req)mg->mg_ptr : 0;
206     }
207    
208 root 1.49 static void aio_grp_feed (aio_req grp)
209     {
210 root 1.115 if (grp->sv2 && SvOK (grp->sv2))
211 root 1.49 {
212 root 1.115 dSP;
213 root 1.49
214 root 1.115 ENTER;
215     SAVETMPS;
216     PUSHMARK (SP);
217     XPUSHs (req_sv (grp, AIO_GRP_KLASS));
218     PUTBACK;
219     call_sv (grp->sv2, G_VOID | G_EVAL | G_KEEPERR);
220     SPAGAIN;
221     FREETMPS;
222     LEAVE;
223 root 1.50 }
224     }
225    
226 root 1.117 static void req_submit (eio_req *req)
227     {
228     eio_submit (req);
229    
230 root 1.127 if (expect_false (on_next_submit))
231 root 1.117 {
232     dSP;
233     SV *cb = sv_2mortal (on_next_submit);
234    
235     on_next_submit = 0;
236    
237     PUSHMARK (SP);
238     PUTBACK;
239     call_sv (cb, G_DISCARD | G_EVAL);
240     }
241     }
242    
243 root 1.115 static int req_invoke (eio_req *req)
244 root 1.45 {
245     dSP;
246    
247 root 1.100 if (req->flags & FLAG_SV2_RO_OFF)
248     SvREADONLY_off (req->sv2);
249 root 1.86
250 root 1.128 if (!EIO_CANCELLED (req) && req->callback)
251 root 1.67 {
252     ENTER;
253     SAVETMPS;
254     PUSHMARK (SP);
255     EXTEND (SP, 1);
256 root 1.45
257 root 1.67 switch (req->type)
258 root 1.45 {
259 root 1.115 case EIO_READDIR:
260 root 1.45 {
261 root 1.67 SV *rv = &PL_sv_undef;
262 root 1.45
263 root 1.67 if (req->result >= 0)
264 root 1.45 {
265 root 1.71 int i;
266 root 1.147 char *names = (char *)req->ptr2;
267     eio_dirent *ent = (eio_dirent *)req->ptr1; /* might be 0 */
268 root 1.67 AV *av = newAV ();
269    
270 root 1.71 av_extend (av, req->result - 1);
271    
272     for (i = 0; i < req->result; ++i)
273 root 1.67 {
274 root 1.141 if (req->int1 & EIO_READDIR_DENTS)
275     {
276 root 1.147 SV *namesv = newSVpvn (names + ent->nameofs, ent->namelen);
277 root 1.141
278     if (req->int1 & EIO_READDIR_CUSTOM2)
279     {
280     static SV *sv_type [EIO_DT_MAX + 1]; /* type sv cache */
281     AV *avent = newAV ();
282    
283     av_extend (avent, 2);
284    
285     if (!sv_type [ent->type])
286     {
287     sv_type [ent->type] = newSViv (ent->type);
288     SvREADONLY_on (sv_type [ent->type]);
289     }
290    
291     av_store (avent, 0, namesv);
292 root 1.143 av_store (avent, 1, SvREFCNT_inc (sv_type [ent->type]));
293     av_store (avent, 2, IVSIZE >= 8 ? newSVuv (ent->inode) : newSVnv (ent->inode));
294 root 1.141
295     av_store (av, i, newRV_noinc ((SV *)avent));
296     }
297     else
298     av_store (av, i, namesv);
299    
300 root 1.147 ++ent;
301 root 1.141 }
302     else
303     {
304 root 1.147 SV *name = newSVpv (names, 0);
305 root 1.141 av_store (av, i, name);
306 root 1.147 names += SvCUR (name) + 1;
307 root 1.141 }
308 root 1.67 }
309 root 1.45
310 root 1.67 rv = sv_2mortal (newRV_noinc ((SV *)av));
311 root 1.45 }
312    
313 root 1.67 PUSHs (rv);
314 root 1.141
315     if (req->int1 & EIO_READDIR_CUSTOM1)
316     XPUSHs (sv_2mortal (newSViv (req->int1 & ~(EIO_READDIR_CUSTOM1 | EIO_READDIR_CUSTOM2))));
317 root 1.45 }
318 root 1.67 break;
319 root 1.45
320 root 1.115 case EIO_OPEN:
321 root 1.67 {
322     /* convert fd to fh */
323 root 1.107 SV *fh = &PL_sv_undef;
324    
325     if (req->result >= 0)
326     {
327     GV *gv = (GV *)sv_newmortal ();
328     int flags = req->int1 & (O_RDONLY | O_WRONLY | O_RDWR);
329     char sym [64];
330     int symlen;
331    
332 root 1.146 symlen = snprintf (sym, sizeof (sym), "fd#%d", (int)req->result);
333 root 1.107 gv_init (gv, stash, sym, symlen, 0);
334    
335     symlen = snprintf (
336     sym,
337     sizeof (sym),
338 root 1.146 "%s&=%d",
339 root 1.107 flags == O_RDONLY ? "<" : flags == O_WRONLY ? ">" : "+<",
340 root 1.146 (int)req->result
341 root 1.107 );
342 root 1.45
343 root 1.107 if (do_open (gv, sym, symlen, 0, 0, 0, 0))
344 root 1.108 fh = (SV *)gv;
345 root 1.107 }
346 root 1.45
347 root 1.109 PUSHs (fh);
348 root 1.67 }
349     break;
350 root 1.45
351 root 1.115 case EIO_GROUP:
352 root 1.86 req->int1 = 2; /* mark group as finished */
353 root 1.45
354 root 1.86 if (req->sv1)
355 root 1.67 {
356     int i;
357 root 1.86 AV *av = (AV *)req->sv1;
358 root 1.49
359 root 1.67 EXTEND (SP, AvFILL (av) + 1);
360     for (i = 0; i <= AvFILL (av); ++i)
361     PUSHs (*av_fetch (av, i, 0));
362     }
363     break;
364 root 1.48
365 root 1.115 case EIO_NOP:
366     case EIO_BUSY:
367 root 1.67 break;
368 root 1.48
369 root 1.115 case EIO_READLINK:
370 root 1.86 if (req->result > 0)
371 root 1.119 PUSHs (sv_2mortal (newSVpvn (req->ptr2, req->result)));
372 root 1.86 break;
373    
374 root 1.115 case EIO_STAT:
375     case EIO_LSTAT:
376     case EIO_FSTAT:
377     PL_laststype = req->type == EIO_LSTAT ? OP_LSTAT : OP_STAT;
378 root 1.87 PL_laststatval = req->result;
379 root 1.115 PL_statcache = *(EIO_STRUCT_STAT *)(req->ptr2);
380 root 1.87 PUSHs (sv_2mortal (newSViv (req->result)));
381     break;
382    
383 root 1.115 case EIO_READ:
384 root 1.138 {
385     SvCUR_set (req->sv2, req->stroffset + (req->result > 0 ? req->result : 0));
386     *SvEND (req->sv2) = 0;
387     SvPOK_only (req->sv2);
388     SvSETMAGIC (req->sv2);
389     PUSHs (sv_2mortal (newSViv (req->result)));
390     }
391 root 1.87 break;
392    
393 root 1.115 case EIO_DUP2:
394     if (req->result > 0)
395     req->result = 0;
396     /* FALLTHROUGH */
397    
398 root 1.67 default:
399     PUSHs (sv_2mortal (newSViv (req->result)));
400     break;
401     }
402 root 1.45
403 root 1.79 errno = req->errorno;
404 root 1.51
405 root 1.67 PUTBACK;
406 root 1.109 call_sv (req->callback, G_VOID | G_EVAL | G_DISCARD);
407 root 1.67 SPAGAIN;
408 root 1.51
409 root 1.67 FREETMPS;
410     LEAVE;
411 root 1.109
412     PUTBACK;
413 root 1.45 }
414    
415 root 1.115 return !!SvTRUE (ERRSV);
416 root 1.67 }
417    
418 root 1.101 static void req_destroy (aio_req req)
419 root 1.67 {
420 root 1.43 if (req->self)
421     {
422     sv_unmagic (req->self, PERL_MAGIC_ext);
423     SvREFCNT_dec (req->self);
424     }
425    
426 root 1.86 SvREFCNT_dec (req->sv1);
427     SvREFCNT_dec (req->sv2);
428 root 1.49 SvREFCNT_dec (req->callback);
429 root 1.26
430     Safefree (req);
431     }
432    
433 root 1.72 static void req_cancel_subs (aio_req grp)
434     {
435     aio_req sub;
436    
437 root 1.115 if (grp->type != EIO_GROUP)
438 root 1.72 return;
439    
440 root 1.86 SvREFCNT_dec (grp->sv2);
441     grp->sv2 = 0;
442 root 1.72
443 root 1.115 eio_grp_cancel (grp);
444 root 1.1 }
445    
446 root 1.104 #ifdef USE_SOCKETS_AS_HANDLES
447     # define TO_SOCKET(x) (win32_get_osfhandle (x))
448     #else
449     # define TO_SOCKET(x) (x)
450     #endif
451    
452     static void
453 root 1.111 create_respipe (void)
454 root 1.104 {
455 root 1.106 int old_readfd = respipe [0];
456    
457     if (respipe [1] >= 0)
458     respipe_close (TO_SOCKET (respipe [1]));
459    
460     #ifdef _WIN32
461     if (PerlSock_socketpair (AF_UNIX, SOCK_STREAM, 0, respipe))
462     #else
463     if (pipe (respipe))
464     #endif
465     croak ("unable to initialize result pipe");
466    
467     if (old_readfd >= 0)
468     {
469     if (dup2 (TO_SOCKET (respipe [0]), TO_SOCKET (old_readfd)) < 0)
470     croak ("unable to initialize result pipe(2)");
471    
472     respipe_close (respipe [0]);
473     respipe [0] = old_readfd;
474     }
475    
476 root 1.104 #ifdef _WIN32
477     int arg = 1;
478 root 1.106 if (ioctlsocket (TO_SOCKET (respipe [0]), FIONBIO, &arg)
479     || ioctlsocket (TO_SOCKET (respipe [1]), FIONBIO, &arg))
480 root 1.104 #else
481 root 1.106 if (fcntl (respipe [0], F_SETFL, O_NONBLOCK)
482     || fcntl (respipe [1], F_SETFL, O_NONBLOCK))
483 root 1.104 #endif
484 root 1.106 croak ("unable to initialize result pipe(3)");
485 root 1.104
486     respipe_osf [0] = TO_SOCKET (respipe [0]);
487     respipe_osf [1] = TO_SOCKET (respipe [1]);
488     }
489    
490 root 1.111 static void poll_wait (void)
491 root 1.80 {
492     fd_set rfd;
493    
494 root 1.115 while (eio_nreqs ())
495 root 1.30 {
496 root 1.80 int size;
497 root 1.115
498     X_LOCK (reslock);
499 root 1.80 size = res_queue.size;
500 root 1.115 X_UNLOCK (reslock);
501 root 1.80
502     if (size)
503     return;
504    
505 root 1.123 etp_maybe_start_thread ();
506 root 1.80
507 root 1.104 FD_ZERO (&rfd);
508     FD_SET (respipe [0], &rfd);
509 root 1.80
510 root 1.104 PerlSock_select (respipe [0] + 1, &rfd, 0, 0, 0);
511 root 1.22 }
512 root 1.80 }
513    
514 root 1.111 static int poll_cb (void)
515 root 1.80 {
516 root 1.124 for (;;)
517 root 1.116 {
518 root 1.124 int res = eio_poll ();
519 root 1.116
520     if (res > 0)
521     croak (0);
522 root 1.124
523     if (!max_outstanding || max_outstanding > eio_nreqs ())
524     return res;
525    
526     poll_wait ();
527 root 1.116 }
528 root 1.17 }
529 root 1.37
530 root 1.116 static void atfork_child (void)
531     {
532     create_respipe ();
533     }
534    
535 root 1.128 static SV *
536     get_cb (SV *cb_sv)
537     {
538 root 1.152 SvGETMAGIC (cb_sv);
539     return SvOK (cb_sv) ? s_get_cv_croak (cb_sv) : 0;
540 root 1.128 }
541    
542 root 1.22 #define dREQ \
543 root 1.128 SV *cb_cv; \
544 root 1.22 aio_req req; \
545 root 1.60 int req_pri = next_pri; \
546 root 1.121 next_pri = EIO_PRI_DEFAULT; \
547 root 1.22 \
548 root 1.128 cb_cv = get_cb (callback); \
549 root 1.22 \
550 root 1.115 Newz (0, req, 1, eio_req); \
551 root 1.22 if (!req) \
552 root 1.115 croak ("out of memory during eio_req allocation"); \
553 root 1.22 \
554 root 1.128 req->callback = SvREFCNT_inc (cb_cv); \
555 root 1.60 req->pri = req_pri
556 root 1.43
557     #define REQ_SEND \
558 root 1.126 PUTBACK; \
559 root 1.117 req_submit (req); \
560 root 1.126 SPAGAIN; \
561 root 1.43 \
562     if (GIMME_V != G_VOID) \
563 root 1.44 XPUSHs (req_sv (req, AIO_REQ_KLASS));
564 root 1.136
565 root 1.1 MODULE = IO::AIO PACKAGE = IO::AIO
566    
567 root 1.8 PROTOTYPES: ENABLE
568    
569 root 1.1 BOOT:
570     {
571 root 1.141 static const struct {
572     const char *name;
573     IV iv;
574     } *civ, const_iv[] = {
575     # define const_iv(name, value) { # name, (IV) value },
576     # define const_eio(name) { # name, (IV) EIO_ ## name },
577     const_iv (EXDEV , EXDEV)
578     const_iv (ENOSYS , ENOSYS)
579     const_iv (O_RDONLY, O_RDONLY)
580     const_iv (O_WRONLY, O_WRONLY)
581     const_iv (O_CREAT , O_CREAT)
582     const_iv (O_TRUNC , O_TRUNC)
583 root 1.122 #ifndef _WIN32
584 root 1.141 const_iv (S_IFIFO , S_IFIFO)
585 root 1.103 #endif
586 root 1.148 const_iv (FADV_NORMAL , POSIX_FADV_NORMAL)
587     const_iv (FADV_SEQUENTIAL, POSIX_FADV_SEQUENTIAL)
588     const_iv (FADV_RANDOM , POSIX_FADV_RANDOM)
589     const_iv (FADV_NOREUSE , POSIX_FADV_NOREUSE)
590     const_iv (FADV_WILLNEED , POSIX_FADV_WILLNEED)
591     const_iv (FADV_DONTNEED , POSIX_FADV_DONTNEED)
592    
593 root 1.141 const_eio (SYNC_FILE_RANGE_WAIT_BEFORE)
594     const_eio (SYNC_FILE_RANGE_WRITE)
595     const_eio (SYNC_FILE_RANGE_WAIT_AFTER)
596    
597     const_eio (READDIR_DENTS)
598     const_eio (READDIR_DIRS_FIRST)
599     const_eio (READDIR_STAT_ORDER)
600     const_eio (READDIR_FOUND_UNKNOWN)
601    
602     const_eio (DT_UNKNOWN)
603     const_eio (DT_FIFO)
604     const_eio (DT_CHR)
605     const_eio (DT_DIR)
606     const_eio (DT_BLK)
607     const_eio (DT_REG)
608     const_eio (DT_LNK)
609     const_eio (DT_SOCK)
610     const_eio (DT_WHT)
611     };
612 root 1.103
613 root 1.146 stash = gv_stashpv ("IO::AIO", 1);
614    
615 root 1.141 for (civ = const_iv + sizeof (const_iv) / sizeof (const_iv [0]); civ-- > const_iv; )
616     newCONSTSUB (stash, (char *)civ->name, newSViv (civ->iv));
617    
618     create_respipe ();
619 root 1.41
620 root 1.141 if (eio_init (want_poll, done_poll) < 0)
621     croak ("IO::AIO: unable to initialise eio library");
622 root 1.116
623 root 1.141 /* atfork child called in fifo order, so before eio's handler */
624     X_THREAD_ATFORK (0, 0, atfork_child);
625 root 1.1 }
626    
627     void
628 root 1.85 max_poll_reqs (int nreqs)
629     PROTOTYPE: $
630     CODE:
631 root 1.115 eio_set_max_poll_reqs (nreqs);
632 root 1.85
633     void
634     max_poll_time (double nseconds)
635     PROTOTYPE: $
636     CODE:
637 root 1.115 eio_set_max_poll_time (nseconds);
638 root 1.85
639     void
640 root 1.78 min_parallel (int nthreads)
641 root 1.1 PROTOTYPE: $
642 root 1.115 CODE:
643     eio_set_min_parallel (nthreads);
644 root 1.1
645     void
646 root 1.78 max_parallel (int nthreads)
647     PROTOTYPE: $
648 root 1.115 CODE:
649     eio_set_max_parallel (nthreads);
650 root 1.78
651 root 1.85 void
652     max_idle (int nthreads)
653     PROTOTYPE: $
654     CODE:
655 root 1.115 eio_set_max_idle (nthreads);
656 root 1.85
657 root 1.115 void
658 root 1.78 max_outstanding (int maxreqs)
659 root 1.1 PROTOTYPE: $
660 root 1.78 CODE:
661     max_outstanding = maxreqs;
662 root 1.1
663     void
664 root 1.99 aio_open (SV8 *pathname, int flags, int mode, SV *callback=&PL_sv_undef)
665 root 1.8 PROTOTYPE: $$$;$
666 root 1.43 PPCODE:
667 root 1.1 {
668 root 1.22 dREQ;
669 root 1.1
670 root 1.115 req->type = EIO_OPEN;
671 root 1.86 req->sv1 = newSVsv (pathname);
672 root 1.89 req->ptr1 = SvPVbyte_nolen (req->sv1);
673 root 1.86 req->int1 = flags;
674 root 1.115 req->int2 = mode;
675 root 1.1
676 root 1.43 REQ_SEND;
677 root 1.1 }
678    
679     void
680 root 1.107 aio_fsync (SV *fh, SV *callback=&PL_sv_undef)
681 root 1.8 PROTOTYPE: $;$
682 root 1.1 ALIAS:
683 root 1.115 aio_fsync = EIO_FSYNC
684     aio_fdatasync = EIO_FDATASYNC
685 root 1.43 PPCODE:
686 root 1.1 {
687 root 1.151 int fd = s_fileno_croak (fh, 0);
688 root 1.22 dREQ;
689 root 1.1
690     req->type = ix;
691 root 1.99 req->sv1 = newSVsv (fh);
692 root 1.136 req->int1 = fd;
693 root 1.1
694 root 1.43 REQ_SEND (req);
695 root 1.1 }
696    
697     void
698 root 1.148 aio_sync_file_range (SV *fh, off_t offset, size_t nbytes, UV flags, SV *callback=&PL_sv_undef)
699 root 1.133 PROTOTYPE: $$$$;$
700     PPCODE:
701     {
702 root 1.151 int fd = s_fileno_croak (fh, 0);
703 root 1.133 dREQ;
704    
705     req->type = EIO_SYNC_FILE_RANGE;
706     req->sv1 = newSVsv (fh);
707 root 1.136 req->int1 = fd;
708 root 1.148 req->offs = offset;
709     req->size = nbytes;
710 root 1.133 req->int2 = flags;
711    
712     REQ_SEND (req);
713     }
714    
715     void
716 root 1.113 aio_close (SV *fh, SV *callback=&PL_sv_undef)
717 root 1.107 PROTOTYPE: $;$
718     PPCODE:
719     {
720 root 1.115 static int close_pipe = -1; /* dummy fd to close fds via dup2 */
721 root 1.151 int fd = s_fileno_croak (fh, 0);
722 root 1.109 dREQ;
723 root 1.107
724 root 1.115 if (close_pipe < 0)
725     {
726     int pipefd [2];
727    
728     if (pipe (pipefd) < 0
729     || close (pipefd [1]) < 0
730     || fcntl (pipefd [0], F_SETFD, FD_CLOEXEC) < 0)
731     abort (); /*D*/
732    
733     close_pipe = pipefd [0];
734     }
735    
736     req->type = EIO_DUP2;
737     req->int1 = close_pipe;
738     req->sv2 = newSVsv (fh);
739 root 1.136 req->int2 = fd;
740 root 1.107
741 root 1.109 REQ_SEND (req);
742 root 1.107 }
743    
744     void
745 root 1.102 aio_read (SV *fh, SV *offset, SV *length, SV8 *data, IV dataoffset, SV *callback=&PL_sv_undef)
746 root 1.13 ALIAS:
747 root 1.115 aio_read = EIO_READ
748     aio_write = EIO_WRITE
749 root 1.8 PROTOTYPE: $$$$$;$
750 root 1.43 PPCODE:
751 root 1.13 {
752     STRLEN svlen;
753 root 1.151 int fd = s_fileno_croak (fh, ix == EIO_WRITE);
754 root 1.21 char *svptr = SvPVbyte (data, svlen);
755 root 1.102 UV len = SvUV (length);
756 root 1.134
757 root 1.13 if (dataoffset < 0)
758     dataoffset += svlen;
759    
760     if (dataoffset < 0 || dataoffset > svlen)
761 root 1.102 croak ("dataoffset outside of data scalar");
762 root 1.13
763 root 1.115 if (ix == EIO_WRITE)
764 root 1.13 {
765     /* write: check length and adjust. */
766 root 1.102 if (!SvOK (length) || len + dataoffset > svlen)
767     len = svlen - dataoffset;
768 root 1.13 }
769     else
770     {
771 root 1.138 /* read: check type and grow scalar as necessary */
772     SvUPGRADE (data, SVt_PV);
773 root 1.102 svptr = SvGROW (data, len + dataoffset + 1);
774 root 1.13 }
775    
776 root 1.22 {
777     dREQ;
778 root 1.13
779 root 1.22 req->type = ix;
780 root 1.99 req->sv1 = newSVsv (fh);
781 root 1.135 req->int1 = fd;
782 root 1.101 req->offs = SvOK (offset) ? SvVAL64 (offset) : -1;
783 root 1.102 req->size = len;
784 root 1.132 req->sv2 = SvREFCNT_inc (data);
785 root 1.115 req->ptr2 = (char *)svptr + dataoffset;
786 root 1.86 req->stroffset = dataoffset;
787 root 1.13
788 root 1.28 if (!SvREADONLY (data))
789     {
790     SvREADONLY_on (data);
791 root 1.100 req->flags |= FLAG_SV2_RO_OFF;
792 root 1.28 }
793    
794 root 1.43 REQ_SEND;
795 root 1.22 }
796 root 1.13 }
797 root 1.1
798     void
799 root 1.99 aio_readlink (SV8 *path, SV *callback=&PL_sv_undef)
800 root 1.86 PROTOTYPE: $$;$
801     PPCODE:
802     {
803     SV *data;
804     dREQ;
805    
806 root 1.115 req->type = EIO_READLINK;
807 root 1.99 req->sv1 = newSVsv (path);
808 root 1.115 req->ptr1 = SvPVbyte_nolen (req->sv1);
809 root 1.86
810     REQ_SEND;
811     }
812    
813     void
814 root 1.148 aio_sendfile (SV *out_fh, SV *in_fh, off_t in_offset, size_t length, SV *callback=&PL_sv_undef)
815 root 1.32 PROTOTYPE: $$$$;$
816 root 1.43 PPCODE:
817 root 1.32 {
818 root 1.151 int ifd = s_fileno_croak (in_fh , 0);
819     int ofd = s_fileno_croak (out_fh, 1);
820 root 1.32 dREQ;
821    
822 root 1.115 req->type = EIO_SENDFILE;
823 root 1.99 req->sv1 = newSVsv (out_fh);
824 root 1.136 req->int1 = ofd;
825 root 1.86 req->sv2 = newSVsv (in_fh);
826 root 1.136 req->int2 = ifd;
827 root 1.148 req->offs = in_offset;
828 root 1.86 req->size = length;
829 root 1.32
830 root 1.43 REQ_SEND;
831 root 1.32 }
832    
833     void
834 root 1.148 aio_readahead (SV *fh, off_t offset, size_t length, SV *callback=&PL_sv_undef)
835 root 1.8 PROTOTYPE: $$$;$
836 root 1.43 PPCODE:
837 root 1.1 {
838 root 1.151 int fd = s_fileno_croak (fh, 0);
839 root 1.22 dREQ;
840 root 1.1
841 root 1.115 req->type = EIO_READAHEAD;
842 root 1.99 req->sv1 = newSVsv (fh);
843 root 1.136 req->int1 = fd;
844 root 1.148 req->offs = offset;
845 root 1.86 req->size = length;
846 root 1.1
847 root 1.43 REQ_SEND;
848 root 1.1 }
849    
850     void
851 root 1.99 aio_stat (SV8 *fh_or_path, SV *callback=&PL_sv_undef)
852 root 1.1 ALIAS:
853 root 1.115 aio_stat = EIO_STAT
854     aio_lstat = EIO_LSTAT
855 root 1.43 PPCODE:
856 root 1.1 {
857 root 1.22 dREQ;
858 root 1.1
859 root 1.99 req->sv1 = newSVsv (fh_or_path);
860 root 1.87
861 root 1.119 if (SvPOK (req->sv1))
862 root 1.1 {
863 root 1.8 req->type = ix;
864 root 1.89 req->ptr1 = SvPVbyte_nolen (req->sv1);
865 root 1.1 }
866     else
867     {
868 root 1.115 req->type = EIO_FSTAT;
869 root 1.86 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
870 root 1.1 }
871    
872 root 1.43 REQ_SEND;
873 root 1.1 }
874    
875     void
876 root 1.99 aio_utime (SV8 *fh_or_path, SV *atime, SV *mtime, SV *callback=&PL_sv_undef)
877     PPCODE:
878     {
879     dREQ;
880    
881     req->nv1 = SvOK (atime) ? SvNV (atime) : -1.;
882     req->nv2 = SvOK (mtime) ? SvNV (mtime) : -1.;
883     req->sv1 = newSVsv (fh_or_path);
884    
885 root 1.119 if (SvPOK (req->sv1))
886 root 1.99 {
887 root 1.115 req->type = EIO_UTIME;
888 root 1.99 req->ptr1 = SvPVbyte_nolen (req->sv1);
889     }
890     else
891     {
892 root 1.115 req->type = EIO_FUTIME;
893 root 1.99 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
894     }
895    
896     REQ_SEND;
897     }
898    
899     void
900 root 1.103 aio_truncate (SV8 *fh_or_path, SV *offset, SV *callback=&PL_sv_undef)
901     PPCODE:
902     {
903     dREQ;
904    
905     req->sv1 = newSVsv (fh_or_path);
906     req->offs = SvOK (offset) ? SvVAL64 (offset) : -1;
907    
908 root 1.119 if (SvPOK (req->sv1))
909 root 1.103 {
910 root 1.115 req->type = EIO_TRUNCATE;
911 root 1.103 req->ptr1 = SvPVbyte_nolen (req->sv1);
912     }
913     else
914     {
915 root 1.115 req->type = EIO_FTRUNCATE;
916 root 1.103 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
917     }
918    
919     REQ_SEND;
920     }
921    
922     void
923 root 1.99 aio_chmod (SV8 *fh_or_path, int mode, SV *callback=&PL_sv_undef)
924 root 1.115 ALIAS:
925     aio_chmod = EIO_CHMOD
926     aio_mkdir = EIO_MKDIR
927 root 1.99 PPCODE:
928     {
929     dREQ;
930    
931 root 1.115 req->int2 = mode;
932 root 1.99 req->sv1 = newSVsv (fh_or_path);
933    
934 root 1.119 if (SvPOK (req->sv1))
935     {
936 root 1.120 req->type = ix;
937     req->ptr1 = SvPVbyte_nolen (req->sv1);
938 root 1.119 }
939 root 1.99 else
940 root 1.119 {
941 root 1.120 req->type = EIO_FCHMOD;
942     req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
943 root 1.119 }
944 root 1.99
945     REQ_SEND;
946     }
947    
948     void
949     aio_chown (SV8 *fh_or_path, SV *uid, SV *gid, SV *callback=&PL_sv_undef)
950     PPCODE:
951     {
952     dREQ;
953    
954     req->int2 = SvOK (uid) ? SvIV (uid) : -1;
955     req->int3 = SvOK (gid) ? SvIV (gid) : -1;
956     req->sv1 = newSVsv (fh_or_path);
957    
958 root 1.119 if (SvPOK (req->sv1))
959 root 1.99 {
960 root 1.115 req->type = EIO_CHOWN;
961 root 1.99 req->ptr1 = SvPVbyte_nolen (req->sv1);
962     }
963     else
964     {
965 root 1.115 req->type = EIO_FCHOWN;
966 root 1.99 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
967     }
968    
969     REQ_SEND;
970     }
971    
972     void
973 root 1.141 aio_readdirx (SV8 *pathname, IV flags, SV *callback=&PL_sv_undef)
974     PPCODE:
975     {
976     dREQ;
977    
978     req->type = EIO_READDIR;
979     req->sv1 = newSVsv (pathname);
980     req->ptr1 = SvPVbyte_nolen (req->sv1);
981     req->int1 = flags | EIO_READDIR_DENTS | EIO_READDIR_CUSTOM1;
982    
983     if (flags & EIO_READDIR_DENTS)
984 root 1.142 req->int1 |= EIO_READDIR_CUSTOM2;
985 root 1.141
986     REQ_SEND;
987     }
988    
989     void
990 root 1.99 aio_unlink (SV8 *pathname, SV *callback=&PL_sv_undef)
991 root 1.22 ALIAS:
992 root 1.115 aio_unlink = EIO_UNLINK
993     aio_rmdir = EIO_RMDIR
994     aio_readdir = EIO_READDIR
995 root 1.43 PPCODE:
996 root 1.1 {
997 root 1.22 dREQ;
998 root 1.1
999 root 1.22 req->type = ix;
1000 root 1.86 req->sv1 = newSVsv (pathname);
1001 root 1.89 req->ptr1 = SvPVbyte_nolen (req->sv1);
1002 root 1.87
1003 root 1.43 REQ_SEND;
1004 root 1.22 }
1005    
1006     void
1007 root 1.99 aio_link (SV8 *oldpath, SV8 *newpath, SV *callback=&PL_sv_undef)
1008 root 1.40 ALIAS:
1009 root 1.115 aio_link = EIO_LINK
1010     aio_symlink = EIO_SYMLINK
1011     aio_rename = EIO_RENAME
1012 root 1.43 PPCODE:
1013 root 1.22 {
1014     dREQ;
1015 root 1.1
1016 root 1.40 req->type = ix;
1017 root 1.115 req->sv1 = newSVsv (oldpath);
1018     req->ptr1 = SvPVbyte_nolen (req->sv1);
1019     req->sv2 = newSVsv (newpath);
1020 root 1.99 req->ptr2 = SvPVbyte_nolen (req->sv2);
1021 root 1.1
1022 root 1.43 REQ_SEND;
1023 root 1.1 }
1024    
1025 root 1.42 void
1026 root 1.99 aio_mknod (SV8 *pathname, int mode, UV dev, SV *callback=&PL_sv_undef)
1027 root 1.81 PPCODE:
1028     {
1029     dREQ;
1030    
1031 root 1.115 req->type = EIO_MKNOD;
1032 root 1.86 req->sv1 = newSVsv (pathname);
1033 root 1.89 req->ptr1 = SvPVbyte_nolen (req->sv1);
1034 root 1.115 req->int2 = (mode_t)mode;
1035 root 1.86 req->offs = dev;
1036 root 1.81
1037     REQ_SEND;
1038     }
1039    
1040     void
1041 root 1.99 aio_busy (double delay, SV *callback=&PL_sv_undef)
1042 root 1.45 PPCODE:
1043     {
1044     dREQ;
1045    
1046 root 1.115 req->type = EIO_BUSY;
1047 root 1.99 req->nv1 = delay < 0. ? 0. : delay;
1048 root 1.45
1049     REQ_SEND;
1050     }
1051    
1052     void
1053 root 1.99 aio_group (SV *callback=&PL_sv_undef)
1054 root 1.46 PROTOTYPE: ;$
1055 root 1.44 PPCODE:
1056 root 1.42 {
1057 root 1.44 dREQ;
1058 root 1.60
1059 root 1.115 req->type = EIO_GROUP;
1060 root 1.86
1061 root 1.127 req_submit (req);
1062 root 1.44 XPUSHs (req_sv (req, AIO_GRP_KLASS));
1063 root 1.42 }
1064    
1065 root 1.6 void
1066 root 1.99 aio_nop (SV *callback=&PL_sv_undef)
1067 root 1.110 ALIAS:
1068 root 1.115 aio_nop = EIO_NOP
1069     aio_sync = EIO_SYNC
1070 root 1.54 PPCODE:
1071     {
1072     dREQ;
1073    
1074 root 1.110 req->type = ix;
1075 root 1.54
1076     REQ_SEND;
1077     }
1078    
1079 root 1.79 int
1080     aioreq_pri (int pri = 0)
1081     PROTOTYPE: ;$
1082     CODE:
1083 root 1.121 RETVAL = next_pri;
1084 root 1.79 if (items > 0)
1085     {
1086 root 1.115 if (pri < EIO_PRI_MIN) pri = EIO_PRI_MIN;
1087     if (pri > EIO_PRI_MAX) pri = EIO_PRI_MAX;
1088 root 1.121 next_pri = pri;
1089 root 1.79 }
1090     OUTPUT:
1091     RETVAL
1092 root 1.68
1093     void
1094     aioreq_nice (int nice = 0)
1095 root 1.79 CODE:
1096     nice = next_pri - nice;
1097 root 1.115 if (nice < EIO_PRI_MIN) nice = EIO_PRI_MIN;
1098     if (nice > EIO_PRI_MAX) nice = EIO_PRI_MAX;
1099 root 1.121 next_pri = nice;
1100 root 1.60
1101 root 1.54 void
1102 root 1.40 flush ()
1103 root 1.6 PROTOTYPE:
1104     CODE:
1105 root 1.116 while (eio_nreqs ())
1106 root 1.6 {
1107     poll_wait ();
1108 root 1.91 poll_cb ();
1109 root 1.6 }
1110    
1111 root 1.91 int
1112 root 1.7 poll()
1113     PROTOTYPE:
1114     CODE:
1115 root 1.92 poll_wait ();
1116     RETVAL = poll_cb ();
1117 root 1.91 OUTPUT:
1118     RETVAL
1119 root 1.7
1120 root 1.1 int
1121     poll_fileno()
1122     PROTOTYPE:
1123     CODE:
1124 root 1.3 RETVAL = respipe [0];
1125 root 1.1 OUTPUT:
1126     RETVAL
1127    
1128     int
1129     poll_cb(...)
1130     PROTOTYPE:
1131     CODE:
1132 root 1.85 RETVAL = poll_cb ();
1133 root 1.1 OUTPUT:
1134     RETVAL
1135    
1136     void
1137     poll_wait()
1138     PROTOTYPE:
1139     CODE:
1140 root 1.92 poll_wait ();
1141 root 1.1
1142     int
1143     nreqs()
1144     PROTOTYPE:
1145     CODE:
1146 root 1.115 RETVAL = eio_nreqs ();
1147 root 1.1 OUTPUT:
1148     RETVAL
1149    
1150 root 1.79 int
1151     nready()
1152     PROTOTYPE:
1153     CODE:
1154 root 1.115 RETVAL = eio_nready ();
1155 root 1.79 OUTPUT:
1156     RETVAL
1157    
1158     int
1159     npending()
1160     PROTOTYPE:
1161     CODE:
1162 root 1.115 RETVAL = eio_npending ();
1163 root 1.79 OUTPUT:
1164     RETVAL
1165    
1166 root 1.85 int
1167     nthreads()
1168     PROTOTYPE:
1169     CODE:
1170 root 1.122 RETVAL = eio_nthreads ();
1171 root 1.85 OUTPUT:
1172     RETVAL
1173    
1174 root 1.148 int
1175     fadvise (aio_rfd fh, off_t offset, off_t length, IV advice)
1176     PROTOTYPE: $$$$
1177     CODE:
1178 root 1.150 #if _XOPEN_SOURCE >= 600 && !NO_FADVISE
1179 root 1.148 RETVAL = posix_fadvise (fh, offset, length, advice);
1180 root 1.150 #else
1181 root 1.148 RETVAL = errno = ENOSYS;
1182 root 1.150 #endif
1183 root 1.148 OUTPUT:
1184     RETVAL
1185    
1186     ssize_t
1187     sendfile (aio_wfd ofh, aio_rfd ifh, off_t offset, size_t count)
1188     PROTOTYPE: $$$$
1189     CODE:
1190     eio_sendfile_sync (ofh, ifh, offset, count);
1191    
1192 root 1.117 void _on_next_submit (SV *cb)
1193     CODE:
1194     SvREFCNT_dec (on_next_submit);
1195     on_next_submit = SvOK (cb) ? newSVsv (cb) : 0;
1196    
1197 root 1.48 PROTOTYPES: DISABLE
1198    
1199 root 1.44 MODULE = IO::AIO PACKAGE = IO::AIO::REQ
1200 root 1.43
1201     void
1202     cancel (aio_req_ornot req)
1203     CODE:
1204 root 1.115 eio_cancel (req);
1205 root 1.45
1206 root 1.56 void
1207 root 1.59 cb (aio_req_ornot req, SV *callback=&PL_sv_undef)
1208 root 1.128 PPCODE:
1209     {
1210     if (GIMME_V != G_VOID)
1211     XPUSHs (req->callback ? sv_2mortal (newRV_inc (req->callback)) : &PL_sv_undef);
1212    
1213     if (items > 1)
1214     {
1215 root 1.152 SV *cb_cv =get_cb (callback);
1216 root 1.128
1217     SvREFCNT_dec (req->callback);
1218     req->callback = SvREFCNT_inc (cb_cv);
1219     }
1220     }
1221 root 1.56
1222 root 1.45 MODULE = IO::AIO PACKAGE = IO::AIO::GRP
1223    
1224     void
1225     add (aio_req grp, ...)
1226     PPCODE:
1227     {
1228     int i;
1229 root 1.94
1230 root 1.86 if (grp->int1 == 2)
1231 root 1.49 croak ("cannot add requests to IO::AIO::GRP after the group finished");
1232    
1233 root 1.45 for (i = 1; i < items; ++i )
1234     {
1235 root 1.115 aio_req req;
1236    
1237 root 1.46 if (GIMME_V != G_VOID)
1238     XPUSHs (sv_2mortal (newSVsv (ST (i))));
1239    
1240 root 1.53 req = SvAIO_REQ (ST (i));
1241 root 1.45
1242 root 1.46 if (req)
1243 root 1.115 eio_grp_add (grp, req);
1244 root 1.45 }
1245     }
1246 root 1.43
1247 root 1.48 void
1248 root 1.72 cancel_subs (aio_req_ornot req)
1249     CODE:
1250     req_cancel_subs (req);
1251    
1252     void
1253 root 1.51 result (aio_req grp, ...)
1254     CODE:
1255     {
1256     int i;
1257 root 1.79 AV *av;
1258    
1259     grp->errorno = errno;
1260    
1261     av = newAV ();
1262 root 1.141 av_extend (av, items - 1);
1263 root 1.51
1264     for (i = 1; i < items; ++i )
1265     av_push (av, newSVsv (ST (i)));
1266    
1267 root 1.86 SvREFCNT_dec (grp->sv1);
1268     grp->sv1 = (SV *)av;
1269 root 1.51 }
1270    
1271     void
1272 root 1.79 errno (aio_req grp, int errorno = errno)
1273     CODE:
1274     grp->errorno = errorno;
1275    
1276     void
1277 root 1.67 limit (aio_req grp, int limit)
1278 root 1.49 CODE:
1279 root 1.115 eio_grp_limit (grp, limit);
1280 root 1.49
1281     void
1282 root 1.56 feed (aio_req grp, SV *callback=&PL_sv_undef)
1283 root 1.49 CODE:
1284     {
1285 root 1.86 SvREFCNT_dec (grp->sv2);
1286 root 1.131 grp->sv2 = newSVsv (callback);
1287     grp->feed = aio_grp_feed;
1288 root 1.49
1289 root 1.86 if (grp->int2 <= 0)
1290     grp->int2 = 2;
1291 root 1.49
1292 root 1.115 eio_grp_limit (grp, grp->int2);
1293 root 1.49 }
1294