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