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.32 by root, Wed Aug 17 05:26:20 2005 UTC vs.
Revision 1.54 by root, Sun Oct 22 00:19:05 2006 UTC

14 aio_unlink "/tmp/file", sub { }; 14 aio_unlink "/tmp/file", sub { };
15 15
16 aio_read $fh, 30000, 1024, $buffer, 0, sub { 16 aio_read $fh, 30000, 1024, $buffer, 0, sub {
17 $_[0] > 0 or die "read error: $!"; 17 $_[0] > 0 or die "read error: $!";
18 }; 18 };
19
20 use IO::AIO 2; # version has aio objects
21
22 my $req = aio_unlink "/tmp/file", sub { };
23 $req->cancel; # cancel request if still in queue
24
25 # AnyEvent
26 open my $fh, "<&=" . IO::AIO::poll_fileno or die "$!";
27 my $w = AnyEvent->io (fh => $fh, poll => 'r', cb => sub { IO::AIO::poll_cb });
19 28
20 # Event 29 # Event
21 Event->io (fd => IO::AIO::poll_fileno, 30 Event->io (fd => IO::AIO::poll_fileno,
22 poll => 'r', 31 poll => 'r',
23 cb => \&IO::AIO::poll_cb); 32 cb => \&IO::AIO::poll_cb);
57=cut 66=cut
58 67
59package IO::AIO; 68package IO::AIO;
60 69
61no warnings; 70no warnings;
71use strict 'vars';
62 72
63use base 'Exporter'; 73use base 'Exporter';
64 74
65use Fcntl ();
66
67BEGIN { 75BEGIN {
68 $VERSION = 1.3; 76 our $VERSION = '1.99';
69 77
70
71 @EXPORT = qw(aio_read aio_write aio_open aio_close aio_stat aio_lstat aio_unlink 78 our @EXPORT = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat
72 aio_rmdir aio_symlink aio_fsync aio_fdatasync aio_readahead); 79 aio_lstat aio_unlink aio_rmdir aio_readdir aio_scandir aio_symlink
80 aio_fsync aio_fdatasync aio_readahead aio_rename aio_link aio_move
81 aio_group);
73 @EXPORT_OK = qw(poll_fileno poll_cb min_parallel max_parallel max_outstanding nreqs); 82 our @EXPORT_OK = qw(poll_fileno poll_cb min_parallel max_parallel max_outstanding nreqs);
83
84 @IO::AIO::GRP::ISA = 'IO::AIO::REQ';
74 85
75 require XSLoader; 86 require XSLoader;
76 XSLoader::load IO::AIO, $VERSION; 87 XSLoader::load ("IO::AIO", $VERSION);
77} 88}
78 89
79=head1 FUNCTIONS 90=head1 FUNCTIONS
80 91
81=head2 AIO FUNCTIONS 92=head2 AIO FUNCTIONS
89syscall has been executed asynchronously. 100syscall has been executed asynchronously.
90 101
91All functions expecting a filehandle keep a copy of the filehandle 102All functions expecting a filehandle keep a copy of the filehandle
92internally until the request has finished. 103internally until the request has finished.
93 104
105All non-composite requests (requests that are not broken down into
106multiple requests) return objects of type L<IO::AIO::REQ> that allow
107further manipulation of running requests.
108
94The pathnames you pass to these routines I<must> be absolute and 109The pathnames you pass to these routines I<must> be absolute and
95encoded in byte form. The reason for the former is that at the time the 110encoded in byte form. The reason for the former is that at the time the
96request is being executed, the current working directory could have 111request is being executed, the current working directory could have
97changed. Alternatively, you can make sure that you never change the 112changed. Alternatively, you can make sure that you never change the
98current working directory. 113current working directory.
104environment, d) use Glib::filename_from_unicode on unicode filenames or e) 119environment, d) use Glib::filename_from_unicode on unicode filenames or e)
105use something else. 120use something else.
106 121
107=over 4 122=over 4
108 123
109=item aio_open $pathname, $flags, $mode, $callback 124=item aio_open $pathname, $flags, $mode, $callback->($fh)
110 125
111Asynchronously open or create a file and call the callback with a newly 126Asynchronously open or create a file and call the callback with a newly
112created filehandle for the file. 127created filehandle for the file.
113 128
114The pathname passed to C<aio_open> must be absolute. See API NOTES, above, 129The pathname passed to C<aio_open> must be absolute. See API NOTES, above,
131 } else { 146 } else {
132 die "open failed: $!\n"; 147 die "open failed: $!\n";
133 } 148 }
134 }; 149 };
135 150
136=item aio_close $fh, $callback 151=item aio_close $fh, $callback->($status)
137 152
138Asynchronously close a file and call the callback with the result 153Asynchronously close a file and call the callback with the result
139code. I<WARNING:> although accepted, you should not pass in a perl 154code. I<WARNING:> although accepted, you should not pass in a perl
140filehandle here, as perl will likely close the file descriptor another 155filehandle here, as perl will likely close the file descriptor another
141time when the filehandle is destroyed. Normally, you can safely call perls 156time when the filehandle is destroyed. Normally, you can safely call perls
142C<close> or just let filehandles go out of scope. 157C<close> or just let filehandles go out of scope.
143 158
144This is supposed to be a bug in the API, so that might change. It's 159This is supposed to be a bug in the API, so that might change. It's
145therefore best to avoid this function. 160therefore best to avoid this function.
146 161
147=item aio_read $fh,$offset,$length, $data,$dataoffset,$callback 162=item aio_read $fh,$offset,$length, $data,$dataoffset, $callback->($retval)
148 163
149=item aio_write $fh,$offset,$length, $data,$dataoffset,$callback 164=item aio_write $fh,$offset,$length, $data,$dataoffset, $callback->($retval)
150 165
151Reads or writes C<length> bytes from the specified C<fh> and C<offset> 166Reads or writes C<length> bytes from the specified C<fh> and C<offset>
152into the scalar given by C<data> and offset C<dataoffset> and calls the 167into the scalar given by C<data> and offset C<dataoffset> and calls the
153callback without the actual number of bytes read (or -1 on error, just 168callback without the actual number of bytes read (or -1 on error, just
154like the syscall). 169like the syscall).
163 aio_read $fh, 7, 15, $buffer, 0, sub { 178 aio_read $fh, 7, 15, $buffer, 0, sub {
164 $_[0] > 0 or die "read error: $!"; 179 $_[0] > 0 or die "read error: $!";
165 print "read $_[0] bytes: <$buffer>\n"; 180 print "read $_[0] bytes: <$buffer>\n";
166 }; 181 };
167 182
183=item aio_move $srcpath, $dstpath, $callback->($status)
184
185Try to move the I<file> (directories not supported as either source or
186destination) from C<$srcpath> to C<$dstpath> and call the callback with
187the C<0> (error) or C<-1> ok.
188
189This is a composite request that tries to rename(2) the file first. If
190rename files with C<EXDEV>, it creates the destination file with mode 0200
191and copies the contents of the source file into it using C<aio_sendfile>,
192followed by restoring atime, mtime, access mode and uid/gid, in that
193order, and unlinking the C<$srcpath>.
194
195If an error occurs, the partial destination file will be unlinked, if
196possible, except when setting atime, mtime, access mode and uid/gid, where
197errors are being ignored.
198
199=cut
200
201sub aio_move($$$) {
202 my ($src, $dst, $cb) = @_;
203
204 aio_rename $src, $dst, sub {
205 if ($_[0] && $! == EXDEV) {
206 aio_open $src, O_RDONLY, 0, sub {
207 if (my $src_fh = $_[0]) {
208 my @stat = stat $src_fh;
209
210 aio_open $dst, O_WRONLY, 0200, sub {
211 if (my $dst_fh = $_[0]) {
212 aio_sendfile $dst_fh, $src_fh, 0, $stat[7], sub {
213 close $src_fh;
214
215 if ($_[0] == $stat[7]) {
216 utime $stat[8], $stat[9], $dst;
217 chmod $stat[2] & 07777, $dst_fh;
218 chown $stat[4], $stat[5], $dst_fh;
219 close $dst_fh;
220
221 aio_unlink $src, sub {
222 $cb->($_[0]);
223 };
224 } else {
225 my $errno = $!;
226 aio_unlink $dst, sub {
227 $! = $errno;
228 $cb->(-1);
229 };
230 }
231 };
232 } else {
233 $cb->(-1);
234 }
235 },
236
237 } else {
238 $cb->(-1);
239 }
240 };
241 } else {
242 $cb->($_[0]);
243 }
244 };
245}
246
247=item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback->($retval)
248
249Tries to copy C<$length> bytes from C<$in_fh> to C<$out_fh>. It starts
250reading at byte offset C<$in_offset>, and starts writing at the current
251file offset of C<$out_fh>. Because of that, it is not safe to issue more
252than one C<aio_sendfile> per C<$out_fh>, as they will interfere with each
253other.
254
255This call tries to make use of a native C<sendfile> syscall to provide
256zero-copy operation. For this to work, C<$out_fh> should refer to a
257socket, and C<$in_fh> should refer to mmap'able file.
258
259If the native sendfile call fails or is not implemented, it will be
260emulated, so you can call C<aio_sendfile> on any type of filehandle
261regardless of the limitations of the operating system.
262
263Please note, however, that C<aio_sendfile> can read more bytes from
264C<$in_fh> than are written, and there is no way to find out how many
265bytes have been read from C<aio_sendfile> alone, as C<aio_sendfile> only
266provides the number of bytes written to C<$out_fh>. Only if the result
267value equals C<$length> one can assume that C<$length> bytes have been
268read.
269
168=item aio_readahead $fh,$offset,$length, $callback 270=item aio_readahead $fh,$offset,$length, $callback->($retval)
169 271
170C<aio_readahead> populates the page cache with data from a file so that 272C<aio_readahead> populates the page cache with data from a file so that
171subsequent reads from that file will not block on disk I/O. The C<$offset> 273subsequent reads from that file will not block on disk I/O. The C<$offset>
172argument specifies the starting point from which data is to be read and 274argument specifies the starting point from which data is to be read and
173C<$length> specifies the number of bytes to be read. I/O is performed in 275C<$length> specifies the number of bytes to be read. I/O is performed in
177file. The current file offset of the file is left unchanged. 279file. The current file offset of the file is left unchanged.
178 280
179If that syscall doesn't exist (likely if your OS isn't Linux) it will be 281If that syscall doesn't exist (likely if your OS isn't Linux) it will be
180emulated by simply reading the data, which would have a similar effect. 282emulated by simply reading the data, which would have a similar effect.
181 283
182=item aio_stat $fh_or_path, $callback 284=item aio_stat $fh_or_path, $callback->($status)
183 285
184=item aio_lstat $fh, $callback 286=item aio_lstat $fh, $callback->($status)
185 287
186Works like perl's C<stat> or C<lstat> in void context. The callback will 288Works like perl's C<stat> or C<lstat> in void context. The callback will
187be called after the stat and the results will be available using C<stat _> 289be called after the stat and the results will be available using C<stat _>
188or C<-s _> etc... 290or C<-s _> etc...
189 291
199 aio_stat "/etc/passwd", sub { 301 aio_stat "/etc/passwd", sub {
200 $_[0] and die "stat failed: $!"; 302 $_[0] and die "stat failed: $!";
201 print "size is ", -s _, "\n"; 303 print "size is ", -s _, "\n";
202 }; 304 };
203 305
204=item aio_unlink $pathname, $callback 306=item aio_unlink $pathname, $callback->($status)
205 307
206Asynchronously unlink (delete) a file and call the callback with the 308Asynchronously unlink (delete) a file and call the callback with the
207result code. 309result code.
208 310
311=item aio_link $srcpath, $dstpath, $callback->($status)
312
313Asynchronously create a new link to the existing object at C<$srcpath> at
314the path C<$dstpath> and call the callback with the result code.
315
316=item aio_symlink $srcpath, $dstpath, $callback->($status)
317
318Asynchronously create a new symbolic link to the existing object at C<$srcpath> at
319the path C<$dstpath> and call the callback with the result code.
320
321=item aio_rename $srcpath, $dstpath, $callback->($status)
322
323Asynchronously rename the object at C<$srcpath> to C<$dstpath>, just as
324rename(2) and call the callback with the result code.
325
209=item aio_rmdir $pathname, $callback 326=item aio_rmdir $pathname, $callback->($status)
210 327
211Asynchronously rmdir (delete) a directory and call the callback with the 328Asynchronously rmdir (delete) a directory and call the callback with the
212result code. 329result code.
213 330
331=item aio_readdir $pathname, $callback->($entries)
332
333Unlike the POSIX call of the same name, C<aio_readdir> reads an entire
334directory (i.e. opendir + readdir + closedir). The entries will not be
335sorted, and will B<NOT> include the C<.> and C<..> entries.
336
337The callback a single argument which is either C<undef> or an array-ref
338with the filenames.
339
340=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)
341
342Scans a directory (similar to C<aio_readdir>) but additionally tries to
343separate the entries of directory C<$path> into two sets of names, ones
344you can recurse into (directories or links to them), and ones you cannot
345recurse into (everything else).
346
347C<aio_scandir> is a composite request that consists of many sub
348requests. C<$maxreq> specifies the maximum number of outstanding aio
349requests that this function generates. If it is C<< <= 0 >>, then a
350suitable default will be chosen (currently 8).
351
352On error, the callback is called without arguments, otherwise it receives
353two array-refs with path-relative entry names.
354
355Example:
356
357 aio_scandir $dir, 0, sub {
358 my ($dirs, $nondirs) = @_;
359 print "real directories: @$dirs\n";
360 print "everything else: @$nondirs\n";
361 };
362
363Implementation notes.
364
365The C<aio_readdir> cannot be avoided, but C<stat()>'ing every entry can.
366
367After reading the directory, the modification time, size etc. of the
368directory before and after the readdir is checked, and if they match (and
369isn't the current time), the link count will be used to decide how many
370entries are directories (if >= 2). Otherwise, no knowledge of the number
371of subdirectories will be assumed.
372
373Then entries will be sorted into likely directories (everything without
374a non-initial dot currently) and likely non-directories (everything
375else). Then every entry plus an appended C</.> will be C<stat>'ed,
376likely directories first. If that succeeds, it assumes that the entry
377is a directory or a symlink to directory (which will be checked
378seperately). This is often faster than stat'ing the entry itself because
379filesystems might detect the type of the entry without reading the inode
380data (e.g. ext2fs filetype feature).
381
382If the known number of directories (link count - 2) has been reached, the
383rest of the entries is assumed to be non-directories.
384
385This only works with certainty on POSIX (= UNIX) filesystems, which
386fortunately are the vast majority of filesystems around.
387
388It will also likely work on non-POSIX filesystems with reduced efficiency
389as those tend to return 0 or 1 as link counts, which disables the
390directory counting heuristic.
391
392=cut
393
394sub aio_scandir($$$) {
395 my ($path, $maxreq, $cb) = @_;
396
397 $maxreq = 8 if $maxreq <= 0;
398
399 # stat once
400 aio_stat $path, sub {
401 return $cb->() if $_[0];
402 my $now = time;
403 my $hash1 = join ":", (stat _)[0,1,3,7,9];
404
405 # read the directory entries
406 aio_readdir $path, sub {
407 my $entries = shift
408 or return $cb->();
409
410 # stat the dir another time
411 aio_stat $path, sub {
412 my $hash2 = join ":", (stat _)[0,1,3,7,9];
413
414 my $ndirs;
415
416 # take the slow route if anything looks fishy
417 if ($hash1 ne $hash2 or (stat _)[9] == $now) {
418 $ndirs = -1;
419 } else {
420 # if nlink == 2, we are finished
421 # on non-posix-fs's, we rely on nlink < 2
422 $ndirs = (stat _)[3] - 2
423 or return $cb->([], $entries);
424 }
425
426 # sort into likely dirs and likely nondirs
427 # dirs == files without ".", short entries first
428 $entries = [map $_->[0],
429 sort { $b->[1] cmp $a->[1] }
430 map [$_, sprintf "%s%04d", (/.\./ ? "1" : "0"), length],
431 @$entries];
432
433 my (@dirs, @nondirs);
434
435 my ($statcb, $schedcb);
436 my $nreq = 0;
437
438 $schedcb = sub {
439 if (@$entries) {
440 if ($nreq < $maxreq) {
441 my $ent = pop @$entries;
442 $nreq++;
443 aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) };
444 }
445 } elsif (!$nreq) {
446 # finished
447 undef $statcb;
448 undef $schedcb;
449 $cb->(\@dirs, \@nondirs) if $cb;
450 undef $cb;
451 }
452 };
453 $statcb = sub {
454 my ($status, $entry) = @_;
455
456 if ($status < 0) {
457 $nreq--;
458 push @nondirs, $entry;
459 &$schedcb;
460 } else {
461 # need to check for real directory
462 aio_lstat "$path/$entry", sub {
463 $nreq--;
464
465 if (-d _) {
466 push @dirs, $entry;
467
468 if (!--$ndirs) {
469 push @nondirs, @$entries;
470 $entries = [];
471 }
472 } else {
473 push @nondirs, $entry;
474 }
475
476 &$schedcb;
477 }
478 }
479 };
480
481 &$schedcb while @$entries && $nreq < $maxreq;
482 };
483 };
484 };
485}
486
214=item aio_fsync $fh, $callback 487=item aio_fsync $fh, $callback->($status)
215 488
216Asynchronously call fsync on the given filehandle and call the callback 489Asynchronously call fsync on the given filehandle and call the callback
217with the fsync result code. 490with the fsync result code.
218 491
219=item aio_fdatasync $fh, $callback 492=item aio_fdatasync $fh, $callback->($status)
220 493
221Asynchronously call fdatasync on the given filehandle and call the 494Asynchronously call fdatasync on the given filehandle and call the
222callback with the fdatasync result code. 495callback with the fdatasync result code.
223 496
224If this call isn't available because your OS lacks it or it couldn't be 497If this call isn't available because your OS lacks it or it couldn't be
225detected, it will be emulated by calling C<fsync> instead. 498detected, it will be emulated by calling C<fsync> instead.
499
500=item aio_group $callback->()
501
502=item aio_sleep $fractional_seconds, $callback->() *NOT EXPORTED*
503
504Mainly used for debugging and benchmarking, this aio request puts one of
505the request workers to sleep for the given time.
506
507=back
508
509=head2 IO::AIO::REQ CLASS
510
511All non-aggregate C<aio_*> functions return an object of this class when
512called in non-void context.
513
514A request always moves through the following five states in its lifetime,
515in order: B<ready> (request has been created, but has not been executed
516yet), B<execute> (request is currently being executed), B<pending>
517(request has been executed but callback has not been called yet),
518B<result> (results are being processed synchronously, includes calling the
519callback) and B<done> (request has reached the end of its lifetime and
520holds no resources anymore).
521
522=over 4
523
524=item $req->cancel
525
526Cancels the request, if possible. Has the effect of skipping execution
527when entering the B<execute> state and skipping calling the callback when
528entering the the B<result> state, but will leave the request otherwise
529untouched. That means that requests that currently execute will not be
530stopped and resources held by the request will not be freed prematurely.
226 531
227=back 532=back
228 533
229=head2 SUPPORT FUNCTIONS 534=head2 SUPPORT FUNCTIONS
230 535
288 IO::AIO::poll_wait, IO::AIO::poll_cb 593 IO::AIO::poll_wait, IO::AIO::poll_cb
289 if IO::AIO::nreqs; 594 if IO::AIO::nreqs;
290 595
291=item IO::AIO::min_parallel $nthreads 596=item IO::AIO::min_parallel $nthreads
292 597
293Set the minimum number of AIO threads to C<$nthreads>. The default is 598Set the minimum number of AIO threads to C<$nthreads>. The current default
294C<1>, which means a single asynchronous operation can be done at one time 599is C<4>, which means four asynchronous operations can be done at one time
295(the number of outstanding operations, however, is unlimited). 600(the number of outstanding operations, however, is unlimited).
601
602IO::AIO starts threads only on demand, when an AIO request is queued and
603no free thread exists.
296 604
297It is recommended to keep the number of threads low, as some Linux 605It is recommended to keep the number of threads low, as some Linux
298kernel versions will scale negatively with the number of threads (higher 606kernel versions will scale negatively with the number of threads (higher
299parallelity => MUCH higher latency). With current Linux 2.6 versions, 4-32 607parallelity => MUCH higher latency). With current Linux 2.6 versions, 4-32
300threads should be fine. 608threads should be fine.
301 609
302Under normal circumstances you don't need to call this function, as this 610Under most circumstances you don't need to call this function, as the
303module automatically starts some threads (the exact number might change, 611module selects a default that is suitable for low to moderate load.
304and is currently 4).
305 612
306=item IO::AIO::max_parallel $nthreads 613=item IO::AIO::max_parallel $nthreads
307 614
308Sets the maximum number of AIO threads to C<$nthreads>. If more than 615Sets the maximum number of AIO threads to C<$nthreads>. If more than the
309the specified number of threads are currently running, kill them. This 616specified number of threads are currently running, this function kills
310function blocks until the limit is reached. 617them. This function blocks until the limit is reached.
618
619While C<$nthreads> are zero, aio requests get queued but not executed
620until the number of threads has been increased again.
311 621
312This module automatically runs C<max_parallel 0> at program end, to ensure 622This module automatically runs C<max_parallel 0> at program end, to ensure
313that all threads are killed and that there are no outstanding requests. 623that all threads are killed and that there are no outstanding requests.
314 624
315Under normal circumstances you don't need to call this function. 625Under normal circumstances you don't need to call this function.
319Sets the maximum number of outstanding requests to C<$nreqs>. If you 629Sets the maximum number of outstanding requests to C<$nreqs>. If you
320try to queue up more than this number of requests, the caller will block until 630try to queue up more than this number of requests, the caller will block until
321some requests have been handled. 631some requests have been handled.
322 632
323The default is very large, so normally there is no practical limit. If you 633The default is very large, so normally there is no practical limit. If you
324queue up many requests in a loop it it often improves speed if you set 634queue up many requests in a loop it often improves speed if you set
325this to a relatively low number, such as C<100>. 635this to a relatively low number, such as C<100>.
326 636
327Under normal circumstances you don't need to call this function. 637Under normal circumstances you don't need to call this function.
328 638
329=back 639=back
354 664
3551; 6651;
356 666
357=head2 FORK BEHAVIOUR 667=head2 FORK BEHAVIOUR
358 668
669This module should do "the right thing" when the process using it forks:
670
359Before the fork IO::AIO enters a quiescent state where no requests can be 671Before the fork, IO::AIO enters a quiescent state where no requests
360added in other threads and no results will be processed. After the fork 672can be added in other threads and no results will be processed. After
361the parent simply leaves the quiescent state and continues request/result 673the fork the parent simply leaves the quiescent state and continues
362processing, while the child clears the request/result queue and starts the 674request/result processing, while the child clears the request/result
363same number of threads as were in use by the parent. 675queue (so the requests started before the fork will only be handled in
676the parent). Threads will be started on demand until the limit ste in the
677parent process has been reached again.
678
679In short: the parent will, after a short pause, continue as if fork had
680not been called, while the child will act as if IO::AIO has not been used
681yet.
364 682
365=head1 SEE ALSO 683=head1 SEE ALSO
366 684
367L<Coro>, L<Linux::AIO>. 685L<Coro>, L<Linux::AIO> (obsolete).
368 686
369=head1 AUTHOR 687=head1 AUTHOR
370 688
371 Marc Lehmann <schmorp@schmorp.de> 689 Marc Lehmann <schmorp@schmorp.de>
372 http://home.schmorp.de/ 690 http://home.schmorp.de/

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines