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

r77 - trunk/diswork-daemon
by bleny＠users.nuiton.org June 16, 2010

June 16, 2010

Author: bleny Date: 2010-06-16 18:02:13 +0200 (Wed, 16 Jun 2010) New Revision: 77 Url: http://nuiton.org/repositories/revision/diswork/77 Log: menage eclipse (boulette) Removed: trunk/diswork-daemon/.classpath trunk/diswork-daemon/.settings/ Deleted: trunk/diswork-daemon/.classpath =================================================================== --- trunk/diswork-daemon/.classpath 2010-06-16 15:55:53 UTC (rev 76) +++ trunk/diswork-daemon/.classpath 2010-06-16 16:02:13 UTC (rev 77) @@ -1,8 +0,0 @@ -<?xml version="1.0" encoding="UTF-8"?> -<classpath> - <classpathentry kind="src" output="target/classes" path="src/main/java"/> - <classpathentry kind="src" output="target/test-classes" path="src/test/java"/> - <classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.6"/> - <classpathentry kind="con" path="org.maven.ide.eclipse.MAVEN2_CLASSPATH_CONTAINER"/> - <classpathentry kind="output" path="target/classes"/> -</classpath>

1 0

r76 - in trunk/diswork-daemon: . .settings src/main/java/org/nuiton/diswork/daemon src/test/java/org/nuiton/diswork/daemon
by bleny＠users.nuiton.org June 16, 2010

June 16, 2010

Author: bleny Date: 2010-06-16 17:55:53 +0200 (Wed, 16 Jun 2010) New Revision: 76 Url: http://nuiton.org/repositories/revision/diswork/76 Log: statistiques globales, parsing JSDL, activity strategy, config, documentation Added: trunk/diswork-daemon/.classpath trunk/diswork-daemon/.project trunk/diswork-daemon/.settings/ trunk/diswork-daemon/.settings/org.eclipse.jdt.core.prefs trunk/diswork-daemon/.settings/org.maven.ide.eclipse.prefs trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/ActivityStrategy.java trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/JobDescriptionTest.java Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkConfig.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/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 Added: trunk/diswork-daemon/.classpath =================================================================== --- trunk/diswork-daemon/.classpath (rev 0) +++ trunk/diswork-daemon/.classpath 2010-06-16 15:55:53 UTC (rev 76) @@ -0,0 +1,8 @@ +<?xml version="1.0" encoding="UTF-8"?> +<classpath> + <classpathentry kind="src" output="target/classes" path="src/main/java"/> + <classpathentry kind="src" output="target/test-classes" path="src/test/java"/> + <classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.6"/> + <classpathentry kind="con" path="org.maven.ide.eclipse.MAVEN2_CLASSPATH_CONTAINER"/> + <classpathentry kind="output" path="target/classes"/> +</classpath> Added: trunk/diswork-daemon/.project =================================================================== --- trunk/diswork-daemon/.project (rev 0) +++ trunk/diswork-daemon/.project 2010-06-16 15:55:53 UTC (rev 76) @@ -0,0 +1,23 @@ +<?xml version="1.0" encoding="UTF-8"?> +<projectDescription> + <name>diswork-daemon</name> + <comment></comment> + <projects> + </projects> + <buildSpec> + <buildCommand> + <name>org.eclipse.jdt.core.javabuilder</name> + <arguments> + </arguments> + </buildCommand> + <buildCommand> + <name>org.maven.ide.eclipse.maven2Builder</name> + <arguments> + </arguments> + </buildCommand> + </buildSpec> + <natures> + <nature>org.eclipse.jdt.core.javanature</nature> + <nature>org.maven.ide.eclipse.maven2Nature</nature> + </natures> +</projectDescription> Added: trunk/diswork-daemon/.settings/org.eclipse.jdt.core.prefs =================================================================== --- trunk/diswork-daemon/.settings/org.eclipse.jdt.core.prefs (rev 0) +++ trunk/diswork-daemon/.settings/org.eclipse.jdt.core.prefs 2010-06-16 15:55:53 UTC (rev 76) @@ -0,0 +1,6 @@ +#Tue Jun 08 14:19:53 CEST 2010 +eclipse.preferences.version=1 +org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.6 +org.eclipse.jdt.core.compiler.compliance=1.6 +org.eclipse.jdt.core.compiler.problem.forbiddenReference=warning +org.eclipse.jdt.core.compiler.source=1.6 Added: trunk/diswork-daemon/.settings/org.maven.ide.eclipse.prefs =================================================================== --- trunk/diswork-daemon/.settings/org.maven.ide.eclipse.prefs (rev 0) +++ trunk/diswork-daemon/.settings/org.maven.ide.eclipse.prefs 2010-06-16 15:55:53 UTC (rev 76) @@ -0,0 +1,9 @@ +#Mon Jun 07 18:20:47 CEST 2010 +activeProfiles= +eclipse.preferences.version=1 +fullBuildGoals=process-test-resources +includeModules=false +resolveWorkspaceProjects=true +resourceFilterGoals=process-resources resources\:testResources +skipCompilerPlugin=true +version=1 Added: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/ActivityStrategy.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/ActivityStrategy.java (rev 0) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/ActivityStrategy.java 2010-06-16 15:55:53 UTC (rev 76) @@ -0,0 +1,89 @@ +package org.nuiton.diswork.daemon; + +import java.lang.management.ManagementFactory; +import java.lang.management.OperatingSystemMXBean; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +public interface ActivityStrategy { + + /** use this strategy to never run a job */ + public static class NoActivity implements ActivityStrategy { + @Override + public boolean canWork() { + return false; + } + } + + /** use this strategy to always run a job */ + public static class UnlimitedActivity implements ActivityStrategy { + @Override + public boolean canWork() { + return true; + } + } + + /** use this strategy to run a job only if load average is low */ + public static class LimitedActivity implements ActivityStrategy { + + private static final Log log = LogFactory.getLog(LimitedActivity.class); + + protected static class LoadAverageMonitoring extends Thread { + + OperatingSystemMXBean os = ManagementFactory.getOperatingSystemMXBean(); + + Double loadAverageNow = 1.0; + Double loadAverage5MinutesBefore = 1.0; + Double loadAverage10MinutesBefore = 1.0; + + @Override + public void run() { + while (true) { + try { + Thread.sleep(5 * 60 * 1000); // 5 min + } catch (InterruptedException e) { + // TODO 20100615 bleny Auto-generated catch block + log.info("exception catch", e); + e.printStackTrace(); + } + loadAverage10MinutesBefore = loadAverage5MinutesBefore; + loadAverage5MinutesBefore = loadAverageNow; + loadAverageNow = os.getSystemLoadAverage(); + log.info("load averages : " + loadAverageNow + " " + + loadAverage5MinutesBefore + " " + + loadAverage10MinutesBefore); + } + } + } + + LoadAverageMonitoring monitoring; + + public LimitedActivity() { + monitoring = new LoadAverageMonitoring(); + monitoring.start(); + } + + @Override + public boolean canWork() { + boolean canWork = monitoring.loadAverageNow < 1.0 + && monitoring.loadAverage5MinutesBefore < 1.0 + && monitoring.loadAverage10MinutesBefore < 1.0; + return canWork; + } + } + + /** use this strategy to run a job only at fixed times of the week */ + public static class ScheduledActivity implements ActivityStrategy { + + @Override + public boolean canWork() { + // TODO 20100615 bleny Auto-generated method stub + throw new UnsupportedOperationException("not yet implemented"); + } + } + + + /** return true if a job can be run */ + boolean canWork(); +} \ No newline at end of file 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-14 08:53:35 UTC (rev 75) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkConfig.java 2010-06-16 15:55:53 UTC (rev 76) @@ -24,21 +24,68 @@ */ package org.nuiton.diswork.daemon; +import java.io.File; +import java.lang.management.ManagementFactory; + import org.nuiton.diswork.fs.DisworkFileSystemConfig; import org.nuiton.util.ApplicationConfig; +/** + * + * <dl> + * <dt>diswork.workers_number</dt> + * <dd>The number of jobs the daemon can run at the same time. Should be set + * to a value that consider the number of processors, core and RAM + * available. By default, it's the number processors available to the + * JVM. Set it to 0 will disable job-processing.</dd> + * <dt>diswork.temp_directory</dt> + * <dd>Diswork need a temporary directory to store temporary data for each + * jobs. By default, the temp dir of the OS is used (ie "/tmp/diswork" + * under Linux).</dd> + * <dt>diswork.activity_strategy</dt> + * <dd>This is the way the daemon is started, different values are + * available: + * <dl> + * <dt>none</dt> + * <dd>never try to do a job</dd> + * <dt>unlimited</dt> + * <dd>always try to job, whatever the cost</dd> + * <dt>limited</dt> + * <dd>run a job only if hardware resources are available (based + * on the system load average)</dd> + * <dt>scheduled</dt> + * <dd>run a job only at fixed time of the week (for example, + * nights, week-end, etc.). It needs to define a pattern.</dd> + * </dl> + * </dd> + * </dl> + * + * @author bleny + */ public class DisworkConfig extends ApplicationConfig { protected DisworkFileSystemConfig fileSystemConfig; public DisworkConfig() { setConfigFileName("diswork.config"); + + Integer availableProcessors = ManagementFactory + . getOperatingSystemMXBean() + . getAvailableProcessors(); + setDefaultOption("diswork.workers_number", availableProcessors.toString()); + + setDefaultOption("diswork.temp_directory", + System.getProperty("java.io.tmpdir") + + File.separator + "diswork"); + + setDefaultOption("diswork.activity_strategy", "unlimited"); + + // if no total_uptime saved, consider daemon has never run + setDefaultOption("diswork.total_uptime", "0"); } public String getTempDirectory() { - return System.getProperty("java.io.tmpdir", - System.getProperty("user.dir", - ".")) + "/diswork"; + return getOption("diswork.temp_directory"); } public static DisworkConfig newConfig() { @@ -90,5 +137,35 @@ public void setFileSystemConfig(DisworkFileSystemConfig fileSystemConfig) { this.fileSystemConfig = fileSystemConfig; } + + public Integer getNumberOfWorkers() { + return getOptionAsInt("diswork.workers_number"); + } + + public String getActivityStrategy() { + return getOption("diswork.activity_strategy"); + } + + public void setActivityStrategy(String activityStrategyName) { + setOption("diswork.activity_strategy", activityStrategyName); + } + + public Long getTotalUptime() { + String upTime = getOption("diswork.total_uptime"); + return Long.parseLong(upTime); + } + + public void setTotalUptime(Long upTime) { + setOption("diswork.total_uptime", upTime.toString()); + } + + public void setFirstRunTime(Long time) { + setOption("diswork.first_run_time", time.toString()); + } + public Long getFirstRunTime() { + String firstRunTime = getOption("diswork.first_run_time"); + return Long.parseLong(firstRunTime); + } + } 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-14 08:53:35 UTC (rev 75) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemon.java 2010-06-16 15:55:53 UTC (rev 76) @@ -28,6 +28,8 @@ import java.io.FileNotFoundException; import java.io.IOException; import java.io.InputStream; +import java.lang.management.ManagementFactory; +import java.lang.management.OperatingSystemMXBean; import java.net.UnknownHostException; import java.util.ConcurrentModificationException; import java.util.HashMap; @@ -41,6 +43,7 @@ import org.apache.commons.logging.LogFactory; import org.nuiton.diswork.fs.DisworkFileSystem; import org.nuiton.diswork.fs.DisworkFileSystemConfig; +import org.nuiton.util.FileUtil; /** * @@ -55,16 +58,6 @@ private static final Log log = LogFactory.getLog(DisworkDaemon.class); - protected DisworkFileSystem fileSystem; - - protected DisworkConfig config; - - /** owner id for this node */ - protected String ownerId; - - /** path to owned directory on fileSystem */ - protected String homeDir; - /** contains applications */ protected static final String BIN = "/bin"; @@ -94,9 +87,6 @@ /** a place where are all user-directories */ protected static final String HOME = "/home"; - - /** worker-manager will make the daemon accomplish jobs */ - protected WorkersManager workers; /** in a job directory, the place where the JSDL must be placed */ protected static final String JSDL_PATH = ".diswork/job.jsdl"; @@ -104,6 +94,28 @@ /** in a job directory, the place where the log must be placed */ protected static final String LOG_PATH = ".diswork/job.log"; + /** in a home directory, the place where the hardware info must be placed */ + protected static final String HARDINFO_PATH = "hardinfo"; + + + /** the distributed file system where jobs, data and results are stored */ + protected DisworkFileSystem fileSystem; + + /** provide the configuration data about the daemon */ + protected DisworkConfig config; + + /** time when the deamon started this time, used for total uptime stat */ + protected Long sessionStartTime; + + /** owner id for this node */ + protected String ownerId; + + /** path to owned directory on fileSystem */ + protected String homeDir; + + /** worker-manager will make the daemon accomplish jobs */ + protected WorkersManager workers; + public DisworkDaemon(DisworkConfig config) throws DisworkException { this.config = config; @@ -122,39 +134,79 @@ throw new DisworkException("booting diswork file system failed", e); } - - ownerId = config.getOwnerId(); // get job owner id from config - - if (ownerId == null) { + ownerId = config.getOwnerId(); // get job owner id from config + + if (ownerId == null) { // first time running the daemon log.info("can't find owner id, generating a new one"); - // generate a new one by cheking if home dir exists - ownerId = System.getProperty("user.name", "anonymous"); + // check home dir do not exists + try { + Random random = new Random(); + // generate a new one by cheking if home dir exists + String simpleName = System.getProperty("user.name", "anonymous"); + ownerId = simpleName; + boolean alreadyExists = fileSystem.exists(HOME + "/" + ownerId); + + // if simpleName is already taken, try simpleName + random + while (alreadyExists) { + alreadyExists = fileSystem.exists(HOME + "/" + ownerId); + ownerId = simpleName + random.nextInt(); + } + homeDir = HOME + "/" + ownerId; + if (!fileSystem.exists(homeDir)) { + fileSystem.createDirectory(homeDir); + } + } catch (ConcurrentModificationException e) { + log.info("can't create home dir", e); + throw new DisworkException("can't create home dir", e); + } catch (IOException e) { + log.info("can't create home dir", e); + throw new DisworkException("can't create home dir", e); + } + config.setOwnerId(ownerId); - + + config.setFirstRunTime(System.currentTimeMillis()); + // config.saveForUser(); } - + log.info("owner id is " + ownerId); - + + // check if config implies to run a worker + if (config.getNumberOfWorkers() >= 0) { + workers = new WorkersManager(fileSystem, config); + } else { + log.info("worker manager disabled"); + } + + sessionStartTime = System.currentTimeMillis(); + + // writing hardware info to homeDir try { - homeDir = HOME + "/" + ownerId; - if (!fileSystem.exists(homeDir)) { - fileSystem.createDirectory(homeDir); - } + OperatingSystemMXBean os = ManagementFactory.getOperatingSystemMXBean(); + String hardinfos = os.getName() + "\n" + os.getArch() + "\n" + + os.getAvailableProcessors(); + + // TODO 20100615 bleny add RAM size and HDD capacities to hardinfos + + fileSystem.write(homeDir + "/" + HARDINFO_PATH, + IOUtils.toInputStream(hardinfos)); + log.info("writing hardware infos " + hardinfos); } catch (ConcurrentModificationException e) { - log.info("can't create home dir", e); - throw new DisworkException("can't create home dir", e); + log.info("can't write hardware info", e); + throw new DisworkException("can't write hardware info", e); } catch (IOException e) { - log.info("can't create home dir", e); - throw new DisworkException("can't create home dir", e); + log.info("can't write hardware info", e); + throw new DisworkException("can't write hardware info", e); } - - // check if config implies to run a worker - workers = new WorkersManager(fileSystem, config); - workers.start(); } + /** + * Create all base directories on the File System + * @throws ConcurrentModificationException + * @throws IOException + */ protected void initFileSystem() throws ConcurrentModificationException, IOException { String[] directories = { TODO, TODO_RUNNING, FAILED_1, FAILED_1_RUNNING, @@ -170,6 +222,8 @@ } } + /* *** methods for defining some usual paths *** */ + /** * given a name and a version for an application, returns the path where * application data can be found on diswork file system. @@ -188,6 +242,27 @@ } /** + * Given a job description, returns the place on disworkFS where all data + * for this jobs should be stored + * @param jobDescription + * @return an absolute path + */ + protected String getPathForJob(JobDescription jobDescription) { + return getPathForJob(jobDescription.getJobId()); + } + + /** + * Given a job id, returns the place on disworkFS where all data + * for this jobs should be stored + * @param jobDescription + * @return a path + */ + protected String getPathForJob(String jobId) { + // all jobs are stored in home dir + return homeDir + "/" + jobId; + } + + /** * every-time a link to a job is created or modified in the job, his name * has to be generated by this method * @return the name to use for a link @@ -196,9 +271,22 @@ return ((Long) System.currentTimeMillis()).toString(); } - public void submitApplication(String applicationName, - String applicationVersion, - InputStream applicationData) throws DisworkException { + /** + * Provide an application to all nodes. Once provided, all nodes will be + * 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 + * is done : the application file is NOT replaced. + * + * @param applicationName the name of the application + * @param applicationVersion the version of the application + * @param applicationData an InputStream on the application .zip file + * @throws DisworkException if an error occurs while uploading the file + */ + public void submitApplication(String applicationName, + String applicationVersion, InputStream applicationData) + throws DisworkException { // the place where dependency will be stored String path = getPathForDependency(applicationName, applicationVersion); @@ -221,108 +309,87 @@ throw new DisworkException("unable to write", e); } } - - /** - * Given a job description, returns the place on disworkFS where all data - * for this jobs should be stored - * @param jobDescription - * @return a path - */ - protected String getJobPath(JobDescription jobDescription) { - return getJobPath(jobDescription.getJobId()); - } - /** - * Given a job description, returns the place on disworkFS where all data - * for this jobs should be stored - * @param jobDescription - * @return a path - */ - protected String getJobPath(String jobId) { - // all jobs are stored in home dir - return homeDir + "/" + jobId; - } - - /** - * - */ - public JobDescription newJob() throws IOException { - Random random = new Random(); - - boolean alreadyExists = true; - String newJobIntendifier = null; - while (alreadyExists) { - Integer randomInteger = random.nextInt(); - newJobIntendifier = "job_" + randomInteger.toString(); - alreadyExists = fileSystem.exists(getJobPath(newJobIntendifier)); - } - - // create both job path and sub-directory .diswork - fileSystem.createDirectories(getJobPath(newJobIntendifier) + "/" + ".diswork"); - log.info("created new job " + newJobIntendifier); - return new JobDescription(newJobIntendifier); - } - public void submitJob(JobDescription jobDescription) throws DisworkException { - submitJob(jobDescription, new HashMap<String, InputStream>()); - } - - public void submitJob(JobDescription jobDescription, - Map<String, InputStream> inputFiles) throws DisworkException { - // check dependencies, throw exception - - if (inputFiles.size() + jobDescription.getStagingInputUrls().size() + if (jobDescription.getInputData().size() + jobDescription.getStagingInputUrls().size() < jobDescription.getStagingInput().size()) { // dependencies are missing } try { - String dependencyPath = - getPathForDependency(jobDescription.getApplicationName(), - jobDescription.getApplicationVersion()); - log.info("looking for " + dependencyPath); - if (!fileSystem.exists(dependencyPath)) { - throw new DisworkException("job require a dependency " + - jobDescription.getApplicationName() + "-" + - jobDescription.getApplicationVersion() + " that is " + - "not available"); + // trying to put the job in a new directory of home + Random random = new Random(); + boolean alreadyExists = true; + String newJobIntendifier = null; + while (alreadyExists) { + Integer randomInteger = random.nextInt(); + newJobIntendifier = "job_" + randomInteger.toString(); + alreadyExists = fileSystem.exists(getPathForJob(newJobIntendifier)); } + + jobDescription.setJobId(newJobIntendifier); - String jobDir = getJobPath(jobDescription); + // create both job path and sub-directory .diswork + fileSystem.createDirectories( + getPathForJob(jobDescription) + "/" + ".diswork"); + + if (jobDescription.applicationName != null) { + String dependencyPath = getPathForDependency( + jobDescription.getApplicationName(), + jobDescription.getApplicationVersion()); + log.info("looking for " + dependencyPath); + + if (!fileSystem.exists(dependencyPath)) { + throw new DisworkException("job require a dependency " + + jobDescription.getApplicationName() + "-" + + jobDescription.getApplicationVersion() + + " that is not available"); + } + } else { + log.info("no dependency specified for " + jobDescription); + } + String jobDir = getPathForJob(jobDescription); + if(!fileSystem.exists(jobDir)) { // strange ! } + // creating an empty log file + log.info("creating log file " + jobDir + "/" + LOG_PATH); fileSystem.write(jobDir + "/" + LOG_PATH, IOUtils.toInputStream("")); - + + // writing the JSDL file InputStream jobJSDL = IOUtils.toInputStream(jobDescription.toJSDL()); fileSystem.write(jobDir + "/" + JSDL_PATH, jobJSDL); // file staging - for (String fileName : inputFiles.keySet()) { - fileSystem.write(jobDir + "/" + fileName, inputFiles.get(fileName)); + for (String fileName : jobDescription.getInputData().keySet()) { + fileSystem.write(jobDir + "/" + fileName, + jobDescription.getInputData().get(fileName)); } - + + // FIXME 20100609 bleny may throws exception if jobs are proposed + // at a same time + // propose job String linkName = newJobLinkName(); - - // FIXME 20100609 bleny may throws exception if jobs are proposed - // at a same time fileSystem.createSymbolicLink(TODO + "/" + linkName, jobDir); + log.info("job submited"); + } catch (IOException e) { log.error("file system error", e); throw new DisworkFileSystemException("file system error", e); } } - public boolean checkLogContains(JobDescription job, + protected boolean checkLogContains(JobDescription job, String pattern) throws DisworkException { try { - String jobPath = getJobPath(job); + String jobPath = getPathForJob(job); List<?> entries = IOUtils.readLines(fileSystem.read(jobPath + "/" + LOG_PATH)); return entries.contains(pattern); } catch (FileNotFoundException e) { @@ -345,28 +412,107 @@ public boolean isFailed(JobDescription job) throws DisworkException { return isFinished(job) && !isSuccessful(job); } + + public Map<String, InputStream> getResults(JobDescription job) + throws DisworkException { + if (isFinished(job)) { + Map<String, InputStream> results = new HashMap<String, InputStream>(); + for (String fileName : job.getStagingOutput()) { + String jobPath = getPathForJob(job); + try { + InputStream result = fileSystem.read(jobPath + "/" + fileName); + results.put(fileName, result); + } catch (FileNotFoundException e) { + throw new DisworkException("an expected file is missing", e); + } catch (IOException e) { + log.info("file system error ", e); + throw new DisworkException("file system error ", e); + } + } + return results; + } else { + throw new DisworkException("can't get results for unfinished job " + + job); + } + } + + /** close the daemon (stop all workers) + * update statistices and delete all temporary data + */ @Override public void close() throws IOException { workers.stop(); + + // updating total uptime statistic + Long totalUptime = getTotalUptime(); + log.info("saving total uptime: " + totalUptime); + config.setTotalUptime(totalUptime); + //config.saveForUser(); + fileSystem.close(); + + FileUtil.deleteRecursively(config.getTempDirectory()); } - public Map<String, InputStream> getResults(JobDescription job) - throws DisworkException { - Map<String, InputStream> results = new HashMap<String, InputStream>(); - for (String fileName : job.getStagingOutput()) { - String jobPath = getJobPath(job); - try { - InputStream result = fileSystem.read(jobPath + "/" + fileName); - results.put(fileName, result); - } catch (FileNotFoundException e) { - throw new DisworkException("an expected file is missing", e); - } catch (IOException e) { - log.info("file system error ", e); - throw new DisworkException("file system error ", e); + /* *** methods about statistics *** */ + + protected Long getTotalUptime() { + Long currentTime = System.currentTimeMillis(); + Long sessionUptime = currentTime - sessionStartTime; + Long totalUptime = config.getTotalUptime() + sessionUptime; + return totalUptime; + } + + /** + * return a ratio of the total uptime on the time since the first demon + * run. For exemple, if daemon was up 2 hours and was installed 4 hours ago + * this method return 0.5. + * @return + */ + public Double getUptimeRatio() { + Double uptimeRatio = (double) getTotalUptime().longValue() + / ((double) System.currentTimeMillis() + - (double) config.getFirstRunTime().longValue()); + return uptimeRatio; + } + + /** get informations on hardware available on the global Diswork system + * + * @return + * @throws DisworkException + */ + public Map<String, Integer> getGlobalStats() throws DisworkException { + try { + Map<String, Integer> stats = new HashMap<String, Integer>(); + stats.put("available_processors", 0); + + List<String> homeDirs = fileSystem.readDirectory(HOME); + for (String homeDir : homeDirs) { + String hardinfo = IOUtils.toString(fileSystem.read( + HOME + "/" + homeDir + "/" + HARDINFO_PATH)); + String[] infos = hardinfo.split("\n"); + + // reading the OS name + if (!stats.containsKey(infos[0])) { + stats.put(infos[0], 0); + } + // reading the architecture + stats.put(infos[0], stats.get(infos[0]) + 1); + if (!stats.containsKey(infos[1])) { + stats.put(infos[1], 0); + } + stats.put(infos[1], stats.get(infos[1]) + 1); + + // reading the number of processors + stats.put("available_processors", + stats.get("available_processors") + + Integer.parseInt(infos[2])); } + return stats; + } catch (IOException e) { + log.info("file system error ", e); + throw new DisworkException("file system error ", e); } - return results; } } \ No newline at end of file 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-14 08:53:35 UTC (rev 75) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemonRunner.java 2010-06-16 15:55:53 UTC (rev 76) @@ -24,6 +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; @@ -34,12 +37,21 @@ public class DisworkDaemonRunner { private static final Log log = LogFactory.getLog(DisworkDaemonRunner.class); - + /** * @param args */ public static void main(String[] args) { + 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); 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-14 08:53:35 UTC (rev 75) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/JobDescription.java 2010-06-16 15:55:53 UTC (rev 76) @@ -24,46 +24,86 @@ */ package org.nuiton.diswork.daemon; -import java.io.Serializable; +import java.io.IOException; +import java.io.InputStream; +import java.net.MalformedURLException; import java.net.URL; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; +import org.apache.commons.io.IOUtils; +import org.apache.commons.io.input.NullInputStream; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.jdom.Attribute; +import org.jdom.Document; +import org.jdom.Element; +import org.jdom.JDOMException; +import org.jdom.Namespace; +import org.jdom.input.SAXBuilder; + /** * + * When writing your command line, you should use provided tokens. Token + * are piece of command that will be replaced by the job-executor when + * needed. * + * <dl> + * <dt>%java</dt> + * </dd>will be replaced by the actual path of java executable + * into the JRE</dd> + * </dl> * + * This class provides methods to read and parse those data to an XML file + * following (as far as possible), the Job Submission Description Language + * Specification. + * + * @see http://en.wikipedia.org/wiki/Job_Submission_Description_Language + * @see http://www.gridforum.org/documents/GFD.56.pdf + * * @author bleny */ -public class JobDescription implements Serializable { +public class JobDescription { - private static final long serialVersionUID = -8493700934802808925L; - - /** an id for diswork */ + private static final Log log = LogFactory.getLog(JobDescription.class); + + /** an id for diswork, not stored in the JSDL file */ protected String jobId; + /** a name for the job, it's a convenience for the user */ protected String jobName; + /** the name of the application needed for this job + * can be null if no application is needed for complete this job + */ protected String applicationName; + + /** the version of the application + * can't be null applicationName is set + */ protected String applicationVersion; + /** the command line to execute this job */ protected String commandLine; /** all files expected at the beginning of the job */ - protected List<String> stagingInput = new ArrayList<String>(); + protected List<String> input = new ArrayList<String>(); /** all files expected at the end of the job */ - protected List<String> stagingOutput = new ArrayList<String>(); + protected List<String> output = new ArrayList<String>(); - /** the name of a file and the URI where to get it */ - protected Map<String, URL> stagingInputUrls = new HashMap<String, URL>(); + /** the name of some input files and the URI where to get it */ + protected Map<String, URL> inputUrls = new HashMap<String, URL>(); + /** */ + protected Map<String, InputStream> inputData = new HashMap<String, InputStream>(); + /** file where to read the standard input, may be null */ protected String standardInput; - /** file where to write the standard output */ + /** file where to write the standard output, may be null */ protected String standardOutput; /** TOKENS are piece of string you can use for writing command lines */ @@ -85,9 +125,15 @@ this.jobId = jobId; } + public JobDescription() {} + public String getJobId() { return jobId; } + + protected void setJobId(String jobId) { + this.jobId = jobId; + } public String getCommandLine() { String result = commandLine; @@ -108,19 +154,39 @@ public String getApplicationName() { return applicationName; } + + /** + * this method is protected to force the use of + * {@link #setApplication(String, String)} which need a version. + * This method is still here because it is used by + * {@link #parseJSDL(String)} + */ + protected void setApplicationName(String applicationName) { + this.applicationName = applicationName; + } public String getApplicationVersion() { return applicationVersion; } + /** + * this method is protected to force the use of + * {@link #setApplication(String, String)} which need a version. + * This method is still here because it is used by + * {@link #parseJSDL(String)} + */ + protected void setApplicationVersion(String applicationVersion) { + this.applicationVersion = applicationVersion; + } + public void setJobName(String jobName) { this.jobName = jobName; } public void setApplication(String applicationName, String applicationVersion) { - this.applicationName = applicationName; - this.applicationVersion = applicationVersion; + setApplicationName(applicationName); + setApplicationVersion(applicationVersion); } @Override @@ -132,38 +198,192 @@ protected static Map<String, JobDescription> map = new HashMap<String, JobDescription>(); public String toJSDL() { + /* count += 1; map.put(count.toString(), this); return count.toString(); - } + */ + + String jsdl = + "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n" + + "<jsdl:JobDefinition xmlns=\"http://www.example.org/\"\n" + + " xmlns:jsdl=\"http://schemas.ggf.org/jsdl/2005/11/jsdl\"\n" + + " xmlns:jsdl-posix=\"http://schemas.ggf.org/jsdl/2005/11/jsdl-posix\"\n" + + " xmlns:diswork=\"http://nuiton.org/projects/show/diswork\"\n" + + " xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\">\n" + + "<jsdl:JobDescription>\n" + + " <jsdl:JobIdentification>\n" + + " <jsdl:JobName>" + jobName + "</jsdl:JobName>\n" + + " </jsdl:JobIdentification>\n" + + " <jsdl:Application>\n"; + if (applicationName != null) { + jsdl += " <jsdl:ApplicationName>" + applicationName + "</jsdl:ApplicationName>\n" + + " <jsdl:ApplicationVersion>" + applicationVersion + "</jsdl:ApplicationVersion>\n"; + } + jsdl += " <jsdl-posix:POSIXApplication>\n" + + " <jsdl-posix:Executable />\n" + + " <jsdl-posix:Argument>" + commandLine + "</jsdl-posix:Argument>\n"; + if (standardInput != null) { + jsdl += " <jsdl-posix:Input>" + standardInput + "</jsdl-posix:Input>\n"; + } + if (standardOutput != null) { + jsdl += " <jsdl-posix:Output>" + standardOutput + "</jsdl-posix:Output>\n"; + } + jsdl += " </jsdl-posix:POSIXApplication>\n" + + " </jsdl:Application>\n"; + + for (String inputName : input) { + jsdl += " <jsdl:DataStaging diswork:type=\"in\">\n" + + " <jsdl:FileName>" + inputName + "</jsdl:FileName>\n"; + if (inputUrls.containsKey(inputName)) { + jsdl += + " <jsdl:Source>\n" + + " <jsdl:URI>" + inputUrls.get(inputName) + "</jsdl:URI>\n" + + " </jsdl:Source>\n"; + } + jsdl += " </jsdl:DataStaging>\n"; + } - public static JobDescription parseJSDL(String jsdl) { - return map.get(jsdl); + for (String outputName : output) { + jsdl += " <jsdl:DataStaging diswork:type=\"out\">\n" + + " <jsdl:FileName>" + outputName + "</jsdl:FileName>\n" + + " </jsdl:DataStaging>\n"; + } + + jsdl += + "</jsdl:JobDescription>\n" + + "</jsdl:JobDefinition>\n"; + + return jsdl; + } - public void addStagingInput(String fileName, URL source) { - stagingInput.add(fileName); - stagingInputUrls.put(fileName, source); + /** + * Factory method to get a JobDescription from JSDL + * @param jsdl the content of the JSDL file + * @return a job description representing the content of the JSDL + * @throws IOException if JSDL is malformed or if an URL is malformed + */ + public static JobDescription parseJSDL(String jsdl) throws IOException { + // TODO 20100616 bleny correctly set dependency to JDOM in pom.xml + JobDescription result = new JobDescription(); + + try { + SAXBuilder builder = new SAXBuilder(); + Document document = builder.build(IOUtils.toInputStream(jsdl)); + Element jobDefinition = document.getRootElement(); + + // namespaces + Namespace jsdlNamespace = jobDefinition.getNamespace("jsdl"); + Namespace jsdlPosixNamespace = jobDefinition.getNamespace("jsdl-posix"); + Namespace disworkNamespace = jobDefinition.getNamespace("diswork"); + + // main element + Element jobDescription = jobDefinition.getChild("JobDescription", + jsdlNamespace); + + // job identification + Element jobIdentification = jobDescription.getChild + ("JobIdentification", jsdlNamespace); + Element jobName = jobIdentification.getChild("JobName", + jsdlNamespace); + result.setJobName(jobName.getText()); + + // application + Element application = jobDescription.getChild("Application", + jsdlNamespace); + Element applicationName = application.getChild("ApplicationName", + jsdlNamespace); + if (application != null) { + Element applicationVersion = application.getChild + ("ApplicationVersion", jsdlNamespace); + result.setApplication(applicationName.getText(), + applicationVersion.getText()); + } + + Element POSIXApplication = application.getChild("POSIXApplication", + jsdlPosixNamespace); + + Element argument = POSIXApplication.getChild("Argument", + jsdlPosixNamespace); + result.setCommandLine(argument.getText()); + + Element input = POSIXApplication.getChild("Input", + jsdlPosixNamespace); + if (input != null) { + result.setStandardInput(input.getText()); + } + Element output = POSIXApplication.getChild("Output", + jsdlPosixNamespace); + if (input != 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" + Element source = dataStaging.getChild("Source", + jsdlNamespace); + if (source != null) { + Element URI = source.getChild("URI", jsdlNamespace); + result.addInput(fileName.getText(), new URL(URI.getText())); + } else { + result.addInput(fileName.getText(), new NullInputStream(0)); + } + + // type not set + if (type == null) { + 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); + } catch (MalformedURLException e) { + log.error("malformed URL", e); + throw new IOException("malformed URL", e); + } + return result; } - public void addStagingInput(String fileName) { - stagingInput.add(fileName); + public void addInput(String fileName, URL source) { + input.add(fileName); + inputUrls.put(fileName, source); } + + public void addInput(String fileName, InputStream source) { + input.add(fileName); + inputData.put(fileName, source); + } - public void addStagingOutput(String fileName) { - stagingOutput.add(fileName); + public void addOutput(String fileName) { + output.add(fileName); } public List<String> getStagingInput() { - return stagingInput; + return input; } public List<String> getStagingOutput() { - return stagingOutput; + return output; } public Map<String, URL> getStagingInputUrls() { - return stagingInputUrls; + return inputUrls; } public String getStandardInput() { @@ -182,4 +402,8 @@ standardOutput = fileName; } + public Map<String, InputStream> getInputData() { + return inputData; + } + } \ No newline at end of file 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-14 08:53:35 UTC (rev 75) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java 2010-06-16 15:55:53 UTC (rev 76) @@ -32,16 +32,19 @@ import java.io.InputStream; import java.io.OutputStream; import java.net.URL; +import java.util.ArrayList; import java.util.Collections; 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; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.nuiton.diswork.fs.DisworkFileSystem; +import org.nuiton.util.FileUtil; import org.nuiton.util.ZipUtil; /** @@ -57,14 +60,18 @@ /** 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 { 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, DisworkDaemon.FAILED_1); - RUNNING_MOVE.put(DisworkDaemon.FAILED_1, DisworkDaemon.FAILED_2); - RUNNING_MOVE.put(DisworkDaemon.FAILED_2, DisworkDaemon.FAILED_3); + 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); } private static final Log log = LogFactory.getLog(WorkersManager.class); @@ -72,15 +79,17 @@ protected DisworkFileSystem fileSystem; protected DisworkConfig config; - // Pool of workers ? - protected Worker worker; + // Pool of workers + protected List<Worker> workers = new ArrayList<Worker>(); - // Activity strategy ? - - protected class Worker implements Runnable { + protected ActivityStrategy activityStrategy; + + protected class Worker extends Thread { public boolean shouldStop = false; + public WorkersManager manager; + protected void log(String jobPath, String message) throws IOException { String logPath = jobPath + "/" + DisworkDaemon.LOG_PATH; InputStream oldLogAsStream = fileSystem.read(logPath); @@ -113,24 +122,32 @@ log.info("will run job " + jobDescription); // create temp dir - File jobDir = new File(config.getTempDirectory(), jobDescription.getJobId()); + Random random = new Random(); + File jobDir = new File(config.getTempDirectory(), + String.valueOf(random.nextInt())); jobDir.mkdirs(); // download application - String applicationPath = DisworkDaemon.getPathForDependency( + if (jobDescription.getApplicationName() != null) { + log.info("dependency needed for " + jobDescription + " (" + + jobDescription.getApplicationName() + "-" + + jobDescription.getApplicationVersion() + ")"); + String applicationPath = DisworkDaemon.getPathForDependency( jobDescription.getApplicationName(), jobDescription.getApplicationVersion()); - InputStream applicationData = fileSystem.read(applicationPath); - - - File application = new File(jobDir, FilenameUtils.getName(applicationPath)); - application.createNewFile(); - log.info("will create " + application.getAbsolutePath()); - OutputStream out = new FileOutputStream(application); - IOUtils.copy(applicationData, out); - // unzip application - ZipUtil.uncompress(application, jobDir); - + InputStream applicationData = fileSystem.read(applicationPath); + + File application = new File(jobDir, + FilenameUtils.getName(applicationPath)); + application.createNewFile(); + log.info("will create " + application.getAbsolutePath()); + OutputStream out = new FileOutputStream(application); + IOUtils.copy(applicationData, out); + // unzip application + ZipUtil.uncompress(application, jobDir); + } else { + log.info("no dependency specified for " + jobDescription); + } // staging input files for (String fileName : jobDescription.getStagingInput()) { File localCopy = new File(jobDir, fileName); @@ -151,16 +168,17 @@ // executing the job String commandLine = jobDescription.getCommandLine(); log.info("calling " + commandLine); - String[] bidule = commandLine.split(" "); - ProcessBuilder builder = new ProcessBuilder(bidule); + String[] commandLineElements = commandLine.split(" "); + ProcessBuilder builder = new ProcessBuilder(commandLineElements); builder.directory(jobDir); builder.redirectErrorStream(true); Process job = builder.start(); - // plugin a file on the standard input + // plugging a file on the standard input String standardInputFileName = jobDescription.getStandardInput(); if (standardInputFileName != null) { - InputStream input = new FileInputStream(new File(jobDir, standardInputFileName)); + InputStream input = new FileInputStream( + new File(jobDir, standardInputFileName)); IOUtils.copy(input, job.getOutputStream()); } @@ -177,7 +195,8 @@ // dump the standard output in a file String standardOutputFileName = jobDescription.getStandardOutput(); if (standardOutputFileName != null) { - OutputStream output = new FileOutputStream(new File(jobDir, standardOutputFileName)); + OutputStream output = new FileOutputStream( + new File(jobDir, standardOutputFileName)); IOUtils.copy(job.getInputStream(), output); } @@ -201,78 +220,207 @@ } // clean up the job directory - // FileUtil.deleteRecursively(jobDir); + FileUtil.deleteRecursively(jobDir); boolean success = exitValue == 0; if (success) { - log(jobPath, "DONE\nFINISHED"); + log(jobPath, "DONE"); } else { - log(jobPath, "FAILED\nFINISHED"); + log(jobPath, "FAILED"); } return success; } + + public String getFistJobName(String path) throws IOException { + List<String> jobsNames = fileSystem.readDirectory(path); + if (jobsNames.size() == 0) { + return null; + } else { + Collections.sort(jobsNames); + return jobsNames.get(0); + } + } - @Override - public void run() { - while (! shouldStop) { - // try to find a new job - try { - // TODO 20100609 bleny watch for other jobs - List<String> jobsNames = - fileSystem.readDirectory(DisworkDaemon.TODO); - if (jobsNames.size() != 0) { - // sort and choose the first, due to names, it should be - // the more ancient one - Collections.sort(jobsNames); - String oldName = jobsNames.get(0); - - String newName = DisworkDaemon.newJobLinkName(); - fileSystem.move(DisworkDaemon.TODO + "/" + oldName, - DisworkDaemon.TODO_RUNNING + "/" + newName); - - boolean jobSuccess = runJob(DisworkDaemon.TODO_RUNNING + "/" + newName); - - oldName = newName; - newName = DisworkDaemon.newJobLinkName(); - - if (jobSuccess) { - fileSystem.move( - DisworkDaemon.TODO_RUNNING + "/" + oldName, - DisworkDaemon.DONE + "/" + newName); - } else { - fileSystem.move( - DisworkDaemon.TODO_RUNNING + "/" + oldName, - DisworkDaemon.FAILED_3 + "/" + newName); + protected void findAJobAndRunIt() { + // try to find a new job + try { + String jobLinkDir = null; + String jobLinkName = null; + + String[] runningJobsDirs = { DisworkDaemon.FAILED_2_RUNNING, + DisworkDaemon.FAILED_1_RUNNING, + DisworkDaemon.TODO_RUNNING + }; + 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; } + } + } + + 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 (jobLinkDir == null) { + log.info("nothing to do"); + Thread.sleep(JOB_WAIT); + } else { + // move the link before running the job + String oldPath = jobLinkDir + "/" + jobLinkName; + log.info("job found at " + oldPath); + + String newPath = RUNNING_MOVE.get(jobLinkDir) + "/" + + DisworkDaemon.newJobLinkName(); + + 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 { - log.info("nothing to do"); - Thread.sleep(JOB_WAIT); + newDir = FAILED_MOVE.get(jobLinkDir); } + newPath = newDir + "/" + DisworkDaemon.newJobLinkName(); - } catch (IOException e) { - log.error("error while reading jobs", e); - // TODO 20100611 bleny manage exception + log.info("moving " + oldPath + " to " + newPath); + fileSystem.move(oldPath, newPath); + + // mark the job has finished if done or failed too many + // times + if ( newDir == DisworkDaemon.DONE || + newDir == DisworkDaemon.FAILED_3) { + log.info("marking " + newPath + " as finished"); + log(newPath, "FINISHED"); + } + } + + + } catch (IOException e) { + log.error("error while reading jobs", e); + // TODO 20100611 bleny manage exception + } catch (InterruptedException e) { + log.info("exception catch", e); + // TODO 20100611 bleny manage exception + } + + } + + @Override + public void run() { + while (! shouldStop) { + if (manager.getActivityStrategy().canWork()) { + findAJobAndRunIt(); + } + try { + Thread.sleep(10*1000); } catch (InterruptedException e) { + // TODO 20100615 bleny Auto-generated catch block log.info("exception catch", e); - // TODO 20100611 bleny manage exception + e.printStackTrace(); } } } - } public WorkersManager(DisworkFileSystem fileSystem, DisworkConfig config) { this.fileSystem = fileSystem; this.config = config; - } - public void start() { - worker = new Worker(); - Thread t = new Thread(worker); - t.start(); + 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)) { + activeNoActivityStrategy(); + } else if ("unlimited".equals(initialStrategy)) { + activeUnlimitedActivityStrategy(); + } else if ("limited".equals(initialStrategy)) { + activeLimitedActivityStrategy(); + } else if ("scheduled".equals(initialStrategy)) { + activeScheduledActivityStrategy(); + } else { + log.error("wrong config directive " + initialStrategy); + activeNoActivityStrategy(); + } + } public void stop() { - worker.shouldStop = true; + stop(false); } + + public void stop(boolean now) { + // asking to all threads to stop + for (Worker worker : workers) { + worker.shouldStop = true; + } + + if( !now ) { + // waiting for them to actually have finished + for (Worker worker : workers) { + while (worker.isAlive()) { + try { + Thread.sleep(10 * 1000); + } catch (InterruptedException e) { + // TODO 20100615 bleny Auto-generated catch block + log.info("exception catch", e); + e.printStackTrace(); + } + } + } + } + } + + public ActivityStrategy getActivityStrategy() { + return activityStrategy; + } + + public void setActivityStrategy(ActivityStrategy activityStrategy) { + this.activityStrategy = activityStrategy; + } + + public void activeNoActivityStrategy() { + activityStrategy = new ActivityStrategy.NoActivity(); + } + + public void activeUnlimitedActivityStrategy() { + activityStrategy = new ActivityStrategy.UnlimitedActivity(); + } + + public void activeLimitedActivityStrategy() { + activityStrategy = new ActivityStrategy.LimitedActivity(); + } + + public void activeScheduledActivityStrategy() { + activityStrategy = new ActivityStrategy.ScheduledActivity(); + } + } \ No newline at end of file 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-14 08:53:35 UTC (rev 75) +++ trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonTest.java 2010-06-16 15:55:53 UTC (rev 76) @@ -5,7 +5,6 @@ import java.io.InputStream; import java.net.URL; -import java.util.HashMap; import java.util.Map; import org.apache.commons.io.IOUtils; @@ -15,9 +14,9 @@ public class DisworkDaemonTest { - protected DisworkDaemon daemon; + protected static DisworkDaemon daemon; - protected static int port = 39999; + protected static int port = 49999; @Before public void setUp() throws Exception { @@ -31,19 +30,27 @@ @After public void tearDown() throws Exception { + Thread.sleep(5000); daemon.close(); } + @Test + public void simpleSubmit() throws Exception { + JobDescription job = new JobDescription(); + job.setCommandLine("java -version"); + daemon.submitJob(job); + } + @Test(expected = DisworkException.class) public void testSubmitWithoutDependency() throws Exception { - JobDescription job = daemon.newJob(); + JobDescription job = new JobDescription(); job.setApplication("non-existing-application", "0.0"); daemon.submitJob(job); } @Test public void testSubmitSuccessfulJob() throws Exception { - JobDescription job = daemon.newJob(); + JobDescription job = new JobDescription(); job.setApplication("fake-app", "1.0"); job.setCommandLine("%java -jar fake-app.jar"); daemon.submitJob(job); @@ -53,11 +60,12 @@ } assertTrue(daemon.isSuccessful(job)); + } @Test public void testSubmitFailJob() throws Exception { - JobDescription job = daemon.newJob(); + JobDescription job = new JobDescription(); job.setApplication("fake-app", "1.0"); job.setCommandLine("%java -jar fake-app.jar fail"); daemon.submitJob(job); @@ -69,38 +77,73 @@ assertTrue(daemon.isFailed(job)); } + /** + * Create a complex job, submit it and check the results: + * <ul> + * <li>the job need an application (fake-app version 1.0);</li> + * <li>the job come with input data (input.txt);</li> + * <li>the job need another input file to be downloaded from http;</li> + * <li>the job ask for a file to be provided as a result at the end + * of the job.</li> + * <li>standard output is asked has result</li> + * </ul> + * + */ @Test public void testStaging() throws Exception { - JobDescription job = daemon.newJob(); + JobDescription job = new JobDescription(); job.setJobName("My Job"); job.setApplication("fake-app", "1.0"); job.setCommandLine("%java -jar fake-app.jar"); - job.addStagingInput("example.com_index", new URL("http://www.example.com/")); - job.addStagingInput("input.txt"); - job.addStagingOutput("output.txt"); + // defining data in input + job.addInput("example.com_index", new URL("http://www.example.com/")); + job.addInput("input.txt", ClassLoader.getSystemResourceAsStream("input.txt")); + // defining expected data in output + job.addOutput("output.txt"); + job.addOutput("example.com_index"); + + // setting standard input and output file job.setStandardInput("input.txt"); - job.setStandardOutput("output.txt"); - - Map<String, InputStream> data = new HashMap<String, InputStream>(); - data.put("input.txt", ClassLoader.getSystemResourceAsStream("input.txt")); - - daemon.submitJob(job, data); + job.setStandardOutput("output.txt"); + + // submit the job + daemon.submitJob(job); + // waiting for the job to finish while(! daemon.isFinished(job)) { - Thread.sleep(1 * 1000); + Thread.sleep(5 * 1000); } + // check that job is successful assertTrue(daemon.isSuccessful(job)); + // checking the presence of results Map<String, InputStream> results = daemon.getResults(job); - + assertEquals(2, results.size()); assertTrue(results.containsKey("output.txt")); + assertTrue(results.containsKey("example.com_index")); + // checking that results are what was expected String output = IOUtils.toString(results.get("output.txt")); assertEquals("a print on standard output\n", output); - + output = IOUtils.toString(results.get("example.com_index")); + assertTrue(output.contains("Example Web Page")); } + /** + * tests the stats given by the daemon + * @throws Exception + */ + @Test + public void testStats() throws Exception { + daemon.getUptimeRatio(); + + Map<String, Integer> stats = daemon.getGlobalStats(); + // deamon should read 3 stats : 1 OS, 1 architecture and the number + // of processors + assertEquals(3, stats); + } + } Added: trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/JobDescriptionTest.java =================================================================== --- trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/JobDescriptionTest.java (rev 0) +++ trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/JobDescriptionTest.java 2010-06-16 15:55:53 UTC (rev 76) @@ -0,0 +1,98 @@ +package org.nuiton.diswork.daemon; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.IOException; +import java.net.URL; + +import org.apache.commons.collections.ListUtils; +import org.apache.commons.io.IOUtils; +import org.junit.Before; +import org.junit.Test; + +/** + * Tests in this class create a complex JobDescription and check that + * transformation to JSDL and back to an object keep all informations + * + * @author bleny + */ +public class JobDescriptionTest { + + protected static JobDescription job; + protected static JobDescription job2; + + @Before + public void setUp() throws Exception { + job = new JobDescription(); + job.setJobName("My Job"); + job.setApplication("fake-app", "1.0"); + job.setCommandLine("%java -jar fake-app.jar"); + + job2 = new JobDescription(); + job2.setJobName("My Job"); + job2.setApplication("fake-app", "1.0"); + job2.setCommandLine("%java -jar fake-app.jar"); + + + // defining data in input + job2.addInput("example.com_index", new URL("http://www.example.com/")); + job2.addInput("input.txt", IOUtils.toInputStream("")); + + // defining expected data in output + job2.addOutput("output.txt"); + job2.addOutput("example.com_index"); + + // setting standard input and output file + job2.setStandardInput("input.txt"); + job2.setStandardOutput("output.txt"); + + } + + @Test + public void testParseJSDL1() throws Exception { + try { + JobDescription jobCopy = JobDescription.parseJSDL(job.toJSDL()); + assertNotNull(jobCopy); + assertEquals(job.getJobName(), jobCopy.getJobName()); + assertEquals(job.getApplicationName(), + jobCopy.getApplicationName()); + assertEquals(job.getApplicationVersion(), + jobCopy.getApplicationVersion()); + assertEquals(job.getCommandLine(), jobCopy.getCommandLine()); + assertEquals(job.getStandardInput(), jobCopy.getStandardInput()); + assertEquals(job.getStandardOutput(), jobCopy.getStandardOutput()); + } catch (IOException e) { + fail(); + throw e; + } + } + + @Test + public void testParseJSDL2() throws Exception { + try { + JobDescription job2Copy = JobDescription.parseJSDL(job2.toJSDL()); + assertEquals(job2.getCommandLine(), job2Copy.getCommandLine()); + assertEquals(job2.getStandardInput(), job2Copy.getStandardInput()); + assertEquals(job2.getStandardOutput(), + job2Copy.getStandardOutput()); + + assertTrue(ListUtils.isEqualList(job2.getStagingInput(), + job2Copy.getStagingInput())); + assertTrue(ListUtils.isEqualList(job2.getStagingOutput(), + job2Copy.getStagingOutput())); + assertTrue(ListUtils.isEqualList( + job2.getStagingInputUrls().keySet(), + job2Copy.getStagingInputUrls().keySet())); + assertTrue(ListUtils.isEqualList( + job2.getStagingInputUrls().values(), + job2Copy.getStagingInputUrls().values())); + } catch (IOException e) { + fail(); + throw e; + } + } + +}

1 0

r75 - trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs
by bleny＠users.nuiton.org June 14, 2010

June 14, 2010

Author: bleny Date: 2010-06-14 10:53:35 +0200 (Mon, 14 Jun 2010) New Revision: 75 Url: http://nuiton.org/repositories/revision/diswork/75 Log: mauvaise utilisation du FS dans les tests Modified: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java 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-14 08:12:34 UTC (rev 74) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java 2010-06-14 08:53:35 UTC (rev 75) @@ -83,7 +83,7 @@ */ @Test public void testWrite() throws Exception { - fileSystem.write("/", "my_file", new FileInputStream(randomFilePath)); + fileSystem.write("/my_file", new FileInputStream(randomFilePath)); } /** @@ -92,7 +92,7 @@ */ @Test public void testExists() throws Exception { - fileSystem.write("/", "my_file", new FileInputStream(randomFilePath)); + fileSystem.write("/my_file", new FileInputStream(randomFilePath)); assertTrue(fileSystem.exists("/my_file")); assertFalse(fileSystem.exists("/my_other_file")); } @@ -119,7 +119,7 @@ InputStream source; source = new ByteArrayInputStream(bytes); - fileSystem.write("/", "my_file", source); + fileSystem.write("/my_file", source); source.close(); @@ -148,7 +148,7 @@ InputStream source = new FileInputStream(randomFilePath); - fileSystem.write("/", "my_file", source); + fileSystem.write("/my_file", source); source.close(); @@ -179,7 +179,7 @@ */ @Test(expected = IOException.class) public void testFailAtWrite() throws Exception { - fileSystem.write("/my_folder", "my_file", + fileSystem.write("/my_folder/my_file", new FileInputStream(randomFilePath)); } @@ -220,10 +220,10 @@ @Test public void testWriteInFolder() throws Exception { fileSystem.createDirectory("/my_folder"); - fileSystem.write("/my_folder", "my_file", + 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", + fileSystem.write("/my_folder/my_sub_folder/my_file", new FileInputStream(randomFilePath)); assertTrue(fileSystem.exists("/my_folder/my_sub_folder/my_file")); } @@ -236,7 +236,7 @@ @Test public void testLinking() throws Exception { fileSystem.createDirectory("/my_folder"); - fileSystem.write("/my_folder", "my_file", + fileSystem.write("/my_folder/my_file", new FileInputStream(randomFilePath)); fileSystem.createSymbolicLink("/my_link", "/my_folder/my_file"); @@ -267,7 +267,7 @@ @Test public void testRemove() throws Exception { fileSystem.createDirectory("/my_folder"); - fileSystem.write("/my_folder", "my_file", + fileSystem.write("/my_folder/my_file", new FileInputStream(randomFilePath)); fileSystem.delete("/my_folder/my_file"); assertTrue(fileSystem.exists("/my_folder")); @@ -284,7 +284,7 @@ @Test(expected = IOException.class) public void testFailAtRemove() throws Exception { fileSystem.createDirectory("/my_folder"); - fileSystem.write("/my_folder", "my_file", + fileSystem.write("/my_folder/my_file", new FileInputStream(randomFilePath)); // trying to remove a non-empty directory should raise an exception @@ -298,7 +298,7 @@ public void testListDirectory() throws Exception { fileSystem.createDirectory("/my_folder"); fileSystem.createDirectory("/my_folder/my_sub_dir"); - fileSystem.write("/my_folder", "my_file", + fileSystem.write("/my_folder/my_file", new FileInputStream(randomFilePath)); fileSystem.createSymbolicLink("/my_folder/my_link", "my_file"); @@ -397,7 +397,11 @@ @Test public void testCreateDirectories() throws Exception { - fileSystem.createDirectories("/dir/subdir/subsubdir"); - assertTrue(fileSystem.exists("/dir/subdir/subsubdir")); + try { + fileSystem.createDirectories("/dir/subdir/subsubdir"); + assertTrue(fileSystem.exists("/dir/subdir/subsubdir")); + } catch (IOException e) { + fail(); + } } }

1 0

r74 - in trunk/diswork-fs/src: main/java/org/nuiton/diswork/fs/storage test/java/org/nuiton/diswork/fs
by bleny＠users.nuiton.org June 14, 2010

June 14, 2010

Author: bleny Date: 2010-06-14 10:12:34 +0200 (Mon, 14 Jun 2010) New Revision: 74 Url: http://nuiton.org/repositories/revision/diswork/74 Log: correction fuite m?\195?\169moire avec test Modified: 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/DisworkFileSystemInMemoryTest.java 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-11 15:52:09 UTC (rev 73) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/Storage.java 2010-06-14 08:12:34 UTC (rev 74) @@ -249,17 +249,17 @@ public void removeDirectory(String id) throws IOException { log.debug("removeDirectory(\"" + id + "\")"); - removeKey(id); + remove(id); } public void removeFile(String id) throws IOException { log.debug("removeFile(\"" + id + "\")"); - removeKey(id); + remove(id); } public void removeLink(String id) { log.debug("removeLink(\"" + id + "\")"); - map.remove(id); + removeKey(id); } @@ -465,7 +465,11 @@ } public void setMap(DisworkMap map) { - this.map = map; + this.map = map; } + + public DisworkMap getMap() { + return map; + } } \ No newline at end of file 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-11 15:52:09 UTC (rev 73) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java 2010-06-14 08:12:34 UTC (rev 74) @@ -43,7 +43,7 @@ /** * The file will have this fixed size */ - static protected int randomFileSize = 9999; + static protected int randomFileSize = 25 * 1000; static protected DisworkFileSystem fileSystem; 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-11 15:52:09 UTC (rev 73) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemInMemoryTest.java 2010-06-14 08:12:34 UTC (rev 74) @@ -1,6 +1,11 @@ package org.nuiton.diswork.fs; +import static org.junit.Assert.assertEquals; + +import java.io.FileInputStream; + import org.junit.Before; +import org.junit.Test; import org.nuiton.diswork.fs.DisworkFileSystem; import org.nuiton.diswork.fs.DisworkFileSystemConfig; @@ -18,4 +23,22 @@ fileSystem = new DisworkFileSystem(disworkConfig); } + @Test + public void testMemoryLeak() throws Exception { + int initialSize = fileSystem.storage.getMap().size(); + + fileSystem.createDirectory("/dir"); + fileSystem.write("/dir/file", new FileInputStream(randomFilePath)); + fileSystem.createSymbolicLink("/dir/link", "/dir/file"); + + fileSystem.delete("/dir/link"); + fileSystem.delete("/dir/file"); + fileSystem.delete("/dir"); + + fileSystem.storage.clean(); + + int finalSize = fileSystem.storage.getMap().size(); + assertEquals(initialSize, finalSize); + } + }

1 0

June 11, 2010

Author: bleny Date: 2010-06-11 17:52:09 +0200 (Fri, 11 Jun 2010) New Revision: 73 Url: http://nuiton.org/repositories/revision/diswork/73 Log: impl?\195?\169mentation du d?\195?\169mon diswork ; ajout d'op?\195?\169rations dans diswork-fs Added: trunk/diswork-daemon/src/main/java/org/ trunk/diswork-daemon/src/main/java/org/nuiton/ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkConfig.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/DisworkException.java trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkFileSystemException.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/ trunk/diswork-daemon/src/test/java/org/nuiton/ trunk/diswork-daemon/src/test/java/org/nuiton/diswork/ trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/ trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonTest.java trunk/diswork-daemon/src/test/resources/ trunk/diswork-daemon/src/test/resources/fake-app-1.0.zip trunk/diswork-daemon/src/test/resources/fake-app.jar trunk/diswork-daemon/src/test/resources/input.txt trunk/diswork-daemon/src/test/resources/log4j.properties trunk/diswork-fs/src/test/resources/ trunk/diswork-fs/src/test/resources/log4j.properties Removed: trunk/diswork-fs/src/main/resources/log4j.properties Modified: trunk/diswork-daemon/pom.xml trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/Demo.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystem.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/InMemoryDisworkMap.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/KademliaDisworkMap.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/KademliaDisworkMapTest.java Modified: trunk/diswork-daemon/pom.xml =================================================================== --- trunk/diswork-daemon/pom.xml 2010-06-10 09:32:59 UTC (rev 72) +++ trunk/diswork-daemon/pom.xml 2010-06-11 15:52:09 UTC (rev 73) @@ -31,6 +31,10 @@ <artifactId>commons-io</artifactId> </dependency> <dependency> + <groupId>commons-lang</groupId> + <artifactId>commons-lang</artifactId> + </dependency> + <dependency> <groupId>org.nuiton</groupId> <artifactId>nuiton-utils</artifactId> </dependency> Added: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkConfig.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkConfig.java (rev 0) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkConfig.java 2010-06-11 15:52:09 UTC (rev 73) @@ -0,0 +1,94 @@ +/* + * #%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 org.nuiton.diswork.fs.DisworkFileSystemConfig; +import org.nuiton.util.ApplicationConfig; + +public class DisworkConfig extends ApplicationConfig { + + protected DisworkFileSystemConfig fileSystemConfig; + + public DisworkConfig() { + setConfigFileName("diswork.config"); + } + + public String getTempDirectory() { + return System.getProperty("java.io.tmpdir", + System.getProperty("user.dir", + ".")) + "/diswork"; + } + + public static DisworkConfig newConfig() { + DisworkConfig newConfig = new DisworkConfig(); + newConfig.setFileSystemConfig(DisworkFileSystemConfig.newKademliaDisworkConfig()); + return newConfig; + } + + public String getOwnerId() { + return getOption("diswork.owner"); + } + + public void setOwnerId(String ownerId) { + setOption("diswork.owner", ownerId); + } + + + + + + public String getBootstrapIp() { + return fileSystemConfig.getBootstrapIp(); + } + + public Integer getBootstrapPort() { + return fileSystemConfig.getBootstrapPort(); + } + + public Integer getUsedPort() { + return fileSystemConfig.getUsedPort(); + } + + public void setBootstrapIp(String ip) { + fileSystemConfig.setBootstrapIp(ip); + } + + public void setBootstrapPort(Integer port) { + fileSystemConfig.setBootstrapPort(port); + } + + public void setUsedPort(Integer port) { + fileSystemConfig.setUsedPort(port); + } + + public DisworkFileSystemConfig getFileSystemConfig() { + return fileSystemConfig; + } + + public void setFileSystemConfig(DisworkFileSystemConfig fileSystemConfig) { + this.fileSystemConfig = fileSystemConfig; + } + +} Added: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemon.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemon.java (rev 0) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemon.java 2010-06-11 15:52:09 UTC (rev 73) @@ -0,0 +1,372 @@ +/* + * #%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.Closeable; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.net.UnknownHostException; +import java.util.ConcurrentModificationException; +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; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.nuiton.diswork.fs.DisworkFileSystem; +import org.nuiton.diswork.fs.DisworkFileSystemConfig; + +/** + * + * As far as possible, the use of the file system follow the UNIX Filesystem + * Hierarchy Standard + * + * @see http://en.wikipedia.org/wiki/Filesystem_Hierarchy_Standard + * + * @author bleny + */ +public class DisworkDaemon implements Closeable { + + private static final Log log = LogFactory.getLog(DisworkDaemon.class); + + protected DisworkFileSystem fileSystem; + + protected DisworkConfig config; + + /** owner id for this node */ + protected String ownerId; + + /** path to owned directory on fileSystem */ + protected String homeDir; + + /** contains applications */ + protected static final String BIN = "/bin"; + + /** new jobs */ + protected static final String TODO = "/var/jobs/todo"; + + /** jobs taken that was never tried */ + protected static final String TODO_RUNNING = "/var/jobs/todo_running"; + + /** jobs failed once */ + protected static final String FAILED_1 = "/var/jobs/failed_1"; + + /** jobs taken that was failed once */ + protected static final String FAILED_1_RUNNING = "/var/jobs/failed_1_running"; + + /** jobs failed two times */ + protected static final String FAILED_2 = "/var/jobs/failed_2"; + + /** jobs taken that was failed two times */ + protected static final String FAILED_2_RUNNING = "/var/jobs/failed_2_running"; + + /** jobs failed 3 times */ + protected static final String FAILED_3 = "/var/jobs/failed_3"; + + /** jobs done */ + protected static final String DONE = "/var/jobs/done"; + + /** a place where are all user-directories */ + protected static final String HOME = "/home"; + + /** worker-manager will make the daemon accomplish jobs */ + protected WorkersManager workers; + + /** in a job directory, the place where the JSDL must be placed */ + protected static final String JSDL_PATH = ".diswork/job.jsdl"; + + /** in a job directory, the place where the log must be placed */ + protected static final String LOG_PATH = ".diswork/job.log"; + + public DisworkDaemon(DisworkConfig config) throws DisworkException { + this.config = config; + + // init fileSystem with all needed directories + try { + DisworkFileSystemConfig fileSystemConfig = + config.getFileSystemConfig(); + + fileSystem = new DisworkFileSystem(fileSystemConfig); + initFileSystem(); + } catch (UnknownHostException e) { + log.error("bootstrap failed", e); + throw new DisworkException("bootstrap failed", e); + } catch (IOException e) { + log.error("booting diswork file system failed", e); + throw new DisworkException("booting diswork file system failed", e); + } + + + ownerId = config.getOwnerId(); // get job owner id from config + + if (ownerId == null) { + log.info("can't find owner id, generating a new one"); + // generate a new one by cheking if home dir exists + ownerId = System.getProperty("user.name", "anonymous"); + // check home dir do not exists + config.setOwnerId(ownerId); + + // config.saveForUser(); + } + + log.info("owner id is " + ownerId); + + try { + homeDir = HOME + "/" + ownerId; + if (!fileSystem.exists(homeDir)) { + fileSystem.createDirectory(homeDir); + } + } catch (ConcurrentModificationException e) { + log.info("can't create home dir", e); + throw new DisworkException("can't create home dir", e); + } catch (IOException e) { + log.info("can't create home dir", e); + throw new DisworkException("can't create home dir", e); + } + + // check if config implies to run a worker + workers = new WorkersManager(fileSystem, config); + workers.start(); + } + + protected void initFileSystem() throws ConcurrentModificationException, + IOException { + String[] directories = { TODO, TODO_RUNNING, FAILED_1, FAILED_1_RUNNING, + FAILED_2, FAILED_2_RUNNING, FAILED_3, DONE, HOME, BIN }; + // if HOME exists, we suppose all others exists + if (! fileSystem.exists(HOME)) { + for (String directory : directories) { + if (! fileSystem.exists(directory)) { + fileSystem.createDirectories(directory); + log.info("created " + directory); + } + } + } + } + + /** + * given a name and a version for an application, returns the path where + * application data can be found on diswork file system. + * @param applicationName + * @param applicationVersion + * @return a path + */ + protected static String getPathForDependency(String applicationName, + String applicationVersion) { + + String result = "/bin/" + applicationName // application directory + + "/" + // application file name + + applicationName + "-" + applicationVersion + ".zip"; + return result; + } + + /** + * every-time a link to a job is created or modified in the job, his name + * has to be generated by this method + * @return the name to use for a link + */ + protected static String newJobLinkName() { + return ((Long) System.currentTimeMillis()).toString(); + } + + public void submitApplication(String applicationName, + String applicationVersion, + InputStream applicationData) throws DisworkException { + + // the place where dependency will be stored + String path = getPathForDependency(applicationName, applicationVersion); + + String applicationDirectory = FilenameUtils.getFullPathNoEndSeparator(path); + + try { + if (!fileSystem.exists(applicationDirectory)) { + fileSystem.createDirectory(applicationDirectory); + } + + if (!fileSystem.exists(path)) { + fileSystem.write(path, applicationData); + } + } catch (ConcurrentModificationException e) { + log.info("unable to write", e); + throw new DisworkException("unable to write", e); + } catch (IOException e) { + log.info("unable to write", e); + throw new DisworkException("unable to write", e); + } + } + + /** + * Given a job description, returns the place on disworkFS where all data + * for this jobs should be stored + * @param jobDescription + * @return a path + */ + protected String getJobPath(JobDescription jobDescription) { + return getJobPath(jobDescription.getJobId()); + } + + /** + * Given a job description, returns the place on disworkFS where all data + * for this jobs should be stored + * @param jobDescription + * @return a path + */ + protected String getJobPath(String jobId) { + // all jobs are stored in home dir + return homeDir + "/" + jobId; + } + + /** + * + */ + public JobDescription newJob() throws IOException { + Random random = new Random(); + + boolean alreadyExists = true; + String newJobIntendifier = null; + while (alreadyExists) { + Integer randomInteger = random.nextInt(); + newJobIntendifier = "job_" + randomInteger.toString(); + alreadyExists = fileSystem.exists(getJobPath(newJobIntendifier)); + } + + // create both job path and sub-directory .diswork + fileSystem.createDirectories(getJobPath(newJobIntendifier) + "/" + ".diswork"); + log.info("created new job " + newJobIntendifier); + return new JobDescription(newJobIntendifier); + } + + public void submitJob(JobDescription jobDescription) throws DisworkException { + submitJob(jobDescription, new HashMap<String, InputStream>()); + } + + public void submitJob(JobDescription jobDescription, + Map<String, InputStream> inputFiles) throws DisworkException { + + // check dependencies, throw exception + + if (inputFiles.size() + jobDescription.getStagingInputUrls().size() + < jobDescription.getStagingInput().size()) { + // dependencies are missing + } + + try { + String dependencyPath = + getPathForDependency(jobDescription.getApplicationName(), + jobDescription.getApplicationVersion()); + log.info("looking for " + dependencyPath); + + if (!fileSystem.exists(dependencyPath)) { + throw new DisworkException("job require a dependency " + + jobDescription.getApplicationName() + "-" + + jobDescription.getApplicationVersion() + " that is " + + "not available"); + } + + String jobDir = getJobPath(jobDescription); + + if(!fileSystem.exists(jobDir)) { + // strange ! + } + + fileSystem.write(jobDir + "/" + LOG_PATH, IOUtils.toInputStream("")); + + InputStream jobJSDL = IOUtils.toInputStream(jobDescription.toJSDL()); + fileSystem.write(jobDir + "/" + JSDL_PATH, jobJSDL); + + // file staging + for (String fileName : inputFiles.keySet()) { + fileSystem.write(jobDir + "/" + fileName, inputFiles.get(fileName)); + } + + // propose job + String linkName = newJobLinkName(); + + // FIXME 20100609 bleny may throws exception if jobs are proposed + // at a same time + fileSystem.createSymbolicLink(TODO + "/" + linkName, jobDir); + + } catch (IOException e) { + log.error("file system error", e); + throw new DisworkFileSystemException("file system error", e); + } + } + + public boolean checkLogContains(JobDescription job, + String pattern) throws DisworkException { + try { + String jobPath = getJobPath(job); + List<?> entries = IOUtils.readLines(fileSystem.read(jobPath + "/" + LOG_PATH)); + return entries.contains(pattern); + } catch (FileNotFoundException e) { + log.info("log file was not found in job " + job, e); + throw new DisworkException("log file was not found in job " + job, e); + } catch (IOException e) { + log.info("file system error ", e); + throw new DisworkException("file system error ", e); + } + } + + public boolean isFinished(JobDescription job) throws DisworkException { + return checkLogContains(job, "FINISHED"); + } + + public boolean isSuccessful(JobDescription job) throws DisworkException { + return checkLogContains(job, "DONE"); + } + + public boolean isFailed(JobDescription job) throws DisworkException { + return isFinished(job) && !isSuccessful(job); + } + + @Override + public void close() throws IOException { + workers.stop(); + fileSystem.close(); + } + + public Map<String, InputStream> getResults(JobDescription job) + throws DisworkException { + Map<String, InputStream> results = new HashMap<String, InputStream>(); + for (String fileName : job.getStagingOutput()) { + String jobPath = getJobPath(job); + try { + InputStream result = fileSystem.read(jobPath + "/" + fileName); + results.put(fileName, result); + } catch (FileNotFoundException e) { + throw new DisworkException("an expected file is missing", e); + } catch (IOException e) { + log.info("file system error ", e); + throw new DisworkException("file system error ", e); + } + } + return results; + } +} \ No newline at end of file Added: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemonRunner.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemonRunner.java (rev 0) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemonRunner.java 2010-06-11 15:52:09 UTC (rev 73) @@ -0,0 +1,48 @@ +/* + * #%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 org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +/** + * + * @author bleny + */ +public class DisworkDaemonRunner { + + private static final Log log = LogFactory.getLog(DisworkDaemonRunner.class); + + /** + * @param args + */ + public static void main(String[] args) { + + // consider args + + // DisworkDaemon node = new DisworkDaemon(config); + + } +} Added: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkException.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkException.java (rev 0) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkException.java 2010-06-11 15:52:09 UTC (rev 73) @@ -0,0 +1,47 @@ +/* + * #%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; + +/** + * + * @author bleny + */ +public class DisworkException extends Exception { + + private static final long serialVersionUID = -6434751198109021511L; + + public DisworkException(String message, Throwable cause) { + super(message, cause); + } + + public DisworkException(String message) { + super(message); + } + + public DisworkException(Throwable cause) { + super(cause); + } + +} Added: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkFileSystemException.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkFileSystemException.java (rev 0) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkFileSystemException.java 2010-06-11 15:52:09 UTC (rev 73) @@ -0,0 +1,47 @@ +/* + * #%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; + +/** + * + * @author bleny + */ +public class DisworkFileSystemException extends DisworkException { + + private static final long serialVersionUID = -4027003687525235092L; + + public DisworkFileSystemException(String message, Throwable cause) { + super(message, cause); + } + + public DisworkFileSystemException(String message) { + super(message); + } + + public DisworkFileSystemException(Throwable cause) { + super(cause); + } + +} Added: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/JobDescription.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/JobDescription.java (rev 0) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/JobDescription.java 2010-06-11 15:52:09 UTC (rev 73) @@ -0,0 +1,185 @@ +/* + * #%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.Serializable; +import java.net.URL; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * + * + * + * @author bleny + */ +public class JobDescription implements Serializable { + + private static final long serialVersionUID = -8493700934802808925L; + + /** an id for diswork */ + protected String jobId; + + protected String jobName; + + protected String applicationName; + protected String applicationVersion; + + protected String commandLine; + + /** all files expected at the beginning of the job */ + protected List<String> stagingInput = new ArrayList<String>(); + + /** all files expected at the end of the job */ + protected List<String> stagingOutput = new ArrayList<String>(); + + /** the name of a file and the URI where to get it */ + protected Map<String, URL> stagingInputUrls = new HashMap<String, URL>(); + + /** file where to read the standard input, may be null */ + protected String standardInput; + + /** file where to write the standard output */ + protected String standardOutput; + + /** TOKENS are piece of string you can use for writing command lines */ + protected static Map<String, String> TOKENS; + + static { + TOKENS = new HashMap<String, String>(); + + TOKENS.put("%java", System.getProperty("java.home") + "/bin/java"); + } + + /** + * 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 + * @param jobId + */ + protected JobDescription(String jobId) { + this.jobId = jobId; + } + + public String getJobId() { + return jobId; + } + + public String getCommandLine() { + String result = commandLine; + for (String token : TOKENS.keySet()) { + result = result.replace(token, TOKENS.get(token)); + } + return result; + } + + public void setCommandLine(String commandLine) { + this.commandLine = commandLine; + } + + public String getJobName() { + return jobName; + } + + public String getApplicationName() { + return applicationName; + } + + public String getApplicationVersion() { + return applicationVersion; + } + + public void setJobName(String jobName) { + this.jobName = jobName; + } + + public void setApplication(String applicationName, + String applicationVersion) { + this.applicationName = applicationName; + this.applicationVersion = applicationVersion; + } + + @Override + public String toString() { + 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 static JobDescription parseJSDL(String jsdl) { + return map.get(jsdl); + } + + public void addStagingInput(String fileName, URL source) { + stagingInput.add(fileName); + stagingInputUrls.put(fileName, source); + } + + public void addStagingInput(String fileName) { + stagingInput.add(fileName); + } + + public void addStagingOutput(String fileName) { + stagingOutput.add(fileName); + } + + public List<String> getStagingInput() { + return stagingInput; + } + + public List<String> getStagingOutput() { + return stagingOutput; + } + + public Map<String, URL> getStagingInputUrls() { + return stagingInputUrls; + } + + public String getStandardInput() { + return standardInput; + } + + public void setStandardInput(String fileName) { + standardInput = fileName; + } + + public String getStandardOutput() { + return standardOutput; + } + + public void setStandardOutput(String fileName) { + standardOutput = fileName; + } + +} \ No newline at end of file Added: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java (rev 0) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java 2010-06-11 15:52:09 UTC (rev 73) @@ -0,0 +1,278 @@ +/* + * #%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.File; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.FileOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.URL; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.commons.io.FilenameUtils; +import org.apache.commons.io.IOUtils; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.nuiton.diswork.fs.DisworkFileSystem; +import org.nuiton.util.ZipUtil; + +/** + * + * @author bleny + */ +public class WorkersManager { + + protected static Map<String, String> RUNNING_MOVE = new HashMap<String, String>(); + 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; + + static { + 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, DisworkDaemon.FAILED_1); + RUNNING_MOVE.put(DisworkDaemon.FAILED_1, DisworkDaemon.FAILED_2); + RUNNING_MOVE.put(DisworkDaemon.FAILED_2, DisworkDaemon.FAILED_3); + } + + private static final Log log = LogFactory.getLog(WorkersManager.class); + + protected DisworkFileSystem fileSystem; + protected DisworkConfig config; + + // Pool of workers ? + protected Worker worker; + + // Activity strategy ? + + protected class Worker implements Runnable { + + public boolean shouldStop = false; + + 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)); + } + + protected boolean runJob(String jobPath) throws IOException { + log.info("running job at " + jobPath); + + String jsdl; + try { + String jsdlPath = jobPath + "/" + DisworkDaemon.JSDL_PATH; + jsdl = IOUtils.toString(fileSystem.read(jsdlPath)); + } catch (FileNotFoundException e) { + log.warn("job " + jobPath + " misses a job description"); + return false; + } + + log.info("read jsdl " + jsdl); + + JobDescription jobDescription = JobDescription.parseJSDL(jsdl); + + log.info("will run job " + jobDescription); + + // create temp dir + File jobDir = new File(config.getTempDirectory(), jobDescription.getJobId()); + jobDir.mkdirs(); + + // download application + String applicationPath = DisworkDaemon.getPathForDependency( + jobDescription.getApplicationName(), + jobDescription.getApplicationVersion()); + InputStream applicationData = fileSystem.read(applicationPath); + + + File application = new File(jobDir, FilenameUtils.getName(applicationPath)); + application.createNewFile(); + log.info("will create " + application.getAbsolutePath()); + OutputStream out = new FileOutputStream(application); + IOUtils.copy(applicationData, out); + // unzip application + ZipUtil.uncompress(application, jobDir); + + // staging input files + for (String fileName : jobDescription.getStagingInput()) { + File localCopy = new File(jobDir, fileName); + localCopy.createNewFile(); + InputStream source = null; + if (jobDescription.getStagingInputUrls().containsKey(fileName)) { + // download this file from URL + URL url = jobDescription.getStagingInputUrls().get(fileName); + log.info("downloading from " + url); + source = url.openStream(); + } else { + // download this file from diswork + source = fileSystem.read(jobPath + "/" + fileName); + } + IOUtils.copy(source, new FileOutputStream(localCopy)); + } + + // executing the job + String commandLine = jobDescription.getCommandLine(); + log.info("calling " + commandLine); + String[] bidule = commandLine.split(" "); + ProcessBuilder builder = new ProcessBuilder(bidule); + builder.directory(jobDir); + builder.redirectErrorStream(true); + Process job = builder.start(); + + // plugin a file on the standard input + String standardInputFileName = jobDescription.getStandardInput(); + if (standardInputFileName != null) { + InputStream input = new FileInputStream(new File(jobDir, standardInputFileName)); + IOUtils.copy(input, job.getOutputStream()); + } + + int exitValue = -1; + try { + exitValue = job.waitFor(); + } catch (InterruptedException e) { + log.error("job " + jobDescription + " was interupted", e); + + // FIXME 20100611 bleny job is considered has failed + exitValue = 1; + } + + // dump the standard output in a file + String standardOutputFileName = jobDescription.getStandardOutput(); + if (standardOutputFileName != null) { + OutputStream output = new FileOutputStream(new File(jobDir, standardOutputFileName)); + IOUtils.copy(job.getInputStream(), output); + } + + log.info("job returned " + exitValue); + + // output file staging + for (String fileName : jobDescription.getStagingOutput()) { + File localCopy = new File(jobDir, fileName); + InputStream localCopyStream = new FileInputStream(localCopy); + + String filePath = jobPath + "/" + fileName; + + log.info("out-staging " + fileName); + // erase before write + if (fileSystem.exists(filePath)) { + fileSystem.delete(filePath); + } + + fileSystem.write(filePath, localCopyStream); + localCopyStream.close(); + } + + // clean up the job directory + // FileUtil.deleteRecursively(jobDir); + + boolean success = exitValue == 0; + if (success) { + log(jobPath, "DONE\nFINISHED"); + } else { + log(jobPath, "FAILED\nFINISHED"); + } + return success; + } + + @Override + public void run() { + while (! shouldStop) { + // try to find a new job + try { + // TODO 20100609 bleny watch for other jobs + List<String> jobsNames = + fileSystem.readDirectory(DisworkDaemon.TODO); + if (jobsNames.size() != 0) { + // sort and choose the first, due to names, it should be + // the more ancient one + Collections.sort(jobsNames); + String oldName = jobsNames.get(0); + + String newName = DisworkDaemon.newJobLinkName(); + fileSystem.move(DisworkDaemon.TODO + "/" + oldName, + DisworkDaemon.TODO_RUNNING + "/" + newName); + + boolean jobSuccess = runJob(DisworkDaemon.TODO_RUNNING + "/" + newName); + + oldName = newName; + newName = DisworkDaemon.newJobLinkName(); + + if (jobSuccess) { + fileSystem.move( + DisworkDaemon.TODO_RUNNING + "/" + oldName, + DisworkDaemon.DONE + "/" + newName); + } else { + fileSystem.move( + DisworkDaemon.TODO_RUNNING + "/" + oldName, + DisworkDaemon.FAILED_3 + "/" + newName); + } + } else { + log.info("nothing to do"); + Thread.sleep(JOB_WAIT); + } + + } catch (IOException e) { + log.error("error while reading jobs", e); + // TODO 20100611 bleny manage exception + } catch (InterruptedException e) { + log.info("exception catch", e); + // TODO 20100611 bleny manage exception + } + } + } + + } + + public WorkersManager(DisworkFileSystem fileSystem, DisworkConfig config) { + this.fileSystem = fileSystem; + this.config = config; + } + + public void start() { + worker = new Worker(); + Thread t = new Thread(worker); + t.start(); + } + + public void stop() { + worker.shouldStop = true; + } +} \ No newline at end of file Added: trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonTest.java =================================================================== --- trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonTest.java (rev 0) +++ trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonTest.java 2010-06-11 15:52:09 UTC (rev 73) @@ -0,0 +1,106 @@ +package org.nuiton.diswork.daemon; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.io.InputStream; +import java.net.URL; +import java.util.HashMap; +import java.util.Map; + +import org.apache.commons.io.IOUtils; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +public class DisworkDaemonTest { + + protected DisworkDaemon daemon; + + protected static int port = 39999; + + @Before + public void setUp() throws Exception { + DisworkConfig config = DisworkConfig.newConfig(); + port += 1; + config.setUsedPort(port); + daemon = new DisworkDaemon(config); + InputStream application = ClassLoader.getSystemResourceAsStream("fake-app-1.0.zip"); + daemon.submitApplication("fake-app", "1.0", application); + } + + @After + public void tearDown() throws Exception { + daemon.close(); + } + + @Test(expected = DisworkException.class) + public void testSubmitWithoutDependency() throws Exception { + JobDescription job = daemon.newJob(); + job.setApplication("non-existing-application", "0.0"); + daemon.submitJob(job); + } + + @Test + public void testSubmitSuccessfulJob() throws Exception { + JobDescription job = daemon.newJob(); + job.setApplication("fake-app", "1.0"); + job.setCommandLine("%java -jar fake-app.jar"); + daemon.submitJob(job); + + while(! daemon.isFinished(job)) { + Thread.sleep(5 * 1000); + } + + assertTrue(daemon.isSuccessful(job)); + } + + @Test + public void testSubmitFailJob() throws Exception { + JobDescription job = daemon.newJob(); + job.setApplication("fake-app", "1.0"); + job.setCommandLine("%java -jar fake-app.jar fail"); + daemon.submitJob(job); + + while(! daemon.isFinished(job)) { + Thread.sleep(5 * 1000); + } + + assertTrue(daemon.isFailed(job)); + } + + @Test + public void testStaging() throws Exception { + JobDescription job = daemon.newJob(); + job.setJobName("My Job"); + job.setApplication("fake-app", "1.0"); + job.setCommandLine("%java -jar fake-app.jar"); + + job.addStagingInput("example.com_index", new URL("http://www.example.com/")); + job.addStagingInput("input.txt"); + job.addStagingOutput("output.txt"); + + job.setStandardInput("input.txt"); + job.setStandardOutput("output.txt"); + + Map<String, InputStream> data = new HashMap<String, InputStream>(); + data.put("input.txt", ClassLoader.getSystemResourceAsStream("input.txt")); + + daemon.submitJob(job, data); + + while(! daemon.isFinished(job)) { + Thread.sleep(1 * 1000); + } + + assertTrue(daemon.isSuccessful(job)); + + Map<String, InputStream> results = daemon.getResults(job); + + assertTrue(results.containsKey("output.txt")); + + String output = IOUtils.toString(results.get("output.txt")); + assertEquals("a print on standard output\n", output); + + } + +} Added: trunk/diswork-daemon/src/test/resources/fake-app-1.0.zip =================================================================== (Binary files differ) Property changes on: trunk/diswork-daemon/src/test/resources/fake-app-1.0.zip ___________________________________________________________________ Added: svn:mime-type + application/octet-stream Added: trunk/diswork-daemon/src/test/resources/fake-app.jar =================================================================== (Binary files differ) Property changes on: trunk/diswork-daemon/src/test/resources/fake-app.jar ___________________________________________________________________ Added: svn:mime-type + application/octet-stream Added: trunk/diswork-daemon/src/test/resources/input.txt =================================================================== --- trunk/diswork-daemon/src/test/resources/input.txt (rev 0) +++ trunk/diswork-daemon/src/test/resources/input.txt 2010-06-11 15:52:09 UTC (rev 73) @@ -0,0 +1 @@ +Hello Added: trunk/diswork-daemon/src/test/resources/log4j.properties =================================================================== --- trunk/diswork-daemon/src/test/resources/log4j.properties (rev 0) +++ trunk/diswork-daemon/src/test/resources/log4j.properties 2010-06-11 15:52:09 UTC (rev 73) @@ -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 Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/Demo.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/Demo.java 2010-06-10 09:32:59 UTC (rev 72) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/Demo.java 2010-06-11 15:52:09 UTC (rev 73) @@ -111,6 +111,7 @@ List<String> todos = fileSystem.readDirectory("/todo"); if (todos.isEmpty()) { log.info("nothing to do"); + System.out.println("nothing to do"); } else { // taking a random job Random random = new Random(); 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-10 09:32:59 UTC (rev 72) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystem.java 2010-06-11 15:52:09 UTC (rev 73) @@ -680,7 +680,7 @@ */ protected void checkPathSyntax(String path) throws IOException { if (!path.startsWith(EntryUtil.ROOT_DIRECTORY)) { - throw new IOException("\" + path + \" is not correct, all pathes " + + throw new IOException("\"" + path + "\" is not correct, all pathes " + "have to be absolute (thus, starts with)" + EntryUtil.ROOT_DIRECTORY); } @@ -688,7 +688,7 @@ String doubleSeparator = EntryUtil.PATH_SEPARATOR + EntryUtil.PATH_SEPARATOR; if (path.contains(doubleSeparator)) { - throw new IOException("\" + path + \" is not correct, it contains " + throw new IOException("\"" + path + "\" is not correct, it contains " + doubleSeparator); } } @@ -809,4 +809,21 @@ } } + + public void createDirectories(String path) + throws ConcurrentModificationException, + IOException { + log.info("trying create directories for " + path); + + String pathWithoutRoot = path.substring(1, path.length()); + + String[] dirs = pathWithoutRoot.split(EntryUtil.PATH_SEPARATOR); + String dirPath = ""; + for (String dir : dirs) { + dirPath += EntryUtil.PATH_SEPARATOR + dir; + if (!exists(dirPath)) { + createDirectory(dirPath); + } + } + } } \ 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-10 09:32:59 UTC (rev 72) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java 2010-06-11 15:52:09 UTC (rev 73) @@ -81,8 +81,15 @@ * @return * @throws UnknownHostException */ - public static String getIp() throws UnknownHostException { - InetAddress result = InetAddress.getLocalHost(); + public static String getIp() { + InetAddress result = null; + try { + result = InetAddress.getLocalHost(); + } catch (UnknownHostException e) { + // LocalHost is always known + log.error(e); + } + if (result.isLoopbackAddress()) { try { Socket temp = new Socket("microsoft.com", 80); @@ -140,8 +147,7 @@ * @return * @throws UnknownHostException */ - public static DisworkFileSystemConfig newPastryDisworkConfig() - throws UnknownHostException { + public static DisworkFileSystemConfig newPastryDisworkConfig() { return newPastryDisworkConfig(null); } @@ -153,8 +159,8 @@ * @return a complete config * @throws UnknownHostException */ - public static DisworkFileSystemConfig newPastryDisworkConfig(Integer bootstrapPort) - throws UnknownHostException { + public static DisworkFileSystemConfig + newPastryDisworkConfig(Integer bootstrapPort) { DisworkFileSystemConfig result = new DisworkFileSystemConfig(); String port = getPort().toString(); String ip = getIp(); @@ -169,13 +175,12 @@ return result; } - public static DisworkFileSystemConfig newKademliaDisworkConfig() - throws UnknownHostException { + public static DisworkFileSystemConfig newKademliaDisworkConfig() { return newKademliaDisworkConfig(null); } - public static DisworkFileSystemConfig newKademliaDisworkConfig(Integer bootstrapPort) - throws UnknownHostException { + public static DisworkFileSystemConfig + newKademliaDisworkConfig (Integer bootstrapPort) { DisworkFileSystemConfig result = new DisworkFileSystemConfig(); String port = getPort().toString(); String ip = getIp(); Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/InMemoryDisworkMap.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/InMemoryDisworkMap.java 2010-06-10 09:32:59 UTC (rev 72) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/InMemoryDisworkMap.java 2010-06-11 15:52:09 UTC (rev 73) @@ -39,6 +39,8 @@ public class InMemoryDisworkMap extends HashMap<String, byte[]> implements DisworkMap { + private static final long serialVersionUID = 2467699175129909832L; + private final Log log = LogFactory.getLog(InMemoryDisworkMap.class); @Override 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-10 09:32:59 UTC (rev 72) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/KademliaDisworkMap.java 2010-06-11 15:52:09 UTC (rev 73) @@ -49,14 +49,15 @@ protected Kademlia kad; public KademliaDisworkMap(DisworkFileSystemConfig config) throws IOException { - log.info("booting kademlia dht on port " + config.getUsedPort()); + kad = new Kademlia(Identifier.randomIdentifier(), config.getUsedPort()); - + if (config.getBootstrapIp() != null) { InetSocketAddress bootstrap = new InetSocketAddress( config.getBootstrapIp(), config.getBootstrapPort()); try { + log.info("trying to connect to " + bootstrap); kad.connect(bootstrap); } catch (RoutingException e) { log.error("bootstrap node is unreachable", e); @@ -64,6 +65,8 @@ } } + log.info("kademlia status : " + kad); + //Identifier.IDSIZE = 1024; log.info("using " + Identifier.IDSIZE + " bytes for identifiers"); } @@ -93,7 +96,7 @@ } - log.info("key for string " + s + " is " + id); + log.debug("key for string " + s + " is " + id); return id; } Deleted: trunk/diswork-fs/src/main/resources/log4j.properties =================================================================== --- trunk/diswork-fs/src/main/resources/log4j.properties 2010-06-10 09:32:59 UTC (rev 72) +++ trunk/diswork-fs/src/main/resources/log4j.properties 2010-06-11 15:52:09 UTC (rev 73) @@ -1,9 +0,0 @@ -# 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=WARN -log4j.logger.org.nuiton.diswork.fs.Demo=INFO \ No newline at end of file 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-10 09:32:59 UTC (rev 72) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java 2010-06-11 15:52:09 UTC (rev 73) @@ -394,4 +394,10 @@ } */ } + + @Test + public void testCreateDirectories() throws Exception { + fileSystem.createDirectories("/dir/subdir/subsubdir"); + assertTrue(fileSystem.exists("/dir/subdir/subsubdir")); + } } 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-10 09:32:59 UTC (rev 72) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java 2010-06-11 15:52:09 UTC (rev 73) @@ -27,7 +27,7 @@ // finally, initiate the fileSystem DisworkFileSystemConfig disworkConfig1 = DisworkFileSystemConfig.newKademliaDisworkConfig(); - bootstrapPort = disworkConfig1.getUsedPort(); + bootstrapPort = disworkConfig1.getUsedPort() + 1; 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-10 09:32:59 UTC (rev 72) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/KademliaDisworkMapTest.java 2010-06-11 15:52:09 UTC (rev 73) @@ -3,7 +3,6 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; -import java.io.IOException; import java.util.ArrayList; import java.util.HashSet; import java.util.List; @@ -12,7 +11,6 @@ import org.junit.Before; import org.junit.Test; import org.nuiton.diswork.fs.DisworkFileSystemConfig; -import org.nuiton.diswork.fs.storage.KademliaDisworkMap; import org.planx.xmlstore.routing.Identifier; public class KademliaDisworkMapTest extends AbstractDisworkMapTest { @@ -24,7 +22,7 @@ map1 = new KademliaDisworkMap(config1); DisworkFileSystemConfig config2 = DisworkFileSystemConfig - .newKademliaDisworkConfig(config1.getUsedPort()); + .newKademliaDisworkConfig(config1.getUsedPort()+1); map2 = new KademliaDisworkMap(config2); } @@ -72,7 +70,7 @@ * a bad bootstrap * @throws Exception */ - @Test(expected = IOException.class) + @Test(expected = org.planx.xmlstore.routing.RoutingException.class) public void testBadBootrap() throws Exception { DisworkFileSystemConfig config1 = DisworkFileSystemConfig.newKademliaDisworkConfig(); Copied: trunk/diswork-fs/src/test/resources/log4j.properties (from rev 72, trunk/diswork-fs/src/main/resources/log4j.properties) =================================================================== --- trunk/diswork-fs/src/test/resources/log4j.properties (rev 0) +++ trunk/diswork-fs/src/test/resources/log4j.properties 2010-06-11 15:52:09 UTC (rev 73) @@ -0,0 +1,10 @@ +# 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=WARN +log4j.logger.org.nuiton.diswork.fs.storage.KademliaDisworkMap=INFO +log4j.logger.org.nuiton.diswork.fs.Demo=INFO \ No newline at end of file

1 0

r72 - trunk/diswork-fs/src/main/resources
by bleny＠users.nuiton.org June 10, 2010

June 10, 2010

Author: bleny Date: 2010-06-10 11:32:59 +0200 (Thu, 10 Jun 2010) New Revision: 72 Url: http://nuiton.org/repositories/revision/diswork/72 Log: fs : update log4j.prop Modified: trunk/diswork-fs/src/main/resources/log4j.properties Modified: trunk/diswork-fs/src/main/resources/log4j.properties =================================================================== --- trunk/diswork-fs/src/main/resources/log4j.properties 2010-06-10 09:30:17 UTC (rev 71) +++ trunk/diswork-fs/src/main/resources/log4j.properties 2010-06-10 09:32:59 UTC (rev 72) @@ -5,6 +5,5 @@ 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.disworkfs=WARN -#log4j.logger.org.nuiton.diswork.fs.storage.KademliaDisworkMap=INFO -log4j.logger.org.nuiton.disworkfs.Demo=INFO \ No newline at end of file +log4j.logger.org.nuiton.diswork.fs=WARN +log4j.logger.org.nuiton.diswork.fs.Demo=INFO \ No newline at end of file

1 0

r71 - in trunk/diswork-fs/src: main/java/org/nuiton/diswork/fs main/java/org/nuiton/diswork/fs/storage main/resources test/java/org/nuiton/diswork/fs test/java/org/nuiton/diswork/fs/storage
by bleny＠users.nuiton.org June 10, 2010

June 10, 2010

Author: bleny Date: 2010-06-10 11:30:17 +0200 (Thu, 10 Jun 2010) New Revision: 71 Url: http://nuiton.org/repositories/revision/diswork/71 Log: implementation move + test, gestion de l'echec du bootstrap dans kademlia + test, traces dans la demo Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/Demo.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystem.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/KademliaDisworkMap.java trunk/diswork-fs/src/main/resources/log4j.properties trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/KademliaDisworkMapTest.java Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/Demo.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/Demo.java 2010-06-09 08:40:08 UTC (rev 70) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/Demo.java 2010-06-10 09:30:17 UTC (rev 71) @@ -90,6 +90,7 @@ } } catch (IOException e) { log.error(e); + System.out.println("error occured " + e); } try { Thread.sleep(15 * 1000); @@ -142,11 +143,13 @@ } } catch (IOException e) { log.error(e); + System.out.println("error occured " + e); } try { Thread.sleep(10 * 1000); } catch (InterruptedException e) { log.info("sleep interrupted", e); + System.out.println("error occured " + e); } } } 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-09 08:40:08 UTC (rev 70) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystem.java 2010-06-10 09:30:17 UTC (rev 71) @@ -692,4 +692,121 @@ + doubleSeparator); } } + + /** + * move a file, a directory a link. May be used for renaming purpose + * @param path the path to the object to move + * @param destination the full path of the target + * @throws IOException + */ + public void move(String path, String destination) throws IOException { + checkPathSyntax(path); + checkPathSyntax(destination); + + String pathParent = EntryUtil.getParentFromPath(path); + String pathName = EntryUtil.getNameFromPath(path); + String destinationParent = EntryUtil.getParentFromPath(destination); + String destinationName = EntryUtil.getNameFromPath(destination); + + move(pathParent, pathName, destinationParent, destinationName); + + } + + protected void move(String pathParent, String pathName, + String destinationParent, String destinationName) throws IOException { + + String entryParent = walk(pathParent); + + if (entryParent == null) { + throw new IOException(pathParent + " directory doesn't exists"); + } + + if (EntryUtil.isDirectory(entryParent)) { + + String destinationParentEntry = walk(destinationParent); + if (destinationParentEntry == null) { + throw new IOException(destinationParentEntry + " directory" + + "doesn't exists"); + } + + if (EntryUtil.isDirectory(destinationParentEntry)) { + + // we are now in actual directories for old path and destination + + String parentContent = storage.getDirectory( + EntryUtil.getIdFromEntry(entryParent)); + String oldEntry = EntryUtil.findEntryInDirectory(parentContent, + pathName); + if (oldEntry == null) { + throw new IOException("no element " + pathName + " in " + + pathParent); + } + + String parentId = EntryUtil.getIdFromEntry(entryParent); + String destinationParentId = EntryUtil.getIdFromEntry(destinationParentEntry); + // update directory content + int numberOfTry = 0; + boolean lockAcquired = false; + 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); + String newContent = EntryUtil.removeEntryFromEntries(parentContent, pathName); + storage.putDirectory(parentId, newContent); + + + String destinationParentContent = storage.getDirectory(destinationParentId); + newContent = EntryUtil.addEntryToDirectoryContent(destinationParentContent, + EntryUtil.getTypeFromEntry(oldEntry), + destinationName, + EntryUtil.getIdFromEntry(oldEntry)); + storage.putDirectory(destinationParentId, newContent); + + storage.unLock(parentId); + storage.unLock(destinationParentId); + } else { + log.info(pathParent + " is locked"); + try { + Thread.sleep(LOCK_WAIT); + } catch (InterruptedException e) { + log.info("wait for lock interrupted", e); + throw new IOException + ("interrupted while trying to acquire lock", e); + } + } + } + + if (!lockAcquired) { + // fail, parent dir have not been written + throw new ConcurrentModificationException + ("can't write " + pathParent + " directory is locked"); + } + + } else if (EntryUtil.isLink(destinationParentEntry)) { + String linkTarget = storage.getLink( + EntryUtil.getIdFromEntry(destinationParentEntry)); + String newTarget = EntryUtil.resolveLink(destinationParent, linkTarget); + move(newTarget, pathName, destinationParent, destinationName); + } else if (EntryUtil.isFile(destinationParentEntry)) { + throw new IOException(destinationParent + " is not a directory"); + } else { + log.warn("strange entry" + destinationParentEntry); + throw new IOException("strange entry" + destinationParentEntry); + } + + } else if (EntryUtil.isLink(entryParent)) { + String linkTarget = storage.getLink( + EntryUtil.getIdFromEntry(entryParent)); + String newTarget = EntryUtil.resolveLink(pathParent, linkTarget); + move(pathParent, pathName, newTarget, destinationName); + } else if (EntryUtil.isFile(entryParent)) { + throw new IOException(pathParent + " is not a directory"); + } else { + log.warn("strange entry" + entryParent); + throw new IOException("strange entry" + entryParent); + } + + } } \ No newline at end of file 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-09 08:40:08 UTC (rev 70) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/KademliaDisworkMap.java 2010-06-10 09:30:17 UTC (rev 71) @@ -40,6 +40,7 @@ import org.nuiton.diswork.fs.DisworkFileSystemConfig; import org.planx.xmlstore.routing.Identifier; import org.planx.xmlstore.routing.Kademlia; +import org.planx.xmlstore.routing.RoutingException; public class KademliaDisworkMap implements DisworkMap { @@ -48,13 +49,19 @@ protected Kademlia kad; public KademliaDisworkMap(DisworkFileSystemConfig config) throws IOException { + log.info("booting kademlia dht on port " + config.getUsedPort()); kad = new Kademlia(Identifier.randomIdentifier(), config.getUsedPort()); if (config.getBootstrapIp() != null) { InetSocketAddress bootstrap = new InetSocketAddress( config.getBootstrapIp(), config.getBootstrapPort()); - kad.connect(bootstrap); + try { + kad.connect(bootstrap); + } catch (RoutingException e) { + log.error("bootstrap node is unreachable", e); + throw e; + } } //Identifier.IDSIZE = 1024; Modified: trunk/diswork-fs/src/main/resources/log4j.properties =================================================================== --- trunk/diswork-fs/src/main/resources/log4j.properties 2010-06-09 08:40:08 UTC (rev 70) +++ trunk/diswork-fs/src/main/resources/log4j.properties 2010-06-10 09:30:17 UTC (rev 71) @@ -6,4 +6,5 @@ log4j.appender.stdout.layout.ConversionPattern=%d %5p [%t] (%F:%L) %M - %m%n # package level log4j.logger.org.nuiton.disworkfs=WARN +#log4j.logger.org.nuiton.diswork.fs.storage.KademliaDisworkMap=INFO log4j.logger.org.nuiton.disworkfs.Demo=INFO \ No newline at end of file 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-09 08:40:08 UTC (rev 70) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java 2010-06-10 09:30:17 UTC (rev 71) @@ -327,7 +327,7 @@ } @Test - public void testLinks() throws Exception { + public void testLinks() throws Exception { fileSystem.createDirectory("/dir"); fileSystem.createDirectory("/dir/subdir"); fileSystem.createDirectory("/otherdir"); @@ -358,4 +358,40 @@ assertEquals(1, readResult.size()); assertTrue(readResult.contains("subdir")); } + + @Test + public void testMove() throws Exception { + fileSystem.createDirectory("/dir"); + fileSystem.createDirectory("/dir/subdir"); + fileSystem.createDirectory("/dir/subdir/subsubdir"); + fileSystem.createSymbolicLink("/dir/link", "subdir"); + fileSystem.write("/dir/file", new FileInputStream(randomFilePath)); + + fileSystem.createDirectory("/otherdir"); + fileSystem.move("/dir/subdir", "/otherdir/subdir"); + fileSystem.move("/dir/link", "/otherdir/link"); + fileSystem.move("/dir/file", "/otherdir/newfile"); + + List<String> readDir = fileSystem.readDirectory("/dir"); + for (String c : readDir) { + System.out.println(c); + } + assertTrue(readDir.isEmpty()); + + readDir = fileSystem.readDirectory("/otherdir"); + assertEquals(3, readDir.size()); + + assertTrue(fileSystem.exists("/otherdir/newfile")); + assertTrue(fileSystem.exists("/otherdir/subdir/subsubdir")); + + // FIXME 20100610 bleny after moving a link, it still points the old target + + //assertTrue(fileSystem.exists("/otherdir/link/subsubdir")); + /* + readDir = fileSystem.readDirectory("/otherdir/link"); + for (String e : readDir) { + System.out.println(e); + } + */ + } } 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-09 08:40:08 UTC (rev 70) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/KademliaDisworkMapTest.java 2010-06-10 09:30:17 UTC (rev 71) @@ -3,6 +3,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import java.io.IOException; import java.util.ArrayList; import java.util.HashSet; import java.util.List; @@ -18,10 +19,12 @@ @Before public void setUp() throws Exception { - DisworkFileSystemConfig config1 = DisworkFileSystemConfig.newKademliaDisworkConfig(); + DisworkFileSystemConfig config1 = + DisworkFileSystemConfig.newKademliaDisworkConfig(); map1 = new KademliaDisworkMap(config1); - DisworkFileSystemConfig config2 = DisworkFileSystemConfig.newKademliaDisworkConfig(config1.getUsedPort()); + DisworkFileSystemConfig config2 = DisworkFileSystemConfig + .newKademliaDisworkConfig(config1.getUsedPort()); map2 = new KademliaDisworkMap(config2); } @@ -63,4 +66,18 @@ // if keys set is too small, there are collisions assertEquals(paths.size(), keys.size()); } + + /** + * Tests that in exception is raised when attempting to connect using + * a bad bootstrap + * @throws Exception + */ + @Test(expected = IOException.class) + public void testBadBootrap() throws Exception { + DisworkFileSystemConfig config1 = + DisworkFileSystemConfig.newKademliaDisworkConfig(); + config1.setBootstrapIp("microsoft.com"); + config1.setBootstrapPort(80); + new KademliaDisworkMap(config1); + } }

1 0

r70 - in trunk/diswork-fs/src: main/java/org/nuiton/diswork/fs main/java/org/nuiton/diswork/fs/storage test/java/org/nuiton/diswork/fs
by bleny＠users.nuiton.org June 9, 2010

June 9, 2010

Author: bleny Date: 2010-06-09 10:40:08 +0200 (Wed, 09 Jun 2010) New Revision: 70 Url: http://nuiton.org/repositories/revision/diswork/70 Log: demo avec traces ; style Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/Demo.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 Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/Demo.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/Demo.java 2010-06-08 14:35:06 UTC (rev 69) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/Demo.java 2010-06-09 08:40:08 UTC (rev 70) @@ -63,6 +63,7 @@ InputStream source = IOUtils.toInputStream(job); log.info("writing job " + waitingForJob + " (" + randomInteger + ")"); + System.out.println("submited job " + randomInteger); fileSystem.write(waitingForJob, source); source.close(); } else { @@ -76,6 +77,8 @@ Integer result = Integer.parseInt(resultString); log.info("result is " + result + " (expected : " + expectedResult + ")"); + System.out.println("job result published " + + result); fileSystem.delete(resultPath); fileSystem.delete(waitingForJob); waitingForJob = null; @@ -118,11 +121,15 @@ log.info("reading the job " + jobPath); InputStream in = fileSystem.read(jobPath); String operation = IOUtils.toString(in); - // in.close(); + in.close(); log.info("operation to do " + operation); + System.out.println("job found " + + operation); Integer i = Integer.parseInt(operation); i += 1; String result = i.toString(); + System.out.println("published result " + + result); log.info("result is " + result); InputStream source = IOUtils.toInputStream(result); 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-08 14:35:06 UTC (rev 69) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/Storage.java 2010-06-09 08:40:08 UTC (rev 70) @@ -449,7 +449,7 @@ } } - private boolean isLockStillOwned(String key) { + protected boolean isLockStillOwned(String key) { String lockKey = keyToLockKey(key); byte[] lock = map.get(lockKey); String lockOwner = EntryUtil.getOwnerFromLock(lock); 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-08 14:35:06 UTC (rev 69) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java 2010-06-09 08:40:08 UTC (rev 70) @@ -25,7 +25,7 @@ import org.nuiton.util.FileUtil; -public abstract class AbstractDisworkFileSystemTest { +public abstract class AbstractDisworkFileSystemTest { /** * a place to store files for the test it's a subdirectory of the OS temp

1 0

r69 - in trunk/diswork-fs: . src/main/java/org/nuiton/diswork/fs src/main/java/org/nuiton/diswork/fs/storage
by bleny＠users.nuiton.org June 8, 2010

June 8, 2010

Author: bleny Date: 2010-06-08 16:35:06 +0200 (Tue, 08 Jun 2010) New Revision: 69 Url: http://nuiton.org/repositories/revision/diswork/69 Log: fs : syntax check on paths Modified: trunk/diswork-fs/run_demo.sh trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystem.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/Storage.java Modified: trunk/diswork-fs/run_demo.sh =================================================================== --- trunk/diswork-fs/run_demo.sh 2010-06-08 13:16:35 UTC (rev 68) +++ trunk/diswork-fs/run_demo.sh 2010-06-08 14:35:06 UTC (rev 69) @@ -1,2 +1,2 @@ -mvn -e exec:java -Dexec.mainClass=org.nuiton.disworkfs.Demo -Dexec.args="consumer 9001" -Dexec.classpathScope=test -mvn -e exec:java -Dexec.mainClass=org.nuiton.disworkfs.Demo -Dexec.args="producer 9002 127.0.0.1 9001" -Dexec.classpathScope=test +mvn -e exec:java -Dexec.mainClass=org.nuiton.diswork.fs.Demo -Dexec.args="consumer 9001" -Dexec.classpathScope=test +mvn -e exec:java -Dexec.mainClass=org.nuiton.diswork.fs.Demo -Dexec.args="producer 9002 127.0.0.1 9001" -Dexec.classpathScope=test 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-08 13:16:35 UTC (rev 68) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystem.java 2010-06-08 14:35:06 UTC (rev 69) @@ -89,6 +89,7 @@ * @throws IOException */ public boolean exists(String path) throws IOException { + checkPathSyntax(path); String entry = walk(path); boolean result = entry != null; return result; @@ -104,6 +105,7 @@ */ public InputStream read(String path) throws FileNotFoundException, IOException { + checkPathSyntax(path); String entry = walk(path); if (entry == null) { @@ -141,8 +143,9 @@ public void write(String path, InputStream source) throws IOException, ConcurrentModificationException { + checkPathSyntax(path); if (source == null) { - throw new IOException("source is not readable (null stream)"); + throw new NullPointerException("source stream is null"); } String parent = EntryUtil.getParentFromPath(path); String name = EntryUtil.getNameFromPath(path); @@ -224,6 +227,7 @@ public void createDirectory(String path) throws IOException, ConcurrentModificationException { + checkPathSyntax(path); String parent = EntryUtil.getParentFromPath(path); String dirName = EntryUtil.getNameFromPath(path); createDirectory(parent, dirName); @@ -309,6 +313,7 @@ public void createSymbolicLink(String path, String target) throws IOException, ConcurrentModificationException { + checkPathSyntax(path); String parent = EntryUtil.getParentFromPath(path); String name = EntryUtil.getNameFromPath(path); createSymbolicLink(parent, name, target); @@ -411,6 +416,7 @@ */ public void delete(String path) throws IOException, ConcurrentModificationException { + checkPathSyntax(path); String parent = EntryUtil.getParentFromPath(path); String name = EntryUtil.getNameFromPath(path); log.info("trying to remove " + path); @@ -510,6 +516,7 @@ * @throws IOException if path doesn't point to a directory */ public List<String> readDirectory(String path) throws IOException { + checkPathSyntax(path); String entry = walk(path); List<String> result = null; @@ -660,11 +667,29 @@ @Override public void close() throws IOException { - storage.close(); + storage.close(); } protected void setMap(DisworkMap map) { storage.setMap(map); } -} + /** + * check a path is absolute and syntactically correct, throw exception if + * that's not the case. + */ + protected void checkPathSyntax(String path) throws IOException { + if (!path.startsWith(EntryUtil.ROOT_DIRECTORY)) { + throw new IOException("\" + path + \" is not correct, all pathes " + + "have to be absolute (thus, starts with)" + + EntryUtil.ROOT_DIRECTORY); + } + + String doubleSeparator = EntryUtil.PATH_SEPARATOR + + EntryUtil.PATH_SEPARATOR; + if (path.contains(doubleSeparator)) { + throw new IOException("\" + path + \" is not correct, it contains " + + doubleSeparator); + } + } +} \ No newline at end of file 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-08 13:16:35 UTC (rev 68) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/Storage.java 2010-06-08 14:35:06 UTC (rev 69) @@ -51,12 +51,16 @@ protected static final long LOCK_VALID_TIME = 60 * 60 * 1000; + /** this id will be used to sign locks */ protected String ownerId = EntryUtil.generateId(); + /** a key where the lock for key should be put at the returned value */ protected static String keyToLockKey(String key) { return key + "_lock"; } - + + /** when using copy-on-write new data for the value of key should be stored + * to the returned value */ protected static String keyToNewDataKey(String key) { return key + "_new_data"; }

1 0

r68 - in trunk/diswork-fs/src: main/java/org/nuiton main/java/org/nuiton/diswork main/java/org/nuiton/diswork/fs main/java/org/nuiton/diswork/fs/peerunit main/java/org/nuiton/diswork/fs/storage test/java/org/nuiton test/java/org/nuiton/diswork test/java/org/nuiton/diswork/fs test/java/org/nuiton/diswork/fs/storage
by bleny＠users.nuiton.org June 8, 2010

June 8, 2010

Author: bleny Date: 2010-06-08 15:16:35 +0200 (Tue, 08 Jun 2010) New Revision: 68 Url: http://nuiton.org/repositories/revision/diswork/68 Log: changement du package de org.nuiton.disworkfs en org.nuiton.diswork.fs Added: trunk/diswork-fs/src/main/java/org/nuiton/diswork/ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/Demo.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystem.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/package-info.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/peerunit/ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/ trunk/diswork-fs/src/test/java/org/nuiton/diswork/ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemInMemoryTest.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemPastryTest.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/ Removed: trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/ trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/ Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/peerunit/DisworkFileSystemTest.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/peerunit/package-info.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/DisworkMap.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/EntryUtil.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/InMemoryDisworkMap.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/KademliaDisworkMap.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/PastryDisworkMap.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/Storage.java trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/package-info.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/EntryUtilTest.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/InMemoryDisworkMapTest.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/KademliaDisworkMapTest.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/PastryDisworkMapTest.java Copied: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/Demo.java (from rev 66, trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/Demo.java) =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/Demo.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/Demo.java 2010-06-08 13:16:35 UTC (rev 68) @@ -0,0 +1,214 @@ +/* + * #%L + * disworkfs + * + * $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.fs; + +import java.io.IOException; +import java.io.InputStream; +import java.util.List; +import java.util.Random; + +import org.apache.commons.io.IOUtils; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +public class Demo { + + private static final Log log = LogFactory.getLog(Demo.class); + + protected static final String USAGE = + "Usage :\n" + + "org.nuiton.disworkfs.Demo producer|consumer" + + " usePort [bootStrapIp bootStrapPort]"; + + protected static DisworkFileSystem fileSystem; + + protected static class Producer implements Runnable { + + protected String waitingForJob = null; + protected Integer expectedResult = null; + + @Override + public void run() { + while (true) { + try { + if (fileSystem.exists("/todo")) { + if (waitingForJob == null) { + Random random = new Random(); + waitingForJob = "/todo/job-" + random.nextInt(); + Integer randomInteger = random.nextInt(); + expectedResult = randomInteger + 1; + String job = randomInteger.toString(); + InputStream source = IOUtils.toInputStream(job); + log.info("writing job " + waitingForJob + " (" + + randomInteger + ")"); + fileSystem.write(waitingForJob, source); + source.close(); + } else { + String resultPath = waitingForJob + ".result"; + if (fileSystem.exists(resultPath)) { + log.info("results has been published"); + InputStream resultStream = + fileSystem.read(resultPath); + String resultString = + IOUtils.toString(resultStream); + Integer result = Integer.parseInt(resultString); + log.info("result is " + result + " (expected : " + + expectedResult + ")"); + fileSystem.delete(resultPath); + fileSystem.delete(waitingForJob); + waitingForJob = null; + expectedResult = null; + } + } + } else { + fileSystem.createDirectory("/todo"); + } + } catch (IOException e) { + log.error(e); + } + try { + Thread.sleep(15 * 1000); + } catch (InterruptedException e) { + log.info("sleep interrupted", e); + } + } + } + } + + protected static class Consumer implements Runnable { + + @Override + public void run() { + while (true) { + try { + if (fileSystem.exists("/todo")) { + List<String> todos = fileSystem.readDirectory("/todo"); + if (todos.isEmpty()) { + log.info("nothing to do"); + } else { + // taking a random job + Random random = new Random(); + String todo = todos.get( + random.nextInt(todos.size()) + ); + if (!todo.endsWith(".result")) { + String jobPath = "/todo/" + todo; + log.info("reading the job " + jobPath); + InputStream in = fileSystem.read(jobPath); + String operation = IOUtils.toString(in); + // in.close(); + log.info("operation to do " + operation); + Integer i = Integer.parseInt(operation); + i += 1; + String result = i.toString(); + log.info("result is " + result); + InputStream source = + IOUtils.toInputStream(result); + fileSystem.write(jobPath + ".result", source); + source.close(); + } + } + } else { + fileSystem.createDirectory("/todo"); + } + } catch (IOException e) { + log.error(e); + } + try { + Thread.sleep(10 * 1000); + } catch (InterruptedException e) { + log.info("sleep interrupted", e); + } + } + } + } + + public static void printUsage() { + System.out.println(USAGE); + } + + /** + * + * <dl> + * <dt>producer|consumer (required)</dt> + * <dd>a producer will produce operations todo, a consumer may read + * operations and try to execute them + * </dd> + * <dt>port (required)</dt> + * <dd>the port to use</dd> + * <dt>bootstrapip (optional)</dt> + * <dd>if not provided, the node will bootstrap itself, creating a + * new network. If provided, the node will first try to join a + * network</dd> + * <dt>bootstrap port (required if bootstrapip is set)</dt> + * <dd>the port used by the node to join</dd> + * </dl> + * + * Example : + * org.nuiton.disworkfs.Demo consumer 9001 + * org.nuiton.disworkfs.Demo producer 9002 127.0.0.1 9001 + * org.nuiton.disworkfs.Demo consumer 9003 127.0.0.1 9002 + * org.nuiton.disworkfs.Demo producer 9004 127.0.0.1 9001 + * org.nuiton.disworkfs.Demo consumer 9005 127.0.0.1 9003 + * + * @param args + * @throws IOException + */ + public static void main(String[] args) throws IOException { + if (args.length == 2) { + DisworkFileSystemConfig config = + DisworkFileSystemConfig.newKademliaDisworkConfig(); + config.setOption("diswork.fs.use_port", args[1]); + fileSystem = new DisworkFileSystem(config); + if ("producer".equals(args[0])) { + Thread t = new Thread(new Producer()); + log.info("starting a producer"); + t.start(); + } else if ("consumer".equals(args[0])) { + Thread t = new Thread(new Consumer()); + log.info("starting a consumer"); + t.start(); + } + } else if (args.length == 4) { + DisworkFileSystemConfig config = + DisworkFileSystemConfig.newKademliaDisworkConfig(); + config.setOption("diswork.fs.use_port", args[1]); + config.setOption("diswork.fs.bootstrap.ip", args[2]); + config.setOption("diswork.fs.bootstrap.port", args[3]); + fileSystem = new DisworkFileSystem(config); + if ("producer".equals(args[0])) { + Thread t = new Thread(new Producer()); + log.info("starting a producer"); + t.start(); + } else if ("consumer".equals(args[0])) { + Thread t = new Thread(new Consumer()); + log.info("starting a consumer"); + t.start(); + } + } else { + printUsage(); + } + } +} \ No newline at end of file Copied: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystem.java (from rev 66, trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystem.java) =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystem.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystem.java 2010-06-08 13:16:35 UTC (rev 68) @@ -0,0 +1,670 @@ +/* + * #%L + * disworkfs + * + * $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.fs; + +import java.io.Closeable; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.util.ArrayList; +import java.util.ConcurrentModificationException; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.nuiton.diswork.fs.storage.DisworkMap; +import org.nuiton.diswork.fs.storage.EntryUtil; +import org.nuiton.diswork.fs.storage.Storage; + +/** Main class of Diswork File System, provide methods for all operations + * You can use: + * <ul> + * <li>{@link #createDirectory(String)} and {@link #readDirectory(String)} + * to create and browse directories</li> + * <li>{@link #write(String, InputStream)} and {@link #read(String)} + * to write a file and read it</li> + * <li>{@link #createSymbolicLink(String, String)} to create symbolic links</li> + * <li>{@link #exists(String)} and {@link #delete(String)} can be used + * on directories, files and symlinks</li> + * </ul> + */ +public class DisworkFileSystem implements Closeable { + + private static final Log log = LogFactory.getLog(DisworkFileSystem.class); + + /** storage will permit to save and read directories, files and links */ + protected Storage storage; + + /** number of try to acquire a lock before giving up */ + protected static final int LOCK_MAX_NUMBER_OF_TRY = 10; + + /** time (ms) to wait between two try to acquire a lock */ + protected static final int LOCK_WAIT = 10 * 1000; // ten seconds + + /** default constructor (uses default configuration parameters) + * @throws IOException caused by network issue + */ + public DisworkFileSystem() throws IOException { + this(new DisworkFileSystemConfig()); + } + + /** constructor allowing to provide another configuration to use + * + * @param config the configuration + * @throws IOException caused by network issue + */ + public DisworkFileSystem(DisworkFileSystemConfig config) + throws IOException { + storage = new Storage(config); + } + + /** tests the existence of a file/dir/link at a given path + * return true if something exists at path p, it will be true if a call + * to <code>mkdir(p)</code>, <code>write(p, ?)</code> or + * <code>ln(p, ?)</code> has been done before. + * @param path a path in the virtual FS + * @return true is something (a link, a file, or a directory) exists at path + * @throws IOException + */ + public boolean exists(String path) throws IOException { + String entry = walk(path); + boolean result = entry != null; + return result; + } + + /** + * use this method to read the content of a file. A call of read may be + * done after a call to {@link #write} + * @param path the path to the file to read + * @return an InputStream on the file + * @throws FileNotFoundException if no file exists at this path + * @throws IOException if path exists but is a directory + */ + public InputStream read(String path) throws FileNotFoundException, + IOException { + + String entry = walk(path); + if (entry == null) { + throw new FileNotFoundException(path); + } + + InputStream result = null; + + if (EntryUtil.isLink(entry)) { + log.info("reading link " + path); + String id = EntryUtil.getIdFromEntry(entry); + String link = storage.getLink(id); + String newTarget = EntryUtil.resolveLink(path, link); + result = read(newTarget); + } else if (EntryUtil.isDirectory(entry)) { + throw new IOException("target is not a file: " + path); + } else if (EntryUtil.isFile(entry)) { + log.info("reading file " + path); + String id = EntryUtil.getIdFromEntry(entry); + result = storage.getFile(id); + } + + log.info("read " + path + " returns " + result.available() + " bytes"); + + return result; + } + + /** + * write a file on the file system at a given place + * @param path + * @param source + * @throws IOException if file already exists + * @throws ConcurrentModificationException if file is already being written + */ + public void write(String path, InputStream source) + throws IOException, + ConcurrentModificationException { + if (source == null) { + throw new IOException("source is not readable (null stream)"); + } + String parent = EntryUtil.getParentFromPath(path); + String name = EntryUtil.getNameFromPath(path); + log.info("writing " + source.available() + " bytes at " + path); + write(parent, name, source); + } + + protected void write(String parent, String fileName, InputStream source) + throws IOException { + String entryParent = walk(parent); + + if (entryParent == null) { + throw new IOException(parent + " directory doesn't exists"); + } + + 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(); + storage.putFile(newFileId, source); + + // update directory content + int numberOfTry = 0; + boolean lockAcquired = false; + while (numberOfTry <= LOCK_MAX_NUMBER_OF_TRY && !lockAcquired) { + lockAcquired = storage.tryToLock(parentId); + if (lockAcquired) { + // parent dir is locked, do the update + content = storage.getDirectory(parentId); + String newContent = EntryUtil.addEntryToDirectoryContent( + content, EntryUtil.TYPE.F, fileName, newFileId); + storage.putDirectory(parentId, newContent); + storage.unLock(parentId); + } else { + log.info(parent + " is locked and can't be written"); + try { + Thread.sleep(LOCK_WAIT); + } catch (InterruptedException e) { + log.info("wait for lock interrupted", e); + throw new IOException + ("interrupted while trying to acquire lock", e); + } + } + } + + if (!lockAcquired) { + // fail, parent dir have not been written + throw new ConcurrentModificationException + ("can't write " + parent + " directory is locked"); + } + } else if (EntryUtil.isLink(entryParent)) { + String linkTarget = storage.getLink( + EntryUtil.getIdFromEntry(entryParent)); + String newTarget = EntryUtil.resolveLink(parent, linkTarget); + write(newTarget, fileName, source); + } else if (EntryUtil.isFile(entryParent)) { + throw new IOException(parent + " is not a directory"); + } else { + log.warn("strange entry" + entryParent); + throw new IOException("strange entry" + entryParent); + } + } + + /** + * creates a new empty directory + * @param path the absolute path (ending by the name) of the directory + * @throws IOException if parent path is not correct + */ + public void createDirectory(String path) + throws IOException, + ConcurrentModificationException { + String parent = EntryUtil.getParentFromPath(path); + String dirName = EntryUtil.getNameFromPath(path); + createDirectory(parent, dirName); + } + + protected void createDirectory(String parent, String dirName) + throws IOException { + + log.info("trying to create directory " + dirName + " in " + parent); + + String entryParent = walk(parent); + + if (entryParent == null) { + throw new IOException(parent + " directory doesn't exists"); + } + + if (EntryUtil.isDirectory(entryParent)) { + String parentId = EntryUtil.getIdFromEntry(entryParent); + String content = storage.getDirectory(parentId); + + // first, check if nothing already exists with this name + String findResult = EntryUtil.findEntryInDirectory + (content, dirName); + if (findResult != null) { + throw new IOException + (parent + " already contains an element named " + dirName); + } + + // store file before meta info + String newDirectoryId = EntryUtil.generateId(); + storage.putDirectory(newDirectoryId, + EntryUtil.EMPTY_DIRECTORY_CONTENT); + + // update directory content + int numberOfTry = 0; + boolean lockAcquired = false; + while (numberOfTry <= LOCK_MAX_NUMBER_OF_TRY && !lockAcquired) { + lockAcquired = storage.tryToLock(parentId); + if (lockAcquired) { + // we have locked, do the update + content = storage.getDirectory(parentId); + String newContent = EntryUtil.addEntryToDirectoryContent( + content, EntryUtil.TYPE.D, dirName, newDirectoryId); + storage.putDirectory(parentId, newContent); + storage.unLock(parentId); + } else { + log.info(parent + " is locked and can't be written"); + try { + Thread.sleep(LOCK_WAIT); + } catch (InterruptedException e) { + log.info("wait for lock interrupted", e); + throw new IOException + ("interrupted while trying to acquire lock", e); + } + } + } + + if (!lockAcquired) { + // fail, parent dir have not been written + throw new ConcurrentModificationException + ("can't write " + parent + " directory is locked"); + } + } else if (EntryUtil.isLink(entryParent)) { + String linkTarget = storage.getLink( + EntryUtil.getIdFromEntry(entryParent)); + String newTarget = EntryUtil.resolveLink(parent, linkTarget); + createDirectory(newTarget, dirName); + } else if (EntryUtil.isFile(entryParent)) { + throw new IOException(parent + " is not a directory"); + } else { + log.warn("strange entry" + entryParent); + throw new IOException("strange entry" + entryParent); + } + } + + /** + * create a new symbolic link + * @param path the path (ending by the name) where the link will be created + * @param target the path where the link point to + * (may be relative or absolute) + * @throws IOException if parent path is not correct + */ + public void createSymbolicLink(String path, String target) + throws IOException, + ConcurrentModificationException { + String parent = EntryUtil.getParentFromPath(path); + String name = EntryUtil.getNameFromPath(path); + createSymbolicLink(parent, name, target); + } + + /** + * @see #createSymbolicLink(String, String) + */ + protected void createSymbolicLink(String parent, String name, String target) + throws IOException { + + if (target.startsWith(EntryUtil.PATH_SEPARATOR)) { + // target is absolute + } else { + // target is relative, taking this in consideration + target = EntryUtil.resolveLink(parent, target); + } + + log.info("trying to create symbolic link named " + name + " at path " + + parent + " with target \"" + target + "\""); + + if (exists(target)) { + + String entryParent = walk(parent); + + if (entryParent == null) { + throw new IOException(parent + " directory doesn't exists"); + } + + if (EntryUtil.isDirectory(entryParent)) { + + String parentId = EntryUtil.getIdFromEntry(entryParent); + String content = storage.getDirectory(parentId); + + // first, check if nothing already exists with this name + String findResult = EntryUtil.findEntryInDirectory + (content, name); + if (findResult != null) { + throw new IOException + (parent + " already contains an element named " + name); + } + + // store file before meta info + String newLinkId = EntryUtil.generateId(); + storage.putLink(newLinkId, target); + + // update directory content + int numberOfTry = 0; + boolean lockAcquired = false; + while (numberOfTry <= LOCK_MAX_NUMBER_OF_TRY && !lockAcquired) { + lockAcquired = storage.tryToLock(parentId); + if (lockAcquired) { + // we have locked, do the update + content = storage.getDirectory(parentId); + String newContent = EntryUtil.addEntryToDirectoryContent( + content, EntryUtil.TYPE.L, name, newLinkId); + storage.putDirectory(parentId, newContent); + storage.unLock(parentId); + } else { + log.info(parent + " is locked and can't be written"); + try { + Thread.sleep(LOCK_WAIT); + } catch (InterruptedException e) { + log.info("wait for lock interrupted", e); + throw new IOException + ("interrupted while trying to acquire lock", e); + } + } + } + + if (!lockAcquired) { + // fail, parent dir have not been written + throw new ConcurrentModificationException + ("can't write " + parent + " directory is locked"); + } + + } else if (EntryUtil.isLink(entryParent)) { + String linkTarget = storage.getLink( + EntryUtil.getIdFromEntry(entryParent)); + String newTarget = EntryUtil.resolveLink(parent, linkTarget); + createSymbolicLink(newTarget, name, target); + } else if (EntryUtil.isFile(entryParent)) { + throw new IOException(parent + " is not a directory"); + } else { + log.warn("strange entry" + entryParent); + throw new IOException("strange entry" + entryParent); + } + + + } else { + throw new IOException(target + " is not a valid target"); + } + } + + /** + * remove a file, directory, or link. Non-empty directories can't be + * 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 { + String parent = EntryUtil.getParentFromPath(path); + String name = EntryUtil.getNameFromPath(path); + log.info("trying to remove " + path); + delete(parent, name); + } + + /** + * @see #delete(String) + */ + protected void delete(String parent, String name) throws IOException { + String entryParent = walk(parent); + + if (entryParent == null) { + throw new IOException(parent + " directory doesn't exists"); + } + + if (EntryUtil.isDirectory(entryParent)) { + String parentId = EntryUtil.getIdFromEntry(entryParent); + String content = storage.getDirectory(parentId); + + String entry = EntryUtil.findEntryInDirectory(content, name); + String idToRemove = EntryUtil.getIdFromEntry(entry); + + // according to the type of the entry name, delete it + if (EntryUtil.isDirectory(entry)) { + // check if not removing a non-empty directory + String innerDirectoryId = EntryUtil.getIdFromEntry(entry); + String innerDirectoryContent = + storage.getDirectory(innerDirectoryId); + // checking the emptiness of the directory + if (!innerDirectoryContent.equals( + EntryUtil.EMPTY_DIRECTORY_CONTENT)) { + // directory is not empty + throw new IOException + ("trying to remove a non-empty directory"); + } + + // remove it + storage.removeDirectory(idToRemove); + + } else if (EntryUtil.isFile(entry)) { + storage.removeFile(idToRemove); + } else if (EntryUtil.isLink(entry)) { + storage.removeLink(idToRemove); + } else { + log.warn("strange entry" + entryParent); + throw new IOException("strange entry" + entryParent); + } + + + // update directory content + int numberOfTry = 0; + boolean lockAcquired = false; + while (numberOfTry <= LOCK_MAX_NUMBER_OF_TRY && !lockAcquired) { + lockAcquired = storage.tryToLock(parentId); + if (lockAcquired) { + // we have locked, do the update + content = storage.getDirectory(parentId); + String newContent = EntryUtil.removeEntryFromEntries(content, name); + storage.putDirectory(parentId, newContent); + storage.unLock(parentId); + } else { + log.info(parent + " is locked and can't be written"); + try { + Thread.sleep(LOCK_WAIT); + } catch (InterruptedException e) { + log.info("wait for lock interrupted", e); + throw new IOException + ("interrupted while trying to acquire lock", e); + } + } + } + + if (!lockAcquired) { + // fail, parent dir have not been written + throw new ConcurrentModificationException + ("can't write " + parent + " directory is locked"); + } + + } else if (EntryUtil.isLink(entryParent)) { + String linkTarget = storage.getLink( + EntryUtil.getIdFromEntry(entryParent)); + String newTarget = EntryUtil.resolveLink(parent, linkTarget); + delete(newTarget, name); + } else if (EntryUtil.isFile(entryParent)) { + throw new IOException(parent + " is not a directory"); + } else { + log.warn("strange entry" + entryParent); + throw new IOException("strange entry" + entryParent); + } + } + + /** + * list the content of a directory + * @param path the complete path to the directory + * @return a list of the names of the elements in <code>path</code> + * @throws IOException if path doesn't point to a directory + */ + public List<String> readDirectory(String path) throws IOException { + String entry = walk(path); + List<String> result = null; + + if (entry == null) { + throw new IOException(path + " doesn't exists"); + } else { + // path may be a link, if it's the case, + // entry become the actual directory + if (EntryUtil.isLink(entry)) { + String target = storage.getLink( + EntryUtil.getIdFromEntry(entry)); + return readDirectory(target); + } else if (EntryUtil.isDirectory(entry)) { + result = new ArrayList<String>(); + String content = storage.getDirectory( + EntryUtil.getIdFromEntry(entry)); + if (EntryUtil.EMPTY_DIRECTORY_CONTENT.equals(content)) { + // directory is empty, add nothing + } else { + String[] entries = content.split( + EntryUtil.ENTRIES_SEPARATOR); + for (String elementEntry : entries) { + result.add(EntryUtil.getNameFromEntry(elementEntry)); + } + } + } else if (EntryUtil.isFile(entry)) { + throw new IOException(path + " is not a directory but a file"); + } else { + log.warn("strange entry" + entry); + throw new IOException("strange entry" + entry); + } + } + + log.info("readDirectory " + path + " returns " + result.size() + " results"); + + return result; + } + + /** + * return the entry of the element at the end of <code>path</code> + * @param path + * @return null if path is not valid + */ + protected String walk(String path) throws IOException { + String result = walk(path, null, null); + log.info("walking to " + path + " returns " + result); + return result; + } + + /** + * This method is a recursive function to walk through the tree structure + * of the directories and starting for root directory, following the + * symbolic links to reach the given path + * @param path + * @param current + * @param content + * @return null if path is not valid or the entry corresponding to path + * @throws IOException + */ + protected String walk(String path, String current, String content) + throws IOException { + // FIXME 20105021 bleny works fine but is not understandable + String result = null; + + // if path is "/", recursion can be initiated with this value : + // (it returns "/" as the id where to get the content of "/" + if (path.equals(EntryUtil.ROOT_DIRECTORY)) { + return EntryUtil.TYPE.D + EntryUtil.ENTRY_SEPARATOR + + EntryUtil.ROOT_DIRECTORY + EntryUtil.ENTRY_SEPARATOR + + EntryUtil.ROOT_DIRECTORY; + } + + String parentPath = EntryUtil.getParentFromPath(path); + + if (content == null) { + // start the recursion from root directory + content = storage.getRootDirectory(); + result = walk(path, EntryUtil.ROOT_DIRECTORY, content); + } else if (parentPath.equals(current)) { + // we are now in the last directory + // ie if path is a/b/p we are at a/b + + // p is the name of the element in a/b we search for + String tail = path.substring(current.length()); + String p = EntryUtil.getNameFromPath(tail); + + log.info("in final dir " + current + ", looking for " + p); + + String entry = EntryUtil.findEntryInDirectory(content, p); + result = entry; + } else { + // in middle of path + // if path is a/b/c/d/e/f, we may be at b, c, d... + + String tail; // the path still remaining when in current + // ie if we are in c, tail is "d/e/f" + + // if we are at root directory, deal with the "/" + if (current.equals(EntryUtil.ROOT_DIRECTORY)) { + tail = path.substring(current.length()); + } else { + tail = path.substring(current.length() + 1); + } + + log.debug("current = " + current); + log.debug("tail = " + tail); + String[] elementsNames = tail.split(EntryUtil.PATH_SEPARATOR); + String p = elementsNames[0]; + + log.info("in intermediate dir " + current + ", looking for " + p); + + // updating current for recursion + if (current.equals(EntryUtil.ROOT_DIRECTORY)) { + current = ""; // avoid "//path" next line + } + current += EntryUtil.PATH_SEPARATOR + p; + + String entry = EntryUtil.findEntryInDirectory(content, p); + if (entry == null) { + result = null; + } else { + // we have found the entry to call recursion + + if (EntryUtil.isDirectory(entry)) { + String id = EntryUtil.getIdFromEntry(entry); + content = storage.getDirectory(id); + result = walk(path, current, content); + } else if (EntryUtil.isLink(entry)) { + String id = EntryUtil.getIdFromEntry(entry); + String linkContent = storage.getLink(id); + String newTarget = + EntryUtil.resolveLink(current, linkContent); + newTarget += path.substring(current.length()); + + // restart walk from / + result = walk(newTarget, null, null); + } else if (EntryUtil.isFile(entry)) { + // error, found file in path like '/dir1/dir2/filename/dir3' + result = null; + } else { + log.warn("strange case: " + entry); + result = null; + } + } + } + return result; + } + + @Override + public void close() throws IOException { + storage.close(); + } + + protected void setMap(DisworkMap map) { + storage.setMap(map); + } + +} Copied: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java (from rev 66, trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/DisworkFileSystemConfig.java) =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemConfig.java 2010-06-08 13:16:35 UTC (rev 68) @@ -0,0 +1,190 @@ +/* + * #%L + * disworkfs + * + * $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.fs; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.Socket; +import java.net.UnknownHostException; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.nuiton.util.ApplicationConfig; + +/** + * This object contains parameters used in {@link DisworkFileSystem}. + * + * Available parameters are: + * <dl> + * <dt>blocks_size</dt> + * <dd> + * The size of blocks to create when splitting big data in bytes + * (by default, 10 MiB) + * </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 + * // will bootstrap by joining the + * // first node + * c3 = newDisworkConfig(c.getUsedPort()) + * </pre> + * + */ +public class DisworkFileSystemConfig extends ApplicationConfig { + + private static final Log log = + LogFactory.getLog(DisworkFileSystemConfig.class); + + protected static Integer port = 19000; + + /** + * returns a new port, returned value change at each call. + * @return + */ + public static Integer getPort() { + port += 1; + return port; + } + + /** + * returns the IP on the local machine. Trying to get an public IP or a LAN + * IP or the loopback IP if there is no other interface + * @return + * @throws UnknownHostException + */ + public static String getIp() throws UnknownHostException { + InetAddress result = InetAddress.getLocalHost(); + if (result.isLoopbackAddress()) { + try { + Socket temp = new Socket("microsoft.com", 80); + result = temp.getLocalAddress(); + temp.close(); + } catch (IOException e) { + log.warn("can't get external IP address"); + } + } + return result.getHostAddress(); + } + + public DisworkFileSystemConfig() { + setDefaultOption("diswork.fs.blocks_size", "10485760"); // 10 MiB + + setDefaultOption("diswork.fs.map_type", "inmemory"); + setDefaultOption("diswork.fs.use_port", port.toString()); + } + + public int getBlockSize() { + return getOptionAsInt("diswork.fs.blocks_size"); + } + + public Integer getUsedPort() { + return getOptionAsInt("diswork.fs.use_port"); + } + + public void setUsedPort(Integer port) { + setOption("diswork.fs.use_port", port.toString()); + } + + public String getBootstrapIp() { + return getOption("diswork.fs.bootstrap.ip"); + } + + public void setBootstrapIp(String ip) { + setOption("diswork.fs.bootstrap.ip", ip); + } + + public Integer getBootstrapPort() { + return getOptionAsInt("diswork.fs.bootstrap.port"); + } + + public void setBootstrapPort(Integer port) { + setOption("diswork.fs.bootstrap.port", port.toString()); + } + + public String getMapType() { + return getOption("diswork.fs.map_type"); + } + + /** + * returns a @link {@link DisworkFileSystemConfig} ready to be use as a + * config for a single-node instance of DisworkFS + * @return + * @throws UnknownHostException + */ + public static DisworkFileSystemConfig newPastryDisworkConfig() + throws UnknownHostException { + return newPastryDisworkConfig(null); + } + + /** + * returns a @link {@link DisworkFileSystemConfig} ready to be use as a + * config for a multiple-node instance of DisworkFS on a same machine. + * @param bootstrapPort the port on the same machine where another node can + * be found to bootstrap + * @return a complete config + * @throws UnknownHostException + */ + public static DisworkFileSystemConfig newPastryDisworkConfig(Integer bootstrapPort) + throws UnknownHostException { + DisworkFileSystemConfig result = new DisworkFileSystemConfig(); + String port = getPort().toString(); + String ip = getIp(); + result.setOption("diswork.fs.map_type", "pastry"); + result.setOption("diswork.fs.use_port", port); + result.setOption("diswork.fs.bootstrap.ip", ip); + if (bootstrapPort == null) { + result.setOption("diswork.fs.bootstrap.port", port); + } else { + result.setOption("diswork.fs.bootstrap.port", bootstrapPort.toString()); + } + return result; + } + + public static DisworkFileSystemConfig newKademliaDisworkConfig() + throws UnknownHostException { + return newKademliaDisworkConfig(null); + } + + public static DisworkFileSystemConfig newKademliaDisworkConfig(Integer bootstrapPort) + throws UnknownHostException { + 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); + } + return result; + } +} Copied: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/package-info.java (from rev 66, trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/package-info.java) =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/package-info.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/package-info.java 2010-06-08 13:16:35 UTC (rev 68) @@ -0,0 +1,35 @@ +/* + * #%L + * disworkfs + * + * $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% + */ +/** + * DisworkFS is a distributed file system. You can use it by instantiating + * {@link org.nuiton.diswork.fs.DisworkFileSystem} and use it to store + * directories and their content. + * + * You can change the default DisworkFileSystem behavior by provide a + * {@link org.nuiton.diswork.fs.DisworkFileSystemConfig} instance at + * construction. If you don't provide one, a instance will be created + */ + +package org.nuiton.diswork.fs; \ No newline at end of file Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/peerunit/DisworkFileSystemTest.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/peerunit/DisworkFileSystemTest.java 2010-06-08 12:13:22 UTC (rev 66) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/peerunit/DisworkFileSystemTest.java 2010-06-08 13:16:35 UTC (rev 68) @@ -22,7 +22,7 @@ * <http://www.gnu.org/licenses/lgpl-3.0.html>. * #L% */ -package org.nuiton.disworkfs.peerunit; +package org.nuiton.diswork.fs.peerunit; import static fr.inria.peerunit.test.assertion.Assert.assertTrue; import static fr.inria.peerunit.test.assertion.Assert.assertEquals; @@ -36,8 +36,8 @@ import org.apache.commons.io.IOUtils; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.nuiton.disworkfs.DisworkFileSystem; -import org.nuiton.disworkfs.DisworkFileSystemConfig; +import org.nuiton.diswork.fs.DisworkFileSystem; +import org.nuiton.diswork.fs.DisworkFileSystemConfig; import fr.inria.peerunit.TestCaseImpl; import fr.inria.peerunit.parser.TestStep; Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/peerunit/package-info.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/peerunit/package-info.java 2010-06-08 12:13:22 UTC (rev 66) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/peerunit/package-info.java 2010-06-08 13:16:35 UTC (rev 68) @@ -26,4 +26,4 @@ * In this package can be found some Test classes for PeerUnit framework */ -package org.nuiton.disworkfs.peerunit; \ No newline at end of file +package org.nuiton.diswork.fs.peerunit; \ No newline at end of file Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/DisworkMap.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/storage/DisworkMap.java 2010-06-08 12:13:22 UTC (rev 66) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/DisworkMap.java 2010-06-08 13:16:35 UTC (rev 68) @@ -22,7 +22,7 @@ * <http://www.gnu.org/licenses/lgpl-3.0.html>. * #L% */ -package org.nuiton.disworkfs.storage; +package org.nuiton.diswork.fs.storage; import java.io.Closeable; import java.util.Map; Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/EntryUtil.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/storage/EntryUtil.java 2010-06-08 12:13:22 UTC (rev 66) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/EntryUtil.java 2010-06-08 13:16:35 UTC (rev 68) @@ -22,7 +22,7 @@ * <http://www.gnu.org/licenses/lgpl-3.0.html>. * #L% */ -package org.nuiton.disworkfs.storage; +package org.nuiton.diswork.fs.storage; import java.nio.charset.Charset; Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/InMemoryDisworkMap.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/storage/InMemoryDisworkMap.java 2010-06-08 12:13:22 UTC (rev 66) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/InMemoryDisworkMap.java 2010-06-08 13:16:35 UTC (rev 68) @@ -22,7 +22,7 @@ * <http://www.gnu.org/licenses/lgpl-3.0.html>. * #L% */ -package org.nuiton.disworkfs.storage; +package org.nuiton.diswork.fs.storage; import java.io.IOException; import java.util.Arrays; Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/KademliaDisworkMap.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/storage/KademliaDisworkMap.java 2010-06-08 12:13:22 UTC (rev 66) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/KademliaDisworkMap.java 2010-06-08 13:16:35 UTC (rev 68) @@ -22,7 +22,7 @@ * <http://www.gnu.org/licenses/lgpl-3.0.html>. * #L% */ -package org.nuiton.disworkfs.storage; +package org.nuiton.diswork.fs.storage; import java.io.DataInput; import java.io.DataInputStream; @@ -37,7 +37,7 @@ import org.apache.commons.io.IOUtils; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.nuiton.disworkfs.DisworkFileSystemConfig; +import org.nuiton.diswork.fs.DisworkFileSystemConfig; import org.planx.xmlstore.routing.Identifier; import org.planx.xmlstore.routing.Kademlia; Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/PastryDisworkMap.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/storage/PastryDisworkMap.java 2010-06-08 12:13:22 UTC (rev 66) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/PastryDisworkMap.java 2010-06-08 13:16:35 UTC (rev 68) @@ -22,7 +22,7 @@ * <http://www.gnu.org/licenses/lgpl-3.0.html>. * #L% */ -package org.nuiton.disworkfs.storage; +package org.nuiton.diswork.fs.storage; import java.io.IOException; import java.net.BindException; @@ -36,7 +36,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.nuiton.disworkfs.DisworkFileSystemConfig; +import org.nuiton.diswork.fs.DisworkFileSystemConfig; import rice.Continuation; import rice.environment.Environment; Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/Storage.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/storage/Storage.java 2010-06-08 12:13:22 UTC (rev 66) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/Storage.java 2010-06-08 13:16:35 UTC (rev 68) @@ -22,7 +22,7 @@ * <http://www.gnu.org/licenses/lgpl-3.0.html>. * #L% */ -package org.nuiton.disworkfs.storage; +package org.nuiton.diswork.fs.storage; import java.io.ByteArrayInputStream; import java.io.Closeable; @@ -36,7 +36,7 @@ import org.apache.commons.io.IOUtils; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.nuiton.disworkfs.DisworkFileSystemConfig; +import org.nuiton.diswork.fs.DisworkFileSystemConfig; /** * This class is the middle layer between the File System operations and Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/package-info.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/disworkfs/storage/package-info.java 2010-06-08 12:13:22 UTC (rev 66) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/package-info.java 2010-06-08 13:16:35 UTC (rev 68) @@ -26,7 +26,7 @@ * * This package provides to {@link org.nuiton.disworkfs.DisworkFileSystem} a * way to persistently store data. This is done by the - * {@link org.nuiton.disworkfs.storage.Storage} class + * {@link org.nuiton.diswork.fs.storage.Storage} class * which permit to store different type of data. * * @@ -108,13 +108,13 @@ * Link), the name of the element, and an ID to be used as * a key on the map to get the actual content. Those three informations are * separated by ":" which is - * {@link org.nuiton.disworkfs.storage.EntryUtil#ENTRY_SEPARATOR}. + * {@link org.nuiton.diswork.fs.storage.EntryUtil#ENTRY_SEPARATOR}. * * * * A directory way have multiple entries. Entries of a directory are * separated by "\n" which is - * {@link org.nuiton.disworkfs.storage.EntryUtil#ENTRIES_SEPARATOR}. + * {@link org.nuiton.diswork.fs.storage.EntryUtil#ENTRIES_SEPARATOR}. * * * The above description shows the main principle used to store a tree @@ -198,7 +198,7 @@ * A metablock is composed of the total size of the file followed by * an ordered lists of IDs of the different blocks composing the file. * Those informations are separated by ";" (see - * {@link org.nuiton.disworkfs.storage.EntryUtil#BLOCKIDS_SEPARATOR}). + * {@link org.nuiton.diswork.fs.storage.EntryUtil#BLOCKIDS_SEPARATOR}). * * * In the above example: @@ -222,10 +222,10 @@ * When reading and writing in storage, split is done transparently. When * reading, a Stream is returned: it loads data blocks after blocks * when needed (see - * {@link org.nuiton.disworkfs.storage.Storage.SplitBlocksInputStream}). + * {@link org.nuiton.diswork.fs.storage.Storage.SplitBlocksInputStream}). * When writing, data are split in blocks of a maximum configurable size * (see {@link org.nuiton.disworkfs.DisworkFileSystemConfig}). * */ -package org.nuiton.disworkfs.storage; \ No newline at end of file +package org.nuiton.diswork.fs.storage; \ No newline at end of file Copied: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java (from rev 66, trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/AbstractDisworkFileSystemTest.java) =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java (rev 0) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java 2010-06-08 13:16:35 UTC (rev 68) @@ -0,0 +1,361 @@ +package org.nuiton.diswork.fs; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.ByteArrayInputStream; +import java.io.File; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.util.ConcurrentModificationException; +import java.util.List; +import java.util.Random; + +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; + + +public abstract class AbstractDisworkFileSystemTest { + + /** + * a place to store files for the test it's a subdirectory of the OS temp + * dir e.g. under linux /tmp/disworkfs/tests/ + */ + static protected String tempDirectoryPath = + System.getProperty("java.io.tmpdir", ".") // temp directory + + "/disworkfs/tests"; + + /** + * We will create a file at this path for test purpose + */ + static protected String randomFilePath = tempDirectoryPath + "/randomfile"; + + /** + * The file will have this fixed size + */ + static protected int randomFileSize = 9999; + + static protected DisworkFileSystem fileSystem; + + /** + * This setUp creates in a temp directory a file of size + * {@link randomFileSize} (fulfilling it with random bytes) + * This file can be found at {@link #randomFilePath} + * At the end of the method {@link #fileSystem} is ready to be used + * @throws Exception + */ + @Before + public void setUp() throws Exception { + // create a temp directory for our test + File tempDirectory = new File(tempDirectoryPath); + tempDirectory.mkdir(); + + // creating random data for the file + Random random = new Random(); + byte[] randomBytes = new byte[randomFileSize]; + random.nextBytes(randomBytes); + + // dumping random data into the file + File randomFile = new File(randomFilePath); + FileUtils.writeByteArrayToFile(randomFile, randomBytes); + + } + + @After + public void tearDown() throws Exception { + // cleaning + FileUtil.deleteRecursively(tempDirectoryPath); + } + + /** + * writing a file at root directory should not raise any exception + * @throws Exception + */ + @Test + public void testWrite() throws Exception { + fileSystem.write("/", "my_file", new FileInputStream(randomFilePath)); + } + + /** + * First, write a file then test if exists return true. + * @throws Exception + */ + @Test + public void testExists() throws Exception { + fileSystem.write("/", "my_file", new FileInputStream(randomFilePath)); + assertTrue(fileSystem.exists("/my_file")); + assertFalse(fileSystem.exists("/my_other_file")); + } + + /** + * try to read a file that as never been created nor written + */ + @Test(expected = FileNotFoundException.class) + public void testFailAtRead() throws Exception { + fileSystem.read("/not_existing_file"); + } + + /** + * tests {@link org.nuiton.diswork.fs.storage.Storage#SplitBlocksInputStream} + * by storing a "-1" byte, can be buggy due to the use of read() + * @throws IOException + */ + @Test + public void testSplit() throws IOException { + + byte[] bytes = new byte[1]; + bytes[0] = -0x1; + + InputStream source; + + source = new ByteArrayInputStream(bytes); + fileSystem.write("/", "my_file", source); + + source.close(); + + + source = new ByteArrayInputStream(bytes); + InputStream readResult = fileSystem.read("/my_file"); + + int read = 0; + byte[] b = new byte[1]; + + read = readResult.read(b); + + assertEquals(1, read); + assertArrayEquals(bytes, b); + + } + + /** + * writing a file at the root directory and reading it. + * finally, compare original source and read result + * byte-to-byte: contents should be equals + * @throws Exception + */ + @Test + public void testWriteRead() throws Exception { + + InputStream source = new FileInputStream(randomFilePath); + + fileSystem.write("/", "my_file", 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()); + + byte[] sourceAsBytes = IOUtils.toByteArray(source); + byte[] readResultAsBytes = IOUtils.toByteArray(readResult); + + assertArrayEquals(sourceAsBytes, readResultAsBytes); + + source.close(); + readResult.close(); + + } + + /** + * this use case should raise an exception because my_folder + * doesn't exists + */ + @Test(expected = IOException.class) + public void testFailAtWrite() throws Exception { + fileSystem.write("/my_folder", "my_file", + new FileInputStream(randomFilePath)); + } + + /** + * this use case should raise an exception because writing to a file + * that already exists + */ + @Test(expected = IOException.class) + 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(); + } + } + + /** + * This test uses mkdir to create dirs and sub-dirs + * @throws Exception + */ + @Test + public void testCreateDirectory() throws Exception { + fileSystem.createDirectory("/my_folder"); + assertTrue(fileSystem.exists("/my_folder")); + fileSystem.createDirectory("/my_folder/my_sub_folder"); + assertTrue(fileSystem.exists("/my_folder/my_sub_folder")); + } + + /** + * Create some folders with mkdir and write files in those directories + * @throws Exception + */ + @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")); + } + + /** + * create a symbolic link to a file. This test show that we can read the + * file using the link + * @throws Exception + */ + @Test + public void testLinking() throws Exception { + fileSystem.createDirectory("/my_folder"); + fileSystem.write("/my_folder", "my_file", + new FileInputStream(randomFilePath)); + fileSystem.createSymbolicLink("/my_link", "/my_folder/my_file"); + + InputStream source = new FileInputStream(randomFilePath); + InputStream readResult = fileSystem.read("/my_link"); + + boolean actualContentEquality = + IOUtils.contentEquals(source, readResult); + source.close(); + readResult.close(); + + assertTrue(actualContentEquality); + } + + /** + * Trying to create a link to a wrong target, this sould raise an exception + * @throws Exception + */ + @Test(expected = IOException.class) + public void testFailAtLinking() throws Exception { + fileSystem.createSymbolicLink("/my_link", "/wrong_target_path"); + } + + /** + * Trying to remove files and directories + * @throws Exception + */ + @Test + public void testRemove() throws Exception { + fileSystem.createDirectory("/my_folder"); + fileSystem.write("/my_folder", "my_file", + new FileInputStream(randomFilePath)); + fileSystem.delete("/my_folder/my_file"); + assertTrue(fileSystem.exists("/my_folder")); + assertFalse(fileSystem.exists("/my_folder/my_file")); + fileSystem.delete("/my_folder"); + assertFalse(fileSystem.exists("/my_folder")); + } + + /** + * By trying to remove a non-empty directory, this test should raise an + * exception + * @throws Exception + */ + @Test(expected = IOException.class) + public void testFailAtRemove() throws Exception { + fileSystem.createDirectory("/my_folder"); + fileSystem.write("/my_folder", "my_file", + new FileInputStream(randomFilePath)); + + // trying to remove a non-empty directory should raise an exception + fileSystem.delete("/my_folder"); + } + + /** + * This tests uses ls + */ + @Test + 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)); + fileSystem.createSymbolicLink("/my_folder/my_link", "my_file"); + + List<String> lsResult = fileSystem.readDirectory("/my_folder"); + + // checking that result contains all the required data + assertTrue(lsResult.contains("my_sub_dir")); + assertTrue(lsResult.contains("my_file")); + assertTrue(lsResult.contains("my_link")); + + // ... and only those + assertEquals(3, lsResult.size()); + + lsResult = fileSystem.readDirectory("/"); + assertEquals(1, lsResult.size()); + } + + @Test + public void testConcurrency() throws Exception { + fileSystem.createDirectory("/mydir"); + try { + fileSystem.createDirectory("/myseconddir"); + } catch (ConcurrentModificationException e) { + fail(); + } + } + + @Test + public void testLinks() throws Exception { + fileSystem.createDirectory("/dir"); + fileSystem.createDirectory("/dir/subdir"); + fileSystem.createDirectory("/otherdir"); + + fileSystem.createSymbolicLink("/link", "dir/subdir"); + fileSystem.createSymbolicLink("/link/subsubdirlink", "/otherdir"); + List<String> readResult = fileSystem.readDirectory("/dir/subdir"); + assertEquals(1, readResult.size()); + assertTrue(readResult.contains("subsubdirlink")); + + fileSystem.createDirectory("/link/subsubdirlink/finaldir"); + readResult = fileSystem.readDirectory("/otherdir"); + assertEquals(1, readResult.size()); + assertTrue(readResult.contains("finaldir")); + + // tests that delete remove the link itself and not the target + fileSystem.delete("/link/subsubdirlink"); + assertTrue(fileSystem.exists("/otherdir")); + fileSystem.delete("/link"); + assertTrue(fileSystem.exists("/dir/subdir")); + + // testing the use of multiple links at the end of a path + fileSystem.createSymbolicLink("/link1", "/dir"); + fileSystem.createSymbolicLink("/link2", "/link1"); + fileSystem.createSymbolicLink("/link3", "/link2"); + + readResult = fileSystem.readDirectory("/link3"); + assertEquals(1, readResult.size()); + assertTrue(readResult.contains("subdir")); + } +} Copied: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemInMemoryTest.java (from rev 66, trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/DisworkFileSystemInMemoryTest.java) =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemInMemoryTest.java (rev 0) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemInMemoryTest.java 2010-06-08 13:16:35 UTC (rev 68) @@ -0,0 +1,21 @@ +package org.nuiton.diswork.fs; + +import org.junit.Before; +import org.nuiton.diswork.fs.DisworkFileSystem; +import org.nuiton.diswork.fs.DisworkFileSystemConfig; + +public class DisworkFileSystemInMemoryTest extends AbstractDisworkFileSystemTest { + + /** + * this code executed after {@link AbstractDisworkFileSystemTest#setUp()} + * @throws Exception + */ + @Before + public void setUpFileSystem() throws Exception { + // finally, initiate the fileSystem + DisworkFileSystemConfig disworkConfig = new DisworkFileSystemConfig(); + disworkConfig.setOption("diswork.fs.map_type", "inmemory"); + fileSystem = new DisworkFileSystem(disworkConfig); + } + +} Copied: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java (from rev 66, trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/DisworkFileSystemKademliaTest.java) =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java (rev 0) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java 2010-06-08 13:16:35 UTC (rev 68) @@ -0,0 +1,79 @@ +package org.nuiton.diswork.fs; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.ByteArrayInputStream; +import java.util.ConcurrentModificationException; + +import org.apache.commons.io.IOUtils; +import org.junit.Before; +import org.junit.Test; +import org.nuiton.diswork.fs.DisworkFileSystem; +import org.nuiton.diswork.fs.DisworkFileSystemConfig; + +public class DisworkFileSystemKademliaTest extends AbstractDisworkFileSystemTest { + + protected Integer bootstrapPort; + + /** + * this code executed after {@link AbstractDisworkFileSystemTest#setUp()} + * @throws Exception + */ + @Before + public void setUpFileSystem() throws Exception { + // finally, initiate the fileSystem + DisworkFileSystemConfig disworkConfig1 = + DisworkFileSystemConfig.newKademliaDisworkConfig(); + bootstrapPort = disworkConfig1.getUsedPort(); + fileSystem = new DisworkFileSystem(disworkConfig1); + } + + + @Test + public void testMultipleNodes1() throws Exception { + DisworkFileSystemConfig disworkConfig = + DisworkFileSystemConfig.newKademliaDisworkConfig(bootstrapPort); + DisworkFileSystem fileSystem2 = new DisworkFileSystem(disworkConfig); + + assertTrue(fileSystem.exists("/")); + assertTrue(fileSystem2.exists("/")); + } + + @Test + public void testMultipleNodes2() throws Exception { + 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)); + + assertTrue(fileSystem.exists("/my_file")); + assertTrue(fileSystem2.exists("/my_file")); + + assertEquals(1, fileSystem2.readDirectory("/").size()); + + byte[] getResult = IOUtils.toByteArray(fileSystem2.read("/my_file")); + + assertArrayEquals(bytes, getResult); + + } + + @Test + public void testMultipleNodes3() throws Exception { + + fileSystem.createDirectory("/mydir"); + try { + fileSystem.createDirectory("/myseconddir"); + } catch (ConcurrentModificationException e) { + fail(); + } + + } + + +} Copied: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemPastryTest.java (from rev 66, trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/DisworkFileSystemPastryTest.java) =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemPastryTest.java (rev 0) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemPastryTest.java 2010-06-08 13:16:35 UTC (rev 68) @@ -0,0 +1,77 @@ +package org.nuiton.diswork.fs; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.ByteArrayInputStream; +import java.util.ConcurrentModificationException; + +import org.apache.commons.io.IOUtils; +import org.junit.Before; +import org.junit.Test; +import org.nuiton.diswork.fs.DisworkFileSystem; +import org.nuiton.diswork.fs.DisworkFileSystemConfig; + +public class DisworkFileSystemPastryTest extends AbstractDisworkFileSystemTest { + + protected Integer bootstrapPort; + + /** + * this code executed after {@link AbstractDisworkFileSystemTest#setUp()} + * @throws Exception + */ + @Before + public void setUpFileSystem() throws Exception { + // finally, initiate the fileSystem + DisworkFileSystemConfig disworkConfig1 = DisworkFileSystemConfig.newPastryDisworkConfig(); + bootstrapPort = disworkConfig1.getUsedPort(); + fileSystem = new DisworkFileSystem(disworkConfig1); + + } + + + @Test + public void testMultipleNodes1() throws Exception { + DisworkFileSystemConfig disworkConfig = DisworkFileSystemConfig.newPastryDisworkConfig(bootstrapPort); + DisworkFileSystem fileSystem2 = new DisworkFileSystem(disworkConfig); + + assertTrue(fileSystem.exists("/")); + assertTrue(fileSystem2.exists("/")); + } + + @Test + public void testMultipleNodes2() throws Exception { + DisworkFileSystemConfig disworkConfig = DisworkFileSystemConfig.newPastryDisworkConfig(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)); + + assertTrue(fileSystem.exists("/my_file")); + assertTrue(fileSystem2.exists("/my_file")); + + assertEquals(1, fileSystem2.readDirectory("/").size()); + + byte[] getResult = IOUtils.toByteArray(fileSystem2.read("/my_file")); + + assertArrayEquals(bytes, getResult); + + } + + @Test + public void testMultipleNodes3() throws Exception { + + fileSystem.createDirectory("/mydir"); + try { + fileSystem.createDirectory("/myseconddir"); + } catch (ConcurrentModificationException e) { + fail(); + } + + } + + +} Modified: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/AbstractDisworkMapTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/storage/AbstractDisworkMapTest.java 2010-06-08 12:13:22 UTC (rev 66) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/AbstractDisworkMapTest.java 2010-06-08 13:16:35 UTC (rev 68) @@ -1,4 +1,4 @@ -package org.nuiton.disworkfs.storage; +package org.nuiton.diswork.fs.storage; import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertFalse; @@ -11,6 +11,7 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.nuiton.diswork.fs.storage.DisworkMap; public abstract class AbstractDisworkMapTest { Modified: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/EntryUtilTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/storage/EntryUtilTest.java 2010-06-08 12:13:22 UTC (rev 66) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/EntryUtilTest.java 2010-06-08 13:16:35 UTC (rev 68) @@ -15,7 +15,7 @@ * along with this program. If not, see <http://www.gnu.org/licenses/>. *##%*/ -package org.nuiton.disworkfs.storage; +package org.nuiton.diswork.fs.storage; import static org.junit.Assert.assertArrayEquals; @@ -24,7 +24,7 @@ import static org.junit.Assert.assertTrue; import org.junit.Test; -import org.nuiton.disworkfs.storage.EntryUtil; +import org.nuiton.diswork.fs.storage.EntryUtil; /** * Modified: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/InMemoryDisworkMapTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/storage/InMemoryDisworkMapTest.java 2010-06-08 12:13:22 UTC (rev 66) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/InMemoryDisworkMapTest.java 2010-06-08 13:16:35 UTC (rev 68) @@ -1,6 +1,7 @@ -package org.nuiton.disworkfs.storage; +package org.nuiton.diswork.fs.storage; import org.junit.Before; +import org.nuiton.diswork.fs.storage.InMemoryDisworkMap; public class InMemoryDisworkMapTest extends AbstractDisworkMapTest { Modified: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/KademliaDisworkMapTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/storage/KademliaDisworkMapTest.java 2010-06-08 12:13:22 UTC (rev 66) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/KademliaDisworkMapTest.java 2010-06-08 13:16:35 UTC (rev 68) @@ -1,4 +1,4 @@ -package org.nuiton.disworkfs.storage; +package org.nuiton.diswork.fs.storage; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; @@ -10,7 +10,8 @@ import org.junit.Before; import org.junit.Test; -import org.nuiton.disworkfs.DisworkFileSystemConfig; +import org.nuiton.diswork.fs.DisworkFileSystemConfig; +import org.nuiton.diswork.fs.storage.KademliaDisworkMap; import org.planx.xmlstore.routing.Identifier; public class KademliaDisworkMapTest extends AbstractDisworkMapTest { Modified: trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/PastryDisworkMapTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/disworkfs/storage/PastryDisworkMapTest.java 2010-06-08 12:13:22 UTC (rev 66) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/PastryDisworkMapTest.java 2010-06-08 13:16:35 UTC (rev 68) @@ -1,3 +1,4 @@ +package org.nuiton.diswork.fs.storage; //package org.nuiton.disworkfs.storage; // //import org.junit.Before;

1 0