- Diswork-commits - mailman.cloud.codelutin.com

r87 - trunk
by bleny＠users.nuiton.org 25 Jun '10

25 Jun '10

Author: bleny Date: 2010-06-25 11:38:13 +0200 (Fri, 25 Jun 2010) New Revision: 87 Url: http://nuiton.org/repositories/revision/diswork/87 Log: changement de version xmlstore Modified: trunk/pom.xml Modified: trunk/pom.xml =================================================================== --- trunk/pom.xml 2010-06-24 14:13:06 UTC (rev 86) +++ trunk/pom.xml 2010-06-25 09:38:13 UTC (rev 87) @@ -82,7 +82,7 @@ <dependency> <groupId>org.planx</groupId> <artifactId>koala-xmlstore</artifactId> - <version>0.4.12_cl20100612</version> + <version>0.4.12_cl20100625</version> </dependency> <dependency> <groupId>commons-digester</groupId>

1 0

r86 - in trunk: . diswork-daemon diswork-daemon/src/main/java/org/nuiton/diswork/daemon diswork-daemon/src/test/java/org/nuiton/diswork/daemon diswork-daemon/src/test/resources diswork-fs/src/main/java/org/nuiton/diswork/fs diswork-fs/src/test/java/org/nuiton/diswork/fs diswork-fs/src/test/java/org/nuiton/diswork/fs/storage diswork-fs/src/test/resources
by bleny＠users.nuiton.org 24 Jun '10

24 Jun '10

Author: bleny Date: 2010-06-24 16:13:06 +0200 (Thu, 24 Jun 2010) New Revision: 86 Url: http://nuiton.org/repositories/revision/diswork/86 Log: lecture continue de la sortie standard, tests avec plusieurs d?\195?\169mons, m?\195?\160j pom (mina) Added: trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonMultipleNodesTest.java Modified: trunk/diswork-daemon/pom.xml trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemonRunner.java trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkSimpleClient.java trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java trunk/diswork-daemon/src/test/resources/log4j.properties trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/AbstractDisworkMapTest.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/KademliaDisworkMapTest.java trunk/diswork-fs/src/test/resources/log4j.properties trunk/pom.xml Modified: trunk/diswork-daemon/pom.xml =================================================================== --- trunk/diswork-daemon/pom.xml 2010-06-21 10:14:34 UTC (rev 85) +++ trunk/diswork-daemon/pom.xml 2010-06-24 14:13:06 UTC (rev 86) @@ -42,6 +42,10 @@ <groupId>commons-digester</groupId> <artifactId>commons-digester</artifactId> </dependency> + <dependency> + <groupId>org.freemarker</groupId> + <artifactId>freemarker</artifactId> + </dependency>  <dependency> Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemonRunner.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemonRunner.java 2010-06-21 10:14:34 UTC (rev 85) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemonRunner.java 2010-06-24 14:13:06 UTC (rev 86) @@ -50,6 +50,7 @@ Integer.parseInt(args[1]))); } config.setActivityStrategy("unlimited"); + config.fileSystemConfig.setBlockSize(65000); DisworkDaemon daemon = new DisworkDaemon(config); } } Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkSimpleClient.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkSimpleClient.java 2010-06-21 10:14:34 UTC (rev 85) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkSimpleClient.java 2010-06-24 14:13:06 UTC (rev 86) @@ -28,7 +28,7 @@ import java.io.File; import java.io.FileInputStream; import java.io.FileOutputStream; -import java.io.IOException; +import java.io.InputStream; import java.io.InputStreamReader; import java.util.Map; @@ -36,7 +36,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.nuiton.diswork.fs.DisworkFileSystemConfig; -import org.nuiton.util.FileUtil; /** * @@ -80,36 +79,49 @@ } public static void isisSubmit() throws Exception { - File isis = new File("/tmp/isis/isis-fish-3.3.0.3-bin.zip"); - daemon.submitApplication("isis-fish", "3.3.0.3", new FileInputStream(isis)); + File isis = new File("/tmp/isis/isis-fish-3.3.0.4-dev3069-bin.zip"); + // File isis = new File("/tmp/isis/fake-app-1.0.zip"); + + InputStream in = new FileInputStream(isis); + log.info("isis is " + in.available() + " bytes"); + daemon.submitApplication("isis-fish", "3.3.0.4-dev3069", in); JobDescription jobDescription = new JobDescription(); - jobDescription.setJobName("Mon Job Isis"); + jobDescription.setApplication("isis-fish", "3.3.0.4-dev3069"); - jobDescription.setApplication("isis-fish", "3.3.0.3"); + File input = new File("/tmp/isis/isisfish-simulation-ica10ans.zip"); + jobDescription.addInput("isisfish-simulation-ica10ans.zip", new FileInputStream(input)); + jobDescription.addOutput("isisfish-simulation-ica10ans-result.zip"); - File input = new File("/tmp/isis/sim_test-gdg-3.2-3.3.zip"); - jobDescription.addInput("sim_test-gdg-3.2-3.3.zip", new FileInputStream(input)); - jobDescription.addOutput("sim_test-gdg-3.2-3.3-result.zip"); + jobDescription.setCommandLine( + "%java -Xmx2500M -jar isis-fish-3.3.0.4-SNAPSHOT/isis-fish-3.3.0.4-SNAPSHOT.jar" + + " --option launch.ui false" + + " --option perform.vcsupdate false" + + " --option perform.migration false" + + " --option perform.cron false" + + " --simulateRemotelly my_isis_job isisfish-simulation-ica10ans.zip" + + " isisfish-simulation-ica10ans-result.zip"); - jobDescription.setCommandLine("%java -Xmx2500M -jar isis-fish*.jar" + - " --option launch.ui false --option perform.vcsupdate false" + - " --option perform.migration false --option perform.cron false" - + " --simulateRemotelly my_isis_job sim_test-gdg-3.2-3.3.zip" - + " sim_test-gdg-3.2-3.3-result.zip"); + jobDescription.setStandardOutput("output.txt"); + jobDescription.addOutput("output.txt"); daemon.submitJob(jobDescription); while (! daemon.isFinished(jobDescription)) { - Thread.sleep(5 * 1000); + Thread.sleep(15 * 1000); System.out.print("."); } - + + System.out.println("getting result"); File result = new File("/tmp/isis/sim_test-gdg-3.2-3.3-result.zip"); + IOUtils.copy(daemon.getResults(jobDescription).get("isisfish-simulation-ica10ans-result.zip"), new FileOutputStream(result)); + + System.out.println("getting output"); + result = new File("/tmp/isis/output.txt"); + IOUtils.copy(daemon.getResults(jobDescription).get("output.txt"), new FileOutputStream(result)); - IOUtils.copy(daemon.getResults(jobDescription).get("sim_test-gdg-3.2-3.3-result.zip"), new FileOutputStream(result)); - + System.out.println("end"); } /** @@ -124,6 +136,7 @@ DisworkFileSystemConfig.newKademliaDisworkConfig(args[0], port)); config.setActivityStrategy("none"); config.setUsedPort(30000); + config.fileSystemConfig.setBlockSize(65000); daemon = new DisworkDaemon(config); //userPrompt(); 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-06-21 10:14:34 UTC (rev 85) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java 2010-06-24 14:13:06 UTC (rev 86) @@ -24,13 +24,17 @@ */ package org.nuiton.diswork.daemon; +import java.io.BufferedReader; +import java.io.BufferedWriter; import java.io.File; import java.io.FileInputStream; import java.io.FileNotFoundException; import java.io.FileOutputStream; import java.io.IOException; import java.io.InputStream; +import java.io.InputStreamReader; import java.io.OutputStream; +import java.io.OutputStreamWriter; import java.net.URL; import java.util.ArrayList; import java.util.Collections; @@ -127,6 +131,81 @@ /** set this field to true will make run() return and thread stop */ protected boolean shouldStop = false; + /** read the standard output of the subprocess + * + * By reading the standard output, this thread has multiple goals : + * <ul> + * <li>if the process produces many data on standard output, the + * process will block (and waitfor will never return). This + * thread, by reading the output, unblock it. + * http://www.javaworld.com/javaworld/jw-12-2000/jw-1229-traps.html?page=4 + * </li> + * <li>if asked, write the output to a file</li> + * <li>add the output to the log</li> + * </ul> + * + * @author bleny + * + */ + protected class OutputReader extends Thread { + + protected InputStream output; + protected OutputStream outputFile; + + /** + * @param output the stream of the standard output of the process + * @param outputFile the file where to copy, null if you don't + * care about the what is on the standard output + */ + public OutputReader(InputStream output, OutputStream outputFile) { + this.output = output; + this.outputFile = outputFile; + } + + @Override + public void run() { + InputStreamReader osr = new InputStreamReader(output); + BufferedReader br = new BufferedReader(osr); + + BufferedWriter wr = null; + OutputStreamWriter outputStreamWriter = null; + if (outputFile != null) { + outputStreamWriter = new OutputStreamWriter(outputFile); + wr = new BufferedWriter(outputStreamWriter); + } + + String line = null; + try { + // read a line from standard output + while ((line = br.readLine()) != null) { + // copy this line to output file + if (wr != null) { + wr.write(line + "\n"); + } + + // add reading line to logging output + if (log.isInfoEnabled()) { + log.debug(this.toString() + ">" + line); + } + } + } catch (IOException e) { + log.warn("error while reading the output of the subprocess", e); + } finally { + try { + if (wr != null) { + wr.close(); + } + if (outputStreamWriter != null) { + outputStreamWriter.close(); + } + } catch (IOException e) { + log.warn("error while closing the output of the subprocess", e); + } + } + } + + } + /** * this method add a line to a job-specific log * @param jobPath the path to the job concerned @@ -142,6 +221,24 @@ fileSystem.write(logPath, IOUtils.toInputStream(newLog)); } + protected void streamCopy(InputStream in, OutputStream out) throws IOException { + System.out.println("hop"); + int read = 0; + int totalRead = 0; + byte[] buffer = new byte[50000]; + // BufferedInputStream hop = new BufferedInputStream(in); + + while ((read = in.read(buffer)) != -1) { + + totalRead += read; + System.out.println("read " + read + " bytes (total=" + totalRead + ")"); + + out.write(buffer, 0, read); + } + in.close(); + out.close(); + } + /** * Download all the files needed for a job in a temp directory, run * the job, wait for it to end, write all the results. Mark the job @@ -189,14 +286,20 @@ application.createNewFile(); log.info("will create " + application.getAbsolutePath()); OutputStream out = new FileOutputStream(application); - IOUtils.copy(applicationData, out); + log.debug("starting copy of " + applicationData.available() + " bytes"); + // IOUtils.copy(applicationData, out); + streamCopy(applicationData, out); + log.info("unzip application start"); // unzip application ZipUtil.uncompress(application, jobDir); + log.info("unzip application finished"); } else { log.info("no dependency specified for " + jobDescription); } + // staging input files for (String fileName : jobDescription.getStagingInput()) { + log.info("staging " + fileName); File localCopy = new File(jobDir, fileName); localCopy.createNewFile(); InputStream source = null; @@ -212,6 +315,7 @@ IOUtils.copy(source, new FileOutputStream(localCopy)); } + log.info("preparing the job"); // prepare and run it the job String commandLine = jobDescription.getCommandLine(); String[] commandLineElements = commandLine.split(" "); @@ -220,10 +324,25 @@ builder.redirectErrorStream(true); log.info("calling " + commandLine); Process job = builder.start(); + + // dump the standard output in a file + String standardOutputFileName = jobDescription.getStandardOutput(); + log.info("standardOutputFileName is " + standardOutputFileName); + OutputStream outputFileStream = null; + if (standardOutputFileName != null) { + File outputFile = new File(jobDir, standardOutputFileName); + log.info("writing standard output in " + outputFile); + outputFileStream = new FileOutputStream(outputFile); + } + + OutputReader outputReader = new OutputReader(job.getInputStream(), outputFileStream); + outputReader.start(); // plugging a file on the standard input String standardInputFileName = jobDescription.getStandardInput(); if (standardInputFileName != null) { + log.info("writing " + standardInputFileName + " on standard " + + "input"); InputStream input = new FileInputStream( new File(jobDir, standardInputFileName)); IOUtils.copy(input, job.getOutputStream()); @@ -232,6 +351,7 @@ // run the process and wait for it to return int exitValue = -1; try { + log.info("waiting for the end of the process"); exitValue = job.waitFor(); } catch (InterruptedException e) { log.error("job " + jobDescription + " was interupted", e); @@ -240,27 +360,19 @@ exitValue = 1; } - // dump the standard output in a file - String standardOutputFileName = jobDescription.getStandardOutput(); - log.info("standardOutputFileName is " + standardOutputFileName); - if (standardOutputFileName != null) { - File outputFile = new File(jobDir, standardOutputFileName); - OutputStream output = new FileOutputStream(outputFile); - log.info("writing standard output in " + outputFile); - IOUtils.copy(job.getInputStream(), output); - } + log.info("job returned " + exitValue); // output file staging for (String fileName : jobDescription.getStagingOutput()) { + log.info("staging file " + fileName); File localCopy = new File(jobDir, fileName); // FIXME 20100616 bleny may not exists if job has fail InputStream localCopyStream = new FileInputStream(localCopy); String filePath = jobPath + "/" + fileName; - log.info("out-staging " + fileName); // erase before write if (fileSystem.exists(filePath)) { fileSystem.delete(filePath); Added: trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonMultipleNodesTest.java =================================================================== --- trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonMultipleNodesTest.java (rev 0) +++ trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonMultipleNodesTest.java 2010-06-24 14:13:06 UTC (rev 86) @@ -0,0 +1,40 @@ +package org.nuiton.diswork.daemon; + +import java.io.InputStream; + +import org.junit.Before; +import org.junit.Test; +import org.nuiton.diswork.fs.DisworkFileSystemConfig; + +public class DisworkDaemonMultipleNodesTest extends DisworkDaemonTest { + + /** another deamon + * this second deamon will be connected to {@link DisworkDaemonTest#daemon} + * and will to his jobs. + */ + protected static DisworkDaemon daemon2; + + @Before + @Override + public void setUp() throws Exception { + DisworkConfig config = DisworkConfig.newConfig(); + port += 1; + config.setUsedPort(port); + config.setActivityStrategy("none"); + + config.fileSystemConfig.setBlockSize(500); + daemon = new DisworkDaemon(config); + InputStream application = ClassLoader.getSystemResourceAsStream("fake-app-1.0.zip"); + daemon.submitApplication("fake-app", "1.0", application); + + + DisworkConfig config2 = DisworkConfig.newConfig(); + config2.setBootstrapIp(DisworkFileSystemConfig.getIp()); + config2.setBootstrapPort(port); + port += 1; + config.setUsedPort(port); + config.setActivityStrategy("unlimited"); + daemon2 = new DisworkDaemon(config2); + } + +} Modified: trunk/diswork-daemon/src/test/resources/log4j.properties =================================================================== --- trunk/diswork-daemon/src/test/resources/log4j.properties 2010-06-21 10:14:34 UTC (rev 85) +++ trunk/diswork-daemon/src/test/resources/log4j.properties 2010-06-24 14:13:06 UTC (rev 86) @@ -6,4 +6,5 @@ log4j.appender.stdout.layout.ConversionPattern=%d %5p [%t] (%F:%L) %M - %m%n # package level log4j.logger.org.nuiton.diswork.fs.storage.KademliaDisworkMap=INFO -log4j.logger.org.nuiton.diswork.daemon=INFO \ No newline at end of file +log4j.logger.org.nuiton.diswork.daemon=DEBUG +log4j.logger.org.planx.xmlstore.routing.messaging.MessageServer=INFO \ No newline at end of file Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java 2010-06-21 10:14:34 UTC (rev 85) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java 2010-06-24 14:13:06 UTC (rev 86) @@ -102,13 +102,16 @@ } public DisworkFileSystemConfig() { - // setDefaultOption("diswork.fs.blocks_size", "10485760"); // 10 MiB - setDefaultOption("diswork.fs.blocks_size", "50000"); + setDefaultOption("diswork.fs.blocks_size", "10485760"); // 10 MiB setDefaultOption("diswork.fs.map_type", "inmemory"); setDefaultOption("diswork.fs.use_port", port.toString()); } + public void setBlockSize(Integer size) { + setOption("diswork.fs.blocks_size", size.toString()); + } + public int getBlockSize() { return getOptionAsInt("diswork.fs.blocks_size"); } Modified: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java 2010-06-21 10:14:34 UTC (rev 85) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java 2010-06-24 14:13:06 UTC (rev 86) @@ -16,14 +16,11 @@ import java.util.List; import java.util.Random; -import javax.xml.transform.Source; - import org.apache.commons.io.FileUtils; import org.apache.commons.io.IOUtils; import org.junit.After; import org.junit.Before; import org.junit.Test; -import org.nuiton.diswork.fs.DisworkFileSystem; import org.nuiton.util.FileUtil; @@ -45,7 +42,7 @@ /** * The file will have this fixed size */ - static protected int randomFileSize = 11 * 1000 * 1000; + static protected int randomFileSize = 45 * 1000 * 1000; static protected DisworkFileSystem fileSystem; @@ -75,6 +72,7 @@ @After public void tearDown() throws Exception { + fileSystem.close(); // cleaning FileUtil.deleteRecursively(tempDirectoryPath); } Modified: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java 2010-06-21 10:14:34 UTC (rev 85) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java 2010-06-24 14:13:06 UTC (rev 86) @@ -7,9 +7,11 @@ import java.io.ByteArrayInputStream; import java.io.FileInputStream; +import java.io.InputStream; import java.util.ConcurrentModificationException; import org.apache.commons.io.IOUtils; +import org.apache.log4j.lf5.util.StreamUtils; import org.junit.Before; import org.junit.Test; import org.nuiton.diswork.fs.DisworkFileSystem; @@ -48,19 +50,41 @@ DisworkFileSystemConfig disworkConfig = DisworkFileSystemConfig.newKademliaDisworkConfig(bootstrapPort); DisworkFileSystem fileSystem2 = new DisworkFileSystem(disworkConfig); - - byte[] bytes = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}; - fileSystem.write("/my_file", new ByteArrayInputStream(bytes)); + InputStream source = null; + try { + source = new FileInputStream(randomFilePath); + fileSystem.write("/my_file", source); + } finally { + IOUtils.closeQuietly(source); + } assertTrue(fileSystem.exists("/my_file")); assertTrue(fileSystem2.exists("/my_file")); - assertEquals(1, fileSystem2.readDirectory("/").size()); - byte[] getResult = IOUtils.toByteArray(fileSystem2.read("/my_file")); + // now, the checks. We read the original file and the result of + // a read() and then compare it byte-to-byte + + InputStream readResult = null; + try { + source = new FileInputStream(randomFilePath); + readResult = fileSystem2.read("/my_file"); + + assertEquals(randomFileSize, source.available()); + assertEquals(randomFileSize, readResult.available()); + + assertTrue(IOUtils.contentEquals(source, readResult)); + /* + byte[] sourceAsBytes = IOUtils.toByteArray(source); + byte[] readResultAsBytes = IOUtils.toByteArray(readResult); - assertArrayEquals(bytes, getResult); + assertArrayEquals(sourceAsBytes, readResultAsBytes); + */ + } finally { + IOUtils.closeQuietly(source); + IOUtils.closeQuietly(readResult); + } } Modified: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/AbstractDisworkMapTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/AbstractDisworkMapTest.java 2010-06-21 10:14:34 UTC (rev 85) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/AbstractDisworkMapTest.java 2010-06-24 14:13:06 UTC (rev 86) @@ -1,10 +1,13 @@ package org.nuiton.diswork.fs.storage; import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import java.util.Arrays; +import java.util.Random; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -96,4 +99,55 @@ } + @Test + public void testStoreLargeValue() { + + // 237499 + // 312499 + // 351561 + // 249999 + + int min = 0; + int max = 10 * 1000 * 1000; + + + int milieu = (min + max) / 2; + + while (max - min > 1) { + + + byte[] newBytes = new byte[milieu]; + Random random = new Random(); + random.nextBytes(newBytes); + + map1.put("test", newBytes); + + byte[] getResult = map1.get("test"); + assertNotNull(getResult); + assertEquals(milieu, getResult.length); + + getResult = map2.get("test"); + if (getResult == null) { + System.out.println(max + " trop grand"); + max = milieu; + } else { + System.out.println(min + " trop petit"); + min = milieu; + } + + milieu = (min + max) / 2; + System.out.println(min + " < " + milieu + " < " + max); + } + + System.out.println(milieu); + /* + assertNotNull(getResult); + assertEquals(bytesSize, getResult.length); + + assertArrayEquals(newBytes, getResult); + */ + + + } + } Modified: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/KademliaDisworkMapTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/KademliaDisworkMapTest.java 2010-06-21 10:14:34 UTC (rev 85) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/KademliaDisworkMapTest.java 2010-06-24 14:13:06 UTC (rev 86) @@ -70,6 +70,7 @@ * a bad bootstrap * @throws Exception */ + /* @Test(expected = org.planx.xmlstore.routing.RoutingException.class) public void testBadBootrap() throws Exception { DisworkFileSystemConfig config1 = @@ -78,4 +79,5 @@ config1.setBootstrapPort(80); new KademliaDisworkMap(config1); } + */ } Modified: trunk/diswork-fs/src/test/resources/log4j.properties =================================================================== --- trunk/diswork-fs/src/test/resources/log4j.properties 2010-06-21 10:14:34 UTC (rev 85) +++ trunk/diswork-fs/src/test/resources/log4j.properties 2010-06-24 14:13:06 UTC (rev 86) @@ -7,4 +7,5 @@ # package level log4j.logger.org.nuiton.diswork.fs=WARN log4j.logger.org.nuiton.diswork.fs.storage.KademliaDisworkMap=INFO -log4j.logger.org.nuiton.diswork.fs.Demo=INFO \ No newline at end of file +log4j.logger.org.nuiton.diswork.fs.Demo=INFO +log4j.logger.org.planx.xmlstore.routing=DEBUG \ No newline at end of file Modified: trunk/pom.xml =================================================================== --- trunk/pom.xml 2010-06-21 10:14:34 UTC (rev 85) +++ trunk/pom.xml 2010-06-24 14:13:06 UTC (rev 86) @@ -1,5 +1,6 @@ <?xml version="1.0" encoding="UTF-8"?> -<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> +<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> @@ -32,7 +33,7 @@ <dependency> <groupId>log4j</groupId> <artifactId>log4j</artifactId> - <version>1.2.14</version> + <version>1.2.16</version> <scope>test</scope> </dependency> <dependency> @@ -88,7 +89,13 @@ <artifactId>commons-digester</artifactId> <version>2.0</version> </dependency> + <dependency> + <groupId>org.freemarker</groupId> + <artifactId>freemarker</artifactId> + <version>2.3.16</version> + </dependency> +  <dependency> <groupId>junit</groupId>

1 0

r85 - in trunk: . diswork-daemon/src/main/java/org/nuiton/diswork/daemon diswork-daemon/src/test/java/org/nuiton/diswork/daemon diswork-fs/src/main/java/org/nuiton/diswork/fs diswork-fs/src/main/java/org/nuiton/diswork/fs/storage diswork-fs/src/test/java/org/nuiton/diswork/fs
by bleny＠users.nuiton.org 21 Jun '10

21 Jun '10

Author: bleny Date: 2010-06-21 12:14:34 +0200 (Mon, 21 Jun 2010) New Revision: 85 Url: http://nuiton.org/repositories/revision/diswork/85 Log: documentation, client simple, lanceur de d?\195?\169mon Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/ActivityStrategy.java trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemon.java trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemonRunner.java trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkSimpleClient.java trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/JobDescription.java trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonTest.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/Storage.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java trunk/pom.xml Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/ActivityStrategy.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/ActivityStrategy.java 2010-06-19 18:55:14 UTC (rev 84) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/ActivityStrategy.java 2010-06-21 10:14:34 UTC (rev 85) @@ -30,6 +30,34 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +/** + * This interface introduce the concept of "activity strategy" in + * diswork, an activity strategy answer to "Can i run a job now ?". + * + * This provide a simple method that returns true if Diswork can run a job + * and take hardware resources. If false is returned, it means that resources + * are not available or we want diswork not to consume more resources. + * + * This interface come with 4 implementations. Two first are very simple, + * one is used when diswork is "shut down", it always returns false. The + * second is used when diswork is in "free-wheel" and take as much resources + * as jobs need (always return true). + * + * LimitedActivity implementation is user-activity aware and returns true + * only if it seems that the user incompletely use the resources of + * his computer. This implementation is based on the a load-average + * computation of the computer. If load-average is low, computer is + * idling : it returns true and diswork will use remaining resources. + * If load-average is high, false is returned and diswork don't make + * the computer over-loaded. + * + * ScheduledActivity implementation make diswork run jobs only at + * fixed period of time. An user can configure it to make + * diswork run job only at night are during the week-end, for example. + * + * @author bleny + * + */ public interface ActivityStrategy { /** use this strategy to never run a job */ @@ -48,7 +76,7 @@ } } - /** use this strategy to run a job only if load average is low */ + /** use this strategy to run a job only if computer is idling */ public static class LimitedActivity implements ActivityStrategy { private static final Log log = LogFactory.getLog(LimitedActivity.class); 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-06-19 18:55:14 UTC (rev 84) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemon.java 2010-06-21 10:14:34 UTC (rev 85) @@ -31,6 +31,7 @@ import java.lang.management.ManagementFactory; import java.lang.management.OperatingSystemMXBean; import java.net.UnknownHostException; +import java.util.ArrayList; import java.util.ConcurrentModificationException; import java.util.HashMap; import java.util.List; @@ -46,9 +47,88 @@ import org.nuiton.util.FileUtil; /** + * The diswork daemon is the gateway to the global diswork system. Instantiate + * this class creates a new node on the system. The new node can be used + * to submit jobs and monitor them. * + * For how-to use considerations see ... + * + * Here is a list of what the daemon do when booting : + * <ul> + * <li>join the network, given some information (IP+port) provided + * with diswork-config (a diswork-config is an instance of + * {@link DisworkConfig})</li> + * <li>define what is the ownedId for this node. It has to be + * recovered from the last run of the daemon. It's + * the link between two different sessions : if you submit a job, stop + * and restart the daemon with the same ownerId, you will find your + * job back. If ownerId can't be recovered or if it's the first + * run, an ownerId is generated and saved for future run.</li> + * <li>define the home-directory, it's deduced from the ownerId. Thus, if you + * change you ownerId, you'll change your home-dir and loose your jobs + * because they are all stored in your home-dir. If it's the first + * time the daemon is run by this ownerId, the homeDir is created</li> + * <li>If diswork is configured for, the daemon will start a workers-manager. + * This manager don't care about the jobs you submit but look + * for jobs that are proposed on the global system and try to + * help. For more details, see {@link WorkersManager}</li> + * <li>Write some stats about itself. Since all nodes do that, each nodes + * can performs statistics computation to estimate the powerfulness of + * the global diswork system. Each node write those info in a file + * in his home-dir at {@link #HARDINFO_PATH}</li> + * </ul> + * + * The main use of the daemon is to submit job to the global diswork + * system. + * + * A job is a task that need an application. An application is a simple file + * that should be provided to the daemon once for all + * (using {@link #submitApplication(String, String, InputStream)}. Then, all + * nodes that would like to perform this job can download the application. + * + * You can create your application ready for diswork, submit it to the daemon + * and then submit as much jobs as you want that depends of this application. + * Think an application as the common stuff (programs, scripts, files etc.) + * all the different job will need. An application is just a set of file. + * + * TODO 20100617 bleny explain it better + * + * Once the application has been made available, jobs can be created and + * submitted. To do so, a programmer must write a class that + * + * <ol> + * <li>may republish the application (it's always good to be sure an + * application is still present if the last time that it has + * been used is too long)</li> + * <li>Create a new {@link JobDescription} and set it right for his + * new job.</li> + * <li>Submit the job by calling {@link #submitJob(JobDescription)}</li> + * <li>Monitor the job with {@link #isFinished(JobDescription)}</li> + * <li>Check the job state with {@link #isFailed(JobDescription)} + * and {@link #isSuccessful(JobDescription)} and get the results with + * {@link #getResults(JobDescription)}</li> + * <li>Finally, remove the job</li> + * </ol> + * + * Jobs will be saved in home-dir, in a path generated with + * {@link #getPathForJob(JobDescription)}. In this directory will be placed + * <ul> + * <li>All the data provided with the job {@link JobDescription#inputData}</li> + * <li>The JSDL file that describe the job (placed according to the + * value of {@link #JSDL_PATH})</li> + * <li>A log file placed at {@link #LOG_PATH}, it will be empty at the + * submission but lines will be added each time a node try to do + * this job. If the job is finished, the log will contain "FINISHED", + * if it has been completed, it will contain "DONE", if it has been + * failed, it will contain "FAILED" one or few times.</li> + * </ul> + * + * The daemon should be closed properly by using the {@link #close()} method. + * It will release resources, update stats and stop the worker-manager. + * * As far as possible, the use of the file system follow the UNIX Filesystem - * Hierarchy Standard + * Hierarchy Standard. All usual paths used by Diswork are defined in + * constants. * * @see http://en.wikipedia.org/wiki/Filesystem_Hierarchy_Standard * @@ -87,6 +167,9 @@ /** a place where are all user-directories */ protected static final String HOME = "/home"; + + /** in home, where jobs should be placed */ + protected static final String JOBS_DIR = "jobs"; /** in a job directory, the place where the JSDL must be placed */ protected static final String JSDL_PATH = ".diswork/job.jsdl"; @@ -259,7 +342,7 @@ */ protected String getPathForJob(String jobId) { // all jobs are stored in home dir - return homeDir + "/" + jobId; + return homeDir + "/" + JOBS_DIR + "/" + jobId; } /** @@ -273,7 +356,7 @@ /** * Provide an application to all nodes. Once provided, all nodes will be - * able to perform a job with this application + * able to perform a job with this application. * * An application can be uploaded in different version. If a given * application in the given version is already on the File System, nothing @@ -285,7 +368,7 @@ * @throws DisworkException if an error occurs while uploading the file */ public void submitApplication(String applicationName, - String applicationVersion, InputStream applicationData) + String applicationVersion, InputStream applicationData) throws DisworkException { // the place where dependency will be stored @@ -310,6 +393,39 @@ } } + /** + * Returns a list with all jobs submitted before. + * @return + * @throws DisworkException + */ + public List<JobDescription> getAllJobs() throws DisworkException { + List<JobDescription> result = new ArrayList<JobDescription>(); + try { + List<String> jobs = fileSystem.readDirectory(homeDir + "/" + JOBS_DIR); + for (String jobId : jobs) { + String jsdl = IOUtils.toString(fileSystem.read( + getPathForJob(jobId) + "/" + JSDL_PATH)); + JobDescription jobDescription = JobDescription.parseJSDL(jsdl); + jobDescription.setJobId(jobId); + result.add(jobDescription); + } + } catch (IOException e) { + log.info("error file reading home-dir", e); + throw new DisworkException("error file reading home-dir", e); + } + return result; + } + + /** + * Cancel the submission of a job + * @param jobDescription + * @throws DisworkException + */ + public void deleteJob(JobDescription jobDescription) { + // TODO 20100618 bleny stub + throw new UnsupportedOperationException(); + } + public void submitJob(JobDescription jobDescription) throws DisworkException { if (jobDescription.getInputData().size() + jobDescription.getStagingInputUrls().size() @@ -413,6 +529,14 @@ return isFinished(job) && !isSuccessful(job); } + /** + * Must not be called until {@link #isFinished(JobDescription)} returns + * true for this job. + * @param job + * @return a map associating the file-name of a result to a stream to + * read the file + * @throws DisworkException if job is not finished + */ public Map<String, InputStream> getResults(JobDescription job) throws DisworkException { if (isFinished(job)) { Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemonRunner.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemonRunner.java 2010-06-19 18:55:14 UTC (rev 84) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemonRunner.java 2010-06-21 10:14:34 UTC (rev 85) @@ -42,10 +42,9 @@ public static void main(String[] args) throws Exception { DisworkConfig config = new DisworkConfig(); - if (args.length == 0) { config.setFileSystemConfig( DisworkFileSystemConfig.newKademliaDisworkConfig()); - } else if (args.length == 2){ + if (args.length == 2) { config.setFileSystemConfig( DisworkFileSystemConfig.newKademliaDisworkConfig(args[0], Integer.parseInt(args[1]))); Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkSimpleClient.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkSimpleClient.java 2010-06-19 18:55:14 UTC (rev 84) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkSimpleClient.java 2010-06-21 10:14:34 UTC (rev 85) @@ -25,6 +25,10 @@ package org.nuiton.diswork.daemon; import java.io.BufferedReader; +import java.io.File; +import java.io.FileInputStream; +import java.io.FileOutputStream; +import java.io.IOException; import java.io.InputStreamReader; import java.util.Map; @@ -32,36 +36,25 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.nuiton.diswork.fs.DisworkFileSystemConfig; +import org.nuiton.util.FileUtil; /** * * @author bleny */ public class DisworkSimpleClient { - + private static final Log log = LogFactory.getLog(DisworkSimpleClient.class); + + static DisworkDaemon daemon; - /** - * @param args - * @throws Exception - */ - public static void main(String[] args) throws Exception { + public static void userPrompt() throws Exception { - DisworkConfig config = new DisworkConfig(); - config.setFileSystemConfig( - DisworkFileSystemConfig.newKademliaDisworkConfig(args[0], - Integer.parseInt(args[1]))); - config.setActivityStrategy("none"); - config.setUsedPort(30000); - DisworkDaemon daemon = new DisworkDaemon(config); - - // prompt the user to enter their name - // open up standard input BufferedReader br = new BufferedReader(new InputStreamReader(System.in)); String userEntry = ""; - do { + while (!userEntry.equals("quit")) { System.out.print("diswork: "); userEntry = br.readLine(); @@ -83,8 +76,57 @@ System.out.println(IOUtils.toString(daemon.getResults(job).get("output.txt"))); } - - } while (!userEntry.equals("quit")); + } + } + + public static void isisSubmit() throws Exception { + File isis = new File("/tmp/isis/isis-fish-3.3.0.3-bin.zip"); + daemon.submitApplication("isis-fish", "3.3.0.3", new FileInputStream(isis)); + + JobDescription jobDescription = new JobDescription(); + + jobDescription.setJobName("Mon Job Isis"); + + jobDescription.setApplication("isis-fish", "3.3.0.3"); + + File input = new File("/tmp/isis/sim_test-gdg-3.2-3.3.zip"); + jobDescription.addInput("sim_test-gdg-3.2-3.3.zip", new FileInputStream(input)); + jobDescription.addOutput("sim_test-gdg-3.2-3.3-result.zip"); + + jobDescription.setCommandLine("%java -Xmx2500M -jar isis-fish*.jar" + + " --option launch.ui false --option perform.vcsupdate false" + + " --option perform.migration false --option perform.cron false" + + " --simulateRemotelly my_isis_job sim_test-gdg-3.2-3.3.zip" + + " sim_test-gdg-3.2-3.3-result.zip"); + + daemon.submitJob(jobDescription); + + while (! daemon.isFinished(jobDescription)) { + Thread.sleep(5 * 1000); + System.out.print("."); + } + + File result = new File("/tmp/isis/sim_test-gdg-3.2-3.3-result.zip"); + + IOUtils.copy(daemon.getResults(jobDescription).get("sim_test-gdg-3.2-3.3-result.zip"), new FileOutputStream(result)); + + } + + /** + * @param args + * @throws Exception + */ + public static void main(String[] args) throws Exception { + DisworkConfig config = new DisworkConfig(); + Integer port = Integer.parseInt(args[1]); + System.out.println("port = " + port); + config.setFileSystemConfig( + DisworkFileSystemConfig.newKademliaDisworkConfig(args[0], port)); + config.setActivityStrategy("none"); + config.setUsedPort(30000); + daemon = new DisworkDaemon(config); + //userPrompt(); + isisSubmit(); } } Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/JobDescription.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/JobDescription.java 2010-06-19 18:55:14 UTC (rev 84) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/JobDescription.java 2010-06-21 10:14:34 UTC (rev 85) @@ -52,7 +52,7 @@ * * <dl> * <dt>%java</dt> - * </dd>will be replaced by the actual path of java executable + * <dd>will be replaced by the actual path of java executable * into the JRE</dd> * </dl> * @@ -117,8 +117,8 @@ /** * constructor is protected to prevent bad jobIds. To get a JobDescription - * instance, a client should use the {@link DisworkDaemon#newJob()} - * factory method. The given instance will have a valid jobId + * instance, a client should use the {@link #JobDescription()} constructor + * The given instance will have a valid jobId when needed. * @param jobId */ protected JobDescription(String jobId) { @@ -194,16 +194,7 @@ return "job : " + jobName + " (" + jobId + ")"; } - protected static Integer count = 0; - protected static Map<String, JobDescription> map = new HashMap<String, JobDescription>(); - - public String toJSDL() { - /* - count += 1; - map.put(count.toString(), this); - return count.toString(); - */ - + public String toJSDL() { String jsdl = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n" + "<jsdl:JobDefinition xmlns=\"http://www.example.org/\"\n" @@ -288,7 +279,7 @@ Element jobName = jobIdentification.getChild("JobName", jsdlNamespace); result.setJobName(jobName.getText()); - + // application Element application = jobDescription.getChild("Application", jsdlNamespace); @@ -300,10 +291,10 @@ result.setApplication(applicationName.getText(), applicationVersion.getText()); } - + Element POSIXApplication = application.getChild("POSIXApplication", jsdlPosixNamespace); - + Element argument = POSIXApplication.getChild("Argument", jsdlPosixNamespace); result.setCommandLine(argument.getText()); @@ -318,22 +309,16 @@ if (output != null) { result.setStandardOutput(output.getText()); } - + // staging List<Element> dataStagings = jobDescription.getChildren ("DataStaging", jsdlNamespace); for (Element dataStaging : dataStagings) { Attribute type = dataStaging.getAttribute("type", disworkNamespace); - Element fileName = dataStaging.getChild("FileName", jsdlNamespace); - - if (type != null && "out".equals(type.getValue())) { - // type="out" - result.addOutput(fileName.getText()); - } else { - // type="in" + if ("in".equals(type.getValue())) { Element source = dataStaging.getChild("Source", jsdlNamespace); if (source != null) { @@ -342,14 +327,11 @@ } else { result.addInput(fileName.getText(), new NullInputStream(0)); } - - // type not set - if (type == null) { - result.addOutput(fileName.getText()); - } } + if ("out".equals(type.getValue())) { + result.addOutput(fileName.getText()); + } } - } catch (JDOMException e) { log.error("can't read malformed JSDL file", e); throw new IOException("can't read malformed JSDL file", e); @@ -359,7 +341,7 @@ } return result; } - + public void addInput(String fileName, URL source) { input.add(fileName); inputUrls.put(fileName, source); @@ -369,7 +351,7 @@ input.add(fileName); inputData.put(fileName, source); } - + public void addOutput(String fileName) { output.add(fileName); } 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-06-19 18:55:14 UTC (rev 84) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java 2010-06-21 10:14:34 UTC (rev 85) @@ -37,7 +37,6 @@ import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Random; import org.apache.commons.io.FilenameUtils; import org.apache.commons.io.IOUtils; @@ -48,23 +47,37 @@ import org.nuiton.util.ZipUtil; /** + * The workers-manager aims to run and manage the different workers. A worker + * is a thread that try to find a jobs and execute them. * + * The manager can manage multiple workers : one worker + * doing one job at a time. By running n workers, the daemon can run up + * to n job at the same time. The number of workers should be set carefully + * (according to the actual hardware resources available). The number of workers + * can be set through {@link DisworkConfig#setNumberOfWorkers(Integer)}. + * + * Workers are not always active, before trying to find and job, they check + * if they will have enough resources to execute the job without bugging + * the other processes of the host machine. To do so, the worker-manager + * allow to define an activity-strategy to be followed by each worker. The + * activity strategy tell the worker if they are allowed to take resources or + * not, different strategies are available (see {@link ActivityStrategy}), + * current strategy can be changed on the fly, while the daemon run. + * * @author bleny */ public class WorkersManager { + private static final Log log = LogFactory.getLog(WorkersManager.class); + + /** a job found in dir "key", should be move to "value" before running */ protected static Map<String, String> RUNNING_MOVE = new HashMap<String, String>(); + + /** a job found in dir "key" that fail should be moved to "value" */ protected static Map<String, String> FAILED_MOVE = new HashMap<String, String>(); - - // TODO 20100611 bleny make it configurable - /** time to wait beetween two look for a job */ - protected static final int JOB_WAIT = 10 * 1000; - // TODO 20100614 bleny make it configurable - /** after this time (ms), a job is considered as no longer running */ - protected static final long MAX_JOB_RUNNING_TIME = 24 * 60 * 60 * 1000; - static { + // initialize RUNNING_MOVE and FAILED_MOVE constants RUNNING_MOVE.put(DisworkDaemon.TODO, DisworkDaemon.TODO_RUNNING); RUNNING_MOVE.put(DisworkDaemon.FAILED_1, DisworkDaemon.FAILED_1_RUNNING); RUNNING_MOVE.put(DisworkDaemon.FAILED_2, DisworkDaemon.FAILED_2_RUNNING); @@ -76,23 +89,44 @@ FAILED_MOVE.put(DisworkDaemon.FAILED_1_RUNNING, DisworkDaemon.FAILED_2); FAILED_MOVE.put(DisworkDaemon.FAILED_2_RUNNING, DisworkDaemon.FAILED_3); } - - private static final Log log = LogFactory.getLog(WorkersManager.class); protected DisworkFileSystem fileSystem; protected DisworkConfig config; - // Pool of workers + /** a pool of workers */ protected List<Worker> workers = new ArrayList<Worker>(); + /** the current activity strategy followed by all the workers */ protected ActivityStrategy activityStrategy; + /** A worker search a job and execute it. + * + * Jobs are found on the file-system by browsing some special directories, + * like {@link DisworkDaemon#TODO}. They do not contains jobs but symbolic + * links to jobs. If a worker find a job, it must move it to another + * directory ({@link DisworkDaemon#TODO_RUNNING}) for no other worker to + * take it. Once finished, the job link, if successful, will be moved to + * {@link DisworkDaemon#DONE}. If failed, the link will be moved to + * {@link DisworkDaemon#FAILED_1} (read "failed once") for another worker + * to try it and move it the same may. If the job is failed three times, + * it's moved to {@link DisworkDaemon#FAILED_3} meaning it will not been + * tried again. + * + * @author bleny + */ protected class Worker extends Thread { - + + // TODO 20100611 bleny make it configurable + /** time to wait between two look for a job */ + protected static final int JOB_WAIT = 10 * 1000; + + // TODO 20100614 bleny make it configurable + /** after this time (ms), a job is considered as no longer running */ + protected static final long MAX_JOB_RUNNING_TIME = 24 * 60 * 60 * 1000; + + /** set this field to true will make run() return and thread stop */ protected boolean shouldStop = false; - - protected WorkersManager manager; - + /** * this method add a line to a job-specific log * @param jobPath the path to the job concerned @@ -135,10 +169,9 @@ log.info("will run job " + jobDescription); - // create temp dir - Random random = new Random(); - File jobDir = new File(config.getTempDirectory(), - String.valueOf(random.nextInt())); + // create temp dir + File jobDir = FileUtil.createTempDirectory("job", "", + new File(config.getTempDirectory())); jobDir.mkdirs(); // download application @@ -249,6 +282,10 @@ return success; } + /** + * In a directory, list the content, sort the content, and returns + * the first element. + */ protected String getFistJobName(String path) throws IOException { List<String> jobsNames = fileSystem.readDirectory(path); if (jobsNames.size() == 0) { @@ -369,19 +406,15 @@ } } + /** + * find + */ @Override public void run() { while (! shouldStop) { - if (manager.getActivityStrategy().canWork()) { + if (activityStrategy.canWork()) { findAJobAndRunIt(); } - try { - Thread.sleep(10*1000); - } catch (InterruptedException e) { - // TODO 20100615 bleny Auto-generated catch block - log.info("exception catch", e); - e.printStackTrace(); - } } } } @@ -389,7 +422,8 @@ public WorkersManager(DisworkFileSystem fileSystem, DisworkConfig config) { this.fileSystem = fileSystem; this.config = config; - + + // initialize activityStrategy according to config String initialStrategy = config.getActivityStrategy(); if ( "none".equals(initialStrategy)) { activeNoActivityStrategy(); @@ -405,14 +439,13 @@ activeNoActivityStrategy(); } + // start as many workers as needed log.info("will start " + config.getNumberOfWorkers() + " workers"); for (int i = 1 ; i <= config.getNumberOfWorkers() ; i++) { Worker worker = new Worker(); - worker.manager = this; worker.start(); workers.add(worker); } - } public void stop() { Modified: trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonTest.java =================================================================== --- trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonTest.java 2010-06-19 18:55:14 UTC (rev 84) +++ trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonTest.java 2010-06-21 10:14:34 UTC (rev 85) @@ -5,6 +5,7 @@ import java.io.InputStream; import java.net.URL; +import java.util.List; import java.util.Map; import org.apache.commons.io.IOUtils; @@ -39,6 +40,7 @@ JobDescription job = new JobDescription(); job.setCommandLine("java -version"); daemon.submitJob(job); + } @Test(expected = DisworkException.class) @@ -60,6 +62,14 @@ } assertTrue(daemon.isSuccessful(job)); + + // check getAllJobs return + List<JobDescription> currentJobs = daemon.getAllJobs(); + assertEquals(1, currentJobs.size()); + + JobDescription currentJob = currentJobs.get(0); + assertEquals(job.getJobId(), currentJob.getJobId()); + assertEquals(job.getCommandLine(), currentJob.getCommandLine()); } Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java 2010-06-19 18:55:14 UTC (rev 84) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java 2010-06-21 10:14:34 UTC (rev 85) @@ -45,11 +45,11 @@ * </dd> * </dl> * - * * This class provides utilities to write tests easily. * * You can get multiples diswork configs ready to use. All the instances uses * a new port and the local IP. + * * <pre> * c = newDisworkConfig(); // create a config for a bootstrap node * c2 = newDisworkConfig(c.getUsedPort()) // creates a config for a node that @@ -57,7 +57,6 @@ * // first node * c3 = newDisworkConfig(c.getUsedPort()) * </pre> - * */ public class DisworkFileSystemConfig extends ApplicationConfig { @@ -103,7 +102,8 @@ } public DisworkFileSystemConfig() { - setDefaultOption("diswork.fs.blocks_size", "10485760"); // 10 MiB + // setDefaultOption("diswork.fs.blocks_size", "10485760"); // 10 MiB + setDefaultOption("diswork.fs.blocks_size", "50000"); setDefaultOption("diswork.fs.map_type", "inmemory"); setDefaultOption("diswork.fs.use_port", port.toString()); Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/Storage.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/Storage.java 2010-06-19 18:55:14 UTC (rev 84) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/Storage.java 2010-06-21 10:14:34 UTC (rev 85) @@ -152,16 +152,16 @@ protected DisworkMap map; - protected DisworkFileSystemConfig disworkConfig; + protected DisworkFileSystemConfig config; - public Storage(DisworkFileSystemConfig disworkConfig, DisworkMap map) + public Storage(DisworkFileSystemConfig config, DisworkMap map) throws IOException { - this.disworkConfig = disworkConfig; + this.config = config; if (map == null) { // instantiating a map according to config directives - String mapType = disworkConfig.getOption("diswork.fs.map_type"); + String mapType = config.getOption("diswork.fs.map_type"); if (mapType == null) { log.info("no map type specified"); @@ -172,10 +172,10 @@ this.map = new InMemoryDisworkMap(); } else if ("pastry".equals(mapType)) { log.info("using Pastry map"); - this.map = new PastryDisworkMap(disworkConfig); + this.map = new PastryDisworkMap(config); } else if ("kademlia".equals(mapType)) { log.info("using Kademlia map"); - this.map = new KademliaDisworkMap(disworkConfig); + this.map = new KademliaDisworkMap(config); } } } else { @@ -191,8 +191,8 @@ } } - public Storage(DisworkFileSystemConfig disworkConfig) throws IOException { - this(disworkConfig, null); + public Storage(DisworkFileSystemConfig config) throws IOException { + this(config, null); } /** @@ -298,7 +298,7 @@ map.put(newDataKey, EntryUtil.stringToBytes(metaBlock)); // creating a buffer of the size of a block - int bufferSize = disworkConfig.getBlockSize(); + int bufferSize = config.getBlockSize(); byte[] buffer = new byte[bufferSize]; while ((readResult = value.read(buffer)) != -1) { Modified: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java 2010-06-19 18:55:14 UTC (rev 84) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java 2010-06-21 10:14:34 UTC (rev 85) @@ -16,12 +16,13 @@ import java.util.List; import java.util.Random; +import javax.xml.transform.Source; + import org.apache.commons.io.FileUtils; import org.apache.commons.io.IOUtils; import org.junit.After; import org.junit.Before; import org.junit.Test; -import org.junit.experimental.categories.Categories.ExcludeCategory; import org.nuiton.diswork.fs.DisworkFileSystem; import org.nuiton.util.FileUtil; @@ -44,7 +45,7 @@ /** * The file will have this fixed size */ - static protected int randomFileSize = 25 * 1000; + static protected int randomFileSize = 11 * 1000 * 1000; static protected DisworkFileSystem fileSystem; @@ -84,7 +85,13 @@ */ @Test public void testWrite() throws Exception { - fileSystem.write("/my_file", new FileInputStream(randomFilePath)); + InputStream source = null; + try { + source = new FileInputStream(randomFilePath); + fileSystem.write("/my_file", source); + } finally { + IOUtils.closeQuietly(source); + } } /** @@ -93,9 +100,15 @@ */ @Test public void testExists() throws Exception { - fileSystem.write("/my_file", new FileInputStream(randomFilePath)); - assertTrue(fileSystem.exists("/my_file")); - assertFalse(fileSystem.exists("/my_other_file")); + InputStream source = null; + try { + source = new FileInputStream(randomFilePath); + fileSystem.write("/my_file", new FileInputStream(randomFilePath)); + assertTrue(fileSystem.exists("/my_file")); + assertFalse(fileSystem.exists("/my_other_file")); + } finally { + IOUtils.closeQuietly(source); + } } /** @@ -117,25 +130,28 @@ byte[] bytes = new byte[1]; bytes[0] = -0x1; - InputStream source; + InputStream source = null; + try { + source = new ByteArrayInputStream(bytes); + fileSystem.write("/my_file", source); + } finally { + IOUtils.closeQuietly(source); + } - source = new ByteArrayInputStream(bytes); - fileSystem.write("/my_file", source); + try { + source = new ByteArrayInputStream(bytes); + InputStream readResult = fileSystem.read("/my_file"); - source.close(); + int read = 0; + byte[] b = new byte[1]; - - source = new ByteArrayInputStream(bytes); - InputStream readResult = fileSystem.read("/my_file"); - - int read = 0; - byte[] b = new byte[1]; - - read = readResult.read(b); + read = readResult.read(b); - assertEquals(1, read); - assertArrayEquals(bytes, b); - + assertEquals(1, read); + assertArrayEquals(bytes, b); + } finally { + IOUtils.closeQuietly(source); + } } /** @@ -147,31 +163,35 @@ @Test public void testWriteRead() throws Exception { - InputStream source = new FileInputStream(randomFilePath); - - fileSystem.write("/my_file", source); + InputStream source = null; + try { + source = new FileInputStream(randomFilePath); + fileSystem.write("/my_file", source); + } finally { + IOUtils.closeQuietly(source); + } - source.close(); - - InputStream readResult; // now, the checks. We read the original file and the result of // a read() and then compare it byte-to-byte - - source = new FileInputStream(randomFilePath); - readResult = fileSystem.read("/my_file"); - assertEquals(randomFileSize, source.available()); - assertEquals(randomFileSize, readResult.available()); + InputStream readResult = null; + try { + source = new FileInputStream(randomFilePath); + readResult = fileSystem.read("/my_file"); + + assertEquals(randomFileSize, source.available()); + assertEquals(randomFileSize, readResult.available()); - byte[] sourceAsBytes = IOUtils.toByteArray(source); - byte[] readResultAsBytes = IOUtils.toByteArray(readResult); + byte[] sourceAsBytes = IOUtils.toByteArray(source); + byte[] readResultAsBytes = IOUtils.toByteArray(readResult); - assertArrayEquals(sourceAsBytes, readResultAsBytes); + assertArrayEquals(sourceAsBytes, readResultAsBytes); + } finally { + IOUtils.closeQuietly(source); + IOUtils.closeQuietly(readResult); + } - source.close(); - readResult.close(); - } /** @@ -180,22 +200,34 @@ */ @Test(expected = IOException.class) public void testFailAtWrite() throws Exception { - fileSystem.write("/my_folder/my_file", - new FileInputStream(randomFilePath)); + InputStream source = null; + try { + source = new FileInputStream(randomFilePath); + fileSystem.write("/my_folder/my_file", source); + } finally { + IOUtils.closeQuietly(source); + } } /** * this writing a file that already exists should not be a problem */ @Test - public void testFailAtDoubleWrite() throws Exception { - InputStream source = new FileInputStream(randomFilePath); - fileSystem.write("/my_file", source); - source.close(); + public void testDoubleWrite() throws Exception { + InputStream source = null; + try { + source = new FileInputStream(randomFilePath); + fileSystem.write("/my_file", source); + } finally { + IOUtils.closeQuietly(source); + } - source = new FileInputStream(randomFilePath); - fileSystem.write("/my_file", source); - source.close(); + try { + source = new FileInputStream(randomFilePath); + fileSystem.write("/my_file", source); + } finally { + IOUtils.closeQuietly(source); + } } /** @@ -217,12 +249,23 @@ @Test public void testWriteInFolder() throws Exception { fileSystem.createDirectory("/my_folder"); - fileSystem.write("/my_folder/my_file", - new FileInputStream(randomFilePath)); fileSystem.createDirectory("/my_folder/my_sub_folder"); - fileSystem.write("/my_folder/my_sub_folder/my_file", - new FileInputStream(randomFilePath)); - assertTrue(fileSystem.exists("/my_folder/my_sub_folder/my_file")); + + InputStream source = null; + try { + source = new FileInputStream(randomFilePath); + fileSystem.write("/my_folder/my_file", source); + } finally { + IOUtils.closeQuietly(source); + } + + try { + source = new FileInputStream(randomFilePath); + fileSystem.write("/my_folder/my_sub_folder/my_file", source); + assertTrue(fileSystem.exists("/my_folder/my_sub_folder/my_file")); + } finally { + IOUtils.closeQuietly(source); + } } /** @@ -233,19 +276,29 @@ @Test public void testLinking() throws Exception { fileSystem.createDirectory("/my_folder"); - fileSystem.write("/my_folder/my_file", - new FileInputStream(randomFilePath)); + + InputStream source = null; + try { + source = new FileInputStream(randomFilePath); + fileSystem.write("/my_folder/my_file", source); + } finally { + IOUtils.closeQuietly(source); + } + fileSystem.createSymbolicLink("/my_link", "/my_folder/my_file"); - InputStream source = new FileInputStream(randomFilePath); - InputStream readResult = fileSystem.read("/my_link"); + InputStream readResult = null; + try { + source = new FileInputStream(randomFilePath); + readResult = fileSystem.read("/my_link"); - boolean actualContentEquality = - IOUtils.contentEquals(source, readResult); - source.close(); - readResult.close(); - - assertTrue(actualContentEquality); + boolean actualContentEquality = + IOUtils.contentEquals(source, readResult); + assertTrue(actualContentEquality); + } finally { + IOUtils.closeQuietly(source); + IOUtils.closeQuietly(readResult); + } } /** @@ -264,8 +317,14 @@ @Test public void testRemove() throws Exception { fileSystem.createDirectory("/my_folder"); - fileSystem.write("/my_folder/my_file", - new FileInputStream(randomFilePath)); + + InputStream source = null; + try { + source = new FileInputStream(randomFilePath); + fileSystem.write("/my_folder/my_file", source); + } finally { + IOUtils.closeQuietly(source); + } fileSystem.delete("/my_folder/my_file"); assertTrue(fileSystem.exists("/my_folder")); assertFalse(fileSystem.exists("/my_folder/my_file")); @@ -281,9 +340,15 @@ @Test(expected = IOException.class) public void testFailAtRemove() throws Exception { fileSystem.createDirectory("/my_folder"); - fileSystem.write("/my_folder/my_file", - new FileInputStream(randomFilePath)); + InputStream source = null; + try { + source = new FileInputStream(randomFilePath); + fileSystem.write("/my_folder/my_file", source); + } finally { + IOUtils.closeQuietly(source); + } + // trying to remove a non-empty directory should raise an exception fileSystem.delete("/my_folder"); } @@ -295,8 +360,15 @@ public void testListDirectory() throws Exception { fileSystem.createDirectory("/my_folder"); fileSystem.createDirectory("/my_folder/my_sub_dir"); - fileSystem.write("/my_folder/my_file", - new FileInputStream(randomFilePath)); + + InputStream source = null; + try { + source = new FileInputStream(randomFilePath); + fileSystem.write("/my_folder/my_file", source); + } finally { + IOUtils.closeQuietly(source); + } + fileSystem.createSymbolicLink("/my_folder/my_link", "my_file"); List<String> lsResult = fileSystem.readDirectory("/my_folder"); @@ -362,7 +434,14 @@ fileSystem.createDirectory("/dir/subdir"); fileSystem.createDirectory("/dir/subdir/subsubdir"); fileSystem.createSymbolicLink("/dir/link", "subdir"); - fileSystem.write("/dir/file", new FileInputStream(randomFilePath)); + + InputStream source = null; + try { + source = new FileInputStream(randomFilePath); + fileSystem.write("/dir/file", source); + } finally { + IOUtils.closeQuietly(source); + } fileSystem.createDirectory("/otherdir"); fileSystem.move("/dir/subdir", "/otherdir/subdir"); Modified: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java 2010-06-19 18:55:14 UTC (rev 84) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java 2010-06-21 10:14:34 UTC (rev 85) @@ -6,6 +6,7 @@ import static org.junit.Assert.fail; import java.io.ByteArrayInputStream; +import java.io.FileInputStream; import java.util.ConcurrentModificationException; import org.apache.commons.io.IOUtils; @@ -38,8 +39,8 @@ DisworkFileSystemConfig.newKademliaDisworkConfig(bootstrapPort); DisworkFileSystem fileSystem2 = new DisworkFileSystem(disworkConfig); - assertTrue(fileSystem.exists("/")); - assertTrue(fileSystem2.exists("/")); + fileSystem.write("/file", new FileInputStream(randomFilePath)); + assertTrue(fileSystem2.exists("/file")); } @Test Modified: trunk/pom.xml =================================================================== --- trunk/pom.xml 2010-06-19 18:55:14 UTC (rev 84) +++ trunk/pom.xml 2010-06-21 10:14:34 UTC (rev 85) @@ -81,7 +81,7 @@ <dependency> <groupId>org.planx</groupId> <artifactId>koala-xmlstore</artifactId> - <version>0.4.12</version> + <version>0.4.12_cl20100612</version> </dependency> <dependency> <groupId>commons-digester</groupId>

1 0

r84 - trunk
by tchemit＠users.nuiton.org 19 Jun '10

19 Jun '10

Author: tchemit Date: 2010-06-19 20:55:14 +0200 (Sat, 19 Jun 2010) New Revision: 84 Url: http://nuiton.org/repositories/revision/diswork/84 Log: Utilisation de mavenpom4redmine 2.2 Modified: trunk/pom.xml Modified: trunk/pom.xml =================================================================== --- trunk/pom.xml 2010-06-17 13:37:59 UTC (rev 83) +++ trunk/pom.xml 2010-06-19 18:55:14 UTC (rev 84) @@ -1,6 +1,5 @@ <?xml version="1.0" encoding="UTF-8"?> -<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" - xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> +<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> @@ -11,7 +10,7 @@ <parent> <groupId>org.nuiton</groupId> <artifactId>mavenpom4redmine</artifactId> - <version>2.1.5</version> + <version>2.2</version> </parent> <artifactId>diswork</artifactId>

1 0

r83 - in trunk/diswork-fs/src: main/java/org/nuiton/diswork/fs test/java/org/nuiton/diswork/fs
by bleny＠users.nuiton.org 17 Jun '10

17 Jun '10

Author: bleny Date: 2010-06-17 15:37:59 +0200 (Thu, 17 Jun 2010) New Revision: 83 Url: http://nuiton.org/repositories/revision/diswork/83 Log: bugfix move, overwrite Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystem.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemInMemoryTest.java Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystem.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystem.java 2010-06-17 13:33:48 UTC (rev 82) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystem.java 2010-06-17 13:37:59 UTC (rev 83) @@ -49,6 +49,11 @@ * <li>{@link #exists(String)} and {@link #delete(String)} can be used * on directories, files and symlinks</li> * </ul> + * + * Those methods are not thread-safe, so manipulating the content of + * same directory in two different thread using the same instance may + * lead to errors. + * */ public class DisworkFileSystem implements Closeable { @@ -164,14 +169,6 @@ if (EntryUtil.isDirectory(entryParent)) { String parentId = EntryUtil.getIdFromEntry(entryParent); - // checking that file do not already exists in this directory - String content = storage.getDirectory(parentId); - String findResult = EntryUtil.findEntryInDirectory - (content, fileName); - if (findResult != null) { - throw new IOException - (parent + " already contains an element named " + fileName); - } // file do not exists, write file on the FS String newFileId = EntryUtil.generateId(); @@ -184,11 +181,26 @@ lockAcquired = storage.tryToLock(parentId); if (lockAcquired) { // parent dir is locked, do the update - content = storage.getDirectory(parentId); - String newContent = EntryUtil.addEntryToDirectoryContent( + + // checking that file do not already exists in this directory + String content = storage.getDirectory(parentId); + String oldEntry = EntryUtil.findEntryInDirectory + (content, fileName); + + if (oldEntry != null) { + content = EntryUtil.removeEntryFromEntries(content, fileName); + } + + content = EntryUtil.addEntryToDirectoryContent( content, EntryUtil.TYPE.F, fileName, newFileId); - storage.putDirectory(parentId, newContent); + storage.putDirectory(parentId, content); storage.unLock(parentId); + + // removing old data + if (oldEntry != null) { + storage.removeFile(EntryUtil.getIdFromEntry(oldEntry)); + } + } else { log.info(parent + " is locked and can't be written"); try { @@ -410,13 +422,16 @@ /** * remove a file, directory, or link. Non-empty directories can't be - * removed + * removed. * @param path the complete path to the entity to remove * @throws IOException if path is incorrect or directory not empty */ public void delete(String path) throws IOException, ConcurrentModificationException { checkPathSyntax(path); + if (!exists(path)) { + throw new IOException(path + " doesn't exists"); + } String parent = EntryUtil.getParentFromPath(path); String name = EntryUtil.getNameFromPath(path); log.info("trying to remove " + path); @@ -703,6 +718,10 @@ checkPathSyntax(path); checkPathSyntax(destination); + if (exists(destination)) { + throw new IOException(destination + " already exists"); + } + String pathParent = EntryUtil.getParentFromPath(path); String pathName = EntryUtil.getNameFromPath(path); String destinationParent = EntryUtil.getParentFromPath(destination); @@ -750,6 +769,7 @@ while (numberOfTry <= LOCK_MAX_NUMBER_OF_TRY && !lockAcquired) { lockAcquired = storage.tryToLock(parentId) && storage.tryToLock(destinationParentId); + if (lockAcquired) { // we have locked, do the update parentContent = storage.getDirectory(parentId); Modified: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java 2010-06-17 13:33:48 UTC (rev 82) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java 2010-06-17 13:37:59 UTC (rev 83) @@ -21,6 +21,7 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.junit.experimental.categories.Categories.ExcludeCategory; import org.nuiton.diswork.fs.DisworkFileSystem; import org.nuiton.util.FileUtil; @@ -184,21 +185,17 @@ } /** - * this use case should raise an exception because writing to a file - * that already exists + * this writing a file that already exists should not be a problem */ - @Test(expected = IOException.class) + @Test public void testFailAtDoubleWrite() throws Exception { InputStream source = new FileInputStream(randomFilePath); fileSystem.write("/my_file", source); source.close(); source = new FileInputStream(randomFilePath); - try { - fileSystem.write("/my_file", source); - } finally { - source.close(); - } + fileSystem.write("/my_file", source); + source.close(); } /** @@ -395,6 +392,13 @@ */ } + @Test(expected = IOException.class) + public void failAtMove() throws Exception { + fileSystem.createDirectory("/dir"); + fileSystem.createDirectory("/dir2"); + fileSystem.move("/dir", "/dir2"); + } + @Test public void testCreateDirectories() throws Exception { try { Modified: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemInMemoryTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemInMemoryTest.java 2010-06-17 13:33:48 UTC (rev 82) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemInMemoryTest.java 2010-06-17 13:37:59 UTC (rev 83) @@ -29,6 +29,7 @@ fileSystem.createDirectory("/dir"); fileSystem.write("/dir/file", new FileInputStream(randomFilePath)); + fileSystem.write("/dir/file", new FileInputStream(randomFilePath)); fileSystem.createSymbolicLink("/dir/link", "/dir/file"); fileSystem.delete("/dir/link");

1 0

r82 - trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon
by bleny＠users.nuiton.org 17 Jun '10

17 Jun '10

Author: bleny Date: 2010-06-17 15:33:48 +0200 (Thu, 17 Jun 2010) New Revision: 82 Url: http://nuiton.org/repositories/revision/diswork/82 Log: bugfix concurrence Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java 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-06-17 09:00:26 UTC (rev 81) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java 2010-06-17 13:33:48 UTC (rev 82) @@ -72,9 +72,9 @@ RUNNING_MOVE.put(DisworkDaemon.FAILED_1_RUNNING, DisworkDaemon.FAILED_1_RUNNING); RUNNING_MOVE.put(DisworkDaemon.FAILED_2_RUNNING, DisworkDaemon.FAILED_2_RUNNING); - FAILED_MOVE.put(DisworkDaemon.TODO, DisworkDaemon.FAILED_1); - FAILED_MOVE.put(DisworkDaemon.FAILED_1, DisworkDaemon.FAILED_2); - FAILED_MOVE.put(DisworkDaemon.FAILED_2, DisworkDaemon.FAILED_3); + FAILED_MOVE.put(DisworkDaemon.TODO_RUNNING, DisworkDaemon.FAILED_1); + FAILED_MOVE.put(DisworkDaemon.FAILED_1_RUNNING, DisworkDaemon.FAILED_2); + FAILED_MOVE.put(DisworkDaemon.FAILED_2_RUNNING, DisworkDaemon.FAILED_3); } private static final Log log = LogFactory.getLog(WorkersManager.class); @@ -89,23 +89,34 @@ protected class Worker extends Thread { - public boolean shouldStop = false; + protected boolean shouldStop = false; - public WorkersManager manager; + protected WorkersManager manager; + /** + * this method add a line to a job-specific log + * @param jobPath the path to the job concerned + * @param message the line to add the the log + * @throws IOException if an error occurred while writing the log + */ protected void log(String jobPath, String message) throws IOException { String logPath = jobPath + "/" + DisworkDaemon.LOG_PATH; InputStream oldLogAsStream = fileSystem.read(logPath); String oldLog = IOUtils.toString(oldLogAsStream); - String logEntry = message + "\n"; - String newLog = oldLog + logEntry; - - fileSystem.delete(logPath); fileSystem.write(logPath, IOUtils.toInputStream(newLog)); } + /** + * Download all the files needed for a job in a temp directory, run + * the job, wait for it to end, write all the results. Mark the job + * as running at the beginning and move it to DONE or FAILED at + * the end, depending of the results + * @param jobPath + * @return + * @throws IOException + */ protected boolean runJob(String jobPath) throws IOException { log.info("running job at " + jobPath); @@ -168,13 +179,13 @@ IOUtils.copy(source, new FileOutputStream(localCopy)); } - // executing the job + // prepare and run it the job String commandLine = jobDescription.getCommandLine(); - log.info("calling " + commandLine); String[] commandLineElements = commandLine.split(" "); ProcessBuilder builder = new ProcessBuilder(commandLineElements); builder.directory(jobDir); builder.redirectErrorStream(true); + log.info("calling " + commandLine); Process job = builder.start(); // plugging a file on the standard input @@ -185,6 +196,7 @@ IOUtils.copy(input, job.getOutputStream()); } + // run the process and wait for it to return int exitValue = -1; try { exitValue = job.waitFor(); @@ -237,7 +249,7 @@ return success; } - public String getFistJobName(String path) throws IOException { + protected String getFistJobName(String path) throws IOException { List<String> jobsNames = fileSystem.readDirectory(path); if (jobsNames.size() == 0) { return null; @@ -250,71 +262,93 @@ protected void findAJobAndRunIt() { // try to find a new job try { + + // Once a job is found, those two var will be set String jobLinkDir = null; String jobLinkName = null; - String[] runningJobsDirs = { DisworkDaemon.FAILED_2_RUNNING, - DisworkDaemon.FAILED_1_RUNNING, - DisworkDaemon.TODO_RUNNING - }; - - // FIXME 20100617 bleny stop loops when found - for (String path : runningJobsDirs) { - String oldName = getFistJobName(path); - if (oldName != null) { - Long linkAge = System.currentTimeMillis() - - Long.parseLong(oldName); - if (linkAge > MAX_JOB_RUNNING_TIME) { - log.info("taking old job (age = " + linkAge + ")"); - jobLinkDir = path; - jobLinkName = oldName; + // use a synchronized block because multiple workers + // may try to take a same job + synchronized (fileSystem) { + + // fist, try to find a job declared has running since + // too long to re-run it + String[] runningJobsDirs = { DisworkDaemon.FAILED_2_RUNNING, + DisworkDaemon.FAILED_1_RUNNING, + DisworkDaemon.TODO_RUNNING + }; + // browsing all "running" dirs + for (String path : runningJobsDirs) { + String oldName = getFistJobName(path); + if (oldName != null) { + Long linkAge = System.currentTimeMillis() + - Long.parseLong(oldName); + // check is oldest job is too old and should be + // considered has to-be-rerun + if (linkAge > MAX_JOB_RUNNING_TIME) { + log.info("taking old job (age = " + linkAge + ")"); + jobLinkDir = path; + jobLinkName = oldName; + // FIXME 20100617 bleny break s*cks + break; + } } } - } - - if (jobLinkDir == null) { - String[] jobsDirs = { DisworkDaemon.FAILED_2, - DisworkDaemon.FAILED_1, - DisworkDaemon.TODO - }; - for (String path : jobsDirs) { - String oldName = getFistJobName(path); - if (oldName != null) { // take it - jobLinkDir = path; - jobLinkName = oldName; + + // if no job was found, search now in not running jobs + if (jobLinkDir == null) { + String[] jobsDirs = { DisworkDaemon.FAILED_2, + DisworkDaemon.FAILED_1, + DisworkDaemon.TODO + }; + for (String path : jobsDirs) { + String oldName = getFistJobName(path); + if (oldName != null) { // take it + jobLinkDir = path; + jobLinkName = oldName; + // FIXME 20100617 bleny break s*cks + break; + } } } + + if (jobLinkDir != null) { + // move the link before running the job + String oldPath = jobLinkDir + "/" + jobLinkName; + log.info("job found at " + oldPath); + + jobLinkDir = RUNNING_MOVE.get(jobLinkDir); + jobLinkName = DisworkDaemon.newJobLinkName(); + String newPath = jobLinkDir + "/" + jobLinkName; + + log.info("moving " + oldPath + " to " + newPath); + fileSystem.move(oldPath, newPath); + + } } + // now, if no job was found, do nothing if (jobLinkDir == null) { log.info("nothing to do"); Thread.sleep(JOB_WAIT); + // if a job was found, take it } else { - // move the link before running the job - String oldPath = jobLinkDir + "/" + jobLinkName; - log.info("job found at " + oldPath); + + String jobPath = jobLinkDir + "/" + jobLinkName; - String newPath = RUNNING_MOVE.get(jobLinkDir) + "/" + - DisworkDaemon.newJobLinkName(); + boolean jobSuccess = runJob(jobPath); - log.info("moving " + oldPath + " to " + newPath); - fileSystem.move(oldPath, newPath); - - // run the job - boolean jobSuccess = runJob(newPath); - // move the link after the job - oldPath = newPath; String newDir = null; if (jobSuccess) { newDir = DisworkDaemon.DONE; } else { newDir = FAILED_MOVE.get(jobLinkDir); } - newPath = newDir + "/" + DisworkDaemon.newJobLinkName(); + String newPath = newDir + "/" + DisworkDaemon.newJobLinkName(); - log.info("moving " + oldPath + " to " + newPath); - fileSystem.move(oldPath, newPath); + log.info("moving " + jobPath + " to " + newPath); + fileSystem.move(jobPath, newPath); // mark the job has finished if done or failed too many // times @@ -332,8 +366,7 @@ } catch (InterruptedException e) { log.info("exception catch", e); // TODO 20100611 bleny manage exception - } - + } } @Override @@ -367,7 +400,8 @@ } else if ("scheduled".equals(initialStrategy)) { activeScheduledActivityStrategy(); } else { - log.error("wrong config directive " + initialStrategy); + log.error("wrong config directive for initial strategy" + + initialStrategy); activeNoActivityStrategy(); }

1 0

r81 - in trunk/diswork-daemon/src: main/java/org/nuiton/diswork/daemon test/java/org/nuiton/diswork/daemon
by bleny＠users.nuiton.org 17 Jun '10

17 Jun '10

Author: bleny Date: 2010-06-17 11:00:26 +0200 (Thu, 17 Jun 2010) New Revision: 81 Url: http://nuiton.org/repositories/revision/diswork/81 Log: bugfix dans la concurrence avec test Added: trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonConcurrencyTest.java Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/ActivityStrategy.java trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkConfig.java trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonTest.java Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/ActivityStrategy.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/ActivityStrategy.java 2010-06-17 08:08:19 UTC (rev 80) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/ActivityStrategy.java 2010-06-17 09:00:26 UTC (rev 81) @@ -1,3 +1,27 @@ +/* + * #%L + * Diswork daemon + * + * $Id$ + * $HeadURL$ + * %% + * Copyright (C) 2010 CodeLutin + * %% + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Lesser Public License for more details. + * + * You should have received a copy of the GNU General Lesser Public + * License along with this program. If not, see + * <http://www.gnu.org/licenses/lgpl-3.0.html>. + * #L% + */ package org.nuiton.diswork.daemon; import java.lang.management.ManagementFactory; Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkConfig.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkConfig.java 2010-06-17 08:08:19 UTC (rev 80) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkConfig.java 2010-06-17 09:00:26 UTC (rev 81) @@ -142,6 +142,10 @@ return getOptionAsInt("diswork.workers_number"); } + public void setNumberOfWorkers(Integer number) { + setOption("diswork.workers_number", number.toString()); + } + public String getActivityStrategy() { return getOption("diswork.activity_strategy"); } 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-06-17 08:08:19 UTC (rev 80) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java 2010-06-17 09:00:26 UTC (rev 81) @@ -68,6 +68,9 @@ RUNNING_MOVE.put(DisworkDaemon.TODO, DisworkDaemon.TODO_RUNNING); RUNNING_MOVE.put(DisworkDaemon.FAILED_1, DisworkDaemon.FAILED_1_RUNNING); RUNNING_MOVE.put(DisworkDaemon.FAILED_2, DisworkDaemon.FAILED_2_RUNNING); + RUNNING_MOVE.put(DisworkDaemon.TODO_RUNNING, DisworkDaemon.TODO_RUNNING); + RUNNING_MOVE.put(DisworkDaemon.FAILED_1_RUNNING, DisworkDaemon.FAILED_1_RUNNING); + RUNNING_MOVE.put(DisworkDaemon.FAILED_2_RUNNING, DisworkDaemon.FAILED_2_RUNNING); FAILED_MOVE.put(DisworkDaemon.TODO, DisworkDaemon.FAILED_1); FAILED_MOVE.put(DisworkDaemon.FAILED_1, DisworkDaemon.FAILED_2); @@ -254,12 +257,14 @@ DisworkDaemon.FAILED_1_RUNNING, DisworkDaemon.TODO_RUNNING }; + + // FIXME 20100617 bleny stop loops when found for (String path : runningJobsDirs) { String oldName = getFistJobName(path); if (oldName != null) { Long linkAge = System.currentTimeMillis() - Long.parseLong(oldName); - if (linkAge <= MAX_JOB_RUNNING_TIME) { + if (linkAge > MAX_JOB_RUNNING_TIME) { log.info("taking old job (age = " + linkAge + ")"); jobLinkDir = path; jobLinkName = oldName; @@ -351,14 +356,6 @@ public WorkersManager(DisworkFileSystem fileSystem, DisworkConfig config) { this.fileSystem = fileSystem; this.config = config; - - log.info("will start " + config.getNumberOfWorkers() + " workers"); - for (int i = 1 ; i <= config.getNumberOfWorkers() ; i++) { - Worker worker = new Worker(); - worker.manager = this; - worker.start(); - workers.add(worker); - } String initialStrategy = config.getActivityStrategy(); if ( "none".equals(initialStrategy)) { @@ -373,6 +370,14 @@ log.error("wrong config directive " + initialStrategy); activeNoActivityStrategy(); } + + log.info("will start " + config.getNumberOfWorkers() + " workers"); + for (int i = 1 ; i <= config.getNumberOfWorkers() ; i++) { + Worker worker = new Worker(); + worker.manager = this; + worker.start(); + workers.add(worker); + } } Added: trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonConcurrencyTest.java =================================================================== --- trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonConcurrencyTest.java (rev 0) +++ trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonConcurrencyTest.java 2010-06-17 09:00:26 UTC (rev 81) @@ -0,0 +1,21 @@ +package org.nuiton.diswork.daemon; + +import java.io.InputStream; + +import org.junit.Before; + +public class DisworkDaemonConcurrencyTest extends DisworkDaemonTest { + + @Before + public void setUp() throws Exception { + DisworkConfig config = DisworkConfig.newConfig(); + port += 1; + config.setUsedPort(port); + config.setNumberOfWorkers(32); + daemon = new DisworkDaemon(config); + InputStream application = ClassLoader.getSystemResourceAsStream("fake-app-1.0.zip"); + daemon.submitApplication("fake-app", "1.0", application); + } + + +} Modified: trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonTest.java =================================================================== --- trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonTest.java 2010-06-17 08:08:19 UTC (rev 80) +++ trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonTest.java 2010-06-17 09:00:26 UTC (rev 81) @@ -143,7 +143,7 @@ Map<String, Integer> stats = daemon.getGlobalStats(); // deamon should read 3 stats : 1 OS, 1 architecture and the number // of processors - assertEquals(3, stats); + assertEquals(3, stats.size()); } }

1 0

r80 - in trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs: . storage
by bleny＠users.nuiton.org 17 Jun '10

17 Jun '10

Author: bleny Date: 2010-06-17 10:08:19 +0200 (Thu, 17 Jun 2010) New Revision: 80 Url: http://nuiton.org/repositories/revision/diswork/80 Log: - bug dans les tests kad Modified: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/KademliaDisworkMapTest.java Modified: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java 2010-06-17 07:53:30 UTC (rev 79) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java 2010-06-17 08:08:19 UTC (rev 80) @@ -27,7 +27,7 @@ // finally, initiate the fileSystem DisworkFileSystemConfig disworkConfig1 = DisworkFileSystemConfig.newKademliaDisworkConfig(); - bootstrapPort = disworkConfig1.getUsedPort() + 1; + bootstrapPort = disworkConfig1.getUsedPort(); fileSystem = new DisworkFileSystem(disworkConfig1); } Modified: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/KademliaDisworkMapTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/KademliaDisworkMapTest.java 2010-06-17 07:53:30 UTC (rev 79) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/KademliaDisworkMapTest.java 2010-06-17 08:08:19 UTC (rev 80) @@ -22,7 +22,7 @@ map1 = new KademliaDisworkMap(config1); DisworkFileSystemConfig config2 = DisworkFileSystemConfig - .newKademliaDisworkConfig(config1.getUsedPort()+1); + .newKademliaDisworkConfig(config1.getUsedPort()); map2 = new KademliaDisworkMap(config2); }

1 0

r79 - in trunk/diswork-daemon/src/main: . resources
by bleny＠users.nuiton.org 17 Jun '10

17 Jun '10

Author: bleny Date: 2010-06-17 09:53:30 +0200 (Thu, 17 Jun 2010) New Revision: 79 Url: http://nuiton.org/repositories/revision/diswork/79 Log: config log4j pour les executables Added: trunk/diswork-daemon/src/main/resources/ trunk/diswork-daemon/src/main/resources/log4j.properties Added: trunk/diswork-daemon/src/main/resources/log4j.properties =================================================================== --- trunk/diswork-daemon/src/main/resources/log4j.properties (rev 0) +++ trunk/diswork-daemon/src/main/resources/log4j.properties 2010-06-17 07:53:30 UTC (rev 79) @@ -0,0 +1,9 @@ +# Global logging configuration +log4j.rootLogger=WARN, stdout +# Console output... +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%d %5p [%t] (%F:%L) %M - %m%n +# package level +log4j.logger.org.nuiton.diswork.fs.storage.KademliaDisworkMap=INFO +log4j.logger.org.nuiton.diswork.daemon=INFO \ No newline at end of file

1 0

r78 - in trunk: diswork-daemon diswork-daemon/src/main/java/org/nuiton/diswork/daemon diswork-fs/src/main/java/org/nuiton/diswork/fs diswork-fs/src/main/java/org/nuiton/diswork/fs/storage
by bleny＠users.nuiton.org 17 Jun '10

17 Jun '10

Author: bleny Date: 2010-06-17 09:51:19 +0200 (Thu, 17 Jun 2010) New Revision: 78 Url: http://nuiton.org/repositories/revision/diswork/78 Log: DaemonRunner et SimpleClient Added: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkSimpleClient.java Modified: trunk/diswork-daemon/ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemonRunner.java trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/JobDescription.java trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/KademliaDisworkMap.java Property changes on: trunk/diswork-daemon ___________________________________________________________________ Modified: svn:ignore - target + target .classpath .settings Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemonRunner.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemonRunner.java 2010-06-16 16:02:13 UTC (rev 77) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemonRunner.java 2010-06-17 07:51:19 UTC (rev 78) @@ -24,11 +24,9 @@ */ package org.nuiton.diswork.daemon; -import java.lang.management.ManagementFactory; -import java.lang.management.OperatingSystemMXBean; -import java.lang.management.RuntimeMXBean; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.nuiton.diswork.fs.DisworkFileSystemConfig; /** * @@ -41,20 +39,18 @@ /** * @param args */ - public static void main(String[] args) { + public static void main(String[] args) throws Exception { - OperatingSystemMXBean os = ManagementFactory.getOperatingSystemMXBean(); - System.out.println(os.getArch()); - System.out.println(os.getAvailableProcessors()); - System.out.println(os.getName()); - System.out.println(os.getVersion()); - - RuntimeMXBean run = ManagementFactory.getRuntimeMXBean(); - System.out.println(run.getUptime()); - - // consider args - - // DisworkDaemon node = new DisworkDaemon(config); - + DisworkConfig config = new DisworkConfig(); + if (args.length == 0) { + config.setFileSystemConfig( + DisworkFileSystemConfig.newKademliaDisworkConfig()); + } else if (args.length == 2){ + config.setFileSystemConfig( + DisworkFileSystemConfig.newKademliaDisworkConfig(args[0], + Integer.parseInt(args[1]))); + } + config.setActivityStrategy("unlimited"); + DisworkDaemon daemon = new DisworkDaemon(config); } } Added: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkSimpleClient.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkSimpleClient.java (rev 0) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkSimpleClient.java 2010-06-17 07:51:19 UTC (rev 78) @@ -0,0 +1,90 @@ +/* + * #%L + * Diswork daemon + * + * $Id$ + * $HeadURL$ + * %% + * Copyright (C) 2010 CodeLutin + * %% + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Lesser Public License for more details. + * + * You should have received a copy of the GNU General Lesser Public + * License along with this program. If not, see + * <http://www.gnu.org/licenses/lgpl-3.0.html>. + * #L% + */ +package org.nuiton.diswork.daemon; + +import java.io.BufferedReader; +import java.io.InputStreamReader; +import java.util.Map; + +import org.apache.commons.io.IOUtils; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.nuiton.diswork.fs.DisworkFileSystemConfig; + +/** + * + * @author bleny + */ +public class DisworkSimpleClient { + + private static final Log log = LogFactory.getLog(DisworkSimpleClient.class); + + /** + * @param args + * @throws Exception + */ + public static void main(String[] args) throws Exception { + + DisworkConfig config = new DisworkConfig(); + config.setFileSystemConfig( + DisworkFileSystemConfig.newKademliaDisworkConfig(args[0], + Integer.parseInt(args[1]))); + config.setActivityStrategy("none"); + config.setUsedPort(30000); + DisworkDaemon daemon = new DisworkDaemon(config); + + // prompt the user to enter their name + + // open up standard input + BufferedReader br = new BufferedReader(new InputStreamReader(System.in)); + + String userEntry = ""; + do { + System.out.print("diswork: "); + userEntry = br.readLine(); + + if ("stats".equals(userEntry)) { + Map<String, Integer> stats = daemon.getGlobalStats(); + for (String key : stats.keySet()) { + System.out.println(key + " => " + stats.get(key)); + } + } else if (!"quit".equals(userEntry)) { + JobDescription job = new JobDescription(); + job.setStandardOutput("output.txt"); + job.addOutput("output.txt"); + job.setCommandLine(userEntry); + daemon.submitJob(job); + + while(! daemon.isFinished(job)) { + Thread.sleep(5 * 1000); + } + + System.out.println(IOUtils.toString(daemon.getResults(job).get("output.txt"))); + } + + } while (!userEntry.equals("quit")); + + } +} Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/JobDescription.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/JobDescription.java 2010-06-16 16:02:13 UTC (rev 77) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/JobDescription.java 2010-06-17 07:51:19 UTC (rev 78) @@ -294,7 +294,7 @@ jsdlNamespace); Element applicationName = application.getChild("ApplicationName", jsdlNamespace); - if (application != null) { + if (applicationName != null) { Element applicationVersion = application.getChild ("ApplicationVersion", jsdlNamespace); result.setApplication(applicationName.getText(), @@ -315,7 +315,7 @@ } Element output = POSIXApplication.getChild("Output", jsdlPosixNamespace); - if (input != null) { + if (output != null) { result.setStandardOutput(output.getText()); } 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-06-16 16:02:13 UTC (rev 77) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java 2010-06-17 07:51:19 UTC (rev 78) @@ -194,10 +194,12 @@ // dump the standard output in a file String standardOutputFileName = jobDescription.getStandardOutput(); + log.info("standardOutputFileName is " + standardOutputFileName); if (standardOutputFileName != null) { - OutputStream output = new FileOutputStream( - new File(jobDir, standardOutputFileName)); - IOUtils.copy(job.getInputStream(), output); + File outputFile = new File(jobDir, standardOutputFileName); + OutputStream output = new FileOutputStream(outputFile); + log.info("writing standard output in " + outputFile); + IOUtils.copy(job.getInputStream(), output); } log.info("job returned " + exitValue); @@ -205,6 +207,7 @@ // output file staging for (String fileName : jobDescription.getStagingOutput()) { File localCopy = new File(jobDir, fileName); + // FIXME 20100616 bleny may not exists if job has fail InputStream localCopyStream = new FileInputStream(localCopy); String filePath = jobPath + "/" + fileName; Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java 2010-06-16 16:02:13 UTC (rev 77) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java 2010-06-17 07:51:19 UTC (rev 78) @@ -181,14 +181,19 @@ public static DisworkFileSystemConfig newKademliaDisworkConfig (Integer bootstrapPort) { + return newKademliaDisworkConfig(getIp(), bootstrapPort); + } + + public static DisworkFileSystemConfig + newKademliaDisworkConfig (String bootstrapIp, + Integer bootstrapPort) { DisworkFileSystemConfig result = new DisworkFileSystemConfig(); String port = getPort().toString(); - String ip = getIp(); result.setOption("diswork.fs.map_type", "kademlia"); result.setOption("diswork.fs.use_port", port); if (bootstrapPort != null) { result.setOption("diswork.fs.bootstrap.port", bootstrapPort.toString()); - result.setOption("diswork.fs.bootstrap.ip", ip); + result.setOption("diswork.fs.bootstrap.ip", bootstrapIp); } return result; } Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/KademliaDisworkMap.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/KademliaDisworkMap.java 2010-06-16 16:02:13 UTC (rev 77) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/KademliaDisworkMap.java 2010-06-17 07:51:19 UTC (rev 78) @@ -50,6 +50,7 @@ public KademliaDisworkMap(DisworkFileSystemConfig config) throws IOException { + log.info("booting on port " + config.getUsedPort()); kad = new Kademlia(Identifier.randomIdentifier(), config.getUsedPort()); if (config.getBootstrapIp() != null) {

1 0