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.39 by root, Sun Aug 28 11:05:50 2005 UTC vs.
Revision 1.55 by root, Sun Oct 22 00:49:29 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.6; 76 our $VERSION = '2.0';
69 77
70 @EXPORT = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat 78 our @EXPORT = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat
71 aio_lstat aio_unlink aio_rmdir aio_readdir aio_symlink 79 aio_lstat aio_unlink aio_rmdir aio_readdir aio_scandir aio_symlink
72 aio_fsync aio_fdatasync aio_readahead); 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 82 our @EXPORT_OK = qw(poll_fileno poll_cb min_parallel max_parallel max_outstanding nreqs);
74 max_outstanding nreqs); 83
84 @IO::AIO::GRP::ISA = 'IO::AIO::REQ';
75 85
76 require XSLoader; 86 require XSLoader;
77 XSLoader::load IO::AIO, $VERSION; 87 XSLoader::load ("IO::AIO", $VERSION);
78} 88}
79 89
80=head1 FUNCTIONS 90=head1 FUNCTIONS
81 91
82=head2 AIO FUNCTIONS 92=head2 AIO FUNCTIONS
90syscall has been executed asynchronously. 100syscall has been executed asynchronously.
91 101
92All functions expecting a filehandle keep a copy of the filehandle 102All functions expecting a filehandle keep a copy of the filehandle
93internally until the request has finished. 103internally until the request has finished.
94 104
105All requests return objects of type L<IO::AIO::REQ> that allow further
106manipulation of those requests while they are in-flight.
107
95The pathnames you pass to these routines I<must> be absolute and 108The pathnames you pass to these routines I<must> be absolute and
96encoded in byte form. The reason for the former is that at the time the 109encoded in byte form. The reason for the former is that at the time the
97request is being executed, the current working directory could have 110request is being executed, the current working directory could have
98changed. Alternatively, you can make sure that you never change the 111changed. Alternatively, you can make sure that you never change the
99current working directory. 112current working directory.
105environment, d) use Glib::filename_from_unicode on unicode filenames or e) 118environment, d) use Glib::filename_from_unicode on unicode filenames or e)
106use something else. 119use something else.
107 120
108=over 4 121=over 4
109 122
110=item aio_open $pathname, $flags, $mode, $callback 123=item aio_open $pathname, $flags, $mode, $callback->($fh)
111 124
112Asynchronously open or create a file and call the callback with a newly 125Asynchronously open or create a file and call the callback with a newly
113created filehandle for the file. 126created filehandle for the file.
114 127
115The pathname passed to C<aio_open> must be absolute. See API NOTES, above, 128The pathname passed to C<aio_open> must be absolute. See API NOTES, above,
132 } else { 145 } else {
133 die "open failed: $!\n"; 146 die "open failed: $!\n";
134 } 147 }
135 }; 148 };
136 149
137=item aio_close $fh, $callback 150=item aio_close $fh, $callback->($status)
138 151
139Asynchronously close a file and call the callback with the result 152Asynchronously close a file and call the callback with the result
140code. I<WARNING:> although accepted, you should not pass in a perl 153code. I<WARNING:> although accepted, you should not pass in a perl
141filehandle here, as perl will likely close the file descriptor another 154filehandle here, as perl will likely close the file descriptor another
142time when the filehandle is destroyed. Normally, you can safely call perls 155time when the filehandle is destroyed. Normally, you can safely call perls
143C<close> or just let filehandles go out of scope. 156C<close> or just let filehandles go out of scope.
144 157
145This is supposed to be a bug in the API, so that might change. It's 158This is supposed to be a bug in the API, so that might change. It's
146therefore best to avoid this function. 159therefore best to avoid this function.
147 160
148=item aio_read $fh,$offset,$length, $data,$dataoffset,$callback 161=item aio_read $fh,$offset,$length, $data,$dataoffset, $callback->($retval)
149 162
150=item aio_write $fh,$offset,$length, $data,$dataoffset,$callback 163=item aio_write $fh,$offset,$length, $data,$dataoffset, $callback->($retval)
151 164
152Reads or writes C<length> bytes from the specified C<fh> and C<offset> 165Reads or writes C<length> bytes from the specified C<fh> and C<offset>
153into the scalar given by C<data> and offset C<dataoffset> and calls the 166into the scalar given by C<data> and offset C<dataoffset> and calls the
154callback without the actual number of bytes read (or -1 on error, just 167callback without the actual number of bytes read (or -1 on error, just
155like the syscall). 168like the syscall).
164 aio_read $fh, 7, 15, $buffer, 0, sub { 177 aio_read $fh, 7, 15, $buffer, 0, sub {
165 $_[0] > 0 or die "read error: $!"; 178 $_[0] > 0 or die "read error: $!";
166 print "read $_[0] bytes: <$buffer>\n"; 179 print "read $_[0] bytes: <$buffer>\n";
167 }; 180 };
168 181
182=item aio_move $srcpath, $dstpath, $callback->($status)
183
184Try to move the I<file> (directories not supported as either source or
185destination) from C<$srcpath> to C<$dstpath> and call the callback with
186the C<0> (error) or C<-1> ok.
187
188This is a composite request that tries to rename(2) the file first. If
189rename files with C<EXDEV>, it creates the destination file with mode 0200
190and copies the contents of the source file into it using C<aio_sendfile>,
191followed by restoring atime, mtime, access mode and uid/gid, in that
192order, and unlinking the C<$srcpath>.
193
194If an error occurs, the partial destination file will be unlinked, if
195possible, except when setting atime, mtime, access mode and uid/gid, where
196errors are being ignored.
197
198=cut
199
200sub aio_move($$$) {
201 my ($src, $dst, $cb) = @_;
202
203 my $grp = aio_group;
204
205 add $grp aio_rename $src, $dst, sub {
206 if ($_[0] && $! == EXDEV) {
207 add $grp aio_open $src, O_RDONLY, 0, sub {
208 if (my $src_fh = $_[0]) {
209 my @stat = stat $src_fh;
210
211 add $grp aio_open $dst, O_WRONLY, 0200, sub {
212 if (my $dst_fh = $_[0]) {
213 add $grp aio_sendfile $dst_fh, $src_fh, 0, $stat[7], sub {
214 close $src_fh;
215
216 if ($_[0] == $stat[7]) {
217 utime $stat[8], $stat[9], $dst;
218 chmod $stat[2] & 07777, $dst_fh;
219 chown $stat[4], $stat[5], $dst_fh;
220 close $dst_fh;
221
222 add $grp aio_unlink $src, sub {
223 $cb->($_[0]);
224 };
225 } else {
226 my $errno = $!;
227 add $grp aio_unlink $dst, sub {
228 $! = $errno;
229 $cb->(-1);
230 };
231 }
232 };
233 } else {
234 $cb->(-1);
235 }
236 },
237
238 } else {
239 $cb->(-1);
240 }
241 };
242 } else {
243 $cb->($_[0]);
244 }
245 };
246
247 $grp
248}
249
169=item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback 250=item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback->($retval)
170 251
171Tries to copy C<$length> bytes from C<$in_fh> to C<$out_fh>. It starts 252Tries to copy C<$length> bytes from C<$in_fh> to C<$out_fh>. It starts
172reading at byte offset C<$in_offset>, and starts writing at the current 253reading at byte offset C<$in_offset>, and starts writing at the current
173file offset of C<$out_fh>. Because of that, it is not safe to issue more 254file offset of C<$out_fh>. Because of that, it is not safe to issue more
174than one C<aio_sendfile> per C<$out_fh>, as they will interfere with each 255than one C<aio_sendfile> per C<$out_fh>, as they will interfere with each
187bytes have been read from C<aio_sendfile> alone, as C<aio_sendfile> only 268bytes have been read from C<aio_sendfile> alone, as C<aio_sendfile> only
188provides the number of bytes written to C<$out_fh>. Only if the result 269provides the number of bytes written to C<$out_fh>. Only if the result
189value equals C<$length> one can assume that C<$length> bytes have been 270value equals C<$length> one can assume that C<$length> bytes have been
190read. 271read.
191 272
192=item aio_readahead $fh,$offset,$length, $callback 273=item aio_readahead $fh,$offset,$length, $callback->($retval)
193 274
194C<aio_readahead> populates the page cache with data from a file so that 275C<aio_readahead> populates the page cache with data from a file so that
195subsequent reads from that file will not block on disk I/O. The C<$offset> 276subsequent reads from that file will not block on disk I/O. The C<$offset>
196argument specifies the starting point from which data is to be read and 277argument specifies the starting point from which data is to be read and
197C<$length> specifies the number of bytes to be read. I/O is performed in 278C<$length> specifies the number of bytes to be read. I/O is performed in
201file. The current file offset of the file is left unchanged. 282file. The current file offset of the file is left unchanged.
202 283
203If that syscall doesn't exist (likely if your OS isn't Linux) it will be 284If that syscall doesn't exist (likely if your OS isn't Linux) it will be
204emulated by simply reading the data, which would have a similar effect. 285emulated by simply reading the data, which would have a similar effect.
205 286
206=item aio_stat $fh_or_path, $callback 287=item aio_stat $fh_or_path, $callback->($status)
207 288
208=item aio_lstat $fh, $callback 289=item aio_lstat $fh, $callback->($status)
209 290
210Works like perl's C<stat> or C<lstat> in void context. The callback will 291Works like perl's C<stat> or C<lstat> in void context. The callback will
211be called after the stat and the results will be available using C<stat _> 292be called after the stat and the results will be available using C<stat _>
212or C<-s _> etc... 293or C<-s _> etc...
213 294
223 aio_stat "/etc/passwd", sub { 304 aio_stat "/etc/passwd", sub {
224 $_[0] and die "stat failed: $!"; 305 $_[0] and die "stat failed: $!";
225 print "size is ", -s _, "\n"; 306 print "size is ", -s _, "\n";
226 }; 307 };
227 308
228=item aio_unlink $pathname, $callback 309=item aio_unlink $pathname, $callback->($status)
229 310
230Asynchronously unlink (delete) a file and call the callback with the 311Asynchronously unlink (delete) a file and call the callback with the
231result code. 312result code.
232 313
314=item aio_link $srcpath, $dstpath, $callback->($status)
315
316Asynchronously create a new link to the existing object at C<$srcpath> at
317the path C<$dstpath> and call the callback with the result code.
318
319=item aio_symlink $srcpath, $dstpath, $callback->($status)
320
321Asynchronously create a new symbolic link to the existing object at C<$srcpath> at
322the path C<$dstpath> and call the callback with the result code.
323
324=item aio_rename $srcpath, $dstpath, $callback->($status)
325
326Asynchronously rename the object at C<$srcpath> to C<$dstpath>, just as
327rename(2) and call the callback with the result code.
328
233=item aio_rmdir $pathname, $callback 329=item aio_rmdir $pathname, $callback->($status)
234 330
235Asynchronously rmdir (delete) a directory and call the callback with the 331Asynchronously rmdir (delete) a directory and call the callback with the
236result code. 332result code.
237 333
238=item aio_readdir $pathname $callback 334=item aio_readdir $pathname, $callback->($entries)
239 335
240Unlike the POSIX call of the same name, C<aio_readdir> reads an entire 336Unlike the POSIX call of the same name, C<aio_readdir> reads an entire
241directory (i.e. opendir + readdir + closedir). The entries will not be 337directory (i.e. opendir + readdir + closedir). The entries will not be
242sorted, and will B<NOT> include the C<.> and C<..> entries. 338sorted, and will B<NOT> include the C<.> and C<..> entries.
243 339
244The callback a single argument which is either C<undef> or an array-ref 340The callback a single argument which is either C<undef> or an array-ref
245with the filenames. 341with the filenames.
246 342
343=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)
344
345Scans a directory (similar to C<aio_readdir>) but additionally tries to
346separate the entries of directory C<$path> into two sets of names, ones
347you can recurse into (directories or links to them), and ones you cannot
348recurse into (everything else).
349
350C<aio_scandir> is a composite request that consists of many sub
351requests. C<$maxreq> specifies the maximum number of outstanding aio
352requests that this function generates. If it is C<< <= 0 >>, then a
353suitable default will be chosen (currently 8).
354
355On error, the callback is called without arguments, otherwise it receives
356two array-refs with path-relative entry names.
357
358Example:
359
360 aio_scandir $dir, 0, sub {
361 my ($dirs, $nondirs) = @_;
362 print "real directories: @$dirs\n";
363 print "everything else: @$nondirs\n";
364 };
365
366Implementation notes.
367
368The C<aio_readdir> cannot be avoided, but C<stat()>'ing every entry can.
369
370After reading the directory, the modification time, size etc. of the
371directory before and after the readdir is checked, and if they match (and
372isn't the current time), the link count will be used to decide how many
373entries are directories (if >= 2). Otherwise, no knowledge of the number
374of subdirectories will be assumed.
375
376Then entries will be sorted into likely directories (everything without
377a non-initial dot currently) and likely non-directories (everything
378else). Then every entry plus an appended C</.> will be C<stat>'ed,
379likely directories first. If that succeeds, it assumes that the entry
380is a directory or a symlink to directory (which will be checked
381seperately). This is often faster than stat'ing the entry itself because
382filesystems might detect the type of the entry without reading the inode
383data (e.g. ext2fs filetype feature).
384
385If the known number of directories (link count - 2) has been reached, the
386rest of the entries is assumed to be non-directories.
387
388This only works with certainty on POSIX (= UNIX) filesystems, which
389fortunately are the vast majority of filesystems around.
390
391It will also likely work on non-POSIX filesystems with reduced efficiency
392as those tend to return 0 or 1 as link counts, which disables the
393directory counting heuristic.
394
395=cut
396
397sub aio_scandir($$$) {
398 my ($path, $maxreq, $cb) = @_;
399
400 my $grp = aio_group;
401
402 $maxreq = 8 if $maxreq <= 0;
403
404 # stat once
405 add $grp aio_stat $path, sub {
406 return $cb->() if $_[0];
407 my $now = time;
408 my $hash1 = join ":", (stat _)[0,1,3,7,9];
409
410 # read the directory entries
411 add $grp aio_readdir $path, sub {
412 my $entries = shift
413 or return $cb->();
414
415 # stat the dir another time
416 add $grp aio_stat $path, sub {
417 my $hash2 = join ":", (stat _)[0,1,3,7,9];
418
419 my $ndirs;
420
421 # take the slow route if anything looks fishy
422 if ($hash1 ne $hash2 or (stat _)[9] == $now) {
423 $ndirs = -1;
424 } else {
425 # if nlink == 2, we are finished
426 # on non-posix-fs's, we rely on nlink < 2
427 $ndirs = (stat _)[3] - 2
428 or return $cb->([], $entries);
429 }
430
431 # sort into likely dirs and likely nondirs
432 # dirs == files without ".", short entries first
433 $entries = [map $_->[0],
434 sort { $b->[1] cmp $a->[1] }
435 map [$_, sprintf "%s%04d", (/.\./ ? "1" : "0"), length],
436 @$entries];
437
438 my (@dirs, @nondirs);
439
440 my ($statcb, $schedcb);
441 my $nreq = 0;
442
443 $schedcb = sub {
444 if (@$entries) {
445 if ($nreq < $maxreq) {
446 my $ent = pop @$entries;
447 $nreq++;
448 add $grp aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) };
449 }
450 } elsif (!$nreq) {
451 # finished
452 undef $statcb;
453 undef $schedcb;
454 $cb->(\@dirs, \@nondirs) if $cb;
455 undef $cb;
456 }
457 };
458 $statcb = sub {
459 my ($status, $entry) = @_;
460
461 if ($status < 0) {
462 $nreq--;
463 push @nondirs, $entry;
464 &$schedcb;
465 } else {
466 # need to check for real directory
467 add $grp aio_lstat "$path/$entry", sub {
468 $nreq--;
469
470 if (-d _) {
471 push @dirs, $entry;
472
473 if (!--$ndirs) {
474 push @nondirs, @$entries;
475 $entries = [];
476 }
477 } else {
478 push @nondirs, $entry;
479 }
480
481 &$schedcb;
482 }
483 }
484 };
485
486 &$schedcb while @$entries && $nreq < $maxreq;
487 };
488 };
489 };
490
491 $grp
492}
493
247=item aio_fsync $fh, $callback 494=item aio_fsync $fh, $callback->($status)
248 495
249Asynchronously call fsync on the given filehandle and call the callback 496Asynchronously call fsync on the given filehandle and call the callback
250with the fsync result code. 497with the fsync result code.
251 498
252=item aio_fdatasync $fh, $callback 499=item aio_fdatasync $fh, $callback->($status)
253 500
254Asynchronously call fdatasync on the given filehandle and call the 501Asynchronously call fdatasync on the given filehandle and call the
255callback with the fdatasync result code. 502callback with the fdatasync result code.
256 503
257If this call isn't available because your OS lacks it or it couldn't be 504If this call isn't available because your OS lacks it or it couldn't be
258detected, it will be emulated by calling C<fsync> instead. 505detected, it will be emulated by calling C<fsync> instead.
506
507=item aio_group $callback->()
508
509[EXPERIMENTAL]
510
511This is a very special aio request: Instead of doing something, it is a
512container for other aio requests, which is useful if you want to bundle
513many requests into a single, composite, request.
514
515Returns an object of class L<IO::AIO::GRP>. See its documentation below
516for more info.
517
518Example:
519
520 my $grp = aio_group sub {
521 print "all stats done\n";
522 };
523
524 add $grp
525 (aio_stat ...),
526 (aio_stat ...),
527 ...;
528
529=item aio_sleep $fractional_seconds, $callback->() *NOT EXPORTED*
530
531Mainly used for debugging and benchmarking, this aio request puts one of
532the request workers to sleep for the given time.
533
534=back
535
536=head2 IO::AIO::REQ CLASS
537
538All non-aggregate C<aio_*> functions return an object of this class when
539called in non-void context.
540
541A request always moves through the following five states in its lifetime,
542in order: B<ready> (request has been created, but has not been executed
543yet), B<execute> (request is currently being executed), B<pending>
544(request has been executed but callback has not been called yet),
545B<result> (results are being processed synchronously, includes calling the
546callback) and B<done> (request has reached the end of its lifetime and
547holds no resources anymore).
548
549=over 4
550
551=item $req->cancel
552
553Cancels the request, if possible. Has the effect of skipping execution
554when entering the B<execute> state and skipping calling the callback when
555entering the the B<result> state, but will leave the request otherwise
556untouched. That means that requests that currently execute will not be
557stopped and resources held by the request will not be freed prematurely.
558
559=back
560
561=head2 IO::AIO::GRP CLASS
562
563This class is a subclass of L<IO::AIO::REQ>, so all its methods apply to
564objects of this class, too.
565
566A IO::AIO::GRP object is a special request that can contain multiple other
567aio requests.
568
569You create one by calling the C<aio_group> constructing function with a
570callback that will be called when all contained requests have entered the
571C<done> state:
572
573 my $grp = aio_group sub {
574 print "all requests are done\n";
575 };
576
577You add requests by calling the C<add> method with one or more
578C<IO::AIO::REQ> objects:
579
580 $grp->add (aio_unlink "...");
581
582 add $grp aio_stat "...", sub { ... };
583
584This makes it very easy to create composite requests (see the source of
585C<aio_move> for an application) that work and feel like simple requests.
586
587The IO::AIO::GRP objects will be cleaned up during calls to
588C<IO::AIO::poll_cb>, just like any other request.
589
590They can be canceled like any other request. Canceling will cancel not
591just the request itself, but also all requests it contains.
592
593They can also can also be added to other IO::AIO::GRP objects.
594
595Their lifetime, simplified, looks like this: when they are empty, they
596will finish very quickly. If they contain only requests that are in the
597C<done> state, they will also finish. Otherwise they will continue to
598exist.
599
600=over 4
601
602=item $grp->add (...)
603
604=item add $grp ...
605
606Add one or more
607Cancels the request, if possible. Has the effect of skipping execution
608when entering the B<execute> state and skipping calling the callback when
609entering the the B<result> state, but will leave the request otherwise
610untouched. That means that requests that currently execute will not be
611stopped and resources held by the request will not be freed prematurely.
259 612
260=back 613=back
261 614
262=head2 SUPPORT FUNCTIONS 615=head2 SUPPORT FUNCTIONS
263 616
391} 744}
392 745
3931; 7461;
394 747
395=head2 FORK BEHAVIOUR 748=head2 FORK BEHAVIOUR
749
750This module should do "the right thing" when the process using it forks:
396 751
397Before the fork, IO::AIO enters a quiescent state where no requests 752Before the fork, IO::AIO enters a quiescent state where no requests
398can be added in other threads and no results will be processed. After 753can be added in other threads and no results will be processed. After
399the fork the parent simply leaves the quiescent state and continues 754the fork the parent simply leaves the quiescent state and continues
400request/result processing, while the child clears the request/result 755request/result processing, while the child clears the request/result
401queue (so the requests started before the fork will only be handled in 756queue (so the requests started before the fork will only be handled in
402the parent). Threats will be started on demand until the limit ste in the 757the parent). Threads will be started on demand until the limit ste in the
403parent process has been reached again. 758parent process has been reached again.
404 759
760In short: the parent will, after a short pause, continue as if fork had
761not been called, while the child will act as if IO::AIO has not been used
762yet.
763
405=head1 SEE ALSO 764=head1 SEE ALSO
406 765
407L<Coro>, L<Linux::AIO>. 766L<Coro>, L<Linux::AIO> (obsolete).
408 767
409=head1 AUTHOR 768=head1 AUTHOR
410 769
411 Marc Lehmann <schmorp@schmorp.de> 770 Marc Lehmann <schmorp@schmorp.de>
412 http://home.schmorp.de/ 771 http://home.schmorp.de/

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines