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 read buffer (this is an lvalue, so you
can change the read buffer if you like).

You can use this function 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(shift, join "", @_);
}

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.13
Committed:	Tue Oct 9 23:51:39 2001 UTC (22 years, 8 months ago) by root
Branch:	MAIN
Changes since 1.12:	+2 -2 lines
Log Message:	* empty log message *
#	Content
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	no warnings qw(uninitialized);
22
23	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	sub readable { Coro::Handle::FH::readable(tied ${$_[0]}) }
71	sub writable { Coro::Handle::FH::writable(tied ${$_[0]}) }
72
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	=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
100	=cut
101
102	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
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	my $self = tied(${$_[0]});
121	if (@_ > 1) {
122	$self->[2] = $_[1];
123	$self->[5]->timeout($_[1]) if $self->[5];
124	$self->[6]->timeout($_[1]) if $self->[6];
125	}
126	$self->[2];
127	}
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	=item $fh->rbuf
136
137	Returns the current contents of the read buffer (this is an lvalue, so you
138	can change the read buffer if you like).
139
140	You can use this function 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
162	=cut
163
164	sub fh {
165	(tied ${$_[0]})->[0];
166	}
167
168	sub rbuf : lvalue {
169	tied(${$_[0]})->[3];
170	}
171
172	package Coro::Handle::FH;
173
174	no warnings qw(uninitialized);
175
176	use Fcntl ();
177	use Errno ();
178	use Carp 'croak';
179
180	use Coro::Event;
181	use Event::Watcher qw(R W E);
182
183	# 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	# 4 wb # unused
191	# 5 rw
192	# 6 ww
193
194	sub TIEHANDLE {
195	my $class = shift;
196	my %args = @_;
197
198	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
205	fcntl $self->[0], &Fcntl::F_SETFL, &Fcntl::O_NONBLOCK
206	or croak "fcntl(O_NONBLOCK): $!";
207
208	$self;
209	}
210
211	sub cleanup {
212	$_[0][3] = "";
213	($_[0][5])->cancel if defined $_[0][5]; $_[0][5] = undef;
214
215	$_[0][4] = "";
216	($_[0][6])->cancel if defined $_[0][6]; $_[0][6] = undef;
217	}
218
219	sub OPEN {
220	&cleanup;
221	my $self = shift;
222	my $r = @_ == 2 ? open $self->[0], $_[0], $_[1]
223	: open $self->[0], $_[0], $_[1], $_[2];
224	if ($r) {
225	fcntl $self->[0], &Fcntl::F_SETFL, &Fcntl::O_NONBLOCK
226	or croak "fcntl(O_NONBLOCK): $!";
227	}
228	$r;
229	}
230
231	sub PRINT {
232	WRITE(shift, join "", @_);
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	}
260
261	sub CLOSE {
262	&cleanup;
263	close $_[0][0];
264	}
265
266	sub DESTROY {
267	&cleanup;
268	}
269
270	sub FILENO {
271	fileno $_[0][0];
272	}
273
274	# 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	sub readable {
281	($_[0][5] \|\|= Coro::Event->io(
282	fd => $_[0][0],
283	desc => "$_[0][1] R",
284	timeout => $_[0][2],
285	poll => R+E,
286	))->next->{Coro::Event}[5] & R;
287	}
288
289	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	sub WRITE {
299	my $len = defined $_[2] ? $_[2] : length $_[1];
300	my $ofs = $_[3];
301	my $res = 0;
302
303	while() {
304	my $r = syswrite $_[0][0], $_[1], $len, $ofs;
305	if (defined $r) {
306	$len -= $r;
307	$ofs += $r;
308	$res += $r;
309	last unless $len;
310	} elsif ($! != Errno::EAGAIN) {
311	last;
312	}
313	last unless &writable;
314	}
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	if (defined $_[0][3]) {
326	my $l = length $_[0][3];
327	if ($l <= $len) {
328	substr($_[1], $ofs) = $_[0][3]; undef $_[0][3];
329	$len -= $l;
330	$res += $l;
331	return $res unless $len;
332	} else {
333	substr($_[1], $ofs) = substr($_[0][3], 0, $len);
334	substr($_[0][3], 0, $len) = "";
335	return $len;
336	}
337	}
338
339	while() {
340	my $r = sysread $_[0][0], $_[1], $len, $ofs;
341	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	last unless &readable;
350	}
351
352	return $res;
353	}
354
355	sub READLINE {
356	my $irs = @_ > 1 ? $_[1] : $/;
357
358	while() {
359	my $pos = index $_[0][3], $irs;
360	if ($pos >= 0) {
361	$pos += length $irs;
362	my $res = substr $_[0][3], 0, $pos;
363	substr ($_[0][3], 0, $pos) = "";
364	return $res;
365	}
366
367	my $r = sysread $_[0][0], $_[0][3], 8192, length $_[0][3];
368	if (defined $r) {
369	return undef unless $r;
370	} elsif ($! != Errno::EAGAIN \|\| !&readable) {
371	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