ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/IO-AIO/AIO.xs
(Generate patch)

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.119 by root, Sun May 11 19:11:05 2008 UTC vs.
Revision 1.153 by root, Wed Jul 15 01:36:04 2009 UTC

3#include <errno.h> 3#include <errno.h>
4 4
5#include "EXTERN.h" 5#include "EXTERN.h"
6#include "perl.h" 6#include "perl.h"
7#include "XSUB.h" 7#include "XSUB.h"
8
9#include "schmorp.h"
8 10
9#include <stddef.h> 11#include <stddef.h>
10#include <stdlib.h> 12#include <stdlib.h>
11#include <errno.h> 13#include <errno.h>
12#include <sys/types.h> 14#include <sys/types.h>
65# include <signal.h> 67# include <signal.h>
66# define EIO_STRUCT_DIRENT struct dirent 68# define EIO_STRUCT_DIRENT struct dirent
67 69
68#endif 70#endif
69 71
72/* perl stupidly overrides readdir and maybe others */
73/* with thread-unsafe versions, imagine that :( */
74#undef readdir
75#undef opendir
76#undef closedir
77
70#define EIO_STRUCT_STAT Stat_t 78#define EIO_STRUCT_STAT Stat_t
71 79
72/* use NV for 32 bit perls as it allows larger offsets */ 80/* use NV for 32 bit perls as it allows larger offsets */
73#if IVSIZE >= 8 81#if IVSIZE >= 8
82# define VAL64 IV
74# define SvVAL64 SvIV 83# define SvVAL64 SvIV
84# define newSVval64 newSViv
75#else 85#else
86# define VAL64 NV
76# define SvVAL64 SvNV 87# define SvVAL64 SvNV
88# define newSVval64 newSVnv
77#endif 89#endif
90
91/*****************************************************************************/
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/*****************************************************************************/
78 103
79static HV *stash; 104static HV *stash;
80typedef SV SV8; /* byte-sv, used for argument-checking */ 105typedef SV SV8; /* byte-sv, used for argument-checking */
106typedef int aio_rfd; /* read file desriptor */
107typedef int aio_wfd; /* write file descriptor */
81 108
82#define AIO_REQ_KLASS "IO::AIO::REQ" 109#define AIO_REQ_KLASS "IO::AIO::REQ"
83#define AIO_GRP_KLASS "IO::AIO::GRP" 110#define AIO_GRP_KLASS "IO::AIO::GRP"
84 111
85#define EIO_COMMON \ 112#define EIO_REQ_MEMBERS \
86 SV *callback; \ 113 SV *callback; \
87 SV *sv1, *sv2; \ 114 SV *sv1, *sv2; \
88 STRLEN stroffset; \ 115 STRLEN stroffset; \
89 SV *self 116 SV *self;
117
118#define EIO_NO_WRAPPERS 1
90 119
91#include "libeio/eio.h" 120#include "libeio/eio.h"
121
122#ifndef POSIX_FADV_NORMAL
123# define POSIX_FADV_NORMAL 0
124# define NO_FADVISE 1
125#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
92 146
93static int req_invoke (eio_req *req); 147static int req_invoke (eio_req *req);
94#define EIO_FINISH(req) req_invoke (req) 148#define EIO_FINISH(req) req_invoke (req)
95static void aio_grp_feed (eio_req *grp);
96#define EIO_FEED(req) aio_grp_feed (req)
97static void req_destroy (eio_req *grp); 149static void req_destroy (eio_req *grp);
98#define EIO_DESTROY(req) req_destroy (req) 150#define EIO_DESTROY(req) req_destroy (req)
99 151
100enum { 152enum {
101 FLAG_SV2_RO_OFF = 0x40, /* data was set readonly */ 153 FLAG_SV2_RO_OFF = 0x40, /* data was set readonly */
105 157
106typedef eio_req *aio_req; 158typedef eio_req *aio_req;
107typedef eio_req *aio_req_ornot; 159typedef eio_req *aio_req_ornot;
108 160
109static SV *on_next_submit; 161static SV *on_next_submit;
110static int next_pri = EIO_DEFAULT_PRI + EIO_PRI_BIAS; 162static int next_pri = EIO_PRI_DEFAULT;
111static int max_outstanding; 163static int max_outstanding;
112 164
113static int respipe_osf [2], respipe [2] = { -1, -1 }; 165static int respipe_osf [2];
166static s_epipe respipe;
114 167
115static void req_destroy (aio_req req); 168static void req_destroy (aio_req req);
116static void req_cancel (aio_req req); 169static void req_cancel (aio_req req);
117 170
118static void want_poll (void) 171static void want_poll (void)
119{ 172{
120 /* write a dummy byte to the pipe so fh becomes ready */ 173 /* write a dummy byte to the pipe so fh becomes ready */
121 respipe_write (respipe_osf [1], (const void *)&respipe_osf, 1); 174 s_epipe_signal (&respipe);
122} 175}
123 176
124static void done_poll (void) 177static void done_poll (void)
125{ 178{
126 /* read any signals sent by the worker threads */ 179 /* read any signals sent by the worker threads */
127 char buf [4]; 180 s_epipe_drain (&respipe);
128 while (respipe_read (respipe [0], buf, 4) == 4)
129 ;
130} 181}
131 182
132/* must be called at most once */ 183/* must be called at most once */
133static SV *req_sv (aio_req req, const char *klass) 184static SV *req_sv (aio_req req, const char *klass)
134{ 185{
173 224
174static void req_submit (eio_req *req) 225static void req_submit (eio_req *req)
175{ 226{
176 eio_submit (req); 227 eio_submit (req);
177 228
178 if (on_next_submit) 229 if (expect_false (on_next_submit))
179 { 230 {
180 dSP; 231 dSP;
181 SV *cb = sv_2mortal (on_next_submit); 232 SV *cb = sv_2mortal (on_next_submit);
182 233
183 on_next_submit = 0; 234 on_next_submit = 0;
193 dSP; 244 dSP;
194 245
195 if (req->flags & FLAG_SV2_RO_OFF) 246 if (req->flags & FLAG_SV2_RO_OFF)
196 SvREADONLY_off (req->sv2); 247 SvREADONLY_off (req->sv2);
197 248
198 if (!EIO_CANCELLED (req) && SvOK (req->callback)) 249 if (!EIO_CANCELLED (req) && req->callback)
199 { 250 {
200 ENTER; 251 ENTER;
201 SAVETMPS; 252 SAVETMPS;
202 PUSHMARK (SP); 253 PUSHMARK (SP);
203 EXTEND (SP, 1); 254 EXTEND (SP, 1);
209 SV *rv = &PL_sv_undef; 260 SV *rv = &PL_sv_undef;
210 261
211 if (req->result >= 0) 262 if (req->result >= 0)
212 { 263 {
213 int i; 264 int i;
214 char *buf = req->ptr2; 265 char *names = (char *)req->ptr2;
266 eio_dirent *ent = (eio_dirent *)req->ptr1; /* might be 0 */
215 AV *av = newAV (); 267 AV *av = newAV ();
216 268
217 av_extend (av, req->result - 1); 269 av_extend (av, req->result - 1);
218 270
219 for (i = 0; i < req->result; ++i) 271 for (i = 0; i < req->result; ++i)
220 { 272 {
221 SV *sv = newSVpv (buf, 0); 273 if (req->int1 & EIO_READDIR_DENTS)
274 {
275 SV *namesv = newSVpvn (names + ent->nameofs, ent->namelen);
222 276
277 if (req->int1 & EIO_READDIR_CUSTOM2)
278 {
279 static SV *sv_type [EIO_DT_MAX + 1]; /* type sv cache */
280 AV *avent = newAV ();
281
282 av_extend (avent, 2);
283
284 if (!sv_type [ent->type])
285 {
286 sv_type [ent->type] = newSViv (ent->type);
287 SvREADONLY_on (sv_type [ent->type]);
288 }
289
290 av_store (avent, 0, namesv);
291 av_store (avent, 1, SvREFCNT_inc (sv_type [ent->type]));
292 av_store (avent, 2, IVSIZE >= 8 ? newSVuv (ent->inode) : newSVnv (ent->inode));
293
294 av_store (av, i, newRV_noinc ((SV *)avent));
295 }
296 else
223 av_store (av, i, sv); 297 av_store (av, i, namesv);
298
299 ++ent;
300 }
301 else
302 {
303 SV *name = newSVpv (names, 0);
304 av_store (av, i, name);
224 buf += SvCUR (sv) + 1; 305 names += SvCUR (name) + 1;
306 }
225 } 307 }
226 308
227 rv = sv_2mortal (newRV_noinc ((SV *)av)); 309 rv = sv_2mortal (newRV_noinc ((SV *)av));
228 } 310 }
229 311
230 PUSHs (rv); 312 PUSHs (rv);
313
314 if (req->int1 & EIO_READDIR_CUSTOM1)
315 XPUSHs (sv_2mortal (newSViv (req->int1 & ~(EIO_READDIR_CUSTOM1 | EIO_READDIR_CUSTOM2))));
231 } 316 }
232 break; 317 break;
233 318
234 case EIO_OPEN: 319 case EIO_OPEN:
235 { 320 {
241 GV *gv = (GV *)sv_newmortal (); 326 GV *gv = (GV *)sv_newmortal ();
242 int flags = req->int1 & (O_RDONLY | O_WRONLY | O_RDWR); 327 int flags = req->int1 & (O_RDONLY | O_WRONLY | O_RDWR);
243 char sym [64]; 328 char sym [64];
244 int symlen; 329 int symlen;
245 330
246 symlen = snprintf (sym, sizeof (sym), "fd#%d", req->result); 331 symlen = snprintf (sym, sizeof (sym), "fd#%d", (int)req->result);
247 gv_init (gv, stash, sym, symlen, 0); 332 gv_init (gv, stash, sym, symlen, 0);
248 333
249 symlen = snprintf ( 334 symlen = snprintf (
250 sym, 335 sym,
251 sizeof (sym), 336 sizeof (sym),
252 "%s&=%d", 337 "%s&=%d",
253 flags == O_RDONLY ? "<" : flags == O_WRONLY ? ">" : "+<", 338 flags == O_RDONLY ? "<" : flags == O_WRONLY ? ">" : "+<",
254 req->result 339 (int)req->result
255 ); 340 );
256 341
257 if (do_open (gv, sym, symlen, 0, 0, 0, 0)) 342 if (do_open (gv, sym, symlen, 0, 0, 0, 0))
258 fh = (SV *)gv; 343 fh = (SV *)gv;
259 } 344 }
293 PL_statcache = *(EIO_STRUCT_STAT *)(req->ptr2); 378 PL_statcache = *(EIO_STRUCT_STAT *)(req->ptr2);
294 PUSHs (sv_2mortal (newSViv (req->result))); 379 PUSHs (sv_2mortal (newSViv (req->result)));
295 break; 380 break;
296 381
297 case EIO_READ: 382 case EIO_READ:
383 {
298 SvCUR_set (req->sv2, req->stroffset + (req->result > 0 ? req->result : 0)); 384 SvCUR_set (req->sv2, req->stroffset + (req->result > 0 ? req->result : 0));
299 *SvEND (req->sv2) = 0; 385 *SvEND (req->sv2) = 0;
386 SvPOK_only (req->sv2);
387 SvSETMAGIC (req->sv2);
300 PUSHs (sv_2mortal (newSViv (req->result))); 388 PUSHs (sv_2mortal (newSViv (req->result)));
389 }
301 break; 390 break;
302 391
303 case EIO_DUP2: 392 case EIO_DUP2:
304 if (req->result > 0) 393 if (req->result > 0)
305 req->result = 0; 394 req->result = 0;
351 grp->sv2 = 0; 440 grp->sv2 = 0;
352 441
353 eio_grp_cancel (grp); 442 eio_grp_cancel (grp);
354} 443}
355 444
356#ifdef USE_SOCKETS_AS_HANDLES
357# define TO_SOCKET(x) (win32_get_osfhandle (x))
358#else
359# define TO_SOCKET(x) (x)
360#endif
361
362static void 445static void
363create_respipe (void) 446create_respipe (void)
364{ 447{
365 int old_readfd = respipe [0];
366
367 if (respipe [1] >= 0)
368 respipe_close (TO_SOCKET (respipe [1]));
369
370#ifdef _WIN32
371 if (PerlSock_socketpair (AF_UNIX, SOCK_STREAM, 0, respipe))
372#else
373 if (pipe (respipe)) 448 if (s_epipe_renew (&respipe))
374#endif
375 croak ("unable to initialize result pipe"); 449 croak ("unable to initialize result pipe");
376 450
377 if (old_readfd >= 0)
378 {
379 if (dup2 (TO_SOCKET (respipe [0]), TO_SOCKET (old_readfd)) < 0)
380 croak ("unable to initialize result pipe(2)");
381
382 respipe_close (respipe [0]);
383 respipe [0] = old_readfd;
384 }
385
386#ifdef _WIN32
387 int arg = 1;
388 if (ioctlsocket (TO_SOCKET (respipe [0]), FIONBIO, &arg)
389 || ioctlsocket (TO_SOCKET (respipe [1]), FIONBIO, &arg))
390#else
391 if (fcntl (respipe [0], F_SETFL, O_NONBLOCK)
392 || fcntl (respipe [1], F_SETFL, O_NONBLOCK))
393#endif
394 croak ("unable to initialize result pipe(3)");
395
396 respipe_osf [0] = TO_SOCKET (respipe [0]); 451 respipe_osf [0] = S_TO_SOCKET (respipe.fd [0]);
397 respipe_osf [1] = TO_SOCKET (respipe [1]); 452 respipe_osf [1] = S_TO_SOCKET (respipe.fd [1]);
398} 453}
399 454
400static void poll_wait (void) 455static void poll_wait (void)
401{ 456{
402 fd_set rfd;
403
404 while (eio_nreqs ()) 457 while (eio_nreqs ())
405 { 458 {
406 int size; 459 int size;
407 460
408 X_LOCK (reslock); 461 X_LOCK (reslock);
410 X_UNLOCK (reslock); 463 X_UNLOCK (reslock);
411 464
412 if (size) 465 if (size)
413 return; 466 return;
414 467
415 maybe_start_thread (); 468 etp_maybe_start_thread ();
416 469
417 FD_ZERO (&rfd); 470 s_epipe_wait (&respipe);
418 FD_SET (respipe [0], &rfd);
419
420 PerlSock_select (respipe [0] + 1, &rfd, 0, 0, 0);
421 } 471 }
422} 472}
423 473
424static int poll_cb (void) 474static int poll_cb (void)
425{ 475{
426 int res; 476 for (;;)
427
428 do
429 { 477 {
430 res = eio_poll (); 478 int res = eio_poll ();
431 479
432 if (res > 0) 480 if (res > 0)
433 croak (0); 481 croak (0);
482
483 if (!max_outstanding || max_outstanding > eio_nreqs ())
484 return res;
485
486 poll_wait ();
434 } 487 }
435 while (max_outstanding && max_outstanding <= eio_nreqs ());
436
437 return res;
438} 488}
439 489
440static void atfork_child (void) 490static void atfork_child (void)
441{ 491{
442 create_respipe (); 492 create_respipe ();
443} 493}
444 494
495static SV *
496get_cb (SV *cb_sv)
497{
498 SvGETMAGIC (cb_sv);
499 return SvOK (cb_sv) ? s_get_cv_croak (cb_sv) : 0;
500}
501
445#define dREQ \ 502#define dREQ \
503 SV *cb_cv; \
446 aio_req req; \ 504 aio_req req; \
447 int req_pri = next_pri; \ 505 int req_pri = next_pri; \
448 next_pri = EIO_DEFAULT_PRI + EIO_PRI_BIAS; \ 506 next_pri = EIO_PRI_DEFAULT; \
449 \ 507 \
450 if (SvOK (callback) && !SvROK (callback)) \ 508 cb_cv = get_cb (callback); \
451 croak ("callback must be undef or of reference type"); \
452 \ 509 \
453 Newz (0, req, 1, eio_req); \ 510 Newz (0, req, 1, eio_req); \
454 if (!req) \ 511 if (!req) \
455 croak ("out of memory during eio_req allocation"); \ 512 croak ("out of memory during eio_req allocation"); \
456 \ 513 \
457 req->callback = newSVsv (callback); \ 514 req->callback = SvREFCNT_inc (cb_cv); \
458 req->pri = req_pri 515 req->pri = req_pri
459 516
460#define REQ_SEND \ 517#define REQ_SEND \
518 PUTBACK; \
461 req_submit (req); \ 519 req_submit (req); \
520 SPAGAIN; \
462 \ 521 \
463 if (GIMME_V != G_VOID) \ 522 if (GIMME_V != G_VOID) \
464 XPUSHs (req_sv (req, AIO_REQ_KLASS)); 523 XPUSHs (req_sv (req, AIO_REQ_KLASS));
465 524
466MODULE = IO::AIO PACKAGE = IO::AIO 525MODULE = IO::AIO PACKAGE = IO::AIO
467 526
468PROTOTYPES: ENABLE 527PROTOTYPES: ENABLE
469 528
470BOOT: 529BOOT:
471{ 530{
472 stash = gv_stashpv ("IO::AIO", 1); 531 static const struct {
473 532 const char *name;
474 newCONSTSUB (stash, "EXDEV", newSViv (EXDEV)); 533 IV iv;
475 newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY)); 534 } *civ, const_iv[] = {
476 newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY)); 535# define const_iv(name, value) { # name, (IV) value },
477 newCONSTSUB (stash, "O_CREAT", newSViv (O_CREAT)); 536# define const_eio(name) { # name, (IV) EIO_ ## name },
478 newCONSTSUB (stash, "O_TRUNC", newSViv (O_TRUNC)); 537 const_iv (EXDEV , EXDEV)
538 const_iv (ENOSYS , ENOSYS)
539 const_iv (O_RDONLY, O_RDONLY)
540 const_iv (O_WRONLY, O_WRONLY)
541 const_iv (O_CREAT , O_CREAT)
542 const_iv (O_TRUNC , O_TRUNC)
479#ifdef _WIN32 543#ifndef _WIN32
480 X_MUTEX_CHECK (wrklock); 544 const_iv (S_IFIFO , S_IFIFO)
481 X_MUTEX_CHECK (reslock);
482 X_MUTEX_CHECK (reqlock);
483 X_MUTEX_CHECK (reqwait);
484 X_MUTEX_CHECK (preadwritelock);
485 X_MUTEX_CHECK (readdirlock);
486
487 X_COND_CHECK (reqwait);
488#else
489 newCONSTSUB (stash, "S_IFIFO", newSViv (S_IFIFO));
490#endif 545#endif
546 const_iv (FADV_NORMAL , POSIX_FADV_NORMAL)
547 const_iv (FADV_SEQUENTIAL, POSIX_FADV_SEQUENTIAL)
548 const_iv (FADV_RANDOM , POSIX_FADV_RANDOM)
549 const_iv (FADV_NOREUSE , POSIX_FADV_NOREUSE)
550 const_iv (FADV_WILLNEED , POSIX_FADV_WILLNEED)
551 const_iv (FADV_DONTNEED , POSIX_FADV_DONTNEED)
491 552
553 const_eio (SYNC_FILE_RANGE_WAIT_BEFORE)
554 const_eio (SYNC_FILE_RANGE_WRITE)
555 const_eio (SYNC_FILE_RANGE_WAIT_AFTER)
556
557 const_eio (READDIR_DENTS)
558 const_eio (READDIR_DIRS_FIRST)
559 const_eio (READDIR_STAT_ORDER)
560 const_eio (READDIR_FOUND_UNKNOWN)
561
562 const_eio (DT_UNKNOWN)
563 const_eio (DT_FIFO)
564 const_eio (DT_CHR)
565 const_eio (DT_DIR)
566 const_eio (DT_BLK)
567 const_eio (DT_REG)
568 const_eio (DT_LNK)
569 const_eio (DT_SOCK)
570 const_eio (DT_WHT)
571 };
572
573 stash = gv_stashpv ("IO::AIO", 1);
574
575 for (civ = const_iv + sizeof (const_iv) / sizeof (const_iv [0]); civ-- > const_iv; )
576 newCONSTSUB (stash, (char *)civ->name, newSViv (civ->iv));
577
492 create_respipe (); 578 create_respipe ();
493 579
494 if (eio_init (want_poll, done_poll) < 0) 580 if (eio_init (want_poll, done_poll) < 0)
495 croak ("IO::AIO: unable to initialise eio library"); 581 croak ("IO::AIO: unable to initialise eio library");
496 582
497 /* atfork child called in fifo order, so before eio's handler */ 583 /* atfork child called in fifo order, so before eio's handler */
498 X_THREAD_ATFORK (0, 0, atfork_child); 584 X_THREAD_ATFORK (0, 0, atfork_child);
499} 585}
500 586
501void 587void
502max_poll_reqs (int nreqs) 588max_poll_reqs (int nreqs)
503 PROTOTYPE: $ 589 PROTOTYPE: $
556 ALIAS: 642 ALIAS:
557 aio_fsync = EIO_FSYNC 643 aio_fsync = EIO_FSYNC
558 aio_fdatasync = EIO_FDATASYNC 644 aio_fdatasync = EIO_FDATASYNC
559 PPCODE: 645 PPCODE:
560{ 646{
647 int fd = s_fileno_croak (fh, 0);
561 dREQ; 648 dREQ;
562 649
563 req->type = ix; 650 req->type = ix;
564 req->sv1 = newSVsv (fh); 651 req->sv1 = newSVsv (fh);
565 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh))); 652 req->int1 = fd;
653
654 REQ_SEND (req);
655}
656
657void
658aio_sync_file_range (SV *fh, off_t offset, size_t nbytes, UV flags, SV *callback=&PL_sv_undef)
659 PROTOTYPE: $$$$;$
660 PPCODE:
661{
662 int fd = s_fileno_croak (fh, 0);
663 dREQ;
664
665 req->type = EIO_SYNC_FILE_RANGE;
666 req->sv1 = newSVsv (fh);
667 req->int1 = fd;
668 req->offs = offset;
669 req->size = nbytes;
670 req->int2 = flags;
566 671
567 REQ_SEND (req); 672 REQ_SEND (req);
568} 673}
569 674
570void 675void
571aio_close (SV *fh, SV *callback=&PL_sv_undef) 676aio_close (SV *fh, SV *callback=&PL_sv_undef)
572 PROTOTYPE: $;$ 677 PROTOTYPE: $;$
573 PPCODE: 678 PPCODE:
574{ 679{
575 static int close_pipe = -1; /* dummy fd to close fds via dup2 */ 680 static int close_pipe = -1; /* dummy fd to close fds via dup2 */
681 int fd = s_fileno_croak (fh, 0);
576 dREQ; 682 dREQ;
577 683
578 if (close_pipe < 0) 684 if (close_pipe < 0)
579 { 685 {
580 int pipefd [2]; 686 int pipefd [2];
588 } 694 }
589 695
590 req->type = EIO_DUP2; 696 req->type = EIO_DUP2;
591 req->int1 = close_pipe; 697 req->int1 = close_pipe;
592 req->sv2 = newSVsv (fh); 698 req->sv2 = newSVsv (fh);
593 req->int2 = PerlIO_fileno (IoIFP (sv_2io (fh))); 699 req->int2 = fd;
594 700
595 REQ_SEND (req); 701 REQ_SEND (req);
596} 702}
597 703
598void 704void
602 aio_write = EIO_WRITE 708 aio_write = EIO_WRITE
603 PROTOTYPE: $$$$$;$ 709 PROTOTYPE: $$$$$;$
604 PPCODE: 710 PPCODE:
605{ 711{
606 STRLEN svlen; 712 STRLEN svlen;
713 int fd = s_fileno_croak (fh, ix == EIO_WRITE);
607 char *svptr = SvPVbyte (data, svlen); 714 char *svptr = SvPVbyte (data, svlen);
608 UV len = SvUV (length); 715 UV len = SvUV (length);
609
610 SvUPGRADE (data, SVt_PV);
611 SvPOK_on (data);
612 716
613 if (dataoffset < 0) 717 if (dataoffset < 0)
614 dataoffset += svlen; 718 dataoffset += svlen;
615 719
616 if (dataoffset < 0 || dataoffset > svlen) 720 if (dataoffset < 0 || dataoffset > svlen)
622 if (!SvOK (length) || len + dataoffset > svlen) 726 if (!SvOK (length) || len + dataoffset > svlen)
623 len = svlen - dataoffset; 727 len = svlen - dataoffset;
624 } 728 }
625 else 729 else
626 { 730 {
627 /* read: grow scalar as necessary */ 731 /* read: check type and grow scalar as necessary */
732 SvUPGRADE (data, SVt_PV);
628 svptr = SvGROW (data, len + dataoffset + 1); 733 svptr = SvGROW (data, len + dataoffset + 1);
629 } 734 }
630
631 if (len < 0)
632 croak ("length must not be negative");
633 735
634 { 736 {
635 dREQ; 737 dREQ;
636 738
637 req->type = ix; 739 req->type = ix;
638 req->sv1 = newSVsv (fh); 740 req->sv1 = newSVsv (fh);
639 req->int1 = PerlIO_fileno (ix == EIO_READ ? IoIFP (sv_2io (fh)) 741 req->int1 = fd;
640 : IoOFP (sv_2io (fh)));
641 req->offs = SvOK (offset) ? SvVAL64 (offset) : -1; 742 req->offs = SvOK (offset) ? SvVAL64 (offset) : -1;
642 req->size = len; 743 req->size = len;
643 req->sv2 = SvREFCNT_inc (data); 744 req->sv2 = SvREFCNT_inc (data);
644 req->ptr2 = (char *)svptr + dataoffset; 745 req->ptr2 = (char *)svptr + dataoffset;
645 req->stroffset = dataoffset; 746 req->stroffset = dataoffset;
668 769
669 REQ_SEND; 770 REQ_SEND;
670} 771}
671 772
672void 773void
673aio_sendfile (SV *out_fh, SV *in_fh, SV *in_offset, UV length, SV *callback=&PL_sv_undef) 774aio_sendfile (SV *out_fh, SV *in_fh, off_t in_offset, size_t length, SV *callback=&PL_sv_undef)
674 PROTOTYPE: $$$$;$ 775 PROTOTYPE: $$$$;$
675 PPCODE: 776 PPCODE:
676{ 777{
778 int ifd = s_fileno_croak (in_fh , 0);
779 int ofd = s_fileno_croak (out_fh, 1);
677 dREQ; 780 dREQ;
678 781
679 req->type = EIO_SENDFILE; 782 req->type = EIO_SENDFILE;
680 req->sv1 = newSVsv (out_fh); 783 req->sv1 = newSVsv (out_fh);
681 req->int1 = PerlIO_fileno (IoIFP (sv_2io (out_fh))); 784 req->int1 = ofd;
682 req->sv2 = newSVsv (in_fh); 785 req->sv2 = newSVsv (in_fh);
683 req->int2 = PerlIO_fileno (IoIFP (sv_2io (in_fh))); 786 req->int2 = ifd;
684 req->offs = SvVAL64 (in_offset); 787 req->offs = in_offset;
685 req->size = length; 788 req->size = length;
686 789
687 REQ_SEND; 790 REQ_SEND;
688} 791}
689 792
690void 793void
691aio_readahead (SV *fh, SV *offset, IV length, SV *callback=&PL_sv_undef) 794aio_readahead (SV *fh, off_t offset, size_t length, SV *callback=&PL_sv_undef)
692 PROTOTYPE: $$$;$ 795 PROTOTYPE: $$$;$
693 PPCODE: 796 PPCODE:
694{ 797{
798 int fd = s_fileno_croak (fh, 0);
695 dREQ; 799 dREQ;
696 800
697 req->type = EIO_READAHEAD; 801 req->type = EIO_READAHEAD;
698 req->sv1 = newSVsv (fh); 802 req->sv1 = newSVsv (fh);
699 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh))); 803 req->int1 = fd;
700 req->offs = SvVAL64 (offset); 804 req->offs = offset;
701 req->size = length; 805 req->size = length;
702 806
703 REQ_SEND; 807 REQ_SEND;
704} 808}
705 809
787 req->int2 = mode; 891 req->int2 = mode;
788 req->sv1 = newSVsv (fh_or_path); 892 req->sv1 = newSVsv (fh_or_path);
789 893
790 if (SvPOK (req->sv1)) 894 if (SvPOK (req->sv1))
791 { 895 {
896 req->type = ix;
897 req->ptr1 = SvPVbyte_nolen (req->sv1);
898 }
899 else
900 {
792 req->type = EIO_FCHMOD; 901 req->type = EIO_FCHMOD;
793 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path))); 902 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
794 }
795 else
796 {
797 req->type = ix;
798 req->ptr1 = SvPVbyte_nolen (req->sv1);
799 } 903 }
800 904
801 REQ_SEND; 905 REQ_SEND;
802} 906}
803 907
821 req->type = EIO_FCHOWN; 925 req->type = EIO_FCHOWN;
822 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path))); 926 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
823 } 927 }
824 928
825 REQ_SEND; 929 REQ_SEND;
930}
931
932void
933aio_readdirx (SV8 *pathname, IV flags, SV *callback=&PL_sv_undef)
934 PPCODE:
935{
936 dREQ;
937
938 req->type = EIO_READDIR;
939 req->sv1 = newSVsv (pathname);
940 req->ptr1 = SvPVbyte_nolen (req->sv1);
941 req->int1 = flags | EIO_READDIR_DENTS | EIO_READDIR_CUSTOM1;
942
943 if (flags & EIO_READDIR_DENTS)
944 req->int1 |= EIO_READDIR_CUSTOM2;
945
946 REQ_SEND;
826} 947}
827 948
828void 949void
829aio_unlink (SV8 *pathname, SV *callback=&PL_sv_undef) 950aio_unlink (SV8 *pathname, SV *callback=&PL_sv_undef)
830 ALIAS: 951 ALIAS:
895{ 1016{
896 dREQ; 1017 dREQ;
897 1018
898 req->type = EIO_GROUP; 1019 req->type = EIO_GROUP;
899 1020
900 eio_submit (req); 1021 req_submit (req);
901 XPUSHs (req_sv (req, AIO_GRP_KLASS)); 1022 XPUSHs (req_sv (req, AIO_GRP_KLASS));
902} 1023}
903 1024
904void 1025void
905aio_nop (SV *callback=&PL_sv_undef) 1026aio_nop (SV *callback=&PL_sv_undef)
917 1038
918int 1039int
919aioreq_pri (int pri = 0) 1040aioreq_pri (int pri = 0)
920 PROTOTYPE: ;$ 1041 PROTOTYPE: ;$
921 CODE: 1042 CODE:
922 RETVAL = next_pri - EIO_PRI_BIAS; 1043 RETVAL = next_pri;
923 if (items > 0) 1044 if (items > 0)
924 { 1045 {
925 if (pri < EIO_PRI_MIN) pri = EIO_PRI_MIN; 1046 if (pri < EIO_PRI_MIN) pri = EIO_PRI_MIN;
926 if (pri > EIO_PRI_MAX) pri = EIO_PRI_MAX; 1047 if (pri > EIO_PRI_MAX) pri = EIO_PRI_MAX;
927 next_pri = pri + EIO_PRI_BIAS; 1048 next_pri = pri;
928 } 1049 }
929 OUTPUT: 1050 OUTPUT:
930 RETVAL 1051 RETVAL
931 1052
932void 1053void
933aioreq_nice (int nice = 0) 1054aioreq_nice (int nice = 0)
934 CODE: 1055 CODE:
935 nice = next_pri - nice; 1056 nice = next_pri - nice;
936 if (nice < EIO_PRI_MIN) nice = EIO_PRI_MIN; 1057 if (nice < EIO_PRI_MIN) nice = EIO_PRI_MIN;
937 if (nice > EIO_PRI_MAX) nice = EIO_PRI_MAX; 1058 if (nice > EIO_PRI_MAX) nice = EIO_PRI_MAX;
938 next_pri = nice + EIO_PRI_BIAS; 1059 next_pri = nice;
939 1060
940void 1061void
941flush () 1062flush ()
942 PROTOTYPE: 1063 PROTOTYPE:
943 CODE: 1064 CODE:
958 1079
959int 1080int
960poll_fileno() 1081poll_fileno()
961 PROTOTYPE: 1082 PROTOTYPE:
962 CODE: 1083 CODE:
963 RETVAL = respipe [0]; 1084 RETVAL = s_epipe_fd (&respipe);
964 OUTPUT: 1085 OUTPUT:
965 RETVAL 1086 RETVAL
966 1087
967int 1088int
968poll_cb(...) 1089poll_cb(...)
1004 1125
1005int 1126int
1006nthreads() 1127nthreads()
1007 PROTOTYPE: 1128 PROTOTYPE:
1008 CODE: 1129 CODE:
1009 if (WORDACCESS_UNSAFE) X_LOCK (wrklock); 1130 RETVAL = eio_nthreads ();
1010 RETVAL = started;
1011 if (WORDACCESS_UNSAFE) X_UNLOCK (wrklock);
1012 OUTPUT: 1131 OUTPUT:
1013 RETVAL 1132 RETVAL
1014 1133
1134int
1135fadvise (aio_rfd fh, off_t offset, off_t length, IV advice)
1136 PROTOTYPE: $$$$
1137 CODE:
1138#if _XOPEN_SOURCE >= 600 && !NO_FADVISE
1139 RETVAL = posix_fadvise (fh, offset, length, advice);
1140#else
1141 RETVAL = errno = ENOSYS;
1142#endif
1143 OUTPUT:
1144 RETVAL
1145
1146ssize_t
1147sendfile (aio_wfd ofh, aio_rfd ifh, off_t offset, size_t count)
1148 PROTOTYPE: $$$$
1149 CODE:
1150 eio_sendfile_sync (ofh, ifh, offset, count);
1151
1015void _on_next_submit (SV *cb) 1152void _on_next_submit (SV *cb)
1016 CODE: 1153 CODE:
1017 SvREFCNT_dec (on_next_submit); 1154 SvREFCNT_dec (on_next_submit);
1018 on_next_submit = SvOK (cb) ? newSVsv (cb) : 0; 1155 on_next_submit = SvOK (cb) ? newSVsv (cb) : 0;
1019 1156
1026 CODE: 1163 CODE:
1027 eio_cancel (req); 1164 eio_cancel (req);
1028 1165
1029void 1166void
1030cb (aio_req_ornot req, SV *callback=&PL_sv_undef) 1167cb (aio_req_ornot req, SV *callback=&PL_sv_undef)
1031 CODE: 1168 PPCODE:
1169{
1170 if (GIMME_V != G_VOID)
1171 XPUSHs (req->callback ? sv_2mortal (newRV_inc (req->callback)) : &PL_sv_undef);
1172
1173 if (items > 1)
1174 {
1175 SV *cb_cv =get_cb (callback);
1176
1032 SvREFCNT_dec (req->callback); 1177 SvREFCNT_dec (req->callback);
1033 req->callback = newSVsv (callback); 1178 req->callback = SvREFCNT_inc (cb_cv);
1179 }
1180}
1034 1181
1035MODULE = IO::AIO PACKAGE = IO::AIO::GRP 1182MODULE = IO::AIO PACKAGE = IO::AIO::GRP
1036 1183
1037void 1184void
1038add (aio_req grp, ...) 1185add (aio_req grp, ...)
1070 AV *av; 1217 AV *av;
1071 1218
1072 grp->errorno = errno; 1219 grp->errorno = errno;
1073 1220
1074 av = newAV (); 1221 av = newAV ();
1222 av_extend (av, items - 1);
1075 1223
1076 for (i = 1; i < items; ++i ) 1224 for (i = 1; i < items; ++i )
1077 av_push (av, newSVsv (ST (i))); 1225 av_push (av, newSVsv (ST (i)));
1078 1226
1079 SvREFCNT_dec (grp->sv1); 1227 SvREFCNT_dec (grp->sv1);
1093void 1241void
1094feed (aio_req grp, SV *callback=&PL_sv_undef) 1242feed (aio_req grp, SV *callback=&PL_sv_undef)
1095 CODE: 1243 CODE:
1096{ 1244{
1097 SvREFCNT_dec (grp->sv2); 1245 SvREFCNT_dec (grp->sv2);
1098 grp->sv2 = newSVsv (callback); 1246 grp->sv2 = newSVsv (callback);
1247 grp->feed = aio_grp_feed;
1099 1248
1100 if (grp->int2 <= 0) 1249 if (grp->int2 <= 0)
1101 grp->int2 = 2; 1250 grp->int2 = 2;
1102 1251
1103 eio_grp_limit (grp, grp->int2); 1252 eio_grp_limit (grp, grp->int2);

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines