kahing#63 threadpool configuration for command line

src/catfs/flags.rs

@@ -74,6 +74,8 @@ pub struct FlagStorage {
     pub free_space: DiskSpace,
     pub uid: libc::uid_t,
     pub gid: libc::gid_t,
+    pub catfs_threadpool_size: usize,
+    pub pcatfs_threadpool_size: usize
 }
 
 #[cfg(test)]

src/catfs/mod.rs

@@ -110,7 +110,7 @@ pub fn make_self<T>(s: &mut T) -> &'static T {
 }
 
 impl CatFS {
-    pub fn new(from: &dyn AsRef<Path>, to: &dyn AsRef<Path>) -> error::Result<CatFS> {
+    pub fn new(from: &dyn AsRef<Path>, to: &dyn AsRef<Path>, n_threads : usize) -> error::Result<CatFS> {
         let src_dir = rlibc::open(from, rlibc::O_RDONLY, 0)?;
         let cache_dir = rlibc::open(to, rlibc::O_RDONLY, 0)?;
 
@@ -123,7 +123,7 @@ impl CatFS {
             store: Mutex::new(Default::default()),
             dh_store: Mutex::new(Default::default()),
             fh_store: Mutex::new(Default::default()),
-            tp: Mutex::new(ThreadPool::new(5)),
+            tp: Mutex::new(ThreadPool::new(n_threads)),
         };
 
         catfs.make_root()?;

src/flags.rs

@@ -86,6 +86,11 @@ pub fn parse_options<'a, 'b>(mut app: clap::App<'a, 'a>, flags: &'b mut [Flag<'a
                 *v = String::from(s);
                 continue;
             }
+            if let Some(v) = f.value.downcast_mut::<usize>() {
+                let s = matches.value_of(name).unwrap();
+                *v = s.parse().unwrap();
+                continue;
+            }
             if let Some(v) = f.value.downcast_mut::<OsString>() {
                 let s = matches.value_of_os(name).unwrap();
                 *v = s.to_os_string();

src/main.rs

@@ -101,6 +101,8 @@ fn main_internal() -> error::Result<()> {
     flags.mount_options.push(
         OsString::from("default_permissions"),
     );
+    flags.catfs_threadpool_size = 5;
+    flags.pcatfs_threadpool_size = 100;
 
     let app = App::new("catfs")
         .about("Cache Anything FileSystem")
@@ -193,6 +195,20 @@ fn main_internal() -> error::Result<()> {
                     .validator(path_validator),
                 value: &mut flags.mount_point,
             },
+            flags::Flag{
+                arg: Arg::with_name("catfs-threadpool-size")
+                    .short("t")
+                    .takes_value(true)
+                    .help("Catfs thread pool size"),
+                value: &mut flags.catfs_threadpool_size,
+            },
+            flags::Flag{
+                arg: Arg::with_name("pcatfs-threadpool-size")
+                    .short("p")
+                    .takes_value(true)
+                    .help("PCatfs thread pool size"),
+                value: &mut flags.pcatfs_threadpool_size,
+            },
         ];
 
 
@@ -233,14 +249,14 @@ fn main_internal() -> error::Result<()> {
     let signal = chan_signal::notify(&[Signal::INT, Signal::TERM]);
     let path_from = Path::new(&flags.cat_from).canonicalize()?;
     let path_to = Path::new(&flags.cat_to).canonicalize()?;
-    let fs = catfs::CatFS::new(&path_from, &path_to)?;
-    let fs = pcatfs::PCatFS::new(fs);
+    let fs = catfs::CatFS::new(&path_from, &path_to, flags.catfs_threadpool_size)?;
+    let fs = pcatfs::PCatFS::new(fs, flags.pcatfs_threadpool_size);
     let cache_dir = fs.get_cache_dir()?;
     let mut options: Vec<&OsStr> = Vec::new();
     for i in 0..flags.mount_options.len() {
         options.push(&flags.mount_options[i]);
     }
-
+    debug!("threadpool options are catfs:{:?} pcatfs:{:?}", flags.catfs_threadpool_size, flags.pcatfs_threadpool_size);
     debug!("options are {:?}", flags.mount_options);
 
     {

src/pcatfs/mod.rs

@@ -28,9 +28,9 @@ pub fn make_self<T>(s: &mut T) -> &'static mut T {
 }
 
 impl PCatFS {
-    pub fn new(fs: CatFS) -> PCatFS {
+    pub fn new(fs: CatFS, n_threads : usize) -> PCatFS {
         PCatFS {
-            tp: ThreadPool::new(100),
+            tp: ThreadPool::new(n_threads),
             fs: fs,
         }
     }