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

Comparing IO-AIO/AIO.pm (file contents):
Revision 1.87 by root, Sun Oct 29 00:52:02 2006 UTC vs.
Revision 1.116 by root, Wed Oct 3 21:27:51 2007 UTC

… …
5=head1 SYNOPSIS 5=head1 SYNOPSIS
6 6
7 use IO::AIO; 7 use IO::AIO;
8 8
9 aio_open "/etc/passwd", O_RDONLY, 0, sub { 9 aio_open "/etc/passwd", O_RDONLY, 0, sub {
10 my ($fh) = @_; 10 my $fh = shift
11 or die "/etc/passwd: $!";
11 ... 12 ...
12 }; 13 };
13 14
14 aio_unlink "/tmp/file", sub { }; 15 aio_unlink "/tmp/file", sub { };
15 16
… …
61etc.), but can also be used to easily do operations in parallel that are 62etc.), but can also be used to easily do operations in parallel that are
62normally done sequentially, e.g. stat'ing many files, which is much faster 63normally done sequentially, e.g. stat'ing many files, which is much faster
63on a RAID volume or over NFS when you do a number of stat operations 64on a RAID volume or over NFS when you do a number of stat operations
64concurrently. 65concurrently.
65 66
66While this works on all types of file descriptors (for example sockets), 67While most of this works on all types of file descriptors (for
67using these functions on file descriptors that support nonblocking 68example sockets), using these functions on file descriptors that
68operation (again, sockets, pipes etc.) is very inefficient. Use an event 69support nonblocking operation (again, sockets, pipes etc.) is very
69loop for that (such as the L<Event|Event> module): IO::AIO will naturally 70inefficient. Use an event loop for that (such as the L<Event|Event>
70fit into such an event loop itself. 71module): IO::AIO will naturally fit into such an event loop itself.
71 72
72In this version, a number of threads are started that execute your 73In this version, a number of threads are started that execute your
73requests and signal their completion. You don't need thread support 74requests and signal their completion. You don't need thread support
74in perl, and the threads created by this module will not be visible 75in perl, and the threads created by this module will not be visible
75to perl. In the future, this module might make use of the native aio 76to perl. In the future, this module might make use of the native aio
… …
77not well-supported or restricted (GNU/Linux doesn't allow them on normal 78not well-supported or restricted (GNU/Linux doesn't allow them on normal
78files currently, for example), and they would only support aio_read and 79files currently, for example), and they would only support aio_read and
79aio_write, so the remaining functionality would have to be implemented 80aio_write, so the remaining functionality would have to be implemented
80using threads anyway. 81using threads anyway.
81 82
82Although the module will work with in the presence of other (Perl-) 83Although the module will work in the presence of other (Perl-) threads,
83threads, it is currently not reentrant in any way, so use appropriate 84it is currently not reentrant in any way, so use appropriate locking
84locking yourself, always call C<poll_cb> from within the same thread, or 85yourself, always call C<poll_cb> from within the same thread, or never
85never call C<poll_cb> (or other C<aio_> functions) recursively. 86call C<poll_cb> (or other C<aio_> functions) recursively.
86 87
87=head2 EXAMPLE 88=head2 EXAMPLE
88 89
89This is a simple example that uses the Event module and loads 90This is a simple example that uses the Event module and loads
90F</etc/passwd> asynchronously: 91F</etc/passwd> asynchronously:
… …
98 poll => 'r', 99 poll => 'r',
99 cb => \&IO::AIO::poll_cb); 100 cb => \&IO::AIO::poll_cb);
100 101
101 # queue the request to open /etc/passwd 102 # queue the request to open /etc/passwd
102 aio_open "/etc/passwd", O_RDONLY, 0, sub { 103 aio_open "/etc/passwd", O_RDONLY, 0, sub {
103 my $fh = $_[0] 104 my $fh = shift
104 or die "error while opening: $!"; 105 or die "error while opening: $!";
105 106
106 # stat'ing filehandles is generally non-blocking 107 # stat'ing filehandles is generally non-blocking
107 my $size = -s $fh; 108 my $size = -s $fh;
108 109
… …
176Request has reached the end of its lifetime and holds no resources anymore 177Request has reached the end of its lifetime and holds no resources anymore
177(except possibly for the Perl object, but its connection to the actual 178(except possibly for the Perl object, but its connection to the actual
178aio request is severed and calling its methods will either do nothing or 179aio request is severed and calling its methods will either do nothing or
179result in a runtime error). 180result in a runtime error).
180 181
182=back
183
181=cut 184=cut
182 185
183package IO::AIO; 186package IO::AIO;
184 187
185no warnings; 188no warnings;
186use strict 'vars'; 189use strict 'vars';
187 190
188use base 'Exporter'; 191use base 'Exporter';
189 192
190BEGIN { 193BEGIN {
191 our $VERSION = '2.1'; 194 our $VERSION = '2.5';
192 195
193 our @AIO_REQ = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat 196 our @AIO_REQ = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat
194 aio_lstat aio_unlink aio_rmdir aio_readdir aio_scandir aio_symlink 197 aio_lstat aio_unlink aio_rmdir aio_readdir aio_scandir aio_symlink
195 aio_fsync aio_fdatasync aio_readahead aio_rename aio_link aio_move 198 aio_readlink aio_fsync aio_fdatasync aio_readahead aio_rename aio_link
196 aio_copy aio_group aio_nop aio_mknod); 199 aio_move aio_copy aio_group aio_nop aio_mknod aio_load aio_rmtree aio_mkdir
200 aio_chown aio_chmod aio_utime aio_truncate);
197 our @EXPORT = (@AIO_REQ, qw(aioreq_pri aioreq_nice)); 201 our @EXPORT = (@AIO_REQ, qw(aioreq_pri aioreq_nice aio_block));
198 our @EXPORT_OK = qw(poll_fileno poll_cb poll_wait flush 202 our @EXPORT_OK = qw(poll_fileno poll_cb poll_wait flush
199 min_parallel max_parallel max_idle 203 min_parallel max_parallel max_idle
200 nreqs nready npending nthreads 204 nreqs nready npending nthreads
201 max_poll_time max_poll_reqs); 205 max_poll_time max_poll_reqs);
202 206
… …
267 aio_read $_[0], ..., sub { 271 aio_read $_[0], ..., sub {
268 ... 272 ...
269 }; 273 };
270 }; 274 };
271 275
276
272=item aioreq_nice $pri_adjust 277=item aioreq_nice $pri_adjust
273 278
274Similar to C<aioreq_pri>, but subtracts the given value from the current 279Similar to C<aioreq_pri>, but subtracts the given value from the current
275priority, so the effect is cumulative. 280priority, so the effect is cumulative.
281
276 282
277=item aio_open $pathname, $flags, $mode, $callback->($fh) 283=item aio_open $pathname, $flags, $mode, $callback->($fh)
278 284
279Asynchronously open or create a file and call the callback with a newly 285Asynchronously open or create a file and call the callback with a newly
280created filehandle for the file. 286created filehandle for the file.
… …
286list. They are the same as used by C<sysopen>. 292list. They are the same as used by C<sysopen>.
287 293
288Likewise, C<$mode> specifies the mode of the newly created file, if it 294Likewise, C<$mode> specifies the mode of the newly created file, if it
289didn't exist and C<O_CREAT> has been given, just like perl's C<sysopen>, 295didn't exist and C<O_CREAT> has been given, just like perl's C<sysopen>,
290except that it is mandatory (i.e. use C<0> if you don't create new files, 296except that it is mandatory (i.e. use C<0> if you don't create new files,
291and C<0666> or C<0777> if you do). 297and C<0666> or C<0777> if you do). Note that the C<$mode> will be modified
298by the umask in effect then the request is being executed, so better never
299change the umask.
292 300
293Example: 301Example:
294 302
295 aio_open "/etc/passwd", O_RDONLY, 0, sub { 303 aio_open "/etc/passwd", O_RDONLY, 0, sub {
296 if ($_[0]) { 304 if ($_[0]) {
… …
299 } else { 307 } else {
300 die "open failed: $!\n"; 308 die "open failed: $!\n";
301 } 309 }
302 }; 310 };
303 311
312
304=item aio_close $fh, $callback->($status) 313=item aio_close $fh, $callback->($status)
305 314
306Asynchronously close a file and call the callback with the result 315Asynchronously close a file and call the callback with the result
307code. I<WARNING:> although accepted, you should not pass in a perl 316code.
308filehandle here, as perl will likely close the file descriptor another
309time when the filehandle is destroyed. Normally, you can safely call perls
310C<close> or just let filehandles go out of scope.
311 317
312This is supposed to be a bug in the API, so that might change. It's 318Unlike the other functions operating on files, this function uses the
313therefore best to avoid this function. 319PerlIO layer to close the filehandle. The reason is that the PerlIO API
320insists on closing the underlying fd itself, no matter what, and doesn't
321allow modifications to the fd. Unfortunately, it is not clear that you can
322call PerlIO from different threads (actually, its quite clear that this
323won't work in some cases), so while it likely works perfectly with simple
324file handles (such as the ones created by C<aio_open>) it might fail in
325interesting ways for others.
326
327Having said that, aio_close tries to clean up the filehandle as much as
328possible before handing it to an io thread, and generally does work.
329
314 330
315=item aio_read $fh,$offset,$length, $data,$dataoffset, $callback->($retval) 331=item aio_read $fh,$offset,$length, $data,$dataoffset, $callback->($retval)
316 332
317=item aio_write $fh,$offset,$length, $data,$dataoffset, $callback->($retval) 333=item aio_write $fh,$offset,$length, $data,$dataoffset, $callback->($retval)
318 334
319Reads or writes C<length> bytes from the specified C<fh> and C<offset> 335Reads or writes C<$length> bytes from the specified C<$fh> and C<$offset>
320into the scalar given by C<data> and offset C<dataoffset> and calls the 336into the scalar given by C<$data> and offset C<$dataoffset> and calls the
321callback without the actual number of bytes read (or -1 on error, just 337callback without the actual number of bytes read (or -1 on error, just
322like the syscall). 338like the syscall).
323 339
340If C<$offset> is undefined, then the current file descriptor offset will
341be used (and updated), otherwise the file descriptor offset will not be
342changed by these calls.
343
344If C<$length> is undefined in C<aio_write>, use the remaining length of C<$data>.
345
346If C<$dataoffset> is less than zero, it will be counted from the end of
347C<$data>.
348
324The C<$data> scalar I<MUST NOT> be modified in any way while the request 349The C<$data> scalar I<MUST NOT> be modified in any way while the request
325is outstanding. Modifying it can result in segfaults or WW3 (if the 350is outstanding. Modifying it can result in segfaults or World War III (if
326necessary/optional hardware is installed). 351the necessary/optional hardware is installed).
327 352
328Example: Read 15 bytes at offset 7 into scalar C<$buffer>, starting at 353Example: Read 15 bytes at offset 7 into scalar C<$buffer>, starting at
329offset C<0> within the scalar: 354offset C<0> within the scalar:
330 355
331 aio_read $fh, 7, 15, $buffer, 0, sub { 356 aio_read $fh, 7, 15, $buffer, 0, sub {
332 $_[0] > 0 or die "read error: $!"; 357 $_[0] > 0 or die "read error: $!";
333 print "read $_[0] bytes: <$buffer>\n"; 358 print "read $_[0] bytes: <$buffer>\n";
334 }; 359 };
360
335 361
336=item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback->($retval) 362=item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback->($retval)
337 363
338Tries to copy C<$length> bytes from C<$in_fh> to C<$out_fh>. It starts 364Tries to copy C<$length> bytes from C<$in_fh> to C<$out_fh>. It starts
339reading at byte offset C<$in_offset>, and starts writing at the current 365reading at byte offset C<$in_offset>, and starts writing at the current
… …
353C<$in_fh> than are written, and there is no way to find out how many 379C<$in_fh> than are written, and there is no way to find out how many
354bytes have been read from C<aio_sendfile> alone, as C<aio_sendfile> only 380bytes have been read from C<aio_sendfile> alone, as C<aio_sendfile> only
355provides the number of bytes written to C<$out_fh>. Only if the result 381provides the number of bytes written to C<$out_fh>. Only if the result
356value equals C<$length> one can assume that C<$length> bytes have been 382value equals C<$length> one can assume that C<$length> bytes have been
357read. 383read.
384
358 385
359=item aio_readahead $fh,$offset,$length, $callback->($retval) 386=item aio_readahead $fh,$offset,$length, $callback->($retval)
360 387
361C<aio_readahead> populates the page cache with data from a file so that 388C<aio_readahead> populates the page cache with data from a file so that
362subsequent reads from that file will not block on disk I/O. The C<$offset> 389subsequent reads from that file will not block on disk I/O. The C<$offset>
… …
368file. The current file offset of the file is left unchanged. 395file. The current file offset of the file is left unchanged.
369 396
370If that syscall doesn't exist (likely if your OS isn't Linux) it will be 397If that syscall doesn't exist (likely if your OS isn't Linux) it will be
371emulated by simply reading the data, which would have a similar effect. 398emulated by simply reading the data, which would have a similar effect.
372 399
400
373=item aio_stat $fh_or_path, $callback->($status) 401=item aio_stat $fh_or_path, $callback->($status)
374 402
375=item aio_lstat $fh, $callback->($status) 403=item aio_lstat $fh, $callback->($status)
376 404
377Works like perl's C<stat> or C<lstat> in void context. The callback will 405Works like perl's C<stat> or C<lstat> in void context. The callback will
… …
390 aio_stat "/etc/passwd", sub { 418 aio_stat "/etc/passwd", sub {
391 $_[0] and die "stat failed: $!"; 419 $_[0] and die "stat failed: $!";
392 print "size is ", -s _, "\n"; 420 print "size is ", -s _, "\n";
393 }; 421 };
394 422
423
424=item aio_utime $fh_or_path, $atime, $mtime, $callback->($status)
425
426Works like perl's C<utime> function (including the special case of $atime
427and $mtime being undef). Fractional times are supported if the underlying
428syscalls support them.
429
430When called with a pathname, uses utimes(2) if available, otherwise
431utime(2). If called on a file descriptor, uses futimes(2) if available,
432otherwise returns ENOSYS, so this is not portable.
433
434Examples:
435
436 # set atime and mtime to current time (basically touch(1)):
437 aio_utime "path", undef, undef;
438 # set atime to current time and mtime to beginning of the epoch:
439 aio_utime "path", time, undef; # undef==0
440
441
442=item aio_chown $fh_or_path, $uid, $gid, $callback->($status)
443
444Works like perl's C<chown> function, except that C<undef> for either $uid
445or $gid is being interpreted as "do not change" (but -1 can also be used).
446
447Examples:
448
449 # same as "chown root path" in the shell:
450 aio_chown "path", 0, -1;
451 # same as above:
452 aio_chown "path", 0, undef;
453
454
455=item aio_truncate $fh_or_path, $offset, $callback->($status)
456
457Works like truncate(2) or ftruncate(2).
458
459
460=item aio_chmod $fh_or_path, $mode, $callback->($status)
461
462Works like perl's C<chmod> function.
463
464
395=item aio_unlink $pathname, $callback->($status) 465=item aio_unlink $pathname, $callback->($status)
396 466
397Asynchronously unlink (delete) a file and call the callback with the 467Asynchronously unlink (delete) a file and call the callback with the
398result code. 468result code.
399 469
470
400=item aio_mknod $path, $mode, $dev, $callback->($status) 471=item aio_mknod $path, $mode, $dev, $callback->($status)
401 472
402[EXPERIMENTAL] 473[EXPERIMENTAL]
403 474
404Asynchronously create a device node (or fifo). See mknod(2). 475Asynchronously create a device node (or fifo). See mknod(2).
405 476
406The only (POSIX-) portable way of calling this function is: 477The only (POSIX-) portable way of calling this function is:
407 478
408 aio_mknod $path, IO::AIO::S_IFIFO | $mode, 0, sub { ... 479 aio_mknod $path, IO::AIO::S_IFIFO | $mode, 0, sub { ...
480
409 481
410=item aio_link $srcpath, $dstpath, $callback->($status) 482=item aio_link $srcpath, $dstpath, $callback->($status)
411 483
412Asynchronously create a new link to the existing object at C<$srcpath> at 484Asynchronously create a new link to the existing object at C<$srcpath> at
413the path C<$dstpath> and call the callback with the result code. 485the path C<$dstpath> and call the callback with the result code.
414 486
487
415=item aio_symlink $srcpath, $dstpath, $callback->($status) 488=item aio_symlink $srcpath, $dstpath, $callback->($status)
416 489
417Asynchronously create a new symbolic link to the existing object at C<$srcpath> at 490Asynchronously create a new symbolic link to the existing object at C<$srcpath> at
418the path C<$dstpath> and call the callback with the result code. 491the path C<$dstpath> and call the callback with the result code.
419 492
493
494=item aio_readlink $path, $callback->($link)
495
496Asynchronously read the symlink specified by C<$path> and pass it to
497the callback. If an error occurs, nothing or undef gets passed to the
498callback.
499
500
420=item aio_rename $srcpath, $dstpath, $callback->($status) 501=item aio_rename $srcpath, $dstpath, $callback->($status)
421 502
422Asynchronously rename the object at C<$srcpath> to C<$dstpath>, just as 503Asynchronously rename the object at C<$srcpath> to C<$dstpath>, just as
423rename(2) and call the callback with the result code. 504rename(2) and call the callback with the result code.
424 505
506
507=item aio_mkdir $pathname, $mode, $callback->($status)
508
509Asynchronously mkdir (create) a directory and call the callback with
510the result code. C<$mode> will be modified by the umask at the time the
511request is executed, so do not change your umask.
512
513
425=item aio_rmdir $pathname, $callback->($status) 514=item aio_rmdir $pathname, $callback->($status)
426 515
427Asynchronously rmdir (delete) a directory and call the callback with the 516Asynchronously rmdir (delete) a directory and call the callback with the
428result code. 517result code.
518
429 519
430=item aio_readdir $pathname, $callback->($entries) 520=item aio_readdir $pathname, $callback->($entries)
431 521
432Unlike the POSIX call of the same name, C<aio_readdir> reads an entire 522Unlike the POSIX call of the same name, C<aio_readdir> reads an entire
433directory (i.e. opendir + readdir + closedir). The entries will not be 523directory (i.e. opendir + readdir + closedir). The entries will not be
434sorted, and will B<NOT> include the C<.> and C<..> entries. 524sorted, and will B<NOT> include the C<.> and C<..> entries.
435 525
436The callback a single argument which is either C<undef> or an array-ref 526The callback a single argument which is either C<undef> or an array-ref
437with the filenames. 527with the filenames.
528
529
530=item aio_load $path, $data, $callback->($status)
531
532This is a composite request that tries to fully load the given file into
533memory. Status is the same as with aio_read.
534
535=cut
536
537sub aio_load($$;$) {
538 aio_block {
539 my ($path, undef, $cb) = @_;
540 my $data = \$_[1];
541
542 my $pri = aioreq_pri;
543 my $grp = aio_group $cb;
544
545 aioreq_pri $pri;
546 add $grp aio_open $path, O_RDONLY, 0, sub {
547 my $fh = shift
548 or return $grp->result (-1);
549
550 aioreq_pri $pri;
551 add $grp aio_read $fh, 0, (-s $fh), $$data, 0, sub {
552 $grp->result ($_[0]);
553 };
554 };
555
556 $grp
557 }
558}
438 559
439=item aio_copy $srcpath, $dstpath, $callback->($status) 560=item aio_copy $srcpath, $dstpath, $callback->($status)
440 561
441Try to copy the I<file> (directories not supported as either source or 562Try to copy the I<file> (directories not supported as either source or
442destination) from C<$srcpath> to C<$dstpath> and call the callback with 563destination) from C<$srcpath> to C<$dstpath> and call the callback with
… …
452errors are being ignored. 573errors are being ignored.
453 574
454=cut 575=cut
455 576
456sub aio_copy($$;$) { 577sub aio_copy($$;$) {
578 aio_block {
457 my ($src, $dst, $cb) = @_; 579 my ($src, $dst, $cb) = @_;
458 580
459 my $pri = aioreq_pri; 581 my $pri = aioreq_pri;
460 my $grp = aio_group $cb; 582 my $grp = aio_group $cb;
461 583
462 aioreq_pri $pri; 584 aioreq_pri $pri;
463 add $grp aio_open $src, O_RDONLY, 0, sub { 585 add $grp aio_open $src, O_RDONLY, 0, sub {
464 if (my $src_fh = $_[0]) { 586 if (my $src_fh = $_[0]) {
465 my @stat = stat $src_fh; 587 my @stat = stat $src_fh;
466 588
467 aioreq_pri $pri; 589 aioreq_pri $pri;
468 add $grp aio_open $dst, O_CREAT | O_WRONLY | O_TRUNC, 0200, sub { 590 add $grp aio_open $dst, O_CREAT | O_WRONLY | O_TRUNC, 0200, sub {
469 if (my $dst_fh = $_[0]) { 591 if (my $dst_fh = $_[0]) {
470 aioreq_pri $pri; 592 aioreq_pri $pri;
471 add $grp aio_sendfile $dst_fh, $src_fh, 0, $stat[7], sub { 593 add $grp aio_sendfile $dst_fh, $src_fh, 0, $stat[7], sub {
472 if ($_[0] == $stat[7]) { 594 if ($_[0] == $stat[7]) {
473 $grp->result (0); 595 $grp->result (0);
474 close $src_fh; 596 close $src_fh;
475 597
476 # those should not normally block. should. should. 598 # those should not normally block. should. should.
477 utime $stat[8], $stat[9], $dst; 599 utime $stat[8], $stat[9], $dst;
478 chmod $stat[2] & 07777, $dst_fh; 600 chmod $stat[2] & 07777, $dst_fh;
479 chown $stat[4], $stat[5], $dst_fh; 601 chown $stat[4], $stat[5], $dst_fh;
480 close $dst_fh; 602 close $dst_fh;
481 } else { 603 } else {
482 $grp->result (-1); 604 $grp->result (-1);
483 close $src_fh; 605 close $src_fh;
484 close $dst_fh; 606 close $dst_fh;
485 607
486 aioreq $pri; 608 aioreq $pri;
487 add $grp aio_unlink $dst; 609 add $grp aio_unlink $dst;
610 }
488 } 611 };
612 } else {
613 $grp->result (-1);
489 }; 614 }
490 } else {
491 $grp->result (-1);
492 } 615 },
616
617 } else {
618 $grp->result (-1);
493 }, 619 }
494
495 } else {
496 $grp->result (-1);
497 } 620 };
621
622 $grp
498 }; 623 }
499
500 $grp
501} 624}
502 625
503=item aio_move $srcpath, $dstpath, $callback->($status) 626=item aio_move $srcpath, $dstpath, $callback->($status)
504 627
505Try to move the I<file> (directories not supported as either source or 628Try to move the I<file> (directories not supported as either source or
… …
511that is successful, unlinking the C<$srcpath>. 634that is successful, unlinking the C<$srcpath>.
512 635
513=cut 636=cut
514 637
515sub aio_move($$;$) { 638sub aio_move($$;$) {
639 aio_block {
516 my ($src, $dst, $cb) = @_; 640 my ($src, $dst, $cb) = @_;
517 641
518 my $pri = aioreq_pri; 642 my $pri = aioreq_pri;
519 my $grp = aio_group $cb; 643 my $grp = aio_group $cb;
520 644
521 aioreq_pri $pri; 645 aioreq_pri $pri;
522 add $grp aio_rename $src, $dst, sub { 646 add $grp aio_rename $src, $dst, sub {
523 if ($_[0] && $! == EXDEV) { 647 if ($_[0] && $! == EXDEV) {
524 aioreq_pri $pri; 648 aioreq_pri $pri;
525 add $grp aio_copy $src, $dst, sub { 649 add $grp aio_copy $src, $dst, sub {
650 $grp->result ($_[0]);
651
652 if (!$_[0]) {
653 aioreq_pri $pri;
654 add $grp aio_unlink $src;
655 }
656 };
657 } else {
526 $grp->result ($_[0]); 658 $grp->result ($_[0]);
527
528 if (!$_[0]) {
529 aioreq_pri $pri;
530 add $grp aio_unlink $src;
531 }
532 }; 659 }
533 } else {
534 $grp->result ($_[0]);
535 } 660 };
661
662 $grp
536 }; 663 }
537
538 $grp
539} 664}
540 665
541=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs) 666=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)
542 667
543Scans a directory (similar to C<aio_readdir>) but additionally tries to 668Scans a directory (similar to C<aio_readdir>) but additionally tries to
… …
590as those tend to return 0 or 1 as link counts, which disables the 715as those tend to return 0 or 1 as link counts, which disables the
591directory counting heuristic. 716directory counting heuristic.
592 717
593=cut 718=cut
594 719
595sub aio_scandir($$$) { 720sub aio_scandir($$;$) {
721 aio_block {
596 my ($path, $maxreq, $cb) = @_; 722 my ($path, $maxreq, $cb) = @_;
597 723
598 my $pri = aioreq_pri; 724 my $pri = aioreq_pri;
599 725
600 my $grp = aio_group $cb; 726 my $grp = aio_group $cb;
601 727
602 $maxreq = 4 if $maxreq <= 0; 728 $maxreq = 4 if $maxreq <= 0;
603 729
604 # stat once 730 # stat once
605 aioreq_pri $pri;
606 add $grp aio_stat $path, sub {
607 return $grp->result () if $_[0];
608 my $now = time;
609 my $hash1 = join ":", (stat _)[0,1,3,7,9];
610
611 # read the directory entries
612 aioreq_pri $pri; 731 aioreq_pri $pri;
613 add $grp aio_readdir $path, sub { 732 add $grp aio_stat $path, sub {
614 my $entries = shift
615 or return $grp->result (); 733 return $grp->result () if $_[0];
734 my $now = time;
735 my $hash1 = join ":", (stat _)[0,1,3,7,9];
616 736
617 # stat the dir another time 737 # read the directory entries
618 aioreq_pri $pri; 738 aioreq_pri $pri;
739 add $grp aio_readdir $path, sub {
740 my $entries = shift
741 or return $grp->result ();
742
743 # stat the dir another time
744 aioreq_pri $pri;
619 add $grp aio_stat $path, sub { 745 add $grp aio_stat $path, sub {
620 my $hash2 = join ":", (stat _)[0,1,3,7,9]; 746 my $hash2 = join ":", (stat _)[0,1,3,7,9];
621 747
622 my $ndirs; 748 my $ndirs;
623 749
624 # take the slow route if anything looks fishy 750 # take the slow route if anything looks fishy
625 if ($hash1 ne $hash2 or (stat _)[9] == $now) { 751 if ($hash1 ne $hash2 or (stat _)[9] == $now) {
626 $ndirs = -1; 752 $ndirs = -1;
627 } else { 753 } else {
628 # if nlink == 2, we are finished 754 # if nlink == 2, we are finished
629 # on non-posix-fs's, we rely on nlink < 2 755 # on non-posix-fs's, we rely on nlink < 2
630 $ndirs = (stat _)[3] - 2 756 $ndirs = (stat _)[3] - 2
631 or return $grp->result ([], $entries); 757 or return $grp->result ([], $entries);
632 } 758 }
633 759
634 # sort into likely dirs and likely nondirs 760 # sort into likely dirs and likely nondirs
635 # dirs == files without ".", short entries first 761 # dirs == files without ".", short entries first
636 $entries = [map $_->[0], 762 $entries = [map $_->[0],
637 sort { $b->[1] cmp $a->[1] } 763 sort { $b->[1] cmp $a->[1] }
638 map [$_, sprintf "%s%04d", (/.\./ ? "1" : "0"), length], 764 map [$_, sprintf "%s%04d", (/.\./ ? "1" : "0"), length],
639 @$entries]; 765 @$entries];
640 766
641 my (@dirs, @nondirs); 767 my (@dirs, @nondirs);
642 768
643 my $statgrp = add $grp aio_group sub { 769 my $statgrp = add $grp aio_group sub {
644 $grp->result (\@dirs, \@nondirs); 770 $grp->result (\@dirs, \@nondirs);
645 }; 771 };
646 772
647 limit $statgrp $maxreq; 773 limit $statgrp $maxreq;
648 feed $statgrp sub { 774 feed $statgrp sub {
649 return unless @$entries; 775 return unless @$entries;
650 my $entry = pop @$entries; 776 my $entry = pop @$entries;
651 777
652 aioreq_pri $pri; 778 aioreq_pri $pri;
653 add $statgrp aio_stat "$path/$entry/.", sub { 779 add $statgrp aio_stat "$path/$entry/.", sub {
654 if ($_[0] < 0) { 780 if ($_[0] < 0) {
655 push @nondirs, $entry; 781 push @nondirs, $entry;
656 } else { 782 } else {
657 # need to check for real directory 783 # need to check for real directory
658 aioreq_pri $pri; 784 aioreq_pri $pri;
659 add $statgrp aio_lstat "$path/$entry", sub { 785 add $statgrp aio_lstat "$path/$entry", sub {
660 if (-d _) { 786 if (-d _) {
661 push @dirs, $entry; 787 push @dirs, $entry;
662 788
663 unless (--$ndirs) { 789 unless (--$ndirs) {
664 push @nondirs, @$entries; 790 push @nondirs, @$entries;
665 feed $statgrp; 791 feed $statgrp;
792 }
793 } else {
794 push @nondirs, $entry;
666 } 795 }
667 } else {
668 push @nondirs, $entry;
669 } 796 }
670 } 797 }
671 } 798 };
672 }; 799 };
673 }; 800 };
674 }; 801 };
675 }; 802 };
803
804 $grp
676 }; 805 }
806}
677 807
808=item aio_rmtree $path, $callback->($status)
809
810Delete a directory tree starting (and including) C<$path>, return the
811status of the final C<rmdir> only. This is a composite request that
812uses C<aio_scandir> to recurse into and rmdir directories, and unlink
813everything else.
814
815=cut
816
817sub aio_rmtree;
818sub aio_rmtree($;$) {
819 aio_block {
820 my ($path, $cb) = @_;
821
822 my $pri = aioreq_pri;
823 my $grp = aio_group $cb;
824
825 aioreq_pri $pri;
826 add $grp aio_scandir $path, 0, sub {
827 my ($dirs, $nondirs) = @_;
828
829 my $dirgrp = aio_group sub {
830 add $grp aio_rmdir $path, sub {
831 $grp->result ($_[0]);
832 };
833 };
834
835 (aioreq_pri $pri), add $dirgrp aio_rmtree "$path/$_" for @$dirs;
836 (aioreq_pri $pri), add $dirgrp aio_unlink "$path/$_" for @$nondirs;
837
838 add $grp $dirgrp;
839 };
840
678 $grp 841 $grp
842 }
679} 843}
680 844
681=item aio_fsync $fh, $callback->($status) 845=item aio_fsync $fh, $callback->($status)
682 846
683Asynchronously call fsync on the given filehandle and call the callback 847Asynchronously call fsync on the given filehandle and call the callback
… …
937that are being processed by C<IO::AIO::poll_cb> in one call, respectively 1101that are being processed by C<IO::AIO::poll_cb> in one call, respectively
938the maximum amount of time (default C<0>, meaning infinity) spent in 1102the maximum amount of time (default C<0>, meaning infinity) spent in
939C<IO::AIO::poll_cb> to process requests (more correctly the mininum amount 1103C<IO::AIO::poll_cb> to process requests (more correctly the mininum amount
940of time C<poll_cb> is allowed to use). 1104of time C<poll_cb> is allowed to use).
941 1105
1106Setting C<max_poll_time> to a non-zero value creates an overhead of one
1107syscall per request processed, which is not normally a problem unless your
1108callbacks are really really fast or your OS is really really slow (I am
1109not mentioning Solaris here). Using C<max_poll_reqs> incurs no overhead.
1110
942Setting these is useful if you want to ensure some level of 1111Setting these is useful if you want to ensure some level of
943interactiveness when perl is not fast enough to process all requests in 1112interactiveness when perl is not fast enough to process all requests in
944time. 1113time.
945 1114
946For interactive programs, values such as C<0.01> to C<0.1> should be fine. 1115For interactive programs, values such as C<0.01> to C<0.1> should be fine.
947 1116
948Example: Install an Event watcher that automatically calls 1117Example: Install an Event watcher that automatically calls
949IO::AIO::poll_some with low priority, to ensure that other parts of the 1118IO::AIO::poll_cb with low priority, to ensure that other parts of the
950program get the CPU sometimes even under high AIO load. 1119program get the CPU sometimes even under high AIO load.
951 1120
952 # try not to spend much more than 0.1s in poll_cb 1121 # try not to spend much more than 0.1s in poll_cb
953 IO::AIO::max_poll_time 0.1; 1122 IO::AIO::max_poll_time 0.1;
954 1123
… …
957 poll => 'r', nice => 1, 1126 poll => 'r', nice => 1,
958 cb => &IO::AIO::poll_cb); 1127 cb => &IO::AIO::poll_cb);
959 1128
960=item IO::AIO::poll_wait 1129=item IO::AIO::poll_wait
961 1130
1131If there are any outstanding requests and none of them in the result
962Wait till the result filehandle becomes ready for reading (simply does a 1132phase, wait till the result filehandle becomes ready for reading (simply
963C<select> on the filehandle. This is useful if you want to synchronously 1133does a C<select> on the filehandle. This is useful if you want to
964wait for some requests to finish). 1134synchronously wait for some requests to finish).
965 1135
966See C<nreqs> for an example. 1136See C<nreqs> for an example.
967 1137
968=item IO::AIO::poll 1138=item IO::AIO::poll
969 1139
970Waits until some requests have been handled. 1140Waits until some requests have been handled.
971 1141
1142Returns the number of requests processed, but is otherwise strictly
972Strictly equivalent to: 1143equivalent to:
973 1144
974 IO::AIO::poll_wait, IO::AIO::poll_cb 1145 IO::AIO::poll_wait, IO::AIO::poll_cb
975 if IO::AIO::nreqs;
976 1146
977=item IO::AIO::flush 1147=item IO::AIO::flush
978 1148
979Wait till all outstanding AIO requests have been handled. 1149Wait till all outstanding AIO requests have been handled.
980 1150
981Strictly equivalent to: 1151Strictly equivalent to:
982 1152
983 IO::AIO::poll_wait, IO::AIO::poll_cb 1153 IO::AIO::poll_wait, IO::AIO::poll_cb
984 while IO::AIO::nreqs; 1154 while IO::AIO::nreqs;
985 1155
1156=back
1157
986=head3 CONTROLLING THE NUMBER OF THREADS 1158=head3 CONTROLLING THE NUMBER OF THREADS
1159
1160=over
987 1161
988=item IO::AIO::min_parallel $nthreads 1162=item IO::AIO::min_parallel $nthreads
989 1163
990Set the minimum number of AIO threads to C<$nthreads>. The current 1164Set the minimum number of AIO threads to C<$nthreads>. The current
991default is C<8>, which means eight asynchronous operations can execute 1165default is C<8>, which means eight asynchronous operations can execute
… …
1039This is a very bad function to use in interactive programs because it 1213This is a very bad function to use in interactive programs because it
1040blocks, and a bad way to reduce concurrency because it is inexact: Better 1214blocks, and a bad way to reduce concurrency because it is inexact: Better
1041use an C<aio_group> together with a feed callback. 1215use an C<aio_group> together with a feed callback.
1042 1216
1043Sets the maximum number of outstanding requests to C<$nreqs>. If you 1217Sets the maximum number of outstanding requests to C<$nreqs>. If you
1044to queue up more than this number of requests, the next call to the 1218do queue up more than this number of requests, the next call to the
1045C<poll_cb> (and C<poll_some> and other functions calling C<poll_cb>) 1219C<poll_cb> (and C<poll_some> and other functions calling C<poll_cb>)
1046function will block until the limit is no longer exceeded. 1220function will block until the limit is no longer exceeded.
1047 1221
1048The default value is very large, so there is no practical limit on the 1222The default value is very large, so there is no practical limit on the
1049number of outstanding requests. 1223number of outstanding requests.
1050 1224
1051You can still queue as many requests as you want. Therefore, 1225You can still queue as many requests as you want. Therefore,
1052C<max_oustsanding> is mainly useful in simple scripts (with low values) or 1226C<max_oustsanding> is mainly useful in simple scripts (with low values) or
1053as a stop gap to shield against fatal memory overflow (with large values). 1227as a stop gap to shield against fatal memory overflow (with large values).
1054 1228
1229=back
1230
1055=head3 STATISTICAL INFORMATION 1231=head3 STATISTICAL INFORMATION
1232
1233=over
1056 1234
1057=item IO::AIO::nreqs 1235=item IO::AIO::nreqs
1058 1236
1059Returns the number of requests currently in the ready, execute or pending 1237Returns the number of requests currently in the ready, execute or pending
1060states (i.e. for which their callback has not been invoked yet). 1238states (i.e. for which their callback has not been invoked yet).
… …
1076 1254
1077=back 1255=back
1078 1256
1079=cut 1257=cut
1080 1258
1081# support function to convert a fd into a perl filehandle
1082sub _fd2fh {
1083 return undef if $_[0] < 0;
1084
1085 # try to generate nice filehandles
1086 my $sym = "IO::AIO::fd#$_[0]";
1087 local *$sym;
1088
1089 open *$sym, "+<&=$_[0]" # usually works under any unix
1090 or open *$sym, "<&=$_[0]" # cygwin needs this
1091 or open *$sym, ">&=$_[0]" # or this
1092 or return undef;
1093
1094 *$sym
1095}
1096
1097min_parallel 8; 1259min_parallel 8;
1098 1260
1099END { 1261END { flush }
1100 min_parallel 1;
1101 flush;
1102};
1103 1262
11041; 12631;
1105 1264
1106=head2 FORK BEHAVIOUR 1265=head2 FORK BEHAVIOUR
1107 1266
… …
1127bytes of memory. In addition, stat requests need a stat buffer (possibly 1286bytes of memory. In addition, stat requests need a stat buffer (possibly
1128a few hundred bytes), readdir requires a result buffer and so on. Perl 1287a few hundred bytes), readdir requires a result buffer and so on. Perl
1129scalars and other data passed into aio requests will also be locked and 1288scalars and other data passed into aio requests will also be locked and
1130will consume memory till the request has entered the done state. 1289will consume memory till the request has entered the done state.
1131 1290
1132This is now awfully much, so queuing lots of requests is not usually a 1291This is not awfully much, so queuing lots of requests is not usually a
1133problem. 1292problem.
1134 1293
1135Per-thread usage: 1294Per-thread usage:
1136 1295
1137In the execution phase, some aio requests require more memory for 1296In the execution phase, some aio requests require more memory for

Diff Legend

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