ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/IO-AIO/AIO.pm
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 ***

File Contents

# Content
1 =head1 NAME
2
3 IO::AIO - Asynchronous Input/Output
4
5 =head1 SYNOPSIS
6
7 use IO::AIO;
8
9 aio_open "/etc/passwd", O_RDONLY, 0, sub {
10 my ($fh) = @_;
11 ...
12 };
13
14 aio_unlink "/tmp/file", sub { };
15
16 aio_read $fh, 30000, 1024, $buffer, 0, sub {
17 $_[0] > 0 or die "read error: $!";
18 };
19
20 # Event
21 Event->io (fd => IO::AIO::poll_fileno,
22 poll => 'r',
23 cb => \&IO::AIO::poll_cb);
24
25 # Glib/Gtk2
26 add_watch Glib::IO IO::AIO::poll_fileno,
27 in => sub { IO::AIO::poll_cb; 1 };
28
29 # Tk
30 Tk::Event::IO->fileevent (IO::AIO::poll_fileno, "",
31 readable => \&IO::AIO::poll_cb);
32
33 # Danga::Socket
34 Danga::Socket->AddOtherFds (IO::AIO::poll_fileno =>
35 \&IO::AIO::poll_cb);
36
37
38 =head1 DESCRIPTION
39
40 This module implements asynchronous I/O using whatever means your
41 operating system supports.
42
43 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
52 Although the module will work with in the presence of other threads, it is
53 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
57 =cut
58
59 package IO::AIO;
60
61 no warnings;
62
63 use base 'Exporter';
64
65 use Fcntl ();
66
67 BEGIN {
68 $VERSION = '1.61';
69
70 @EXPORT = qw(aio_sendfile aio_read aio_write aio_open aio_close aio_stat
71 aio_lstat aio_unlink aio_rmdir aio_readdir aio_scandir aio_symlink
72 aio_fsync aio_fdatasync aio_readahead);
73 @EXPORT_OK = qw(poll_fileno poll_cb min_parallel max_parallel
74 max_outstanding nreqs);
75
76 require XSLoader;
77 XSLoader::load IO::AIO, $VERSION;
78 }
79
80 =head1 FUNCTIONS
81
82 =head2 AIO FUNCTIONS
83
84 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 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
92 All functions expecting a filehandle keep a copy of the filehandle
93 internally until the request has finished.
94
95 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
108 =over 4
109
110 =item aio_open $pathname, $flags, $mode, $callback->($fh)
111
112 Asynchronously open or create a file and call the callback with a newly
113 created filehandle for the file.
114
115 The pathname passed to C<aio_open> must be absolute. See API NOTES, above,
116 for an explanation.
117
118 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
126 Example:
127
128 aio_open "/etc/passwd", O_RDONLY, 0, sub {
129 if ($_[0]) {
130 print "open successful, fh is $_[0]\n";
131 ...
132 } else {
133 die "open failed: $!\n";
134 }
135 };
136
137 =item aio_close $fh, $callback->($status)
138
139 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 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
148 =item aio_read $fh,$offset,$length, $data,$dataoffset, $callback->($retval)
149
150 =item aio_write $fh,$offset,$length, $data,$dataoffset, $callback->($retval)
151
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 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 Example: Read 15 bytes at offset 7 into scalar C<$buffer>, starting at
162 offset C<0> within the scalar:
163
164 aio_read $fh, 7, 15, $buffer, 0, sub {
165 $_[0] > 0 or die "read error: $!";
166 print "read $_[0] bytes: <$buffer>\n";
167 };
168
169 =item aio_sendfile $out_fh, $in_fh, $in_offset, $length, $callback->($retval)
170
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 emulated, so you can call C<aio_sendfile> on any type of filehandle
183 regardless of the limitations of the operating system.
184
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 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
192 =item aio_readahead $fh,$offset,$length, $callback->($retval)
193
194 C<aio_readahead> populates the page cache with data from a file so that
195 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 (off-set+length). C<aio_readahead> does not read beyond the end of the
201 file. The current file offset of the file is left unchanged.
202
203 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 =item aio_stat $fh_or_path, $callback->($status)
207
208 =item aio_lstat $fh, $callback->($status)
209
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 =item aio_unlink $pathname, $callback->($status)
229
230 Asynchronously unlink (delete) a file and call the callback with the
231 result code.
232
233 =item aio_rmdir $pathname, $callback->($status)
234
235 Asynchronously rmdir (delete) a directory and call the callback with the
236 result code.
237
238 =item aio_readdir $pathname $callback->($entries)
239
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 =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
386 Asynchronously call fsync on the given filehandle and call the callback
387 with the fsync result code.
388
389 =item aio_fdatasync $fh, $callback->($status)
390
391 Asynchronously call fdatasync on the given filehandle and call the
392 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
397 =back
398
399 =head2 SUPPORT FUNCTIONS
400
401 =over 4
402
403 =item $fileno = IO::AIO::poll_fileno
404
405 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
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 Example: Install an Event watcher that automatically calls
419 IO::AIO::poll_cb with high priority:
420
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 C<select> on the filehandle. This is useful if you want to synchronously wait
429 for some requests to finish).
430
431 See C<nreqs> for an example.
432
433 =item IO::AIO::nreqs
434
435 Returns the number of requests currently outstanding (i.e. for which their
436 callback has not been invoked yet).
437
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 =item IO::AIO::flush
444
445 Wait till all outstanding AIO requests have been handled.
446
447 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 =item IO::AIO::min_parallel $nthreads
462
463 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 (the number of outstanding operations, however, is unlimited).
466
467 IO::AIO starts threads only on demand, when an AIO request is queued and
468 no free thread exists.
469
470 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 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
478 =item IO::AIO::max_parallel $nthreads
479
480 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
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 queue up many requests in a loop it often improves speed if you set
500 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 =cut
507
508 # support function to convert a fd into a perl filehandle
509 sub _fd2fh {
510 return undef if $_[0] < 0;
511
512 # try to generate nice filehandles
513 my $sym = "IO::AIO::fd#$_[0]";
514 local *$sym;
515
516 open *$sym, "+<&=$_[0]" # usually works under any unix
517 or open *$sym, "<&=$_[0]" # cygwin needs this
518 or open *$sym, ">&=$_[0]" # or this
519 or return undef;
520
521 *$sym
522 }
523
524 min_parallel 4;
525
526 END {
527 max_parallel 0;
528 }
529
530 1;
531
532 =head2 FORK BEHAVIOUR
533
534 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
542 =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