Coro/Event/Handle.pm

=head1 NAME

Coro::Handle - non-blocking io with a blocking interface.

=head1 SYNOPSIS

 use Coro::Handle;

=head1 DESCRIPTION

This module implements io-handles in a coroutine-compatible way, that is,
other coroutines can run while reads or writes block on the handle. It
does NOT inherit from IO::Handle but uses tied objects.

=over 4

=cut

package Coro::Handle;

no warnings qw(uninitialized);

use Errno ();
use base 'Exporter';

$VERSION = 0.45;

@EXPORT = qw(unblock);

=item $fh = new_from_fh Coro::Handle $fhandle [, arg => value...]

Create a new non-blocking io-handle using the given
perl-filehandle. Returns undef if no fhandle is given. The only other
supported argument is "timeout", which sets a timeout for each operation.

=cut

sub new_from_fh {
   my $class = shift;
   my $fh = shift or return;
   my $self = do { local *Coro::Handle };

   my ($package, $filename, $line) = caller;
   $filename =~ s/^.*[\/\\]//;

   tie $self, Coro::Handle::FH, fh => $fh, desc => "$filename:$line", @_;

   my $_fh = select bless \$self, $class; $| = 1; select $_fh;
}

=item $fh = unblock $fh

This is a convinience function that just calls C<new_from_fh> on the given
filehandle. Use it to replace a normal perl filehandle by a non-blocking
equivalent.

=cut

sub unblock($) {
   new_from_fh Coro::Handle $_[0];
}

=item $fh->writable, $fh->readable

Wait until the filehandle is readable or writable (and return true) or
until an error condition happens (and return false).

=cut

sub readable    { Coro::Handle::FH::readable(tied ${$_[0]}) }
sub writable    { Coro::Handle::FH::writable(tied ${$_[0]}) }

=item $fh->readline([$terminator])

Like the builtin of the same name, but allows you to specify the input
record separator in a coroutine-safe manner (i.e. not using a global
variable).

=cut

sub readline    { tied(${+shift})->READLINE(@_) }

=item $fh->autoflush([...])

Always returns true, arguments are being ignored (exists for compatibility
only). Might change in the future.

=cut

sub autoflush   { !0 }

=item $fh->fileno, $fh->close,
$fh->read, $fh->sysread, $fh->syswrite,
$fh->print, $fh->printf

Work like their function equivalents (except read, which works like
sysread. You should not use the read function with Coro::Handles, it will
work but it's not efficient).

=cut

sub read        { Coro::Handle::FH::READ  (tied ${$_[0]}, $_[1], $_[2], $_[3]) }
sub sysread     { Coro::Handle::FH::READ  (tied ${$_[0]}, $_[1], $_[2], $_[3]) }
sub syswrite    { Coro::Handle::FH::WRITE (tied ${$_[0]}, $_[1], $_[2], $_[3]) }
sub print       { Coro::Handle::FH::WRITE (tied ${+shift}, join "", @_) }
sub printf      { Coro::Handle::FH::PRINTF(tied ${+shift}, @_) }
sub fileno      { Coro::Handle::FH::FILENO(tied ${$_[0]}) }
sub close       { Coro::Handle::FH::CLOSE (tied ${$_[0]}) }

=item $fh->timeout([...])

The optional agrument sets the new timeout (in seconds) for this
handle. Returns the current (new) value.

C<0> is a valid timeout, use C<undef> to disable the timeout.

=cut

sub timeout {
   my $self = tied(${$_[0]});
   if (@_ > 1) {
      $self->[2] = $_[1];
      $self->[5]->timeout($_[1]) if $self->[5];
      $self->[6]->timeout($_[1]) if $self->[6];
   }
   $self->[2];
}

=item $fh->fh

Returns the "real" (non-blocking) filehandle. Use this if you want to
do operations on the file handle you cannot do using the Coro::Handle
interface.

=item $fh->rbuf

Returns the current contents of the raed buffer (this is an lvalue, so you
can change the read buffer if you like).

You can use this fucntion to implement your own optimized reader when neither
readline nor sysread are viable candidates, like this:

  # first get the _real_ non-blocking filehandle
  # and fetch the current contents of the read buffer
  my $nb_fh = $fh->fh;
  my $buf = $fh->rbuf;

  for(;;) {
     # now use buffer contents, modifying
     # if necessary to reflect the removed data

     last if $buf ne ""; # we have leftover data

     # read another buffer full of data
     $fh->readable or die "end of file";
     sysread $nb_fh, $buf, 8192;
  }

  # put the read buffer back
  $fh->rbuf = $buf;

=cut

sub fh {
   (tied ${$_[0]})->[0];
}

sub rbuf : lvalue {
   tied(${$_[0]})->[3];
}

package Coro::Handle::FH;

no warnings qw(uninitialized);

use Fcntl ();
use Errno ();
use Carp 'croak';

use Coro::Event;
use Event::Watcher qw(R W E);

# formerly a hash, but we are speed-critical, so try
# to be faster even if it hurts.
#
# 0 FH
# 1 desc
# 2 timeout
# 3 rb
# 4 wb # unused
# 5 rw
# 6 ww

sub TIEHANDLE {
   my $class = shift;
   my %args = @_;

   my $self = bless [], $class;
   $self->[0] = $args{fh};
   $self->[1] = $args{desc};
   $self->[2] = $args{timeout};
   $self->[3] = "";
   $self->[4] = "";

   fcntl $self->[0], &Fcntl::F_SETFL, &Fcntl::O_NONBLOCK
      or croak "fcntl(O_NONBLOCK): $!";

   $self;
}

sub cleanup {
   $_[0][3] = "";
   ($_[0][5])->cancel if defined $_[0][5]; $_[0][5] = undef;

   $_[0][4] = "";
   ($_[0][6])->cancel if defined $_[0][6]; $_[0][6] = undef;
}

sub OPEN {
   &cleanup;
   my $self = shift;
   my $r = @_ == 2 ? open $self->[0], $_[0], $_[1]
                   : open $self->[0], $_[0], $_[1], $_[2];
   if ($r) {
      fcntl $self->[0], &Fcntl::F_SETFL, &Fcntl::O_NONBLOCK
         or croak "fcntl(O_NONBLOCK): $!";
   }
   $r;
}

sub PRINT {
   WRITE($_[0], $_[1]);
}

sub PRINTF {
   WRITE(shift, sprintf(shift,@_));
}

sub GETC {
   my $buf;
   READ($_[0], $buf, 1);
   $buf;
}

sub BINMODE {
   binmode $_[0][0];
}

sub TELL {
   use Carp (); Carp::croak("Coro::Handle's don't support tell()");
}

sub SEEK {
   use Carp (); Carp::croak("Coro::Handle's don't support seek()");
}

sub EOF {
   use Carp (); Carp::croak("Coro::Handle's don't support eof()");
}

sub CLOSE {
   &cleanup;
   close $_[0][0];
}

sub DESTROY {
   &cleanup;
}

sub FILENO {
   fileno $_[0][0];
}

# seems to be called for stringification (how weird), at least
# when DumpValue::dumpValue is used to print this.
sub FETCH {
   "$_[0]<$_[0][1]>";
}

sub readable {
   ($_[0][5] ||= Coro::Event->io(
      fd      => $_[0][0],
      desc    => "$_[0][1] R",
      timeout => $_[0][2],
      poll    => R+E,
   ))->next->{Coro::Event}[5] & R;
}

sub writable {
   ($_[0][6] ||= Coro::Event->io(
      fd      => $_[0][0],
      desc    => "$_[0][1] W",
      timeout => $_[0][2],
      poll    => W+E,
   ))->next->{Coro::Event}[5] & W;
}

sub WRITE {
   my $len = defined $_[2] ? $_[2] : length $_[1];
   my $ofs = $_[3];
   my $res = 0;

   while() {
      my $r = syswrite $_[0][0], $_[1], $len, $ofs;
      if (defined $r) {
         $len -= $r;
         $ofs += $r;
         $res += $r;
         last unless $len;
      } elsif ($! != Errno::EAGAIN) {
         last;
      }
      last unless &writable;
   }

   return $res;
}

sub READ {
   my $len = $_[2];
   my $ofs = $_[3];
   my $res = 0;

   # first deplete the read buffer
   if (defined $_[0][3]) {
      my $l = length $_[0][3];
      if ($l <= $len) {
         substr($_[1], $ofs) = $_[0][3]; undef $_[0][3];
         $len -= $l;
         $res += $l;
         return $res unless $len;
      } else {
         substr($_[1], $ofs) = substr($_[0][3], 0, $len);
         substr($_[0][3], 0, $len) = "";
         return $len;
      }
   }

   while() {
      my $r = sysread $_[0][0], $_[1], $len, $ofs;
      if (defined $r) {
         $len -= $r;
         $ofs += $r;
         $res += $r;
         last unless $len && $r;
      } elsif ($! != Errno::EAGAIN) {
         last;
      }
      last unless &readable;
   }

   return $res;
}

sub READLINE {
   my $irs = @_ > 1 ? $_[1] : $/;

   while() {
      my $pos = index $_[0][3], $irs;
      if ($pos >= 0) {
         $pos += length $irs;
         my $res = substr $_[0][3], 0, $pos;
         substr ($_[0][3], 0, $pos) = "";
         return $res;
      }

      my $r = sysread $_[0][0], $_[0][3], 8192, length $_[0][3];
      if (defined $r) {
         return undef unless $r;
      } elsif ($! != Errno::EAGAIN || !&readable) {
         return undef;
      }
   }
}

1;

=head1 BUGS

 - Perl's IO-Handle model is THE bug.

=head1 AUTHOR

 Marc Lehmann <pcg@goof.com>
 http://www.goof.com/pcg/marc/

=cut

Revision:	1.11
Committed:	Mon Sep 24 02:05:44 2001 UTC (22 years, 8 months ago) by root
Branch:	MAIN
Changes since 1.10:	+27 -5 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3			Coro::Handle - non-blocking io with a blocking interface.
4
5			=head1 SYNOPSIS
6
7			use Coro::Handle;
8
9			=head1 DESCRIPTION
10
11			This module implements io-handles in a coroutine-compatible way, that is,
12			other coroutines can run while reads or writes block on the handle. It
13			does NOT inherit from IO::Handle but uses tied objects.
14
15			=over 4
16
17			=cut
18
19			package Coro::Handle;
20
21	root	1.10	no warnings qw(uninitialized);
22	root	1.9
23	root	1.1	use Errno ();
24			use base 'Exporter';
25
26			$VERSION = 0.45;
27
28			@EXPORT = qw(unblock);
29
30			=item $fh = new_from_fh Coro::Handle $fhandle [, arg => value...]
31
32			Create a new non-blocking io-handle using the given
33			perl-filehandle. Returns undef if no fhandle is given. The only other
34			supported argument is "timeout", which sets a timeout for each operation.
35
36			=cut
37
38			sub new_from_fh {
39			my $class = shift;
40			my $fh = shift or return;
41			my $self = do { local *Coro::Handle };
42
43			my ($package, $filename, $line) = caller;
44			$filename =~ s/^.*[\/\\]//;
45
46			tie $self, Coro::Handle::FH, fh => $fh, desc => "$filename:$line", @_;
47
48			my $_fh = select bless \$self, $class; $\| = 1; select $_fh;
49			}
50
51			=item $fh = unblock $fh
52
53			This is a convinience function that just calls C<new_from_fh> on the given
54			filehandle. Use it to replace a normal perl filehandle by a non-blocking
55			equivalent.
56
57			=cut
58
59			sub unblock($) {
60			new_from_fh Coro::Handle $_[0];
61			}
62
63			=item $fh->writable, $fh->readable
64
65			Wait until the filehandle is readable or writable (and return true) or
66			until an error condition happens (and return false).
67
68			=cut
69
70	root	1.5	sub readable { Coro::Handle::FH::readable(tied ${$_[0]}) }
71			sub writable { Coro::Handle::FH::writable(tied ${$_[0]}) }
72	root	1.1
73			=item $fh->readline([$terminator])
74
75			Like the builtin of the same name, but allows you to specify the input
76			record separator in a coroutine-safe manner (i.e. not using a global
77			variable).
78
79			=cut
80
81			sub readline { tied(${+shift})->READLINE(@_) }
82
83			=item $fh->autoflush([...])
84
85			Always returns true, arguments are being ignored (exists for compatibility
86			only). Might change in the future.
87
88			=cut
89
90			sub autoflush { !0 }
91
92	root	1.9	=item $fh->fileno, $fh->close,
93			$fh->read, $fh->sysread, $fh->syswrite,
94			$fh->print, $fh->printf
95
96			Work like their function equivalents (except read, which works like
97			sysread. You should not use the read function with Coro::Handles, it will
98			work but it's not efficient).
99	root	1.1
100			=cut
101
102	root	1.9	sub read { Coro::Handle::FH::READ (tied ${$_[0]}, $_[1], $_[2], $_[3]) }
103			sub sysread { Coro::Handle::FH::READ (tied ${$_[0]}, $_[1], $_[2], $_[3]) }
104			sub syswrite { Coro::Handle::FH::WRITE (tied ${$_[0]}, $_[1], $_[2], $_[3]) }
105			sub print { Coro::Handle::FH::WRITE (tied ${+shift}, join "", @_) }
106			sub printf { Coro::Handle::FH::PRINTF(tied ${+shift}, @_) }
107			sub fileno { Coro::Handle::FH::FILENO(tied ${$_[0]}) }
108			sub close { Coro::Handle::FH::CLOSE (tied ${$_[0]}) }
109	root	1.1
110			=item $fh->timeout([...])
111
112			The optional agrument sets the new timeout (in seconds) for this
113			handle. Returns the current (new) value.
114
115			C<0> is a valid timeout, use C<undef> to disable the timeout.
116
117			=cut
118
119			sub timeout {
120	root	1.4	my $self = tied(${$_[0]});
121	root	1.6	if (@_ > 1) {
122			$self->[2] = $_[1];
123			$self->[5]->timeout($_[1]) if $self->[5];
124			$self->[6]->timeout($_[1]) if $self->[6];
125	root	1.1	}
126	root	1.6	$self->[2];
127	root	1.4	}
128
129			=item $fh->fh
130
131			Returns the "real" (non-blocking) filehandle. Use this if you want to
132			do operations on the file handle you cannot do using the Coro::Handle
133			interface.
134
135	root	1.11	=item $fh->rbuf
136	root	1.10
137	root	1.11	Returns the current contents of the raed buffer (this is an lvalue, so you
138			can change the read buffer if you like).
139
140			You can use this fucntion to implement your own optimized reader when neither
141			readline nor sysread are viable candidates, like this:
142
143			# first get the _real_ non-blocking filehandle
144			# and fetch the current contents of the read buffer
145			my $nb_fh = $fh->fh;
146			my $buf = $fh->rbuf;
147
148			for(;;) {
149			# now use buffer contents, modifying
150			# if necessary to reflect the removed data
151
152			last if $buf ne ""; # we have leftover data
153
154			# read another buffer full of data
155			$fh->readable or die "end of file";
156			sysread $nb_fh, $buf, 8192;
157			}
158
159			# put the read buffer back
160			$fh->rbuf = $buf;
161	root	1.10
162	root	1.4	=cut
163
164			sub fh {
165	root	1.10	(tied ${$_[0]})->[0];
166			}
167
168	root	1.11	sub rbuf : lvalue {
169			tied(${$_[0]})->[3];
170	root	1.1	}
171
172			package Coro::Handle::FH;
173
174	root	1.10	no warnings qw(uninitialized);
175	root	1.9
176	root	1.1	use Fcntl ();
177			use Errno ();
178			use Carp 'croak';
179
180			use Coro::Event;
181			use Event::Watcher qw(R W E);
182
183	root	1.6	# formerly a hash, but we are speed-critical, so try
184			# to be faster even if it hurts.
185			#
186			# 0 FH
187			# 1 desc
188			# 2 timeout
189			# 3 rb
190	root	1.10	# 4 wb # unused
191	root	1.6	# 5 rw
192			# 6 ww
193
194	root	1.1	sub TIEHANDLE {
195			my $class = shift;
196	root	1.6	my %args = @_;
197	root	1.1
198	root	1.6	my $self = bless [], $class;
199			$self->[0] = $args{fh};
200			$self->[1] = $args{desc};
201			$self->[2] = $args{timeout};
202			$self->[3] = "";
203			$self->[4] = "";
204	root	1.1
205	root	1.6	fcntl $self->[0], &Fcntl::F_SETFL, &Fcntl::O_NONBLOCK
206	root	1.1	or croak "fcntl(O_NONBLOCK): $!";
207
208			$self;
209			}
210
211	root	1.6	sub cleanup {
212			$_[0][3] = "";
213	root	1.7	($_[0][5])->cancel if defined $_[0][5]; $_[0][5] = undef;
214	root	1.6
215			$_[0][4] = "";
216	root	1.7	($_[0][6])->cancel if defined $_[0][6]; $_[0][6] = undef;
217	root	1.6	}
218
219	root	1.1	sub OPEN {
220	root	1.6	&cleanup;
221	root	1.1	my $self = shift;
222	root	1.6	my $r = @_ == 2 ? open $self->[0], $_[0], $_[1]
223			: open $self->[0], $_[0], $_[1], $_[2];
224	root	1.1	if ($r) {
225	root	1.6	fcntl $self->[0], &Fcntl::F_SETFL, &Fcntl::O_NONBLOCK
226	root	1.1	or croak "fcntl(O_NONBLOCK): $!";
227			}
228			$r;
229	root	1.9	}
230
231			sub PRINT {
232			WRITE($_[0], $_[1]);
233			}
234
235			sub PRINTF {
236			WRITE(shift, sprintf(shift,@_));
237			}
238
239			sub GETC {
240			my $buf;
241			READ($_[0], $buf, 1);
242			$buf;
243			}
244
245			sub BINMODE {
246			binmode $_[0][0];
247			}
248
249			sub TELL {
250			use Carp (); Carp::croak("Coro::Handle's don't support tell()");
251			}
252
253			sub SEEK {
254			use Carp (); Carp::croak("Coro::Handle's don't support seek()");
255			}
256
257			sub EOF {
258			use Carp (); Carp::croak("Coro::Handle's don't support eof()");
259	root	1.1	}
260
261			sub CLOSE {
262	root	1.6	&cleanup;
263			close $_[0][0];
264	root	1.1	}
265
266	root	1.6	sub DESTROY {
267			&cleanup;
268	root	1.1	}
269
270	root	1.6	sub FILENO {
271			fileno $_[0][0];
272	root	1.1	}
273
274	root	1.8	# seems to be called for stringification (how weird), at least
275			# when DumpValue::dumpValue is used to print this.
276			sub FETCH {
277			"$_[0]<$_[0][1]>";
278			}
279
280	root	1.1	sub readable {
281	root	1.6	($_[0][5] \|\|= Coro::Event->io(
282			fd => $_[0][0],
283			desc => "$_[0][1] R",
284			timeout => $_[0][2],
285	root	1.1	poll => R+E,
286	root	1.3	))->next->{Coro::Event}[5] & R;
287	root	1.1	}
288
289	root	1.6	sub writable {
290			($_[0][6] \|\|= Coro::Event->io(
291			fd => $_[0][0],
292			desc => "$_[0][1] W",
293			timeout => $_[0][2],
294			poll => W+E,
295			))->next->{Coro::Event}[5] & W;
296			}
297
298	root	1.1	sub WRITE {
299			my $len = defined $_[2] ? $_[2] : length $_[1];
300			my $ofs = $_[3];
301			my $res = 0;
302
303			while() {
304	root	1.6	my $r = syswrite $_[0][0], $_[1], $len, $ofs;
305	root	1.1	if (defined $r) {
306			$len -= $r;
307			$ofs += $r;
308			$res += $r;
309			last unless $len;
310			} elsif ($! != Errno::EAGAIN) {
311			last;
312			}
313	root	1.6	last unless &writable;
314	root	1.1	}
315
316			return $res;
317			}
318
319			sub READ {
320			my $len = $_[2];
321			my $ofs = $_[3];
322			my $res = 0;
323
324			# first deplete the read buffer
325	root	1.6	if (defined $_[0][3]) {
326			my $l = length $_[0][3];
327	root	1.1	if ($l <= $len) {
328	root	1.6	substr($_[1], $ofs) = $_[0][3]; undef $_[0][3];
329	root	1.1	$len -= $l;
330			$res += $l;
331			return $res unless $len;
332			} else {
333	root	1.6	substr($_[1], $ofs) = substr($_[0][3], 0, $len);
334			substr($_[0][3], 0, $len) = "";
335	root	1.1	return $len;
336			}
337			}
338
339			while() {
340	root	1.6	my $r = sysread $_[0][0], $_[1], $len, $ofs;
341	root	1.1	if (defined $r) {
342			$len -= $r;
343			$ofs += $r;
344			$res += $r;
345			last unless $len && $r;
346			} elsif ($! != Errno::EAGAIN) {
347			last;
348			}
349	root	1.6	last unless &readable;
350	root	1.1	}
351
352			return $res;
353			}
354
355			sub READLINE {
356	root	1.6	my $irs = @_ > 1 ? $_[1] : $/;
357	root	1.1
358			while() {
359	root	1.6	my $pos = index $_[0][3], $irs;
360	root	1.1	if ($pos >= 0) {
361			$pos += length $irs;
362	root	1.6	my $res = substr $_[0][3], 0, $pos;
363			substr ($_[0][3], 0, $pos) = "";
364	root	1.1	return $res;
365			}
366	root	1.6
367			my $r = sysread $_[0][0], $_[0][3], 8192, length $_[0][3];
368	root	1.1	if (defined $r) {
369			return undef unless $r;
370	root	1.6	} elsif ($! != Errno::EAGAIN \|\| !&readable) {
371	root	1.1	return undef;
372			}
373			}
374			}
375
376			1;
377
378			=head1 BUGS
379
380			- Perl's IO-Handle model is THE bug.
381
382			=head1 AUTHOR
383
384			Marc Lehmann <pcg@goof.com>
385			http://www.goof.com/pcg/marc/
386
387			=cut
388