Coro/Coro/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.13;

@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    { tied(${$_[0]})->readable }
sub writable    { tied(${$_[0]})->writable }

=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 usign a global
variable).

=cut

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

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

Always returns true, arguments are being ignored (exists for compatibility
only).

=cut

sub autoflush   { !0 }

package Coro::Handle::FH;

use Fcntl ();
use Errno ();

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

use base 'Tie::Handle';

sub TIEHANDLE {
   my $class = shift;

   my $self = bless {
      rb => "",
      wb => "",
      @_,
   }, $class;

   fcntl $self->{fh}, &Fcntl::F_SETFL, &Fcntl::O_NONBLOCK
      or die "fcntl(O_NONBLOCK): $!";

   $self;
}

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

sub CLOSE {
   my $self = shift;
   $self->{rb} =
   $self->{wb} = "";
   (delete $self->{rw})->cancel if $self->{rw};
   (delete $self->{ww})->cancel if $self->{ww};
   close $self->{fh};
}

sub writable {
   ($_[0]->{ww} ||= Coro::Event->io(
      fd      => $_[0]->{fh},
      desc    => "$_[0]->{desc} WW",
      timeout => $_[0]->{timeout},
      poll    => W+E,
   ))->next->got & W;
}

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

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

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

   return $res;
}

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

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

   return $res;
}

sub READLINE {
   my $self = shift;
   my $irs = @_ ? shift : $/;

   while() {
      my $pos = index $self->{rb}, $irs;
      if ($pos >= 0) {
         $pos += length $/;
         my $res = substr $self->{rb}, 0, $pos;
         substr ($self->{rb}, 0, $pos) = "";
         return $res;
      }
      my $r = sysread $self->{fh}, $self->{rb}, 8192, length $self->{rb};
      if (defined $r) {
         return undef unless $r;
      } elsif ($! != Errno::EAGAIN || !$self->readable) {
         return undef;
      }
   }
}

sub DESTROY {
   &CLOSE;
}

1;

=head1 BUGS

 - Perl's IO-Handle model is THE bug.
 - READLINE cannot be mixed with other forms of input.

=head1 AUTHOR

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

=cut

Revision:	1.6
Committed:	Fri Aug 3 12:51:56 2001 UTC (22 years, 10 months ago) by root
Branch:	MAIN
Changes since 1.5:	+32 -13 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			use Errno ();
22	root	1.5	use base 'Exporter';
23	root	1.1
24	root	1.5	$VERSION = 0.13;
25
26			@EXPORT = qw(unblock);
27	root	1.1
28	root	1.6	=item $fh = new_from_fh Coro::Handle $fhandle [, arg => value...]
29	root	1.1
30			Create a new non-blocking io-handle using the given
31	root	1.6	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	root	1.1
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	root	1.6	my ($package, $filename, $line) = caller;
42			$filename =~ s/^.*[\/\\]//;
43
44			tie $self, Coro::Handle::FH, fh => $fh, desc => "$filename:$line", @_;
45	root	1.1
46			my $_fh = select bless \$self, $class; $\| = 1; select $_fh;
47			}
48
49	root	1.5	=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	root	1.1	=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 { tied(${$_[0]})->readable }
73			sub writable { tied(${$_[0]})->writable }
74
75	root	1.4	=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 usign 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).
89
90			=cut
91
92			sub autoflush { !0 }
93
94	root	1.1	package Coro::Handle::FH;
95
96			use Fcntl ();
97			use Errno ();
98
99			use Coro::Event;
100			use Event::Watcher qw(R W E);
101
102			use base 'Tie::Handle';
103
104			sub TIEHANDLE {
105	root	1.6	my $class = shift;
106	root	1.1
107	root	1.6	my $self = bless {
108	root	1.1	rb => "",
109			wb => "",
110	root	1.6	@_,
111			}, $class;
112	root	1.1
113	root	1.6	fcntl $self->{fh}, &Fcntl::F_SETFL, &Fcntl::O_NONBLOCK
114			or die "fcntl(O_NONBLOCK): $!";
115
116			$self;
117	root	1.1	}
118
119			sub OPEN {
120			my $self = shift;
121			$self->CLOSE;
122			my $r = @_ == 2 ? open $self->{fh}, $_[0], $_[1]
123			: open $self->{fh}, $_[0], $_[1], $_[2];
124			if ($r) {
125			fcntl $self->{fh}, &Fcntl::F_SETFL, &Fcntl::O_NONBLOCK
126			or die "fcntl(O_NONBLOCK): $!";
127			}
128			$r;
129			}
130
131			sub CLOSE {
132			my $self = shift;
133			$self->{rb} =
134			$self->{wb} = "";
135	root	1.5	(delete $self->{rw})->cancel if $self->{rw};
136	root	1.6	(delete $self->{ww})->cancel if $self->{ww};
137	root	1.1	close $self->{fh};
138			}
139
140			sub writable {
141	root	1.6	($_[0]->{ww} \|\|= Coro::Event->io(
142			fd => $_[0]->{fh},
143			desc => "$_[0]->{desc} WW",
144			timeout => $_[0]->{timeout},
145			poll => W+E,
146			))->next->got & W;
147	root	1.1	}
148
149			sub readable {
150	root	1.6	($_[0]->{rw} \|\|= Coro::Event->io(
151			fd => $_[0]->{fh},
152			desc => "$_[0]->{desc} RW",
153			timeout => $_[0]->{timeout},
154			poll => R+E,
155			))->next->got & R;
156	root	1.1	}
157
158			sub WRITE {
159			my $self = $_[0];
160			my $len = defined $_[2] ? $_[2] : length $_[1];
161			my $ofs = $_[3];
162			my $res = 0;
163
164	root	1.2	while() {
165	root	1.1	my $r = syswrite $self->{fh}, $_[1], $len, $ofs;
166			if (defined $r) {
167			$len -= $r;
168			$ofs += $r;
169			$res += $r;
170			last unless $len;
171			} elsif ($! != Errno::EAGAIN) {
172			last;
173			}
174	root	1.2	last unless $self->writable;
175	root	1.1	}
176
177			return $res;
178			}
179
180			sub READ {
181			my $self = $_[0];
182			my $len = $_[2];
183			my $ofs = $_[3];
184			my $res = 0;
185
186	root	1.2	while() {
187	root	1.1	my $r = sysread $self->{fh}, $_[1], $len, $ofs;
188			if (defined $r) {
189			$len -= $r;
190			$ofs += $r;
191			$res += $r;
192			last unless $len && $r;
193			} elsif ($! != Errno::EAGAIN) {
194			last;
195			}
196	root	1.2	last unless $self->readable;
197	root	1.1	}
198
199			return $res;
200			}
201
202			sub READLINE {
203			my $self = shift;
204	root	1.4	my $irs = @_ ? shift : $/;
205	root	1.1
206			while() {
207	root	1.4	my $pos = index $self->{rb}, $irs;
208	root	1.1	if ($pos >= 0) {
209			$pos += length $/;
210			my $res = substr $self->{rb}, 0, $pos;
211			substr ($self->{rb}, 0, $pos) = "";
212			return $res;
213			}
214			my $r = sysread $self->{fh}, $self->{rb}, 8192, length $self->{rb};
215			if (defined $r) {
216			return undef unless $r;
217	root	1.2	} elsif ($! != Errno::EAGAIN \|\| !$self->readable) {
218	root	1.1	return undef;
219			}
220			}
221	root	1.6	}
222
223			sub DESTROY {
224			&CLOSE;
225	root	1.1	}
226
227			1;
228
229			=head1 BUGS
230
231			- Perl's IO-Handle model is THE bug.
232			- READLINE cannot be mixed with other forms of input.
233
234			=head1 AUTHOR
235
236			Marc Lehmann <pcg@goof.com>
237			http://www.goof.com/pcg/marc/
238
239			=cut
240