ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/IO-AIO/AIO.pm
Revision: 1.123
Committed: Sat May 10 18:06:41 2008 UTC (16 years ago) by root
Branch: MAIN
Changes since 1.122: +171 -183 lines
Log Message:
*** empty log message ***

File Contents

# Content
1 =head1 NAME
2
3 IO::AIO - Asynchronous Input/Output
4
5 =head1 SYNOPSIS
6
7 use IO::AIO;
8
9 aio_open "/etc/passwd", O_RDONLY, 0, sub {
10 my $fh = shift
11 or die "/etc/passwd: $!";
12 ...
13 };
14
15 aio_unlink "/tmp/file", sub { };
16
17 aio_read $fh, 30000, 1024, $buffer, 0, sub {
18 $_[0] > 0 or die "read error: $!";
19 };
20
21 # version 2+ has request and group objects
22 use IO::AIO 2;
23
24 aioreq_pri 4; # give next request a very high priority
25 my $req = aio_unlink "/tmp/file", sub { };
26 $req->cancel; # cancel request if still in queue
27
28 my $grp = aio_group sub { print "all stats done\n" };
29 add $grp aio_stat "..." for ...;
30
31 # AnyEvent integration (EV, Event, Glib, Tk, urxvt, pureperl...)
32 open my $fh, "<&=" . IO::AIO::poll_fileno or die "$!";
33 my $w = AnyEvent->io (fh => $fh, poll => 'r', cb => sub { IO::AIO::poll_cb });
34
35 # EV integration
36 my $w = EV::io IO::AIO::poll_fileno, EV::READ, \&IO::AIO::poll_cb;
37
38 # Event integration
39 Event->io (fd => IO::AIO::poll_fileno,
40 poll => 'r',
41 cb => \&IO::AIO::poll_cb);
42
43 # Glib/Gtk2 integration
44 add_watch Glib::IO IO::AIO::poll_fileno,
45 in => sub { IO::AIO::poll_cb; 1 };
46
47 # Tk integration
48 Tk::Event::IO->fileevent (IO::AIO::poll_fileno, "",
49 readable => \&IO::AIO::poll_cb);
50
51 # Danga::Socket integration
52 Danga::Socket->AddOtherFds (IO::AIO::poll_fileno =>
53 \&IO::AIO::poll_cb);
54
55 =head1 DESCRIPTION
56
57 This module implements asynchronous I/O using whatever means your
58 operating system supports.
59
60 Asynchronous means that operations that can normally block your program
61 (e.g. reading from disk) will be done asynchronously: the operation
62 will still block, but you can do something else in the meantime. This
63 is extremely useful for programs that need to stay interactive even
64 when doing heavy I/O (GUI programs, high performance network servers
65 etc.), but can also be used to easily do operations in parallel that are
66 normally done sequentially, e.g. stat'ing many files, which is much faster
67 on a RAID volume or over NFS when you do a number of stat operations
68 concurrently.
69
70 While most of this works on all types of file descriptors (for
71 example sockets), using these functions on file descriptors that
72 support nonblocking operation (again, sockets, pipes etc.) is very
73 inefficient. Use an event loop for that (such as the L<Event|Event>
74 module): IO::AIO will naturally fit into such an event loop itself.
75
76 In this version, a number of threads are started that execute your
77 requests and signal their completion. You don't need thread support
78 in perl, and the threads created by this module will not be visible
79 to perl. In the future, this module might make use of the native aio
80 functions available on many operating systems. However, they are often
81 not well-supported or restricted (GNU/Linux doesn't allow them on normal
82 files currently, for example), and they would only support aio_read and
83 aio_write, so the remaining functionality would have to be implemented
84 using threads anyway.
85
86 Although the module will work in the presence of other (Perl-) threads,
87 it is currently not reentrant in any way, so use appropriate locking
88 yourself, always call C<poll_cb> from within the same thread, or never
89 call C<poll_cb> (or other C<aio_> functions) recursively.
90
91 =head2 EXAMPLE
92
93 This is a simple example that uses the Event module and loads
94 F</etc/passwd> asynchronously:
95
96 use Fcntl;
97 use Event;
98 use IO::AIO;
99
100 # register the IO::AIO callback with Event
101 Event->io (fd => IO::AIO::poll_fileno,
102 poll => 'r',
103 cb => \&IO::AIO::poll_cb);
104
105 # queue the request to open /etc/passwd
106 aio_open "/etc/passwd", O_RDONLY, 0, sub {
107 my $fh = shift
108 or die "error while opening: $!";
109
110 # stat'ing filehandles is generally non-blocking
111 my $size = -s $fh;
112
113 # queue a request to read the file
114 my $contents;
115 aio_read $fh, 0, $size, $contents, 0, sub {
116 $_[0] == $size
117 or die "short read: $!";
118
119 close $fh;
120
121 # file contents now in $contents
122 print $contents;
123
124 # exit event loop and program
125 Event::unloop;
126 };
127 };
128
129 # possibly queue up other requests, or open GUI windows,
130 # check for sockets etc. etc.
131
132 # process events as long as there are some:
133 Event::loop;
134
135 =head1 REQUEST ANATOMY AND LIFETIME
136
137 Every C<aio_*> function creates a request. which is a C data structure not
138 directly visible to Perl.
139
140 If called in non-void context, every request function returns a Perl
141 object representing the request. In void context, nothing is returned,
142 which saves a bit of memory.
143
144 The perl object is a fairly standard ref-to-hash object. The hash contents
145 are not used by IO::AIO so you are free to store anything you like in it.
146
147 During their existance, aio requests travel through the following states,
148 in order:
149
150 =over 4
151
152 =item ready
153
154 Immediately after a request is created it is put into the ready state,
155 waiting for a thread to execute it.
156
157 =item execute
158
159 A thread has accepted the request for processing and is currently
160 executing it (e.g. blocking in read).
161
162 =item pending
163
164 The request has been executed and is waiting for result processing.
165
166 While request submission and execution is fully asynchronous, result
167 processing is not and relies on the perl interpreter calling C<poll_cb>
168 (or another function with the same effect).
169
170 =item result
171
172 The request results are processed synchronously by C<poll_cb>.
173
174 The C<poll_cb> function will process all outstanding aio requests by
175 calling their callbacks, freeing memory associated with them and managing
176 any groups they are contained in.
177
178 =item done
179
180 Request has reached the end of its lifetime and holds no resources anymore
181 (except possibly for the Perl object, but its connection to the actual
182 aio request is severed and calling its methods will either do nothing or
183 result in a runtime error).
184
185 =back
186
187 =cut
188
189 package IO::AIO;
190
191 use Carp ();
192
193 no warnings;
194 use strict 'vars';
195
196 use base 'Exporter';
197
198 BEGIN {
199 our $VERSION = '2.62';
200
201 our @AIO_REQ = qw(aio_sendfile aio_read aio_write aio_open aio_close
202 aio_stat aio_lstat aio_unlink aio_rmdir aio_readdir
203 aio_scandir aio_symlink aio_readlink aio_sync aio_fsync
204 aio_fdatasync aio_pathsync aio_readahead
205 aio_rename aio_link aio_move aio_copy aio_group
206 aio_nop aio_mknod aio_load aio_rmtree aio_mkdir aio_chown
207 aio_chmod aio_utime aio_truncate);
208
209 our @EXPORT = (@AIO_REQ, qw(aioreq_pri aioreq_nice));
210 our @EXPORT_OK = qw(poll_fileno poll_cb poll_wait flush
211 min_parallel max_parallel max_idle
212 nreqs nready npending nthreads
213 max_poll_time max_poll_reqs);
214
215 @IO::AIO::GRP::ISA = 'IO::AIO::REQ';
216
217 require XSLoader;
218 XSLoader::load ("IO::AIO", $VERSION);
219 }
220
221 =head1 FUNCTIONS
222
223 =head2 AIO REQUEST FUNCTIONS
224
225 All the C<aio_*> calls are more or less thin wrappers around the syscall
226 with the same name (sans C<aio_>). The arguments are similar or identical,
227 and they all accept an additional (and optional) C<$callback> argument
228 which must be a code reference. This code reference will get called with
229 the syscall return code (e.g. most syscalls return C<-1> on error, unlike
230 perl, which usually delivers "false") as it's sole argument when the given
231 syscall has been executed asynchronously.
232
233 All functions expecting a filehandle keep a copy of the filehandle
234 internally until the request has finished.
235
236 All functions return request objects of type L<IO::AIO::REQ> that allow
237 further manipulation of those requests while they are in-flight.
238
239 The pathnames you pass to these routines I<must> be absolute and
240 encoded as octets. The reason for the former is that at the time the
241 request is being executed, the current working directory could have
242 changed. Alternatively, you can make sure that you never change the
243 current working directory anywhere in the program and then use relative
244 paths.
245
246 To encode pathnames as octets, either make sure you either: a) always pass
247 in filenames you got from outside (command line, readdir etc.) without
248 tinkering, b) are ASCII or ISO 8859-1, c) use the Encode module and encode
249 your pathnames to the locale (or other) encoding in effect in the user
250 environment, d) use Glib::filename_from_unicode on unicode filenames or e)
251 use something else to ensure your scalar has the correct contents.
252
253 This works, btw. independent of the internal UTF-8 bit, which IO::AIO
254 handles correctly wether it is set or not.
255
256 =over 4
257
258 =item $prev_pri = aioreq_pri [$pri]
259
260 Returns the priority value that would be used for the next request and, if
261 C<$pri> is given, sets the priority for the next aio request.
262
263 The default priority is C<0>, the minimum and maximum priorities are C<-4>
264 and C<4>, respectively. Requests with higher priority will be serviced
265 first.
266
267 The priority will be reset to C<0> after each call to one of the C<aio_*>
268 functions.
269
270 Example: open a file with low priority, then read something from it with
271 higher priority so the read request is serviced before other low priority
272 open requests (potentially spamming the cache):
273
274 aioreq_pri -3;
275 aio_open ..., sub {
276 return unless $_[0];
277
278 aioreq_pri -2;
279 aio_read $_[0], ..., sub {
280 ...
281 };
282 };
283
284
285 =item aioreq_nice $pri_adjust
286
287 Similar to C<aioreq_pri>, but subtracts the given value from the current
288 priority, so the effect is cumulative.
289
290
291 =item aio_open $pathname, $flags, $mode, $callback->($fh)
292
293 Asynchronously open or create a file and call the callback with a newly
294 created filehandle for the file.
295
296 The pathname passed to C<aio_open> must be absolute. See API NOTES, above,
297 for an explanation.
298
299 The C<$flags> argument is a bitmask. See the C<Fcntl> module for a
300 list. They are the same as used by C<sysopen>.
301
302 Likewise, C<$mode> specifies the mode of the newly created file, if it
303 didn't exist and C<O_CREAT> has been given, just like perl's C<sysopen>,
304 except that it is mandatory (i.e. use C<0> if you don't create new files,
305 and C<0666> or C<0777> if you do). Note that the C<$mode> will be modified
306 by the umask in effect then the request is being executed, so better never
307 change the umask.
308
309 Example:
310
311 aio_open "/etc/passwd", O_RDONLY, 0, sub {
312 if ($_[0]) {
313 print "open successful, fh is $_[0]\n";
314 ...
315 } else {
316 die "open failed: $!\n";
317 }
318 };
319
320
321 =item aio_close $fh, $callback->($status)
322
323 Asynchronously close a file and call the callback with the result
324 code.
325
326 Unfortunately, you can't do this to perl. Perl I<insists> very strongly on
327 closing the file descriptor associated with the filehandle itself.
328
329 Therefore, C<aio_close> will not close the filehandle - instead it will
330 use dup2 to overwrite the file descriptor with the write-end of a pipe
331 (the pipe fd will be created on demand and will be cached).
332
333 Or in other words: the file descriptor will be closed, but it will not be
334 free for reuse until the perl filehandle is closed.
335
336 =cut
337
338 =item aio_read $fh,$offset,$length, $data,$dataoffset, $callback->($retval)
339
340 =item aio_write $fh,$offset,$length, $data,$dataoffset, $callback->($retval)
341
342 Reads or writes C<$length> bytes from the specified C<$fh> and C<$offset>
343 into the scalar given by C<$data> and offset C<$dataoffset> and calls the
344 callback without the actual number of bytes read (or -1 on error, just
345 like the syscall).
346
347 If C<$offset> is undefined, then the current file descriptor offset will
348 be used (and updated), otherwise the file descriptor offset will not be
349 changed by these calls.
350
351 If C<$length> is undefined in C<aio_write>, use the remaining length of C<$data>.
352
353 If C<$dataoffset> is less than zero, it will be counted from the end of
354 C<$data>.
355
356 The C<$data> scalar I<MUST NOT> be modified in any way while the request
357 is outstanding. Modifying it can result in segfaults or World War III (if
358 the necessary/optional hardware is installed).
359
360 Example: Read 15 bytes at offset 7 into scalar C<$buffer>, starting at
361 offset C<0> within the scalar:
362
363 aio_read $fh, 7, 15, $buffer, 0, sub {
364 $_[0] > 0 or die "read error: $!";
365 print "read $_[0] bytes: <$buffer>\n";
366 };
367
368
369 =item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback->($retval)
370
371 Tries to copy C<$length> bytes from C<$in_fh> to C<$out_fh>. It starts
372 reading at byte offset C<$in_offset>, and starts writing at the current
373 file offset of C<$out_fh>. Because of that, it is not safe to issue more
374 than one C<aio_sendfile> per C<$out_fh>, as they will interfere with each
375 other.
376
377 This call tries to make use of a native C<sendfile> syscall to provide
378 zero-copy operation. For this to work, C<$out_fh> should refer to a
379 socket, and C<$in_fh> should refer to mmap'able file.
380
381 If the native sendfile call fails or is not implemented, it will be
382 emulated, so you can call C<aio_sendfile> on any type of filehandle
383 regardless of the limitations of the operating system.
384
385 Please note, however, that C<aio_sendfile> can read more bytes from
386 C<$in_fh> than are written, and there is no way to find out how many
387 bytes have been read from C<aio_sendfile> alone, as C<aio_sendfile> only
388 provides the number of bytes written to C<$out_fh>. Only if the result
389 value equals C<$length> one can assume that C<$length> bytes have been
390 read.
391
392
393 =item aio_readahead $fh,$offset,$length, $callback->($retval)
394
395 C<aio_readahead> populates the page cache with data from a file so that
396 subsequent reads from that file will not block on disk I/O. The C<$offset>
397 argument specifies the starting point from which data is to be read and
398 C<$length> specifies the number of bytes to be read. I/O is performed in
399 whole pages, so that offset is effectively rounded down to a page boundary
400 and bytes are read up to the next page boundary greater than or equal to
401 (off-set+length). C<aio_readahead> does not read beyond the end of the
402 file. The current file offset of the file is left unchanged.
403
404 If that syscall doesn't exist (likely if your OS isn't Linux) it will be
405 emulated by simply reading the data, which would have a similar effect.
406
407
408 =item aio_stat $fh_or_path, $callback->($status)
409
410 =item aio_lstat $fh, $callback->($status)
411
412 Works like perl's C<stat> or C<lstat> in void context. The callback will
413 be called after the stat and the results will be available using C<stat _>
414 or C<-s _> etc...
415
416 The pathname passed to C<aio_stat> must be absolute. See API NOTES, above,
417 for an explanation.
418
419 Currently, the stats are always 64-bit-stats, i.e. instead of returning an
420 error when stat'ing a large file, the results will be silently truncated
421 unless perl itself is compiled with large file support.
422
423 Example: Print the length of F</etc/passwd>:
424
425 aio_stat "/etc/passwd", sub {
426 $_[0] and die "stat failed: $!";
427 print "size is ", -s _, "\n";
428 };
429
430
431 =item aio_utime $fh_or_path, $atime, $mtime, $callback->($status)
432
433 Works like perl's C<utime> function (including the special case of $atime
434 and $mtime being undef). Fractional times are supported if the underlying
435 syscalls support them.
436
437 When called with a pathname, uses utimes(2) if available, otherwise
438 utime(2). If called on a file descriptor, uses futimes(2) if available,
439 otherwise returns ENOSYS, so this is not portable.
440
441 Examples:
442
443 # set atime and mtime to current time (basically touch(1)):
444 aio_utime "path", undef, undef;
445 # set atime to current time and mtime to beginning of the epoch:
446 aio_utime "path", time, undef; # undef==0
447
448
449 =item aio_chown $fh_or_path, $uid, $gid, $callback->($status)
450
451 Works like perl's C<chown> function, except that C<undef> for either $uid
452 or $gid is being interpreted as "do not change" (but -1 can also be used).
453
454 Examples:
455
456 # same as "chown root path" in the shell:
457 aio_chown "path", 0, -1;
458 # same as above:
459 aio_chown "path", 0, undef;
460
461
462 =item aio_truncate $fh_or_path, $offset, $callback->($status)
463
464 Works like truncate(2) or ftruncate(2).
465
466
467 =item aio_chmod $fh_or_path, $mode, $callback->($status)
468
469 Works like perl's C<chmod> function.
470
471
472 =item aio_unlink $pathname, $callback->($status)
473
474 Asynchronously unlink (delete) a file and call the callback with the
475 result code.
476
477
478 =item aio_mknod $path, $mode, $dev, $callback->($status)
479
480 [EXPERIMENTAL]
481
482 Asynchronously create a device node (or fifo). See mknod(2).
483
484 The only (POSIX-) portable way of calling this function is:
485
486 aio_mknod $path, IO::AIO::S_IFIFO | $mode, 0, sub { ...
487
488
489 =item aio_link $srcpath, $dstpath, $callback->($status)
490
491 Asynchronously create a new link to the existing object at C<$srcpath> at
492 the path C<$dstpath> and call the callback with the result code.
493
494
495 =item aio_symlink $srcpath, $dstpath, $callback->($status)
496
497 Asynchronously create a new symbolic link to the existing object at C<$srcpath> at
498 the path C<$dstpath> and call the callback with the result code.
499
500
501 =item aio_readlink $path, $callback->($link)
502
503 Asynchronously read the symlink specified by C<$path> and pass it to
504 the callback. If an error occurs, nothing or undef gets passed to the
505 callback.
506
507
508 =item aio_rename $srcpath, $dstpath, $callback->($status)
509
510 Asynchronously rename the object at C<$srcpath> to C<$dstpath>, just as
511 rename(2) and call the callback with the result code.
512
513
514 =item aio_mkdir $pathname, $mode, $callback->($status)
515
516 Asynchronously mkdir (create) a directory and call the callback with
517 the result code. C<$mode> will be modified by the umask at the time the
518 request is executed, so do not change your umask.
519
520
521 =item aio_rmdir $pathname, $callback->($status)
522
523 Asynchronously rmdir (delete) a directory and call the callback with the
524 result code.
525
526
527 =item aio_readdir $pathname, $callback->($entries)
528
529 Unlike the POSIX call of the same name, C<aio_readdir> reads an entire
530 directory (i.e. opendir + readdir + closedir). The entries will not be
531 sorted, and will B<NOT> include the C<.> and C<..> entries.
532
533 The callback a single argument which is either C<undef> or an array-ref
534 with the filenames.
535
536
537 =item aio_load $path, $data, $callback->($status)
538
539 This is a composite request that tries to fully load the given file into
540 memory. Status is the same as with aio_read.
541
542 =cut
543
544 sub aio_load($$;$) {
545 my ($path, undef, $cb) = @_;
546 my $data = \$_[1];
547
548 my $pri = aioreq_pri;
549 my $grp = aio_group $cb;
550
551 aioreq_pri $pri;
552 add $grp aio_open $path, O_RDONLY, 0, sub {
553 my $fh = shift
554 or return $grp->result (-1);
555
556 aioreq_pri $pri;
557 add $grp aio_read $fh, 0, (-s $fh), $$data, 0, sub {
558 $grp->result ($_[0]);
559 };
560 };
561
562 $grp
563 }
564
565 =item aio_copy $srcpath, $dstpath, $callback->($status)
566
567 Try to copy the I<file> (directories not supported as either source or
568 destination) from C<$srcpath> to C<$dstpath> and call the callback with
569 the C<0> (error) or C<-1> ok.
570
571 This is a composite request that it creates the destination file with
572 mode 0200 and copies the contents of the source file into it using
573 C<aio_sendfile>, followed by restoring atime, mtime, access mode and
574 uid/gid, in that order.
575
576 If an error occurs, the partial destination file will be unlinked, if
577 possible, except when setting atime, mtime, access mode and uid/gid, where
578 errors are being ignored.
579
580 =cut
581
582 sub aio_copy($$;$) {
583 my ($src, $dst, $cb) = @_;
584
585 my $pri = aioreq_pri;
586 my $grp = aio_group $cb;
587
588 aioreq_pri $pri;
589 add $grp aio_open $src, O_RDONLY, 0, sub {
590 if (my $src_fh = $_[0]) {
591 my @stat = stat $src_fh;
592
593 aioreq_pri $pri;
594 add $grp aio_open $dst, O_CREAT | O_WRONLY | O_TRUNC, 0200, sub {
595 if (my $dst_fh = $_[0]) {
596 aioreq_pri $pri;
597 add $grp aio_sendfile $dst_fh, $src_fh, 0, $stat[7], sub {
598 if ($_[0] == $stat[7]) {
599 $grp->result (0);
600 close $src_fh;
601
602 # those should not normally block. should. should.
603 utime $stat[8], $stat[9], $dst;
604 chmod $stat[2] & 07777, $dst_fh;
605 chown $stat[4], $stat[5], $dst_fh;
606
607 aioreq_pri $pri;
608 add $grp aio_close $dst_fh;
609 } else {
610 $grp->result (-1);
611 close $src_fh;
612 close $dst_fh;
613
614 aioreq $pri;
615 add $grp aio_unlink $dst;
616 }
617 };
618 } else {
619 $grp->result (-1);
620 }
621 },
622
623 } else {
624 $grp->result (-1);
625 }
626 };
627
628 $grp
629 }
630
631 =item aio_move $srcpath, $dstpath, $callback->($status)
632
633 Try to move the I<file> (directories not supported as either source or
634 destination) from C<$srcpath> to C<$dstpath> and call the callback with
635 the C<0> (error) or C<-1> ok.
636
637 This is a composite request that tries to rename(2) the file first. If
638 rename files with C<EXDEV>, it copies the file with C<aio_copy> and, if
639 that is successful, unlinking the C<$srcpath>.
640
641 =cut
642
643 sub aio_move($$;$) {
644 my ($src, $dst, $cb) = @_;
645
646 my $pri = aioreq_pri;
647 my $grp = aio_group $cb;
648
649 aioreq_pri $pri;
650 add $grp aio_rename $src, $dst, sub {
651 if ($_[0] && $! == EXDEV) {
652 aioreq_pri $pri;
653 add $grp aio_copy $src, $dst, sub {
654 $grp->result ($_[0]);
655
656 if (!$_[0]) {
657 aioreq_pri $pri;
658 add $grp aio_unlink $src;
659 }
660 };
661 } else {
662 $grp->result ($_[0]);
663 }
664 };
665
666 $grp
667 }
668
669 =item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)
670
671 Scans a directory (similar to C<aio_readdir>) but additionally tries to
672 efficiently separate the entries of directory C<$path> into two sets of
673 names, directories you can recurse into (directories), and ones you cannot
674 recurse into (everything else, including symlinks to directories).
675
676 C<aio_scandir> is a composite request that creates of many sub requests_
677 C<$maxreq> specifies the maximum number of outstanding aio requests that
678 this function generates. If it is C<< <= 0 >>, then a suitable default
679 will be chosen (currently 4).
680
681 On error, the callback is called without arguments, otherwise it receives
682 two array-refs with path-relative entry names.
683
684 Example:
685
686 aio_scandir $dir, 0, sub {
687 my ($dirs, $nondirs) = @_;
688 print "real directories: @$dirs\n";
689 print "everything else: @$nondirs\n";
690 };
691
692 Implementation notes.
693
694 The C<aio_readdir> cannot be avoided, but C<stat()>'ing every entry can.
695
696 After reading the directory, the modification time, size etc. of the
697 directory before and after the readdir is checked, and if they match (and
698 isn't the current time), the link count will be used to decide how many
699 entries are directories (if >= 2). Otherwise, no knowledge of the number
700 of subdirectories will be assumed.
701
702 Then entries will be sorted into likely directories (everything without
703 a non-initial dot currently) and likely non-directories (everything
704 else). Then every entry plus an appended C</.> will be C<stat>'ed,
705 likely directories first. If that succeeds, it assumes that the entry
706 is a directory or a symlink to directory (which will be checked
707 seperately). This is often faster than stat'ing the entry itself because
708 filesystems might detect the type of the entry without reading the inode
709 data (e.g. ext2fs filetype feature).
710
711 If the known number of directories (link count - 2) has been reached, the
712 rest of the entries is assumed to be non-directories.
713
714 This only works with certainty on POSIX (= UNIX) filesystems, which
715 fortunately are the vast majority of filesystems around.
716
717 It will also likely work on non-POSIX filesystems with reduced efficiency
718 as those tend to return 0 or 1 as link counts, which disables the
719 directory counting heuristic.
720
721 =cut
722
723 sub aio_scandir($$;$) {
724 my ($path, $maxreq, $cb) = @_;
725
726 my $pri = aioreq_pri;
727
728 my $grp = aio_group $cb;
729
730 $maxreq = 4 if $maxreq <= 0;
731
732 # stat once
733 aioreq_pri $pri;
734 add $grp aio_stat $path, sub {
735 return $grp->result () if $_[0];
736 my $now = time;
737 my $hash1 = join ":", (stat _)[0,1,3,7,9];
738
739 # read the directory entries
740 aioreq_pri $pri;
741 add $grp aio_readdir $path, sub {
742 my $entries = shift
743 or return $grp->result ();
744
745 # stat the dir another time
746 aioreq_pri $pri;
747 add $grp aio_stat $path, sub {
748 my $hash2 = join ":", (stat _)[0,1,3,7,9];
749
750 my $ndirs;
751
752 # take the slow route if anything looks fishy
753 if ($hash1 ne $hash2 or (stat _)[9] == $now) {
754 $ndirs = -1;
755 } else {
756 # if nlink == 2, we are finished
757 # on non-posix-fs's, we rely on nlink < 2
758 $ndirs = (stat _)[3] - 2
759 or return $grp->result ([], $entries);
760 }
761
762 # sort into likely dirs and likely nondirs
763 # dirs == files without ".", short entries first
764 $entries = [map $_->[0],
765 sort { $b->[1] cmp $a->[1] }
766 map [$_, sprintf "%s%04d", (/.\./ ? "1" : "0"), length],
767 @$entries];
768
769 my (@dirs, @nondirs);
770
771 my $statgrp = add $grp aio_group sub {
772 $grp->result (\@dirs, \@nondirs);
773 };
774
775 limit $statgrp $maxreq;
776 feed $statgrp sub {
777 return unless @$entries;
778 my $entry = pop @$entries;
779
780 aioreq_pri $pri;
781 add $statgrp aio_stat "$path/$entry/.", sub {
782 if ($_[0] < 0) {
783 push @nondirs, $entry;
784 } else {
785 # need to check for real directory
786 aioreq_pri $pri;
787 add $statgrp aio_lstat "$path/$entry", sub {
788 if (-d _) {
789 push @dirs, $entry;
790
791 unless (--$ndirs) {
792 push @nondirs, @$entries;
793 feed $statgrp;
794 }
795 } else {
796 push @nondirs, $entry;
797 }
798 }
799 }
800 };
801 };
802 };
803 };
804 };
805
806 $grp
807 }
808
809 =item aio_rmtree $path, $callback->($status)
810
811 Delete a directory tree starting (and including) C<$path>, return the
812 status of the final C<rmdir> only. This is a composite request that
813 uses C<aio_scandir> to recurse into and rmdir directories, and unlink
814 everything else.
815
816 =cut
817
818 sub aio_rmtree;
819 sub aio_rmtree($;$) {
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
841 $grp
842 }
843
844 =item aio_sync $callback->($status)
845
846 Asynchronously call sync and call the callback when finished.
847
848 =item aio_fsync $fh, $callback->($status)
849
850 Asynchronously call fsync on the given filehandle and call the callback
851 with the fsync result code.
852
853 =item aio_fdatasync $fh, $callback->($status)
854
855 Asynchronously call fdatasync on the given filehandle and call the
856 callback with the fdatasync result code.
857
858 If this call isn't available because your OS lacks it or it couldn't be
859 detected, it will be emulated by calling C<fsync> instead.
860
861 =item aio_pathsync $path, $callback->($status)
862
863 This request tries to open, fsync and close the given path. This is a
864 composite request intended tosync directories after directory operations
865 (E.g. rename). This might not work on all operating systems or have any
866 specific effect, but usually it makes sure that directory changes get
867 written to disc. It works for anything that can be opened for read-only,
868 not just directories.
869
870 Passes C<0> when everything went ok, and C<-1> on error.
871
872 =cut
873
874 sub aio_pathsync($;$) {
875 my ($path, $cb) = @_;
876
877 my $pri = aioreq_pri;
878 my $grp = aio_group $cb;
879
880 aioreq_pri $pri;
881 add $grp aio_open $path, O_RDONLY, 0, sub {
882 my ($fh) = @_;
883 if ($fh) {
884 aioreq_pri $pri;
885 add $grp aio_fsync $fh, sub {
886 $grp->result ($_[0]);
887
888 aioreq_pri $pri;
889 add $grp aio_close $fh;
890 };
891 } else {
892 $grp->result (-1);
893 }
894 };
895
896 $grp
897 }
898
899 =item aio_group $callback->(...)
900
901 This is a very special aio request: Instead of doing something, it is a
902 container for other aio requests, which is useful if you want to bundle
903 many requests into a single, composite, request with a definite callback
904 and the ability to cancel the whole request with its subrequests.
905
906 Returns an object of class L<IO::AIO::GRP>. See its documentation below
907 for more info.
908
909 Example:
910
911 my $grp = aio_group sub {
912 print "all stats done\n";
913 };
914
915 add $grp
916 (aio_stat ...),
917 (aio_stat ...),
918 ...;
919
920 =item aio_nop $callback->()
921
922 This is a special request - it does nothing in itself and is only used for
923 side effects, such as when you want to add a dummy request to a group so
924 that finishing the requests in the group depends on executing the given
925 code.
926
927 While this request does nothing, it still goes through the execution
928 phase and still requires a worker thread. Thus, the callback will not
929 be executed immediately but only after other requests in the queue have
930 entered their execution phase. This can be used to measure request
931 latency.
932
933 =item IO::AIO::aio_busy $fractional_seconds, $callback->() *NOT EXPORTED*
934
935 Mainly used for debugging and benchmarking, this aio request puts one of
936 the request workers to sleep for the given time.
937
938 While it is theoretically handy to have simple I/O scheduling requests
939 like sleep and file handle readable/writable, the overhead this creates is
940 immense (it blocks a thread for a long time) so do not use this function
941 except to put your application under artificial I/O pressure.
942
943 =back
944
945 =head2 IO::AIO::REQ CLASS
946
947 All non-aggregate C<aio_*> functions return an object of this class when
948 called in non-void context.
949
950 =over 4
951
952 =item cancel $req
953
954 Cancels the request, if possible. Has the effect of skipping execution
955 when entering the B<execute> state and skipping calling the callback when
956 entering the the B<result> state, but will leave the request otherwise
957 untouched. That means that requests that currently execute will not be
958 stopped and resources held by the request will not be freed prematurely.
959
960 =item cb $req $callback->(...)
961
962 Replace (or simply set) the callback registered to the request.
963
964 =back
965
966 =head2 IO::AIO::GRP CLASS
967
968 This class is a subclass of L<IO::AIO::REQ>, so all its methods apply to
969 objects of this class, too.
970
971 A IO::AIO::GRP object is a special request that can contain multiple other
972 aio requests.
973
974 You create one by calling the C<aio_group> constructing function with a
975 callback that will be called when all contained requests have entered the
976 C<done> state:
977
978 my $grp = aio_group sub {
979 print "all requests are done\n";
980 };
981
982 You add requests by calling the C<add> method with one or more
983 C<IO::AIO::REQ> objects:
984
985 $grp->add (aio_unlink "...");
986
987 add $grp aio_stat "...", sub {
988 $_[0] or return $grp->result ("error");
989
990 # add another request dynamically, if first succeeded
991 add $grp aio_open "...", sub {
992 $grp->result ("ok");
993 };
994 };
995
996 This makes it very easy to create composite requests (see the source of
997 C<aio_move> for an application) that work and feel like simple requests.
998
999 =over 4
1000
1001 =item * The IO::AIO::GRP objects will be cleaned up during calls to
1002 C<IO::AIO::poll_cb>, just like any other request.
1003
1004 =item * They can be canceled like any other request. Canceling will cancel not
1005 only the request itself, but also all requests it contains.
1006
1007 =item * They can also can also be added to other IO::AIO::GRP objects.
1008
1009 =item * You must not add requests to a group from within the group callback (or
1010 any later time).
1011
1012 =back
1013
1014 Their lifetime, simplified, looks like this: when they are empty, they
1015 will finish very quickly. If they contain only requests that are in the
1016 C<done> state, they will also finish. Otherwise they will continue to
1017 exist.
1018
1019 That means after creating a group you have some time to add requests. And
1020 in the callbacks of those requests, you can add further requests to the
1021 group. And only when all those requests have finished will the the group
1022 itself finish.
1023
1024 =over 4
1025
1026 =item add $grp ...
1027
1028 =item $grp->add (...)
1029
1030 Add one or more requests to the group. Any type of L<IO::AIO::REQ> can
1031 be added, including other groups, as long as you do not create circular
1032 dependencies.
1033
1034 Returns all its arguments.
1035
1036 =item $grp->cancel_subs
1037
1038 Cancel all subrequests and clears any feeder, but not the group request
1039 itself. Useful when you queued a lot of events but got a result early.
1040
1041 =item $grp->result (...)
1042
1043 Set the result value(s) that will be passed to the group callback when all
1044 subrequests have finished and set the groups errno to the current value
1045 of errno (just like calling C<errno> without an error number). By default,
1046 no argument will be passed and errno is zero.
1047
1048 =item $grp->errno ([$errno])
1049
1050 Sets the group errno value to C<$errno>, or the current value of errno
1051 when the argument is missing.
1052
1053 Every aio request has an associated errno value that is restored when
1054 the callback is invoked. This method lets you change this value from its
1055 default (0).
1056
1057 Calling C<result> will also set errno, so make sure you either set C<$!>
1058 before the call to C<result>, or call c<errno> after it.
1059
1060 =item feed $grp $callback->($grp)
1061
1062 Sets a feeder/generator on this group: every group can have an attached
1063 generator that generates requests if idle. The idea behind this is that,
1064 although you could just queue as many requests as you want in a group,
1065 this might starve other requests for a potentially long time. For
1066 example, C<aio_scandir> might generate hundreds of thousands C<aio_stat>
1067 requests, delaying any later requests for a long time.
1068
1069 To avoid this, and allow incremental generation of requests, you can
1070 instead a group and set a feeder on it that generates those requests. The
1071 feed callback will be called whenever there are few enough (see C<limit>,
1072 below) requests active in the group itself and is expected to queue more
1073 requests.
1074
1075 The feed callback can queue as many requests as it likes (i.e. C<add> does
1076 not impose any limits).
1077
1078 If the feed does not queue more requests when called, it will be
1079 automatically removed from the group.
1080
1081 If the feed limit is C<0>, it will be set to C<2> automatically.
1082
1083 Example:
1084
1085 # stat all files in @files, but only ever use four aio requests concurrently:
1086
1087 my $grp = aio_group sub { print "finished\n" };
1088 limit $grp 4;
1089 feed $grp sub {
1090 my $file = pop @files
1091 or return;
1092
1093 add $grp aio_stat $file, sub { ... };
1094 };
1095
1096 =item limit $grp $num
1097
1098 Sets the feeder limit for the group: The feeder will be called whenever
1099 the group contains less than this many requests.
1100
1101 Setting the limit to C<0> will pause the feeding process.
1102
1103 =back
1104
1105 =head2 SUPPORT FUNCTIONS
1106
1107 =head3 EVENT PROCESSING AND EVENT LOOP INTEGRATION
1108
1109 =over 4
1110
1111 =item $fileno = IO::AIO::poll_fileno
1112
1113 Return the I<request result pipe file descriptor>. This filehandle must be
1114 polled for reading by some mechanism outside this module (e.g. Event or
1115 select, see below or the SYNOPSIS). If the pipe becomes readable you have
1116 to call C<poll_cb> to check the results.
1117
1118 See C<poll_cb> for an example.
1119
1120 =item IO::AIO::poll_cb
1121
1122 Process some outstanding events on the result pipe. You have to call this
1123 regularly. Returns the number of events processed. Returns immediately
1124 when no events are outstanding. The amount of events processed depends on
1125 the settings of C<IO::AIO::max_poll_req> and C<IO::AIO::max_poll_time>.
1126
1127 If not all requests were processed for whatever reason, the filehandle
1128 will still be ready when C<poll_cb> returns.
1129
1130 Example: Install an Event watcher that automatically calls
1131 IO::AIO::poll_cb with high priority:
1132
1133 Event->io (fd => IO::AIO::poll_fileno,
1134 poll => 'r', async => 1,
1135 cb => \&IO::AIO::poll_cb);
1136
1137 =item IO::AIO::max_poll_reqs $nreqs
1138
1139 =item IO::AIO::max_poll_time $seconds
1140
1141 These set the maximum number of requests (default C<0>, meaning infinity)
1142 that are being processed by C<IO::AIO::poll_cb> in one call, respectively
1143 the maximum amount of time (default C<0>, meaning infinity) spent in
1144 C<IO::AIO::poll_cb> to process requests (more correctly the mininum amount
1145 of time C<poll_cb> is allowed to use).
1146
1147 Setting C<max_poll_time> to a non-zero value creates an overhead of one
1148 syscall per request processed, which is not normally a problem unless your
1149 callbacks are really really fast or your OS is really really slow (I am
1150 not mentioning Solaris here). Using C<max_poll_reqs> incurs no overhead.
1151
1152 Setting these is useful if you want to ensure some level of
1153 interactiveness when perl is not fast enough to process all requests in
1154 time.
1155
1156 For interactive programs, values such as C<0.01> to C<0.1> should be fine.
1157
1158 Example: Install an Event watcher that automatically calls
1159 IO::AIO::poll_cb with low priority, to ensure that other parts of the
1160 program get the CPU sometimes even under high AIO load.
1161
1162 # try not to spend much more than 0.1s in poll_cb
1163 IO::AIO::max_poll_time 0.1;
1164
1165 # use a low priority so other tasks have priority
1166 Event->io (fd => IO::AIO::poll_fileno,
1167 poll => 'r', nice => 1,
1168 cb => &IO::AIO::poll_cb);
1169
1170 =item IO::AIO::poll_wait
1171
1172 If there are any outstanding requests and none of them in the result
1173 phase, wait till the result filehandle becomes ready for reading (simply
1174 does a C<select> on the filehandle. This is useful if you want to
1175 synchronously wait for some requests to finish).
1176
1177 See C<nreqs> for an example.
1178
1179 =item IO::AIO::poll
1180
1181 Waits until some requests have been handled.
1182
1183 Returns the number of requests processed, but is otherwise strictly
1184 equivalent to:
1185
1186 IO::AIO::poll_wait, IO::AIO::poll_cb
1187
1188 =item IO::AIO::flush
1189
1190 Wait till all outstanding AIO requests have been handled.
1191
1192 Strictly equivalent to:
1193
1194 IO::AIO::poll_wait, IO::AIO::poll_cb
1195 while IO::AIO::nreqs;
1196
1197 =back
1198
1199 =head3 CONTROLLING THE NUMBER OF THREADS
1200
1201 =over
1202
1203 =item IO::AIO::min_parallel $nthreads
1204
1205 Set the minimum number of AIO threads to C<$nthreads>. The current
1206 default is C<8>, which means eight asynchronous operations can execute
1207 concurrently at any one time (the number of outstanding requests,
1208 however, is unlimited).
1209
1210 IO::AIO starts threads only on demand, when an AIO request is queued and
1211 no free thread exists. Please note that queueing up a hundred requests can
1212 create demand for a hundred threads, even if it turns out that everything
1213 is in the cache and could have been processed faster by a single thread.
1214
1215 It is recommended to keep the number of threads relatively low, as some
1216 Linux kernel versions will scale negatively with the number of threads
1217 (higher parallelity => MUCH higher latency). With current Linux 2.6
1218 versions, 4-32 threads should be fine.
1219
1220 Under most circumstances you don't need to call this function, as the
1221 module selects a default that is suitable for low to moderate load.
1222
1223 =item IO::AIO::max_parallel $nthreads
1224
1225 Sets the maximum number of AIO threads to C<$nthreads>. If more than the
1226 specified number of threads are currently running, this function kills
1227 them. This function blocks until the limit is reached.
1228
1229 While C<$nthreads> are zero, aio requests get queued but not executed
1230 until the number of threads has been increased again.
1231
1232 This module automatically runs C<max_parallel 0> at program end, to ensure
1233 that all threads are killed and that there are no outstanding requests.
1234
1235 Under normal circumstances you don't need to call this function.
1236
1237 =item IO::AIO::max_idle $nthreads
1238
1239 Limit the number of threads (default: 4) that are allowed to idle (i.e.,
1240 threads that did not get a request to process within 10 seconds). That
1241 means if a thread becomes idle while C<$nthreads> other threads are also
1242 idle, it will free its resources and exit.
1243
1244 This is useful when you allow a large number of threads (e.g. 100 or 1000)
1245 to allow for extremely high load situations, but want to free resources
1246 under normal circumstances (1000 threads can easily consume 30MB of RAM).
1247
1248 The default is probably ok in most situations, especially if thread
1249 creation is fast. If thread creation is very slow on your system you might
1250 want to use larger values.
1251
1252 =item IO::AIO::max_outstanding $maxreqs
1253
1254 This is a very bad function to use in interactive programs because it
1255 blocks, and a bad way to reduce concurrency because it is inexact: Better
1256 use an C<aio_group> together with a feed callback.
1257
1258 Sets the maximum number of outstanding requests to C<$nreqs>. If you
1259 do queue up more than this number of requests, the next call to the
1260 C<poll_cb> (and C<poll_some> and other functions calling C<poll_cb>)
1261 function will block until the limit is no longer exceeded.
1262
1263 The default value is very large, so there is no practical limit on the
1264 number of outstanding requests.
1265
1266 You can still queue as many requests as you want. Therefore,
1267 C<max_outstanding> is mainly useful in simple scripts (with low values) or
1268 as a stop gap to shield against fatal memory overflow (with large values).
1269
1270 =back
1271
1272 =head3 STATISTICAL INFORMATION
1273
1274 =over
1275
1276 =item IO::AIO::nreqs
1277
1278 Returns the number of requests currently in the ready, execute or pending
1279 states (i.e. for which their callback has not been invoked yet).
1280
1281 Example: wait till there are no outstanding requests anymore:
1282
1283 IO::AIO::poll_wait, IO::AIO::poll_cb
1284 while IO::AIO::nreqs;
1285
1286 =item IO::AIO::nready
1287
1288 Returns the number of requests currently in the ready state (not yet
1289 executed).
1290
1291 =item IO::AIO::npending
1292
1293 Returns the number of requests currently in the pending state (executed,
1294 but not yet processed by poll_cb).
1295
1296 =back
1297
1298 =cut
1299
1300 min_parallel 8;
1301
1302 END { flush }
1303
1304 1;
1305
1306 =head2 FORK BEHAVIOUR
1307
1308 This module should do "the right thing" when the process using it forks:
1309
1310 Before the fork, IO::AIO enters a quiescent state where no requests
1311 can be added in other threads and no results will be processed. After
1312 the fork the parent simply leaves the quiescent state and continues
1313 request/result processing, while the child frees the request/result queue
1314 (so that the requests started before the fork will only be handled in the
1315 parent). Threads will be started on demand until the limit set in the
1316 parent process has been reached again.
1317
1318 In short: the parent will, after a short pause, continue as if fork had
1319 not been called, while the child will act as if IO::AIO has not been used
1320 yet.
1321
1322 =head2 MEMORY USAGE
1323
1324 Per-request usage:
1325
1326 Each aio request uses - depending on your architecture - around 100-200
1327 bytes of memory. In addition, stat requests need a stat buffer (possibly
1328 a few hundred bytes), readdir requires a result buffer and so on. Perl
1329 scalars and other data passed into aio requests will also be locked and
1330 will consume memory till the request has entered the done state.
1331
1332 This is not awfully much, so queuing lots of requests is not usually a
1333 problem.
1334
1335 Per-thread usage:
1336
1337 In the execution phase, some aio requests require more memory for
1338 temporary buffers, and each thread requires a stack and other data
1339 structures (usually around 16k-128k, depending on the OS).
1340
1341 =head1 KNOWN BUGS
1342
1343 Known bugs will be fixed in the next release.
1344
1345 =head1 SEE ALSO
1346
1347 L<Coro::AIO>.
1348
1349 =head1 AUTHOR
1350
1351 Marc Lehmann <schmorp@schmorp.de>
1352 http://home.schmorp.de/
1353
1354 =cut
1355