cogkit-svn Mailing List for Java CoG Kit (Page 5)

SourceForge Headquarters 1320 Columbia Street Suite 310 San Diego, CA 92101 +1 (858) 422-6466

Revision: 2156
          http://cogkit.svn.sourceforge.net/cogkit/?rev=2156&view=rev
Author:   hategan
Date:     2008-09-16 17:33:14 +0000 (Tue, 16 Sep 2008)

Log Message:
-----------
reconnect; heartbeats; proper exit code processing; fixed idle detection

Modified Paths:
--------------
    trunk/current/src/cog/modules/provider-coaster/resources/worker.pl

Modified: trunk/current/src/cog/modules/provider-coaster/resources/worker.pl
===================================================================

--- trunk/current/src/cog/modules/provider-coaster/resources/worker.pl	2008-09-16 17:29:04 UTC (rev 2155)
+++ trunk/current/src/cog/modules/provider-coaster/resources/worker.pl	2008-09-16 17:33:14 UTC (rev 2156)
@@ -1,6 +1,7 @@
 #!/usr/bin/perl
 use IO::Socket;
 use Cwd;
+use POSIX ":sys_wait_h";
 use strict;
 use warnings;
 
@@ -17,22 +18,30 @@
 my $RETRIES = 3;
 my $REPLYTIMEOUT = 60;
 my $MAXFRAGS = 16;
+my $MAX_RECONNECT_ATTEMPTS = 3;
 
-my $IDLETIMEOUT = 300; #Seconds
+my $IDLETIMEOUT = 10 * 60; #Seconds
 my $LASTRECV = 0;
+my $JOB_RUNNING = 0;
 
 my $BUFSZ = 2048;
 
+# 60 seconds by default. Note that since there is no configuration handshake
+# this would have to match the default interval in the service in order to avoid
+# "lost heartbeats".
+my $HEARTBEAT_INTERVAL = 5 * 60;
+
 my %REQUESTS = ();
 my %REPLIES  = ();
 
-my $LOG = "$ENV{HOME}/worker$ARGV[1].log";
+my $LOG = "$ENV{HOME}/.globus/coasters/worker$ARGV[1].log";
 
 
 my %HANDLERS = (
 	"SHUTDOWN" 	=> \&shutdown,
 	"SUBMITJOB" => \&submitjob,
 	"REGISTER"  => \&register,
+	"HEARTBEAT" => \&heartbeat,
 );
 
 my @CMDQ = ();
@@ -51,6 +60,7 @@
 my %JOB;
 my %JOBENV;
 my @JOBARGS;
+my $LAST_HEARTBEAT = 0;
 
 sub wlog {
 	my $msg;
@@ -60,37 +70,51 @@
 	}
 }
 
-sub init() {
+sub reconnect() {
 	my $fail = 0;
-	
-	open(LOG, ">$LOG") or die "Failed to open log file: $!";
-	my $b = select(LOG);
-	$| = 1;
-	select($b);
-	print LOG time(), " Logging started\n";
-
-	wlog "uri=$URI, scheme=$SCHEME, host=$HOSTNAME, port=$PORT, id=$ID\n";
-	for ($_ = 0; $_ < 10; $_++) {
-		$SOCK = IO::Socket::INET->new(Proto=>'tcp', PeerAddr=>$HOSTNAME, PeerPort=>$PORT) || ($fail = 1);
+	my $i;
+	for ($i = 0; $i < $MAX_RECONNECT_ATTEMPTS; $i++) {
+		wlog "Connecting ($i)...\n";
+		$SOCK = IO::Socket::INET->new(Proto=>'tcp', PeerAddr=>$HOSTNAME, PeerPort=>$PORT, Blocking=>1) || ($fail = 1);
 		if (!$fail) {
 			$SOCK->setsockopt(SOL_SOCKET, SO_RCVBUF, 16384);
 			$SOCK->setsockopt(SOL_SOCKET, SO_SNDBUF, 32768);
+			wlog "Connected\n";
+			$SOCK->blocking(0);
+			queueCmd(\&registerCB, "REGISTER", $ID, "wid://$ID");
 			last;
-		}		
+		}
+		else {
+			wlog "Connection failed\n";
+			select(undef, undef, undef, 2 ** $i);
+		}
 	}
 	if ($fail) {
-		die "Failed to create sockets: $!";	
+		die "Failed to connect: $!";
 	}
 }
 
 
+sub init() {
+	open(LOG, ">$LOG") or die "Failed to open log file: $!";
+	my $b = select(LOG);
+	$| = 1;
+	select($b);
+	print LOG time(), " Logging started\n";
+
+	wlog "uri=$URI, scheme=$SCHEME, host=$HOSTNAME, port=$PORT, id=$ID\n";
+	reconnect();
+}
+
 sub sendm {
 	my ($tag, $flags, $msg) = @_;
 	my $len = length($msg);
 	my $buf = pack("VVV", $tag, $flags, $len);
 	$buf = $buf.$msg;
 	wlog("> len=$len, tag=$tag, flags=$flags, $msg\n");
-	$SOCK->send($buf) == length($buf) or die "cannot send to $SOCK: $!";
+
+	#($SOCK->send($buf) == length($buf)) || reconnect();
+	eval {defined($SOCK->send($buf))} or wlog("Send failed: $!\n");
 }
 
 sub sendFrags {
@@ -106,7 +130,7 @@
 	my $cont = shift(@cmd);
 	my $ctag = $TAG++;
 	
-	registerCmd($ctag, $cont); 
+	registerCmd($ctag, $cont);
 	sendFrags($ctag, 0, @cmd);
 }
 
@@ -176,6 +200,7 @@
 		}
 	}
 	else {
+		$LASTRECV = time();
 		if (!exists($REQUESTS{$tag})) {
 			$REQUESTS{$tag} = [\&processRequest, time(), []];
 			wlog "New request ($tag)\n";
@@ -227,8 +252,8 @@
 	checkTimeouts2(\%REPLIES);
 	if ($LASTRECV != 0) {
 		my $dif = time() - $LASTRECV;
-		if ($dif >= $IDLETIMEOUT) {
-			die "Idle time exceeded";	
+		if ($dif >= $IDLETIMEOUT && $JOB_RUNNING == 0) {
+			die "Idle time exceeded";
 		}
 	}
 }
@@ -239,7 +264,6 @@
 	if (length($data) > 0) {
 		wlog "Received $data\n";
 		eval { process(unpackData($data)); } || wlog "$@\n";
-		$LASTRECV = time();
 	}
 	else {
 		#sleep 250ms
@@ -256,16 +280,24 @@
 
 
 sub mainloop {
-	my $cmd;
 	while(1) {
-		foreach $cmd (@CMDQ) {
-			sendCmd(@$cmd);
-		}
-		@CMDQ = ();
-		recvOne();
+		loopOne();
 	}
 }
 
+sub loopOne {
+	my $cmd;
+	if (time() - $LAST_HEARTBEAT > $HEARTBEAT_INTERVAL) {
+		queueCmd(\&heartbeatCB, "HEARTBEAT");
+		$LAST_HEARTBEAT = time();
+	}
+	foreach $cmd (@CMDQ) {
+		sendCmd(@$cmd);
+	}
+	@CMDQ = ();
+	recvOne();
+}
+
 sub queueCmd {
 	push @CMDQ, [@_];
 }
@@ -295,6 +327,24 @@
 	}
 }
 
+sub heartbeatCB {
+	my ($tag, $timeout, $err, $reply) = @_;
+	
+	if ($timeout) {
+		if (time() - $LAST_HEARTBEAT > 2 * $HEARTBEAT_INTERVAL) {
+			wlog "No heartbeat replies in a while. Dying.\n";
+			die "No response to heartbeat\n";
+		}
+	}
+	elsif ($err) {
+		die "Heartbeat failed: $reply\n";
+	}
+	else {
+		wlog "Heartbeat acknowledged\n";
+	}
+}
+
+
 sub register {
 	my ($tag, $timeout, $reply) = @_;
 	sendReply($tag, ("OK"));
@@ -309,6 +359,11 @@
 	exit 0;
 }
 
+sub heartbeat {
+	my ($tag, $timeout, $msgs) = @_;
+	sendReply($tag, ("OK"));
+}
+
 sub submitjob {
 	my ($tag, $timeout, $msgs) = @_;
 	my $desc = $$msgs[0];
@@ -361,6 +416,7 @@
 
 sub forkjob {
 	my ($pid, $status);
+	$JOB_RUNNING = 1;
 	$pid = fork();
 	if (defined($pid)) {
 		if ($pid == 0) {
@@ -368,8 +424,20 @@
 		}
 		else {
 			wlog "Forked process $pid. Waiting for its completion\n";
-			waitpid($pid, 0);
-			$status = $? & 0xff;
+			#waitpid($pid, 0);
+			my $tid;
+			do {
+				$tid = waitpid($pid, &WNOHANG);
+				if ($tid != $pid) {
+					loopOne();
+				}
+				else {
+					# exit code is in MSB and signal in LSB, so
+					# switch them such that status & 0xff is the
+					# exit code
+					$status = $? >> 8 + (($? & 0xff) << 8);
+				}
+			} while $tid != $pid;
 			wlog "Child process $pid terminated. Status is $status. $!\n";
 			queueCmd(\&nullCB, "JOBSTATUS", $JOBID, "$COMPLETED", "$status", "");
 		}
@@ -377,6 +445,7 @@
 	else {
 		queueCmd(\&nullCB, "JOBSTATUS", $JOBID, "$FAILED", "512", "Could not fork child process");
 	}
+	$JOB_RUNNING = 0;
 }
 
 sub runjob {
@@ -421,22 +490,21 @@
 		$ID =$ARGV[$i];
 		my $MSG="0";
 
-		my $myhost=`hostname -i`;
+		my $myhost=`hostname`;
 		$myhost =~ s/\s+$//;
 		init();
 
-		print LOG time(), " Initialized coaster worker $i\n";
-		queueCmd(\&registerCB, "REGISTER", $ID, "wid://$ID");
+		wlog("Initialized coaster worker $i\n");
+		wlog("Running on node $myhost\n");
 
 		mainloop();
 		exit(0);
-	} else {
+	}
+	else {
 		$wp[$i]=$waitpid;
 	}
 }
 
-for($i=1; $i<=$#ARGV ; $i++) {
+for ($i=1; $i<=$#ARGV ; $i++) {
 	waitpid($wp[$i],0);
 }
-
-


This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.




2006	Jan	Feb	Mar	Apr (39)	May (165)	Jun (164)	Jul (127)	Aug (81)	Sep (146)	Oct (375)	Nov (241)	Dec (77)
2007	Jan (42)	Feb (38)	Mar (30)	Apr (6)	May (17)	Jun	Jul (15)	Aug (59)	Sep (31)	Oct (44)	Nov (30)	Dec (12)
2008	Jan (9)	Feb (63)	Mar (18)	Apr (43)	May (28)	Jun (32)	Jul (61)	Aug (5)	Sep (72)	Oct (48)	Nov (6)	Dec

cogkit-svn Mailing List for Java CoG Kit (Page 5)

cogkit-svn — SVN update notifications are sent to this mailing list