]> dxcluster.net Git - spider.git/blob - perl/Msg.pm
Merge branch 'master' of /scm/spider
[spider.git] / perl / Msg.pm
1 #
2 # This has been taken from the 'Advanced Perl Programming' book by Sriram Srinivasan 
3 #
4 # I am presuming that the code is distributed on the same basis as perl itself.
5 #
6 # I have modified it to suit my devious purposes (Dirk Koopman G1TLH)
7 #
8 #
9 #
10
11 package Msg;
12
13 use strict;
14
15 use DXUtil;
16
17 use IO::Select;
18 use DXDebug;
19 use Timer;
20
21 use vars qw(%rd_callbacks %wt_callbacks %er_callbacks $rd_handles $wt_handles $er_handles $now %conns $noconns $blocking_supported $cnum $total_in $total_out $io_socket);
22
23 %rd_callbacks = ();
24 %wt_callbacks = ();
25 %er_callbacks = ();
26 $rd_handles   = IO::Select->new();
27 $wt_handles   = IO::Select->new();
28 $er_handles   = IO::Select->new();
29 $total_in = $total_out = 0;
30
31 $now = time;
32
33 BEGIN {
34     # Checks if blocking is supported
35     eval {
36                 local $^W;
37         require POSIX; POSIX->import(qw(O_NONBLOCK F_SETFL F_GETFL))
38     };
39
40         eval {
41                 local $^W;
42                 require IO::Socket::INET6;
43         };
44
45         if ($@) {
46                 dbg($@);
47                 require IO::Socket;
48                 $io_socket = 'IO::Socket::INET';
49         } else {
50                 $io_socket = 'IO::Socket::INET6';
51         }
52         $io_socket->import;
53
54         if ($@ || $main::is_win) {
55                 $blocking_supported = $io_socket->can('blocking') ? 2 : 0;
56         } else {
57                 $blocking_supported = $io_socket->can('blocking') ? 2 : 1;
58         }
59
60
61         # import as many of these errno values as are available
62         eval {
63                 local $^W;
64                 require Errno; Errno->import(qw(EAGAIN EINPROGRESS EWOULDBLOCK));
65         };
66
67         unless ($^O eq 'MSWin32') {
68                 if ($] >= 5.6) {
69                         eval {
70                                 local $^W;
71                                 require Socket; Socket->import(qw(IPPROTO_TCP TCP_NODELAY));
72                         };
73                 } else {
74                         dbg("IPPROTO_TCP and TCP_NODELAY manually defined");
75                         eval 'sub IPPROTO_TCP {     6 };';
76                         eval 'sub TCP_NODELAY {     1 };';
77                 }
78         }
79         # http://support.microsoft.com/support/kb/articles/Q150/5/37.asp
80         # defines EINPROGRESS as 10035.  We provide it here because some
81         # Win32 users report POSIX::EINPROGRESS is not vendor-supported.
82         if ($^O eq 'MSWin32') { 
83                 eval '*EINPROGRESS = sub { 10036 };' unless defined *EINPROGRESS;
84                 eval '*EWOULDBLOCK = *EAGAIN = sub { 10035 };' unless defined *EWOULDBLOCK;
85                 eval '*F_GETFL     = sub {     0 };';
86                 eval '*F_SETFL     = sub {     0 };';
87                 eval '*IPPROTO_TCP     = sub {     6 };';
88                 eval '*TCP_NODELAY     = sub {     1 };';
89                 $blocking_supported = 0;   # it appears that this DOESN'T work :-(
90         } 
91 }
92
93 my $w = $^W;
94 $^W = 0;
95 my $eagain = eval {EAGAIN()};
96 my $einprogress = eval {EINPROGRESS()};
97 my $ewouldblock = eval {EWOULDBLOCK()};
98 $^W = $w;
99 $cnum = 0;
100
101
102 #
103 #-----------------------------------------------------------------
104 # Generalised initializer
105
106 sub new
107 {
108     my ($pkg, $rproc) = @_;
109         my $obj = ref($pkg);
110         my $class = $obj || $pkg;
111
112     my $conn = {
113         rproc => $rproc,
114                 inqueue => [],
115                 outqueue => [],
116                 state => 0,
117                 lineend => "\r\n",
118                 csort => 'telnet',
119                 timeval => 60,
120                 blocking => 0,
121                 cnum => (($cnum < 999) ? (++$cnum) : ($cnum = 1)),
122     };
123
124         $noconns++;
125         
126         dbg("Connection created ($noconns)") if isdbg('connll');
127         return bless $conn, $class;
128 }
129
130 sub set_error
131 {
132         my $conn = shift;
133         my $callback = shift;
134         $conn->{eproc} = $callback;
135         set_event_handler($conn->{sock}, error => $callback) if exists $conn->{sock};
136 }
137
138 sub set_rproc
139 {
140         my $conn = shift;
141         my $callback = shift;
142         $conn->{rproc} = $callback;
143 }
144
145 sub blocking
146 {
147         return unless $blocking_supported;
148
149         # Make the handle stop blocking, the Windows way.
150         if ($blocking_supported) { 
151                 $_[0]->blocking($_[1]);
152         } else {
153                 my $flags = fcntl ($_[0], F_GETFL, 0);
154                 if ($_[1]) {
155                         $flags &= ~O_NONBLOCK;
156                 } else {
157                         $flags |= O_NONBLOCK;
158                 }
159                 fcntl ($_[0], F_SETFL, $flags);
160         }
161 }
162
163 # save it
164 sub conns
165 {
166         my $pkg = shift;
167         my $call = shift;
168         my $ref;
169         
170         if (ref $pkg) {
171                 $call = $pkg->{call} unless $call;
172                 return undef unless $call;
173                 dbg("changing $pkg->{call} to $call") if isdbg('connll') && exists $pkg->{call} && $call ne $pkg->{call};
174                 delete $conns{$pkg->{call}} if exists $pkg->{call} && exists $conns{$pkg->{call}} && $pkg->{call} ne $call; 
175                 $pkg->{call} = $call;
176                 $ref = $conns{$call} = $pkg;
177                 dbg("Connection $pkg->{cnum} $call stored") if isdbg('connll');
178         } else {
179                 $ref = $conns{$call};
180         }
181         return $ref;
182 }
183
184 # this is only called by any dependent processes going away unexpectedly
185 sub pid_gone
186 {
187         my ($pkg, $pid) = @_;
188         
189         my @pid = grep {$_->{pid} == $pid} values %conns;
190         foreach my $p (@pid) {
191                 &{$p->{eproc}}($p, "$pid has gorn") if exists $p->{eproc};
192                 $p->disconnect;
193         }
194 }
195
196 #-----------------------------------------------------------------
197 # Send side routines
198 sub connect {
199     my ($pkg, $to_host, $to_port, $rproc) = @_;
200
201     # Create a connection end-point object
202     my $conn = $pkg;
203         unless (ref $pkg) {
204                 $conn = $pkg->new($rproc);
205         }
206         $conn->{peerhost} = $to_host;
207         $conn->{peerport} = $to_port;
208         $conn->{sort} = 'Outgoing';
209         
210     # Create a new internet socket
211     my $sock = $io_socket->new();
212     return undef unless $sock;
213         
214         my $proto = getprotobyname('tcp');
215         $sock->socket(AF_INET, SOCK_STREAM, $proto) or return undef;
216         
217         blocking($sock, 0);
218         $conn->{blocking} = 0;
219
220         # does the host resolve?
221         my $ip = gethostbyname($to_host);
222         return undef unless $ip;
223         
224         my $r = connect($sock, pack_sockaddr_in($to_port, $ip));
225         return undef unless $r || _err_will_block($!);
226         
227         $conn->{sock} = $sock;
228     
229     if ($conn->{rproc}) {
230         my $callback = sub {$conn->_rcv};
231         set_event_handler ($sock, read => $callback);
232     }
233     return $conn;
234 }
235
236 sub start_program
237 {
238         my ($conn, $line, $sort) = @_;
239         my $pid;
240         
241         local $^F = 10000;              # make sure it ain't closed on exec
242         my ($a, $b) = $io_socket->socketpair(AF_UNIX, SOCK_STREAM, PF_UNSPEC);
243         if ($a && $b) {
244                 $a->autoflush(1);
245                 $b->autoflush(1);
246                 $pid = fork;
247                 if (defined $pid) {
248                         if ($pid) {
249                                 close $b;
250                                 $conn->{sock} = $a;
251                                 $conn->{csort} = $sort;
252                                 $conn->{lineend} = "\cM" if $sort eq 'ax25';
253                                 $conn->{pid} = $pid;
254                                 if ($conn->{rproc}) {
255                                         my $callback = sub {$conn->_rcv};
256                                         Msg::set_event_handler ($a, read => $callback);
257                                 }
258                                 dbg("connect $conn->{cnum}: started pid: $conn->{pid} as $line") if isdbg('connect');
259                         } else {
260                                 $^W = 0;
261                                 dbgclose();
262                                 STDIN->close;
263                                 STDOUT->close;
264                                 STDOUT->close;
265                                 *STDIN = IO::File->new_from_fd($b, 'r') or die;
266                                 *STDOUT = IO::File->new_from_fd($b, 'w') or die;
267                                 *STDERR = IO::File->new_from_fd($b, 'w') or die;
268                                 close $a;
269                                 unless ($main::is_win) {
270                                         #                                               $SIG{HUP} = 'IGNORE';
271                                         $SIG{HUP} = $SIG{CHLD} = $SIG{TERM} = $SIG{INT} = 'DEFAULT';
272                                         alarm(0);
273                                 }
274                                 exec "$line" or dbg("exec '$line' failed $!");
275                         } 
276                 } else {
277                         dbg("cannot fork for $line");
278                 }
279         } else {
280                 dbg("no socket pair $! for $line");
281         }
282         return $pid;
283 }
284
285 sub disconnect 
286 {
287     my $conn = shift;
288         return if exists $conn->{disconnecting};
289
290         $conn->{disconnecting} = 1;
291     my $sock = delete $conn->{sock};
292         $conn->{state} = 'E';
293         $conn->{timeout}->del if $conn->{timeout};
294
295         # be careful to delete the correct one
296         my $call;
297         if ($call = $conn->{call}) {
298                 my $ref = $conns{$call};
299                 delete $conns{$call} if $ref && $ref == $conn;
300         }
301         $call ||= 'unallocated';
302         dbg("Connection $conn->{cnum} $call disconnected") if isdbg('connll');
303         
304         # get rid of any references
305         for (keys %$conn) {
306                 if (ref($conn->{$_})) {
307                         delete $conn->{$_};
308                 }
309         }
310
311         if (defined($sock)) {
312                 set_event_handler ($sock, read => undef, write => undef, error => undef);
313                 shutdown($sock, 3);
314                 close($sock);
315         }
316         
317         unless ($main::is_win) {
318                 kill 'TERM', $conn->{pid} if exists $conn->{pid};
319         }
320 }
321
322 sub send_now {
323     my ($conn, $msg) = @_;
324     $conn->enqueue($msg);
325     $conn->_send (1); # 1 ==> flush
326 }
327
328 sub send_later {
329     my ($conn, $msg) = @_;
330     $conn->enqueue($msg);
331     my $sock = $conn->{sock};
332     return unless defined($sock);
333     set_event_handler ($sock, write => sub {$conn->_send(0)});
334 }
335
336 sub enqueue {
337     my $conn = shift;
338     push (@{$conn->{outqueue}}, defined $_[0] ? $_[0] : '');
339 }
340
341 sub _send {
342     my ($conn, $flush) = @_;
343     my $sock = $conn->{sock};
344     return unless defined($sock);
345     my $rq = $conn->{outqueue};
346
347     # If $flush is set, set the socket to blocking, and send all
348     # messages in the queue - return only if there's an error
349     # If $flush is 0 (deferred mode) make the socket non-blocking, and
350     # return to the event loop only after every message, or if it
351     # is likely to block in the middle of a message.
352
353 #       if ($conn->{blocking} != $flush) {
354 #               blocking($sock, $flush);
355 #               $conn->{blocking} = $flush;
356 #       }
357     my $offset = (exists $conn->{send_offset}) ? $conn->{send_offset} : 0;
358
359     while (@$rq) {
360         my $msg            = $rq->[0];
361                 my $mlth           = length($msg);
362         my $bytes_to_write = $mlth - $offset;
363         my $bytes_written  = 0;
364                 confess("Negative Length! msg: '$msg' lth: $mlth offset: $offset") if $bytes_to_write < 0;
365         while ($bytes_to_write > 0) {
366             $bytes_written = syswrite ($sock, $msg,
367                                        $bytes_to_write, $offset);
368             if (!defined($bytes_written)) {
369                 if (_err_will_block($!)) {
370                     # Should happen only in deferred mode. Record how
371                     # much we have already sent.
372                     $conn->{send_offset} = $offset;
373                     # Event handler should already be set, so we will
374                     # be called back eventually, and will resume sending
375                     return 1;
376                 } else {    # Uh, oh
377                                         &{$conn->{eproc}}($conn, $!) if exists $conn->{eproc};
378                                         $conn->disconnect;
379                     return 0; # fail. Message remains in queue ..
380                 }
381             } elsif (isdbg('raw')) {
382                                 my $call = $conn->{call} || 'none';
383                                 dbgdump('raw', "$call send $bytes_written: ", $msg);
384                         }
385                         $total_out      += $bytes_written;
386             $offset         += $bytes_written;
387             $bytes_to_write -= $bytes_written;
388         }
389         delete $conn->{send_offset};
390         $offset = 0;
391         shift @$rq;
392         #last unless $flush; # Go back to select and wait
393                             # for it to fire again.
394     }
395     # Call me back if queue has not been drained.
396     unless (@$rq) {
397         set_event_handler ($sock, write => undef);
398                 if (exists $conn->{close_on_empty}) {
399                         &{$conn->{eproc}}($conn, undef) if exists $conn->{eproc};
400                         $conn->disconnect; 
401                 }
402     }
403     1;  # Success
404 }
405
406 sub dup_sock
407 {
408         my $conn = shift;
409         my $oldsock = $conn->{sock};
410         my $rc = $rd_callbacks{$oldsock};
411         my $wc = $wt_callbacks{$oldsock};
412         my $ec = $er_callbacks{$oldsock};
413         my $sock = $oldsock->new_from_fd($oldsock, "w+");
414         if ($sock) {
415                 set_event_handler($oldsock, read=>undef, write=>undef, error=>undef);
416                 $conn->{sock} = $sock;
417                 set_event_handler($sock, read=>$rc, write=>$wc, error=>$ec);
418                 $oldsock->close;
419         }
420 }
421
422 sub _err_will_block {
423         return 0 unless $blocking_supported;
424         return ($_[0] == $eagain || $_[0] == $ewouldblock || $_[0] == $einprogress);
425 }
426
427 sub close_on_empty
428 {
429         my $conn = shift;
430         $conn->{close_on_empty} = 1;
431 }
432
433 #-----------------------------------------------------------------
434 # Receive side routines
435
436 sub new_server {
437     @_ == 4 || die "Msg->new_server (myhost, myport, login_proc\n";
438     my ($pkg, $my_host, $my_port, $login_proc) = @_;
439         my $self = $pkg->new($login_proc);
440         
441     $self->{sock} = $io_socket->new (
442                                           LocalAddr => "$my_host:$my_port",
443 #                                          LocalPort => $my_port,
444                                           Listen    => SOMAXCONN,
445                                           Proto     => 'tcp',
446                                           Reuse => 1);
447     die "Could not create socket: $! \n" unless $self->{sock};
448     set_event_handler ($self->{sock}, read => sub { $self->new_client }  );
449         return $self;
450 }
451
452
453 sub nolinger
454 {
455         my $conn = shift;
456
457         unless ($main::is_win) {
458                 if (isdbg('sock')) {
459                         my ($l, $t) = unpack "ll", getsockopt($conn->{sock}, SOL_SOCKET, SO_LINGER); 
460                         my $k = unpack 'l', getsockopt($conn->{sock}, SOL_SOCKET, SO_KEEPALIVE);
461                         my $n = $main::is_win ? 0 : unpack "l", getsockopt($conn->{sock}, IPPROTO_TCP, TCP_NODELAY);
462                         dbg("Linger is: $l $t, keepalive: $k, nagle: $n");
463                 }
464                 
465                 eval {setsockopt($conn->{sock}, SOL_SOCKET, SO_KEEPALIVE, 1)} or dbg("setsockopt keepalive: $!");
466                 eval {setsockopt($conn->{sock}, SOL_SOCKET, SO_LINGER, pack("ll", 0, 0))} or dbg("setsockopt linger: $!");
467                 eval {setsockopt($conn->{sock}, IPPROTO_TCP, TCP_NODELAY, 1)} or eval {setsockopt($conn->{sock}, SOL_SOCKET, TCP_NODELAY, 1)} or dbg("setsockopt tcp_nodelay: $!");
468                 $conn->{sock}->autoflush(0);
469
470                 if (isdbg('sock')) {
471                         my ($l, $t) = unpack "ll", getsockopt($conn->{sock}, SOL_SOCKET, SO_LINGER); 
472                         my $k = unpack 'l', getsockopt($conn->{sock}, SOL_SOCKET, SO_KEEPALIVE);
473                         my $n = $main::is_win ? 0 : unpack "l", getsockopt($conn->{sock}, IPPROTO_TCP, TCP_NODELAY);
474                         dbg("Linger is: $l $t, keepalive: $k, nagle: $n");
475                 }
476         } 
477 }
478
479 sub dequeue
480 {
481         my $conn = shift;
482
483         if ($conn->{msg} =~ /\n/) {
484                 my @lines = split /\r?\n/, $conn->{msg};
485                 if ($conn->{msg} =~ /\n$/) {
486                         delete $conn->{msg};
487                 } else {
488                         $conn->{msg} = pop @lines;
489                 }
490                 for (@lines) {
491                         &{$conn->{rproc}}($conn, defined $_ ? $_ : '');
492                 }
493         }
494 }
495
496 sub _rcv {                     # Complement to _send
497     my $conn = shift; # $rcv_now complement of $flush
498     # Find out how much has already been received, if at all
499     my ($msg, $offset, $bytes_to_read, $bytes_read);
500     my $sock = $conn->{sock};
501     return unless defined($sock);
502
503         my @lines;
504 #       if ($conn->{blocking}) {
505 #               blocking($sock, 0);
506 #               $conn->{blocking} = 0;
507 #       }
508         $bytes_read = sysread ($sock, $msg, 1024, 0);
509         if (defined ($bytes_read)) {
510                 if ($bytes_read > 0) {
511                         $total_in += $bytes_read;
512                         if (isdbg('raw')) {
513                                 my $call = $conn->{call} || 'none';
514                                 dbgdump('raw', "$call read $bytes_read: ", $msg);
515                         }
516                         if ($conn->{echo}) {
517                                 my @ch = split //, $msg;
518                                 my $out;
519                                 for (@ch) {
520                                         if (/[\cH\x7f]/) {
521                                                 $out .= "\cH \cH";
522                                                 $conn->{msg} =~ s/.$//;
523                                         } else {
524                                                 $out .= $_;
525                                                 $conn->{msg} .= $_;
526                                         }
527                                 }
528                                 if (defined $out) {
529                                         set_event_handler ($sock, write => sub{$conn->_send(0)});
530                                         push @{$conn->{outqueue}}, $out;
531                                 }
532                         } else {
533                                 $conn->{msg} .= $msg;
534                         }
535                 } 
536         } else {
537                 if (_err_will_block($!)) {
538                         return ; 
539                 } else {
540                         $bytes_read = 0;
541                 }
542     }
543
544 FINISH:
545     if (defined $bytes_read && $bytes_read == 0) {
546                 &{$conn->{eproc}}($conn, $!) if exists $conn->{eproc};
547                 $conn->disconnect;
548     } else {
549                 unless ($conn->{disable_read}) {
550                         $conn->dequeue if exists $conn->{msg};
551                 }
552         }
553 }
554
555 sub new_client {
556         my $server_conn = shift;
557     my $sock = $server_conn->{sock}->accept();
558         if ($sock) {
559                 my $conn = $server_conn->new($server_conn->{rproc});
560                 $conn->{sock} = $sock;
561                 blocking($sock, 0);
562                 $conn->nolinger;
563                 $conn->{blocking} = 0;
564                 my ($rproc, $eproc) = &{$server_conn->{rproc}} ($conn, $conn->{peerhost} = $sock->peerhost(), $conn->{peerport} = $sock->peerport());
565                 $conn->{sort} = 'Incoming';
566                 if ($eproc) {
567                         $conn->{eproc} = $eproc;
568                         set_event_handler ($sock, error => $eproc);
569                 }
570                 if ($rproc) {
571                         $conn->{rproc} = $rproc;
572                         my $callback = sub {$conn->_rcv};
573                         set_event_handler ($sock, read => $callback);
574                 } else {  # Login failed
575                         &{$conn->{eproc}}($conn, undef) if exists $conn->{eproc};
576                         $conn->disconnect();
577                 }
578         } else {
579                 dbg("Msg: error on accept ($!)") if isdbg('err');
580         }
581 }
582
583 sub close_server
584 {
585         my $conn = shift;
586         set_event_handler ($conn->{sock}, read => undef, write => undef, error => undef );
587         $conn->{sock}->close;
588 }
589
590 # close all clients (this is for forking really)
591 sub close_all_clients
592 {
593         foreach my $conn (values %conns) {
594                 $conn->disconnect;
595         }
596 }
597
598 sub disable_read
599 {
600         my $conn = shift;
601         set_event_handler ($conn->{sock}, read => undef);
602         return $_[0] ? $conn->{disable_read} = $_[0] : $_[0];
603 }
604
605 #
606 #----------------------------------------------------
607 # Event loop routines used by both client and server
608
609 sub set_event_handler {
610     shift unless ref($_[0]); # shift if first arg is package name
611     my ($handle, %args) = @_;
612     my $callback;
613     if (exists $args{'write'}) {
614         $callback = $args{'write'};
615         if ($callback) {
616             $wt_callbacks{$handle} = $callback;
617             $wt_handles->add($handle);
618         } else {
619             delete $wt_callbacks{$handle};
620             $wt_handles->remove($handle);
621         }
622     }
623     if (exists $args{'read'}) {
624         $callback = $args{'read'};
625         if ($callback) {
626             $rd_callbacks{$handle} = $callback;
627             $rd_handles->add($handle);
628         } else {
629             delete $rd_callbacks{$handle};
630             $rd_handles->remove($handle);
631        }
632     }
633     if (exists $args{'error'}) {
634         $callback = $args{'error'};
635         if ($callback) {
636             $er_callbacks{$handle} = $callback;
637             $er_handles->add($handle);
638         } else {
639             delete $er_callbacks{$handle};
640             $er_handles->remove($handle);
641        }
642     }
643 }
644
645 sub event_loop {
646     my ($pkg, $loop_count, $timeout, $wronly) = @_; # event_loop(1) to process events once
647     my ($conn, $r, $w, $e, $rset, $wset, $eset);
648     while (1) {
649  
650        # Quit the loop if no handles left to process
651                 if ($wronly) {
652                         last unless $wt_handles->count();
653         
654                         ($rset, $wset, $eset) = IO::Select->select(undef, $wt_handles, undef, $timeout);
655                         
656                         foreach $w (@$wset) {
657                                 &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w};
658                         }
659                 } else {
660                         
661                         last unless ($rd_handles->count() || $wt_handles->count());
662         
663                         ($rset, $wset, $eset) = IO::Select->select($rd_handles, $wt_handles, $er_handles, $timeout);
664                         
665                         foreach $e (@$eset) {
666                                 &{$er_callbacks{$e}}($e) if exists $er_callbacks{$e};
667                         }
668                         foreach $r (@$rset) {
669                                 &{$rd_callbacks{$r}}($r) if exists $rd_callbacks{$r};
670                         }
671                         foreach $w (@$wset) {
672                                 &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w};
673                         }
674                 }
675
676                 Timer::handler;
677                 
678         if (defined($loop_count)) {
679             last unless --$loop_count;
680         }
681     }
682 }
683
684 sub sleep
685 {
686         my ($pkg, $interval) = @_;
687         my $now = time;
688         while (time - $now < $interval) {
689                 $pkg->event_loop(10, 0.01);
690         }
691 }
692
693 sub DESTROY
694 {
695         my $conn = shift;
696         my $call = $conn->{call} || 'unallocated';
697         my $host = $conn->{peerhost} || '';
698         my $port = $conn->{peerport} || '';
699         dbg("Connection $conn->{cnum} $call [$host $port] being destroyed") if isdbg('connll');
700         $noconns--;
701 }
702
703 1;
704
705 __END__
706