Menu
▾
▴
cogkit-svn — SVN update notifications are sent to this mailing list
You can subscribe to this list here.
| 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
|
|
From: <ha...@us...> - 2008-10-20 21:44:01
|
Revision: 2236
http://cogkit.svn.sourceforge.net/cogkit/?rev=2236&view=rev
Author: hategan
Date: 2008-10-20 21:43:53 +0000 (Mon, 20 Oct 2008)
Log Message:
-----------
fixed a few issues
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-10-13 20:37:21 UTC (rev 2235)
+++ trunk/current/src/cog/modules/provider-coaster/resources/worker.pl 2008-10-20 21:43:53 UTC (rev 2236)
@@ -70,6 +70,26 @@
}
}
+sub hts {
+ my ($H) = @_;
+
+ my $k;
+ my $s = "{";
+ my $first = 1;
+
+ for $k (keys %$H) {
+ if (!$first) {
+ $s = $s.", ";
+ }
+ else {
+ $first = 0;
+ }
+ $s = $s."$k = $$H{$k}";
+ }
+
+ return $s."}";
+}
+
sub reconnect() {
my $fail = 0;
my $i;
@@ -113,8 +133,9 @@
my $len = length($msg);
my $buf = pack("VVV", $tag, $flags, $len);
$buf = $buf.$msg;
- #wlog("> len=$len, tag=$tag, flags=$flags, $msg\n");
+ wlog("> len=$len, tag=$tag, flags=$flags, $msg\n");
+
#($SOCK->send($buf) == length($buf)) || reconnect();
eval {defined($SOCK->send($buf))} or wlog("Send failed: $!\n");
}
@@ -158,9 +179,19 @@
my $flg = unpack("V", substr($data, 4, 4));
my $len = unpack("V", substr($data, 8, 4));
my $msg;
- $SOCK->recv($msg, $len);
+ my $frag;
+ my $alen = 0;
+ while ($alen < $len) {
+ $SOCK->recv($frag, $len - $alen);
+ $alen = $alen + length($frag);
+ $msg = $msg.$frag;
+ }
- #wlog("< len=$len, tag=$tag, flags=$flg, $data\n");
+ my $actuallen = length($msg);
+ wlog("< len=$len, actuallen=$actuallen, tag=$tag, flags=$flg, $msg\n");
+ if ($len != $actuallen) {
+ wlog("Warning: len != actuallen");
+ }
return ($tag, $flg, $msg);
}
@@ -278,6 +309,8 @@
sub registerCmd {
my ($tag, $cont) = @_;
+ wlog "Replies: ".hts(\%REPLIES)."\n";
+
$REPLIES{$tag} = [$cont, time(), ()];
}
@@ -404,11 +437,10 @@
my $tag = shift;
my $executable = $JOB{"executable"};
if (!(defined $JOBID)) {
- wlog "Job details\n";
- my $name;
- foreach $name (keys %JOB) {
- wlog "key: $name, value: $JOB{$name}\n";
- }
+ my $ds = hts(\%JOB);
+
+ wlog "Job details $ds\n";
+
sendError($tag, ("Missing job identity"));
return 0;
}
@@ -425,13 +457,17 @@
sub forkjob {
my ($pid, $status);
$JOB_RUNNING = 1;
+ pipe(PARENT_R, CHILD_W);
$pid = fork();
if (defined($pid)) {
if ($pid == 0) {
- runjob();
+ close PARENT_R;
+ runjob(\*CHILD_W);
+ close CHILD_W;
}
else {
wlog "Forked process $pid. Waiting for its completion\n";
+ close CHILD_W;
#waitpid($pid, 0);
my $tid;
do {
@@ -447,7 +483,14 @@
}
} while $tid != $pid;
wlog "Child process $pid terminated. Status is $status. $!\n";
- queueCmd(\&nullCB, "JOBSTATUS", $JOBID, "$COMPLETED", "$status", "");
+ my $s = <PARENT_R>;
+ close PARENT_R;
+ if ($s eq "") {
+ queueCmd(\&nullCB, "JOBSTATUS", $JOBID, "$COMPLETED", "$status", $s);
+ }
+ else {
+ queueCmd(\&nullCB, "JOBSTATUS", $JOBID, "$FAILED", "$status", $s);
+ }
}
}
else {
@@ -457,6 +500,7 @@
}
sub runjob {
+ my ($WR) = @_;
my $executable = $JOB{"executable"};
my $stdout = $JOB{"stdout"};
my $stderr = $JOB{"stderr"};
@@ -484,7 +528,8 @@
open STDERR, ">$stderr" or die "Cannot redirect STDERR";
}
wlog "Command: @JOBARGS\n";
- exec { $executable } @JOBARGS or queueCmd(\&nullCB, "JOBSTATUS", $JOBID, "$FAILED", "513", "Could not execute $executable: $!");
+ exec { $executable } @JOBARGS;
+ print $WR "Could not execute $executable: $!\n";
die "Could not execute $executable: $!";
}
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-13 20:37:28
|
Revision: 2235
http://cogkit.svn.sourceforge.net/cogkit/?rev=2235&view=rev
Author: hategan
Date: 2008-10-13 20:37:21 +0000 (Mon, 13 Oct 2008)
Log Message:
-----------
that causes a circular dependency
Modified Paths:
--------------
trunk/current/src/cog/modules/abstraction/dependencies.xml
Modified: trunk/current/src/cog/modules/abstraction/dependencies.xml
===================================================================
--- trunk/current/src/cog/modules/abstraction/dependencies.xml 2008-10-11 17:46:57 UTC (rev 2234)
+++ trunk/current/src/cog/modules/abstraction/dependencies.xml 2008-10-13 20:37:21 UTC (rev 2235)
@@ -29,9 +29,9 @@
<ant antfile="${main.buildfile}" target="dep">
<property name="module" value="provider-localscheduler"/>
</ant>
- <ant antfile="${main.buildfile}" target="dep">
+ <!--<ant antfile="${main.buildfile}" target="dep">
<property name="module" value="provider-coaster"/>
- </ant>
+ </ant>-->
<!-- <ant antfile="${main.buildfile}" target="dep">
<property name="module" value="provider-ws"/>
</ant>-->
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-11 17:47:08
|
Revision: 2234
http://cogkit.svn.sourceforge.net/cogkit/?rev=2234&view=rev
Author: hategan
Date: 2008-10-11 17:46:57 +0000 (Sat, 11 Oct 2008)
Log Message:
-----------
build coasters by default
Modified Paths:
--------------
trunk/current/src/cog/modules/abstraction/dependencies.xml
Modified: trunk/current/src/cog/modules/abstraction/dependencies.xml
===================================================================
--- trunk/current/src/cog/modules/abstraction/dependencies.xml 2008-10-11 00:20:47 UTC (rev 2233)
+++ trunk/current/src/cog/modules/abstraction/dependencies.xml 2008-10-11 17:46:57 UTC (rev 2234)
@@ -29,6 +29,9 @@
<ant antfile="${main.buildfile}" target="dep">
<property name="module" value="provider-localscheduler"/>
</ant>
+ <ant antfile="${main.buildfile}" target="dep">
+ <property name="module" value="provider-coaster"/>
+ </ant>
<!-- <ant antfile="${main.buildfile}" target="dep">
<property name="module" value="provider-ws"/>
</ant>-->
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-11 00:20:56
|
Revision: 2233
http://cogkit.svn.sourceforge.net/cogkit/?rev=2233&view=rev
Author: hategan
Date: 2008-10-11 00:20:47 +0000 (Sat, 11 Oct 2008)
Log Message:
-----------
removed debugging stuff
Modified Paths:
--------------
trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/scheduler/WeightedHostScoreScheduler.java
Modified: trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/scheduler/WeightedHostScoreScheduler.java
===================================================================
--- trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/scheduler/WeightedHostScoreScheduler.java 2008-10-10 23:38:47 UTC (rev 2232)
+++ trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/scheduler/WeightedHostScoreScheduler.java 2008-10-11 00:20:47 UTC (rev 2233)
@@ -202,7 +202,6 @@
double sum = s.getSum();
if (policy == POLICY_WEIGHTED_RANDOM) {
double rand = Math.random() * sum;
- System.err.println("Sorted: " + s);
if (logger.isInfoEnabled() && !s.isEmpty()) {
logger.info("Sorted: " + s);
}
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-10 23:38:52
|
Revision: 2232
http://cogkit.svn.sourceforge.net/cogkit/?rev=2232&view=rev
Author: hategan
Date: 2008-10-10 23:38:47 +0000 (Fri, 10 Oct 2008)
Log Message:
-----------
fixed logging race
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-10-10 23:22:31 UTC (rev 2231)
+++ trunk/current/src/cog/modules/provider-coaster/resources/worker.pl 2008-10-10 23:38:47 UTC (rev 2232)
@@ -65,7 +65,7 @@
sub wlog {
my $msg;
foreach $msg (@_) {
- print LOG time(), " ", $msg;
+ print LOG time(), " $ID $msg";
#print $msg;
}
}
@@ -95,11 +95,11 @@
}
sub initlog() {
- open(LOG, ">$LOG") or die "Failed to open log file: $!";
+ open(LOG, ">>$LOG") or die "Failed to open log file: $!";
my $b = select(LOG);
$| = 1;
select($b);
- print LOG time(), " Logging started\n";
+ print LOG time(), " $ID Logging started\n";
}
@@ -113,7 +113,7 @@
my $len = length($msg);
my $buf = pack("VVV", $tag, $flags, $len);
$buf = $buf.$msg;
- wlog("> len=$len, tag=$tag, flags=$flags, $msg\n");
+ #wlog("> len=$len, tag=$tag, flags=$flags, $msg\n");
#($SOCK->send($buf) == length($buf)) || reconnect();
eval {defined($SOCK->send($buf))} or wlog("Send failed: $!\n");
@@ -144,7 +144,6 @@
sub sendError {
my ($tag, @msgs) = @_;
-
sendFrags($tag, $REPLY_FLAG | $ERROR_FLAG, @msgs);
}
@@ -161,7 +160,7 @@
my $msg;
$SOCK->recv($msg, $len);
- wlog("< len=$len, tag=$tag, flags=$flg, $data\n");
+ #wlog("< len=$len, tag=$tag, flags=$flg, $data\n");
return ($tag, $flg, $msg);
}
@@ -225,6 +224,8 @@
}
$cont->($tag, 0, $err, $frags);
}
+
+ return 1;
}
sub checkTimeouts2 {
@@ -265,7 +266,7 @@
$SOCK->recv($data, 12);
if (length($data) > 0) {
wlog "Received $data\n";
- eval { process(unpackData($data)); } || wlog "Failed to process data: $@\n";
+ eval { process(unpackData($data)); } || (die "Failed to process data: $@" && wlog "Failed to process data: $@\n");
}
else {
#sleep 250ms
@@ -403,6 +404,11 @@
my $tag = shift;
my $executable = $JOB{"executable"};
if (!(defined $JOBID)) {
+ wlog "Job details\n";
+ my $name;
+ foreach $name (keys %JOB) {
+ wlog "key: $name, value: $JOB{$name}\n";
+ }
sendError($tag, ("Missing job identity"));
return 0;
}
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-10 23:22:36
|
Revision: 2231
http://cogkit.svn.sourceforge.net/cogkit/?rev=2231&view=rev
Author: hategan
Date: 2008-10-10 23:22:31 +0000 (Fri, 10 Oct 2008)
Log Message:
-----------
more logging
Modified Paths:
--------------
trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/coaster/service/job/manager/Worker.java
Modified: trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/coaster/service/job/manager/Worker.java
===================================================================
--- trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/coaster/service/job/manager/Worker.java 2008-10-10 23:21:43 UTC (rev 2230)
+++ trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/coaster/service/job/manager/Worker.java 2008-10-10 23:22:31 UTC (rev 2231)
@@ -66,6 +66,10 @@
error = s;
}
else if (code == Status.COMPLETED) {
+ if (logger.isInfoEnabled()) {
+ logger.info("Worker stdout: " + src.getStdOutput());
+ logger.info("Worker stderr: " + src.getStdError());
+ }
if (starting) {
error = new StatusImpl(Status.FAILED,
"Worker ended prematurely", null);
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-10 23:21:47
|
Revision: 2230
http://cogkit.svn.sourceforge.net/cogkit/?rev=2230&view=rev
Author: hategan
Date: 2008-10-10 23:21:43 +0000 (Fri, 10 Oct 2008)
Log Message:
-----------
perhaps safer
Modified Paths:
--------------
trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/channels/ChannelContext.java
Modified: trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/channels/ChannelContext.java
===================================================================
--- trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/channels/ChannelContext.java 2008-10-10 23:21:02 UTC (rev 2229)
+++ trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/channels/ChannelContext.java 2008-10-10 23:21:43 UTC (rev 2230)
@@ -116,7 +116,7 @@
return channelID;
}
- public synchronized int nextCmdSeq() {
+ public int nextCmdSeq() {
cmdseq = cmdseq + 1;
while (activeSenders.containsKey(cmdseq) || activeReceivers.containsKey(cmdseq)) {
cmdseq = cmdseq + 1;
@@ -124,7 +124,7 @@
return cmdseq;
}
- public void registerCommand(Command cmd) throws ProtocolException {
+ public synchronized void registerCommand(Command cmd) throws ProtocolException {
if (cmd.getId() == RequestReply.NOID) {
cmd.setId(nextCmdSeq());
activeSenders.put(cmd.getId(), cmd);
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-10 23:21:07
|
Revision: 2229
http://cogkit.svn.sourceforge.net/cogkit/?rev=2229&view=rev
Author: hategan
Date: 2008-10-10 23:21:02 +0000 (Fri, 10 Oct 2008)
Log Message:
-----------
don't use async timeout if executing synchronously
Modified Paths:
--------------
trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/commands/Command.java
Modified: trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/commands/Command.java
===================================================================
--- trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/commands/Command.java 2008-10-10 23:18:30 UTC (rev 2228)
+++ trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/commands/Command.java 2008-10-10 23:21:02 UTC (rev 2229)
@@ -32,7 +32,7 @@
timer = new Timer();
}
- public static final int DEFAULT_REPLY_TIMEOUT = 10000 * 60;
+ public static final int DEFAULT_REPLY_TIMEOUT = 2 * 60 * 1000; //2 minutes
public static final int DEFAULT_MAX_RETRIES = 2;
private int replyTimeout = DEFAULT_REPLY_TIMEOUT;
private int maxRetries = DEFAULT_MAX_RETRIES;
@@ -57,11 +57,14 @@
}
public void waitForReply() throws ReplyTimeoutException {
- if (!this.isInDataReceived()) {
- synchronized (this) {
+ synchronized (this) {
+ if (!this.isInDataReceived()) {
long start = System.currentTimeMillis();
long left = replyTimeout;
while (!this.isInDataReceived()) {
+ if (left <= 0) {
+ throw new ReplyTimeoutException();
+ }
try {
wait(left);
}
@@ -69,9 +72,6 @@
e.printStackTrace();
}
left = replyTimeout - (System.currentTimeMillis() - start);
- if (left <= 0) {
- throw new ReplyTimeoutException();
- }
}
}
}
@@ -104,27 +104,30 @@
if (logger.isInfoEnabled()) {
logger.info(this + " CMD: " + this);
}
- channel.sendTaggedData(getId(), fin, getOutCmd().getBytes());
+ int id = getId();
+ if (id == NOID) {
+ logger.warn("Command has NOID: " + this, new Throwable());
+ }
+ channel.sendTaggedData(id, fin, getOutCmd().getBytes());
if (!fin) {
Iterator i = outData.iterator();
while (i.hasNext()) {
byte[] buf = (byte[]) i.next();
- channel.sendTaggedData(getId(), !i.hasNext(), buf);
+ channel.sendTaggedData(id, !i.hasNext(), buf);
}
}
- setupReplyTimeoutChecker();
}
catch (ChannelIOException e) {
reexecute(e.getMessage(), e);
}
}
-
+
protected void setupReplyTimeoutChecker() {
- timer.schedule(timeout = new Timeout(), replyTimeout);
+ timer.schedule(timeout = new Timeout(), replyTimeout);
}
public byte[] execute(KarajanChannel channel) throws ProtocolException, IOException {
- executeAsync(channel);
+ send(channel, false);
waitForReply();
if (errorMsg != null) {
throw new ProtocolException(errorMsg, exception);
@@ -142,8 +145,18 @@
}
public void executeAsync(KarajanChannel channel) throws ProtocolException {
+ send(channel, true);
+ }
+
+ protected void send(KarajanChannel channel, boolean async) throws ProtocolException {
channel.registerCommand(this);
+ if (getId() == NOID) {
+ logger.warn("Registration failed for command " + this + " on channel " + channel);
+ }
send();
+ if (async) {
+ setupReplyTimeoutChecker();
+ }
}
public int getReplyTimeout() {
@@ -163,10 +176,7 @@
}
public void receiveCompleted() {
- if (timeout == null) {
- return;
- }
- else {
+ if (timeout != null) {
timeout.cancel();
timeout = null;
}
@@ -207,9 +217,11 @@
protected void reexecute(String message, Exception ex) {
if (++retries > maxRetries) {
+ logger.info(this + ": failed too many times", ex);
errorReceived(message, ex);
}
else {
+ logger.info(this + ": re-sending");
try {
setChannel(ChannelManager.getManager().reserveChannel(
getChannel().getChannelContext()));
@@ -226,6 +238,7 @@
protected void handleReplyTimeout() {
timeout = null;
+ logger.info(this + ": handling reply timeout");
reexecute("Reply timeout", new ReplyTimeoutException());
}
@@ -236,7 +249,7 @@
}
public String toString() {
- return "Command(" + this.getOutCmd() + ")";
+ return "Command(" + this.getId() + ", " + this.getOutCmd() + ")";
}
public static interface Callback {
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-10 23:18:35
|
Revision: 2228
http://cogkit.svn.sourceforge.net/cogkit/?rev=2228&view=rev
Author: hategan
Date: 2008-10-10 23:18:30 +0000 (Fri, 10 Oct 2008)
Log Message:
-----------
no point in doing it otherwise
Modified Paths:
--------------
trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/RequestReply.java
Modified: trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/RequestReply.java
===================================================================
--- trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/RequestReply.java 2008-10-10 23:16:15 UTC (rev 2227)
+++ trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/RequestReply.java 2008-10-10 23:18:30 UTC (rev 2228)
@@ -140,8 +140,8 @@
}
public void receiveCompleted() {
- inDataReceived = true;
synchronized (this) {
+ inDataReceived = true;
notify();
}
}
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
Revision: 2227
http://cogkit.svn.sourceforge.net/cogkit/?rev=2227&view=rev
Author: hategan
Date: 2008-10-10 23:16:15 +0000 (Fri, 10 Oct 2008)
Log Message:
-----------
don't do a isDirectory if it's not a recursive transfer
Modified Paths:
--------------
trunk/current/src/cog/modules/abstraction-common/src/org/globus/cog/abstraction/impl/fileTransfer/DelegatedFileTransferHandler.java
Modified: trunk/current/src/cog/modules/abstraction-common/src/org/globus/cog/abstraction/impl/fileTransfer/DelegatedFileTransferHandler.java
===================================================================
--- trunk/current/src/cog/modules/abstraction-common/src/org/globus/cog/abstraction/impl/fileTransfer/DelegatedFileTransferHandler.java 2008-10-10 23:15:29 UTC (rev 2226)
+++ trunk/current/src/cog/modules/abstraction-common/src/org/globus/cog/abstraction/impl/fileTransfer/DelegatedFileTransferHandler.java 2008-10-10 23:16:15 UTC (rev 2227)
@@ -239,7 +239,7 @@
}
else {
if (this.sourceResource != null) {
- if (this.sourceResource.isDirectory(spec.getSource())) {
+ if (spec.isRecursive() && this.sourceResource.isDirectory(spec.getSource())) {
if (localDestination == null) {
localDestination = File.createTempFile(this.task.getIdentity().getValue(),
null);
@@ -338,7 +338,7 @@
}
else {
if (this.destinationResource != null) {
- if (localSource.isDirectory()) {
+ if (spec.isRecursive() && localSource.isDirectory()) {
if (logger.isDebugEnabled()) {
logger
.debug("Directory transfer with resource local->remote");
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
Revision: 2226
http://cogkit.svn.sourceforge.net/cogkit/?rev=2226&view=rev
Author: hategan
Date: 2008-10-10 23:15:29 +0000 (Fri, 10 Oct 2008)
Log Message:
-----------
added distinction between single file and recursive transfers
Modified Paths:
--------------
trunk/current/src/cog/modules/abstraction-common/src/org/globus/cog/abstraction/impl/common/task/FileTransferSpecificationImpl.java
Modified: trunk/current/src/cog/modules/abstraction-common/src/org/globus/cog/abstraction/impl/common/task/FileTransferSpecificationImpl.java
===================================================================
--- trunk/current/src/cog/modules/abstraction-common/src/org/globus/cog/abstraction/impl/common/task/FileTransferSpecificationImpl.java 2008-10-10 23:15:05 UTC (rev 2225)
+++ trunk/current/src/cog/modules/abstraction-common/src/org/globus/cog/abstraction/impl/common/task/FileTransferSpecificationImpl.java 2008-10-10 23:15:29 UTC (rev 2226)
@@ -20,7 +20,7 @@
private String destinationFile;
private String source;
private String destination;
- private boolean thirdparty;
+ private boolean thirdparty, recursive;
private boolean thirdPartyIfPossible;
private long sourceFileOffset, destinationFileOffset, sourceFileLength;
private Hashtable attributes;
@@ -205,4 +205,12 @@
public void setSourceOffset(long sourceFileOffset) {
this.sourceFileOffset = sourceFileOffset;
}
+
+ public boolean isRecursive() {
+ return recursive;
+ }
+
+ public void setRecursive(boolean recursive) {
+ this.recursive = recursive;
+ }
}
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-10 23:15:13
|
Revision: 2225
http://cogkit.svn.sourceforge.net/cogkit/?rev=2225&view=rev
Author: hategan
Date: 2008-10-10 23:15:05 +0000 (Fri, 10 Oct 2008)
Log Message:
-----------
added distinction between single file and recursive transfers
Modified Paths:
--------------
trunk/current/src/cog/modules/abstraction-common/src/org/globus/cog/abstraction/interfaces/FileTransferSpecification.java
Modified: trunk/current/src/cog/modules/abstraction-common/src/org/globus/cog/abstraction/interfaces/FileTransferSpecification.java
===================================================================
--- trunk/current/src/cog/modules/abstraction-common/src/org/globus/cog/abstraction/interfaces/FileTransferSpecification.java 2008-10-07 22:19:04 UTC (rev 2224)
+++ trunk/current/src/cog/modules/abstraction-common/src/org/globus/cog/abstraction/interfaces/FileTransferSpecification.java 2008-10-10 23:15:05 UTC (rev 2225)
@@ -145,4 +145,12 @@
* be written.
*/
public long getDestinationOffset();
+
+ /**
+ * If set to true, recursive transfer of directories is requested. If set
+ * to false, a single file transfer is requested.
+ */
+ public void setRecursive(boolean recursive);
+
+ public boolean isRecursive();
}
\ No newline at end of file
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-07 22:19:08
|
Revision: 2224
http://cogkit.svn.sourceforge.net/cogkit/?rev=2224&view=rev
Author: hategan
Date: 2008-10-07 22:19:04 +0000 (Tue, 07 Oct 2008)
Log Message:
-----------
bougue fixee
Modified Paths:
--------------
trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/GetFileHandler.java
Modified: trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/GetFileHandler.java
===================================================================
--- trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/GetFileHandler.java 2008-10-07 03:27:27 UTC (rev 2223)
+++ trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/GetFileHandler.java 2008-10-07 22:19:04 UTC (rev 2224)
@@ -69,7 +69,7 @@
else {
int l = is.read(buf);
crt += l;
- if (crt == l) {
+ if (crt == size) {
is.close();
}
if (l == buf.length) {
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-07 03:28:00
|
Revision: 2223
http://cogkit.svn.sourceforge.net/cogkit/?rev=2223&view=rev
Author: hategan
Date: 2008-10-07 03:27:27 +0000 (Tue, 07 Oct 2008)
Log Message:
-----------
make sure delegation is enabled
Modified Paths:
--------------
trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/execution/coaster/ServiceManager.java
Modified: trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/execution/coaster/ServiceManager.java
===================================================================
--- trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/execution/coaster/ServiceManager.java 2008-10-07 03:26:44 UTC (rev 2222)
+++ trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/execution/coaster/ServiceManager.java 2008-10-07 03:27:27 UTC (rev 2223)
@@ -260,6 +260,7 @@
js.addArgument(sc.getHost());
js.setStdOutputLocation(FileLocation.MEMORY);
js.setStdErrorLocation(FileLocation.MEMORY);
+ js.setDelegation(Delegation.FULL_DELEGATION);
t.setSpecification(js);
ExecutionService s = new ExecutionServiceImpl();
s.setServiceContact(sc);
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-07 03:27:31
|
Revision: 2222
http://cogkit.svn.sourceforge.net/cogkit/?rev=2222&view=rev
Author: hategan
Date: 2008-10-07 03:26:44 +0000 (Tue, 07 Oct 2008)
Log Message:
-----------
avoid starting workers for tasks already running
Modified Paths:
--------------
trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/coaster/service/job/manager/WorkerManager.java
Modified: trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/coaster/service/job/manager/WorkerManager.java
===================================================================
--- trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/coaster/service/job/manager/WorkerManager.java 2008-10-05 17:11:29 UTC (rev 2221)
+++ trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/coaster/service/job/manager/WorkerManager.java 2008-10-07 03:26:44 UTC (rev 2222)
@@ -141,6 +141,7 @@
logger.info("Got allocation request: " + req);
}
}
+
try {
startWorker(new Seconds(req.maxWallTime.getSeconds())
.multiply(OVERALLOCATION_FACTOR)
@@ -188,6 +189,11 @@
copyAttributes(t, prototype, maxWallTime);
t.setRequiredService(1);
t.setService(0, buildService(prototype));
+ synchronized(this) {
+ if (!startingTasks.contains(prototype)) {
+ return;
+ }
+ }
Map newlyRequested = new HashMap();
for (int n = 0; n < numWorkers; n++) {
@@ -210,7 +216,6 @@
.getMessage(), e));
}
}
- System.err.println(t.getSpecification());
try {
handler.submit(t);
}
@@ -258,8 +263,6 @@
s.setSecurityContext(AbstractionFactory.newSecurityContext(s
.getProvider()));
}
- System.out.println("Worker start provider: " + s.getProvider());
- System.out.println("Worker start JM: " + s.getJobManager());
return s;
}
@@ -386,7 +389,9 @@
+ "). This worker manager instance does not "
+ "recall requesting a worker with such an id.");
}
- wr.setScheduledTerminationTime(Seconds.now().add(wr.getMaxWallTime()));
+ wr
+ .setScheduledTerminationTime(Seconds.now().add(
+ wr.getMaxWallTime()));
wr.setChannelContext(cc);
if (logger.isInfoEnabled()) {
logger.info("Worker registration received: " + wr);
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-05 19:08:42
|
Revision: 2221
http://cogkit.svn.sourceforge.net/cogkit/?rev=2221&view=rev
Author: hategan
Date: 2008-10-05 17:11:29 +0000 (Sun, 05 Oct 2008)
Log Message:
-----------
cleaned up channel startup; the muxer is a critical process - hard abort if any errors pop up
Modified Paths:
--------------
trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/channels/AbstractStreamKarajanChannel.java
trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/channels/AbstractTCPChannel.java
trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/channels/GSSChannel.java
Modified: trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/channels/AbstractStreamKarajanChannel.java
===================================================================
--- trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/channels/AbstractStreamKarajanChannel.java 2008-10-05 17:08:29 UTC (rev 2220)
+++ trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/channels/AbstractStreamKarajanChannel.java 2008-10-05 17:11:29 UTC (rev 2221)
@@ -156,7 +156,6 @@
}
-
private static final int SENDER_COUNT = 1;
private static Sender[] sender;
private static int crtSender;
@@ -283,7 +282,7 @@
if (multiplexer == null) {
multiplexer = new Multiplexer[MUX_COUNT];
for (int i = 0; i < MUX_COUNT; i++) {
- multiplexer[i] = new Multiplexer();
+ multiplexer[i] = new Multiplexer(i);
multiplexer[i].start();
}
}
@@ -291,11 +290,16 @@
}
protected static class Multiplexer extends Thread {
+ public static final Logger logger = Logger.getLogger(Multiplexer.class);
+
private Set channels;
private List remove, add;
+ private boolean terminated;
+ private int id;
- public Multiplexer() {
- super("Channel multiplexer");
+ public Multiplexer(int id) {
+ super("Channel multiplexer " + id);
+ this.id = id;
setDaemon(true);
channels = new HashSet();
remove = new ArrayList();
@@ -304,9 +308,16 @@
public synchronized void register(AbstractStreamKarajanChannel channel) {
add.add(channel);
+ if (logger.isInfoEnabled()) {
+ logger.info("(" + id + ") Scheduling " + channel + " for addition");
+ }
+ if (terminated) {
+ logger.warn("Trying to add a channel to a stopped multiplexer");
+ }
}
public void run() {
+ logger.info("Multiplexer " + id + " started");
boolean any;
try {
while (true) {
@@ -321,17 +332,24 @@
any |= channel.step();
}
catch (Exception e) {
- shutdown(channel, e);
+ try {
+ shutdown(channel, e);
+ }
+ catch (Exception ee) {
+ logger.warn("Failed to shut down channel", e);
+ }
}
}
synchronized (this) {
i = remove.iterator();
while (i.hasNext()) {
- channels.remove(i.next());
+ Object r = i.next();
+ channels.remove(r);
}
i = add.iterator();
while (i.hasNext()) {
- channels.add(i.next());
+ Object a = i.next();
+ channels.add(a);
}
remove.clear();
add.clear();
@@ -344,6 +362,15 @@
catch (Exception e) {
logger.warn("Exception in channel multiplexer", e);
}
+ catch (Error e) {
+ logger.error("Error in multiplexer", e);
+ e.printStackTrace();
+ System.exit(10);
+ }
+ finally {
+ logger.info("Multiplexer finished");
+ terminated = true;
+ }
}
private void shutdown(AbstractStreamKarajanChannel channel, Exception e) {
Modified: trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/channels/AbstractTCPChannel.java
===================================================================
--- trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/channels/AbstractTCPChannel.java 2008-10-05 17:08:29 UTC (rev 2220)
+++ trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/channels/AbstractTCPChannel.java 2008-10-05 17:11:29 UTC (rev 2221)
@@ -11,14 +11,10 @@
import java.io.IOException;
import java.net.Socket;
-import java.net.URI;
-import org.globus.cog.karajan.workflow.service.ProtocolException;
-import org.globus.cog.karajan.workflow.service.RemoteConfiguration;
import org.globus.cog.karajan.workflow.service.RequestManager;
-import org.globus.cog.karajan.workflow.service.commands.ChannelConfigurationCommand;
-public abstract class AbstractTCPChannel extends AbstractStreamKarajanChannel implements Runnable {
+public abstract class AbstractTCPChannel extends AbstractStreamKarajanChannel {
private Socket socket;
private boolean started;
private Exception startException;
@@ -42,18 +38,7 @@
else {
setName("S(" + socket.getLocalAddress() + ")");
}
- new Thread(this).start();
- while (!isStarted() && !isClosed() && startException == null) {
- try {
- wait();
- }
- catch (InterruptedException e) {
- }
- }
- if (startException != null) {
- logger.debug("Exception while starting channel", startException);
- throw new ChannelException(startException);
- }
+ initialize();
logger.info(getContact() + "Channel started");
if (isClient()) {
try {
@@ -65,30 +50,16 @@
}
}
- public void run() {
+ private void initialize() throws ChannelException {
ChannelContext context = getChannelContext();
try {
- try {
- started = true;
- }
- catch (Exception e) {
- startException = e;
- e.printStackTrace();
- return;
- }
- finally {
- synchronized (this) {
- notifyAll();
- }
- }
initializeConnection();
register();
+ started = true;
}
catch (Exception e) {
- if (!closing) {
- logger.warn("Exception in channel", e);
- context.notifyRegisteredListeners(e);
- }
+ logger.debug("Exception while starting channel", e);
+ throw new ChannelException(e);
}
}
@@ -96,7 +67,6 @@
}
-
public void shutdown() {
if (isLocalShutdown()) {
return;
Modified: trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/channels/GSSChannel.java
===================================================================
--- trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/channels/GSSChannel.java 2008-10-05 17:08:29 UTC (rev 2220)
+++ trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/channels/GSSChannel.java 2008-10-05 17:11:29 UTC (rev 2221)
@@ -30,7 +30,7 @@
import org.ietf.jgss.GSSCredential;
import org.ietf.jgss.GSSManager;
-public class GSSChannel extends AbstractTCPChannel implements Runnable {
+public class GSSChannel extends AbstractTCPChannel {
private static final Logger logger = Logger.getLogger(GSSChannel.class);
private GssSocket socket;
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-05 17:12:28
|
Revision: 2220
http://cogkit.svn.sourceforge.net/cogkit/?rev=2220&view=rev
Author: hategan
Date: 2008-10-05 17:08:29 +0000 (Sun, 05 Oct 2008)
Log Message:
-----------
added some debugging statements
Modified Paths:
--------------
trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/commands/ChannelConfigurationCommand.java
Modified: trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/commands/ChannelConfigurationCommand.java
===================================================================
--- trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/commands/ChannelConfigurationCommand.java 2008-10-05 17:07:01 UTC (rev 2219)
+++ trunk/current/src/cog/modules/karajan/src/org/globus/cog/karajan/workflow/service/commands/ChannelConfigurationCommand.java 2008-10-05 17:08:29 UTC (rev 2220)
@@ -12,6 +12,7 @@
import java.net.URI;
import java.rmi.server.UID;
+import org.apache.log4j.Logger;
import org.globus.cog.karajan.workflow.service.ProtocolException;
import org.globus.cog.karajan.workflow.service.RemoteConfiguration;
import org.globus.cog.karajan.workflow.service.channels.ChannelContext;
@@ -19,6 +20,8 @@
public class ChannelConfigurationCommand extends Command {
+ public static final Logger logger = Logger.getLogger(ChannelConfigurationCommand.class);
+
private final RemoteConfiguration.Entry config;
private final URI callbackURI;
@@ -58,6 +61,9 @@
}
public void receiveCompleted() {
+ if (logger.isInfoEnabled()) {
+ logger.info("Got reply");
+ }
ChannelID cid = getChannel().getChannelContext().getChannelID();
if (cid.getRemoteID() == null) {
cid.setRemoteID(new String(getInData(0)));
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-05 17:10:59
|
Revision: 2219
http://cogkit.svn.sourceforge.net/cogkit/?rev=2219&view=rev
Author: hategan
Date: 2008-10-05 17:07:01 +0000 (Sun, 05 Oct 2008)
Log Message:
-----------
increased max heap
Modified Paths:
--------------
trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/execution/coaster/Bootstrap.java
Modified: trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/execution/coaster/Bootstrap.java
===================================================================
--- trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/execution/coaster/Bootstrap.java 2008-10-04 16:28:32 UTC (rev 2218)
+++ trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/execution/coaster/Bootstrap.java 2008-10-05 17:07:01 UTC (rev 2219)
@@ -187,6 +187,7 @@
List args = new ArrayList();
args.add(java);
addDebuggingOptions(args);
+ args.add("-Xmx128M");
addProperties(args);
args.add("-cp");
args.add(sb.toString());
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-04 16:28:43
|
Revision: 2218
http://cogkit.svn.sourceforge.net/cogkit/?rev=2218&view=rev
Author: hategan
Date: 2008-10-04 16:28:32 +0000 (Sat, 04 Oct 2008)
Log Message:
-----------
initialize log before logging stuff
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-10-03 21:23:22 UTC (rev 2217)
+++ trunk/current/src/cog/modules/provider-coaster/resources/worker.pl 2008-10-04 16:28:32 UTC (rev 2218)
@@ -94,14 +94,16 @@
}
}
-
-sub init() {
+sub initlog() {
open(LOG, ">$LOG") or die "Failed to open log file: $!";
my $b = select(LOG);
$| = 1;
select($b);
print LOG time(), " Logging started\n";
+}
+
+sub init() {
wlog "uri=$URI, scheme=$SCHEME, host=$HOSTNAME, port=$PORT, id=$ID\n";
reconnect();
}
@@ -493,6 +495,8 @@
my $myhost=`hostname`;
$myhost =~ s/\s+$//;
+ initlog();
+
wlog("Initialized coaster worker $i\n");
wlog("Running on node $myhost\n");
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-03 21:23:30
|
Revision: 2217
http://cogkit.svn.sourceforge.net/cogkit/?rev=2217&view=rev
Author: hategan
Date: 2008-10-03 21:23:22 +0000 (Fri, 03 Oct 2008)
Log Message:
-----------
startup watchdog
Modified Paths:
--------------
trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/coaster/service/CoasterService.java
Modified: trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/coaster/service/CoasterService.java
===================================================================
--- trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/coaster/service/CoasterService.java 2008-10-02 18:44:07 UTC (rev 2216)
+++ trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/coaster/service/CoasterService.java 2008-10-03 21:23:22 UTC (rev 2217)
@@ -11,6 +11,8 @@
import java.io.IOException;
import java.net.Socket;
+import java.util.Timer;
+import java.util.TimerTask;
import org.apache.log4j.Logger;
import org.globus.cog.abstraction.coaster.service.job.manager.JobQueue;
@@ -40,6 +42,7 @@
private Exception e;
private boolean done;
private boolean suspended;
+ private static Timer watchdogs = new Timer();
public CoasterService() throws IOException {
this(null, null);
@@ -179,25 +182,25 @@
}
private void startShutdownWatchdog() {
- new Thread() {
- {
- setName("Shutdown watchdog");
- setDaemon(true);
+ watchdogs.schedule(new TimerTask() {
+ public void run() {
+ logger
+ .warn("Shutdown failed after 5 minutes. Forcefully shutting down");
+ System.exit(3);
}
+ }, 5 * 60 * 1000);
+ }
+
+ private static TimerTask startConnectWatchdog() {
+ TimerTask tt = new TimerTask() {
public void run() {
- try {
- Thread.sleep(5 * 60 * 1000);
- logger
- .info("Shutdown failed after 5 minutes. Forcefully shutting down");
- System.exit(3);
- }
- catch (InterruptedException e) {
- e.printStackTrace();
- }
+ logger.warn("Failed to connect after 2 minutes. Shutting down");
+ System.exit(4);
}
-
- }.start();
+ };
+ watchdogs.schedule(tt, 2 * 60 * 1000);
+ return tt;
}
public JobQueue getJobQueue() {
@@ -219,7 +222,9 @@
else {
s = new CoasterService(args[0], args[1]);
}
+ TimerTask t = startConnectWatchdog();
s.start();
+ t.cancel();
s.waitFor();
System.exit(0);
}
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-02 18:44:11
|
Revision: 2216
http://cogkit.svn.sourceforge.net/cogkit/?rev=2216&view=rev
Author: hategan
Date: 2008-10-02 18:44:07 +0000 (Thu, 02 Oct 2008)
Log Message:
-----------
...
Modified Paths:
--------------
trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/FileResourceImpl.java
Modified: trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/FileResourceImpl.java
===================================================================
--- trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/FileResourceImpl.java 2008-10-02 18:43:35 UTC (rev 2215)
+++ trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/FileResourceImpl.java 2008-10-02 18:44:07 UTC (rev 2216)
@@ -150,7 +150,12 @@
throws DirectoryNotFoundException, FileResourceException {
ListCommand lc = new ListCommand(directoryName);
run(lc);
- return lc.getResult();
+ try {
+ return lc.getResult();
+ }
+ catch (ProtocolException e) {
+ throw new FileResourceException(e);
+ }
}
public void putFile(String localFileName, String remoteFileName)
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-02 18:44:06
|
Revision: 2215
http://cogkit.svn.sourceforge.net/cogkit/?rev=2215&view=rev
Author: hategan
Date: 2008-10-02 18:43:35 +0000 (Thu, 02 Oct 2008)
Log Message:
-----------
fixed list
Modified Paths:
--------------
trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/commands/ListCommand.java
Modified: trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/commands/ListCommand.java
===================================================================
--- trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/commands/ListCommand.java 2008-10-02 18:43:12 UTC (rev 2214)
+++ trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/commands/ListCommand.java 2008-10-02 18:43:35 UTC (rev 2215)
@@ -10,14 +10,16 @@
package org.globus.cog.abstraction.impl.file.coaster.commands;
import java.io.File;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Iterator;
import java.util.List;
import org.globus.cog.abstraction.impl.file.GridFileImpl;
import org.globus.cog.abstraction.impl.file.PermissionsImpl;
import org.globus.cog.abstraction.interfaces.GridFile;
+import org.globus.cog.karajan.workflow.service.ProtocolException;
+import org.globus.cog.karajan.workflow.service.channels.AbstractKarajanChannel;
import org.globus.cog.karajan.workflow.service.commands.Command;
public class ListCommand extends Command {
@@ -27,22 +29,21 @@
super(NAME);
addOutData(name);
}
-
- public Collection getResult() {
+
+ public Collection getResult() throws ProtocolException {
List l = new ArrayList();
int i = 0;
int sz = getInDataSize();
if (sz % 7 != 0) {
- errorReceived("Reply size mismatch", new IOException(
- "Reply size mismatch"));
+ throw new ProtocolException("Reply size mismatch");
}
while (i < sz) {
GridFile f = new GridFileImpl();
f.setAbsolutePathName(getInDataAsString(i++));
+ f.setName(new File(f.getAbsolutePathName()).getName());
+ f.setLastModified(getInDataAsString(i++));
f.setFileType((byte) getInDataAsInt(i++));
- f.setLastModified(getInDataAsString(i++));
- f.setName(new File(f.getAbsolutePathName()).getName());
f.setSize(getInDataAsLong(i++));
f.setUserPermissions(new PermissionsImpl(getInDataAsInt(i++)));
f.setGroupPermissions(new PermissionsImpl(getInDataAsInt(i++)));
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-02 18:43:23
|
Revision: 2214
http://cogkit.svn.sourceforge.net/cogkit/?rev=2214&view=rev
Author: hategan
Date: 2008-10-02 18:43:12 +0000 (Thu, 02 Oct 2008)
Log Message:
-----------
fixed get
Modified Paths:
--------------
trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/commands/GetFileCommand.java
Modified: trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/commands/GetFileCommand.java
===================================================================
--- trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/commands/GetFileCommand.java 2008-10-02 18:42:31 UTC (rev 2213)
+++ trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/commands/GetFileCommand.java 2008-10-02 18:43:12 UTC (rev 2214)
@@ -23,7 +23,8 @@
private String local;
private ProgressMonitor pm;
- public GetFileCommand(String remote, String local, ProgressMonitor pm) throws FileNotFoundException {
+ public GetFileCommand(String remote, String local, ProgressMonitor pm)
+ throws FileNotFoundException {
super(NAME);
addOutData(remote);
this.local = local;
@@ -32,15 +33,20 @@
}
protected void addInData(byte[] data) {
- if (len == -1) {
- len = unpackLong(data);
+ if (this.getErrorFlag()) {
+ super.addInData(data);
}
else {
- try {
- fos.write(data);
+ if (len == -1) {
+ len = unpackLong(data);
}
- catch (IOException e) {
- errorReceived(e.getMessage(), e);
+ else {
+ try {
+ fos.write(data);
+ }
+ catch (IOException e) {
+ errorReceived(e.getMessage(), e);
+ }
}
}
}
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-02 18:42:48
|
Revision: 2213
http://cogkit.svn.sourceforge.net/cogkit/?rev=2213&view=rev
Author: hategan
Date: 2008-10-02 18:42:31 +0000 (Thu, 02 Oct 2008)
Log Message:
-----------
updated handlers
Modified Paths:
--------------
trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/ExistsHandler.java
trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/GetFileHandler.java
trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/ListHandler.java
trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/RmdirHandler.java
Modified: trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/ExistsHandler.java
===================================================================
--- trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/ExistsHandler.java 2008-10-02 18:41:23 UTC (rev 2212)
+++ trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/ExistsHandler.java 2008-10-02 18:42:31 UTC (rev 2213)
@@ -11,13 +11,17 @@
import java.io.File;
+import org.apache.log4j.Logger;
import org.globus.cog.karajan.workflow.service.ProtocolException;
public class ExistsHandler extends CoasterFileRequestHandler {
+ public static final Logger logger = Logger.getLogger(ExistsHandler.class);
public void requestComplete() throws ProtocolException {
File f = normalize(getInDataAsString(0));
-
+ if (logger.isInfoEnabled()) {
+ logger.info("Checking if " + f.getAbsolutePath() + " exists");
+ }
if (f.exists()) {
addOutData(true);
}
Modified: trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/GetFileHandler.java
===================================================================
--- trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/GetFileHandler.java 2008-10-02 18:41:23 UTC (rev 2212)
+++ trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/GetFileHandler.java 2008-10-02 18:42:31 UTC (rev 2213)
@@ -17,6 +17,7 @@
import org.apache.log4j.Logger;
import org.globus.cog.karajan.workflow.service.ProtocolException;
+import org.globus.cog.karajan.workflow.service.channels.KarajanChannel;
public class GetFileHandler extends CoasterFileRequestHandler {
public static final Logger logger = Logger
@@ -50,12 +51,13 @@
is = new FileInputStream(f);
}
catch (Exception e) {
+ logger.info("Could not open file", e);
ex = e;
}
}
public boolean hasNext() {
- return first || crt < size && ex == null;
+ return (first || crt < size) && ex == null;
}
public Object next() {
@@ -93,15 +95,31 @@
}
public int size() {
- return chunks + 1;
+ return ex == null ? chunks + 1 : 0;
}
};
}
-
+
public void send() throws ProtocolException {
- super.send();
+ KarajanChannel channel = getChannel();
+ Collection outData = getOutData();
+ if (channel == null) {
+ throw new ProtocolException("Unregistered command");
+ }
+ if (logger.isDebugEnabled()) {
+ logger.debug(ppOutData("HND"));
+ }
+ boolean fin = (outData == null) || (outData.size() == 0);
+ if (!fin) {
+ Iterator i = outData.iterator();
+ while (i.hasNext()) {
+ byte[] buf = (byte[]) i.next();
+ channel.sendTaggedReply(getId(), buf, !i.hasNext(), getErrorFlag());
+ }
+ }
if (ex != null) {
- sendError(ex.getMessage(), ex);
+ logger.info("Transfer exception", ex);
+ channel.sendTaggedReply(getId(), ex.getMessage().getBytes(), true, true);
}
}
}
Modified: trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/ListHandler.java
===================================================================
--- trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/ListHandler.java 2008-10-02 18:41:23 UTC (rev 2212)
+++ trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/ListHandler.java 2008-10-02 18:42:31 UTC (rev 2213)
@@ -10,27 +10,47 @@
package org.globus.cog.abstraction.impl.file.coaster.handlers;
import java.io.File;
+import java.util.Date;
+import org.apache.log4j.Logger;
import org.globus.cog.abstraction.interfaces.GridFile;
import org.globus.cog.karajan.workflow.service.ProtocolException;
public class ListHandler extends CoasterFileRequestHandler {
+ public static final Logger logger = Logger.getLogger(ListHandler.class);
public void requestComplete() throws ProtocolException {
File f = normalize(getInDataAsString(0));
-
+ if (logger.isInfoEnabled()) {
+ logger.info("Listing files in " + f.getAbsolutePath());
+ }
File[] l = f.listFiles();
- for (int i = 0; i < l.length; i++) {
- f = l[i];
- addOutData(f.getAbsolutePath());
- addOutData(f.isDirectory() ? GridFile.DIRECTORY : GridFile.FILE);
- addOutData("");
- addOutData(f.length());
- addOutData(0);
- addOutData(0);
- addOutData(0);
+ if (l == null) {
+ if (logger.isInfoEnabled()) {
+ logger.info("Invalid directory: " + f.getAbsolutePath());
+ }
+ sendError("No such directory: " + f.getAbsolutePath());
}
- sendReply();
+ else {
+ if (logger.isInfoEnabled()) {
+ logger.info("Listing " + l.length + " files");
+ }
+ for (int i = 0; i < l.length; i++) {
+ f = l[i];
+ addOutData(f.getAbsolutePath());
+ addOutData(new Date().toString());
+ addOutData(f.isDirectory() ? GridFile.DIRECTORY
+ : GridFile.FILE);
+ addOutData(f.length());
+ addOutData(0);
+ addOutData(0);
+ addOutData(0);
+ }
+ if (logger.isInfoEnabled()) {
+ logger.info("Sending reply");
+ }
+ sendReply();
+ }
}
}
Modified: trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/RmdirHandler.java
===================================================================
--- trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/RmdirHandler.java 2008-10-02 18:41:23 UTC (rev 2212)
+++ trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/file/coaster/handlers/RmdirHandler.java 2008-10-02 18:42:31 UTC (rev 2213)
@@ -21,7 +21,7 @@
if (!f.isDirectory()) {
sendError("Not a directory: " + f);
}
- if (!f.delete() && f.exists()) {
+ else if (!f.delete() && f.exists()) {
sendError("Failed to delete directory " + f);
}
else {
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
|
From: <ha...@us...> - 2008-10-02 18:41:43
|
Revision: 2212
http://cogkit.svn.sourceforge.net/cogkit/?rev=2212&view=rev
Author: hategan
Date: 2008-10-02 18:41:23 +0000 (Thu, 02 Oct 2008)
Log Message:
-----------
formatting
Modified Paths:
--------------
trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/execution/coaster/ServiceManager.java
Modified: trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/execution/coaster/ServiceManager.java
===================================================================
--- trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/execution/coaster/ServiceManager.java 2008-10-02 18:40:51 UTC (rev 2211)
+++ trunk/current/src/cog/modules/provider-coaster/src/org/globus/cog/abstraction/impl/execution/coaster/ServiceManager.java 2008-10-02 18:41:23 UTC (rev 2212)
@@ -128,7 +128,7 @@
public String reserveService(Task task, String bootHandlerProvider)
throws TaskSubmissionException {
- return reserveService(getContact(task), getSecurityContext(task) ,bootHandlerProvider);
+ return reserveService(getContact(task), getSecurityContext(task), bootHandlerProvider);
}
protected String waitForStart(Object service) throws InterruptedException {
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
