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;

BEGIN { eval { require warnings } && warnings->unimport ("uninitialized") }

use Errno ();
use base 'Exporter';

$VERSION = 1.5;

@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, ref $class ? ref $class : $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 a reference to 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;
  }

=cut

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

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

package Coro::Handle::FH;

BEGIN { eval { require warnings } && warnings->unimport ("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 (length $_[0][3]) {
      my $l = length $_[0][3];
      if ($l <= $len) {
         substr($_[1], $ofs) = $_[0][3]; $_[0][3] = "";
         $len -= $l;
         $ofs += $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 <schmorp@schmorp.de>
 http://home.schmorp.de/

=cut

Revision:	1.34
Committed:	Tue Nov 29 12:36:18 2005 UTC (19 years, 1 month ago) by root
Branch:	MAIN
CVS Tags:	rel-1_5
Changes since 1.33:	+1 -1 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	BEGIN { eval { require warnings } && warnings->unimport ("uninitialized") }
22
23	use Errno ();
24	use base 'Exporter';
25
26	$VERSION = 1.5;
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, ref $class ? ref $class : $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 a reference to 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	=cut
160
161	sub fh {
162	(tied ${$_[0]})->[0];
163	}
164
165	sub rbuf : lvalue {
166	tied(${$_[0]})->[3];
167	}
168
169	package Coro::Handle::FH;
170
171	BEGIN { eval { require warnings } && warnings->unimport ("uninitialized") }
172
173	use Fcntl ();
174	use Errno ();
175	use Carp 'croak';
176
177	use Coro::Event;
178	use Event::Watcher qw(R W E);
179
180	# formerly a hash, but we are speed-critical, so try
181	# to be faster even if it hurts.
182	#
183	# 0 FH
184	# 1 desc
185	# 2 timeout
186	# 3 rb
187	# 4 wb # unused
188	# 5 rw
189	# 6 ww
190
191	sub TIEHANDLE {
192	my $class = shift;
193	my %args = @_;
194
195	my $self = bless [], $class;
196	$self->[0] = $args{fh};
197	$self->[1] = $args{desc};
198	$self->[2] = $args{timeout};
199	$self->[3] = "";
200	$self->[4] = "";
201
202	fcntl $self->[0], &Fcntl::F_SETFL, &Fcntl::O_NONBLOCK
203	or croak "fcntl(O_NONBLOCK): $!";
204
205	$self;
206	}
207
208	sub cleanup {
209	$_[0][3] = "";
210	($_[0][5])->cancel if defined $_[0][5]; $_[0][5] = undef;
211
212	$_[0][4] = "";
213	($_[0][6])->cancel if defined $_[0][6]; $_[0][6] = undef;
214	}
215
216	sub OPEN {
217	&cleanup;
218	my $self = shift;
219	my $r = @_ == 2 ? open $self->[0], $_[0], $_[1]
220	: open $self->[0], $_[0], $_[1], $_[2];
221	if ($r) {
222	fcntl $self->[0], &Fcntl::F_SETFL, &Fcntl::O_NONBLOCK
223	or croak "fcntl(O_NONBLOCK): $!";
224	}
225	$r;
226	}
227
228	sub PRINT {
229	WRITE(shift, join "", @_);
230	}
231
232	sub PRINTF {
233	WRITE(shift, sprintf(shift,@_));
234	}
235
236	sub GETC {
237	my $buf;
238	READ($_[0], $buf, 1);
239	$buf;
240	}
241
242	sub BINMODE {
243	binmode $_[0][0];
244	}
245
246	sub TELL {
247	use Carp (); Carp::croak("Coro::Handle's don't support tell()");
248	}
249
250	sub SEEK {
251	use Carp (); Carp::croak("Coro::Handle's don't support seek()");
252	}
253
254	sub EOF {
255	use Carp (); Carp::croak("Coro::Handle's don't support eof()");
256	}
257
258	sub CLOSE {
259	&cleanup;
260	close $_[0][0];
261	}
262
263	sub DESTROY {
264	&cleanup;
265	}
266
267	sub FILENO {
268	fileno $_[0][0];
269	}
270
271	# seems to be called for stringification (how weird), at least
272	# when DumpValue::dumpValue is used to print this.
273	sub FETCH {
274	"$_[0]<$_[0][1]>";
275	}
276
277	sub readable {
278	($_[0][5] \|\|= Coro::Event->io(
279	fd => $_[0][0],
280	desc => "$_[0][1] R",
281	timeout => $_[0][2],
282	poll => R+E,
283	))->next->{Coro::Event}[5] & R;
284	}
285
286	sub writable {
287	($_[0][6] \|\|= Coro::Event->io(
288	fd => $_[0][0],
289	desc => "$_[0][1] W",
290	timeout => $_[0][2],
291	poll => W+E,
292	))->next->{Coro::Event}[5] & W;
293	}
294
295	sub WRITE {
296	my $len = defined $_[2] ? $_[2] : length $_[1];
297	my $ofs = $_[3];
298	my $res = 0;
299
300	while() {
301	my $r = syswrite $_[0][0], $_[1], $len, $ofs;
302	if (defined $r) {
303	$len -= $r;
304	$ofs += $r;
305	$res += $r;
306	last unless $len;
307	} elsif ($! != Errno::EAGAIN) {
308	last;
309	}
310	last unless &writable;
311	}
312
313	return $res;
314	}
315
316	sub READ {
317	my $len = $_[2];
318	my $ofs = $_[3];
319	my $res = 0;
320
321	# first deplete the read buffer
322	if (length $_[0][3]) {
323	my $l = length $_[0][3];
324	if ($l <= $len) {
325	substr($_[1], $ofs) = $_[0][3]; $_[0][3] = "";
326	$len -= $l;
327	$ofs += $l;
328	$res += $l;
329	return $res unless $len;
330	} else {
331	substr($_[1], $ofs) = substr($_[0][3], 0, $len);
332	substr($_[0][3], 0, $len) = "";
333	return $len;
334	}
335	}
336
337	while() {
338	my $r = sysread $_[0][0], $_[1], $len, $ofs;
339	if (defined $r) {
340	$len -= $r;
341	$ofs += $r;
342	$res += $r;
343	last unless $len && $r;
344	} elsif ($! != Errno::EAGAIN) {
345	last;
346	}
347	last unless &readable;
348	}
349
350	return $res;
351	}
352
353	sub READLINE {
354	my $irs = @_ > 1 ? $_[1] : $/;
355
356	while() {
357	my $pos = index $_[0][3], $irs;
358	if ($pos >= 0) {
359	$pos += length $irs;
360	my $res = substr $_[0][3], 0, $pos;
361	substr ($_[0][3], 0, $pos) = "";
362	return $res;
363	}
364
365	my $r = sysread $_[0][0], $_[0][3], 8192, length $_[0][3];
366	if (defined $r) {
367	return undef unless $r;
368	} elsif ($! != Errno::EAGAIN \|\| !&readable) {
369	return undef;
370	}
371	}
372	}
373
374	1;
375
376	=head1 BUGS
377
378	- Perl's IO-Handle model is THE bug.
379
380	=head1 AUTHOR
381
382	Marc Lehmann <schmorp@schmorp.de>
383	http://home.schmorp.de/
384
385	=cut
386