AnyEvent-Fork-RPC/RPC/Async.pm

package AnyEvent::Fork::RPC::Async;

use common::sense; # actually required to avoid spurious warnings...

our $VERSION = 2; # protocol version

use Errno ();

use AnyEvent;

# declare only
sub AnyEvent::Fork::RPC::event;
sub AnyEvent::Fork::RPC::flush;

sub do_exit { exit } # workaround for perl 5.14 and below

sub run {
   my %kv = splice @_, pop;

   my $rfh = shift;
   my $wfh = fileno $rfh ? $rfh : *STDOUT;

   my $function   = delete $kv{function};
   my $serialiser = delete $kv{serialiser};
   my $rlen       = delete $kv{rlen};
   my $done       = delete $kv{done};

   $0 =~ s/^(\d+).*$/$1 $function/s;

   {
      package main;
      my $init = delete $kv{init};
      &$init if length $init;
      $function = \&$function; # resolve function early for extra speed
   }

   %kv = (); # save some very small amount of memory

   my $busy = 1; # exit when == 0

   my ($f, $t) = eval $serialiser; AE::log fatal => $@ if $@;
   my ($wbuf, $ww);

   my $wcb = sub {
      my $len = syswrite $wfh, $wbuf;

      unless (defined $len) {
         if ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
            undef $ww;
            AE::log fatal => "AnyEvent::Fork::RPC: write error ($!), parent gone?";
         }
      }

      substr $wbuf, 0, $len, "";

      unless (length $wbuf) {
         undef $ww;
         unless ($busy) {
            shutdown $wfh, 1;
            @_ = (); goto &$done;
         }
      }
   };

   my $write = sub {
      $wbuf .= $_[0];
      $ww ||= AE::io $wfh, 1, $wcb;
   };

   *AnyEvent::Fork::RPC::flush = sub {
      while (length $wbuf) {
         my $len = syswrite $wfh, $wbuf;

         if (defined $len) {
            substr $wbuf, 0, $len, "";
         } elsif ($! == Errno::EAGAIN || $! == Errno::EWOULDBLOCK) {
            my $fdset;
            (vec $fdset, fileno $wfh, 1) = 1;
            # buggy windows often sets exceptfds instead of wfds
            select undef, my $wset = $fdset, my $eset = $fdset, undef;
         } else {
            return 0;
         }
      }

      1
   };

   *AnyEvent::Fork::RPC::event = sub {
      $write->(pack "NN/a*", 0, &$f);
   };

   my ($rbuf, $rw);

   my $len;

   $rw = AE::io $rfh, 0, sub {
      $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
      $len = sysread $rfh, $rbuf, $rlen - length $rbuf, length $rbuf;

      if ($len) {
         while (8 <= length $rbuf) {
            (my $id, $len) = unpack "NN", $rbuf;
            8 + $len <= length $rbuf
               or last;

            my @r = $t->(substr $rbuf, 8, $len);
            substr $rbuf, 0, 8 + $len, "";
            
            ++$busy;
            $function->(sub {
               --$busy;
               $write->(pack "NN/a*", $id, &$f);
            }, @r);
         }
      } elsif (defined $len or $! == Errno::EINVAL) { # EINVAL is for microshit windoze
         undef $rw;
         --$busy;
         $ww ||= AE::io $wfh, 1, $wcb;
      } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
         undef $rw;
         AE::log fatal => "AnyEvent::Fork::RPC: read error in child: $!";
      }
   };

   $AnyEvent::MODEL eq "AnyEvent::Impl::EV"
      ? EV::run ()
      : AE::cv->recv;
}

1

Revision:	1.18
Committed:	Sun Sep 15 20:18:14 2019 UTC (7 years ago) by root
Branch:	MAIN
CVS Tags:	HEAD
Changes since 1.17:	+20 -0 lines
Log Message:	2.0
#	User	Rev	Content
1	root	1.1	package AnyEvent::Fork::RPC::Async;
2
3			use common::sense; # actually required to avoid spurious warnings...
4
5	root	1.16	our $VERSION = 2; # protocol version
6
7	root	1.1	use Errno ();
8
9			use AnyEvent;
10
11			# declare only
12			sub AnyEvent::Fork::RPC::event;
13	root	1.18	sub AnyEvent::Fork::RPC::flush;
14	root	1.1
15	root	1.9	sub do_exit { exit } # workaround for perl 5.14 and below
16
17	root	1.1	sub run {
18	root	1.16	my %kv = splice @_, pop;
19
20	root	1.4	my $rfh = shift;
21			my $wfh = fileno $rfh ? $rfh : *STDOUT;
22	root	1.1
23	root	1.16	my $function = delete $kv{function};
24			my $serialiser = delete $kv{serialiser};
25			my $rlen = delete $kv{rlen};
26			my $done = delete $kv{done};
27
28	root	1.15	$0 =~ s/^(\d+).*$/$1 $function/s;
29	root	1.5
30	root	1.1	{
31			package main;
32	root	1.16	my $init = delete $kv{init};
33	root	1.1	&$init if length $init;
34			$function = \&$function; # resolve function early for extra speed
35			}
36
37	root	1.17	%kv = (); # save some very small amount of memory
38
39	root	1.1	my $busy = 1; # exit when == 0
40
41	root	1.8	my ($f, $t) = eval $serialiser; AE::log fatal => $@ if $@;
42	root	1.1	my ($wbuf, $ww);
43
44			my $wcb = sub {
45	root	1.4	my $len = syswrite $wfh, $wbuf;
46	root	1.1
47			unless (defined $len) {
48			if ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
49			undef $ww;
50	root	1.8	AE::log fatal => "AnyEvent::Fork::RPC: write error ($!), parent gone?";
51	root	1.1	}
52			}
53
54			substr $wbuf, 0, $len, "";
55
56			unless (length $wbuf) {
57			undef $ww;
58	root	1.3	unless ($busy) {
59	root	1.4	shutdown $wfh, 1;
60	root	1.6	@_ = (); goto &$done;
61	root	1.3	}
62	root	1.1	}
63			};
64
65			my $write = sub {
66			$wbuf .= $_[0];
67	root	1.4	$ww \|\|= AE::io $wfh, 1, $wcb;
68	root	1.1	};
69
70	root	1.18	*AnyEvent::Fork::RPC::flush = sub {
71			while (length $wbuf) {
72			my $len = syswrite $wfh, $wbuf;
73
74			if (defined $len) {
75			substr $wbuf, 0, $len, "";
76			} elsif ($! == Errno::EAGAIN \|\| $! == Errno::EWOULDBLOCK) {
77			my $fdset;
78			(vec $fdset, fileno $wfh, 1) = 1;
79			# buggy windows often sets exceptfds instead of wfds
80			select undef, my $wset = $fdset, my $eset = $fdset, undef;
81			} else {
82			return 0;
83			}
84			}
85
86			1
87			};
88
89	root	1.1	*AnyEvent::Fork::RPC::event = sub {
90	root	1.4	$write->(pack "NN/a*", 0, &$f);
91	root	1.1	};
92
93	root	1.16	my ($rbuf, $rw);
94	root	1.1
95	root	1.2	my $len;
96	root	1.1
97	root	1.4	$rw = AE::io $rfh, 0, sub {
98	root	1.1	$rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
99	root	1.4	$len = sysread $rfh, $rbuf, $rlen - length $rbuf, length $rbuf;
100	root	1.1
101			if ($len) {
102			while (8 <= length $rbuf) {
103	root	1.4	(my $id, $len) = unpack "NN", $rbuf;
104	root	1.1	8 + $len <= length $rbuf
105			or last;
106
107			my @r = $t->(substr $rbuf, 8, $len);
108			substr $rbuf, 0, 8 + $len, "";
109
110			++$busy;
111			$function->(sub {
112			--$busy;
113	root	1.4	$write->(pack "NN/a*", $id, &$f);
114	root	1.1	}, @r);
115			}
116	root	1.7	} elsif (defined $len or $! == Errno::EINVAL) { # EINVAL is for microshit windoze
117	root	1.1	undef $rw;
118			--$busy;
119	root	1.4	$ww \|\|= AE::io $wfh, 1, $wcb;
120	root	1.1	} elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
121			undef $rw;
122	root	1.8	AE::log fatal => "AnyEvent::Fork::RPC: read error in child: $!";
123	root	1.1	}
124			};
125
126	root	1.14	$AnyEvent::MODEL eq "AnyEvent::Impl::EV"
127	root	1.10	? EV::run ()
128	root	1.1	: AE::cv->recv;
129			}
130
131			1
132