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.103 by root, Sun Jul 8 09:09:34 2007 UTC vs.
Revision 1.107 by root, Wed Oct 3 21:27:51 2007 UTC

… …
24 24
25// perl overrides all those nice win32 functions 25// perl overrides all those nice win32 functions
26# undef open 26# undef open
27# undef read 27# undef read
28# undef write 28# undef write
29# undef send
30# undef recv
29# undef stat 31# undef stat
30# undef fstat 32# undef fstat
31# define lstat stat 33# define lstat stat
32# undef truncate 34# undef truncate
33# undef ftruncate 35# undef ftruncate
… …
94#if IVSIZE >= 8 96#if IVSIZE >= 8
95# define SvVAL64 SvIV 97# define SvVAL64 SvIV
96#else 98#else
97# define SvVAL64 SvNV 99# define SvVAL64 SvNV
98#endif 100#endif
101
102static HV *stash;
99 103
100#define dBUF \ 104#define dBUF \
101 char *aio_buf; \ 105 char *aio_buf; \
102 X_LOCK (wrklock); \ 106 X_LOCK (wrklock); \
103 self->dbuf = aio_buf = malloc (AIO_BUFSIZE); \ 107 self->dbuf = aio_buf = malloc (AIO_BUFSIZE); \
… …
262} 266}
263 267
264static volatile unsigned int nreqs, nready, npending; 268static volatile unsigned int nreqs, nready, npending;
265static volatile unsigned int max_idle = 4; 269static volatile unsigned int max_idle = 4;
266static volatile unsigned int max_outstanding = 0xffffffff; 270static volatile unsigned int max_outstanding = 0xffffffff;
267static int respipe [2]; 271static int respipe_osf [2], respipe [2] = { -1, -1 };
268 272
269static mutex_t reslock = X_MUTEX_INIT; 273static mutex_t reslock = X_MUTEX_INIT;
270static mutex_t reqlock = X_MUTEX_INIT; 274static mutex_t reqlock = X_MUTEX_INIT;
271static cond_t reqwait = X_COND_INIT; 275static cond_t reqwait = X_COND_INIT;
272 276
… …
499 break; 503 break;
500 504
501 case REQ_OPEN: 505 case REQ_OPEN:
502 { 506 {
503 /* convert fd to fh */ 507 /* convert fd to fh */
504 SV *fh; 508 SV *fh = &PL_sv_undef;
505 509
506 PUSHs (sv_2mortal (newSViv (req->result))); 510 if (req->result >= 0)
507 PUTBACK; 511 {
508 call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL); 512 GV *gv = (GV *)sv_newmortal ();
509 SPAGAIN; 513 int flags = req->int1 & (O_RDONLY | O_WRONLY | O_RDWR);
514 char sym [64];
515 int symlen;
516
517 symlen = snprintf (sym, sizeof (sym), "fd#%d", req->result);
518 gv_init (gv, stash, sym, symlen, 0);
510 519
520 symlen = snprintf (
521 sym,
522 sizeof (sym),
523 "%s&=%d",
524 flags == O_RDONLY ? "<" : flags == O_WRONLY ? ">" : "+<",
525 req->result
526 );
527
528 if (do_open (gv, sym, symlen, 0, 0, 0, 0))
511 fh = POPs; 529 fh = gv;
512 PUSHMARK (SP); 530 }
531
513 XPUSHs (fh); 532 XPUSHs (fh);
514 } 533 }
515 break; 534 break;
516 535
517 case REQ_GROUP: 536 case REQ_GROUP:
… …
623static void req_cancel (aio_req req) 642static void req_cancel (aio_req req)
624{ 643{
625 req->flags |= FLAG_CANCELLED; 644 req->flags |= FLAG_CANCELLED;
626 645
627 req_cancel_subs (req); 646 req_cancel_subs (req);
647}
648
649#ifdef USE_SOCKETS_AS_HANDLES
650# define TO_SOCKET(x) (win32_get_osfhandle (x))
651#else
652# define TO_SOCKET(x) (x)
653#endif
654
655static void
656create_respipe ()
657{
658 int old_readfd = respipe [0];
659
660 if (respipe [1] >= 0)
661 respipe_close (TO_SOCKET (respipe [1]));
662
663#ifdef _WIN32
664 if (PerlSock_socketpair (AF_UNIX, SOCK_STREAM, 0, respipe))
665#else
666 if (pipe (respipe))
667#endif
668 croak ("unable to initialize result pipe");
669
670 if (old_readfd >= 0)
671 {
672 if (dup2 (TO_SOCKET (respipe [0]), TO_SOCKET (old_readfd)) < 0)
673 croak ("unable to initialize result pipe(2)");
674
675 respipe_close (respipe [0]);
676 respipe [0] = old_readfd;
677 }
678
679#ifdef _WIN32
680 int arg = 1;
681 if (ioctlsocket (TO_SOCKET (respipe [0]), FIONBIO, &arg)
682 || ioctlsocket (TO_SOCKET (respipe [1]), FIONBIO, &arg))
683#else
684 if (fcntl (respipe [0], F_SETFL, O_NONBLOCK)
685 || fcntl (respipe [1], F_SETFL, O_NONBLOCK))
686#endif
687 croak ("unable to initialize result pipe(3)");
688
689 respipe_osf [0] = TO_SOCKET (respipe [0]);
690 respipe_osf [1] = TO_SOCKET (respipe [1]);
628} 691}
629 692
630X_THREAD_PROC (aio_proc); 693X_THREAD_PROC (aio_proc);
631 694
632static void start_thread (void) 695static void start_thread (void)
… …
736 if (size) 799 if (size)
737 return; 800 return;
738 801
739 maybe_start_thread (); 802 maybe_start_thread ();
740 803
741 FD_ZERO(&rfd); 804 FD_ZERO (&rfd);
742 FD_SET(respipe [0], &rfd); 805 FD_SET (respipe [0], &rfd);
743 806
744 select (respipe [0] + 1, &rfd, 0, 0, 0); 807 PerlSock_select (respipe [0] + 1, &rfd, 0, 0, 0);
745 } 808 }
746} 809}
747 810
748static int poll_cb () 811static int poll_cb ()
749{ 812{
… …
774 837
775 if (!res_queue.size) 838 if (!res_queue.size)
776 { 839 {
777 /* read any signals sent by the worker threads */ 840 /* read any signals sent by the worker threads */
778 char buf [4]; 841 char buf [4];
779 while (read (respipe [0], buf, 4) == 4) 842 while (respipe_read (respipe [0], buf, 4) == 4)
780 ; 843 ;
781 } 844 }
782 } 845 }
783 846
784 X_UNLOCK (reslock); 847 X_UNLOCK (reslock);
… …
1192 case REQ_FCHMOD: req->result = fchmod (req->int1, req->mode); break; 1255 case REQ_FCHMOD: req->result = fchmod (req->int1, req->mode); break;
1193 case REQ_TRUNCATE: req->result = truncate (req->ptr1, req->offs); break; 1256 case REQ_TRUNCATE: req->result = truncate (req->ptr1, req->offs); break;
1194 case REQ_FTRUNCATE: req->result = ftruncate (req->int1, req->offs); break; 1257 case REQ_FTRUNCATE: req->result = ftruncate (req->int1, req->offs); break;
1195 1258
1196 case REQ_OPEN: req->result = open (req->ptr1, req->int1, req->mode); break; 1259 case REQ_OPEN: req->result = open (req->ptr1, req->int1, req->mode); break;
1197 case REQ_CLOSE: req->result = close (req->int1); break; 1260 case REQ_CLOSE: req->result = PerlIO_close ((PerlIO *)req->ptr1); break;
1198 case REQ_UNLINK: req->result = unlink (req->ptr1); break; 1261 case REQ_UNLINK: req->result = unlink (req->ptr1); break;
1199 case REQ_RMDIR: req->result = rmdir (req->ptr1); break; 1262 case REQ_RMDIR: req->result = rmdir (req->ptr1); break;
1200 case REQ_MKDIR: req->result = mkdir (req->ptr1, req->mode); break; 1263 case REQ_MKDIR: req->result = mkdir (req->ptr1, req->mode); break;
1201 case REQ_RENAME: req->result = rename (req->ptr2, req->ptr1); break; 1264 case REQ_RENAME: req->result = rename (req->ptr2, req->ptr1); break;
1202 case REQ_LINK: req->result = link (req->ptr2, req->ptr1); break; 1265 case REQ_LINK: req->result = link (req->ptr2, req->ptr1); break;
… …
1266 ++npending; 1329 ++npending;
1267 1330
1268 if (!reqq_push (&res_queue, req)) 1331 if (!reqq_push (&res_queue, req))
1269 { 1332 {
1270 /* write a dummy byte to the pipe so fh becomes ready */ 1333 /* write a dummy byte to the pipe so fh becomes ready */
1271 write (respipe [1], &respipe, 1); 1334 respipe_write (respipe_osf [1], (const void *)&respipe_osf, 1);
1272 1335
1273 /* optionally signal the main thread asynchronously */ 1336 /* optionally signal the main thread asynchronously */
1274 if (main_sig) 1337 if (main_sig)
1275 pthread_kill (main_tid, main_sig); 1338 pthread_kill (main_tid, main_sig);
1276 } 1339 }
… …
1343 idle = 0; 1406 idle = 0;
1344 nreqs = 0; 1407 nreqs = 0;
1345 nready = 0; 1408 nready = 0;
1346 npending = 0; 1409 npending = 0;
1347 1410
1348 close (respipe [0]); 1411 create_respipe ();
1349 close (respipe [1]);
1350
1351 if (!create_pipe (respipe))
1352 croak ("cannot set result pipe to nonblocking mode");
1353 1412
1354 atfork_parent (); 1413 atfork_parent ();
1355} 1414}
1356 1415
1357#define dREQ \ 1416#define dREQ \
… …
1379 1438
1380PROTOTYPES: ENABLE 1439PROTOTYPES: ENABLE
1381 1440
1382BOOT: 1441BOOT:
1383{ 1442{
1384 HV *stash = gv_stashpv ("IO::AIO", 1); 1443 stash = gv_stashpv ("IO::AIO", 1);
1385 1444
1386 newCONSTSUB (stash, "EXDEV", newSViv (EXDEV)); 1445 newCONSTSUB (stash, "EXDEV", newSViv (EXDEV));
1387 newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY)); 1446 newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY));
1388 newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY)); 1447 newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY));
1389 newCONSTSUB (stash, "O_CREAT", newSViv (O_CREAT)); 1448 newCONSTSUB (stash, "O_CREAT", newSViv (O_CREAT));
… …
1393 X_MUTEX_CHECK (reslock); 1452 X_MUTEX_CHECK (reslock);
1394 X_MUTEX_CHECK (reqlock); 1453 X_MUTEX_CHECK (reqlock);
1395 X_MUTEX_CHECK (reqwait); 1454 X_MUTEX_CHECK (reqwait);
1396 X_MUTEX_CHECK (preadwritelock); 1455 X_MUTEX_CHECK (preadwritelock);
1397 X_MUTEX_CHECK (readdirlock); 1456 X_MUTEX_CHECK (readdirlock);
1457
1458 X_COND_CHECK (reqwait);
1398#else 1459#else
1399 newCONSTSUB (stash, "S_IFIFO", newSViv (S_IFIFO)); 1460 newCONSTSUB (stash, "S_IFIFO", newSViv (S_IFIFO));
1400 newCONSTSUB (stash, "SIGIO", newSViv (SIGIO)); 1461 newCONSTSUB (stash, "SIGIO", newSViv (SIGIO));
1401#endif 1462#endif
1402 1463
1403 if (!create_pipe (respipe)) 1464 create_respipe ();
1404 croak ("cannot set result pipe to nonblocking mode");
1405 1465
1406 X_THREAD_ATFORK (atfork_prepare, atfork_parent, atfork_child); 1466 X_THREAD_ATFORK (atfork_prepare, atfork_parent, atfork_child);
1407} 1467}
1408 1468
1409void 1469void
… …
1456 1516
1457 REQ_SEND; 1517 REQ_SEND;
1458} 1518}
1459 1519
1460void 1520void
1461aio_close (SV *fh, SV *callback=&PL_sv_undef) 1521aio_fsync (SV *fh, SV *callback=&PL_sv_undef)
1462 PROTOTYPE: $;$ 1522 PROTOTYPE: $;$
1463 ALIAS: 1523 ALIAS:
1464 aio_close = REQ_CLOSE
1465 aio_fsync = REQ_FSYNC 1524 aio_fsync = REQ_FSYNC
1466 aio_fdatasync = REQ_FDATASYNC 1525 aio_fdatasync = REQ_FDATASYNC
1467 PPCODE: 1526 PPCODE:
1468{ 1527{
1469 dREQ; 1528 dREQ;
… …
1471 req->type = ix; 1530 req->type = ix;
1472 req->sv1 = newSVsv (fh); 1531 req->sv1 = newSVsv (fh);
1473 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh))); 1532 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh)));
1474 1533
1475 REQ_SEND (req); 1534 REQ_SEND (req);
1535}
1536
1537void
1538aio_close (SV *fh, SV *callback=&PL_sv_undef)
1539 PROTOTYPE: $;$
1540 PPCODE:
1541{
1542 PerlIO *io = IoIFP (sv_2io (fh));
1543 int fd = PerlIO_fileno (io);
1544
1545 if (fd < 0)
1546 croak ("aio_close called with fd-less filehandle");
1547
1548 PerlIO_binmode (io, 0, 0, 0);
1549
1550 {
1551 dREQ;
1552
1553 req->type = REQ_CLOSE;
1554 req->sv1 = newSVsv (fh);
1555 req->ptr1 = (void *)io;
1556
1557 REQ_SEND (req);
1558 }
1476} 1559}
1477 1560
1478void 1561void
1479aio_read (SV *fh, SV *offset, SV *length, SV8 *data, IV dataoffset, SV *callback=&PL_sv_undef) 1562aio_read (SV *fh, SV *offset, SV *length, SV8 *data, IV dataoffset, SV *callback=&PL_sv_undef)
1480 ALIAS: 1563 ALIAS:

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines