[Diswork-commits] r94 - in trunk: diswork-daemon diswork-daemon/src/main/java/org/nuiton/diswork/daemon diswork-daemon/src/test/java/org/nuiton/diswork/daemon diswork-fs diswork-fs/src/main/java/org/nuiton/diswork/fs diswork-fs/src/main/java/org/nuiton/diswork/fs/storage diswork-fs/src/test/java/org/nuiton/diswork/fs diswork-fs/src/test/java/org/nuiton/diswork/fs/storage

Author: bleny Date: 2010-07-01 12:45:20 +0200 (Thu, 01 Jul 2010) New Revision: 94 Url: http://nuiton.org/repositories/revision/diswork/94 Log: refactoring exception management, tokens property files, diswork runnable as an OS service Added: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemException.java Removed: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkFileSystemException.java Modified: trunk/diswork-daemon/ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/ActivityStrategy.java trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkConfig.java trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemon.java trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/JobDescription.java trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonTest.java trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/JobDescriptionTest.java trunk/diswork-fs/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/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/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/AbstractDisworkMapTest.java trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/KademliaDisworkMapTest.java Property changes on: trunk/diswork-daemon ___________________________________________________________________ Modified: svn:ignore - target .classpath .settings + target .classpath .settings .project diswork-daemon.iml Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/ActivityStrategy.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/ActivityStrategy.java 2010-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/ActivityStrategy.java 2010-07-01 10:45:20 UTC (rev 94) @@ -95,9 +95,7 @@ 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(); + log.warn("load average monitoring interrupted", e); } loadAverage10MinutesBefore = loadAverage5MinutesBefore; loadAverage5MinutesBefore = loadAverageNow; 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-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkConfig.java 2010-07-01 10:45:20 UTC (rev 94) @@ -25,8 +25,17 @@ package org.nuiton.diswork.daemon; import java.io.File; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; import java.lang.management.ManagementFactory; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.nuiton.diswork.fs.DisworkFileSystemConfig; import org.nuiton.util.ApplicationConfig; @@ -58,14 +67,28 @@ * nights, week-end, etc.). It needs to define a pattern.</dd> * </dl> * </dd> + * <dt>diswork.tokens_file</dt> + * <dd>a path to a file containing token-names and the replacement strings + * this is a property file</dd> * </dl> * * @author bleny */ public class DisworkConfig extends ApplicationConfig { - + + private static final Log log = LogFactory.getLog(DisworkConfig.class); + protected DisworkFileSystemConfig fileSystemConfig; + + /** TOKENS are piece of string you can use for writing command lines + * some tokens are defined by default, some others can be defined in a + * properties file + * + * @see #initTokens() + */ + protected Map<String, String> tokens; + public DisworkConfig() { setConfigFileName("diswork.config"); @@ -94,6 +117,75 @@ return newConfig; } + /** + * Read the tokens file if one is given in the config and merge the content + * of this file into {@link #tokens} + * + * @throws DisworkException + */ + protected void initTokens() throws DisworkException { + tokens = new HashMap<String, String>(); + + String java = // full java path + System.getProperty("java.home") + + File.separator + "bin" + File.separator + "java" + // setting some system properties + + " -Duser.dir=%tmp" + + " -Duser.home=%tmp" + + " -Djava.io.tmpdir=%tmp" + ; + tokens.put("%java", java); + + File tokensFile = getTokensFile(); + if (tokensFile != null) { + try { + InputStream tokensStream = new FileInputStream(tokensFile); + Properties userTokens = new Properties(); + userTokens.load(tokensStream); + for (String token : userTokens.stringPropertyNames()) { + log.debug("adding token " + token + " → " + + userTokens.getProperty(token)); + tokens.put(token, userTokens.getProperty(token)); + } + } catch (FileNotFoundException e) { + log.warn("tokens file not found, 0 tokens loaded", e); + throw new DisworkException("tokens file not found," + + "no token loaded", e); + } catch (IOException e) { + log.error("can't read tokens file", e); + throw new DisworkException("can't read tokens file", e); + } + } + } + + protected String applyTokensRecursively(String commandLine) { + String result = commandLine; + for (String token : tokens.keySet()) { + result = result.replaceAll(token, tokens.get(token)); + } + if (result.equals(commandLine)) { + return result; + } else { + // a token has been applied, re-do a pass + return applyTokensRecursively(result); + } + } + + protected String parseCommandLine(String commandLine, String tempDir) + throws DisworkException { + if (tokens == null) { + initTokens(); + } + + tokens.put("%tmp", tempDir); + + return applyTokensRecursively(commandLine); + } + + + + + public String getOwnerId() { return getOption("diswork.owner"); } @@ -102,10 +194,6 @@ setOption("diswork.owner", ownerId); } - - - - public String getBootstrapIp() { return fileSystemConfig.getBootstrapIp(); } @@ -171,5 +259,22 @@ String firstRunTime = getOption("diswork.first_run_time"); return Long.parseLong(firstRunTime); } - + + /** + * + * @return null if no file specified + */ + public File getTokensFile() { + String path = getOption("diswork.tokens_file"); + if (path == null) { + return null; + } else { + File file = new File(path); + return file; + } + } + + public void setTokensFile(String tokensFilePath) { + setOption("diswork.tokens_file", tokensFilePath); + } } Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemon.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemon.java 2010-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkDaemon.java 2010-07-01 10:45:20 UTC (rev 94) @@ -25,14 +25,11 @@ package org.nuiton.diswork.daemon; import java.io.Closeable; -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.ArrayList; -import java.util.ConcurrentModificationException; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -44,6 +41,7 @@ import org.apache.commons.logging.LogFactory; import org.nuiton.diswork.fs.DisworkFileSystem; import org.nuiton.diswork.fs.DisworkFileSystemConfig; +import org.nuiton.diswork.fs.DisworkFileSystemException; import org.nuiton.util.FileUtil; /** @@ -130,7 +128,7 @@ * Hierarchy Standard. All usual paths used by Diswork are defined in * constants. * - * @see http://en.wikipedia.org/wiki/Filesystem_Hierarchy_Standard + * @link http://en.wikipedia.org/wiki/Filesystem_Hierarchy_Standard * * @author bleny */ @@ -202,49 +200,81 @@ public DisworkDaemon(DisworkConfig config) throws DisworkException { this.config = config; - // init fileSystem with all needed directories + // step by step, set all the dependencies of the daemon + + // init the file-system + initFileSystem(); + + initOwnerIdAndHomeDir(); + + initWorkersManager(); + + writeHardwareInfos(); + + sessionStartTime = System.currentTimeMillis(); + } + + /* *** init methods, used once by the constructor */ + + /** + * set {@link #fileSystem} and create all base directories + * on the File System + */ + protected void initFileSystem() throws DisworkException { + 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) { + + // init fileSystem with all needed directories + String[] directories = { BIN, HOME, TODO, TODO_RUNNING, FAILED_1, + FAILED_1_RUNNING, FAILED_2, FAILED_2_RUNNING, FAILED_3, DONE}; + // 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); + } + } + } + } catch (DisworkFileSystemException e) { log.error("booting diswork file system failed", e); throw new DisworkException("booting diswork file system failed", e); } + } + /** + * set {@link #ownerId} and {@link #homeDir} + * @throws DisworkException + */ + protected void initOwnerIdAndHomeDir() throws DisworkException { 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"); - + // 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 + Random random = new Random(); while (alreadyExists) { + ownerId = simpleName + random.nextInt(); 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); + + fileSystem.createDirectory(HOME + "/" + ownerId); + + } catch (DisworkFileSystemException e) { + log.error("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); @@ -254,57 +284,42 @@ // config.saveForUser(); } - log.info("owner id is " + ownerId); + homeDir = HOME + "/" + ownerId; + log.info("owner id is " + ownerId); + } + + /** + * + * @throws DisworkException + */ + protected void initWorkersManager() throws DisworkException { // 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(); - + protected void writeHardwareInfos() throws DisworkException { // writing hardware info to homeDir try { OperatingSystemMXBean os = ManagementFactory.getOperatingSystemMXBean(); String hardinfos = os.getName() + "\n" + os.getArch() + "\n" + - os.getAvailableProcessors(); - + config.getNumberOfWorkers(); + // 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 write hardware info", e); - throw new DisworkException("can't write hardware info", e); - } catch (IOException e) { - log.info("can't write hardware info", e); - throw new DisworkException("can't write hardware info", e); + log.info("writing hardware infos " + hardinfos.replaceAll("\n", " ")); + } catch (DisworkFileSystemException e) { + log.error("can't write hardware infos", e); + throw new DisworkException("can't write hardware infos", e); } } - /** - * 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, - 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); - } - } - } - } - /* *** methods for defining some usual paths *** */ /** @@ -317,7 +332,7 @@ protected static String getPathForDependency(String applicationName, String applicationVersion) { - String result = "/bin/" + applicationName // application directory + String result = BIN + "/" + applicationName // application directory + "/" // application file name + applicationName + "-" + applicationVersion + ".zip"; @@ -337,7 +352,7 @@ /** * Given a job id, returns the place on disworkFS where all data * for this jobs should be stored - * @param jobDescription + * @param jobId * @return a path */ protected String getPathForJob(String jobId) { @@ -353,7 +368,10 @@ protected static String newJobLinkName() { return ((Long) System.currentTimeMillis()).toString(); } - + + + /* ** public methods for use of the daemon */ + /** * Provide an application to all nodes. Once provided, all nodes will be * able to perform a job with this application. @@ -384,12 +402,9 @@ 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); + } catch (DisworkFileSystemException e) { + log.error("unable to publish application", e); + throw new DisworkException("unable to publish application", e); } } @@ -410,8 +425,11 @@ result.add(jobDescription); } } catch (IOException e) { - log.info("error file reading home-dir", e); - throw new DisworkException("error file reading home-dir", e); + log.error("error in file while reading home-dir", e); + throw new DisworkException("error in file while reading home-dir", e); + } catch (DisworkFileSystemException e) { + log.error("file-system error", e); + throw new DisworkException("file-system error", e); } return result; } @@ -428,12 +446,12 @@ public void submitJob(JobDescription jobDescription) throws DisworkException { - if (jobDescription.getInputData().size() + jobDescription.getStagingInputUrls().size() - < jobDescription.getStagingInput().size()) { + if (jobDescription.getInputData().size() + jobDescription.getInputUrls().size() + < jobDescription.getInput().size()) { // dependencies are missing } - try { + try { // trying to put the job in a new directory of home Random random = new Random(); @@ -468,10 +486,6 @@ } String jobDir = getPathForJob(jobDescription); - - if(!fileSystem.exists(jobDir)) { - // strange ! - } // creating an empty log file log.info("creating log file " + jobDir + "/" + LOG_PATH); @@ -495,10 +509,10 @@ fileSystem.createSymbolicLink(TODO + "/" + linkName, jobDir); log.info("job submited"); - - } catch (IOException e) { + + } catch (DisworkFileSystemException e) { log.error("file system error", e); - throw new DisworkFileSystemException("file system error", e); + throw new DisworkException(e); } } @@ -508,12 +522,12 @@ String jobPath = getPathForJob(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); + log.info("unable to read log ", e); + throw new DisworkException("unable to read log ", e); + } catch (DisworkFileSystemException e) { + log.error("unable to read log file" + job, e); + throw new DisworkException("unable to read log file" + job, e); } } @@ -541,15 +555,13 @@ throws DisworkException { if (isFinished(job)) { Map<String, InputStream> results = new HashMap<String, InputStream>(); - for (String fileName : job.getStagingOutput()) { + for (String fileName : job.getOutput()) { 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); + } catch (DisworkFileSystemException e) { + log.error("file system error ", e); throw new DisworkException("file system error ", e); } } @@ -562,11 +574,13 @@ } /** close the daemon (stop all workers) - * update statistices and delete all temporary data + * update statistics and delete all temporary data */ @Override public void close() throws IOException { - workers.stop(); + if (workers != null) { + workers.stop(); + } // updating total uptime statistic Long totalUptime = getTotalUptime(); @@ -601,7 +615,7 @@ return uptimeRatio; } - /** get informations on hardware available on the global Diswork system + /** get infos on hardware available on the global Diswork system * * @return * @throws DisworkException @@ -634,9 +648,12 @@ Integer.parseInt(infos[2])); } return stats; + } catch (DisworkFileSystemException e) { + log.error("file system error ", e); + throw new DisworkException("file system error ", e); } catch (IOException e) { - log.info("file system error ", e); - throw new DisworkException("file system error ", e); + log.error("can't read hardware infos ", e); + throw new DisworkException("can't read hardware infos ", e); } } } \ No newline at end of file Deleted: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkFileSystemException.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkFileSystemException.java 2010-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkFileSystemException.java 2010-07-01 10:45:20 UTC (rev 94) @@ -1,47 +0,0 @@ -/* - * #%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); - } - -} 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-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/JobDescription.java 2010-07-01 10:45:20 UTC (rev 94) @@ -54,14 +54,17 @@ * <dt>%java</dt> * <dd>will be replaced by the actual path of java executable * into the JRE</dd> + * <dt>%tmp</dt> + * <dd>will be replaced by the the path to dir where the job + * is executed, it's nice to use this to set a temp directory</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 + * @link http://en.wikipedia.org/wiki/Job_Submission_Description_Language + * @link http://www.gridforum.org/documents/GFD.56.pdf * * @author bleny */ @@ -106,15 +109,6 @@ /** 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 */ - 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 #JobDescription()} constructor @@ -136,11 +130,7 @@ } public String getCommandLine() { - String result = commandLine; - for (String token : TOKENS.keySet()) { - result = result.replace(token, TOKENS.get(token)); - } - return result; + return commandLine; } public void setCommandLine(String commandLine) { @@ -157,9 +147,7 @@ /** * 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)} + * {@link #setApplication(String, String)} which need to give the version. */ protected void setApplicationName(String applicationName) { this.applicationName = applicationName; @@ -172,8 +160,6 @@ /** * 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; @@ -194,6 +180,10 @@ return "job : " + jobName + " (" + jobId + ")"; } + /** + * Get a JSDL that describes this job + * @return a string containing the complete XML + */ public String toJSDL() { String jsdl = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n" @@ -356,15 +346,15 @@ output.add(fileName); } - public List<String> getStagingInput() { + public List<String> getInput() { return input; } - public List<String> getStagingOutput() { + public List<String> getOutput() { return output; } - public Map<String, URL> getStagingInputUrls() { + public Map<String, URL> getInputUrls() { return inputUrls; } Modified: trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java =================================================================== --- trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java 2010-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/WorkersManager.java 2010-07-01 10:45:20 UTC (rev 94) @@ -28,7 +28,6 @@ import java.io.BufferedWriter; import java.io.File; import java.io.FileInputStream; -import java.io.FileNotFoundException; import java.io.FileOutputStream; import java.io.IOException; import java.io.InputStream; @@ -47,6 +46,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.nuiton.diswork.fs.DisworkFileSystem; +import org.nuiton.diswork.fs.DisworkFileSystemException; import org.nuiton.util.FileUtil; import org.nuiton.util.ZipUtil; @@ -161,7 +161,7 @@ this.output = output; this.outputFile = outputFile; } - + @Override public void run() { InputStreamReader osr = new InputStreamReader(output); @@ -189,6 +189,7 @@ } } } catch (IOException e) { + // FIXME 20100701 bleny throw exception log.warn("error while reading the output of the subprocess", e); } finally { try { @@ -210,9 +211,12 @@ * this method add a line to a job-specific log * @param jobPath the path to the job concerned * @param message the line to add the the log - * @throws IOException if an error occurred while writing the log + * @throws DisworkFileSystemException if an error occurred while writing + * the log + * @throws IOException */ - protected void log(String jobPath, String message) throws IOException { + protected void log(String jobPath, String message) + throws DisworkFileSystemException, IOException { String logPath = jobPath + "/" + DisworkDaemon.LOG_PATH; InputStream oldLogAsStream = fileSystem.read(logPath); String oldLog = IOUtils.toString(oldLogAsStream); @@ -221,24 +225,6 @@ fileSystem.write(logPath, IOUtils.toInputStream(newLog)); } - protected void streamCopy(InputStream in, OutputStream out) throws IOException { - System.out.println("hop"); - int read = 0; - int totalRead = 0; - byte[] buffer = new byte[50000]; - // BufferedInputStream hop = new BufferedInputStream(in); - - while ((read = in.read(buffer)) != -1) { - - totalRead += read; - System.out.println("read " + read + " bytes (total=" + totalRead + ")"); - - out.write(buffer, 0, read); - } - in.close(); - out.close(); - } - /** * Download all the files needed for a job in a temp directory, run * the job, wait for it to end, write all the results. Mark the job @@ -246,20 +232,20 @@ * the end, depending of the results * @param jobPath * @return + * @throws DisworkFileSystemException * @throws IOException + * @throws DisworkException */ - protected boolean runJob(String jobPath) throws IOException { + protected boolean runJob(String jobPath) + throws IOException, + DisworkFileSystemException, + DisworkException { 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; - } + String jsdlPath = jobPath + "/" + DisworkDaemon.JSDL_PATH; + String jsdl = IOUtils.toString(fileSystem.read(jsdlPath)); + log.info("read jsdl " + jsdl); JobDescription jobDescription = JobDescription.parseJSDL(jsdl); @@ -287,8 +273,7 @@ log.info("will create " + application.getAbsolutePath()); OutputStream out = new FileOutputStream(application); log.debug("starting copy of " + applicationData.available() + " bytes"); - // IOUtils.copy(applicationData, out); - streamCopy(applicationData, out); + IOUtils.copy(applicationData, out); log.info("unzip application start"); // unzip application ZipUtil.uncompress(application, jobDir); @@ -298,14 +283,14 @@ } // staging input files - for (String fileName : jobDescription.getStagingInput()) { + for (String fileName : jobDescription.getInput()) { log.info("staging " + fileName); File localCopy = new File(jobDir, fileName); localCopy.createNewFile(); InputStream source = null; - if (jobDescription.getStagingInputUrls().containsKey(fileName)) { + if (jobDescription.getInputUrls().containsKey(fileName)) { // download this file from URL - URL url = jobDescription.getStagingInputUrls().get(fileName); + URL url = jobDescription.getInputUrls().get(fileName); log.info("downloading from " + url); source = url.openStream(); } else { @@ -316,8 +301,10 @@ } log.info("preparing the job"); - // prepare and run it the job - String commandLine = jobDescription.getCommandLine(); + // prepare the job and run it + String commandLine = config.parseCommandLine( + jobDescription.getCommandLine(), + jobDir.getAbsolutePath()); String[] commandLineElements = commandLine.split(" "); ProcessBuilder builder = new ProcessBuilder(commandLineElements); builder.directory(jobDir); @@ -325,7 +312,7 @@ log.info("calling " + commandLine); Process job = builder.start(); - // dump the standard output in a file + // start a thread to constantly read on the standard output String standardOutputFileName = jobDescription.getStandardOutput(); log.info("standardOutputFileName is " + standardOutputFileName); OutputStream outputFileStream = null; @@ -335,7 +322,8 @@ outputFileStream = new FileOutputStream(outputFile); } - OutputReader outputReader = new OutputReader(job.getInputStream(), outputFileStream); + OutputReader outputReader = new OutputReader(job.getInputStream(), + outputFileStream); outputReader.start(); // plugging a file on the standard input @@ -354,18 +342,15 @@ log.info("waiting for the end of the process"); exitValue = job.waitFor(); } catch (InterruptedException e) { - log.error("job " + jobDescription + " was interupted", e); + log.error("job " + jobDescription + " was interrupted", e); // FIXME 20100611 bleny job is considered has failed - exitValue = 1; + exitValue = 1; } - - - log.info("job returned " + exitValue); - + // output file staging - for (String fileName : jobDescription.getStagingOutput()) { + for (String fileName : jobDescription.getOutput()) { log.info("staging file " + fileName); File localCopy = new File(jobDir, fileName); // FIXME 20100616 bleny may not exists if job has fail @@ -397,8 +382,10 @@ /** * In a directory, list the content, sort the content, and returns * the first element. + * @throws DisworkFileSystemException */ - protected String getFistJobName(String path) throws IOException { + protected String getFistJobName(String path) + throws DisworkFileSystemException { List<String> jobsNames = fileSystem.readDirectory(path); if (jobsNames.size() == 0) { return null; @@ -408,114 +395,110 @@ } } - protected void findAJobAndRunIt() { - // try to find a new job - try { + protected void findAJobAndRunIt() throws IOException, + DisworkFileSystemException, + DisworkException { - // Once a job is found, those two var will be set - String jobLinkDir = null; - String jobLinkName = null; + // Once a job is found, those two var will be set + String jobLinkDir = null; + String jobLinkName = null; + + // use a synchronized block because multiple workers + // may try to take a same job + synchronized (fileSystem) { - // use a synchronized block because multiple workers - // may try to take a same job - synchronized (fileSystem) { - - // fist, try to find a job declared has running since - // too long to re-run it - String[] runningJobsDirs = { DisworkDaemon.FAILED_2_RUNNING, - DisworkDaemon.FAILED_1_RUNNING, - DisworkDaemon.TODO_RUNNING - }; - // browsing all "running" dirs - for (String path : runningJobsDirs) { - String oldName = getFistJobName(path); - if (oldName != null) { - Long linkAge = System.currentTimeMillis() - - Long.parseLong(oldName); - // check is oldest job is too old and should be - // considered has to-be-rerun - if (linkAge > MAX_JOB_RUNNING_TIME) { - log.info("taking old job (age = " + linkAge + ")"); - jobLinkDir = path; - jobLinkName = oldName; - // FIXME 20100617 bleny break s*cks - break; - } + // fist, try to find a job declared has running since + // too long to re-run it + String[] runningJobsDirs = { DisworkDaemon.FAILED_2_RUNNING, + DisworkDaemon.FAILED_1_RUNNING, + DisworkDaemon.TODO_RUNNING + }; + // browsing all "running" dirs + for (String path : runningJobsDirs) { + String oldName = getFistJobName(path); + if (oldName != null) { + Long linkAge = System.currentTimeMillis() + - Long.parseLong(oldName); + // check is oldest job is too old and should be + // considered has to-be-rerun + if (linkAge > MAX_JOB_RUNNING_TIME) { + log.info("taking old job (age = " + linkAge + ")"); + jobLinkDir = path; + jobLinkName = oldName; + // FIXME 20100617 bleny break s*cks + break; } } - - // if no job was found, search now in not running jobs - if (jobLinkDir == null) { - String[] jobsDirs = { DisworkDaemon.FAILED_2, - DisworkDaemon.FAILED_1, - DisworkDaemon.TODO - }; - for (String path : jobsDirs) { - String oldName = getFistJobName(path); - if (oldName != null) { // take it - jobLinkDir = path; - jobLinkName = oldName; - // FIXME 20100617 bleny break s*cks - break; - } + } + + // if no job was found, search now in not running jobs + if (jobLinkDir == null) { + String[] jobsDirs = { DisworkDaemon.FAILED_2, + DisworkDaemon.FAILED_1, + DisworkDaemon.TODO + }; + for (String path : jobsDirs) { + String oldName = getFistJobName(path); + if (oldName != null) { // take it + jobLinkDir = path; + jobLinkName = oldName; + // FIXME 20100617 bleny break s*cks + break; } } - - if (jobLinkDir != null) { - // move the link before running the job - String oldPath = jobLinkDir + "/" + jobLinkName; - log.info("job found at " + oldPath); + } + + if (jobLinkDir != null) { + // move the link before running the job + String oldPath = jobLinkDir + "/" + jobLinkName; + log.info("job found at " + oldPath); - jobLinkDir = RUNNING_MOVE.get(jobLinkDir); - jobLinkName = DisworkDaemon.newJobLinkName(); - String newPath = jobLinkDir + "/" + jobLinkName; - - log.info("moving " + oldPath + " to " + newPath); - fileSystem.move(oldPath, newPath); - - } + jobLinkDir = RUNNING_MOVE.get(jobLinkDir); + jobLinkName = DisworkDaemon.newJobLinkName(); + String newPath = jobLinkDir + "/" + jobLinkName; + + log.info("moving " + oldPath + " to " + newPath); + fileSystem.move(oldPath, newPath); + } - - // now, if no job was found, do nothing - if (jobLinkDir == null) { - log.info("nothing to do"); + } + + // now, if no job was found, do nothing + if (jobLinkDir == null) { + log.info("nothing to do"); + try { Thread.sleep(JOB_WAIT); + } catch (InterruptedException e) { + log.warn("worker interrupted while waiting", e); + // TODO 20100629 bleny ? + } // if a job was found, take it - } else { + } else { - String jobPath = jobLinkDir + "/" + jobLinkName; - - boolean jobSuccess = runJob(jobPath); - - // move the link after the job - String newDir = null; - if (jobSuccess) { - newDir = DisworkDaemon.DONE; - } else { - newDir = FAILED_MOVE.get(jobLinkDir); - } - String newPath = newDir + "/" + DisworkDaemon.newJobLinkName(); - - log.info("moving " + jobPath + " to " + newPath); - fileSystem.move(jobPath, 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"); - } + String jobPath = jobLinkDir + "/" + jobLinkName; + + boolean jobSuccess = runJob(jobPath); + + // move the link after the job + String newDir = null; + if (jobSuccess) { + newDir = DisworkDaemon.DONE; + } else { + newDir = FAILED_MOVE.get(jobLinkDir); } + String newPath = newDir + "/" + DisworkDaemon.newJobLinkName(); + log.info("moving " + jobPath + " to " + newPath); + fileSystem.move(jobPath, newPath); - } 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 - } + // 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"); + } + } } /** @@ -525,7 +508,18 @@ public void run() { while (! shouldStop) { if (activityStrategy.canWork()) { - findAJobAndRunIt(); + try { + findAJobAndRunIt(); + } catch (DisworkException e) { + log.error("worker error " + e); + throw new RuntimeException("worker error " + e); + } catch (IOException e) { + log.error("worker error " + e); + throw new RuntimeException("worker error " + e); + } catch (DisworkFileSystemException e) { + log.error("worker error " + e); + throw new RuntimeException("worker error " + e); + } } } } @@ -537,7 +531,7 @@ // initialize activityStrategy according to config String initialStrategy = config.getActivityStrategy(); - if ( "none".equals(initialStrategy)) { + if ("none".equals(initialStrategy)) { activeNoActivityStrategy(); } else if ("unlimited".equals(initialStrategy)) { activeUnlimitedActivityStrategy(); @@ -577,9 +571,8 @@ try { Thread.sleep(10 * 1000); } catch (InterruptedException e) { - // TODO 20100615 bleny Auto-generated catch block - log.info("exception catch", e); - e.printStackTrace(); + log.warn("interrupted while waiting for a worker to " + + "stop", e); } } } @@ -595,19 +588,19 @@ } public void activeNoActivityStrategy() { - activityStrategy = new ActivityStrategy.NoActivity(); + setActivityStrategy(new ActivityStrategy.NoActivity()); } public void activeUnlimitedActivityStrategy() { - activityStrategy = new ActivityStrategy.UnlimitedActivity(); + setActivityStrategy(new ActivityStrategy.UnlimitedActivity()); } public void activeLimitedActivityStrategy() { - activityStrategy = new ActivityStrategy.LimitedActivity(); + setActivityStrategy(new ActivityStrategy.LimitedActivity()); } public void activeScheduledActivityStrategy() { - activityStrategy = new ActivityStrategy.ScheduledActivity(); + setActivityStrategy(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-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/DisworkDaemonTest.java 2010-07-01 10:45:20 UTC (rev 94) @@ -93,9 +93,9 @@ * <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> + * of the job;</li> + * <li>standard output is asked has results.</li> + * </ul> * */ @Test Modified: trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/JobDescriptionTest.java =================================================================== --- trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/JobDescriptionTest.java 2010-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-daemon/src/test/java/org/nuiton/diswork/daemon/JobDescriptionTest.java 2010-07-01 10:45:20 UTC (rev 94) @@ -79,16 +79,16 @@ assertEquals(job2.getStandardOutput(), job2Copy.getStandardOutput()); - assertTrue(ListUtils.isEqualList(job2.getStagingInput(), - job2Copy.getStagingInput())); - assertTrue(ListUtils.isEqualList(job2.getStagingOutput(), - job2Copy.getStagingOutput())); + assertTrue(ListUtils.isEqualList(job2.getInput(), + job2Copy.getInput())); + assertTrue(ListUtils.isEqualList(job2.getOutput(), + job2Copy.getOutput())); assertTrue(ListUtils.isEqualList( - job2.getStagingInputUrls().keySet(), - job2Copy.getStagingInputUrls().keySet())); + job2.getInputUrls().keySet(), + job2Copy.getInputUrls().keySet())); assertTrue(ListUtils.isEqualList( - job2.getStagingInputUrls().values(), - job2Copy.getStagingInputUrls().values())); + job2.getInputUrls().values(), + job2Copy.getInputUrls().values())); } catch (IOException e) { fail(); throw e; Modified: trunk/diswork-fs/pom.xml =================================================================== --- trunk/diswork-fs/pom.xml 2010-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-fs/pom.xml 2010-07-01 10:45:20 UTC (rev 94) @@ -11,7 +11,7 @@ <artifactId>diswork-fs</artifactId> <packaging>jar</packaging> - <name>disworkfs</name> + <name>Diswork File-System</name> <dependencies> <dependency> <groupId>commons-logging</groupId> @@ -42,7 +42,6 @@ <artifactId>koala-xmlstore</artifactId> </dependency> - <!-- test --> <dependency> <groupId>junit</groupId> 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-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/Demo.java 2010-07-01 10:45:20 UTC (rev 94) @@ -88,6 +88,9 @@ } else { fileSystem.createDirectory("/todo"); } + } catch (DisworkFileSystemException e) { + log.error(e); + System.out.println("error occured " + e); } catch (IOException e) { log.error(e); System.out.println("error occured " + e); @@ -145,6 +148,9 @@ } catch (IOException e) { log.error(e); System.out.println("error occured " + e); + } catch (DisworkFileSystemException e) { + log.error(e); + System.out.println("error occured " + e); } try { Thread.sleep(10 * 1000); @@ -187,7 +193,7 @@ * @param args * @throws IOException */ - public static void main(String[] args) throws IOException { + public static void main(String[] args) throws Exception { if (args.length == 2) { DisworkFileSystemConfig config = DisworkFileSystemConfig.newKademliaDisworkConfig(); 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-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystem.java 2010-07-01 10:45:20 UTC (rev 94) @@ -34,6 +34,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.nuiton.diswork.fs.DisworkFileSystemException.Type; import org.nuiton.diswork.fs.storage.DisworkMap; import org.nuiton.diswork.fs.storage.EntryUtil; import org.nuiton.diswork.fs.storage.Storage; @@ -59,6 +60,9 @@ private static final Log log = LogFactory.getLog(DisworkFileSystem.class); + /** the symbol to use to separate directories in a path */ + public static final String separator = EntryUtil.PATH_SEPARATOR; + /** storage will permit to save and read directories, files and links */ protected Storage storage; @@ -71,7 +75,7 @@ /** default constructor (uses default configuration parameters) * @throws IOException caused by network issue */ - public DisworkFileSystem() throws IOException { + public DisworkFileSystem() throws DisworkFileSystemException { this(new DisworkFileSystemConfig()); } @@ -81,7 +85,7 @@ * @throws IOException caused by network issue */ public DisworkFileSystem(DisworkFileSystemConfig config) - throws IOException { + throws DisworkFileSystemException { storage = new Storage(config); } @@ -93,7 +97,7 @@ * @return true is something (a link, a file, or a directory) exists at path * @throws IOException */ - public boolean exists(String path) throws IOException { + public boolean exists(String path) throws DisworkFileSystemException { checkPathSyntax(path); String entry = walk(path); boolean result = entry != null; @@ -108,13 +112,12 @@ * @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 { + public InputStream read(String path) throws DisworkFileSystemException { checkPathSyntax(path); String entry = walk(path); if (entry == null) { - throw new FileNotFoundException(path); + throw new DisworkFileSystemException(Type.NO_SUCH_FILE, path); } InputStream result = null; @@ -126,15 +129,13 @@ String newTarget = EntryUtil.resolveLink(path, link); result = read(newTarget); } else if (EntryUtil.isDirectory(entry)) { - throw new IOException("target is not a file: " + path); + throw new DisworkFileSystemException(Type.NOT_FILE, "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; } @@ -146,24 +147,24 @@ * @throws ConcurrentModificationException if file is already being written */ public void write(String path, InputStream source) - throws IOException, - ConcurrentModificationException { + throws DisworkFileSystemException { checkPathSyntax(path); if (source == null) { - throw new NullPointerException("source stream is null"); + throw new DisworkFileSystemException(Type.READ_LOCAL_DATA_FAILURE, + "source stream is null"); } 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 { + throws DisworkFileSystemException { String entryParent = walk(parent); if (entryParent == null) { - throw new IOException(parent + " directory doesn't exists"); + throw new DisworkFileSystemException(Type.NO_SUCH_DIRECTORY, parent + + " directory doesn't exists"); } if (EntryUtil.isDirectory(entryParent)) { @@ -207,8 +208,8 @@ Thread.sleep(LOCK_WAIT); } catch (InterruptedException e) { log.info("wait for lock interrupted", e); - throw new IOException - ("interrupted while trying to acquire lock", e); + throw new DisworkFileSystemException(Type.INTERRUPTION, + "interrupted while trying to acquire lock", e); } } } @@ -224,10 +225,11 @@ String newTarget = EntryUtil.resolveLink(parent, linkTarget); write(newTarget, fileName, source); } else if (EntryUtil.isFile(entryParent)) { - throw new IOException(parent + " is not a directory"); + throw new DisworkFileSystemException(Type.NOT_DIRECTORY, parent); } else { log.warn("strange entry" + entryParent); - throw new IOException("strange entry" + entryParent); + throw new DisworkFileSystemException(Type.CORRUPTED_DATA, + "strange entry" + entryParent); } } @@ -236,9 +238,7 @@ * @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 { + public void createDirectory(String path) throws DisworkFileSystemException { checkPathSyntax(path); String parent = EntryUtil.getParentFromPath(path); String dirName = EntryUtil.getNameFromPath(path); @@ -246,14 +246,14 @@ } protected void createDirectory(String parent, String dirName) - throws IOException { + throws DisworkFileSystemException { log.info("trying to create directory " + dirName + " in " + parent); String entryParent = walk(parent); if (entryParent == null) { - throw new IOException(parent + " directory doesn't exists"); + throw new DisworkFileSystemException(Type.NO_SUCH_DIRECTORY, parent); } if (EntryUtil.isDirectory(entryParent)) { @@ -264,8 +264,8 @@ String findResult = EntryUtil.findEntryInDirectory (content, dirName); if (findResult != null) { - throw new IOException - (parent + " already contains an element named " + dirName); + throw new DisworkFileSystemException(Type.ALREADY_EXISTS, parent + + " already contains an element named " + dirName); } // store file before meta info @@ -291,8 +291,8 @@ Thread.sleep(LOCK_WAIT); } catch (InterruptedException e) { log.info("wait for lock interrupted", e); - throw new IOException - ("interrupted while trying to acquire lock", e); + throw new DisworkFileSystemException(Type.INTERRUPTION, + "interrupted while trying to acquire lock", e); } } } @@ -308,10 +308,11 @@ String newTarget = EntryUtil.resolveLink(parent, linkTarget); createDirectory(newTarget, dirName); } else if (EntryUtil.isFile(entryParent)) { - throw new IOException(parent + " is not a directory"); + throw new DisworkFileSystemException(Type.NOT_DIRECTORY, parent); } else { log.warn("strange entry" + entryParent); - throw new IOException("strange entry" + entryParent); + throw new DisworkFileSystemException(Type.CORRUPTED_DATA, + "strange entry" + entryParent); } } @@ -320,11 +321,10 @@ * @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 + * @throws DisworkFileSystemException */ public void createSymbolicLink(String path, String target) - throws IOException, - ConcurrentModificationException { + throws DisworkFileSystemException { checkPathSyntax(path); String parent = EntryUtil.getParentFromPath(path); String name = EntryUtil.getNameFromPath(path); @@ -335,7 +335,7 @@ * @see #createSymbolicLink(String, String) */ protected void createSymbolicLink(String parent, String name, String target) - throws IOException { + throws DisworkFileSystemException { if (target.startsWith(EntryUtil.PATH_SEPARATOR)) { // target is absolute @@ -352,7 +352,8 @@ String entryParent = walk(parent); if (entryParent == null) { - throw new IOException(parent + " directory doesn't exists"); + throw new DisworkFileSystemException(Type.NO_SUCH_DIRECTORY, + parent); } if (EntryUtil.isDirectory(entryParent)) { @@ -364,8 +365,8 @@ String findResult = EntryUtil.findEntryInDirectory (content, name); if (findResult != null) { - throw new IOException - (parent + " already contains an element named " + name); + throw new DisworkFileSystemException(Type.ALREADY_EXISTS, + parent + " already contains an element named " + name); } // store file before meta info @@ -390,8 +391,8 @@ Thread.sleep(LOCK_WAIT); } catch (InterruptedException e) { log.info("wait for lock interrupted", e); - throw new IOException - ("interrupted while trying to acquire lock", e); + throw new DisworkFileSystemException(Type.INTERRUPTION, + "interrupted while trying to acquire lock", e); } } } @@ -408,15 +409,16 @@ String newTarget = EntryUtil.resolveLink(parent, linkTarget); createSymbolicLink(newTarget, name, target); } else if (EntryUtil.isFile(entryParent)) { - throw new IOException(parent + " is not a directory"); + throw new DisworkFileSystemException(Type.NOT_DIRECTORY, parent); } else { log.warn("strange entry" + entryParent); - throw new IOException("strange entry" + entryParent); + throw new DisworkFileSystemException(Type.CORRUPTED_DATA, + "strange entry" + entryParent); } } else { - throw new IOException(target + " is not a valid target"); + throw new DisworkFileSystemException(Type.INVALID_TARGET, target); } } @@ -424,13 +426,12 @@ * 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 + * @throws DisworkFileSystemException */ - public void delete(String path) throws IOException, - ConcurrentModificationException { + public void delete(String path) throws DisworkFileSystemException { checkPathSyntax(path); if (!exists(path)) { - throw new IOException(path + " doesn't exists"); + throw new DisworkFileSystemException(Type.NO_SUCH_ENTITY, path); } String parent = EntryUtil.getParentFromPath(path); String name = EntryUtil.getNameFromPath(path); @@ -441,13 +442,10 @@ /** * @see #delete(String) */ - protected void delete(String parent, String name) throws IOException { + protected void delete(String parent, String name) + throws DisworkFileSystemException { 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); @@ -465,8 +463,8 @@ if (!innerDirectoryContent.equals( EntryUtil.EMPTY_DIRECTORY_CONTENT)) { // directory is not empty - throw new IOException - ("trying to remove a non-empty directory"); + throw new DisworkFileSystemException(Type.DIRECTORY_NOT_EMPTY, + "trying to remove a non-empty directory"); } // remove it @@ -478,7 +476,8 @@ storage.removeLink(idToRemove); } else { log.warn("strange entry" + entryParent); - throw new IOException("strange entry" + entryParent); + throw new DisworkFileSystemException(Type.CORRUPTED_DATA, + "strange entry" + entryParent); } @@ -499,8 +498,8 @@ Thread.sleep(LOCK_WAIT); } catch (InterruptedException e) { log.info("wait for lock interrupted", e); - throw new IOException - ("interrupted while trying to acquire lock", e); + throw new DisworkFileSystemException(Type.INTERRUPTION, + "interrupted while trying to acquire lock", e); } } } @@ -517,10 +516,11 @@ String newTarget = EntryUtil.resolveLink(parent, linkTarget); delete(newTarget, name); } else if (EntryUtil.isFile(entryParent)) { - throw new IOException(parent + " is not a directory"); + throw new DisworkFileSystemException(Type.NOT_DIRECTORY, parent); } else { log.warn("strange entry" + entryParent); - throw new IOException("strange entry" + entryParent); + throw new DisworkFileSystemException(Type.CORRUPTED_DATA, + "strange entry" + entryParent); } } @@ -528,15 +528,16 @@ * 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 + * @throws DisworkFileSystemException */ - public List<String> readDirectory(String path) throws IOException { + public List<String> readDirectory(String path) + throws DisworkFileSystemException { checkPathSyntax(path); String entry = walk(path); List<String> result = null; if (entry == null) { - throw new IOException(path + " doesn't exists"); + throw new DisworkFileSystemException(Type.NO_SUCH_DIRECTORY, path); } else { // path may be a link, if it's the case, // entry become the actual directory @@ -558,10 +559,12 @@ } } } else if (EntryUtil.isFile(entry)) { - throw new IOException(path + " is not a directory but a file"); + throw new DisworkFileSystemException(Type.NOT_DIRECTORY, path + + " is not a directory but a file"); } else { log.warn("strange entry" + entry); - throw new IOException("strange entry" + entry); + throw new DisworkFileSystemException(Type.CORRUPTED_DATA, + "strange entry" + entry); } } @@ -575,7 +578,7 @@ * @param path * @return null if path is not valid */ - protected String walk(String path) throws IOException { + protected String walk(String path) throws DisworkFileSystemException { String result = walk(path, null, null); log.info("walking to " + path + " returns " + result); return result; @@ -592,7 +595,7 @@ * @throws IOException */ protected String walk(String path, String current, String content) - throws IOException { + throws DisworkFileSystemException { // FIXME 20105021 bleny works fine but is not understandable String result = null; @@ -693,9 +696,11 @@ * check a path is absolute and syntactically correct, throw exception if * that's not the case. */ - protected void checkPathSyntax(String path) throws IOException { + protected void checkPathSyntax(String path) + throws DisworkFileSystemException { if (!path.startsWith(EntryUtil.ROOT_DIRECTORY)) { - throw new IOException("\"" + path + "\" is not correct, all pathes " + + throw new DisworkFileSystemException(Type.INVALID_PATH, + "\"" + path + "\" is not correct, all pathes " + "have to be absolute (thus, starts with)" + EntryUtil.ROOT_DIRECTORY); } @@ -703,7 +708,8 @@ String doubleSeparator = EntryUtil.PATH_SEPARATOR + EntryUtil.PATH_SEPARATOR; if (path.contains(doubleSeparator)) { - throw new IOException("\"" + path + "\" is not correct, it contains " + throw new DisworkFileSystemException(Type.INVALID_PATH, + "\"" + path + "\" is not correct, it contains " + doubleSeparator); } } @@ -712,14 +718,16 @@ * 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 + * @throws DisworkFileSystemException */ - public void move(String path, String destination) throws IOException { + public void move(String path, String destination) + throws DisworkFileSystemException { checkPathSyntax(path); checkPathSyntax(destination); if (exists(destination)) { - throw new IOException(destination + " already exists"); + throw new DisworkFileSystemException(Type.ALREADY_EXISTS, + destination + " already exists"); } String pathParent = EntryUtil.getParentFromPath(path); @@ -732,20 +740,22 @@ } protected void move(String pathParent, String pathName, - String destinationParent, String destinationName) throws IOException { + String destinationParent, String destinationName) + throws DisworkFileSystemException { String entryParent = walk(pathParent); if (entryParent == null) { - throw new IOException(pathParent + " directory doesn't exists"); + throw new DisworkFileSystemException(Type.NO_SUCH_DIRECTORY, + pathParent); } if (EntryUtil.isDirectory(entryParent)) { String destinationParentEntry = walk(destinationParent); if (destinationParentEntry == null) { - throw new IOException(destinationParentEntry + " directory" - + "doesn't exists"); + throw new DisworkFileSystemException(Type.NO_SUCH_DIRECTORY, + destinationParentEntry); } if (EntryUtil.isDirectory(destinationParentEntry)) { @@ -757,8 +767,8 @@ String oldEntry = EntryUtil.findEntryInDirectory(parentContent, pathName); if (oldEntry == null) { - throw new IOException("no element " + pathName + " in " + - pathParent); + throw new DisworkFileSystemException(Type.NO_SUCH_ENTITY, + "no element " + pathName + " in " + pathParent); } String parentId = EntryUtil.getIdFromEntry(entryParent); @@ -792,8 +802,8 @@ Thread.sleep(LOCK_WAIT); } catch (InterruptedException e) { log.info("wait for lock interrupted", e); - throw new IOException - ("interrupted while trying to acquire lock", e); + throw new DisworkFileSystemException(Type.INTERRUPTION, + "interrupted while trying to acquire lock", e); } } } @@ -810,10 +820,11 @@ String newTarget = EntryUtil.resolveLink(destinationParent, linkTarget); move(newTarget, pathName, destinationParent, destinationName); } else if (EntryUtil.isFile(destinationParentEntry)) { - throw new IOException(destinationParent + " is not a directory"); + throw new DisworkFileSystemException(Type.NOT_DIRECTORY, destinationParent); } else { log.warn("strange entry" + destinationParentEntry); - throw new IOException("strange entry" + destinationParentEntry); + throw new DisworkFileSystemException(Type.CORRUPTED_DATA, + "strange entry" + destinationParentEntry); } } else if (EntryUtil.isLink(entryParent)) { @@ -822,17 +833,17 @@ String newTarget = EntryUtil.resolveLink(pathParent, linkTarget); move(pathParent, pathName, newTarget, destinationName); } else if (EntryUtil.isFile(entryParent)) { - throw new IOException(pathParent + " is not a directory"); + throw new DisworkFileSystemException(Type.NOT_DIRECTORY, pathParent); } else { log.warn("strange entry" + entryParent); - throw new IOException("strange entry" + entryParent); + throw new DisworkFileSystemException(Type.CORRUPTED_DATA, + "strange entry" + entryParent); } } public void createDirectories(String path) - throws ConcurrentModificationException, - IOException { + throws DisworkFileSystemException { log.info("trying create directories for " + path); String pathWithoutRoot = path.substring(1, path.length()); Copied: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemException.java (from rev 92, trunk/diswork-daemon/src/main/java/org/nuiton/diswork/daemon/DisworkFileSystemException.java) =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemException.java (rev 0) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/DisworkFileSystemException.java 2010-07-01 10:45:20 UTC (rev 94) @@ -0,0 +1,97 @@ +/* + * #%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.fs; + +/** + * + * A port to Java 1.7 should use java.nio.file.FileSystemException and + * sub-classes. + * + * @author bleny + */ +public class DisworkFileSystemException extends Exception { + + private static final long serialVersionUID = 1L; + + public static enum Type { + /** Something with this path already exists */ + ALREADY_EXISTS, + /** Error while writing because another node is writing at this place */ + CONCURRENT_MODIFICATION, + /** String do not describe a valid path */ + INVALID_PATH, + /** The directory has a content */ + DIRECTORY_NOT_EMPTY, + /** directory doesn't exists */ + NO_SUCH_DIRECTORY, + /** file doesn't exists */ + NO_SUCH_FILE, + /** nothing at a given path */ + NO_SUCH_ENTITY, + /** The path leads to something that exists but is not a directory */ + NOT_DIRECTORY, + /** The path leads to something that exists but is not a file */ + NOT_FILE, + /** The connection to the DHT failed + * this is due to a network failure or a bad configuration + */ + BOOTSTRAP_FAILURE, + /** an IO errors at low-level network occured */ + NETWORK_FAILURE, + /** a source data on the local OS is not readable */ + READ_LOCAL_DATA_FAILURE, + /** error while reading data in the map */ + READ_MAP_DATA_FAILURE, + /** a process was interrupted */ + INTERRUPTION, + /** if map contains corrupted data */ + CORRUPTED_DATA, + /** the symlink point to a target that is not valid */ + INVALID_TARGET + } + + protected Type type; + + public DisworkFileSystemException(Type type, String message, + Throwable cause) { + super(message, cause); + this.type = type; + } + + public DisworkFileSystemException(Type type, String message) { + super(message); + this.type = type; + } + + public DisworkFileSystemException(Type type, Throwable cause) { + super(cause); + this.type = type; + } + + public Type getType() { + return type; + } + +} 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-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/KademliaDisworkMap.java 2010-07-01 10:45:20 UTC (rev 94) @@ -38,6 +38,8 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.nuiton.diswork.fs.DisworkFileSystemConfig; +import org.nuiton.diswork.fs.DisworkFileSystemException; +import org.nuiton.diswork.fs.DisworkFileSystemException.Type; import org.planx.xmlstore.routing.Identifier; import org.planx.xmlstore.routing.Kademlia; import org.planx.xmlstore.routing.RoutingException; @@ -48,28 +50,35 @@ protected Kademlia kad; - public KademliaDisworkMap(DisworkFileSystemConfig config) throws IOException { + public KademliaDisworkMap(DisworkFileSystemConfig config) + throws DisworkFileSystemException { - log.info("booting 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); - throw e; + try { + log.info("booting 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); + throw new DisworkFileSystemException(Type.BOOTSTRAP_FAILURE, + "bootstrap node is unreachable", e); + } } + + log.info("kademlia status : " + kad); + + //Identifier.IDSIZE = 1024; + log.info("using " + Identifier.IDSIZE + " bytes for identifiers"); + } catch (IOException e) { + throw new DisworkFileSystemException(Type.NETWORK_FAILURE, + "failure while bootstraping", e); } - - log.info("kademlia status : " + kad); - - //Identifier.IDSIZE = 1024; - log.info("using " + Identifier.IDSIZE + " bytes for identifiers"); } protected static Identifier stringToIdentifier(String s) { Modified: trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/PastryDisworkMap.java =================================================================== --- trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/PastryDisworkMap.java 2010-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/PastryDisworkMap.java 2010-07-01 10:45:20 UTC (rev 94) @@ -37,6 +37,8 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.nuiton.diswork.fs.DisworkFileSystemConfig; +import org.nuiton.diswork.fs.DisworkFileSystemException; +import org.nuiton.diswork.fs.DisworkFileSystemException.Type; import rice.Continuation; import rice.environment.Environment; @@ -118,104 +120,107 @@ protected PastryIdFactory pastryIdFactory; - public PastryDisworkMap(DisworkFileSystemConfig config) throws IOException { + public PastryDisworkMap(DisworkFileSystemConfig config) + throws DisworkFileSystemException { + + try { + Environment env = new Environment(); - Environment env = new Environment(); + // disable the UPnP setting (in case you are testing this on a NATted LAN) + // env.getParameters().setString("nat_search_policy", "never"); - // disable the UPnP setting (in case you are testing this on a NATted LAN) - // env.getParameters().setString("nat_search_policy", "never"); + // the port to use locally + int bindport = config.getUsedPort(); - // the port to use locally - int bindport = config.getUsedPort(); + InetAddress bootaddr = InetAddress.getByName(config.getBootstrapIp()); + int bootport = config.getBootstrapPort(); - InetAddress bootaddr = InetAddress.getByName(config.getBootstrapIp()); - int bootport = config.getBootstrapPort(); + InetSocketAddress bootaddress = new InetSocketAddress(bootaddr, bootport); - InetSocketAddress bootaddress = new InetSocketAddress(bootaddr, bootport); + log.info("boot address : " + bootaddress); - log.info("boot address : " + bootaddress); + NodeIdFactory nidFactory = new RandomNodeIdFactory(env); - NodeIdFactory nidFactory = new RandomNodeIdFactory(env); - - // construct the PastryNodeFactory, this is how we use rice.pastry.socket - PastryNodeFactory factory = null; + // construct the PastryNodeFactory, this is how we use rice.pastry.socket + PastryNodeFactory factory = null; - // FIXME 20100602 bleny this is a work-around to deal with time out on - // connect exceptions that occurs "sometimes" - int numberOfTry = 0; - while(factory == null && numberOfTry <= 10) { - try { - factory = new SocketPastryNodeFactory(nidFactory, bindport, env); - } catch (BindException e) { - // the bootstrap node can't be joined - throw new IOException(""/*"bootstrap node can't be joined"*/, e); - } catch (IllegalStateException e) { - // the bootstrap node can't be joined - throw new IOException - ("unable to bind to already used port " + bindport, e); - } catch (ConnectException e) { - // this occurs some times, it may work after some time... - numberOfTry += 1; + // FIXME 20100602 bleny this is a work-around to deal with time out on + // connect exceptions that occurs "sometimes" + int numberOfTry = 0; + while(factory == null && numberOfTry <= 10) { try { - Thread.sleep(10 * 1000); - } catch (InterruptedException ee) { - throw new IOException("Pastry boot process interrupted", ee); + factory = new SocketPastryNodeFactory(nidFactory, bindport, env); + } catch (BindException e) { + // the bootstrap node can't be joined + throw new IOException(""/*"bootstrap node can't be joined"*/, e); + } catch (IllegalStateException e) { + // the bootstrap node can't be joined + throw new IOException + ("unable to bind to already used port " + bindport, e); + } catch (ConnectException e) { + // this occurs some times, it may work after some time... + numberOfTry += 1; + try { + Thread.sleep(10 * 1000); + } catch (InterruptedException ee) { + throw new IOException("Pastry boot process interrupted", ee); + } } } - } - if (factory == null) { - throw new IOException("unable to connect"); - } - - // construct a node, but this does not cause it to boot - PastryNode node = factory.newNode(); - - // in later tutorials, we will register applications before calling boot - node.boot(bootaddress); - - // the node may require sending several messages to fully boot into the ring - synchronized(node) { - while(!node.isReady() && !node.joinFailed()) { - // delay so we don't busy-wait - try { - node.wait(500); - } catch (InterruptedException e) { - throw new IOException("Pastry boot process interrupted", e); + if (factory == null) { + throw new IOException("unable to connect"); + } + + // construct a node, but this does not cause it to boot + PastryNode node = factory.newNode(); + + // in later tutorials, we will register applications before calling boot + node.boot(bootaddress); + + // the node may require sending several messages to fully boot into the ring + synchronized(node) { + while(!node.isReady() && !node.joinFailed()) { + // delay so we don't busy-wait + try { + node.wait(500); + } catch (InterruptedException e) { + throw new IOException("Pastry boot process interrupted", e); + } + + // abort if can't join + if (node.joinFailed()) { + throw new IOException("Could not join the FreePastry ring. " + + "Reason : " + node.joinFailedReason()); + } } - - // abort if can't join - if (node.joinFailed()) { - throw new IOException("Could not join the FreePastry ring. " - + "Reason : " + node.joinFailedReason()); - } } + + log.info("finished creating new node " + node); + + pastryIdFactory = new PastryIdFactory(env); + + rice.persistence.Storage storage = new MemoryStorage(pastryIdFactory); + past = new PastImpl( + node, + new StorageManagerImpl( + pastryIdFactory, + storage, + // Using a cache do not permit to remove while putting + /* + new LRUCache( + new MemoryStorage(pastryIdFactory), + 512 * 1024, + node.getEnvironment() + ) + */ + new EmptyCache(pastryIdFactory) + ), + 0, // replication factor + "" + ); + } catch (IOException e) { + throw new DisworkFileSystemException(Type.NETWORK_FAILURE, e); } - - log.info("finished creating new node " + node); - - pastryIdFactory = new PastryIdFactory(env); - - rice.persistence.Storage storage = new MemoryStorage(pastryIdFactory); - past = new PastImpl( - node, - new StorageManagerImpl( - pastryIdFactory, - storage, - // Using a cache do not permit to remove while putting - /* - new LRUCache( - new MemoryStorage(pastryIdFactory), - 512 * 1024, - node.getEnvironment() - ) - */ - new EmptyCache(pastryIdFactory) - ), - 0, // replication factor - "" - ); - - } protected class ContainsKeyContinuation extends 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-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-fs/src/main/java/org/nuiton/diswork/fs/storage/Storage.java 2010-07-01 10:45:20 UTC (rev 94) @@ -29,7 +29,6 @@ import java.io.IOException; import java.io.InputStream; import java.util.ArrayList; -import java.util.ConcurrentModificationException; import java.util.List; import java.util.Map; @@ -37,6 +36,8 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.nuiton.diswork.fs.DisworkFileSystemConfig; +import org.nuiton.diswork.fs.DisworkFileSystemException; +import org.nuiton.diswork.fs.DisworkFileSystemException.Type; /** * This class is the middle layer between the File System operations and @@ -106,8 +107,14 @@ if (blockIdsIndex < blockIds.length) { // closing previously opened stream IOUtils.closeQuietly(currentBlock); - currentBlock = new ByteArrayInputStream( - map.get(blockIds[blockIdsIndex])); + byte[] data = map.get(blockIds[blockIdsIndex]); + if (data == null) { + // maybe data has not been posted or DHT replication + // failed + log.error("a block is missing into the map"); + throw new IOException("missing block"); + } + currentBlock = new ByteArrayInputStream(data); log.debug("new current block (size = " + currentBlock.available() + ")"); } else { @@ -154,8 +161,15 @@ protected DisworkFileSystemConfig config; + /** + * + * @param config an instance of + * @param null at normal use. Will set map according to the content of + * config + * @throws IOException if network fail + */ public Storage(DisworkFileSystemConfig config, DisworkMap map) - throws IOException { + throws DisworkFileSystemException { this.config = config; @@ -191,7 +205,8 @@ } } - public Storage(DisworkFileSystemConfig config) throws IOException { + public Storage(DisworkFileSystemConfig config) + throws DisworkFileSystemException { this(config, null); } @@ -199,7 +214,7 @@ * @return the content (entries) of the root directory * @throws IOException */ - public String getRootDirectory() throws IOException { + public String getRootDirectory() throws DisworkFileSystemException { String result = getDirectory(EntryUtil.ROOT_DIRECTORY); return result; } @@ -211,11 +226,15 @@ * @return a the entries of the directory * @throws IOException */ - public String getDirectory(String id) throws IOException { + public String getDirectory(String id) throws DisworkFileSystemException { InputStream in = get(id); - String content = IOUtils.toString(in); - log.debug("getDirectory(\"" + id + "\") returns \n" + content); - return content; + try { + String content = IOUtils.toString(in); + log.debug("getDirectory(\"" + id + "\") returns \n" + content); + return content; + } catch (IOException e) { + throw new DisworkFileSystemException(Type.READ_MAP_DATA_FAILURE, e); + } } public InputStream getFile(String id) { @@ -223,22 +242,28 @@ return result; } - public String getLink(String id) throws IOException { + public String getLink(String id) { String content = EntryUtil.bytesToString(map.get(id)); - log.debug("getDirectory(\"" + id + "\") returns \"" + content + "\""); + log.debug("getLink(\"" + id + "\") returns \"" + content + "\""); return content; } - public void putDirectory(String id, String content) throws IOException { + public void putDirectory(String id, String content) + throws DisworkFileSystemException { log.debug("putDirectory(\"" + id + "\", \"" + content + "\")"); InputStream value = IOUtils.toInputStream(content); put(id, value); } - public void putFile(String id, InputStream content) throws IOException { - log.debug("putFile(\"" + id + "\", stream of "+ content.available() + - " bytes)"); - put(id, content); + public void putFile(String id, InputStream content) + throws DisworkFileSystemException { + try { + log.debug("putFile(\"" + id + "\", stream of "+ content.available() + + " bytes)"); + put(id, content); + } catch (IOException e) { + throw new DisworkFileSystemException(Type.READ_LOCAL_DATA_FAILURE, e); + } } public void putLink(String id, String content) { @@ -247,12 +272,12 @@ map.put(id, contentAsBytes); } - public void removeDirectory(String id) throws IOException { + public void removeDirectory(String id) throws DisworkFileSystemException { log.debug("removeDirectory(\"" + id + "\")"); remove(id); } - public void removeFile(String id) throws IOException { + public void removeFile(String id) throws DisworkFileSystemException { log.debug("removeFile(\"" + id + "\")"); remove(id); } @@ -277,79 +302,82 @@ * Lock is released at the end * @param key * @param value - * @throws IOException + * @throws DisworkFileSystemException */ protected void put(String key, InputStream value) - throws IOException, - ConcurrentModificationException { + throws DisworkFileSystemException { boolean lockAcquired = tryToLock(key); if (lockAcquired) { - log.info("lock on " + key + " acquired"); + try { + log.info("lock on " + key + " acquired"); - // here, we know we can write or an exception - // would have been thrown earlier - String blocksIds = ""; - int readResult = 0; - int totalSize = 0; + // here, we know we can write or an exception + // would have been thrown earlier + String blocksIds = ""; + int readResult = 0; + int totalSize = 0; - String metaBlock = totalSize + blocksIds; - String newDataKey = keyToNewDataKey(key); - map.put(newDataKey, EntryUtil.stringToBytes(metaBlock)); - - // creating a buffer of the size of a block - int bufferSize = config.getBlockSize(); - byte[] buffer = new byte[bufferSize]; - - while ((readResult = value.read(buffer)) != -1) { - totalSize += readResult; - - byte[] newBlock = buffer; - - // if the buffer is not full, truncate the block - if (readResult < buffer.length) { - newBlock = new byte[readResult]; - System.arraycopy(buffer, 0, newBlock, 0, readResult); + String metaBlock = totalSize + blocksIds; + String newDataKey = keyToNewDataKey(key); + map.put(newDataKey, EntryUtil.stringToBytes(metaBlock)); + + // creating a buffer of the size of a block + int bufferSize = config.getBlockSize(); + byte[] buffer = new byte[bufferSize]; + + while ((readResult = value.read(buffer)) != -1) { + totalSize += readResult; + + byte[] newBlock = buffer; + + // if the buffer is not full, truncate the block + if (readResult < buffer.length) { + newBlock = new byte[readResult]; + System.arraycopy(buffer, 0, newBlock, 0, readResult); + } + + String id = EntryUtil.generateId(); + blocksIds += EntryUtil.BLOCKIDS_SEPARATOR + id; + + log.debug("saving new block (size = " + newBlock.length + + ") at key " + id); + + // copy block in map + map.put(id, newBlock); + + metaBlock = totalSize + blocksIds; + + log.debug("putting metablock " + metaBlock + " at key " + + key); + + map.put(newDataKey, EntryUtil.stringToBytes(metaBlock)); + + // updating lock + updateLock(key); } - - String id = EntryUtil.generateId(); - blocksIds += EntryUtil.BLOCKIDS_SEPARATOR + id; - - log.debug("saving new block (size = " + newBlock.length + ")" - + " at key " + id); - - // copy block in map - map.put(id, newBlock); - - metaBlock = totalSize + blocksIds; - - log.debug("putting metablock " + metaBlock + " at key " + key); - map.put(newDataKey, EntryUtil.stringToBytes(metaBlock)); + eraseDependenciesOfMetablock(key); - // updating lock - updateLock(key); + if (isLockStillOwned(key)) { + map.put(key, map.get(newDataKey)); + } + + map.remove(newDataKey); + + unLock(key); + } catch (IOException e) { + throw new DisworkFileSystemException + (Type.READ_LOCAL_DATA_FAILURE, "can't read value", e); } - eraseDependenciesOfMetablock(key); - - if (isLockStillOwned(key)) { - map.put(key, map.get(newDataKey)); - } - - map.remove(newDataKey); - - unLock(key); - } else { - throw new ConcurrentModificationException("key " + key + - " is locked"); + throw new DisworkFileSystemException(Type.CONCURRENT_MODIFICATION, + "key " + key + " is locked"); } - } - protected void remove(String key) throws IOException, - ConcurrentModificationException { + protected void remove(String key) throws DisworkFileSystemException { eraseDependenciesOfMetablock(key); removeKey(key); } @@ -396,7 +424,7 @@ if (lock == null || (EntryUtil.getOwnerFromLock(lock).equals(ownerId))) { - // file is not locked, we have took the lock + // file was not locked, we have took the lock result = true; } else { // file is locked by other node, check date @@ -464,6 +492,9 @@ return lockAge > LOCK_VALID_TIME; } + /** + * a setter to inject the map dependency + */ public void setMap(DisworkMap map) { this.map = map; } 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-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/AbstractDisworkFileSystemTest.java 2010-07-01 10:45:20 UTC (rev 94) @@ -9,8 +9,6 @@ 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; @@ -21,6 +19,7 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.nuiton.diswork.fs.DisworkFileSystemException.Type; import org.nuiton.util.FileUtil; @@ -42,7 +41,7 @@ /** * The file will have this fixed size */ - static protected int randomFileSize = 45 * 1000 * 1000; + static protected int randomFileSize = 15 * 1000 * 1000; static protected DisworkFileSystem fileSystem; @@ -112,18 +111,22 @@ /** * try to read a file that as never been created nor written */ - @Test(expected = FileNotFoundException.class) + @Test public void testFailAtRead() throws Exception { - fileSystem.read("/not_existing_file"); + try { + fileSystem.read("/not_existing_file"); + fail(); + } catch (DisworkFileSystemException e) { + assertEquals(Type.NO_SUCH_FILE, e.getType()); + } } /** * 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 { + public void testSplit() throws Exception { byte[] bytes = new byte[1]; bytes[0] = -0x1; @@ -196,12 +199,15 @@ * this use case should raise an exception because my_folder * doesn't exists */ - @Test(expected = IOException.class) + @Test public void testFailAtWrite() throws Exception { InputStream source = null; try { source = new FileInputStream(randomFilePath); fileSystem.write("/my_folder/my_file", source); + fail(); + } catch (DisworkFileSystemException e ){ + assertEquals(Type.NO_SUCH_DIRECTORY, e.getType()); } finally { IOUtils.closeQuietly(source); } @@ -303,9 +309,14 @@ * Trying to create a link to a wrong target, this sould raise an exception * @throws Exception */ - @Test(expected = IOException.class) + @Test public void testFailAtLinking() throws Exception { - fileSystem.createSymbolicLink("/my_link", "/wrong_target_path"); + try { + fileSystem.createSymbolicLink("/my_link", "/wrong_target_path"); + fail(); + } catch (DisworkFileSystemException e) { + assertEquals(Type.INVALID_TARGET, e.getType()); + } } /** @@ -335,7 +346,7 @@ * exception * @throws Exception */ - @Test(expected = IOException.class) + @Test public void testFailAtRemove() throws Exception { fileSystem.createDirectory("/my_folder"); @@ -348,7 +359,12 @@ } // trying to remove a non-empty directory should raise an exception - fileSystem.delete("/my_folder"); + try { + fileSystem.delete("/my_folder"); + fail(); + } catch (DisworkFileSystemException e) { + assertEquals(Type.DIRECTORY_NOT_EMPTY, e.getType()); + } } /** @@ -469,11 +485,15 @@ */ } - @Test(expected = IOException.class) + @Test public void failAtMove() throws Exception { fileSystem.createDirectory("/dir"); fileSystem.createDirectory("/dir2"); - fileSystem.move("/dir", "/dir2"); + try { + fileSystem.move("/dir", "/dir2"); + } catch (DisworkFileSystemException e) { + assertEquals(Type.ALREADY_EXISTS, e.getType()); + } } @Test @@ -481,7 +501,7 @@ try { fileSystem.createDirectories("/dir/subdir/subsubdir"); assertTrue(fileSystem.exists("/dir/subdir/subsubdir")); - } catch (IOException e) { + } catch (DisworkFileSystemException e) { fail(); } } 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-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/DisworkFileSystemKademliaTest.java 2010-07-01 10:45:20 UTC (rev 94) @@ -1,21 +1,16 @@ 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.io.FileInputStream; import java.io.InputStream; import java.util.ConcurrentModificationException; import org.apache.commons.io.IOUtils; -import org.apache.log4j.lf5.util.StreamUtils; import org.junit.Before; import org.junit.Test; -import org.nuiton.diswork.fs.DisworkFileSystem; -import org.nuiton.diswork.fs.DisworkFileSystemConfig; public class DisworkFileSystemKademliaTest extends AbstractDisworkFileSystemTest { @@ -90,15 +85,15 @@ @Test public void testMultipleNodes3() throws Exception { - + DisworkFileSystemConfig disworkConfig = + DisworkFileSystemConfig.newKademliaDisworkConfig(bootstrapPort); + DisworkFileSystem fileSystem2 = new DisworkFileSystem(disworkConfig); + fileSystem.createDirectory("/mydir"); try { - fileSystem.createDirectory("/myseconddir"); + fileSystem2.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/diswork/fs/storage/AbstractDisworkMapTest.java 2010-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/AbstractDisworkMapTest.java 2010-07-01 10:45:20 UTC (rev 94) @@ -14,7 +14,6 @@ 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/KademliaDisworkMapTest.java =================================================================== --- trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/KademliaDisworkMapTest.java 2010-07-01 10:38:20 UTC (rev 93) +++ trunk/diswork-fs/src/test/java/org/nuiton/diswork/fs/storage/KademliaDisworkMapTest.java 2010-07-01 10:45:20 UTC (rev 94) @@ -2,6 +2,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.fail; import java.util.ArrayList; import java.util.HashSet; @@ -11,6 +12,8 @@ import org.junit.Before; import org.junit.Test; import org.nuiton.diswork.fs.DisworkFileSystemConfig; +import org.nuiton.diswork.fs.DisworkFileSystemException; +import org.nuiton.diswork.fs.DisworkFileSystemException.Type; import org.planx.xmlstore.routing.Identifier; public class KademliaDisworkMapTest extends AbstractDisworkMapTest { @@ -30,11 +33,11 @@ * this tests the key generation algorithm, it has different properties to * check : * * two call with same param should return same key - * * two call with different parameter should return differents keys + * * two call with different parameter should return different keys * * a call should never return null */ @Test - public void testStringToIndentifier() { + public void testStringToIdentifier() { List<String> paths = new ArrayList<String>(); @@ -66,18 +69,21 @@ } /** - * Tests that in exception is raised when attempting to connect using + * Tests that an exception is raised when attempting to connect using * a bad bootstrap * @throws Exception */ - /* - @Test(expected = org.planx.xmlstore.routing.RoutingException.class) + @Test public void testBadBootrap() throws Exception { DisworkFileSystemConfig config1 = DisworkFileSystemConfig.newKademliaDisworkConfig(); config1.setBootstrapIp("microsoft.com"); config1.setBootstrapPort(80); - new KademliaDisworkMap(config1); + try { + new KademliaDisworkMap(config1); + fail(); + } catch (DisworkFileSystemException e) { + assertEquals(Type.BOOTSTRAP_FAILURE, e.getType()); + } } - */ }