Commit | Line | Data |
12fb4a80 |
1 | package Object::Remote::ReadChannel; |
2 | |
3 | use CPS::Future; |
4 | use Scalar::Util qw(weaken); |
9031635d |
5 | use Object::Remote::Logging qw(:log :dlog); |
90115979 |
6 | use POSIX; |
12fb4a80 |
7 | use Moo; |
8 | |
9 | has fh => ( |
10 | is => 'ro', required => 1, |
11 | trigger => sub { |
12 | my ($self, $fh) = @_; |
13 | weaken($self); |
5d59cb98 |
14 | log_trace { "Watching filehandle via trigger on 'fh' attribute in Object::Remote::ReadChannel" }; |
12fb4a80 |
15 | Object::Remote->current_loop |
16 | ->watch_io( |
17 | handle => $fh, |
18 | on_read_ready => sub { $self->_receive_data_from($fh) } |
19 | ); |
20 | }, |
21 | ); |
22 | |
23 | has on_close_call => ( |
24 | is => 'rw', default => sub { sub {} }, |
25 | ); |
26 | |
27 | has on_line_call => (is => 'rw'); |
28 | |
29 | has _receive_data_buffer => (is => 'ro', default => sub { my $x = ''; \$x }); |
30 | |
9d64d2d9 |
31 | #TODO confirmed this is the point of the hang - sysread() is invoked on a |
32 | #socket inside the controller that blocks and deadlocks the entire system. |
33 | #The remote nodes are all waiting to receive data at that point. |
34 | #Validated this behavior exists in an unmodified Object::Remote from CPAN |
35 | #by wrapping this sysread() with warns that have the pid in them and pounding |
36 | #my local machine with System::Introspector via ssh and 7 remote perl instances |
37 | #It looks like one of the futures is responding to an event regarding the ability |
38 | #to read from a socket and every once in a while an ordering issue means that |
39 | #there is no actual data to read from the socket |
12fb4a80 |
40 | sub _receive_data_from { |
41 | my ($self, $fh) = @_; |
9031635d |
42 | Dlog_trace { "Preparing to read data from $_" } $fh; |
9d64d2d9 |
43 | #use Carp qw(cluck); cluck(); |
12fb4a80 |
44 | my $rb = $self->_receive_data_buffer; |
9d64d2d9 |
45 | #TODO is there a specific reason sysread() and syswrite() aren't |
46 | #a part of ::MiniLoop? It's one spot to handle errors and other |
47 | #logic involving filehandles |
9031635d |
48 | my $len = sysread($fh, $$rb, 32768, length($$rb)); |
12fb4a80 |
49 | my $err = defined($len) ? '' : ": $!"; |
50 | if (defined($len) and $len > 0) { |
5d59cb98 |
51 | log_trace { "Read $len bytes of data" }; |
12fb4a80 |
52 | while (my $cb = $self->on_line_call and $$rb =~ s/^(.*)\n//) { |
53 | $cb->(my $line = $1); |
54 | } |
90115979 |
55 | #TODO this isn't compatible with Windows but would be if |
56 | #EAGAIN was set to something that could never match |
57 | #if on Windows |
58 | } elsif ($! != EAGAIN) { |
5d59cb98 |
59 | log_trace { "Got EOF or error, this read channel is done" }; |
12fb4a80 |
60 | Object::Remote->current_loop |
61 | ->unwatch_io( |
62 | handle => $self->fh, |
63 | on_read_ready => 1 |
64 | ); |
65 | $self->on_close_call->($err); |
66 | } |
67 | } |
68 | |
69 | sub DEMOLISH { |
70 | my ($self, $gd) = @_; |
71 | return if $gd; |
5d59cb98 |
72 | log_trace { "read channel is being demolished" }; |
12fb4a80 |
73 | Object::Remote->current_loop |
74 | ->unwatch_io( |
75 | handle => $self->fh, |
76 | on_read_ready => 1 |
77 | ); |
78 | } |
79 | |
80 | 1; |