src/perl/lib/OpenSRF/Server.pm

   1 # ----------------------------------------------------------------
   2 # Copyright (C) 2010 Equinox Software, Inc.
   3 # Bill Erickson <erickson@esilibrary.com>
   4 #
   5 # This program is free software; you can redistribute it and/or
   6 # modify it under the terms of the GNU General Public License
   7 # as published by the Free Software Foundation; either version 2
   8 # of the License, or (at your option) any later version.
   9 #
  10 # This program is distributed in the hope that it will be useful,
  11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 # GNU General Public License for more details.
  14 # ----------------------------------------------------------------
  15 package OpenSRF::Server;
  16 use strict;
  17 use warnings;
  18 use OpenSRF::Transport;
  19 use OpenSRF::Application;
  20 use OpenSRF::Utils::Config;
  21 use OpenSRF::Transport::PeerHandle;
  22 use OpenSRF::Utils::SettingsClient;
  23 use OpenSRF::Utils::Logger qw($logger);
  24 use OpenSRF::Transport::SlimJabber::Client;
  25 use Encode;
  26 use POSIX qw/:sys_wait_h :errno_h/;
  27 use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK);
  28 use Time::HiRes qw/usleep/;
  29 use IO::Select;
  30 use Socket;
  31 our $chatty = 1; # disable for production
  32
  33 use constant STATUS_PIPE_DATA_SIZE => 12;
  34 use constant WRITE_PIPE_DATA_SIZE  => 12;
  35
  36 sub new {
  37     my($class, $service, %args) = @_;
  38     my $self = bless(\%args, $class);
  39
  40     $self->{service}        = $service; # service name
  41     $self->{num_children}   = 0; # number of child processes
  42     $self->{osrf_handle}    = undef; # xmpp handle
  43     $self->{routers}        = []; # list of registered routers
  44     $self->{active_list}    = []; # list of active children
  45     $self->{idle_list}      = []; # list of idle children
  46     $self->{sighup_pending} = [];
  47     $self->{pid_map}        = {}; # map of child pid to child for cleaner access
  48     $self->{sig_pipe}       = 0;  # true if last syswrite failed
  49
  50     $self->{stderr_log} = $self->{stderr_log_path} . "/${service}_stderr.log"
  51         if $self->{stderr_log_path};
  52
  53     $self->{min_spare_children} ||= 0;
  54
  55     $self->{max_spare_children} = $self->{min_spare_children} + 1 if
  56         $self->{max_spare_children} and
  57         $self->{max_spare_children} <= $self->{min_spare_children};
  58
  59     return $self;
  60 }
  61
  62 # ----------------------------------------------------------------
  63 # Disconnects from routers and waits for child processes to exit.
  64 # ----------------------------------------------------------------
  65 sub cleanup {
  66     my $self = shift;
  67     my $no_exit = shift;
  68     my $graceful = shift;
  69
  70     $logger->info("server: shutting down and cleaning up...");
  71
  72     # de-register routers
  73     $self->unregister_routers;
  74
  75     if ($graceful) {
  76         # graceful shutdown waits for all active
  77         # children to complete their in-process tasks.
  78
  79         while (@{$self->{active_list}}) {
  80             $logger->info("server: graceful shutdown with ".
  81                 @{$self->{active_list}}." active children...");
  82
  83             # block until a child is becomes available
  84             $self->check_status(1);
  85         }
  86         $logger->info("server: all clear for graceful shutdown");
  87     }
  88
  89     # don't get sidetracked by signals while we're cleaning up.
  90     # it could result in unexpected behavior with list traversal
  91     $SIG{CHLD} = 'IGNORE';
  92
  93     # terminate the child processes
  94     $self->kill_child($_) for
  95         (@{$self->{idle_list}}, @{$self->{active_list}});
  96
  97     $self->{osrf_handle}->disconnect;
  98
  99     # clean up our dead children
 100     $self->reap_children(1);
 101
 102     exit(0) unless $no_exit;
 103 }
 104
 105 # ----------------------------------------------------------------
 106 # SIGHUP handler.  Kill all idle children.  Copy list of active
 107 # children into sighup_pending list for later cleanup.
 108 # ----------------------------------------------------------------
 109 sub handle_sighup {
 110     my $self = shift;
 111     $logger->info("server: caught SIGHUP; reloading children");
 112
 113     # reload the opensrf config
 114     # note: calling ::Config->load() results in ever-growing
 115     # package names, which eventually causes an exception
 116     OpenSRF::Utils::Config->current->_load(
 117         force => 1,
 118         config_file => OpenSRF::Utils::Config->current->FILE
 119     );
 120
 121     # force-reload the logger config
 122     OpenSRF::Utils::Logger::set_config(1);
 123
 124     # copy active list into pending list for later cleanup
 125     $self->{sighup_pending} = [ @{$self->{active_list}} ];
 126
 127     # idle_list will be modified as children are reaped.
 128     my @idle = @{$self->{idle_list}};
 129
 130     # idle children are the reaper's plaything
 131     $self->kill_child($_) for @idle;
 132 }
 133
 134 # ----------------------------------------------------------------
 135 # Waits on the jabber socket for inbound data from the router.
 136 # Each new message is passed off to a child process for handling.
 137 # At regular intervals, wake up for min/max spare child maintenance
 138 # ----------------------------------------------------------------
 139 sub run {
 140     my $self = shift;
 141
 142     $logger->set_service($self->{service});
 143
 144     $SIG{$_} = sub { $self->cleanup; } for (qw/INT QUIT/);
 145     $SIG{TERM} = sub { $self->cleanup(0, 1); };
 146     $SIG{CHLD} = sub { $self->reap_children(); };
 147     $SIG{HUP} = sub { $self->handle_sighup(); };
 148     $SIG{USR1} = sub { $self->unregister_routers; };
 149     $SIG{USR2} = sub { $self->register_routers; };
 150
 151     $self->spawn_children;
 152     $self->build_osrf_handle;
 153     $self->register_routers;
 154     my $wait_time = 1;
 155
 156     # main server loop
 157     while(1) {
 158
 159         $self->check_status;
 160         $self->{child_died} = 0;
 161
 162         my $msg = $self->{osrf_handle}->process($wait_time);
 163
 164         # we woke up for any reason, reset the wait time to allow
 165         # for idle maintenance as necessary
 166         $wait_time = 1;
 167
 168         if($msg) {
 169
 170             if ($msg->type and $msg->type eq 'error') {
 171                 $logger->info("server: Listener received an XMPP error ".
 172                     "message.  Likely a bounced message. sender=".$msg->from);
 173
 174             } elsif(my $child = pop(@{$self->{idle_list}})) {
 175
 176                 # we have an idle child to handle the request
 177                 $chatty and $logger->internal("server: passing request to idle child $child");
 178                 push(@{$self->{active_list}}, $child);
 179                 $self->write_child($child, $msg);
 180
 181             } elsif($self->{num_children} < $self->{max_children}) {
 182
 183                 # spawning a child to handle the request
 184                 $chatty and $logger->internal("server: spawning child to handle request");
 185                 $self->write_child($self->spawn_child(1), $msg);
 186
 187             } else {
 188                 $logger->warn("server: no children available, waiting... consider increasing " .
 189                     "max_children for this application higher than $self->{max_children} ".
 190                     "in the OpenSRF configuration if this message occurs frequently");
 191                 $self->check_status(1); # block until child is available
 192
 193                 my $child = pop(@{$self->{idle_list}});
 194                 push(@{$self->{active_list}}, $child);
 195                 $self->write_child($child, $msg);
 196             }
 197
 198         } else {
 199
 200             # don't perform idle maint immediately when woken by SIGCHLD
 201             unless($self->{child_died}) {
 202
 203                 # when we hit equilibrium, there's no need for regular
 204                 # maintenance, so set wait_time to 'forever'
 205                 $wait_time = -1 if
 206                     !$self->perform_idle_maintenance and # no maintenance performed this time
 207                     @{$self->{active_list}} == 0; # no active children
 208             }
 209         }
 210     }
 211 }
 212
 213 # ----------------------------------------------------------------
 214 # Launch a new spare child or kill an extra spare child.  To
 215 # prevent large-scale spawning or die-offs, spawn or kill only
 216 # 1 process per idle maintenance loop.
 217 # Returns true if any idle maintenance occurred, 0 otherwise
 218 # ----------------------------------------------------------------
 219 sub perform_idle_maintenance {
 220     my $self = shift;
 221
 222     $chatty and $logger->internal(
 223         sprintf(
 224             "server: %d idle, %d active, %d min_spare, %d max_spare in idle maintenance",
 225             scalar(@{$self->{idle_list}}),
 226             scalar(@{$self->{active_list}}),
 227             $self->{min_spare_children},
 228             $self->{max_spare_children}
 229         )
 230     );
 231
 232     # spawn 1 spare child per maintenance loop if necessary
 233     if( $self->{min_spare_children} and
 234         $self->{num_children} < $self->{max_children} and
 235         scalar(@{$self->{idle_list}}) < $self->{min_spare_children} ) {
 236
 237         $chatty and $logger->internal("server: spawning spare child");
 238         $self->spawn_child;
 239         return 1;
 240
 241     # kill 1 excess spare child per maintenance loop if necessary
 242     } elsif($self->{max_spare_children} and
 243             $self->{num_children} > $self->{min_children} and
 244             scalar(@{$self->{idle_list}}) > $self->{max_spare_children} ) {
 245
 246         $chatty and $logger->internal("server: killing spare child");
 247         $self->kill_child;
 248         return 1;
 249     }
 250
 251     return 0;
 252 }
 253
 254 sub kill_child {
 255     my $self = shift;
 256     my $child = shift || pop(@{$self->{idle_list}}) or return;
 257     $chatty and $logger->internal("server: killing child $child");
 258     kill('TERM', $child->{pid});
 259 }
 260
 261 # ----------------------------------------------------------------
 262 # Jabber connection inbound message arrive on.
 263 # ----------------------------------------------------------------
 264 sub build_osrf_handle {
 265     my $self = shift;
 266
 267     my $conf = OpenSRF::Utils::Config->current;
 268     my $username = $conf->bootstrap->username;
 269     my $password = $conf->bootstrap->passwd;
 270     my $domain = $conf->bootstrap->domain;
 271     my $port = $conf->bootstrap->port;
 272     my $resource = $self->{service} . '_listener_' . $conf->env->hostname;
 273
 274     $logger->debug("server: inbound connecting as $username\@$domain/$resource on port $port");
 275
 276     $self->{osrf_handle} =
 277         OpenSRF::Transport::SlimJabber::Client->new(
 278             username => $username,
 279             resource => $resource,
 280             password => $password,
 281             host => $domain,
 282             port => $port,
 283         );
 284
 285     $self->{osrf_handle}->initialize;
 286 }
 287
 288
 289 # ----------------------------------------------------------------
 290 # Sends request data to a child process
 291 # ----------------------------------------------------------------
 292 sub write_child {
 293     my($self, $child, $msg) = @_;
 294     my $xml = encode_utf8(decode_utf8($msg->to_xml));
 295
 296     # tell the child how much data to expect, minus the header
 297     my $write_size;
 298     {use bytes; $write_size = length($xml)}
 299     $write_size = sprintf("%*s", WRITE_PIPE_DATA_SIZE, $write_size);
 300
 301     for (0..2) {
 302
 303         $self->{sig_pipe} = 0;
 304         local $SIG{'PIPE'} = sub { $self->{sig_pipe} = 1; };
 305
 306         # send message to child data pipe
 307         syswrite($child->{pipe_to_child}, $write_size . $xml);
 308
 309         last unless $self->{sig_pipe};
 310         $logger->error("server: got SIGPIPE writing to $child, retrying...");
 311         usleep(50000); # 50 msec
 312     }
 313
 314     $logger->error("server: unable to send request message to child $child") if $self->{sig_pipe};
 315 }
 316
 317 # ----------------------------------------------------------------
 318 # Checks to see if any child process has reported its availability
 319 # In blocking mode, blocks until a child has reported.
 320 # ----------------------------------------------------------------
 321 sub check_status {
 322     my($self, $block) = @_;
 323
 324     return unless @{$self->{active_list}};
 325
 326     my @pids;
 327
 328     while (1) {
 329
 330         # if can_read or sysread is interrupted while bloking, go back and
 331         # wait again until we have at least 1 free child
 332
 333         # refresh the read_set handles in case we lost a child in the previous iteration
 334         my $read_set = IO::Select->new;
 335         $read_set->add($_->{pipe_to_child}) for @{$self->{active_list}};
 336
 337         if(my @handles = $read_set->can_read(($block) ? undef : 0)) {
 338             my $pid = '';
 339             for my $pipe (@handles) {
 340                 sysread($pipe, $pid, STATUS_PIPE_DATA_SIZE) or next;
 341                 push(@pids, int($pid));
 342             }
 343         }
 344
 345         last unless $block and !@pids;
 346     }
 347
 348     return unless @pids;
 349
 350     $chatty and $logger->internal("server: ".scalar(@pids)." children reporting for duty: (@pids)");
 351
 352     my $child;
 353     my @new_actives;
 354
 355     # move the children from the active list to the idle list
 356     for my $proc (@{$self->{active_list}}) {
 357         if(grep { $_ == $proc->{pid} } @pids) {
 358             push(@{$self->{idle_list}}, $proc);
 359         } else {
 360             push(@new_actives, $proc);
 361         }
 362     }
 363
 364     $self->{active_list} = [@new_actives];
 365
 366     $chatty and $logger->internal(sprintf(
 367         "server: %d idle and %d active children after status update",
 368             scalar(@{$self->{idle_list}}), scalar(@{$self->{active_list}})));
 369
 370     # some children just went from active to idle. let's see
 371     # if any of them need to be killed from a previous sighup.
 372
 373     for my $child (@{$self->{sighup_pending}}) {
 374         if (grep {$_ == $child->{pid}} @pids) {
 375
 376             $chatty and $logger->internal(
 377                 "server: killing previously-active ".
 378                 "child after receiving SIGHUP: $child");
 379
 380             # remove the pending child
 381             $self->{sighup_pending} = [
 382                 grep {$_->{pid} != $child->{pid}}
 383                     @{$self->{sighup_pending}}
 384             ];
 385
 386             # kill the pending child
 387             $self->kill_child($child)
 388         }
 389     }
 390 }
 391
 392 # ----------------------------------------------------------------
 393 # Cleans up any child processes that have exited.
 394 # In shutdown mode, block until all children have washed ashore
 395 # ----------------------------------------------------------------
 396 sub reap_children {
 397     my($self, $shutdown) = @_;
 398     $self->{child_died} = 1;
 399
 400     while(1) {
 401
 402         my $pid = waitpid(-1, ($shutdown) ? 0 : WNOHANG);
 403         last if $pid <= 0;
 404
 405         $chatty and $logger->internal("server: reaping child $pid");
 406
 407         my $child = $self->{pid_map}->{$pid};
 408
 409         close($child->{pipe_to_parent});
 410         close($child->{pipe_to_child});
 411
 412         $self->{active_list} = [ grep { $_->{pid} != $pid } @{$self->{active_list}} ];
 413         $self->{idle_list} = [ grep { $_->{pid} != $pid } @{$self->{idle_list}} ];
 414
 415         $self->{num_children}--;
 416         delete $self->{pid_map}->{$pid};
 417         delete $child->{$_} for keys %$child; # destroy with a vengeance
 418     }
 419
 420     $self->spawn_children unless $shutdown;
 421
 422     $chatty and $logger->internal(sprintf(
 423         "server: %d idle and %d active children after reap_children",
 424             scalar(@{$self->{idle_list}}), scalar(@{$self->{active_list}})));
 425 }
 426
 427 # ----------------------------------------------------------------
 428 # Spawn up to max_children processes
 429 # ----------------------------------------------------------------
 430 sub spawn_children {
 431     my $self = shift;
 432     $self->spawn_child while $self->{num_children} < $self->{min_children};
 433 }
 434
 435 # ----------------------------------------------------------------
 436 # Spawns a new child.  If $active is set, the child goes directly
 437 # into the active_list.
 438 # ----------------------------------------------------------------
 439 sub spawn_child {
 440     my($self, $active) = @_;
 441
 442     my $child = OpenSRF::Server::Child->new($self);
 443
 444     # socket for sending message data to the child
 445     if(!socketpair(
 446         $child->{pipe_to_child},
 447         $child->{pipe_to_parent},
 448         AF_UNIX, SOCK_STREAM, PF_UNSPEC)) {
 449             $logger->error("server: error creating data socketpair: $!");
 450             return undef;
 451     }
 452
 453     $child->{pipe_to_child}->autoflush(1);
 454     $child->{pipe_to_parent}->autoflush(1);
 455
 456     $child->{pid} = fork();
 457
 458     if($child->{pid}) { # parent process
 459         $self->{num_children}++;
 460         $self->{pid_map}->{$child->{pid}} = $child;
 461
 462         if($active) {
 463             push(@{$self->{active_list}}, $child);
 464         } else {
 465             push(@{$self->{idle_list}}, $child);
 466         }
 467
 468         $chatty and $logger->internal("server: server spawned child $child with ".$self->{num_children}." total children");
 469
 470         return $child;
 471
 472     } else { # child process
 473
 474         # recover default handling for any signal whose handler
 475         # may have been adopted from the parent process.
 476         $SIG{$_} = 'DEFAULT' for qw/TERM INT QUIT HUP CHLD USR1 USR2/;
 477
 478         if($self->{stderr_log}) {
 479
 480             $chatty and $logger->internal("server: redirecting STDERR to " . $self->{stderr_log});
 481
 482             close STDERR;
 483             unless( open(STDERR, '>>' . $self->{stderr_log}) ) {
 484                 $logger->error("server: unable to open STDERR log file: " . $self->{stderr_log} . " : $@");
 485                 open STDERR, '>/dev/null'; # send it back to /dev/null
 486             }
 487         }
 488
 489         $child->{pid} = $$;
 490         eval {
 491             $child->init;
 492             $child->run;
 493             OpenSRF::Transport::PeerHandle->retrieve->disconnect;
 494         };
 495         $logger->error("server: child process died: $@") if $@;
 496         exit(0);
 497     }
 498 }
 499
 500 # ----------------------------------------------------------------
 501 # Sends the register command to the configured routers
 502 # ----------------------------------------------------------------
 503 sub register_routers {
 504     my $self = shift;
 505
 506     my $conf = OpenSRF::Utils::Config->current;
 507     my $routers = $conf->bootstrap->routers;
 508     my $router_name = $conf->bootstrap->router_name;
 509     my @targets;
 510
 511     for my $router (@$routers) {
 512         if(ref $router) {
 513
 514             if( !$router->{services} ||
 515                 !$router->{services}->{service} ||
 516                 (
 517                     ref($router->{services}->{service}) eq 'ARRAY' and
 518                     grep { $_ eq $self->{service} } @{$router->{services}->{service}}
 519                 )  || $router->{services}->{service} eq $self->{service}) {
 520
 521                 my $name = $router->{name};
 522                 my $domain = $router->{domain};
 523                 push(@targets, "$name\@$domain/router");
 524             }
 525
 526         } else {
 527             push(@targets, "$router_name\@$router/router");
 528         }
 529     }
 530
 531     foreach (@targets) {
 532         $logger->info("server: registering with router $_");
 533         $self->{osrf_handle}->send(
 534             to => $_,
 535             body => 'registering',
 536             router_command => 'register',
 537             router_class => $self->{service}
 538         );
 539     }
 540
 541     $self->{routers} = \@targets;
 542 }
 543
 544 # ----------------------------------------------------------------
 545 # Sends the unregister command to any routers we have registered
 546 # with.
 547 # ----------------------------------------------------------------
 548 sub unregister_routers {
 549     my $self = shift;
 550     return unless $self->{osrf_handle}->tcp_connected;
 551
 552     for my $router (@{$self->{routers}}) {
 553         $logger->info("server: disconnecting from router $router");
 554         $self->{osrf_handle}->send(
 555             to => $router,
 556             body => "unregistering",
 557             router_command => "unregister",
 558             router_class => $self->{service}
 559         );
 560     }
 561 }
 562
 563
 564 package OpenSRF::Server::Child;
 565 use strict;
 566 use warnings;
 567 use OpenSRF::Transport;
 568 use OpenSRF::Application;
 569 use OpenSRF::Transport::PeerHandle;
 570 use OpenSRF::Transport::SlimJabber::XMPPMessage;
 571 use OpenSRF::Utils::Logger qw($logger);
 572 use OpenSRF::DomainObject::oilsResponse qw/:status/;
 573 use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK);
 574 use Time::HiRes qw(time usleep);
 575 use POSIX qw/:sys_wait_h :errno_h/;
 576
 577 use overload '""' => sub { return '[' . shift()->{pid} . ']'; };
 578
 579 sub new {
 580     my($class, $parent) = @_;
 581     my $self = bless({}, $class);
 582     $self->{pid} = 0; # my process ID
 583     $self->{parent} = $parent; # Controller parent process
 584     $self->{num_requests} = 0; # total serviced requests
 585     $self->{sig_pipe} = 0;  # true if last syswrite failed
 586     return $self;
 587 }
 588
 589 sub set_nonblock {
 590     my($self, $fh) = @_;
 591     my  $flags = fcntl($fh, F_GETFL, 0);
 592     fcntl($fh, F_SETFL, $flags | O_NONBLOCK);
 593 }
 594
 595 sub set_block {
 596     my($self, $fh) = @_;
 597     my  $flags = fcntl($fh, F_GETFL, 0);
 598     $flags &= ~O_NONBLOCK;
 599     fcntl($fh, F_SETFL, $flags);
 600 }
 601
 602 # ----------------------------------------------------------------
 603 # Connects to Jabber and runs the application child_init
 604 # ----------------------------------------------------------------
 605 sub init {
 606     my $self = shift;
 607     my $service = $self->{parent}->{service};
 608     $0 = "OpenSRF Drone [$service]";
 609     OpenSRF::Transport::PeerHandle->construct($service);
 610     OpenSRF::Application->application_implementation->child_init
 611         if (OpenSRF::Application->application_implementation->can('child_init'));
 612 }
 613
 614 # ----------------------------------------------------------------
 615 # Waits for messages from the parent process, handles the message,
 616 # then goes into the keepalive loop if this is a stateful session.
 617 # When max_requests is hit, the process exits.
 618 # ----------------------------------------------------------------
 619 sub run {
 620     my $self = shift;
 621     my $network = OpenSRF::Transport::PeerHandle->retrieve;
 622
 623     # main child run loop.  Ends when this child hits max requests.
 624     while(1) {
 625
 626         my $data = $self->wait_for_request or next;
 627
 628         # Update process name to show activity
 629         my $orig_name = $0;
 630         $0 = "$0*";
 631
 632         # Discard extraneous data from the jabber socket
 633         if(!$network->flush_socket()) {
 634             $logger->error("server: network disconnected!  child dropping request and exiting: $data");
 635             exit;
 636         }
 637
 638         my $session = OpenSRF::Transport->handler(
 639             $self->{parent}->{service},
 640             OpenSRF::Transport::SlimJabber::XMPPMessage->new(xml => $data)
 641         );
 642
 643         $self->keepalive_loop($session);
 644
 645         last if ++$self->{num_requests} == $self->{parent}->{max_requests};
 646
 647         # Tell the parent process we are available to process requests
 648         $self->send_status;
 649
 650         # Repair process name
 651         $0 = $orig_name;
 652     }
 653
 654     $chatty and $logger->internal("server: child process shutting down after reaching max_requests");
 655
 656     OpenSRF::Application->application_implementation->child_exit
 657         if (OpenSRF::Application->application_implementation->can('child_exit'));
 658 }
 659
 660 # ----------------------------------------------------------------
 661 # waits for a request data on the parent pipe and returns it.
 662 # ----------------------------------------------------------------
 663 sub wait_for_request {
 664     my $self = shift;
 665
 666     my $data = ''; # final request data
 667     my $buf_size = 4096; # default linux pipe_buf (atomic window, not total size)
 668     my $read_pipe = $self->{pipe_to_parent};
 669     my $bytes_needed; # size of the data we are about to receive
 670     my $bytes_recvd; # number of bytes read so far
 671     my $first_read = 1; # true for first loop iteration
 672     my $read_error;
 673
 674     while (1) {
 675
 676         # wait for some data to start arriving
 677         my $read_set = IO::Select->new;
 678         $read_set->add($read_pipe);
 679
 680         while (1) {
 681             # if can_read is interrupted while blocking,
 682             # go back and wait again until it succeeds.
 683             last if $read_set->can_read;
 684         }
 685
 686         # parent started writing, let's start reading
 687         $self->set_nonblock($read_pipe);
 688
 689         while (1) {
 690             # read all of the available data
 691
 692             my $buf = '';
 693             my $nbytes = sysread($self->{pipe_to_parent}, $buf, $buf_size);
 694
 695             unless(defined $nbytes) {
 696                 if ($! != EAGAIN) {
 697                     $logger->error("server: error reading data from parent: $!.  ".
 698                         "bytes_needed=$bytes_needed; bytes_recvd=$bytes_recvd; data=$data");
 699                     $read_error = 1;
 700                 }
 701                 last;
 702             }
 703
 704             last if $nbytes <= 0; # no more data available for reading
 705
 706             $bytes_recvd += $nbytes;
 707             $data .= $buf;
 708         }
 709
 710         $self->set_block($self->{pipe_to_parent});
 711         return undef if $read_error;
 712
 713         # extract the data size and remove the header from the final data
 714         if ($first_read) {
 715             my $wps_size = OpenSRF::Server::WRITE_PIPE_DATA_SIZE;
 716             $bytes_needed = int(substr($data, 0, $wps_size)) + $wps_size;
 717             $data = substr($data, $wps_size);
 718             $first_read = 0;
 719         }
 720
 721
 722         if ($bytes_recvd == $bytes_needed) {
 723             # we've read all the data. Nothing left to do
 724             last;
 725         }
 726
 727         $logger->info("server: child process read all available pipe data.  ".
 728             "waiting for more data from parent.  bytes_needed=$bytes_needed; bytes_recvd=$bytes_recvd");
 729     }
 730
 731     return $data;
 732 }
 733
 734
 735 # ----------------------------------------------------------------
 736 # If this is a stateful opensrf session, wait up to $keepalive
 737 # seconds for subsequent requests from the client
 738 # ----------------------------------------------------------------
 739 sub keepalive_loop {
 740     my($self, $session) = @_;
 741     my $keepalive = $self->{parent}->{keepalive};
 742
 743     while($session->state and $session->state == $session->CONNECTED) {
 744
 745         unless( $session->queue_wait($keepalive) ) {
 746
 747             # client failed to disconnect before timeout
 748             $logger->info("server: no request was received in $keepalive seconds, exiting stateful session");
 749
 750             my $res = OpenSRF::DomainObject::oilsConnectStatus->new(
 751                 status => "Disconnected on timeout",
 752                 statusCode => STATUS_TIMEOUT
 753             );
 754
 755             $session->status($res);
 756             $session->state($session->DISCONNECTED);
 757             last;
 758         }
 759     }
 760
 761     $chatty and $logger->internal("server: child done with request(s)");
 762     $session->kill_me;
 763 }
 764
 765 # ----------------------------------------------------------------
 766 # Report our availability to our parent process
 767 # ----------------------------------------------------------------
 768 sub send_status {
 769     my $self = shift;
 770
 771     for (0..2) {
 772
 773         $self->{sig_pipe} = 0;
 774         local $SIG{'PIPE'} = sub { $self->{sig_pipe} = 1; };
 775
 776         syswrite(
 777             $self->{pipe_to_parent},
 778             sprintf("%*s", OpenSRF::Server::STATUS_PIPE_DATA_SIZE, $self->{pid})
 779         );
 780
 781         last unless $self->{sig_pipe};
 782         $logger->error("server: $self got SIGPIPE writing status to parent, retrying...");
 783         usleep(50000); # 50 msec
 784     }
 785
 786     $logger->error("server: $self unable to send status to parent") if $self->{sig_pipe};
 787 }
 788
 789
 790 1;