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

Comparing IO-AIO/AIO.pm (file contents):
Revision 1.51 by root, Sat Jun 24 19:14:04 2006 UTC vs.
Revision 1.122 by root, Sat Apr 26 12:00:23 2008 UTC

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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines