Author: bleny Date: 2010-07-22 14:45:54 +0200 (Thu, 22 Jul 2010) New Revision: 107 Url: http://nuiton.org/repositories/revision/diswork/107 Log: bug fix Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemon.java trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemon.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemon.java 2010-07-22 10:22:03 UTC (rev 106) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemon.java 2010-07-22 12:45:54 UTC (rev 107) @@ -670,25 +670,25 @@ throw new NullPointerException("job is null"); } - if (isFinished(job)) { - Map<String, InputStream> results = new HashMap<String, InputStream>(); - for (String fileName : job.getOutput()) { - String jobPath = getPathForJob(job); - try { + if (!isFinished(job)) { + throw new DisworkException("can't get results for unfinished job " + + job); + } + + Map<String, InputStream> results = new HashMap<String, InputStream>(); + for (String fileName : job.getOutput()) { + String jobPath = getPathForJob(job); + try { InputStream result = fileSystem.read(jobPath + "/" + fileName); results.put(fileName, result); - } catch (DisworkFileSystemException e) { - log.error("file system error ", e); - throw new DisworkException("file system error ", e); - } catch (FileNotFoundException e) { - log.warn("expected output file was not found : " + fileName, e); - } + } catch (DisworkFileSystemException e) { + log.error("file system error ", e); + throw new DisworkException("file system error ", e); + } catch (FileNotFoundException e) { + log.warn("expected output file was not found : " + fileName, e); } - return results; - } else { - throw new DisworkException("can't get results for unfinished job " - + job); - } + } + return results; } Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java 2010-07-22 10:22:03 UTC (rev 106) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java 2010-07-22 12:45:54 UTC (rev 107) @@ -48,7 +48,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.nuiton.diswork.daemon.ActivityStrategy.ActivityStrategies; -import org.nuiton.diswork.daemon.WorkersManager.Worker; import org.nuiton.diswork.fs.DisworkFileSystem; import org.nuiton.diswork.fs.DisworkFileSystemException; import org.nuiton.util.FileUtil; @@ -269,9 +268,10 @@ } } } catch (IOException e) { - // FIXME 20100701 bleny throw exception + // may occur if process is destroyed log.warn("error while reading the output of the subprocess", e); } finally { + // close file to make it contain as much data as possible try { if (wr != null) { wr.close(); @@ -577,27 +577,26 @@ int returnValue = currentProcess.waitFor(); log.info("process ended"); - // process returned and was not interrupted, output files - // are interesting in both successful and failure case - // upload them - stageOutputFiles(); - - // now check is the process ended as a success or a failure - // and act accordingly - if (returnValue == 0) { - // job is successful - jobIsSuccessful(currentJobPath); + if (shouldStop) { + // process has returned because stop() called Process.destroy() + // so it's not job's fault and should not be considered has a failure + jobIsInterrupted(currentJobPath); } else { - // check if it's due to a destroy call - if (shouldStop) { - // process has returned because stop() called Process.destroy() - // so it's not job's fault and should not be considered has a failure - jobIsInterrupted(currentJobPath); + + // process returned and was not interrupted, output files + // are interesting in both successful and failure case + // upload them + stageOutputFiles(); + + if (returnValue == 0) { + // job is successful + jobIsSuccessful(currentJobPath); } else { // job is a failure, the process returned an error jobIsFailed(currentJobPath); } } + } catch (InterruptedException e) { // job was interrupted maybe by a call to WokersManager#stop() log.debug("process was interrupted", e); @@ -820,7 +819,7 @@ String newJobPath = newDir + "/" + DisworkDaemon.newJobLinkName(); try { - // FIXME 20100720 bleny really useful to move to FAILED_3, a dir never read by anyone + // FIXME 20100720 bleny really useful to move to FAILED_3 ? a dir never read by anyone fileSystem.move(jobPath, newJobPath); log.info("moved " + jobPath + " to " + newJobPath); } catch (DisworkFileSystemException e) { @@ -952,14 +951,25 @@ throws DisworkException { // method is synchronized to prevent multiple workers to download // the same application at the same time + + // a file on local File system is a copy of this application if (!applicationCache.exists()) { applicationCache.mkdirs(); } File cachedApplicationData = new File(applicationCache, applicationName + "-" + applicationVersion + ".zip"); - if (!cachedApplicationData.exists()) { + + if (cachedApplicationData.exists()) { + // file is already available locally, it has been downloaded before + // cache worked + log.debug("cache matches for " + applicationName + "-" + applicationVersion); + } else { + // cache doesn't contains a copy of this application, creating one + // by downloading the application from diswork FS + log.debug("cache fail for " + applicationName + "-" + applicationVersion); synchronized (cachedApplicationData) { + // getting application data from diswork File System String applicationPath = DisworkDaemon.getPathForDependency( applicationName, applicationVersion); InputStream applicationData = null; @@ -974,7 +984,8 @@ } finally { IOUtils.closeQuietly(applicationData); } - + + // writing data to local file system OutputStream out = null; try { cachedApplicationData.createNewFile(); @@ -990,8 +1001,6 @@ IOUtils.closeQuietly(out); } } - } else { - log.debug("cache matches for " + applicationName + "-" + applicationVersion); } return cachedApplicationData; } @@ -1055,8 +1064,11 @@ worker.shouldStop = true; } + // workers should not take new jobs activeNoActivityStrategy(); + // killing all the processes, workers waiting for the end of the process + // will wake up for (Worker worker : workers) { if (worker.currentProcess != null) { log.debug("killing " + worker + " process"); @@ -1064,19 +1076,21 @@ } } + // empty the application-cache if (applicationCache.exists()) { FileUtil.deleteRecursively(applicationCache); } - // waiting for them to actually have finished + // waiting for all the workers to actually have finished for (Worker worker : workers) { while (worker.isAlive()) { + log.debug("waiting for " + worker + " to return"); + // worker may be sleeping, wake it up + synchronized (sem) { + sem.notifyAll(); + } + try { - // worker may be sleeping - synchronized (sem) { - sem.notifyAll(); - } - log.debug("waiting for " + worker + " to return"); Thread.sleep(1000); } catch (InterruptedException e) { log.warn("interrupted while waiting for a worker to " +