[Swift-commit] r4020 - in SwiftApps/SwiftR/Swift: R exec
noreply at svn.ci.uchicago.edu
noreply at svn.ci.uchicago.edu
Thu Jan 20 18:31:06 CST 2011
Author: tga
Date: 2011-01-20 18:31:06 -0600 (Thu, 20 Jan 2011)
New Revision: 4020
Modified:
SwiftApps/SwiftR/Swift/R/Fifo.R
SwiftApps/SwiftR/Swift/R/Swift.R
SwiftApps/SwiftR/Swift/R/Workers.R
SwiftApps/SwiftR/Swift/exec/fifowrite
SwiftApps/SwiftR/Swift/exec/killtree
Log:
An assortment of improvements:
* quiet option for swiftapply
* test now automatically starts up workers if none running
* Fifos now have a no timeout option
* Workers can be shut down individually: swiftInit() returns a handle for the
worker process it starts, up, which can be passed to swiftShutdown()
* Launching workers from within R and running swiftapply works well, at least for local and ssh servers.
Modified: SwiftApps/SwiftR/Swift/R/Fifo.R
===================================================================
--- SwiftApps/SwiftR/Swift/R/Fifo.R 2011-01-20 21:31:55 UTC (rev 4019)
+++ SwiftApps/SwiftR/Swift/R/Fifo.R 2011-01-21 00:31:06 UTC (rev 4020)
@@ -2,12 +2,14 @@
.checkFifoArgs <- function(fifopath, timeout) {
# Catch errors proactively
- tryCatch(
- timeout <- round(timeout),
- error = function (e) {
- stop(paste("Provided timeout value of", timeout, "is not numeric"))
- }
- )
+ if (! is.null(timeout)) {
+ tryCatch(
+ timeout <- round(timeout),
+ error = function (e) {
+ stop(paste("Provided timeout value of", timeout, "is not numeric"))
+ }
+ )
+ }
if (! file.exists(fifopath) ) {
stop(paste("fifo specified does not exist or has wrong permissions:",
@@ -16,26 +18,34 @@
}
-# writes to a fifo with the provided timeout value
+# writes to a fifo with an optional timeout value.
# if the timeout is exceeded, FALSE is returned
# otherwise TRUE is returned
-writeFifo <- function (fifopath, message, timeout=3000) {
+writeFifo <- function (fifopath, message, timeout=NULL) {
.checkFifoArgs(fifopath, timeout)
bin <- file.path(.find.package("Swift"), "exec/fifowrite")
- cmdString <- paste(bin, fifopath, "-t", timeout,
- "-m", paste("'", message, "'", sep=""), "&>/dev/null")
+
+ if (is.null(timeout)) {
+ cmdString <- bin
+ }
+ else {
+ cmdString <- paste(bin, fifopath, "-t", timeout)
+ }
+ cmdString <- paste(cmdString, "-m",
+ paste("'", message, "'", sep=""), "&>/dev/null")
+
retval <- system(cmdString, wait=TRUE)
return (retval == 0)
}
-# reads from a fifo with the provided timeout value
+# reads from a fifo with an optional timeout value
# if the timeout is exceeded, NA is returned
# otherwise a vector of the lines of text is
# read from the fifo
# timeout is in milliseconds.
-readFifo <- function (fifopath, timeout=3000) {
+readFifo <- function (fifopath, timeout=NULL) {
.checkFifoArgs(fifopath, timeout)
# TODO: fork off another process to do reading, to avoid
# the R process blocking irrecoverably
@@ -43,14 +53,19 @@
# if the timeout is exceeded, terminates and writes
# _FIFOTIMEOUT_ to stdout
bin <- file.path(.find.package("Swift"), "exec/fiforead")
- cmdString <- paste(bin, fifopath, "-t", timeout)
+ if (is.null(timeout)) {
+ cmdString <- paste(bin, fifopath)
+ }
+ else {
+ cmdString <- paste(bin, fifopath, "-t", timeout)
+ }
fifodata <- suppressWarnings(system(cmdString, intern=TRUE, ignore.stderr=TRUE))
# stderr is ignored as we will get all relevant information
# through stdout
# want to suppress warning message about exit codea
toutMsg <- "_FIFOTIMEOUT_"
- if (length(fifodata) == 1
- && substr(fifodata[[1]], 1,nchar(toutMsg)) == toutMsg) {
+ if (length(fifodata) == 0
+ || substr(fifodata[[1]], 1,nchar(toutMsg)) == toutMsg) {
return (NA)
}
else {
Modified: SwiftApps/SwiftR/Swift/R/Swift.R
===================================================================
--- SwiftApps/SwiftR/Swift/R/Swift.R 2011-01-20 21:31:55 UTC (rev 4019)
+++ SwiftApps/SwiftR/Swift/R/Swift.R 2011-01-21 00:31:06 UTC (rev 4020)
@@ -1,18 +1,20 @@
swiftapply <- function( func, arglists,
- swiftserver=NULL,
+ server=NULL,
callsperbatch=NULL,
runmode=NULL,
initialexpr=NULL,
workerhosts=NULL,
keepwork=NULL,
- tmpdir=NULL )
+ tmpdir=NULL,
+ timeout=NULL,
+ quiet=FALSE)
{
# Set Swift default options if not passed as keywords or pre-set by user
- if(is.null(swiftserver))
- swiftserver <- getOption("swift.server")
- if(is.null(swiftserver))
- swiftserver<-"local"
+ if(is.null(server))
+ server <- getOption("swift.server")
+ if(is.null(server))
+ server<-"local"
if(is.null(callsperbatch))
callsperbatch <- getOption("swift.callsperbatch")
@@ -48,15 +50,15 @@
tmpdir <- Sys.getenv("SWIFTR_TMP");
if(tmpdir=="")
tmpdir <- "/tmp";
-
- cat("\nSwift properties:\n")
- cat(" swiftserver =", swiftserver,"\n");
- cat(" callsperbatch =", callsperbatch,"\n")
- cat(" runmode =", runmode,"\n")
- cat(" tmpdir =", tmpdir,"\n")
- cat(" workerhosts =", workerhosts,"\n")
- cat(" initialexpr =", initialexpr,"\n\n")
-
+ if (! quiet) {
+ cat("\nSwift properties:\n")
+ cat(" server =", server,"\n");
+ cat(" callsperbatch =", callsperbatch,"\n")
+ cat(" runmode =", runmode,"\n")
+ cat(" tmpdir =", tmpdir,"\n")
+ cat(" workerhosts =", workerhosts,"\n")
+ cat(" initialexpr =", initialexpr,"\n\n")
+ }
user <- Sys.info()[["user"]]
# Initialize globals if first call in this workspace
@@ -81,7 +83,9 @@
# reqdir <- system(paste("mktemp -d ", basedir, "/SwiftR.run.XXXX",sep=""),intern=TRUE)
reqdir = sprintf("%s/R%.7d",requestdirbase,requestid)
dir.create(reqdir,showWarnings=FALSE)
- cat("Swift request is in",reqdir,"\n")
+ if (! quiet) {
+ cat("Swift request is in",reqdir,"\n")
+ }
narglists <- length(arglists) # number of arglists to process
batch <- 1 # Next arglist batch number to fill
@@ -109,7 +113,7 @@
if( runmode == "manual" ) { # Prompt for return (empty line) to continue; assumes user ran a manual R to process the call.
cat("Manual Swift Run:\n run dir: ", getwd(), "/", reqdir,"\n\n")
cat(" swift script: ", RunSwiftScript, "\n")
- cat(" swiftserver: ", swiftserver,"\n")
+ cat(" server: ", server,"\n")
cat(" swiftapplyScript: ", swiftapplyScript,"\n")
cat(" Use RunAllR.sh to process and press return when complete:")
system(paste("cp ", system.file(package="Swift","exec/RunAllR.sh"), reqdir))
@@ -117,13 +121,13 @@
}
else if (runmode == "script") {
RunSwiftScript <- system.file(package="Swift","exec/RunSwiftScript.sh")
- system(paste(RunSwiftScript,reqdir,swiftserver,swiftapplyScript,"\"",workerhosts,"\""))
+ system(paste(RunSwiftScript,reqdir,server,swiftapplyScript,"\"",workerhosts,"\""))
}
else { # runmode == "service" # FIXME: check and post error if not "service"
# Send request to service
- swiftServerDir = paste(tmpdir,"/",user,"/SwiftR/swift.",swiftserver,sep="")
+ swiftServerDir = paste(tmpdir,"/",user,"/SwiftR/swift.",server,sep="")
requestPipeName=paste(swiftServerDir,"/requestpipe",sep="")
resultPipeName=paste(swiftServerDir,"/resultpipe",sep="")
@@ -136,21 +140,35 @@
if (file.exists(requestPipeName)) {
#TODO: there is a race condition here if the fifo disappears in
# between checking for existence and opening the fifo
+ writeTimeout <- 2000
+ success <- writeFifo(requestPipeName,paste(reqdir,"\n",sep=""),
+ timeout=writeTimeout)
+ if (! success) {
+ stop(paste("timeout of", writeTimeout,
+ "ms exceeded when attempting to",
+ "rendezvous with swift server: maybe it is not running or",
+ "it has crashed"))
+ }
- requestPipe <- fifo(requestPipeName,open="w",blocking=TRUE)
- cat(file=requestPipe,paste(reqdir,"\n",sep=""))
- close(requestPipe)
-
# Wait for reply from service
+ res <- readFifo(resultPipeName, timeout=timeout)
+ if ((! length(res) == 0) && is.na(res)) {
+ stop(paste("Timeout of ", timeout, "ms exceeded when waiting",
+ "for response from swift server"))
+ }
- resultPipe <- fifo(resultPipeName,open="r",blocking=TRUE)
- resultStatus <- readLines(con=resultPipe,n=1,ok=TRUE)
- close(resultPipe)
+ # Check that the message was correct
+ successMsg <- "done"
+ if (res[[1]] != successMsg)
+ stop(paste("Got unexpected message '",
+ paste(res, collapse="\n"),"' on fifo ",
+ "aborting job", sep=""))
}
else {
- stop(paste("It appears no SwiftR servers of type", swiftserver,
+ stop(paste("Have you run swiftInit?\n",
+ "It appears that no SwiftR servers of type", server,
"are running, as no request pipe exists in",
- swiftServerDir,"exists"))
+ swiftServerDir))
}
}
@@ -245,7 +263,6 @@
basicSwiftTest <- function() { swiftTest_1.1() }
# .... more tests from below to move here
-
swiftTest_4.1 <- function()
{
sumivars <- function() { initVar1+initVar2 }
@@ -324,310 +341,318 @@
runAllSwiftTests <- function()
{
+ # Launch workerif nothing already running
+ # testPid will be NULL if nothihg launched
+ testPid <- tryCatch(
+ (function() { swiftapply(log, list(list(1)),quiet=TRUE) ;
+ return(NULL); })(),
+ error=function(x) {swiftInit();})
-### FIXME: Save prior options here: restore them when tests are done. Recovery if interrrupted?
+ ### FIXME: Save prior options here: restore them when tests are done.
+ ### Recovery if interrrupted?
-failures=0
+ failures=0
-startTime = proc.time()[["elapsed"]]
+ startTime = proc.time()[["elapsed"]]
-cat("\n*** Starting test group 1 - functions on simple data structures ***\n\n")
+ cat("\n*** Starting test group 1 - functions on simple data structures ***\n\n")
-swiftTest_1.1()
+ swiftTest_1.1()
-##### Test 1.2
+ ##### Test 1.2
-# test 10 remote calls
+ # test 10 remote calls
-sumstuff <- function(treedata,cardata) { sum( treedata$Height, cardata$dist ) }
-data(cars)
-data(trees)
+ sumstuff <- function(treedata,cardata) { sum( treedata$Height, cardata$dist ) }
+ data(cars)
+ data(trees)
-args=list(trees,cars)
-arglist <- rep(list(args),10)
+ args=list(trees,cars)
+ arglist <- rep(list(args),10)
-localres = do.call(sumstuff,args)
+ localres = do.call(sumstuff,args)
-cat("\n*** Test 1.2.1: 10 calls to substuff()\n")
-swiftres <- swiftapply(sumstuff,arglist)
-cat("Swift result:\n")
-format(swiftres)
+ cat("\n*** Test 1.2.1: 10 calls to substuff()\n")
+ swiftres <- swiftapply(sumstuff,arglist)
+ cat("Swift result:\n")
+ format(swiftres)
-diffs <- 0
-for(i in 1:length(swiftres) ) {
- if( !identical(swiftres[[i]],localres) ) {
- diffs <- diffs + 1
- if( diffs < 10 ) cat(sprintf("res[%d]=%s\n",i,format( swiftres[[i]] )))
- }
-}
+ diffs <- 0
+ for(i in 1:length(swiftres) ) {
+ if( !identical(swiftres[[i]],localres) ) {
+ diffs <- diffs + 1
+ if( diffs < 10 ) cat(sprintf("res[%d]=%s\n",i,format( swiftres[[i]] )))
+ }
+ }
-if(diffs == 0) {
- cat("\n==> test 1.2.1 passed\n")
-} else {
- cat("\n!!!==> test 1.2.1 failed.\n")
- cat(sprintf(" %d result elements failed to match.\n",diffs));
- failures=failures+1
-}
+ if(diffs == 0) {
+ cat("\n==> test 1.2.1 passed\n")
+ } else {
+ cat("\n!!!==> test 1.2.1 failed.\n")
+ cat(sprintf(" %d result elements failed to match.\n",diffs));
+ failures=failures+1
+ }
-cat("\n*** Test 1.2.2: 10 calls to substuff() - callsperbatch=10\n")
-swiftres = swiftapply(sumstuff,arglist,callsperbatch=10)
-cat("Swift result:\n")
-format(swiftres)
+ cat("\n*** Test 1.2.2: 10 calls to substuff() - callsperbatch=10\n")
+ swiftres = swiftapply(sumstuff,arglist,callsperbatch=10)
+ cat("Swift result:\n")
+ format(swiftres)
-diffs <- 0
-for(i in 1:length(swiftres) ) {
- if( !identical(swiftres[[i]],localres) ) {
- diffs <- diffs + 1
- if( diffs < 10 ) cat(sprintf("res[%d]=%s\n",i,format( swiftres[[i]] )))
- }
-}
+ diffs <- 0
+ for(i in 1:length(swiftres) ) {
+ if( !identical(swiftres[[i]],localres) ) {
+ diffs <- diffs + 1
+ if( diffs < 10 ) cat(sprintf("res[%d]=%s\n",i,format( swiftres[[i]] )))
+ }
+ }
-if(diffs == 0) {
- cat("\n==> test 1.2.2 passed\n")
-} else {
- cat("\n!!!==> test 1.2.2 failed.\n")
- cat(sprintf(" %d result elements failed to match.\n",diffs));
- failures=failures+1
-}
+ if(diffs == 0) {
+ cat("\n==> test 1.2.2 passed\n")
+ } else {
+ cat("\n!!!==> test 1.2.2 failed.\n")
+ cat(sprintf(" %d result elements failed to match.\n",diffs));
+ failures=failures+1
+ }
-cat("\n*** Test 1.2.3: 10 calls to substuff() - callsperbatch=2\n")
-swiftres = swiftapply(sumstuff,arglist,callsperbatch=2)
-cat("Swift result:\n")
-format(swiftres)
+ cat("\n*** Test 1.2.3: 10 calls to substuff() - callsperbatch=2\n")
+ swiftres = swiftapply(sumstuff,arglist,callsperbatch=2)
+ cat("Swift result:\n")
+ format(swiftres)
-diffs <- 0
-for(i in 1:length(swiftres) ) {
- if( !identical(swiftres[[i]],localres) ) {
- diffs <- diffs + 1
- if( diffs < 10 ) cat(sprintf("res[%d]=%s\n",i, format( swiftres[[i]] )))
- }
-}
+ diffs <- 0
+ for(i in 1:length(swiftres) ) {
+ if( !identical(swiftres[[i]],localres) ) {
+ diffs <- diffs + 1
+ if( diffs < 10 ) cat(sprintf("res[%d]=%s\n",i, format( swiftres[[i]] )))
+ }
+ }
-if(diffs == 0) {
- cat("\n==> test 1.2.3 passed\n")
-} else {
- cat("\n!!!==> test 1.2.3 failed.\n")
- cat(sprintf(" %d result elements failed to match.\n",diffs));
- failures=failures+1
-}
+ if(diffs == 0) {
+ cat("\n==> test 1.2.3 passed\n")
+ } else {
+ cat("\n!!!==> test 1.2.3 failed.\n")
+ cat(sprintf(" %d result elements failed to match.\n",diffs));
+ failures=failures+1
+ }
-cat("\n*** Test 1.2.4: 10 calls to substuff() - callsperbatch=3\n")
-swiftres = swiftapply(sumstuff,arglist,callsperbatch=3)
-swiftres <- swiftapply(sumstuff,arglist)
-cat("Swift result:\n")
-format(swiftres)
+ cat("\n*** Test 1.2.4: 10 calls to substuff() - callsperbatch=3\n")
+ swiftres = swiftapply(sumstuff,arglist,callsperbatch=3)
+ swiftres <- swiftapply(sumstuff,arglist)
+ cat("Swift result:\n")
+ format(swiftres)
-diffs <- 0
-for(i in 1:length(swiftres) ) {
- if( !identical(swiftres[[i]],localres) ) {
- diffs <- diffs + 1
- if( diffs < 10 ) cat(sprintf("res[%d]=%d\n",i, format( swiftres[[i]] )))
- }
-}
+ diffs <- 0
+ for(i in 1:length(swiftres) ) {
+ if( !identical(swiftres[[i]],localres) ) {
+ diffs <- diffs + 1
+ if( diffs < 10 ) cat(sprintf("res[%d]=%d\n",i, format( swiftres[[i]] )))
+ }
+ }
-if(diffs == 0) {
- cat("\n==> test 1.2.4 passed\n")
-} else {
- cat("\n!!!==> test 1.2.4 failed.\n")
- cat(sprintf(" %d result elements failed to match.\n",diffs));
- failures=failures+1
-}
+ if(diffs == 0) {
+ cat("\n==> test 1.2.4 passed\n")
+ } else {
+ cat("\n!!!==> test 1.2.4 failed.\n")
+ cat(sprintf(" %d result elements failed to match.\n",diffs));
+ failures=failures+1
+ }
-# swiftres = swiftapply(sumstuff,arglist,callsperbatch=2,site="pbs")
-# test variations on local vs ssh vs pbs; coasters vs non; etc.
+ # swiftres = swiftapply(sumstuff,arglist,callsperbatch=2,site="pbs")
+ # test variations on local vs ssh vs pbs; coasters vs non; etc.
-##### Test Group 2
+ ##### Test Group 2
-cat("\n*** Starting test group 2 - test matrix passing***\n")
+ cat("\n*** Starting test group 2 - test matrix passing***\n")
-matfunc <- function( m1, m2 )
-{
- (1/m1) %*% m2
-}
+ matfunc <- function( m1, m2 )
+ {
+ (1/m1) %*% m2
+ }
-n <- 5
-m1 <- array(sin(1:n**2), dim=c(n,n))
-m2 <- t(m1)
+ n <- 5
+ m1 <- array(sin(1:n**2), dim=c(n,n))
+ m2 <- t(m1)
-localres = matfunc(m1,m2)
+ localres = matfunc(m1,m2)
-cat("\n*** Test 2.1: 100 calls to matfunc(dim=5x5) - callsperbatch=9\n")
+ cat("\n*** Test 2.1: 100 calls to matfunc(dim=5x5) - callsperbatch=9\n")
-args=list(m1,m2)
-arglist <- rep(list(args),100)
+ args=list(m1,m2)
+ arglist <- rep(list(args),100)
-swiftres = swiftapply(matfunc,arglist,callsperbatch=9)
+ swiftres = swiftapply(matfunc,arglist,callsperbatch=9)
-diffs <- 0
-#for(i in 1:length(swiftres) ) {
-for(i in c(seq(1,100,10),100)) {
- if( !all.equal(swiftres[[i]],localres) ) {
- diffs <- diffs + 1
- if( diffs < 10 ) cat(sprintf("res[%d]=%s\n",i,format(swiftres[[i]])))
- }
-}
+ diffs <- 0
+ #for(i in 1:length(swiftres) ) {
+ for(i in c(seq(1,100,10),100)) {
+ if( !all.equal(swiftres[[i]],localres) ) {
+ diffs <- diffs + 1
+ if( diffs < 10 ) cat(sprintf("res[%d]=%s\n",i,format(swiftres[[i]])))
+ }
+ }
-if(diffs == 0) {
- cat("\n==> test 2.1 passed\n")
-} else {
- cat("\n!!!==> test 2.2 failed.\n")
- cat(sprintf(" %d result elements failed to match.\n",diffs));
- failures=failures+1
-}
+ if(diffs == 0) {
+ cat("\n==> test 2.1 passed\n")
+ } else {
+ cat("\n!!!==> test 2.2 failed.\n")
+ cat(sprintf(" %d result elements failed to match.\n",diffs));
+ failures=failures+1
+ }
-n <- 237
-n <- 50
-m1 <- array(sin(1:n**2), dim=c(n,n))
-m2 <- t(m1)
+ n <- 237
+ n <- 50
+ m1 <- array(sin(1:n**2), dim=c(n,n))
+ m2 <- t(m1)
-localres = matfunc(m1,m2)
+ localres = matfunc(m1,m2)
-cat("\n*** Test 2.2: 123 calls to matfunc(dim=bigger) - callsperbatch=7\n") # FIXME make n easy to adjust and print actual value
+ cat("\n*** Test 2.2: 123 calls to matfunc(dim=bigger) - callsperbatch=7\n") # FIXME make n easy to adjust and print actual value
-args=list(m1,m2)
-arglist <- rep(list(args),123)
+ args=list(m1,m2)
+ arglist <- rep(list(args),123)
-swiftres = swiftapply(matfunc,arglist,callsperbatch=7)
+ swiftres = swiftapply(matfunc,arglist,callsperbatch=7)
-diffs <- 0
-#for(i in 1:length(swiftres) ) {
-for(i in c(seq(1,length(swiftres),10),length(swiftres))) {
+ diffs <- 0
+ #for(i in 1:length(swiftres) ) {
+ for(i in c(seq(1,length(swiftres),10),length(swiftres))) {
- if( !all.equal(swiftres[[i]],localres) ) {
- diffs <- diffs + 1
- if( diffs < 10 ) cat(sprintf("res[%d]=%s\n",i,format(swiftres[[i]])))
- }
-}
+ if( !all.equal(swiftres[[i]],localres) ) {
+ diffs <- diffs + 1
+ if( diffs < 10 ) cat(sprintf("res[%d]=%s\n",i,format(swiftres[[i]])))
+ }
+ }
-if(diffs == 0) {
- cat("\n==> test 2.2 passed\n")
-} else {
- cat("\n!!!==> test 2.2 failed.\n")
- cat(sprintf(" %d result elements failed to match.\n",diffs));
- failures=failures+1
-}
+ if(diffs == 0) {
+ cat("\n==> test 2.2 passed\n")
+ } else {
+ cat("\n!!!==> test 2.2 failed.\n")
+ cat(sprintf(" %d result elements failed to match.\n",diffs));
+ failures=failures+1
+ }
-##### Test Group 3
+ ##### Test Group 3
-cat("\n*** Starting test group 3 - test list element and name passing***\n")
+ cat("\n*** Starting test group 3 - test list element and name passing***\n")
-# Test if list element names are being sent and returned correctly
+ # Test if list element names are being sent and returned correctly
-n <- 5
-m1 <- array(sin(1:n**2), dim=c(n,n))
-m2 <- t(m1)
+ n <- 5
+ m1 <- array(sin(1:n**2), dim=c(n,n))
+ m2 <- t(m1)
-inlist = list()
-inlist[[1]]=123
-inlist[[2]]=456
-inlist$name1=789
-inlist$name2=987
-inlist$mat1 = m1
-inlist[[99]] = m2
+ inlist = list()
+ inlist[[1]]=123
+ inlist[[2]]=456
+ inlist$name1=789
+ inlist$name2=987
+ inlist$mat1 = m1
+ inlist[[99]] = m2
-listfunc <- function(ilist)
-{
- olist = ilist
- olist$sum = ilist[[1]] + ilist[[2]] + ilist$name1 + ilist$name2
- olist$names = names(ilist)
- olist$mprod = ilist$mat1 %*% ilist[[99]]
- return(olist)
-}
-localres = listfunc(inlist)
+ listfunc <- function(ilist)
+ {
+ olist = ilist
+ olist$sum = ilist[[1]] + ilist[[2]] + ilist$name1 + ilist$name2
+ olist$names = names(ilist)
+ olist$mprod = ilist$mat1 %*% ilist[[99]]
+ return(olist)
+ }
+ localres = listfunc(inlist)
-cat("\n*** Starting test 3.1 - 4 calls in one batch of 5 ***\n")
+ cat("\n*** Starting test 3.1 - 4 calls in one batch of 5 ***\n")
-args=list(inlist)
-arglist <- rep(list(args),4)
+ args=list(inlist)
+ arglist <- rep(list(args),4)
-swiftres = swiftapply(listfunc,arglist,callsperbatch=5)
+ swiftres = swiftapply(listfunc,arglist,callsperbatch=5)
-diffs <- 0
-for(i in 1:length(swiftres) ) {
- if( !all.equal(swiftres[[i]],localres) ) {
- diffs <- diffs + 1
- if( diffs < 10 ) cat(sprintf("res[%d=%s\n",i,format(swiftres[[i]])))
- }
-}
+ diffs <- 0
+ for(i in 1:length(swiftres) ) {
+ if( !all.equal(swiftres[[i]],localres) ) {
+ diffs <- diffs + 1
+ if( diffs < 10 ) cat(sprintf("res[%d=%s\n",i,format(swiftres[[i]])))
+ }
+ }
-if(diffs == 0) {
- cat("\n==> test 3.1 passed\n")
-} else {
- cat("\n!!!==> test 3.1 failed.\n")
- cat(sprintf(" %d result elements failed to match.\n",diffs));
- failures=failures+1
-}
+ if(diffs == 0) {
+ cat("\n==> test 3.1 passed\n")
+ } else {
+ cat("\n!!!==> test 3.1 failed.\n")
+ cat(sprintf(" %d result elements failed to match.\n",diffs));
+ failures=failures+1
+ }
-cat("\n*** Starting test 3.2 - 99 calls in batches of 11 ***\n")
+ cat("\n*** Starting test 3.2 - 99 calls in batches of 11 ***\n")
-args=list(inlist)
-arglist <- rep(list(args),99)
+ args=list(inlist)
+ arglist <- rep(list(args),99)
-swiftres = swiftapply(listfunc,arglist,callsperbatch=11)
+ swiftres = swiftapply(listfunc,arglist,callsperbatch=11)
-diffs <- 0
-for(i in 1:length(swiftres) ) {
- if( !all.equal(swiftres[[i]],localres) ) {
- diffs <- diffs + 1
- if( diffs < 10 ) cat(sprintf("res[%d]=%s\n",i,format(swiftres[[i]])))
- }
-}
+ diffs <- 0
+ for(i in 1:length(swiftres) ) {
+ if( !all.equal(swiftres[[i]],localres) ) {
+ diffs <- diffs + 1
+ if( diffs < 10 ) cat(sprintf("res[%d]=%s\n",i,format(swiftres[[i]])))
+ }
+ }
-if(diffs == 0) {
- cat("\n==> test 3.2 passed\n")
-} else {
- cat("\n!!!==> test 3.2 failed.\n")
- cat(sprintf(" %d result elements failed to match.\n",diffs));
- failures=failures+1
-}
+ if(diffs == 0) {
+ cat("\n==> test 3.2 passed\n")
+ } else {
+ cat("\n!!!==> test 3.2 failed.\n")
+ cat(sprintf(" %d result elements failed to match.\n",diffs));
+ failures=failures+1
+ }
-##### Test Group 4 # test initialexpr string
+ ##### Test Group 4 # test initialexpr string
-cat("\n*** Starting test group 4 - test remote R service initialization string ***\n")
+ cat("\n*** Starting test group 4 - test remote R service initialization string ***\n")
-swiftTest_4.1()
-swiftTest_4.2()
+ swiftTest_4.1()
+ swiftTest_4.2()
-##### Test Group 5 # test error handling
+ ##### Test Group 5 # test error handling
-cat("\n*** Starting test group 5 - test remote R service error ***\n")
+ cat("\n*** Starting test group 5 - test remote R service error ***\n")
-arglist = list(list(1.0),list(2.0),list("3.0"),list(4.0),list(5.0))
+ arglist = list(list(1.0),list(2.0),list("3.0"),list(4.0),list(5.0))
-cat("\nTest of swiftapply(sumivars,arglist)\n")
-swiftres = swiftapply(log,arglist)
-cat("Swift result:\n")
-print(swiftres)
+ cat("\nTest of swiftapply(sumivars,arglist)\n")
+ swiftres = swiftapply(log,arglist)
+ cat("Swift result:\n")
+ print(swiftres)
-goodres = c("numeric","numeric","try-error","numeric","numeric")
+ goodres = c("numeric","numeric","try-error","numeric","numeric")
-diffs <- 0
-for(i in 1:length(swiftres) ) {
- if( class(swiftres[[i]]) != goodres[i] ) {
- diffs <- diffs + 1
- if( diffs < 10 ) cat(sprintf("res[%d]=%s\n",i,format(swiftres[[i]])))
- }
-}
+ diffs <- 0
+ for(i in 1:length(swiftres) ) {
+ if( class(swiftres[[i]]) != goodres[i] ) {
+ diffs <- diffs + 1
+ if( diffs < 10 ) cat(sprintf("res[%d]=%s\n",i,format(swiftres[[i]])))
+ }
+ }
-if(diffs == 0) {
- cat("\n==> test 5.1 passed\n")
-} else {
- cat("\n!!!==> test 5.1 failed.\n")
- cat(sprintf(" %d result elements failed to match.\n",diffs));
- failures=failures+1
-}
+ if(diffs == 0) {
+ cat("\n==> test 5.1 passed\n")
+ } else {
+ cat("\n!!!==> test 5.1 failed.\n")
+ cat(sprintf(" %d result elements failed to match.\n",diffs));
+ failures=failures+1
+ }
-endTime <- proc.time()[["elapsed"]]
-runTime <- endTime - startTime
+ endTime <- proc.time()[["elapsed"]]
+ runTime <- endTime - startTime
-cat("\n\n ===> Total elapsed test time = ",runTime," seconds.\n\n")
-
+ cat("\n\n ===> Total elapsed test time = ",runTime," seconds.\n\n")
+ if (!is.null(testPid))
+ swiftShutdown(pid=testPid)
} # end function runAllTests
#options(swift.site="local")
Modified: SwiftApps/SwiftR/Swift/R/Workers.R
===================================================================
--- SwiftApps/SwiftR/Swift/R/Workers.R 2011-01-20 21:31:55 UTC (rev 4019)
+++ SwiftApps/SwiftR/Swift/R/Workers.R 2011-01-21 00:31:06 UTC (rev 4020)
@@ -23,6 +23,8 @@
# DEBUG, TRACE
# Options which are server and site-specific:
# project, queue
+ # Swift returns a descriptor for the worker process, which
+ # can be passed to swiftShutdown to stop the worker.
# In case it was somehow deleted
@@ -86,16 +88,26 @@
# add hook to ensure child process will be killed when
# this process exits
addHook()
+
+ # Sleep to give start-swift time to set up fifos,etc
+ Sys.sleep(2)
+ return (output)
}
-swiftShutdown <- function()
+swiftShutdown <- function(pid=NULL)
{
- if (is.null(.swift.workers)) {
- return
+ if (is.null(pid)) {
+ if (is.null(.swift.workers)) {
+ return
+ }
+ workers = .swift.workers
}
+ else {
+ workers=pid
+ }
cat("Shutting down Swift worker processes\n")
# shut down all worker processes using kill
- for (pid in .swift.workers) {
+ for (pid in workers) {
cmdString <- file.path(.find.package("Swift"), "exec/killtree")
killCmd <- paste(cmdString,pid)
system(killCmd, wait=FALSE)
Modified: SwiftApps/SwiftR/Swift/exec/fifowrite
===================================================================
--- SwiftApps/SwiftR/Swift/exec/fifowrite 2011-01-20 21:31:55 UTC (rev 4019)
+++ SwiftApps/SwiftR/Swift/exec/fifowrite 2011-01-21 00:31:06 UTC (rev 4020)
@@ -28,8 +28,10 @@
# fork off a process to read from fifo and write to
# temp file.
-# Cat reads from stdin and writes to fifo
-echo "$msg" > "$fifo" &
+# echo message to fifo, fork off a process. The remainder
+# of the code handles the case where there is nothing to
+# read it and echo blocks
+echo -n "$msg" > "$fifo" &
catpid=$!
tout_pid=""
if [ "$timeout" -gt 0 ];
Modified: SwiftApps/SwiftR/Swift/exec/killtree
===================================================================
--- SwiftApps/SwiftR/Swift/exec/killtree 2011-01-20 21:31:55 UTC (rev 4019)
+++ SwiftApps/SwiftR/Swift/exec/killtree 2011-01-21 00:31:06 UTC (rev 4020)
@@ -1,6 +1,5 @@
#!/bin/sh
tokill=$1
-echo $tokill
while [ ! -z $tokill ]
do
children=""
More information about the Swift-commit
mailing list