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->unsysread($data)

Pushes the given data into the input buffer. The following calls to
read/sysread will first return this data.

=cut

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

sub unsysread {
   substr tied(${$_[0]})->[3], 0, 0, $_[1];
}

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.10
Committed:	Mon Sep 24 01:36:21 2001 UTC (22 years, 8 months ago) by root
Branch:	MAIN
Changes since 1.9:	+13 -4 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->unsysread($data)
136
137	Pushes the given data into the input buffer. The following calls to
138	read/sysread will first return this data.
139
140	=cut
141
142	sub fh {
143	(tied ${$_[0]})->[0];
144	}
145
146	sub unsysread {
147	substr tied(${$_[0]})->[3], 0, 0, $_[1];
148	}
149
150	package Coro::Handle::FH;
151
152	no warnings qw(uninitialized);
153
154	use Fcntl ();
155	use Errno ();
156	use Carp 'croak';
157
158	use Coro::Event;
159	use Event::Watcher qw(R W E);
160
161	# formerly a hash, but we are speed-critical, so try
162	# to be faster even if it hurts.
163	#
164	# 0 FH
165	# 1 desc
166	# 2 timeout
167	# 3 rb
168	# 4 wb # unused
169	# 5 rw
170	# 6 ww
171
172	sub TIEHANDLE {
173	my $class = shift;
174	my %args = @_;
175
176	my $self = bless [], $class;
177	$self->[0] = $args{fh};
178	$self->[1] = $args{desc};
179	$self->[2] = $args{timeout};
180	$self->[3] = "";
181	$self->[4] = "";
182
183	fcntl $self->[0], &Fcntl::F_SETFL, &Fcntl::O_NONBLOCK
184	or croak "fcntl(O_NONBLOCK): $!";
185
186	$self;
187	}
188
189	sub cleanup {
190	$_[0][3] = "";
191	($_[0][5])->cancel if defined $_[0][5]; $_[0][5] = undef;
192
193	$_[0][4] = "";
194	($_[0][6])->cancel if defined $_[0][6]; $_[0][6] = undef;
195	}
196
197	sub OPEN {
198	&cleanup;
199	my $self = shift;
200	my $r = @_ == 2 ? open $self->[0], $_[0], $_[1]
201	: open $self->[0], $_[0], $_[1], $_[2];
202	if ($r) {
203	fcntl $self->[0], &Fcntl::F_SETFL, &Fcntl::O_NONBLOCK
204	or croak "fcntl(O_NONBLOCK): $!";
205	}
206	$r;
207	}
208
209	sub PRINT {
210	WRITE($_[0], $_[1]);
211	}
212
213	sub PRINTF {
214	WRITE(shift, sprintf(shift,@_));
215	}
216
217	sub GETC {
218	my $buf;
219	READ($_[0], $buf, 1);
220	$buf;
221	}
222
223	sub BINMODE {
224	binmode $_[0][0];
225	}
226
227	sub TELL {
228	use Carp (); Carp::croak("Coro::Handle's don't support tell()");
229	}
230
231	sub SEEK {
232	use Carp (); Carp::croak("Coro::Handle's don't support seek()");
233	}
234
235	sub EOF {
236	use Carp (); Carp::croak("Coro::Handle's don't support eof()");
237	}
238
239	sub CLOSE {
240	&cleanup;
241	close $_[0][0];
242	}
243
244	sub DESTROY {
245	&cleanup;
246	}
247
248	sub FILENO {
249	fileno $_[0][0];
250	}
251
252	# seems to be called for stringification (how weird), at least
253	# when DumpValue::dumpValue is used to print this.
254	sub FETCH {
255	"$_[0]<$_[0][1]>";
256	}
257
258	sub readable {
259	($_[0][5] \|\|= Coro::Event->io(
260	fd => $_[0][0],
261	desc => "$_[0][1] R",
262	timeout => $_[0][2],
263	poll => R+E,
264	))->next->{Coro::Event}[5] & R;
265	}
266
267	sub writable {
268	($_[0][6] \|\|= Coro::Event->io(
269	fd => $_[0][0],
270	desc => "$_[0][1] W",
271	timeout => $_[0][2],
272	poll => W+E,
273	))->next->{Coro::Event}[5] & W;
274	}
275
276	sub WRITE {
277	my $len = defined $_[2] ? $_[2] : length $_[1];
278	my $ofs = $_[3];
279	my $res = 0;
280
281	while() {
282	my $r = syswrite $_[0][0], $_[1], $len, $ofs;
283	if (defined $r) {
284	$len -= $r;
285	$ofs += $r;
286	$res += $r;
287	last unless $len;
288	} elsif ($! != Errno::EAGAIN) {
289	last;
290	}
291	last unless &writable;
292	}
293
294	return $res;
295	}
296
297	sub READ {
298	my $len = $_[2];
299	my $ofs = $_[3];
300	my $res = 0;
301
302	# first deplete the read buffer
303	if (defined $_[0][3]) {
304	my $l = length $_[0][3];
305	if ($l <= $len) {
306	substr($_[1], $ofs) = $_[0][3]; undef $_[0][3];
307	$len -= $l;
308	$res += $l;
309	return $res unless $len;
310	} else {
311	substr($_[1], $ofs) = substr($_[0][3], 0, $len);
312	substr($_[0][3], 0, $len) = "";
313	return $len;
314	}
315	}
316
317	while() {
318	my $r = sysread $_[0][0], $_[1], $len, $ofs;
319	if (defined $r) {
320	$len -= $r;
321	$ofs += $r;
322	$res += $r;
323	last unless $len && $r;
324	} elsif ($! != Errno::EAGAIN) {
325	last;
326	}
327	last unless &readable;
328	}
329
330	return $res;
331	}
332
333	sub READLINE {
334	my $irs = @_ > 1 ? $_[1] : $/;
335
336	while() {
337	my $pos = index $_[0][3], $irs;
338	if ($pos >= 0) {
339	$pos += length $irs;
340	my $res = substr $_[0][3], 0, $pos;
341	substr ($_[0][3], 0, $pos) = "";
342	return $res;
343	}
344
345	my $r = sysread $_[0][0], $_[0][3], 8192, length $_[0][3];
346	if (defined $r) {
347	return undef unless $r;
348	} elsif ($! != Errno::EAGAIN \|\| !&readable) {
349	return undef;
350	}
351	}
352	}
353
354	1;
355
356	=head1 BUGS
357
358	- Perl's IO-Handle model is THE bug.
359
360	=head1 AUTHOR
361
362	Marc Lehmann <pcg@goof.com>
363	http://www.goof.com/pcg/marc/
364
365	=cut
366