Author: spadkins
Date: Tue May 2 06:50:50 2006
New Revision: 6002
Modified:
p5ee/trunk/App-Context/lib/App/Context.pm
Log:
logging/log_file/log-rotation, set_user()/clear_session(), state(), extend_event_loop(), send_async_event()/send_async_message()/send_message(), fork()/exit()/shutdown_unshareable_resources()
Modified: p5ee/trunk/App-Context/lib/App/Context.pm
==============================================================================
--- p5ee/trunk/App-Context/lib/App/Context.pm (original)
+++ p5ee/trunk/App-Context/lib/App/Context.pm Tue May 2 06:50:50 2006
@@ -11,6 +11,9 @@
use App;
use Date::Format;
+use IO::Handle; # for the STDOUT->autoflush() method
+use IO::Socket;
+use IO::Socket::INET;
=head1 NAME
@@ -126,7 +129,7 @@
$context->{Conf}{$user} Info from conf file
[$context->{conf}]
$conf->{$type}{$name} Read-only service conf
- $context->{Session}{$session_id}
+ $context->{sessions}{$session_id}
[$context->{session}]
$session->{store}{$type}{$name} Runtime state which is stored
$session->{cache}{$type}{$name} Instances of services
@@ -225,6 +228,10 @@
$self->{options} = \%options;
$options{context} = $self;
+ $self->{log_level} = $options{log_level};
+ $self->{log_level} = 2 if (!defined $self->{log_level});
+ $self->log_file_open();
+
$conf_class = $options{conf_class};
$conf_class = "App::Conf::File" if (! $conf_class);
@@ -259,9 +266,11 @@
$self->{scheduled_events} = [];
$self->{scheduled_event} = {};
+ $self->{event_loop_extensions} = [];
+
$self->_init(\%options); # allows the subclass to do initialization
- $self->set_current_session($self->session("default"));
+ $self->set_current_session("default");
&App::sub_exit($self) if ($App::trace);
return $self;
@@ -1274,7 +1283,7 @@
* Signature: $context->log(@args);
* Signature: $context->log($options, @args);
* Param: $options HASH [in] (named)
- * Param: log_level integer
+ * Param: level integer
* Param: @args string [in]
* Return: void
* Throws: <none>
@@ -1284,6 +1293,51 @@
$context->log("oops, a bug happened");
+These are the standardized log levels.
+
+ 0 - Context logs nothing (absolutely silent) [???]
+ 1 - only application events [???]
+ 2 - [default] major system-level events [standard level for operations]
+ 3 - internal system-level events [standard level for development]
+ 4 - internal activities [standard level for debugging internals]
+ 5 - internal activities (inside loops) [extreme level for debugging internals]
+
+$self->log("ERROR: send_async_event_now(): node not assigned\n");
+$self->log($@);
+
+$self->log({level=>2},"Starting Cluster Node on $self->{host}:$self->{port}\n");
+$self->log({level=>2},"Stopping Cluster Node\n");
+$self->log({level=>2},"Starting Server on $self->{host}:$self->{port}\n");
+$self->log({level=>2},"Stopping Server.\n");
+$self->log({level=>2},"Starting Cluster Controller on $self->{host}:$self->{port}\n");
+$self->log({level=>2},"Stopping Cluster Controller\n");
+
+$self->log({level=>3},"Send Event: $service_type($name).$method(@args)\n");
+$self->log({level=>3},"Send Event: $method(@args)\n");
+$self->log({level=>3},"$service_type $name instantiated [$service]\n");
+$self->log({level=>3},"Schedule Event (" . join(",",%event) . ")\n";
+$self->log({level=>3},"Caught Signal: @_\n"); };
+$self->log({level=>3},"Caught Signal: @_\n"); };
+$self->log({level=>3},"Caught Signal: @_\n"); };
+$self->log({level=>3},"Caught Signal: @_ (quitting)\n"); $quit = 1; };
+$self->log({level=>3},"Caught Signal: @_ (quitting)\n"); $quit = 1; };
+$self->log({level=>3},"Caught Signal: @_ (quitting)\n"); $quit = 1; };
+$self->log({level=>3},"send_message($host, $port, $message)\n");
+$self->log({level=>3},"send_message($host, $port, ...) => [$response]\n");
+$self->log({level=>3},"process_msg($msg)\n");
+$self->log({level=>3},"process_msg: [$msg]\n");
+$self->log({level=>3},"process_msg($msg)\n");
+
+$self->log({level=>4},"Checking for scheduled events.\n");
+$self->log({level=>4},"Listening on socket: timeout($sleep_interval)\n");
+$self->log({level=>4},"Caught Signal: @_\n"); };
+$self->log({level=>4},"Listening on socket: timeout($sleep_interval)\n");
+$self->log({level=>4},"Child $pid finished [exitval=$exitval,sig=$sig]\n");
+
+$self->log({level=>5},"Checking event: time=$time [$event->{time}, every $event->{interval}] $event->{method}().\n");
+$self->log({level=>5},"Event Rescheduled: time=$time [$event->{time}, every $event->{interval}] $event->{method}().\n");
+$self->log({level=>5},"Event Removed: time=$time [$event->{time}, every $event->{interval}] $event->{method}().\n");
+
=cut
sub log {
@@ -1291,9 +1345,10 @@
my $self = shift;
my ($msg_options);
$msg_options = shift if ($#_ > -1 && ref($_[0]) eq "HASH");
- my $msg_log_level = $msg_options->{log_level} || 1;
+ my $msg_level = $msg_options->{level} || 1;
my $log_level = $self->{options}{log_level};
- if (!defined $log_level || $msg_log_level <= $log_level) {
+ $log_level = 2 if (!defined $log_level);
+ if (!defined $log_level || $msg_level <= $log_level) {
$self->_log(@_);
}
&App::sub_exit() if ($App::trace);
@@ -1302,7 +1357,43 @@
sub _log {
&App::sub_entry if ($App::trace);
my $self = shift;
- print STDERR "[$$] ", time2str("%Y-%m-%d %H:%M:%S", time()), " ", @_;
+ if ($#_ > 0) {
+ my $fmt = "[$$] " . time2str("%Y-%m-%d %H:%M:%S", time()) . " " . shift;
+ printf STDERR $fmt, @_;
+ }
+ elsif ($#_ == 0) {
+ print STDERR "[$$] ", time2str("%Y-%m-%d %H:%M:%S", time()), " ", @_;
+ }
+ &App::sub_exit() if ($App::trace);
+}
+
+# NOTE: log rotation always passes an $overwrite = 0, thus implementing the rule
+# that log rotation should never overwrite a log file, but only append to it.
+sub log_file_open {
+ &App::sub_entry if ($App::trace);
+ my ($self, $overwrite) = @_;
+ my $log_file = $self->{options}{log_file};
+ if ($log_file) {
+ if ($self->{log_fh}) {
+ close($self->{log_fh});
+ delete $self->{log_fh};
+ }
+ if ($log_file =~ /%/) {
+ $log_file = time2str($log_file, time());
+ }
+ if ((defined $overwrite && $overwrite) || (!defined $overwrite && $self->{options}{log_overwrite})) {
+ open(LOG, "> $log_file") || die "Unable to open $log_file log file: $!";
+ }
+ else {
+ open(LOG, ">> $log_file") || die "Unable to open $log_file log file: $!";
+ }
+ open(STDOUT, ">&LOG");
+ open(STDERR, ">&LOG");
+ LOG->autoflush(1);
+ STDOUT->autoflush(1);
+ STDERR->autoflush(1);
+ $self->{log_fh} = \*App::Context::LOG;
+ }
&App::sub_exit() if ($App::trace);
}
@@ -1330,8 +1421,16 @@
sub user {
&App::sub_entry if ($App::trace);
my $self = shift;
- &App::sub_exit("guest") if ($App::trace);
- "guest";
+ my $user = $self->{user} || "guest";
+ &App::sub_exit($user) if ($App::trace);
+ $user;
+}
+
+sub set_user {
+ &App::sub_entry if ($App::trace);
+ my ($self, $user) = @_;
+ $self->{user} = $user;
+ &App::sub_exit() if ($App::trace);
}
#############################################################################
@@ -1418,13 +1517,8 @@
&App::sub_entry if ($App::trace);
my ($self, $session_id, $args) = @_;
my ($session_class, $session, $options);
- if ($session_id) {
- $session = $self->{sessions}{$session_id};
- }
- else {
- $session_id = "default";
- $session = $self->{session};
- }
+ $session_id = "default" if (! defined $session_id || $session_id eq "");
+ $session = $self->{sessions}{$session_id};
if (!$session) {
$options = $self->{options};
$session_class = $options->{session_class} || $self->_default_session_class();
@@ -1459,8 +1553,9 @@
sub set_current_session {
&App::sub_entry if ($App::trace);
- my ($self, $session) = @_;
- $self->{session} = $session;
+ my ($self, $session_id) = @_;
+ $session_id = "default" if (!defined $session_id || $session_id ne "");
+ $self->{session} = $self->session($session_id);
&App::sub_exit() if ($App::trace);
}
@@ -1471,6 +1566,27 @@
&App::sub_exit() if ($App::trace);
}
+sub clear_session {
+ &App::sub_entry if ($App::trace);
+ my ($self, $session_id, @service_types) = @_;
+ $session_id = "default" if (!defined $session_id || $session_id ne "");
+ my $session = $self->{sessions}{$session_id};
+ if ($#service_types > -1) {
+ foreach my $service_type (@service_types) {
+ delete $session->{store}{$service_type};
+ delete $session->{cache}{$service_type};
+ }
+ }
+ else {
+ delete $self->{sessions}{$session_id};
+ if ($session eq $self->{session}) {
+ delete $self->{session};
+ $self->{session} = $self->session($session_id);
+ }
+ }
+ &App::sub_exit() if ($App::trace);
+}
+
#############################################################################
# PUBLIC METHODS
#############################################################################
@@ -1479,6 +1595,54 @@
=cut
+sub state {
+ &App::sub_entry if ($App::trace);
+ my ($self) = @_;
+
+ my $datetime = time2str("%Y-%m-%d %H:%M:%S", time());
+ my $class = ref($self);
+ my $state = "Context: [$class]\n[$datetime]\n";
+ $state .= "\n";
+ $state .= $self->_state();
+
+ &App::sub_exit($state) if ($App::trace);
+ return($state);
+}
+
+sub _state {
+ &App::sub_entry if ($App::trace);
+ my ($self) = @_;
+
+ my $state = "";
+
+ my ($event, @args, $args_str);
+ $state .= "Scheduled Events:\n";
+ foreach $event (@{$self->{scheduled_events}}) {
+ @args = ();
+ @args = @{$event->{args}} if ($event->{args});
+ $args_str = join(",",@args);
+ $state .= sprintf(" %19s %5s %-32s %s\n",
+ time2str("%Y-%m-%d %H:%M:%S",$event->{time}),
+ $event->{interval},
+ $event->{tag},
+ "$event->{name}.$event->{method}($args_str)");
+ }
+
+ $state .= "\n";
+ $state .= "Event Loop Extensions:\n";
+ my ($obj, $method, $args);
+ foreach my $event_loop_extension (@{$self->{event_loop_extensions}}) {
+ ($obj, $method, $args) = @$event_loop_extension;
+ @args = ();
+ @args = @$args if ($args);
+ $args_str = join(",",@args);
+ $state .= sprintf(" %s\n", "$obj->{name}.$method($args_str)");
+ }
+
+ &App::sub_exit($state) if ($App::trace);
+ return($state);
+}
+
#############################################################################
# dbg()
#############################################################################
@@ -1705,7 +1869,7 @@
sub dispatch_events {
&App::sub_entry if ($App::trace);
- my ($self) = @_;
+ my ($self, $max_events_occurred) = @_;
$self->dispatch_events_begin();
@@ -1757,6 +1921,14 @@
&App::sub_exit() if ($App::trace);
}
+sub extend_event_loop {
+ &App::sub_entry if ($App::trace);
+ my ($self, $obj, $method, $args) = @_;
+ $args = [] if (!$args);
+ push(@{$self->{event_loop_extensions}}, [ $obj, $method, $args ]);
+ &App::sub_exit() if ($App::trace);
+}
+
sub call {
&App::sub_entry if ($App::trace);
my ($self, $service_type, $name, $method, $args) = @_;
@@ -1894,7 +2066,7 @@
}
die "schedule_event(): (tag or method) is a required attribute of an event" if (!$event{tag} && !$event{method});
- print "[$$] Schedule Event (", join(",",%event), ")\n" if ($self->{verbose} >= 3);
+ $self->log({level=>3},"Schedule Event (" . join(",",%event) . ")\n");
my $event;
if ($event{tag}) {
@@ -1946,18 +2118,18 @@
# we do unshift() to keep events executing in FIFO order for a particular time
for (my $i = $#$scheduled_events; $i >= 0; $i--) {
$event = $scheduled_events->[$i];
- print "[$$] Checking event: time=$time [$event->{time}, every $event->{interval}] $event->{method}().\n" if ($verbose >= 9);
+ $self->log({level=>5},"Checking event: time=$time [$event->{time}, every $event->{interval}] $event->{method}().\n");
if ($event->{time} <= $time) {
unshift(@$events, $event);
if ($event->{time} && $event->{interval}) {
$event->{time} += $event->{interval}; # reschedule the event
- print "[$$] Event Rescheduled: time=$time [$event->{time}, every $event->{interval}] $event->{method}().\n" if ($verbose >= 9);
+ $self->log({level=>5},"Event Rescheduled: time=$time [$event->{time}, every $event->{interval}] $event->{method}().\n");
if ($time_of_next_event == 0 || $event->{time} < $time_of_next_event) {
$time_of_next_event = $event->{time};
}
}
else {
- print "[$$] Event Removed: time=$time [$event->{time}, every $event->{interval}] $event->{method}().\n" if ($verbose >= 9);
+ $self->log({level=>5},"Event Removed: time=$time [$event->{time}, every $event->{interval}] $event->{method}().\n");
splice(@$scheduled_events, $i, 1); # remove the (one-time) event
$event->{scheduled} = 0;
}
@@ -1979,17 +2151,32 @@
my $method = $event->{method};
my @args = $event->{args} ? @{$event->{args}} : ();
my $service_type = $event->{service_type};
+ my (@results);
if ($service_type) {
my $name = $event->{name};
my $service = $self->service($service_type, $name);
- $self->log("Send Event: $service_type($name).$method(@args)\n") if ($self->{verbose} >= 2);
- $service->$method(@args);
+ $self->log({level=>3},"Send Event: $service_type($name).$method(@args)\n");
+ @results = $service->$method(@args);
}
else {
- $self->log("Send Event: $method(@args)\n") if ($self->{verbose} >= 2);
- $self->$method(@args);
+ $self->log({level=>3},"Send Event: $method(@args)\n");
+ @results = $self->$method(@args);
+ }
+ &App::sub_exit(@results) if ($App::trace);
+ if (wantarray()) {
+ return(@results);
+ }
+ else {
+ if ($#results == -1) {
+ return(undef);
+ }
+ elsif ($#results == 0) {
+ return($results[0]);
+ }
+ else {
+ return(\@results);
+ }
}
- &App::sub_exit() if ($App::trace);
}
# NOTE: The baseline context implements the API for asynchronous events
@@ -2003,17 +2190,122 @@
sub send_async_event {
&App::sub_entry if ($App::trace);
my ($self, $event, $callback_event) = @_;
- $self->send_event($event);
+ my $errnum = 0;
+ my $errmsg = "";
+ my $event_token = "local-$$";
+ my ($returnval);
+ eval {
+ $returnval = $self->send_event($event);
+ };
+ if ($@) {
+ $errmsg = $@;
+ $errnum = 1;
+ }
if ($callback_event) {
- my $event_tag = "local-$$";
- if (! $callback_event->{args}) {
- $callback_event->{args} = [ $event_tag ];
- }
+ $callback_event->{args} = [] if (! $callback_event->{args});
+ push(@{$callback_event->{args}}, {event_token => $event_token, returnval => $returnval, errnum => $errnum, errmsg => $errmsg});
$self->send_event($callback_event);
}
+ &App::sub_exit($event_token) if ($App::trace);
+ return($event_token);
+}
+
+=head2 wait_for_event()
+
+ * Signature: $self->wait_for_event($event_token)
+ * Param: $event_token string
+ * Return: void
+ * Throws: App::Exception
+ * Since: 0.01
+
+ Sample Usage:
+
+ $self->wait_for_event($event_token);
+
+The wait_for_event() method is called when an asynchronous event has been
+sent and no more processing can be completed before it is done.
+
+=cut
+
+sub wait_for_event {
+ &App::sub_entry if ($App::trace);
+ my ($self, $event_token) = @_;
+ &App::sub_exit() if ($App::trace);
+}
+
+# NOTE: This send_message() and send_async_message() can be on the App::Context
+# class to allow a program in any context to send this kind of message.
+# (The only downside is a dependency on IO::Socket::INET.)
+sub send_async_message {
+ &App::sub_entry if ($App::trace);
+ my ($self, $host, $port, $message, $await_return_value, $timeout, $server_close) = @_;
+ my $pid = $self->fork();
+ if (!$pid) { # running in child
+ $self->send_message($host, $port, $message, $await_return_value, $timeout, $server_close);
+ $self->exit(0);
+ }
&App::sub_exit() if ($App::trace);
}
+# NOTE: $messages that start with "RV-" wait for a return value.
+# $messages that start with "SC-" force the server to close the socket first
+# This is to help manage which system has the sockets lingering in TIME_WAIT state.
+# Here is the truth table for $await_return_value, $server_close
+# $await_return_value $server_close = client + server
+# ------------------- ------------- ---------------------- ---------------------
+# 0 0 write/close read/close
+# 0 1 write/read/close read/close
+# 1 0 write/read/write/close read/write/read/close
+# 1 1 write/read/close read/write/close
+sub send_message {
+ &App::sub_entry if ($App::trace);
+ my ($self, $host, $port, $message, $await_return_value, $timeout, $server_close) = @_;
+ my $verbose = $self->{verbose};
+
+ if (!$port && $host =~ /^([^:]+):([0-9]+)$/) {
+ $host = $1;
+ $port = $2;
+ }
+
+ my $send_socket = IO::Socket::INET->new(
+ PeerAddr => $host,
+ PeerPort => $port,
+ Proto => "tcp",
+ Type => SOCK_STREAM,
+ ReuseAddr => 1,
+ );
+ my ($send_fd);
+ $send_fd = fileno($send_socket) if ($send_socket);
+ $self->log({level=>3},"($send_fd) send_message($host, $port, $message)\n");
+
+ my $response = "";
+ my $rv = $await_return_value ? "RV-" : "";
+ my $sc = $server_close ? "SC-" : "";
+ if ($send_socket) {
+ eval {
+ $send_socket->autoflush(1) if ($await_return_value || $server_close);
+ $send_socket->print("$rv$sc$message\n");
+ if ($await_return_value || $server_close) {
+ # $send_socket->timeout($timeout) if ($timeout); # doesn't seem to work
+ $response = $send_socket->getline();
+ $response =~ s/[\r\n]+$//;
+ $send_socket->print("EOF\n") if ($await_return_value && !$server_close);
+ }
+ close($send_socket);
+ };
+ if ($@) {
+ $response = "SEND ERROR: $@";
+ }
+ }
+ else {
+ $response = "CONNECT ERROR: $!";
+ }
+
+ $self->log({level=>3},"send_message($host, $port, ...) => [$response]\n");
+ &App::sub_exit($response) if ($App::trace);
+ return($response);
+}
+
=head2 fork()
* Signature: $pid = $self->fork()
@@ -2032,7 +2324,7 @@
connections to be created if necessary.
Call this after a fork() in the child process.
-It will renew the resources which cannot be shared between a parent and
+It will shut down the resources which cannot be shared between a parent and
a child process.
Currently, this is primarily for database connections.
@@ -2045,19 +2337,29 @@
my ($self) = @_;
my $pid = fork();
if (!$pid) { # in the child process
- $self->renew_process_resources();
+ # $self->{is_child} = 1; # I might need to add this sometime, but not now
+ $self->shutdown_unshareable_resources();
+ }
+ else {
+ $self->log({level=>4},"Child $pid started.\n");
}
&App::sub_exit($pid) if ($App::trace);
return($pid);
}
+sub exit {
+ my ($self, $exitval) = @_;
+ $self->shutdown();
+ exit($exitval);
+}
+
#############################################################################
-# renew_process_resources()
+# shutdown_unshareable_resources()
#############################################################################
-=head2 renew_process_resources()
+=head2 shutdown_unshareable_resources()
- * Signature: $self->renew_process_resources()
+ * Signature: $self->shutdown_unshareable_resources()
* Param: void
* Return: void
* Throws: App::Exception
@@ -2065,15 +2367,15 @@
Sample Usage:
- $self->renew_process_resources();
+ $self->shutdown_unshareable_resources();
-The renew_process_resources() method is called in a child process just after
+The shutdown_unshareable_resources() method is called in a child process just after
it has been fork()ed.
This causes connections to databases, etc. to be closed gracefully and new
connections to be created if necessary.
Call this after a fork() in the child process.
-It will renew the resources which cannot be shared between a parent and
+It will shutdown_unshareable which cannot be shared between a parent and
a child process.
Currently, this is primarily for database connections.
@@ -2081,19 +2383,19 @@
=cut
-sub renew_process_resources {
+sub shutdown_unshareable_resources {
my $self = shift;
my ($conf, $repdef, $repname, $instance);
my ($class, $method, $args, $argidx, $repcache);
- $self->dbgprint("Context->renew_process_resources()")
+ $self->dbgprint("Context->shutdown_unshareable_resources()")
if ($App::DEBUG && $self->dbg(1));
$repcache = $self->{session}{cache}{Repository};
if (defined $repcache && ref($repcache) eq "HASH") {
foreach $repname (keys %$repcache) {
$instance = $repcache->{$repname};
- $instance->_renew_process_resources();
+ $instance->_shutdown_unshareable_resources();
delete $repcache->{$repname};
}
}
lmpx.com only provides a reader for public news (NNTP) servers. It is not
affiliated with the servers or forums shown here and is not responsible for
the content of articles, which is written by their respective authors.