parallel 0.6.4

Command-line CPU load balancer for executing jobs in parallel
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

use std::fs::{read_dir, create_dir_all, remove_file, remove_dir, File};
use std::io::{StderrLock, Write};
use std::path::{Path, PathBuf};
use std::process::exit;

use super::arguments::{FileErr, Args};
use super::filepaths;
use super::input_iterator::InputIterator;

fn remove_preexisting_files() -> Result<(PathBuf, PathBuf, PathBuf), FileErr> {
    // Initialize the base directories of the unprocessed and processed files.
    let path = filepaths::base().ok_or(FileErr::Path)?;

    // Create the directories that are required for storing input files.
    create_dir_all(&path).map_err(|why| FileErr::DirectoryCreate(path.clone(), why))?;

    // Attempt to obtain a listing of all the directories and files within the base directory.
    let directory = read_dir(&path).map_err(|why| FileErr::DirectoryRead(path.clone(), why))?;
    for entry in directory {
        let entry = entry.map_err(|why| FileErr::DirectoryRead(path.clone(), why))?;
        let entry_is_file = entry.file_type().ok().map_or(true, |x| !x.is_dir());
        if entry_is_file {
            remove_file(entry.path()).map_err(|why| FileErr::Remove(path.clone(), why))?;
        } else {
            remove_dir(entry.path()).map_err(|why| FileErr::Remove(path.clone(), why))?;
        }
    }

    // Create empty logs ahead of time
    let errors      = filepaths::errors().ok_or(FileErr::Path)?;
    let unprocessed = filepaths::unprocessed().ok_or(FileErr::Path)?;
    let processed   = filepaths::processed().ok_or(FileErr::Path)?;
    File::create(&errors).map_err(|why| FileErr::Create(errors.clone(), why))?;
    File::create(&unprocessed).map_err(|why| FileErr::Create(unprocessed.clone(), why))?;
    File::create(&processed).map_err(|why| FileErr::Create(processed.clone(), why))?;

    Ok((unprocessed, processed, errors))
}

pub fn cleanup(stderr: &mut StderrLock) -> (PathBuf, PathBuf, PathBuf) {
    match remove_preexisting_files() {
        Ok(values) => values,
        Err(why) => {
            let _ = stderr.write(b"parallel: initialization error: I/O error: ");
            match why {
                FileErr::Create(path, why) => {
                    let _ = write!(stderr, "unable to create file: {:?}: {}", path, why);
                }
                FileErr::DirectoryCreate(path, why) => {
                    let _ = write!(stderr, "unable to create directory: {:?}: {}", path, why);
                },
                FileErr::DirectoryRead(path, why) => {
                    let _ = write!(stderr, "unable to read directory: {:?}: {}", path, why);
                }
                FileErr::Remove(path, why) => {
                    let _ = write!(stderr, "unable to remove file: {:?}: {}", path, why);
                },
                _ => unreachable!()
            }
            exit(1)
        }
    }
}

pub fn parse(args: &mut Args, comm: &mut String, unprocessed: &Path) -> InputIterator {
    match args.parse(comm, unprocessed) {
        Ok(inputs) => {
            args.ninputs = inputs.total_arguments;
            inputs
        },
        Err(why) => why.handle()
    }
}