ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/IO-AIO/AIO.pm
Revision: 1.144
Committed: Sun Apr 19 19:17:59 2009 UTC (15 years, 1 month ago) by root
Branch: MAIN
CVS Tags: rel-3_18
Changes since 1.143: +1 -1 lines
Log Message:
3.18

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