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;

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];
}

sub read        { read     $_[0], $_[1], $_[2], $_[3] }
sub sysread     { sysread  $_[0], $_[1], $_[2], $_[3] }
sub syswrite    { syswrite $_[0], $_[1], $_[2], $_[3] }

=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

Work like their function equivalents.

=cut

sub fileno { tied(${$_[0]})->FILENO }
sub close  { tied(${$_[0]})->CLOSE  }

=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.

=cut

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

package Coro::Handle::FH;

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

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

use base 'Tie::Handle';

# 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
# 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 CLOSE {
   &cleanup;
   close $_[0][0];
}

sub DESTROY {
   &cleanup;
}

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

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.7
Committed:	Fri Sep 14 15:40:56 2001 UTC (22 years, 9 months ago) by root
Branch:	MAIN
Changes since 1.6:	+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	use Errno ();
22	use base 'Exporter';
23
24	$VERSION = 0.45;
25
26	@EXPORT = qw(unblock);
27
28	=item $fh = new_from_fh Coro::Handle $fhandle [, arg => value...]
29
30	Create a new non-blocking io-handle using the given
31	perl-filehandle. Returns undef if no fhandle is given. The only other
32	supported argument is "timeout", which sets a timeout for each operation.
33
34	=cut
35
36	sub new_from_fh {
37	my $class = shift;
38	my $fh = shift or return;
39	my $self = do { local *Coro::Handle };
40
41	my ($package, $filename, $line) = caller;
42	$filename =~ s/^.*[\/\\]//;
43
44	tie $self, Coro::Handle::FH, fh => $fh, desc => "$filename:$line", @_;
45
46	my $_fh = select bless \$self, $class; $\| = 1; select $_fh;
47	}
48
49	=item $fh = unblock $fh
50
51	This is a convinience function that just calls C<new_from_fh> on the given
52	filehandle. Use it to replace a normal perl filehandle by a non-blocking
53	equivalent.
54
55	=cut
56
57	sub unblock($) {
58	new_from_fh Coro::Handle $_[0];
59	}
60
61	sub read { read $_[0], $_[1], $_[2], $_[3] }
62	sub sysread { sysread $_[0], $_[1], $_[2], $_[3] }
63	sub syswrite { syswrite $_[0], $_[1], $_[2], $_[3] }
64
65	=item $fh->writable, $fh->readable
66
67	Wait until the filehandle is readable or writable (and return true) or
68	until an error condition happens (and return false).
69
70	=cut
71
72	sub readable { Coro::Handle::FH::readable(tied ${$_[0]}) }
73	sub writable { Coro::Handle::FH::writable(tied ${$_[0]}) }
74
75	=item $fh->readline([$terminator])
76
77	Like the builtin of the same name, but allows you to specify the input
78	record separator in a coroutine-safe manner (i.e. not using a global
79	variable).
80
81	=cut
82
83	sub readline { tied(${+shift})->READLINE(@_) }
84
85	=item $fh->autoflush([...])
86
87	Always returns true, arguments are being ignored (exists for compatibility
88	only). Might change in the future.
89
90	=cut
91
92	sub autoflush { !0 }
93
94	=item $fh->fileno, $fh->close
95
96	Work like their function equivalents.
97
98	=cut
99
100	sub fileno { tied(${$_[0]})->FILENO }
101	sub close { tied(${$_[0]})->CLOSE }
102
103	=item $fh->timeout([...])
104
105	The optional agrument sets the new timeout (in seconds) for this
106	handle. Returns the current (new) value.
107
108	C<0> is a valid timeout, use C<undef> to disable the timeout.
109
110	=cut
111
112	sub timeout {
113	my $self = tied(${$_[0]});
114	if (@_ > 1) {
115	$self->[2] = $_[1];
116	$self->[5]->timeout($_[1]) if $self->[5];
117	$self->[6]->timeout($_[1]) if $self->[6];
118	}
119	$self->[2];
120	}
121
122	=item $fh->fh
123
124	Returns the "real" (non-blocking) filehandle. Use this if you want to
125	do operations on the file handle you cannot do using the Coro::Handle
126	interface.
127
128	=cut
129
130	sub fh {
131	tied(${$_[0]})->{fh};
132	}
133
134	package Coro::Handle::FH;
135
136	use Fcntl ();
137	use Errno ();
138	use Carp 'croak';
139
140	use Coro::Event;
141	use Event::Watcher qw(R W E);
142
143	use base 'Tie::Handle';
144
145	# formerly a hash, but we are speed-critical, so try
146	# to be faster even if it hurts.
147	#
148	# 0 FH
149	# 1 desc
150	# 2 timeout
151	# 3 rb
152	# 4 wb
153	# 5 rw
154	# 6 ww
155
156	sub TIEHANDLE {
157	my $class = shift;
158	my %args = @_;
159
160	my $self = bless [], $class;
161	$self->[0] = $args{fh};
162	$self->[1] = $args{desc};
163	$self->[2] = $args{timeout};
164	$self->[3] = "";
165	$self->[4] = "";
166
167	fcntl $self->[0], &Fcntl::F_SETFL, &Fcntl::O_NONBLOCK
168	or croak "fcntl(O_NONBLOCK): $!";
169
170	$self;
171	}
172
173	sub cleanup {
174	$_[0][3] = "";
175	($_[0][5])->cancel if defined $_[0][5]; $_[0][5] = undef;
176
177	$_[0][4] = "";
178	($_[0][6])->cancel if defined $_[0][6]; $_[0][6] = undef;
179	}
180
181	sub OPEN {
182	&cleanup;
183	my $self = shift;
184	my $r = @_ == 2 ? open $self->[0], $_[0], $_[1]
185	: open $self->[0], $_[0], $_[1], $_[2];
186	if ($r) {
187	fcntl $self->[0], &Fcntl::F_SETFL, &Fcntl::O_NONBLOCK
188	or croak "fcntl(O_NONBLOCK): $!";
189	}
190	$r;
191	}
192
193	sub CLOSE {
194	&cleanup;
195	close $_[0][0];
196	}
197
198	sub DESTROY {
199	&cleanup;
200	}
201
202	sub FILENO {
203	fileno $_[0][0];
204	}
205
206	sub readable {
207	($_[0][5] \|\|= Coro::Event->io(
208	fd => $_[0][0],
209	desc => "$_[0][1] R",
210	timeout => $_[0][2],
211	poll => R+E,
212	))->next->{Coro::Event}[5] & R;
213	}
214
215	sub writable {
216	($_[0][6] \|\|= Coro::Event->io(
217	fd => $_[0][0],
218	desc => "$_[0][1] W",
219	timeout => $_[0][2],
220	poll => W+E,
221	))->next->{Coro::Event}[5] & W;
222	}
223
224	sub WRITE {
225	my $len = defined $_[2] ? $_[2] : length $_[1];
226	my $ofs = $_[3];
227	my $res = 0;
228
229	while() {
230	my $r = syswrite $_[0][0], $_[1], $len, $ofs;
231	if (defined $r) {
232	$len -= $r;
233	$ofs += $r;
234	$res += $r;
235	last unless $len;
236	} elsif ($! != Errno::EAGAIN) {
237	last;
238	}
239	last unless &writable;
240	}
241
242	return $res;
243	}
244
245	sub READ {
246	my $len = $_[2];
247	my $ofs = $_[3];
248	my $res = 0;
249
250	# first deplete the read buffer
251	if (defined $_[0][3]) {
252	my $l = length $_[0][3];
253	if ($l <= $len) {
254	substr($_[1], $ofs) = $_[0][3]; undef $_[0][3];
255	$len -= $l;
256	$res += $l;
257	return $res unless $len;
258	} else {
259	substr($_[1], $ofs) = substr($_[0][3], 0, $len);
260	substr($_[0][3], 0, $len) = "";
261	return $len;
262	}
263	}
264
265	while() {
266	my $r = sysread $_[0][0], $_[1], $len, $ofs;
267	if (defined $r) {
268	$len -= $r;
269	$ofs += $r;
270	$res += $r;
271	last unless $len && $r;
272	} elsif ($! != Errno::EAGAIN) {
273	last;
274	}
275	last unless &readable;
276	}
277
278	return $res;
279	}
280
281	sub READLINE {
282	my $irs = @_ > 1 ? $_[1] : $/;
283
284	while() {
285	my $pos = index $_[0][3], $irs;
286	if ($pos >= 0) {
287	$pos += length $irs;
288	my $res = substr $_[0][3], 0, $pos;
289	substr ($_[0][3], 0, $pos) = "";
290	return $res;
291	}
292
293	my $r = sysread $_[0][0], $_[0][3], 8192, length $_[0][3];
294	if (defined $r) {
295	return undef unless $r;
296	} elsif ($! != Errno::EAGAIN \|\| !&readable) {
297	return undef;
298	}
299	}
300	}
301
302	1;
303
304	=head1 BUGS
305
306	- Perl's IO-Handle model is THE bug.
307
308	=head1 AUTHOR
309
310	Marc Lehmann <pcg@goof.com>
311	http://www.goof.com/pcg/marc/
312
313	=cut
314