Resolve encoding problem found by attempting to register patrons with Unicode names
[OpenSRF.git] / 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 IO::Select;
29 use Socket;
30 our $chatty = 1; # disable for production
31
32 use constant STATUS_PIPE_DATA_SIZE => 12;
33
34 sub new {
35     my($class, $service, %args) = @_;
36     my $self = bless(\%args, $class);
37
38     $self->{service}        = $service; # service name
39     $self->{num_children}   = 0; # number of child processes
40     $self->{osrf_handle}    = undef; # xmpp handle
41     $self->{routers}        = []; # list of registered routers
42     $self->{active_list}    = []; # list of active children
43     $self->{idle_list}      = []; # list of idle children
44     $self->{pid_map}        = {}; # map of child pid to child for cleaner access
45
46     $self->{stderr_log} = $self->{stderr_log_path} . "/${service}_stderr.log" 
47         if $self->{stderr_log_path};
48
49     $self->{min_spare_children} ||= 0;
50
51     $self->{max_spare_children} = $self->{min_spare_children} + 1 if
52         $self->{max_spare_children} and
53         $self->{max_spare_children} <= $self->{min_spare_children};
54
55     return $self;
56 }
57
58 # ----------------------------------------------------------------
59 # Disconnects from routers and waits for child processes to exit.
60 # ----------------------------------------------------------------
61 sub cleanup {
62     my $self = shift;
63     my $no_exit = shift;
64
65     $logger->info("server: shutting down and cleaning up...");
66
67     # don't get sidetracked by signals while we're cleaning up.
68     # it could result in unexpected behavior with list traversal
69     $SIG{CHLD} = 'IGNORE';
70
71     # terminate the child processes
72     $self->kill_child($_) for
73         (@{$self->{idle_list}}, @{$self->{active_list}});
74
75     # de-register routers
76     $self->unregister_routers;
77
78     $self->{osrf_handle}->disconnect;
79
80     # clean up our dead children
81     $self->reap_children(1);
82
83     exit(0) unless $no_exit;
84 }
85
86
87 # ----------------------------------------------------------------
88 # Waits on the jabber socket for inbound data from the router.
89 # Each new message is passed off to a child process for handling.
90 # At regular intervals, wake up for min/max spare child maintenance
91 # ----------------------------------------------------------------
92 sub run {
93     my $self = shift;
94
95         $logger->set_service($self->{service});
96
97     $SIG{$_} = sub { $self->cleanup; } for (qw/INT TERM QUIT/);
98     $SIG{CHLD} = sub { $self->reap_children(); };
99
100     $self->spawn_children;
101     $self->build_osrf_handle;
102     $self->register_routers;
103     my $wait_time = 1;
104
105     # main server loop
106     while(1) {
107
108         $self->check_status;
109         $self->{child_died} = 0;
110
111         my $msg = $self->{osrf_handle}->process($wait_time);
112
113         # we woke up for any reason, reset the wait time to allow
114         # for idle maintenance as necessary
115         $wait_time = 1;
116
117         if($msg) {
118
119             if(my $child = pop(@{$self->{idle_list}})) {
120
121                 # we have an idle child to handle the request
122                 $chatty and $logger->internal("server: passing request to idle child $child");
123                 push(@{$self->{active_list}}, $child);
124                 $self->write_child($child, $msg);
125
126             } elsif($self->{num_children} < $self->{max_children}) {
127
128                 # spawning a child to handle the request
129                 $chatty and $logger->internal("server: spawning child to handle request");
130                 $self->write_child($self->spawn_child(1), $msg);
131
132             } else {
133
134                 $logger->warn("server: no children available, waiting...");
135                 $self->check_status(1); # block until child is available
136
137                 my $child = pop(@{$self->{idle_list}});
138                 push(@{$self->{active_list}}, $child);
139                 $self->write_child($child, $msg);
140             }
141
142         } else {
143
144             # don't perform idle maint immediately when woken by SIGCHLD
145             unless($self->{child_died}) {
146
147                 # when we hit equilibrium, there's no need for regular
148                 # maintenance, so set wait_time to 'forever'
149                 $wait_time = -1 if 
150                     !$self->perform_idle_maintenance and # no maintenance performed this time
151                     @{$self->{active_list}} == 0; # no active children 
152             }
153         }
154     }
155 }
156
157 # ----------------------------------------------------------------
158 # Launch a new spare child or kill an extra spare child.  To
159 # prevent large-scale spawning or die-offs, spawn or kill only
160 # 1 process per idle maintenance loop.
161 # Returns true if any idle maintenance occurred, 0 otherwise
162 # ----------------------------------------------------------------
163 sub perform_idle_maintenance {
164     my $self = shift;
165
166     $chatty and $logger->internal(
167         sprintf(
168             "server: %d idle, %d active, %d min_spare, %d max_spare in idle maintenance",
169             scalar(@{$self->{idle_list}}), 
170             scalar(@{$self->{active_list}}),
171             $self->{min_spare_children},
172             $self->{max_spare_children}
173         )
174     );
175
176     # spawn 1 spare child per maintenance loop if necessary
177     if( $self->{min_spare_children} and
178         $self->{num_children} < $self->{max_children} and
179         scalar(@{$self->{idle_list}}) < $self->{min_spare_children} ) {
180
181         $chatty and $logger->internal("server: spawning spare child");
182         $self->spawn_child;
183         return 1;
184
185     # kill 1 excess spare child per maintenance loop if necessary
186     } elsif($self->{max_spare_children} and
187             $self->{num_children} > $self->{min_children} and
188             scalar(@{$self->{idle_list}}) > $self->{max_spare_children} ) {
189
190         $chatty and $logger->internal("server: killing spare child");
191         $self->kill_child;
192         return 1;
193     }
194
195     return 0;
196 }
197
198 sub kill_child {
199     my $self = shift;
200     my $child = shift || pop(@{$self->{idle_list}}) or return;
201     $chatty and $logger->internal("server: killing child $child");
202     kill('TERM', $child->{pid});
203 }
204
205 # ----------------------------------------------------------------
206 # Jabber connection inbound message arrive on.
207 # ----------------------------------------------------------------
208 sub build_osrf_handle {
209     my $self = shift;
210
211     my $conf = OpenSRF::Utils::Config->current;
212     my $username = $conf->bootstrap->username;
213     my $password = $conf->bootstrap->passwd;
214     my $domain = $conf->bootstrap->domain;
215     my $port = $conf->bootstrap->port;
216     my $resource = $self->{service} . '_listener_' . $conf->env->hostname;
217
218     $logger->debug("server: inbound connecting as $username\@$domain/$resource on port $port");
219
220     $self->{osrf_handle} =
221         OpenSRF::Transport::SlimJabber::Client->new(
222             username => $username,
223             resource => $resource,
224             password => $password,
225             host => $domain,
226             port => $port,
227         );
228
229     $self->{osrf_handle}->initialize;
230 }
231
232
233 # ----------------------------------------------------------------
234 # Sends request data to a child process
235 # ----------------------------------------------------------------
236 sub write_child {
237     my($self, $child, $msg) = @_;
238     my $xml = decode_utf8($msg->to_xml);
239     syswrite($child->{pipe_to_child}, encode_utf8($xml));
240 }
241
242 # ----------------------------------------------------------------
243 # Checks to see if any child process has reported its availability
244 # In blocking mode, blocks until a child has reported.
245 # ----------------------------------------------------------------
246 sub check_status {
247     my($self, $block) = @_;
248
249     return unless @{$self->{active_list}};
250
251     my $read_set = IO::Select->new;
252     $read_set->add($_->{pipe_to_child}) for @{$self->{active_list}};
253
254     my @pids;
255
256     while (1) {
257
258         # if can_read or sysread is interrupted while bloking, go back and 
259         # wait again until we have at least 1 free child
260
261         if(my @handles = $read_set->can_read(($block) ? undef : 0)) {
262             my $pid = '';
263             for my $pipe (@handles) {
264                 sysread($pipe, $pid, STATUS_PIPE_DATA_SIZE) or next;
265                 push(@pids, int($pid));
266             }
267         }
268
269         last unless $block and !@pids;
270     }
271
272     return unless @pids;
273
274     $chatty and $logger->internal("server: ".scalar(@pids)." children reporting for duty: (@pids)");
275
276     my $child;
277     my @new_actives;
278
279     # move the children from the active list to the idle list
280     for my $proc (@{$self->{active_list}}) {
281         if(grep { $_ == $proc->{pid} } @pids) {
282             push(@{$self->{idle_list}}, $proc);
283         } else {
284             push(@new_actives, $proc);
285         }
286     }
287
288     $self->{active_list} = [@new_actives];
289
290     $chatty and $logger->internal(sprintf(
291         "server: %d idle and %d active children after status update",
292             scalar(@{$self->{idle_list}}), scalar(@{$self->{active_list}})));
293 }
294
295 # ----------------------------------------------------------------
296 # Cleans up any child processes that have exited.
297 # In shutdown mode, block until all children have washed ashore
298 # ----------------------------------------------------------------
299 sub reap_children {
300     my($self, $shutdown) = @_;
301     $self->{child_died} = 1;
302
303     while(1) {
304
305         my $pid = waitpid(-1, ($shutdown) ? 0 : WNOHANG);
306         last if $pid <= 0;
307
308         $chatty and $logger->internal("server: reaping child $pid");
309
310         my $child = $self->{pid_map}->{$pid};
311
312         close($child->{pipe_to_parent});
313         close($child->{pipe_to_child});
314
315         $self->{active_list} = [ grep { $_->{pid} != $pid } @{$self->{active_list}} ];
316         $self->{idle_list} = [ grep { $_->{pid} != $pid } @{$self->{idle_list}} ];
317
318         $self->{num_children}--;
319         delete $self->{pid_map}->{$pid};
320         delete $child->{$_} for keys %$child; # destroy with a vengeance
321     }
322
323     $self->spawn_children unless $shutdown;
324
325     $chatty and $logger->internal(sprintf(
326         "server: %d idle and %d active children after reap_children",
327             scalar(@{$self->{idle_list}}), scalar(@{$self->{active_list}})));
328
329 }
330
331 # ----------------------------------------------------------------
332 # Spawn up to max_children processes
333 # ----------------------------------------------------------------
334 sub spawn_children {
335     my $self = shift;
336     $self->spawn_child while $self->{num_children} < $self->{min_children};
337 }
338
339 # ----------------------------------------------------------------
340 # Spawns a new child.  If $active is set, the child goes directly
341 # into the active_list.
342 # ----------------------------------------------------------------
343 sub spawn_child {
344     my($self, $active) = @_;
345
346     my $child = OpenSRF::Server::Child->new($self);
347
348     # socket for sending message data to the child
349     if(!socketpair(
350         $child->{pipe_to_child},
351         $child->{pipe_to_parent},
352         AF_UNIX, SOCK_STREAM, PF_UNSPEC)) {
353             $logger->error("server: error creating data socketpair: $!");
354             return undef;
355     }
356
357     $child->{pipe_to_child}->autoflush(1);
358     $child->{pipe_to_parent}->autoflush(1);
359
360     $child->{pid} = fork();
361
362     if($child->{pid}) { # parent process
363         $self->{num_children}++;
364         $self->{pid_map}->{$child->{pid}} = $child;
365
366         if($active) {
367             push(@{$self->{active_list}}, $child);
368         } else {
369             push(@{$self->{idle_list}}, $child);
370         }
371
372         $chatty and $logger->internal("server: server spawned child $child with ".$self->{num_children}." total children");
373
374         return $child;
375
376     } else { # child process
377
378         $SIG{$_} = 'DEFAULT' for (qw/INT TERM QUIT HUP/);
379
380         if($self->{stderr_log}) {
381
382             $chatty and $logger->internal("server: redirecting STDERR to " . $self->{stderr_log});
383
384             close STDERR;
385             unless( open(STDERR, '>>' . $self->{stderr_log}) ) {
386                 $logger->error("server: unable to open STDERR log file: " . $self->{stderr_log} . " : $@");
387                 open STDERR, '>/dev/null'; # send it back to /dev/null
388             }
389         }
390
391         $child->{pid} = $$;
392         eval {
393             $child->init;
394             $child->run;
395             OpenSRF::Transport::PeerHandle->retrieve->disconnect;
396         };
397         $logger->error("server: child process died: $@") if $@;
398         exit(0);
399     }
400 }
401
402 # ----------------------------------------------------------------
403 # Sends the register command to the configured routers
404 # ----------------------------------------------------------------
405 sub register_routers {
406     my $self = shift;
407
408     my $conf = OpenSRF::Utils::Config->current;
409     my $routers = $conf->bootstrap->routers;
410     my $router_name = $conf->bootstrap->router_name;
411     my @targets;
412
413     for my $router (@$routers) {
414         if(ref $router) {
415
416             if( !$router->{services} ||
417                 !$router->{services}->{service} ||
418                 (
419                     ref($router->{services}->{service}) eq 'ARRAY' and
420                     grep { $_ eq $self->{service} } @{$router->{services}->{service}}
421                 )  || $router->{services}->{service} eq $self->{service}) {
422
423                 my $name = $router->{name};
424                 my $domain = $router->{domain};
425                 push(@targets, "$name\@$domain/router");
426             }
427
428         } else {
429             push(@targets, "$router_name\@$router/router");
430         }
431     }
432
433     foreach (@targets) {
434         $logger->info("server: registering with router $_");
435         $self->{osrf_handle}->send(
436             to => $_,
437             body => 'registering',
438             router_command => 'register',
439             router_class => $self->{service}
440         );
441     }
442
443     $self->{routers} = \@targets;
444 }
445
446 # ----------------------------------------------------------------
447 # Sends the unregister command to any routers we have registered
448 # with.
449 # ----------------------------------------------------------------
450 sub unregister_routers {
451     my $self = shift;
452     return unless $self->{osrf_handle}->tcp_connected;
453
454         for my $router (@{$self->{routers}}) {
455         $logger->info("server: disconnecting from router $router");
456         $self->{osrf_handle}->send(
457             to => $router,
458             body => "unregistering",
459             router_command => "unregister",
460             router_class => $self->{service}
461         );
462     }
463 }
464
465
466 package OpenSRF::Server::Child;
467 use strict;
468 use warnings;
469 use OpenSRF::Transport;
470 use OpenSRF::Application;
471 use OpenSRF::Transport::PeerHandle;
472 use OpenSRF::Transport::SlimJabber::XMPPMessage;
473 use OpenSRF::Utils::Logger qw($logger);
474 use OpenSRF::DomainObject::oilsResponse qw/:status/;
475 use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK);
476 use Time::HiRes qw(time);
477 use POSIX qw/:sys_wait_h :errno_h/;
478
479 use overload '""' => sub { return '[' . shift()->{pid} . ']'; };
480
481 sub new {
482     my($class, $parent) = @_;
483     my $self = bless({}, $class);
484     $self->{pid} = 0; # my process ID
485     $self->{parent} = $parent; # Controller parent process
486     $self->{num_requests} = 0; # total serviced requests
487     return $self;
488 }
489
490 sub set_nonblock {
491     my($self, $fh) = @_;
492     my  $flags = fcntl($fh, F_GETFL, 0);
493     fcntl($fh, F_SETFL, $flags | O_NONBLOCK);
494 }
495
496 sub set_block {
497     my($self, $fh) = @_;
498     my  $flags = fcntl($fh, F_GETFL, 0);
499     $flags &= ~O_NONBLOCK;
500     fcntl($fh, F_SETFL, $flags);
501 }
502
503 # ----------------------------------------------------------------
504 # Connects to Jabber and runs the application child_init
505 # ----------------------------------------------------------------
506 sub init {
507     my $self = shift;
508     my $service = $self->{parent}->{service};
509     $0 = "OpenSRF Drone [$service]";
510     OpenSRF::Transport::PeerHandle->construct($service);
511         OpenSRF::Application->application_implementation->child_init
512                 if (OpenSRF::Application->application_implementation->can('child_init'));
513 }
514
515 # ----------------------------------------------------------------
516 # Waits for messages from the parent process, handles the message,
517 # then goes into the keepalive loop if this is a stateful session.
518 # When max_requests is hit, the process exits.
519 # ----------------------------------------------------------------
520 sub run {
521     my $self = shift;
522     my $network = OpenSRF::Transport::PeerHandle->retrieve;
523
524     # main child run loop.  Ends when this child hits max requests.
525     while(1) {
526
527         my $data = $self->wait_for_request or next;
528
529         # Update process name to show activity
530         my $orig_name = $0;
531         $0 = "$0*";
532
533         # Discard extraneous data from the jabber socket
534         if(!$network->flush_socket()) {
535             $logger->error("server: network disconnected!  child dropping request and exiting: $data");
536             exit;
537         }
538
539         my $session = OpenSRF::Transport->handler(
540             $self->{parent}->{service},
541             OpenSRF::Transport::SlimJabber::XMPPMessage->new(xml => $data)
542         );
543
544         $self->keepalive_loop($session);
545
546         last if ++$self->{num_requests} == $self->{parent}->{max_requests};
547
548         # Tell the parent process we are available to process requests
549         $self->send_status;
550
551         # Repair process name
552         $0 = $orig_name;
553     }
554
555     $chatty and $logger->internal("server: child process shutting down after reaching max_requests");
556
557         OpenSRF::Application->application_implementation->child_exit
558                 if (OpenSRF::Application->application_implementation->can('child_exit'));
559 }
560
561 # ----------------------------------------------------------------
562 # waits for a request data on the parent pipe and returns it.
563 # ----------------------------------------------------------------
564 sub wait_for_request {
565     my $self = shift;
566
567     my $data = '';
568     my $read_size = 1024;
569     my $nonblock = 0;
570
571     while(1) {
572         # Start out blocking, when data is available, read it all
573
574         my $buf = '';
575         my $n = sysread($self->{pipe_to_parent}, $buf, $read_size);
576
577         unless(defined $n) {
578             $logger->error("server: error reading data pipe: $!") unless EAGAIN == $!; 
579             last;
580         }
581
582         last if $n <= 0; # no data left to read
583
584         $data .= $buf;
585
586         last if $n < $read_size; # done reading all data
587
588         $self->set_nonblock($self->{pipe_to_parent}) unless $nonblock;
589         $nonblock = 1;
590     }
591
592     $self->set_block($self->{pipe_to_parent}) if $nonblock;
593     return $data;
594 }
595
596
597 # ----------------------------------------------------------------
598 # If this is a stateful opensrf session, wait up to $keepalive
599 # seconds for subsequent requests from the client
600 # ----------------------------------------------------------------
601 sub keepalive_loop {
602     my($self, $session) = @_;
603     my $keepalive = $self->{parent}->{keepalive};
604
605     while($session->state and $session->state == $session->CONNECTED) {
606
607         unless( $session->queue_wait($keepalive) ) {
608
609             # client failed to disconnect before timeout
610             $logger->info("server: no request was received in $keepalive seconds, exiting stateful session");
611
612             my $res = OpenSRF::DomainObject::oilsConnectStatus->new(
613                 status => "Disconnected on timeout",
614                 statusCode => STATUS_TIMEOUT
615             );
616
617             $session->status($res);
618             $session->state($session->DISCONNECTED);
619             last;
620         }
621     }
622
623     $chatty and $logger->internal("server: child done with request(s)");
624     $session->kill_me;
625 }
626
627 # ----------------------------------------------------------------
628 # Report our availability to our parent process
629 # ----------------------------------------------------------------
630 sub send_status {
631     my $self = shift;
632     syswrite(
633         $self->{pipe_to_parent},
634         sprintf("%*s", OpenSRF::Server::STATUS_PIPE_DATA_SIZE, $self->{pid})
635     );
636 }
637
638
639 1;