Repository: pkolaczk/fclones (language: rust)

Issue:
Allow to use --cache with --transform
Currently caching is not usable with `--transform`. 
This limitation hurts a lot because `--transform` is actually quite slow, so it would benefit a lot from caching.



Discussion / hints:

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index b480a63..b36f5e9 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -369,6 +369,7 @@ dependencies = [
  "sysinfo",
  "tempfile",
  "thread_local",
+ "typed-sled",
  "uuid",
  "winapi",
  "winapi-util",
@@ -815,6 +816,26 @@ dependencies = [
  "regex",
 ]
 
+[[package]]
+name = "pin-project"
+version = "1.0.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "58ad3879ad3baf4e44784bc6a718a8698867bb991f8ce24d1bcbe2cfb4c3a75e"
+dependencies = [
+ "pin-project-internal",
+]
+
+[[package]]
+name = "pin-project-internal"
+version = "1.0.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "744b6f092ba29c3650faf274db506afd39944f48420f6c86b17cfe0ee1cb36bb"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "ppv-lite86"
 version = "0.2.16"
@@ -1219,6 +1240,19 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "typed-sled"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1362aa15e9df65c77830762d81cbd92bc49f1a315bf17ef4b595539c99ee87c2"
+dependencies = [
+ "bincode",
+ "pin-project",
+ "serde",
+ "sled",
+ "thiserror",
+]
+
 [[package]]
 name = "unicode-segmentation"
 version = "1.9.0"
diff --git a/Cargo.toml b/Cargo.toml
index 0d55fe0..5ee1c9d 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -55,6 +55,7 @@ stfu8 = "0.2"
 structopt = "0.3"
 sysinfo = "0.23.10"
 thread_local = "1.1"
+typed-sled = "0.2.0"
 uuid = { version = "0.8", features = ["v4"] }
 
 [target.'cfg(target_os = "linux")'.dependencies]
diff --git a/src/cache.rs b/src/cache.rs
index 01aff59..8699845 100644
--- a/src/cache.rs
+++ b/src/cache.rs
@@ -5,32 +5,30 @@ use std::fs::create_dir_all;
 use std::time::{Duration, UNIX_EPOCH};
 
 use serde::{Deserialize, Serialize};
-use sled::IVec;
 
 use crate::error::Error;
-use crate::file::{FileChunk, FileHash, FileLen, FileMetadata, FilePos};
+use crate::file::{FileChunk, FileHash, FileId, FileLen, FileMetadata, FilePos};
 use crate::hasher::HashAlgorithm;
 use crate::path::Path;
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct Key {
-    file_id: u128,
-    device_id: u64,
+    file_id: FileId,
     chunk_pos: FilePos,
     chunk_len: FileLen,
-    algorithm: HashAlgorithm,
 }
 
 impl Display for Key {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}:{}", self.device_id, self.file_id)
+        write!(f, "{}:{}", self.file_id.device, self.file_id.inode)
     }
 }
 
 #[derive(Debug, Serialize, Deserialize)]
 struct CachedFileInfo {
-    modified_timestamp_us: u128,
-    len: FileLen,
+    modified_timestamp_ms: u64,
+    file_len: FileLen,
+    data_len: FileLen,
     hash: FileHash,
 }
 
@@ -40,13 +38,17 @@ struct CachedFileInfo {
 /// Usually it is a lot faster to retrieve the hash from an embedded database that to compute
 /// them from file data.
 pub struct HashCache {
-    cache: sled::Db,
+    cache: typed_sled::Tree<Key, CachedFileInfo>,
 }
 
 impl HashCache {
     /// Opens the file hash database located in the given directory.
     /// If the database doesn't exist yet, creates a new one.
-    pub fn open(database_path: &Path) -> Result<HashCache, Error> {
+    pub fn open(
+        database_path: &Path,
+        transform: Option<&str>,
+        algorithm: HashAlgorithm,
+    ) -> Result<HashCache, Error> {
         create_dir_all(&database_path.to_path_buf()).map_err(|e| {
             format!(
                 "Count not create hash database directory {}: {}",
@@ -54,42 +56,53 @@ impl HashCache {
                 e
             )
         })?;
-        let cache = sled::open(&database_path.to_path_buf()).map_err(|e| {
+        let db = sled::open(&database_path.to_path_buf()).map_err(|e| {
             format!(
                 "Failed to open hash database at {}: {}",
                 database_path.to_escaped_string(),
                 e
             )
         })?;
+
+        let tree_id = format!("hash_db:{:?}:{}", algorithm, transform.unwrap_or("<none>"));
+        let cache = typed_sled::Tree::open(&db, tree_id);
         Ok(HashCache { cache })
     }
 
     /// Opens the file hash database located in `fclones` subdir of user cache directory.
     /// If the database doesn't exist yet, creates a new one.
-    pub fn open_default() -> Result<HashCache, Error> {
+    pub fn open_default(
+        transform: Option<&str>,
+        algorithm: HashAlgorithm,
+    ) -> Result<HashCache, Error> {
         let cache_dir =
             dirs::cache_dir().ok_or("Could not obtain user cache directory from the system.")?;
         let hash_db_path = cache_dir.join("fclones");
-        Self::open(&Path::from(hash_db_path))
+        Self::open(&Path::from(hash_db_path), transform, algorithm)
     }
 
     /// Stores the file hash plus some file metadata in the cache.
-    pub fn put(&self, key: &Key, file: &FileMetadata, hash: FileHash) -> Result<(), Error> {
+    pub fn put(
+        &self,
+        key: &Key,
+        file: &FileMetadata,
+        data_len: FileLen,
+        hash: FileHash,
+    ) -> Result<(), Error> {
         let value = CachedFileInfo {
-            modified_timestamp_us: file
+            modified_timestamp_ms: file
                 .modified()
                 .map_err(|e| format!("Unable to get file modification timestamp: {}", e))?
                 .duration_since(UNIX_EPOCH)
                 .unwrap_or(Duration::ZERO)
-                .as_micros(),
-            len: file.len(),
+                .as_millis() as u64,
+            file_len: file.len(),
+            data_len,
             hash,
         };
 
-        let key = bincode::serialize(&key).unwrap();
-        let value = bincode::serialize(&value).unwrap();
         self.cache
-            .insert(key, value)
+            .insert(key, &value)
             .map_err(|e| format!("Failed to write entry to cache: {}", e))?;
         Ok(())
     }
@@ -99,9 +112,12 @@ impl HashCache {
     /// Returns `Ok(None)` if file is not present in the cache or if its current length
     /// or its current modification time do not match the file length and modification time
     /// recorded at insertion time.
-    pub fn get(&self, key: &Key, metadata: &FileMetadata) -> Result<Option<FileHash>, Error> {
-        let key = bincode::serialize(&key).unwrap();
-        let value: Option<IVec> = self
+    pub fn get(
+        &self,
+        key: &Key,
+        metadata: &FileMetadata,
+    ) -> Result<Option<(FileLen, FileHash)>, Error> {
+        let value = self
             .cache
             .get(key)
             .map_err(|e| format!("Failed to retrieve entry from cache: {}", e))?;
@@ -109,20 +125,18 @@ impl HashCache {
             Some(v) => v,
             None => return Ok(None), // not found in cache
         };
-        let value: CachedFileInfo = bincode::deserialize(&value)
-            .map_err(|e| format!("Failed to deserialize value from cache: {}", e))?;
 
         let modified = metadata
             .modified()
             .map_err(|e| format!("Unable to get file modification timestamp: {}", e))?
             .duration_since(UNIX_EPOCH)
             .unwrap_or(Duration::ZERO)
-            .as_micros();
+            .as_millis() as u64;
 
-        if value.modified_timestamp_us != modified || value.len != metadata.len() {
+        if value.modified_timestamp_ms != modified || value.file_len != metadata.len() {
             Ok(None) // found in cache, but the file has changed since it was cached
         } else {
-            Ok(Some(value.hash))
+            Ok(Some((value.data_len, value.hash)))
         }
     }
 
@@ -130,18 +144,11 @@ impl HashCache {
     ///
     /// Using file identifiers as cache keys instead of paths allows the user for moving or renaming
     /// files without losing their cached hash data.
-    pub fn key(
-        &self,
-        chunk: &FileChunk<'_>,
-        metadata: &FileMetadata,
-        algorithm: HashAlgorithm,
-    ) -> Result<Key, Error> {
+    pub fn key(&self, chunk: &FileChunk<'_>, metadata: &FileMetadata) -> Result<Key, Error> {
         let key = Key {
-            file_id: metadata.inode_id() as u128,
-            device_id: metadata.device_id(),
+            file_id: metadata.file_id(),
             chunk_pos: chunk.pos,
             chunk_len: chunk.len,
-            algorithm,
         };
         Ok(key)
     }
@@ -167,16 +174,15 @@ mod test {
             let chunk = FileChunk::new(&path, FilePos(0), FileLen(1000));
 
             let cache_path = Path::from(root.join("cache"));
-            let cache = HashCache::open(&cache_path).unwrap();
-            let key = cache
-                .key(&chunk, &metadata, HashAlgorithm::MetroHash128)
-                .unwrap();
+            let cache = HashCache::open(&cache_path, None, HashAlgorithm::MetroHash128).unwrap();
+            let key = cache.key(&chunk, &metadata).unwrap();
             let orig_hash = FileHash(12345);
 
-            cache.put(&key, &metadata, orig_hash).unwrap();
+            let data_len = FileLen(200);
+            cache.put(&key, &metadata, data_len, orig_hash).unwrap();
             let cached_hash = cache.get(&key, &metadata).unwrap();
 
-            assert_eq!(cached_hash, Some(orig_hash))
+            assert_eq!(cached_hash, Some((data_len, orig_hash)))
         });
     }
 
@@ -190,11 +196,11 @@ mod test {
             let chunk = FileChunk::new(&path, FilePos(0), FileLen(1000));
 
             let cache_path = Path::from(root.join("cache"));
-            let cache = HashCache::open(&cache_path).unwrap();
-            let key = cache
-                .key(&chunk, &metadata, HashAlgorithm::MetroHash128)
+            let cache = HashCache::open(&cache_path, None, HashAlgorithm::MetroHash128).unwrap();
+            let key = cache.key(&chunk, &metadata).unwrap();
+            cache
+                .put(&key, &metadata, chunk.len, FileHash(12345))
                 .unwrap();
-            cache.put(&key, &metadata, FileHash(12345)).unwrap();
 
             // modify the file
             use std::io::Write;
@@ -222,20 +228,50 @@ mod test {
             let chunk = FileChunk::new(&path, FilePos(0), FileLen(1000));
 
             let cache_path = Path::from(root.join("cache"));
-            let cache = HashCache::open(&cache_path).unwrap();
-            let key = cache
-                .key(&chunk, &metadata, HashAlgorithm::MetroHash128)
-                .unwrap();
+            let cache = HashCache::open(&cache_path, None, HashAlgorithm::MetroHash128).unwrap();
+            let key = cache.key(&chunk, &metadata).unwrap();
 
-            cache.put(&key, &metadata, FileHash(12345)).unwrap();
+            cache
+                .put(&key, &metadata, chunk.len, FileHash(12345))
+                .unwrap();
 
             let chunk = FileChunk::new(&path, FilePos(1000), FileLen(2000));
-            let key = cache
-                .key(&chunk, &metadata, HashAlgorithm::MetroHash128)
-                .unwrap();
+            let key = cache.key(&chunk, &metadata).unwrap();
             let cached_hash = cache.get(&key, &metadata).unwrap();
 
             assert_eq!(cached_hash, None)
         });
     }
+
+    #[test]
+    fn return_none_if_different_transform_was_used() {
+        with_dir(
+            "cache/return_none_if_different_transform_was_used",
+            |root| {
+                let path = root.join("file");
+                create_file(&path);
+                let path = Path::from(&path);
+                let metadata = FileMetadata::new(&path).unwrap();
+                let chunk = FileChunk::new(&path, FilePos(0), FileLen(1000));
+
+                let cache_path = Path::from(root.join("cache"));
+                let cache =
+                    HashCache::open(&cache_path, None, HashAlgorithm::MetroHash128).unwrap();
+                let key = cache.key(&chunk, &metadata).unwrap();
+
+                let orig_hash = FileHash(12345);
+                let data_len = FileLen(200);
+                cache.put(&key, &metadata, data_len, orig_hash).unwrap();
+                let cached_hash = cache.get(&key, &metadata).unwrap();
+                assert_eq!(cached_hash, Some((data_len, orig_hash)));
+                drop(cache); // unlock the db so we can open another cache
+
+                let cache =
+                    HashCache::open(&cache_path, Some("transform"), HashAlgorithm::MetroHash128)
+                        .unwrap();
+                let cached_hash = cache.get(&key, &metadata).unwrap();
+                assert_eq!(cached_hash, None);
+            },
+        );
+    }
 }
diff --git a/src/config.rs b/src/config.rs
index d7f88f1..f6ec0c0 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -198,7 +198,7 @@ pub struct GroupConfig {
     /// specified by $OUT and will read output from there.
     /// If the program modifies the original file in-place without writing to the standard output
     /// nor a distinct file, use --in-place flag.
-    #[structopt(long, value_name("command"), conflicts_with("cache"))]
+    #[structopt(long, value_name("command"))]
     pub transform: Option<String>,
 
     /// Set this flag if the command given to --transform transforms the file in-place,
diff --git a/src/file.rs b/src/file.rs
index 8c15e8f..ced8305 100644
--- a/src/file.rs
+++ b/src/file.rs
@@ -186,7 +186,7 @@ pub type InodeId = u64;
 type InodeId = u128;
 
 /// Useful for identifying files in presence of hardlinks
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
 pub struct FileId {
     pub device: u64,
     pub inode: InodeId,
@@ -282,6 +282,10 @@ impl FileMetadata {
         FileLen(self.metadata.len())
     }
 
+    pub fn file_id(&self) -> FileId {
+        self.id
+    }
+
     pub fn device_id(&self) -> u64 {
         self.id.device
     }
diff --git a/src/group.rs b/src/group.rs
index d6d41dc..2f5a783 100644
--- a/src/group.rs
+++ b/src/group.rs
@@ -27,18 +27,16 @@ use sysinfo::DiskType;
 use thread_local::ThreadLocal;
 
 use crate::arg::Arg;
-use crate::cache::HashCache;
 use crate::config::*;
 use crate::device::{DiskDevice, DiskDevices};
 use crate::error::Error;
 use crate::file::*;
-use crate::hasher::{FileHasher, HashAlgorithm};
+use crate::hasher::FileHasher;
 use crate::log::Log;
 use crate::path::Path;
 use crate::report::{FileStats, ReportHeader, ReportWriter};
 use crate::selector::PathSelector;
 use crate::semaphore::Semaphore;
-use crate::transform::Transform;
 use crate::walk::Walk;
 
 /// Groups items by key.
@@ -105,7 +103,6 @@ struct GroupCtx<'a> {
     pub log: &'a Log,
     group_filter: FileGroupFilter,
     devices: DiskDevices,
-    transform: Option<Transform>,
     path_selector: PathSelector,
     hasher: FileHasher<'a>,
 }
@@ -124,17 +121,10 @@ impl<'a> GroupCtx<'a> {
         let path_selector = config
             .path_selector(&base_dir)
             .map_err(|e| format!("Invalid pattern: {}", e))?;
-
-        let cache: Option<HashCache> = if config.cache {
-            Some(HashCache::open_default()?)
+        let hasher = if config.cache {
+            FileHasher::new_cached(transform, log)?
         } else {
-            None
-        };
-        let hasher = FileHasher {
-            cache,
-            algorithm: HashAlgorithm::MetroHash128,
-            buf_len: 65536,
-            log,
+            FileHasher::new(transform, log)
         };
 
         Self::check_pool_config(thread_pool_sizes, &devices)?;
@@ -144,7 +134,6 @@ impl<'a> GroupCtx<'a> {
             log,
             group_filter,
             devices,
-            transform,
             path_selector,
             hasher,
         })
@@ -809,7 +798,6 @@ fn handle_fetch_physical_location_err(
 /// Transforms files by piping them to an external program and groups them by their hashes
 fn group_transformed(
     ctx: &GroupCtx<'_>,
-    transform: &Transform,
     groups: Vec<FileGroup<FileInfo>>,
 ) -> Vec<FileGroup<FileInfo>> {
     let progress = ctx
@@ -823,12 +811,14 @@ fn group_transformed(
         &ctx.devices,
         FileAccess::Sequential,
         |(fi, _)| {
-            let result = transform
-                .run_or_log_err(&fi.path, ctx.log)
-                .map(|(len, hash)| {
-                    fi.len = len;
-                    hash
-                });
+            let chunk = FileChunk::new(&fi.path, FilePos(0), fi.len);
+            let result =
+                ctx.hasher
+                    .hash_transformed_or_log_err(&chunk, |_| {})
+                    .map(|(len, hash)| {
+                        fi.len = len;
+                        hash
+                    });
             progress.tick();
             result
         },
@@ -894,7 +884,7 @@ fn group_by_prefix(
                 ctx.devices[fi.get_device_index()].min_prefix_len()
             };
             let chunk = FileChunk::new(&fi.path, FilePos(0), prefix_len);
-            ctx.hasher.hash(&chunk, |_| {})
+            ctx.hasher.hash_file_or_log_err(&chunk, |_| {})
         },
     );
 
@@ -945,7 +935,7 @@ fn group_by_suffix(
             progress.tick();
             let chunk = FileChunk::new(&fi.path, fi.len.as_pos() - suffix_len, suffix_len);
             ctx.hasher
-                .hash(&chunk, |_| {})
+                .hash_file_or_log_err(&chunk, |_| {})
                 .map(|new_hash| old_hash ^ new_hash)
         },
     );
@@ -978,7 +968,7 @@ fn group_by_contents(
         |(fi, _)| {
             let chunk = FileChunk::new(&fi.path, FilePos(0), fi.len);
             ctx.hasher
-                .hash(&chunk, |bytes_read| progress.inc(bytes_read))
+                .hash_file_or_log_err(&chunk, |bytes_read| progress.inc(bytes_read))
         },
     );
 
@@ -1067,8 +1057,8 @@ pub fn group_files(config: &GroupConfig, log: &Log) -> Result<Vec<FileGroup<File
     let mut size_groups_pruned = remove_same_files(&ctx, size_groups);
     update_file_locations(&ctx, &mut size_groups_pruned);
 
-    let mut groups = match &ctx.transform {
-        Some(transform) => group_transformed(&ctx, transform, size_groups_pruned),
+    let mut groups = match &ctx.hasher.transform {
+        Some(_transform) => group_transformed(&ctx, size_groups_pruned),
         _ => {
             let prefix_len = prefix_len(&ctx.devices, flat_iter(&size_groups_pruned));
             let prefix_groups = group_by_prefix(&ctx, prefix_len, size_groups_pruned);
diff --git a/src/hasher.rs b/src/hasher.rs
index 80047da..f03edf3 100644
--- a/src/hasher.rs
+++ b/src/hasher.rs
@@ -12,6 +12,8 @@ use crate::cache::{HashCache, Key};
 use crate::file::{FileAccess, FileChunk, FileHash, FileLen, FileMetadata, FilePos};
 use crate::log::Log;
 use crate::path::Path;
+use crate::transform::Transform;
+use crate::Error;
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
 pub enum HashAlgorithm {
@@ -20,34 +22,151 @@ pub enum HashAlgorithm {
 
 /// Hashes file contents
 pub struct FileHasher<'a> {
-    pub(crate) algorithm: HashAlgorithm,
+    pub(crate) _algorithm: HashAlgorithm, // to be used later
     pub(crate) buf_len: usize,
     pub(crate) cache: Option<HashCache>,
+    pub(crate) transform: Option<Transform>,
     pub(crate) log: &'a Log,
 }
 
 impl FileHasher<'_> {
+    /// Creates a hasher with no caching
+    pub fn new(transform: Option<Transform>, log: &Log) -> FileHasher<'_> {
+        FileHasher {
+            _algorithm: HashAlgorithm::MetroHash128,
+            buf_len: 65536,
+            cache: None,
+            transform,
+            log,
+        }
+    }
+
+    /// Creates a default hasher with caching enabled
+    pub fn new_cached(transform: Option<Transform>, log: &Log) -> Result<FileHasher<'_>, Error> {
+        let algorithm = HashAlgorithm::MetroHash128;
+        let transform_command_str = transform.as_ref().map(|t| t.command_str.as_str());
+        let cache = HashCache::open_default(transform_command_str, algorithm)?;
+        Ok(FileHasher {
+            _algorithm: algorithm,
+            buf_len: 65536,
+            cache: Some(cache),
+            transform,
+            log,
+        })
+    }
+
     /// Computes the file hash or logs an error and returns none if failed.
     /// If file is not found, no error is logged and `None` is returned.
-    pub fn hash(&self, chunk: &FileChunk<'_>, progress: impl Fn(usize)) -> Option<FileHash> {
+    pub fn hash_file(
+        &self,
+        chunk: &FileChunk<'_>,
+        progress: impl Fn(usize),
+    ) -> io::Result<FileHash> {
         let cache = self.cache.as_ref();
         let metadata = cache.and_then(|_| FileMetadata::new(chunk.path).ok());
         let metadata = metadata.as_ref();
         let key = cache
             .zip(metadata.as_ref())
-            .and_then(|(c, m)| c.key(chunk, m, self.algorithm).ok());
+            .and_then(|(c, m)| c.key(chunk, m).ok());
         let key = key.as_ref();
         let hash = self.load_hash(key, metadata);
-        if hash.is_some() {
+        if let Some((_, hash)) = hash {
             progress(chunk.len.0 as usize);
-            return hash;
+            return Ok(hash);
         }
 
-        match file_hash(chunk, self.buf_len, progress) {
-            Ok(hash) => {
-                self.store_hash(key, metadata, hash);
-                Some(hash)
+        let hash = file_hash(chunk, self.buf_len, progress)?;
+        self.store_hash(key, metadata, chunk.len, hash);
+        Ok(hash)
+    }
+
+    pub fn hash_file_or_log_err(
+        &self,
+        chunk: &FileChunk<'_>,
+        progress: impl Fn(usize),
+    ) -> Option<FileHash> {
+        match self.hash_file(chunk, progress) {
+            Ok(hash) => Some(hash),
+            Err(e) if e.kind() == ErrorKind::NotFound => None,
+            Err(e) => {
+                self.log.warn(format!(
+                    "Failed to compute hash of file {}: {}",
+                    chunk.path.to_escaped_string(),
+                    e
+                ));
+                None
             }
+        }
+    }
+
+    /// Just like `hash_file`, but transforms the file before hashing.
+    pub fn hash_transformed(
+        &self,
+        chunk: &FileChunk<'_>,
+        progress: impl Fn(usize),
+    ) -> io::Result<(FileLen, FileHash)> {
+        assert_eq!(chunk.pos, FilePos::zero());
+        assert!(self.transform.is_some());
+
+        let transform = self.transform.as_ref().unwrap();
+        let cache = self.cache.as_ref();
+        let metadata = cache.and_then(|_| FileMetadata::new(chunk.path).ok());
+        let metadata = metadata.as_ref();
+        let key = cache
+            .zip(metadata.as_ref())
+            .and_then(|(c, m)| c.key(chunk, m).ok());
+        let key = key.as_ref();
+        let hash = self.load_hash(key, metadata);
+        if let Some(hash) = hash {
+            progress(chunk.len.0 as usize);
+            return Ok(hash);
+        }
+
+        let mut transform_output = transform.run(chunk.path)?;
+        let hash_input = &mut transform_output.out_stream;
+
+        // Transformed file may have a different length, so we cannot use stream_hash progress
+        // reporting, as it would report progress of the transformed stream. Instead we advance
+        // progress after doing the full file.
+        let hash = stream_hash(hash_input, chunk.len, self.buf_len, |_| {});
+        progress(chunk.len.0 as usize);
+
+        let hash = hash?;
+        let exit_status = transform_output.child.lock().unwrap().wait()?;
+        if !exit_status.success() {
+            let captured_err = transform_output
+                .err_stream
+                .take()
+                .unwrap()
+                .join()
+                .unwrap_or_else(|_| "".to_owned());
+            let captured_err = format_output_stream(captured_err.as_str());
+            return match exit_status.code() {
+                Some(exit_code) => Err(io::Error::new(
+                    io::ErrorKind::Other,
+                    format!(
+                        "{} failed with non-zero status code: {}{}",
+                        transform.program, exit_code, captured_err
+                    ),
+                )),
+                None => Err(io::Error::new(
+                    io::ErrorKind::Other,
+                    format!("{} failed{}", transform.program, captured_err),
+                )),
+            };
+        }
+
+        self.store_hash(key, metadata, hash.0, hash.1);
+        Ok(hash)
+    }
+
+    pub fn hash_transformed_or_log_err(
+        &self,
+        chunk: &FileChunk<'_>,
+        progress: impl Fn(usize),
+    ) -> Option<(FileLen, FileHash)> {
+        match self.hash_transformed(chunk, progress) {
+            Ok(hash) => Some(hash),
             Err(e) if e.kind() == ErrorKind::NotFound => None,
             Err(e) => {
                 self.log.warn(format!(
@@ -63,13 +182,17 @@ impl FileHasher<'_> {
     /// Loads hash from the cache.
     /// If the hash is not present in the cache, returns `None`.
     /// If the operation fails (e.g. corrupted cache), logs a warning and returns `None`.
-    fn load_hash(&self, key: Option<&Key>, metadata: Option<&FileMetadata>) -> Option<FileHash> {
+    fn load_hash(
+        &self,
+        key: Option<&Key>,
+        metadata: Option<&FileMetadata>,
+    ) -> Option<(FileLen, FileHash)> {
         self.cache
             .as_ref()
             .zip(key)
             .zip(metadata)
             .and_then(|((cache, key), metadata)| match cache.get(key, metadata) {
-                Ok(hash) => hash,
+                Ok(len_and_hash) => len_and_hash,
                 Err(e) => {
                     self.log.warn(format!(
                         "Failed to load hash of file id = {} from the cache: {}",
@@ -82,11 +205,17 @@ impl FileHasher<'_> {
 
     /// Stores the hash in the cache.
     /// If the operation fails (e.g. no space on drive), logs a warning.
-    fn store_hash(&self, key: Option<&Key>, metadata: Option<&FileMetadata>, hash: FileHash) {
+    fn store_hash(
+        &self,
+        key: Option<&Key>,
+        metadata: Option<&FileMetadata>,
+        data_len: FileLen,
+        hash: FileHash,
+    ) {
         if let Some(((cache, key), metadata)) =
             self.cache.as_ref().zip(key.as_ref()).zip(metadata.as_ref())
         {
-            if let Err(e) = cache.put(key, metadata, hash) {
+            if let Err(e) = cache.put(key, metadata, data_len, hash) {
                 self.log.warn(format!(
                     "Failed to store hash of file {} in the cache: {}",
                     key, e
@@ -96,6 +225,15 @@ impl FileHasher<'_> {
     }
 }
 
+fn format_output_stream(output: &str) -> String {
+    let output = output.trim().to_string();
+    if output.is_empty() {
+        output
+    } else {
+        format!("\n{}\n", output)
+    }
+}
+
 #[cfg(target_os = "linux")]
 fn to_off_t(offset: u64) -> libc::off_t {
     min(libc::off_t::MAX as u64, offset) as libc::off_t
@@ -243,7 +381,7 @@ fn scan<F: FnMut(&[u8])>(
 
 /// Computes the hash value over at most `len` bytes of the stream.
 /// Returns the number of the bytes read and a 128-bit hash value.
-pub(crate) fn stream_hash(
+fn stream_hash(
     stream: &mut impl Read,
     len: FileLen,
     buf_len: usize,
@@ -263,7 +401,7 @@ pub(crate) fn stream_hash(
 /// Computes hash of initial `len` bytes of a file.
 /// If the file does not exist or is not readable, print the error to stderr and return `None`.
 /// The returned hash is not cryptograhically secure.
-pub(crate) fn file_hash(
+fn file_hash(
     chunk: &FileChunk<'_>,
     buf_len: usize,
     progress: impl Fn(usize),
diff --git a/src/transform.rs b/src/transform.rs
index 013f705..96f9177 100644
--- a/src/transform.rs
+++ b/src/transform.rs
@@ -1,10 +1,12 @@
 use std::cell::RefCell;
 use std::ffi::OsString;
-use std::fs::{create_dir_all, remove_dir_all, File};
+use std::fs::{create_dir_all, remove_dir_all, File, OpenOptions};
 use std::io;
 use std::io::Read;
 use std::path::PathBuf;
-use std::process::{Command, ExitStatus, Stdio};
+use std::process::{Child, Command, Stdio};
+use std::sync::{Arc, Mutex};
+use std::thread::JoinHandle;
 
 use nom::branch::alt;
 use nom::bytes::complete::tag;
@@ -17,26 +19,13 @@ use nom::IResult;
 use regex::Regex;
 use uuid::Uuid;
 
-use crate::file::{FileHash, FileLen};
-use crate::hasher::stream_hash;
-use crate::log::Log;
 use crate::path::Path;
 
-const BUF_LEN: usize = 64 * 1024;
-
-/// Keeps the results of the transform program execution
-pub struct Output {
-    output_len: FileLen,
-    output_hash: FileHash,
-    status: ExitStatus,
-    stderr: String,
-}
-
 /// Controls how we pass data to the child process.
 /// By default, the file to process is sent to the standard input of the child process.
 /// Some programs do not accept reading input from the stdin, but prefer to be pointed
 /// to a file by a command-line option - in this case `Named` variant is used.
-enum InputConf {
+enum Input {
     /// Pipe the input file from the given path to the stdin of the child
     StdIn(PathBuf),
     /// Pass the original path to the file as $IN param
@@ -45,33 +34,35 @@ enum InputConf {
     Copied(PathBuf, PathBuf),
 }
 
-impl InputConf {
+impl Input {
     fn input_path(&self) -> &PathBuf {
         match self {
-            InputConf::StdIn(path) => path,
-            InputConf::Named(path) => path,
-            InputConf::Copied(_src, target) => target,
+            Input::StdIn(path) => path,
+            Input::Named(path) => path,
+            Input::Copied(_src, target) => target,
         }
     }
 
     fn prepare_input_file(&self) -> io::Result<()> {
         match self {
-            InputConf::StdIn(_path) => Ok(()),
-            InputConf::Named(_path) => Ok(()),
-            InputConf::Copied(src, target) => {
+            Input::StdIn(_path) => Ok(()),
+            Input::Named(_path) => Ok(()),
+            Input::Copied(src, target) => {
                 std::fs::copy(src, target)?;
                 Ok(())
             }
         }
     }
+}
 
+impl Drop for Input {
     /// Removes the temporary file if it was created
-    fn cleanup(&self) -> io::Result<()> {
-        match self {
-            InputConf::StdIn(_) => Ok(()),
-            InputConf::Named(_) => Ok(()),
-            InputConf::Copied(_, target) => std::fs::remove_file(target),
-        }
+    fn drop(&mut self) {
+        let _ = match self {
+            Input::StdIn(_) => Ok(()),
+            Input::Named(_) => Ok(()),
+            Input::Copied(_, target) => std::fs::remove_file(target),
+        };
     }
 }
 
@@ -79,7 +70,7 @@ impl InputConf {
 /// By default we read output directly from the standard output of the child process.
 /// If the preprocessor program can't output data to its stdout, but supports only writing
 /// to files, it can be configured to write to a named pipe, and we read from that named pipe.
-enum OutputConf {
+enum Output {
     /// Pipe data directly to StdOut
     StdOut,
     /// Send data through a named pipe
@@ -88,9 +79,32 @@ enum OutputConf {
     InPlace(PathBuf),
 }
 
+impl Output {
+    /// Returns the path to the named pipe if the process is configured to write to a pipe.
+    /// Returns None if the process is configured to write to stdout or to modify the input file.
+    pub fn pipe_path(&self) -> Option<PathBuf> {
+        match &self {
+            Output::Named(output) => Some(output.clone()),
+            _ => None,
+        }
+    }
+}
+
+impl Drop for Output {
+    /// Removes the output file if it was created
+    fn drop(&mut self) {
+        let _ = match self {
+            Output::StdOut => Ok(()),
+            Output::Named(target) => std::fs::remove_file(target),
+            Output::InPlace(target) => std::fs::remove_file(target),
+        };
+    }
+}
+
 /// Transforms files through an external program.
 /// The `command_str` field contains a path to a program and its space separated arguments.
 /// The command takes a file given in the `$IN` variable and produces an `$OUT` file.
+#[derive(Clone)]
 pub struct Transform {
     /// a path to a program and its space separated arguments
     pub command_str: String,
@@ -101,7 +115,7 @@ pub struct Transform {
     /// read output from the same location as the original
     pub in_place: bool,
     /// will be set to the name of the program, extracted from the command_str
-    program: String,
+    pub program: String,
 }
 
 impl Transform {
@@ -109,7 +123,7 @@ impl Transform {
         let has_in = RefCell::new(false);
         let has_out = RefCell::new(false);
 
-        let parsed = Self::parse_command(&command_str, |s: &str| {
+        let parsed = parse_command(&command_str, |s: &str| {
             match s {
                 "OUT" if cfg!(windows) => *has_out.borrow_mut() = true,
                 "IN" => *has_in.borrow_mut() = true,
@@ -202,72 +216,34 @@ impl Transform {
         self.tmp_dir.join(format!("{:x}", input.hash128()))
     }
 
-    /// Processes the file and logs any errors with the given logger.
-    /// The standard output and error streams of the child process are captured
-    /// and logged in case of a non-zero exit code.
-    pub fn run_or_log_err(&self, input: &Path, log: &Log) -> Option<(FileLen, FileHash)> {
-        let result = self.run(input);
-        match result {
-            Err(e) => {
-                log.err(format!("Failed to execute {}: {}", self.program, e));
-                None
-            }
-            Ok(result) if !result.status.success() => {
-                log.err(format!(
-                    "Failed to transform {}: {} returned non-zero status code: {}{}",
-                    input.display(),
-                    self.program,
-                    result.status.code().unwrap(),
-                    Self::format_output_stream(result.stderr.as_str(), "STDERR")
-                ));
-                None
-            }
-            Ok(result) => Some((result.output_len, result.output_hash)),
-        }
-    }
-
-    fn format_output_stream(output: &str, name: &str) -> String {
-        let output = output.trim().to_string();
-        if output.is_empty() {
-            output
-        } else {
-            format!(
-                "\n------------------------- {} -------------------------\n{}\
-                 \n----------------------------------------------------------",
-                name, output
-            )
-        }
-    }
-
-    /// Processes the input file, and computes the length and hash of the returned output stream
-    pub fn run(&self, input: &Path) -> io::Result<Output> {
+    /// Processes the input file and returns its output and err as as stream
+    pub fn run(&self, input: &Path) -> io::Result<Execution> {
         let (args, input_conf, output_conf) = self.make_args(input);
-        let mut command = Self::build_command(&args, &input_conf, &output_conf)?;
-        let result = Self::execute(&mut command, &output_conf)?;
-        input_conf.cleanup()?;
+        let mut command = build_command(&args, &input_conf, &output_conf)?;
+        let result = execute(&mut command, input_conf, output_conf)?;
         Ok(result)
     }
 
     /// Creates arguments, input and output configuration for processing given input path.
     /// The first element of the argument vector contains the program name.
-    fn make_args(&self, input: &Path) -> (Vec<OsString>, InputConf, OutputConf) {
-        let input_conf = RefCell::<InputConf>::new(InputConf::StdIn(input.to_path_buf()));
-        let output_conf = RefCell::<OutputConf>::new(OutputConf::StdOut);
+    fn make_args(&self, input: &Path) -> (Vec<OsString>, Input, Output) {
+        let input_conf = RefCell::<Input>::new(Input::StdIn(input.to_path_buf()));
+        let output_conf = RefCell::<Output>::new(Output::StdOut);
 
-        let args = Self::parse_command(self.command_str.as_str(), |arg| match arg {
+        let args = parse_command(self.command_str.as_str(), |arg| match arg {
             "IN" if self.copy => {
                 let tmp_target = self.random_tmp_file_name();
-                input_conf.replace(InputConf::Copied(input.to_path_buf(), tmp_target.clone()));
+                input_conf.replace(Input::Copied(input.to_path_buf(), tmp_target.clone()));
                 tmp_target.into_os_string()
             }
             "IN" => {
                 let input = input.to_path_buf();
-                input_conf.replace(InputConf::Named(input.clone()));
+                input_conf.replace(Input::Named(input.clone()));
                 input.into_os_string()
             }
             "OUT" => {
                 let output = self.output(input);
-                output_conf.replace(OutputConf::Named(output.clone()));
+                output_conf.replace(Output::Named(output.clone()));
                 output.into_os_string()
             }
             _ => OsString::from(arg),
@@ -277,168 +253,199 @@ impl Transform {
         let mut output_conf = output_conf.into_inner();
 
         if self.in_place {
-            output_conf = OutputConf::InPlace(input_conf.input_path().clone())
+            output_conf = Output::InPlace(input_conf.input_path().clone())
         }
 
         (args, input_conf, output_conf)
     }
+}
 
-    /// Builds the `Command` struct from the parsed arguments
-    fn build_command(
-        args: &[OsString],
-        input_conf: &InputConf,
-        output_conf: &OutputConf,
-    ) -> io::Result<Command> {
-        let mut args = args.iter();
-        let mut command = Command::new(args.next().unwrap());
-        command.args(args);
-        command.stderr(Stdio::piped());
-
-        input_conf.prepare_input_file()?;
-        if let InputConf::StdIn(_) = input_conf {
-            command.stdin(File::open(input_conf.input_path())?);
-        } else {
-            command.stdin(Stdio::null());
-        }
+/// Cleans up temporary files
+impl Drop for Transform {
+    fn drop(&mut self) {
+        let _ = remove_dir_all(&self.tmp_dir);
+    }
+}
 
-        if let OutputConf::Named(output) = output_conf {
-            command.stdout(Stdio::null());
-            Self::create_named_pipe(output)?;
-        } else {
-            command.stdout(Stdio::piped());
-        }
+/// Keeps the results of the transform program execution
+pub struct Execution {
+    pub(crate) child: Arc<Mutex<Child>>,
+    pub(crate) out_stream: Box<dyn Read>,
+    pub(crate) err_stream: Option<JoinHandle<String>>,
+    _input: Input,   // holds the temporary input file(s) until execution is done
+    _output: Output, // holds the temporary output file(s) until execution is done
+}
 
-        Ok(command)
+impl Drop for Execution {
+    fn drop(&mut self) {
+        let mut buf = [0; 4096];
+        while let Ok(1..) = self.out_stream.read(&mut buf) {}
+        let _ = self.child.lock().unwrap().wait();
     }
+}
 
-    #[cfg(unix)]
-    fn create_named_pipe(path: &std::path::Path) -> io::Result<()> {
-        use nix::sys::stat;
-        use nix::unistd::mkfifo;
-        if let Err(e) = mkfifo(path, stat::Mode::S_IRWXU) {
-            return Err(io::Error::new(
-                io::Error::from(e.as_errno().unwrap()).kind(),
-                format!("Failed to create named pipe {}: {}", path.display(), e),
-            ));
-        }
-        Ok(())
+/// Builds the `Command` struct from the parsed arguments
+fn build_command(
+    args: &[OsString],
+    input_conf: &Input,
+    output_conf: &Output,
+) -> io::Result<Command> {
+    let mut args = args.iter();
+    let mut command = Command::new(args.next().unwrap());
+    command.args(args);
+    command.stderr(Stdio::piped());
+
+    input_conf.prepare_input_file()?;
+    if let Input::StdIn(_) = input_conf {
+        command.stdin(File::open(input_conf.input_path())?);
+    } else {
+        command.stdin(Stdio::null());
     }
 
-    #[cfg(windows)]
-    fn create_named_pipe(_path: &PathBuf) -> io::Result<()> {
-        unimplemented!()
+    if let Output::Named(output) = output_conf {
+        command.stdout(Stdio::null());
+        create_named_pipe(output)?;
+    } else {
+        command.stdout(Stdio::piped());
     }
 
-    /// Spawns the command process,
-    /// computes its output length and its hash and captures its standard error into a string.
-    /// Blocks until the child process terminates.
-    fn execute(command: &mut Command, output_conf: &OutputConf) -> io::Result<Output> {
-        let mut child = command.spawn()?;
-
-        // We call 'take' to avoid borrowing `child` for longer than a single line.
-        // We can't reference stdout/stderr directly, because a mutable borrow of a field
-        // creates a mutable borrow of the containing struct, but later we need to mutably
-        // borrow `child` again to wait on it.
-        let child_out = child.stdout.take();
-        let child_err = child.stderr.take();
-
-        // Capture the stderr in background in order to avoid a deadlock when the child process
-        // would block on writing to stdout, and this process would block on reading stderr
-        // (or the other way round).
-        // The other solution could be to use non-blocking I/O, but threads look simpler.
-        let stderr_reaper = std::thread::spawn(move || {
-            let mut str = String::new();
-            let _ = child_err.unwrap().read_to_string(&mut str);
-            str
-        });
-
-        let result = match output_conf {
-            OutputConf::StdOut => {
-                stream_hash(&mut child_out.unwrap(), FileLen::MAX, BUF_LEN, |_| {})
-            }
-            OutputConf::Named(output) => {
-                stream_hash(&mut File::open(output)?, FileLen::MAX, BUF_LEN, |_| {})
-            }
-            OutputConf::InPlace(output) => {
-                child.wait()?;
-                stream_hash(&mut File::open(output)?, FileLen::MAX, BUF_LEN, |_| {})
-            }
-        }?;
+    Ok(command)
+}
 
-        Ok(Output {
-            output_len: result.0,
-            output_hash: result.1,
-            status: child.wait()?,
-            stderr: stderr_reaper.join().unwrap(),
-        })
+#[cfg(unix)]
+fn create_named_pipe(path: &std::path::Path) -> io::Result<()> {
+    use nix::sys::stat;
+    use nix::unistd::mkfifo;
+    if let Err(e) = mkfifo(path, stat::Mode::S_IRWXU) {
+        return Err(io::Error::new(
+            io::Error::from(e.as_errno().unwrap()).kind(),
+            format!("Failed to create named pipe {}: {}", path.display(), e),
+        ));
     }
+    Ok(())
+}
 
-    /// Compares the input with a regular expression and returns the first match.
-    /// Backported from nom 6.0.0-alpha1. We can't use nom 6.0.0-alpha1 directly,
-    /// because it had some issues with our use of functions in pattern.rs.
-    fn re_find<'s, E>(re: Regex) -> impl Fn(&'s str) -> IResult<&'s str, &'s str, E>
-    where
-        E: ParseError<&'s str>,
-    {
-        move |i| {
-            if let Some(m) = re.find(i) {
-                Ok((&i[m.end()..], &i[m.start()..m.end()]))
-            } else {
-                Err(nom::Err::Error(E::from_error_kind(
-                    i,
-                    ErrorKind::RegexpMatch,
-                )))
-            }
-        }
-    }
+#[cfg(windows)]
+fn create_named_pipe(_path: &PathBuf) -> io::Result<()> {
+    unimplemented!()
+}
 
-    /// Splits the command string into separate arguments and substitutes $params
-    fn parse_command<F>(command: &str, substitute: F) -> Vec<OsString>
-    where
-        F: Fn(&str) -> OsString,
-    {
-        fn join_chars(chars: Vec<char>) -> OsString {
-            let mut result = OsString::new();
-            for c in chars {
-                result.push(c.to_string())
-            }
-            result
+/// Spawns the command process, and returns its output as a stream.
+/// The standard error is captured by a background thread and read to a string.
+fn execute(command: &mut Command, input: Input, output: Output) -> io::Result<Execution> {
+    let child = Arc::new(Mutex::new(command.spawn()?));
+
+    // We call 'take' to avoid borrowing `child` for longer than a single line.
+    // We can't reference stdout/stderr directly, because a mutable borrow of a field
+    // creates a mutable borrow of the containing struct, but later we need to mutably
+    // borrow `child` again to wait on it.
+    let child_out = child.lock().unwrap().stdout.take();
+    let child_err = child.lock().unwrap().stderr.take();
+
+    let output_pipe = output.pipe_path();
+    let child_ref = child.clone();
+
+    // Capture the stderr in background in order to avoid a deadlock when the child process
+    // would block on writing to stdout, and this process would block on reading stderr
+    // (or the other way round).
+    // The other solution could be to use non-blocking I/O, but threads look simpler.
+    let stderr_reaper = std::thread::spawn(move || {
+        let mut str = String::new();
+        if let Some(mut stream) = child_err {
+            let _ = stream.read_to_string(&mut str);
         }
-
-        fn join_str(strings: Vec<OsString>) -> OsString {
-            let mut result = OsString::new();
-            for c in strings {
-                result.push(c)
-            }
-            result
+        // If the child is supposed to communicate its output through a named pipe,
+        // ensure the pipe gets closed and the reader at the other end receives an EOF.
+        // It is possible that due to a misconfiguration
+        // (e.g. wrong arguments given by the user) the child would never open the output file
+        // and the reader at the other end would block forever.
+        if let Some(output_pipe) = output_pipe {
+            // If those fail, we have no way to report the failure.
+            // However if waiting fails here, the child process likely doesn't run, so that's not
+            // a problem.
+            let _ignore = child_ref.lock().unwrap().wait();
+            let _ignore = OpenOptions::new().write(true).open(&output_pipe);
         }
+        str
+    });
+
+    let child_out: Box<dyn Read> = match &output {
+        Output::StdOut => Box::new(child_out.unwrap()),
+        Output::Named(output) => Box::new(File::open(output)?),
+        Output::InPlace(output) => {
+            child.lock().unwrap().wait()?;
+            Box::new(File::open(output)?)
+        }
+    };
+
+    Ok(Execution {
+        child,
+        out_stream: child_out,
+        err_stream: Some(stderr_reaper),
+        _input: input,
+        _output: output,
+    })
+}
 
-        let r_var = Regex::new(r"^([[:alnum:]]|_)+").unwrap();
-        let p_var = map(tuple((tag("$"), Self::re_find(r_var))), |(_, str)| {
-            (substitute)(str)
-        });
-        let p_non_var = map(many1(none_of(" $")), join_chars);
-        let p_arg = map(many1(alt((p_var, p_non_var))), join_str);
-        let p_whitespace = many1(one_of(" \t"));
-        let p_args = separated_list(p_whitespace, p_arg);
-        let result: IResult<&str, Vec<OsString>> = (p_args)(command);
-        result.expect("Parse error").1
+/// Compares the input with a regular expression and returns the first match.
+/// Backported from nom 6.0.0-alpha1. We can't use nom 6.0.0-alpha1 directly,
+/// because it had some issues with our use of functions in pattern.rs.
+fn re_find<'s, E>(re: Regex) -> impl Fn(&'s str) -> IResult<&'s str, &'s str, E>
+where
+    E: ParseError<&'s str>,
+{
+    move |i| {
+        if let Some(m) = re.find(i) {
+            Ok((&i[m.end()..], &i[m.start()..m.end()]))
+        } else {
+            Err(nom::Err::Error(E::from_error_kind(
+                i,
+                ErrorKind::RegexpMatch,
+            )))
+        }
     }
 }
 
-/// Cleans up temporary files
-impl Drop for Transform {
-    fn drop(&mut self) {
-        let _ = remove_dir_all(&self.tmp_dir);
+/// Splits the command string into separate arguments and substitutes $params
+fn parse_command<F>(command: &str, substitute: F) -> Vec<OsString>
+where
+    F: Fn(&str) -> OsString,
+{
+    fn join_chars(chars: Vec<char>) -> OsString {
+        let mut result = OsString::new();
+        for c in chars {
+            result.push(c.to_string())
+        }
+        result
     }
+
+    fn join_str(strings: Vec<OsString>) -> OsString {
+        let mut result = OsString::new();
+        for c in strings {
+            result.push(c)
+        }
+        result
+    }
+
+    let r_var = Regex::new(r"^([[:alnum:]]|_)+").unwrap();
+    let p_var = map(tuple((tag("$"), re_find(r_var))), |(_, str)| {
+        (substitute)(str)
+    });
+    let p_non_var = map(many1(none_of(" $")), join_chars);
+    let p_arg = map(many1(alt((p_var, p_non_var))), join_str);
+    let p_whitespace = many1(one_of(" \t"));
+    let p_args = separated_list(p_whitespace, p_arg);
+    let result: IResult<&str, Vec<OsString>> = (p_args)(command);
+    result.expect("Parse error").1
 }
 
 #[cfg(test)]
 mod test {
     use std::io::Write;
 
-    use crate::file::{FileChunk, FilePos};
-    use crate::hasher::file_hash;
+    use crate::file::{FileChunk, FileLen, FilePos};
+    use crate::hasher::FileHasher;
+    use crate::log::Log;
     use crate::util::test::with_dir;
 
     use super::*;
@@ -452,21 +459,21 @@ mod test {
     #[cfg(unix)]
     fn piped() {
         with_dir("target/test/transform/piped/", |root| {
-            let p = Transform::new(String::from("dd"), false).unwrap();
+            let transform = Transform::new(String::from("dd"), false).unwrap();
             let input_path = root.join("input.txt");
             let mut input = File::create(&input_path).unwrap();
             let content = b"content";
             input.write(content).unwrap();
             drop(input);
 
+            let log = Log::default();
+            let hasher = FileHasher::new(Some(transform), &log);
             let input_path = Path::from(input_path);
             let chunk = FileChunk::new(&input_path, FilePos(0), FileLen::MAX);
-            let good_file_hash = file_hash(&chunk, 4096, |_| {}).unwrap();
-
-            let result = p.run(&input_path).unwrap();
-            assert_eq!(result.status.code(), Some(0));
-            assert_eq!(result.output_len.0, content.len() as u64);
-            assert_eq!(result.output_hash, good_file_hash);
+            let good_file_hash = hasher.hash_file(&chunk, |_| {}).unwrap();
+            let result = hasher.hash_transformed(&chunk, |_| {}).unwrap();
+            assert_eq!(result.0, FileLen(content.len() as u64));
+            assert_eq!(result.1, good_file_hash);
         })
     }
 
@@ -474,34 +481,34 @@ mod test {
     #[cfg(unix)]
     fn parameterized() {
         with_dir("target/test/transform/param/", |root| {
-            let p = Transform::new(String::from("dd if=$IN of=$OUT"), false).unwrap();
+            let transform = Transform::new(String::from("dd if=$IN of=$OUT"), false).unwrap();
             let input_path = root.join("input.txt");
             let mut input = File::create(&input_path).unwrap();
             let content = b"content";
             input.write(content).unwrap();
             drop(input);
 
+            let log = Log::default();
+            let hasher = FileHasher::new(Some(transform), &log);
             let input_path = Path::from(input_path);
-            let result = p.run(&input_path).unwrap();
 
             let chunk = FileChunk::new(&input_path, FilePos(0), FileLen::MAX);
-            let good_file_hash = file_hash(&chunk, 4096, |_| {}).unwrap();
-
-            assert_eq!(result.status.code(), Some(0));
-            assert_eq!(result.output_len.0, content.len() as u64);
-            assert_eq!(result.output_hash, good_file_hash);
+            let good_file_hash = hasher.hash_file(&chunk, |_| {}).unwrap();
+            let result = hasher.hash_transformed(&chunk, |_| {}).unwrap();
+            assert_eq!(result.0, FileLen(content.len() as u64));
+            assert_eq!(result.1, good_file_hash);
         })
     }
 
     #[test]
     fn parse_command() {
-        let result = Transform::parse_command("foo bar", |s| OsString::from(s));
+        let result = super::parse_command("foo bar", |s| OsString::from(s));
         assert_eq!(result, vec![OsString::from("foo"), OsString::from("bar")])
     }
 
     #[test]
     fn parse_command_substitute() {
-        let result = Transform::parse_command("foo bar in=$IN", |s| match s {
+        let result = super::parse_command("foo bar in=$IN", |s| match s {
             "IN" => OsString::from("/input"),
             _ => OsString::from(s),
         });


Test patch: