Some work

Cargo.toml

@@ -2,6 +2,7 @@
 name = "source_query_cacher"
 version = "0.1.0"
 authors = ["Alik Aslanyan <[email protected]>"]
+license = "GPL-3.0"
 
 [profile.release]
 lto = true
@@ -20,4 +21,6 @@ byteorder = "1.2.4"
 fnv = "1.0.3"
 rand = "0.5"
 lru_time_cache = "0.8.0"
-structopt = "0.2"
+structopt = "0.2"
+itertools = "0.7.3"
+tokio-threadpool = "0.1.6"

src/bin/bin.rs

@@ -1,18 +1,36 @@
+//This file is part of source_query_cacher.
+//
+//source_query_cacher is free software: you can redistribute it and/or modify
+//it under the terms of the GNU General Public License as published by
+//the Free Software Foundation, either version 3 of the License, or
+//(at your option) any later version.
+//
+//source_query_cacher 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 Public License for more details.
+//
+//You should have received a copy of the GNU General Public License
+//along with source_query_cacher.  If not, see <https://www.gnu.org/licenses/>.
+
 extern crate env_logger;
 extern crate futures;
+extern crate itertools;
 extern crate source_query_cacher;
 extern crate tokio;
 #[macro_use]
 extern crate structopt;
+extern crate tokio_threadpool;
 
 use source_query_cacher::cacher;
 use std::net::SocketAddr;
 use std::time::Duration;
 use structopt::StructOpt;
 
+use itertools::*;
 use tokio::prelude::future::*;
 
-#[derive(Debug, StructOpt)]
+#[derive(Debug, StructOpt, Clone)]
 struct ServerClientPair {
     /// Server IP:PORT
     proxy: SocketAddr,
@@ -27,8 +45,12 @@ struct Options {
     update_period: u64,
     #[structopt(short = "c", long = "chunk-size", default_value = "5")]
     /// Number of servers to dispatched on the same thread.
-    chunks: usize,
-    #[structopt(short = "l", long = "list", raw(required = "true", min_values = "1"))]
+    chunk_size: usize,
+    #[structopt(
+        short = "l",
+        long = "list",
+        raw(required = "true", min_values = "1")
+    )]
     /// List of strings specified in "PROXY_IP:PORT SERVER_IP:PORT" format
     list: Vec<ServerClientPair>,
 }
@@ -74,16 +96,36 @@ impl std::str::FromStr for ServerClientPair {
     }
 }
 
+use tokio_threadpool::*;
+
 fn main() {
     env_logger::init();
-
     let options = Options::from_args();
+
     let period = options.update_period;
-    tokio::run(
-        join_all(options.list.into_iter().map(move |pair| {
-            cacher::cacher_run(pair.proxy, pair.server, Duration::from_millis(period))
-                .or_else(|()| futures::future::ok::<(), std::io::Error>(()))
-        })).map(|_| {})
-        .map_err(|_| {}),
-    );
+
+    let pool = ThreadPool::new();
+    options
+        .list
+        .iter()
+        .chunks(options.chunk_size)
+        .into_iter()
+        .map(|chunk| {
+            chunk
+                .into_iter()
+                .map(|it| it.clone())
+                .collect::<Vec<ServerClientPair>>()
+        }).collect::<Vec<_>>()
+        .into_iter()
+        .for_each(|chunk| {
+            pool.spawn(
+                join_all(chunk.into_iter().map(move |pair| {
+                    cacher::cacher_run(pair.proxy, pair.server, Duration::from_millis(period))
+                        .or_else(|()| futures::future::ok::<(), std::io::Error>(()))
+                })).map(|_| {})
+                .map_err(|_| {}),
+            )
+        });
+
+    pool.shutdown().wait().unwrap();
 }