IO-AIO/AIO.pm

=head1 NAME

IO::AIO - Asynchronous Input/Output

=head1 SYNOPSIS

 use IO::AIO;

 aio_open "/etc/passwd", O_RDONLY, 0, sub {
    my ($fh) = @_;
    ...
 };

 aio_unlink "/tmp/file", sub { };

 aio_read $fh, 30000, 1024, $buffer, 0, sub {
    $_[0] > 0 or die "read error: $!";
 };

 # Event
 Event->io (fd => IO::AIO::poll_fileno,
            poll => 'r',
            cb => \&IO::AIO::poll_cb);

 # Glib/Gtk2
 add_watch Glib::IO IO::AIO::poll_fileno,
           in => sub { IO::AIO::poll_cb; 1 };

 # Tk
 Tk::Event::IO->fileevent (IO::AIO::poll_fileno, "",
                           readable => \&IO::AIO::poll_cb);

 # Danga::Socket
 Danga::Socket->AddOtherFds (IO::AIO::poll_fileno =>
                             \&IO::AIO::poll_cb);


=head1 DESCRIPTION

This module implements asynchronous I/O using whatever means your
operating system supports.

Currently, a number of threads are started that execute your read/writes
and signal their completion. You don't need thread support in your libc or
perl, and the threads created by this module will not be visible to the
pthreads library. In the future, this module might make use of the native
aio functions available on many operating systems. However, they are often
not well-supported (Linux doesn't allow them on normal files currently,
for example), and they would only support aio_read and aio_write, so the
remaining functionality would have to be implemented using threads anyway.

Although the module will work with in the presence of other threads, it is
currently not reentrant, so use appropriate locking yourself, always call
C<poll_cb> from within the same thread, or never call C<poll_cb> (or other
C<aio_> functions) recursively.

=cut

package IO::AIO;

no warnings;

use base 'Exporter';

use Fcntl ();

BEGIN {
   $VERSION = '1.61';

   @EXPORT = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat
                aio_lstat aio_unlink aio_rmdir aio_readdir aio_scandir aio_symlink
                aio_fsync aio_fdatasync aio_readahead);
   @EXPORT_OK = qw(poll_fileno poll_cb min_parallel max_parallel
                   max_outstanding nreqs);

   require XSLoader;
   XSLoader::load IO::AIO, $VERSION;
}

=head1 FUNCTIONS

=head2 AIO FUNCTIONS

All the C<aio_*> calls are more or less thin wrappers around the syscall
with the same name (sans C<aio_>). The arguments are similar or identical,
and they all accept an additional (and optional) C<$callback> argument
which must be a code reference. This code reference will get called with
the syscall return code (e.g. most syscalls return C<-1> on error, unlike
perl, which usually delivers "false") as it's sole argument when the given
syscall has been executed asynchronously.

All functions expecting a filehandle keep a copy of the filehandle
internally until the request has finished.

The pathnames you pass to these routines I<must> be absolute and
encoded in byte form. The reason for the former is that at the time the
request is being executed, the current working directory could have
changed. Alternatively, you can make sure that you never change the
current working directory.

To encode pathnames to byte form, either make sure you either: a)
always pass in filenames you got from outside (command line, readdir
etc.), b) are ASCII or ISO 8859-1, c) use the Encode module and encode
your pathnames to the locale (or other) encoding in effect in the user
environment, d) use Glib::filename_from_unicode on unicode filenames or e)
use something else.

=over 4

=item aio_open $pathname, $flags, $mode, $callback->($fh)

Asynchronously open or create a file and call the callback with a newly
created filehandle for the file.

The pathname passed to C<aio_open> must be absolute. See API NOTES, above,
for an explanation.

The C<$flags> argument is a bitmask. See the C<Fcntl> module for a
list. They are the same as used by C<sysopen>.

Likewise, C<$mode> specifies the mode of the newly created file, if it
didn't exist and C<O_CREAT> has been given, just like perl's C<sysopen>,
except that it is mandatory (i.e. use C<0> if you don't create new files,
and C<0666> or C<0777> if you do).

Example:

   aio_open "/etc/passwd", O_RDONLY, 0, sub {
      if ($_[0]) {
         print "open successful, fh is $_[0]\n";
         ...
      } else {
         die "open failed: $!\n";
      }
   };

=item aio_close $fh, $callback->($status)

Asynchronously close a file and call the callback with the result
code. I<WARNING:> although accepted, you should not pass in a perl
filehandle here, as perl will likely close the file descriptor another
time when the filehandle is destroyed. Normally, you can safely call perls
C<close> or just let filehandles go out of scope.

This is supposed to be a bug in the API, so that might change. It's
therefore best to avoid this function.

=item aio_read  $fh,$offset,$length, $data,$dataoffset, $callback->($retval)

=item aio_write $fh,$offset,$length, $data,$dataoffset, $callback->($retval)

Reads or writes C<length> bytes from the specified C<fh> and C<offset>
into the scalar given by C<data> and offset C<dataoffset> and calls the
callback without the actual number of bytes read (or -1 on error, just
like the syscall).

The C<$data> scalar I<MUST NOT> be modified in any way while the request
is outstanding. Modifying it can result in segfaults or WW3 (if the
necessary/optional hardware is installed).

Example: Read 15 bytes at offset 7 into scalar C<$buffer>, starting at
offset C<0> within the scalar:

   aio_read $fh, 7, 15, $buffer, 0, sub {
      $_[0] > 0 or die "read error: $!";
      print "read $_[0] bytes: <$buffer>\n";
   };

=item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback->($retval)

Tries to copy C<$length> bytes from C<$in_fh> to C<$out_fh>. It starts
reading at byte offset C<$in_offset>, and starts writing at the current
file offset of C<$out_fh>. Because of that, it is not safe to issue more
than one C<aio_sendfile> per C<$out_fh>, as they will interfere with each
other.

This call tries to make use of a native C<sendfile> syscall to provide
zero-copy operation. For this to work, C<$out_fh> should refer to a
socket, and C<$in_fh> should refer to mmap'able file.

If the native sendfile call fails or is not implemented, it will be
emulated, so you can call C<aio_sendfile> on any type of filehandle
regardless of the limitations of the operating system.

Please note, however, that C<aio_sendfile> can read more bytes from
C<$in_fh> than are written, and there is no way to find out how many
bytes have been read from C<aio_sendfile> alone, as C<aio_sendfile> only
provides the number of bytes written to C<$out_fh>. Only if the result
value equals C<$length> one can assume that C<$length> bytes have been
read.

=item aio_readahead $fh,$offset,$length, $callback->($retval)

C<aio_readahead> populates the page cache with data from a file so that
subsequent reads from that file will not block on disk I/O. The C<$offset>
argument specifies the starting point from which data is to be read and
C<$length> specifies the number of bytes to be read. I/O is performed in
whole pages, so that offset is effectively rounded down to a page boundary
and bytes are read up to the next page boundary greater than or equal to
(off-set+length). C<aio_readahead> does not read beyond the end of the
file. The current file offset of the file is left unchanged.

If that syscall doesn't exist (likely if your OS isn't Linux) it will be
emulated by simply reading the data, which would have a similar effect.

=item aio_stat  $fh_or_path, $callback->($status)

=item aio_lstat $fh, $callback->($status)

Works like perl's C<stat> or C<lstat> in void context. The callback will
be called after the stat and the results will be available using C<stat _>
or C<-s _> etc...

The pathname passed to C<aio_stat> must be absolute. See API NOTES, above,
for an explanation.

Currently, the stats are always 64-bit-stats, i.e. instead of returning an
error when stat'ing a large file, the results will be silently truncated
unless perl itself is compiled with large file support.

Example: Print the length of F</etc/passwd>:

   aio_stat "/etc/passwd", sub {
      $_[0] and die "stat failed: $!";
      print "size is ", -s _, "\n";
   };

=item aio_unlink $pathname, $callback->($status)

Asynchronously unlink (delete) a file and call the callback with the
result code.

=item aio_rmdir $pathname, $callback->($status)

Asynchronously rmdir (delete) a directory and call the callback with the
result code.

=item aio_readdir $pathname $callback->($entries)

Unlike the POSIX call of the same name, C<aio_readdir> reads an entire
directory (i.e. opendir + readdir + closedir). The entries will not be
sorted, and will B<NOT> include the C<.> and C<..> entries.

The callback a single argument which is either C<undef> or an array-ref
with the filenames.

=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)

Scans a directory (similar to C<aio_readdir>) and tries to separate the
entries of directory C<$path> into two sets of names, ones you can recurse
into (directories), and ones you cannot recurse into (everything else).

C<aio_scandir> is a composite request that consists of many
aio-primitives. C<$maxreq> specifies the maximum number of outstanding
aio requests that this function generates. If it is C<< <= 0 >>, then a
suitable default will be chosen (currently 8).

On error, the callback is called without arguments, otherwise it receives
two array-refs with path-relative entry names.

Example:

   aio_scandir $dir, 0, sub {
      my ($dirs, $nondirs) = @_;
      print "real directories: @$dirs\n";
      print "everything else: @$nondirs\n";
   };

Implementation notes.

The C<aio_readdir> cannot be avoided, but C<stat()>'ing every entry can.

After reading the directory, the modification time, size etc. of the
directory before and after the readdir is checked, and if they match, the
link count will be used to decide how many entries are directories (if
>= 2). Otherwise, no knowledge of the number of subdirectories will be
assumed.

Then entires will be sorted into likely directories (everything without a
non-initial dot) and likely non-directories (everything else).  Then every
entry + C</.> will be C<stat>'ed, likely directories first. This is often
faster because filesystems might detect the type of the entry without
reading the inode data (e.g. ext2s filetype feature). If that succeeds,
it assumes that the entry is a directory or a symlink to directory (which
will be checked seperately).

If the known number of directories has been reached, the rest of the
entries is assumed to be non-directories.

=cut

sub aio_scandir($$$) {
   my ($path, $maxreq, $cb) = @_;

   $maxreq = 8 if $maxreq <= 0;

   # stat once
   aio_stat $path, sub {
      $cb->() if $_[0];
      my $hash1 = join ":", (stat _)[0,1,3,7,9];

      # read the directory entries
      aio_readdir $path, sub {
         my $entries = shift
            or return $cb->();

         # stat the dir another time
         aio_stat $path, sub {
            my $hash2 = join ":", (stat _)[0,1,3,7,9];

            my $ndirs;

            # take the slow route if anything looks fishy
            if ($hash1 ne $hash2) {
               $ndirs = -1;
            } else {
               # if nlink == 2, we are finished
               # on non-posix-fs's, we rely on nlink < 2
               $ndirs = (stat _)[3] - 2
                  or $cb->([], $entries);
            }

            # sort into likely dirs and likely nondirs
            # dirs == files without ".", short entries first
            $entries = [map $_->[0],
                           sort { $b->[1] cmp $a->[1] }
                              map [$_, sprintf "%s%04d", (/.\./ ? "1" : "0"), length],
                                 @$entries];

            my (@dirs, @nondirs);

            my ($statcb, $schedcb);
            my $nreq = 0;

            $schedcb = sub {
               if (@$entries) {
                  if ($nreq < $maxreq) {
                     my $ent = pop @$entries;
                     $nreq++;
                     aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) };
                  }
               } elsif (!$nreq) {
                  # finished
                  undef $statcb;
                  undef $schedcb;
                  $cb->(\@dirs, \@nondirs);
                  undef $cb;
               }
            };
            $statcb = sub {
               my ($status, $entry) = @_;

               if ($status < 0) {
                  $nreq--;
                  push @nondirs, $entry;
                  &$schedcb;
               } else {
                  # need to check for real directory
                  aio_lstat "$path/$entry", sub {
                     $nreq--;

                     if (-d _) {
                        push @dirs, $entry;

                        if (!--$ndirs) {
                           push @nondirs, @$entries;
                           $entries = [];
                        }
                     } else {
                        push @nondirs, $entry;
                     }

                     &$schedcb;
                  }
               }
            };

            &$schedcb while @$entries && $nreq < $maxreq;
         };
      };
   };
}

=item aio_fsync $fh, $callback->($status)

Asynchronously call fsync on the given filehandle and call the callback
with the fsync result code.

=item aio_fdatasync $fh, $callback->($status)

Asynchronously call fdatasync on the given filehandle and call the
callback with the fdatasync result code.

If this call isn't available because your OS lacks it or it couldn't be
detected, it will be emulated by calling C<fsync> instead.

=back

=head2 SUPPORT FUNCTIONS

=over 4

=item $fileno = IO::AIO::poll_fileno

Return the I<request result pipe file descriptor>. This filehandle must be
polled for reading by some mechanism outside this module (e.g. Event or
select, see below or the SYNOPSIS). If the pipe becomes readable you have
to call C<poll_cb> to check the results.

See C<poll_cb> for an example.

=item IO::AIO::poll_cb

Process all outstanding events on the result pipe. You have to call this
regularly. Returns the number of events processed. Returns immediately
when no events are outstanding.

Example: Install an Event watcher that automatically calls
IO::AIO::poll_cb with high priority:

   Event->io (fd => IO::AIO::poll_fileno,
              poll => 'r', async => 1,
              cb => \&IO::AIO::poll_cb);

=item IO::AIO::poll_wait

Wait till the result filehandle becomes ready for reading (simply does a
C<select> on the filehandle. This is useful if you want to synchronously wait
for some requests to finish).

See C<nreqs> for an example.

=item IO::AIO::nreqs

Returns the number of requests currently outstanding (i.e. for which their
callback has not been invoked yet).

Example: wait till there are no outstanding requests anymore:

   IO::AIO::poll_wait, IO::AIO::poll_cb
      while IO::AIO::nreqs;

=item IO::AIO::flush

Wait till all outstanding AIO requests have been handled.

Strictly equivalent to:

   IO::AIO::poll_wait, IO::AIO::poll_cb
      while IO::AIO::nreqs;

=item IO::AIO::poll

Waits until some requests have been handled.

Strictly equivalent to:

   IO::AIO::poll_wait, IO::AIO::poll_cb
      if IO::AIO::nreqs;

=item IO::AIO::min_parallel $nthreads

Set the minimum number of AIO threads to C<$nthreads>. The current default
is C<4>, which means four asynchronous operations can be done at one time
(the number of outstanding operations, however, is unlimited).

IO::AIO starts threads only on demand, when an AIO request is queued and
no free thread exists.

It is recommended to keep the number of threads low, as some Linux
kernel versions will scale negatively with the number of threads (higher
parallelity => MUCH higher latency). With current Linux 2.6 versions, 4-32
threads should be fine.

Under most circumstances you don't need to call this function, as the
module selects a default that is suitable for low to moderate load.

=item IO::AIO::max_parallel $nthreads

Sets the maximum number of AIO threads to C<$nthreads>. If more than the
specified number of threads are currently running, this function kills
them. This function blocks until the limit is reached.

While C<$nthreads> are zero, aio requests get queued but not executed
until the number of threads has been increased again.

This module automatically runs C<max_parallel 0> at program end, to ensure
that all threads are killed and that there are no outstanding requests.

Under normal circumstances you don't need to call this function.

=item $oldnreqs = IO::AIO::max_outstanding $nreqs

Sets the maximum number of outstanding requests to C<$nreqs>. If you
try to queue up more than this number of requests, the caller will block until
some requests have been handled.

The default is very large, so normally there is no practical limit. If you
queue up many requests in a loop it often improves speed if you set
this to a relatively low number, such as C<100>.

Under normal circumstances you don't need to call this function.

=back

=cut

# support function to convert a fd into a perl filehandle
sub _fd2fh {
   return undef if $_[0] < 0;

   # try to generate nice filehandles
   my $sym = "IO::AIO::fd#$_[0]";
   local *$sym;

   open *$sym, "+<&=$_[0]"      # usually works under any unix
      or open *$sym, "<&=$_[0]" # cygwin needs this
      or open *$sym, ">&=$_[0]" # or this
      or return undef;

   *$sym
}

min_parallel 4;

END {
   max_parallel 0;
}

1;

=head2 FORK BEHAVIOUR

Before the fork, IO::AIO enters a quiescent state where no requests
can be added in other threads and no results will be processed. After
the fork the parent simply leaves the quiescent state and continues
request/result processing, while the child clears the request/result
queue (so the requests started before the fork will only be handled in
the parent). Threats will be started on demand until the limit ste in the
parent process has been reached again.

=head1 SEE ALSO

L<Coro>, L<Linux::AIO>.

=head1 AUTHOR

 Marc Lehmann <schmorp@schmorp.de>
 http://home.schmorp.de/

=cut

Revision:	1.41
Committed:	Wed Sep 7 17:41:17 2005 UTC (18 years, 8 months ago) by root
Branch:	MAIN
CVS Tags:	rel-1_61
Changes since 1.40:	+1 -1 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3			IO::AIO - Asynchronous Input/Output
4
5			=head1 SYNOPSIS
6
7			use IO::AIO;
8
9	root	1.6	aio_open "/etc/passwd", O_RDONLY, 0, sub {
10			my ($fh) = @_;
11			...
12			};
13
14			aio_unlink "/tmp/file", sub { };
15
16			aio_read $fh, 30000, 1024, $buffer, 0, sub {
17	root	1.8	$_[0] > 0 or die "read error: $!";
18	root	1.6	};
19
20			# Event
21			Event->io (fd => IO::AIO::poll_fileno,
22	root	1.7	poll => 'r',
23	root	1.6	cb => \&IO::AIO::poll_cb);
24
25			# Glib/Gtk2
26			add_watch Glib::IO IO::AIO::poll_fileno,
27	root	1.22	in => sub { IO::AIO::poll_cb; 1 };
28	root	1.6
29			# Tk
30			Tk::Event::IO->fileevent (IO::AIO::poll_fileno, "",
31			readable => \&IO::AIO::poll_cb);
32
33	root	1.11	# Danga::Socket
34			Danga::Socket->AddOtherFds (IO::AIO::poll_fileno =>
35			\&IO::AIO::poll_cb);
36
37
38	root	1.1	=head1 DESCRIPTION
39
40			This module implements asynchronous I/O using whatever means your
41	root	1.2	operating system supports.
42	root	1.1
43	root	1.2	Currently, a number of threads are started that execute your read/writes
44			and signal their completion. You don't need thread support in your libc or
45			perl, and the threads created by this module will not be visible to the
46			pthreads library. In the future, this module might make use of the native
47			aio functions available on many operating systems. However, they are often
48			not well-supported (Linux doesn't allow them on normal files currently,
49			for example), and they would only support aio_read and aio_write, so the
50			remaining functionality would have to be implemented using threads anyway.
51	root	1.1
52			Although the module will work with in the presence of other threads, it is
53	root	1.22	currently not reentrant, so use appropriate locking yourself, always call
54			C<poll_cb> from within the same thread, or never call C<poll_cb> (or other
55			C<aio_> functions) recursively.
56	root	1.1
57			=cut
58
59			package IO::AIO;
60
61	root	1.23	no warnings;
62
63	root	1.1	use base 'Exporter';
64
65	root	1.2	use Fcntl ();
66
67	root	1.1	BEGIN {
68	root	1.41	$VERSION = '1.61';
69	root	1.1
70	root	1.39	@EXPORT = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat
71	root	1.40	aio_lstat aio_unlink aio_rmdir aio_readdir aio_scandir aio_symlink
72	root	1.38	aio_fsync aio_fdatasync aio_readahead);
73			@EXPORT_OK = qw(poll_fileno poll_cb min_parallel max_parallel
74			max_outstanding nreqs);
75	root	1.1
76			require XSLoader;
77			XSLoader::load IO::AIO, $VERSION;
78			}
79
80	root	1.5	=head1 FUNCTIONS
81	root	1.1
82	root	1.5	=head2 AIO FUNCTIONS
83	root	1.1
84	root	1.5	All the C<aio_*> calls are more or less thin wrappers around the syscall
85			with the same name (sans C<aio_>). The arguments are similar or identical,
86	root	1.14	and they all accept an additional (and optional) C<$callback> argument
87			which must be a code reference. This code reference will get called with
88			the syscall return code (e.g. most syscalls return C<-1> on error, unlike
89			perl, which usually delivers "false") as it's sole argument when the given
90			syscall has been executed asynchronously.
91	root	1.1
92	root	1.23	All functions expecting a filehandle keep a copy of the filehandle
93			internally until the request has finished.
94	root	1.1
95	root	1.28	The pathnames you pass to these routines I<must> be absolute and
96			encoded in byte form. The reason for the former is that at the time the
97			request is being executed, the current working directory could have
98			changed. Alternatively, you can make sure that you never change the
99			current working directory.
100
101			To encode pathnames to byte form, either make sure you either: a)
102			always pass in filenames you got from outside (command line, readdir
103			etc.), b) are ASCII or ISO 8859-1, c) use the Encode module and encode
104			your pathnames to the locale (or other) encoding in effect in the user
105			environment, d) use Glib::filename_from_unicode on unicode filenames or e)
106			use something else.
107	root	1.1
108	root	1.5	=over 4
109	root	1.1
110	root	1.40	=item aio_open $pathname, $flags, $mode, $callback->($fh)
111	root	1.1
112	root	1.2	Asynchronously open or create a file and call the callback with a newly
113			created filehandle for the file.
114	root	1.1
115			The pathname passed to C<aio_open> must be absolute. See API NOTES, above,
116			for an explanation.
117
118	root	1.20	The C<$flags> argument is a bitmask. See the C<Fcntl> module for a
119			list. They are the same as used by C<sysopen>.
120
121			Likewise, C<$mode> specifies the mode of the newly created file, if it
122			didn't exist and C<O_CREAT> has been given, just like perl's C<sysopen>,
123			except that it is mandatory (i.e. use C<0> if you don't create new files,
124			and C<0666> or C<0777> if you do).
125	root	1.1
126			Example:
127
128			aio_open "/etc/passwd", O_RDONLY, 0, sub {
129	root	1.2	if ($_[0]) {
130			print "open successful, fh is $_[0]\n";
131	root	1.1	...
132			} else {
133			die "open failed: $!\n";
134			}
135			};
136
137	root	1.40	=item aio_close $fh, $callback->($status)
138	root	1.1
139	root	1.2	Asynchronously close a file and call the callback with the result
140			code. I<WARNING:> although accepted, you should not pass in a perl
141	root	1.20	filehandle here, as perl will likely close the file descriptor another
142			time when the filehandle is destroyed. Normally, you can safely call perls
143			C<close> or just let filehandles go out of scope.
144
145			This is supposed to be a bug in the API, so that might change. It's
146			therefore best to avoid this function.
147	root	1.1
148	root	1.40	=item aio_read $fh,$offset,$length, $data,$dataoffset, $callback->($retval)
149	root	1.1
150	root	1.40	=item aio_write $fh,$offset,$length, $data,$dataoffset, $callback->($retval)
151	root	1.1
152			Reads or writes C<length> bytes from the specified C<fh> and C<offset>
153			into the scalar given by C<data> and offset C<dataoffset> and calls the
154			callback without the actual number of bytes read (or -1 on error, just
155			like the syscall).
156
157	root	1.31	The C<$data> scalar I<MUST NOT> be modified in any way while the request
158			is outstanding. Modifying it can result in segfaults or WW3 (if the
159			necessary/optional hardware is installed).
160
161	root	1.17	Example: Read 15 bytes at offset 7 into scalar C<$buffer>, starting at
162	root	1.1	offset C<0> within the scalar:
163
164			aio_read $fh, 7, 15, $buffer, 0, sub {
165	root	1.9	$_[0] > 0 or die "read error: $!";
166			print "read $_[0] bytes: <$buffer>\n";
167	root	1.1	};
168
169	root	1.40	=item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback->($retval)
170	root	1.35
171			Tries to copy C<$length> bytes from C<$in_fh> to C<$out_fh>. It starts
172			reading at byte offset C<$in_offset>, and starts writing at the current
173			file offset of C<$out_fh>. Because of that, it is not safe to issue more
174			than one C<aio_sendfile> per C<$out_fh>, as they will interfere with each
175			other.
176
177			This call tries to make use of a native C<sendfile> syscall to provide
178			zero-copy operation. For this to work, C<$out_fh> should refer to a
179			socket, and C<$in_fh> should refer to mmap'able file.
180
181			If the native sendfile call fails or is not implemented, it will be
182	root	1.36	emulated, so you can call C<aio_sendfile> on any type of filehandle
183			regardless of the limitations of the operating system.
184	root	1.35
185			Please note, however, that C<aio_sendfile> can read more bytes from
186			C<$in_fh> than are written, and there is no way to find out how many
187	root	1.36	bytes have been read from C<aio_sendfile> alone, as C<aio_sendfile> only
188			provides the number of bytes written to C<$out_fh>. Only if the result
189			value equals C<$length> one can assume that C<$length> bytes have been
190			read.
191	root	1.35
192	root	1.40	=item aio_readahead $fh,$offset,$length, $callback->($retval)
193	root	1.1
194	root	1.20	C<aio_readahead> populates the page cache with data from a file so that
195	root	1.1	subsequent reads from that file will not block on disk I/O. The C<$offset>
196			argument specifies the starting point from which data is to be read and
197			C<$length> specifies the number of bytes to be read. I/O is performed in
198			whole pages, so that offset is effectively rounded down to a page boundary
199			and bytes are read up to the next page boundary greater than or equal to
200	root	1.20	(off-set+length). C<aio_readahead> does not read beyond the end of the
201	root	1.1	file. The current file offset of the file is left unchanged.
202
203	root	1.26	If that syscall doesn't exist (likely if your OS isn't Linux) it will be
204			emulated by simply reading the data, which would have a similar effect.
205
206	root	1.40	=item aio_stat $fh_or_path, $callback->($status)
207	root	1.1
208	root	1.40	=item aio_lstat $fh, $callback->($status)
209	root	1.1
210			Works like perl's C<stat> or C<lstat> in void context. The callback will
211			be called after the stat and the results will be available using C<stat _>
212			or C<-s _> etc...
213
214			The pathname passed to C<aio_stat> must be absolute. See API NOTES, above,
215			for an explanation.
216
217			Currently, the stats are always 64-bit-stats, i.e. instead of returning an
218			error when stat'ing a large file, the results will be silently truncated
219			unless perl itself is compiled with large file support.
220
221			Example: Print the length of F</etc/passwd>:
222
223			aio_stat "/etc/passwd", sub {
224			$_[0] and die "stat failed: $!";
225			print "size is ", -s _, "\n";
226			};
227
228	root	1.40	=item aio_unlink $pathname, $callback->($status)
229	root	1.1
230			Asynchronously unlink (delete) a file and call the callback with the
231			result code.
232
233	root	1.40	=item aio_rmdir $pathname, $callback->($status)
234	root	1.27
235			Asynchronously rmdir (delete) a directory and call the callback with the
236			result code.
237
238	root	1.40	=item aio_readdir $pathname $callback->($entries)
239	root	1.37
240			Unlike the POSIX call of the same name, C<aio_readdir> reads an entire
241			directory (i.e. opendir + readdir + closedir). The entries will not be
242			sorted, and will B<NOT> include the C<.> and C<..> entries.
243
244			The callback a single argument which is either C<undef> or an array-ref
245			with the filenames.
246
247	root	1.40	=item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)
248
249			Scans a directory (similar to C<aio_readdir>) and tries to separate the
250			entries of directory C<$path> into two sets of names, ones you can recurse
251			into (directories), and ones you cannot recurse into (everything else).
252
253			C<aio_scandir> is a composite request that consists of many
254			aio-primitives. C<$maxreq> specifies the maximum number of outstanding
255			aio requests that this function generates. If it is C<< <= 0 >>, then a
256			suitable default will be chosen (currently 8).
257
258			On error, the callback is called without arguments, otherwise it receives
259			two array-refs with path-relative entry names.
260
261			Example:
262
263			aio_scandir $dir, 0, sub {
264			my ($dirs, $nondirs) = @_;
265			print "real directories: @$dirs\n";
266			print "everything else: @$nondirs\n";
267			};
268
269			Implementation notes.
270
271			The C<aio_readdir> cannot be avoided, but C<stat()>'ing every entry can.
272
273			After reading the directory, the modification time, size etc. of the
274			directory before and after the readdir is checked, and if they match, the
275			link count will be used to decide how many entries are directories (if
276			>= 2). Otherwise, no knowledge of the number of subdirectories will be
277			assumed.
278
279			Then entires will be sorted into likely directories (everything without a
280			non-initial dot) and likely non-directories (everything else). Then every
281			entry + C</.> will be C<stat>'ed, likely directories first. This is often
282			faster because filesystems might detect the type of the entry without
283			reading the inode data (e.g. ext2s filetype feature). If that succeeds,
284			it assumes that the entry is a directory or a symlink to directory (which
285			will be checked seperately).
286
287			If the known number of directories has been reached, the rest of the
288			entries is assumed to be non-directories.
289
290			=cut
291
292			sub aio_scandir($$$) {
293			my ($path, $maxreq, $cb) = @_;
294
295			$maxreq = 8 if $maxreq <= 0;
296
297			# stat once
298			aio_stat $path, sub {
299			$cb->() if $_[0];
300			my $hash1 = join ":", (stat _)[0,1,3,7,9];
301
302			# read the directory entries
303			aio_readdir $path, sub {
304			my $entries = shift
305			or return $cb->();
306
307			# stat the dir another time
308			aio_stat $path, sub {
309			my $hash2 = join ":", (stat _)[0,1,3,7,9];
310
311			my $ndirs;
312
313			# take the slow route if anything looks fishy
314			if ($hash1 ne $hash2) {
315			$ndirs = -1;
316			} else {
317			# if nlink == 2, we are finished
318			# on non-posix-fs's, we rely on nlink < 2
319			$ndirs = (stat _)[3] - 2
320			or $cb->([], $entries);
321			}
322
323			# sort into likely dirs and likely nondirs
324			# dirs == files without ".", short entries first
325			$entries = [map $_->[0],
326			sort { $b->[1] cmp $a->[1] }
327			map [$_, sprintf "%s%04d", (/.\./ ? "1" : "0"), length],
328			@$entries];
329
330			my (@dirs, @nondirs);
331
332			my ($statcb, $schedcb);
333			my $nreq = 0;
334
335			$schedcb = sub {
336			if (@$entries) {
337			if ($nreq < $maxreq) {
338			my $ent = pop @$entries;
339			$nreq++;
340			aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) };
341			}
342			} elsif (!$nreq) {
343			# finished
344			undef $statcb;
345			undef $schedcb;
346			$cb->(\@dirs, \@nondirs);
347			undef $cb;
348			}
349			};
350			$statcb = sub {
351			my ($status, $entry) = @_;
352
353			if ($status < 0) {
354			$nreq--;
355			push @nondirs, $entry;
356			&$schedcb;
357			} else {
358			# need to check for real directory
359			aio_lstat "$path/$entry", sub {
360			$nreq--;
361
362			if (-d _) {
363			push @dirs, $entry;
364
365			if (!--$ndirs) {
366			push @nondirs, @$entries;
367			$entries = [];
368			}
369			} else {
370			push @nondirs, $entry;
371			}
372
373			&$schedcb;
374			}
375			}
376			};
377
378			&$schedcb while @$entries && $nreq < $maxreq;
379			};
380			};
381			};
382			}
383
384			=item aio_fsync $fh, $callback->($status)
385	root	1.1
386			Asynchronously call fsync on the given filehandle and call the callback
387			with the fsync result code.
388
389	root	1.40	=item aio_fdatasync $fh, $callback->($status)
390	root	1.1
391			Asynchronously call fdatasync on the given filehandle and call the
392	root	1.26	callback with the fdatasync result code.
393
394			If this call isn't available because your OS lacks it or it couldn't be
395			detected, it will be emulated by calling C<fsync> instead.
396	root	1.1
397	root	1.5	=back
398
399			=head2 SUPPORT FUNCTIONS
400
401			=over 4
402
403			=item $fileno = IO::AIO::poll_fileno
404
405	root	1.20	Return the I<request result pipe file descriptor>. This filehandle must be
406			polled for reading by some mechanism outside this module (e.g. Event or
407			select, see below or the SYNOPSIS). If the pipe becomes readable you have
408			to call C<poll_cb> to check the results.
409	root	1.5
410			See C<poll_cb> for an example.
411
412			=item IO::AIO::poll_cb
413
414			Process all outstanding events on the result pipe. You have to call this
415			regularly. Returns the number of events processed. Returns immediately
416			when no events are outstanding.
417
418	root	1.20	Example: Install an Event watcher that automatically calls
419			IO::AIO::poll_cb with high priority:
420	root	1.5
421			Event->io (fd => IO::AIO::poll_fileno,
422			poll => 'r', async => 1,
423			cb => \&IO::AIO::poll_cb);
424
425			=item IO::AIO::poll_wait
426
427			Wait till the result filehandle becomes ready for reading (simply does a
428	root	1.20	C<select> on the filehandle. This is useful if you want to synchronously wait
429	root	1.5	for some requests to finish).
430
431			See C<nreqs> for an example.
432
433			=item IO::AIO::nreqs
434
435	root	1.20	Returns the number of requests currently outstanding (i.e. for which their
436			callback has not been invoked yet).
437	root	1.5
438			Example: wait till there are no outstanding requests anymore:
439
440			IO::AIO::poll_wait, IO::AIO::poll_cb
441			while IO::AIO::nreqs;
442
443	root	1.12	=item IO::AIO::flush
444
445			Wait till all outstanding AIO requests have been handled.
446
447	root	1.13	Strictly equivalent to:
448
449			IO::AIO::poll_wait, IO::AIO::poll_cb
450			while IO::AIO::nreqs;
451
452			=item IO::AIO::poll
453
454			Waits until some requests have been handled.
455
456			Strictly equivalent to:
457
458			IO::AIO::poll_wait, IO::AIO::poll_cb
459			if IO::AIO::nreqs;
460
461	root	1.5	=item IO::AIO::min_parallel $nthreads
462
463	root	1.34	Set the minimum number of AIO threads to C<$nthreads>. The current default
464			is C<4>, which means four asynchronous operations can be done at one time
465	root	1.5	(the number of outstanding operations, however, is unlimited).
466
467	root	1.34	IO::AIO starts threads only on demand, when an AIO request is queued and
468			no free thread exists.
469
470	root	1.5	It is recommended to keep the number of threads low, as some Linux
471			kernel versions will scale negatively with the number of threads (higher
472			parallelity => MUCH higher latency). With current Linux 2.6 versions, 4-32
473			threads should be fine.
474
475	root	1.34	Under most circumstances you don't need to call this function, as the
476			module selects a default that is suitable for low to moderate load.
477	root	1.5
478			=item IO::AIO::max_parallel $nthreads
479
480	root	1.34	Sets the maximum number of AIO threads to C<$nthreads>. If more than the
481			specified number of threads are currently running, this function kills
482			them. This function blocks until the limit is reached.
483
484			While C<$nthreads> are zero, aio requests get queued but not executed
485			until the number of threads has been increased again.
486	root	1.5
487			This module automatically runs C<max_parallel 0> at program end, to ensure
488			that all threads are killed and that there are no outstanding requests.
489
490			Under normal circumstances you don't need to call this function.
491
492			=item $oldnreqs = IO::AIO::max_outstanding $nreqs
493
494			Sets the maximum number of outstanding requests to C<$nreqs>. If you
495			try to queue up more than this number of requests, the caller will block until
496			some requests have been handled.
497
498			The default is very large, so normally there is no practical limit. If you
499	root	1.34	queue up many requests in a loop it often improves speed if you set
500	root	1.5	this to a relatively low number, such as C<100>.
501
502			Under normal circumstances you don't need to call this function.
503
504			=back
505
506	root	1.1	=cut
507
508	root	1.2	# support function to convert a fd into a perl filehandle
509			sub _fd2fh {
510			return undef if $_[0] < 0;
511
512	root	1.23	# try to generate nice filehandles
513			my $sym = "IO::AIO::fd#$_[0]";
514			local *$sym;
515	root	1.25
516	root	1.27	open *$sym, "+<&=$_[0]" # usually works under any unix
517			or open *$sym, "<&=$_[0]" # cygwin needs this
518			or open *$sym, ">&=$_[0]" # or this
519	root	1.2	or return undef;
520
521	root	1.23	*$sym
522	root	1.2	}
523
524	root	1.1	min_parallel 4;
525
526			END {
527			max_parallel 0;
528			}
529
530			1;
531
532	root	1.27	=head2 FORK BEHAVIOUR
533
534	root	1.34	Before the fork, IO::AIO enters a quiescent state where no requests
535			can be added in other threads and no results will be processed. After
536			the fork the parent simply leaves the quiescent state and continues
537			request/result processing, while the child clears the request/result
538			queue (so the requests started before the fork will only be handled in
539			the parent). Threats will be started on demand until the limit ste in the
540			parent process has been reached again.
541	root	1.27
542	root	1.1	=head1 SEE ALSO
543
544			L<Coro>, L<Linux::AIO>.
545
546			=head1 AUTHOR
547
548			Marc Lehmann <schmorp@schmorp.de>
549			http://home.schmorp.de/
550
551			=cut
552