Speed up searching

* Decode/search in parallel
* Start decoding the first Blob as soon as possible
* Update multiple markers at once

Cargo.lock

  "lazy_static 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "linked-hash-map 0.5.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "num_cpus 1.8.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "osmpbf 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)",
  "regex 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "reqwest 0.8.6 (registry+https://github.com/rust-lang/crates.io-index)",
+ "scoped_threadpool 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)",
  "toml 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 

Cargo.toml

 [dependencies]
 cgmath = "0.16"
 clap = "2.29"
+directories = "0.10"
 env_logger = "0.5.0-rc.2"
 gl = "0.10"
 glutin = "0.16"
 lazy_static = "1.0"
 linked-hash-map = "0.5.0"
 log = "0.4"
+num_cpus = "1.0"
 osmpbf = "0.1"
 regex = "1.0"
 reqwest = "0.8"
+scoped_threadpool = "0.1"
 toml = "0.4"
-directories = "0.10"
 
 [build-dependencies]
 gl_generator = "0.9"

src/main.rs

 extern crate linked_hash_map;
 #[macro_use]
 extern crate log;
+extern crate num_cpus;
 extern crate osmpbf;
 extern crate regex;
 extern crate reqwest;
+extern crate scoped_threadpool;
 extern crate toml;
 
 pub mod args;
     map: &mut MapViewGl,
     input_state: &mut InputState,
     sources: &mut TileSources,
-    marker_rx: &mpsc::Receiver<LatLon>,
+    marker_rx: &mpsc::Receiver<Vec<LatLon>>,
 ) -> Action {
     match *event {
         Event::Awakened => {
-            for pos in marker_rx.try_iter() {
+            for pos in marker_rx.try_iter().flat_map(|c| c.into_iter()) {
                 map.add_marker(pos.into());
             }
             Action::Redraw
     if let (Some(path), Some(pattern)) = (config.pbf_path(), config.search_pattern()) {
         let proxy = events_loop.create_proxy();
 
-        search::search_pbf(
+        search::par_search(
             path,
             pattern,
-            move |latlon| {
-                if marker_tx.send(latlon).is_err() {
-                    return search::ControlFlow::Break;
+            move |coords| {
+                if coords.is_empty() {
+                    search::ControlFlow::Continue
+                } else {
+                    if marker_tx.send(coords).is_err() {
+                        return search::ControlFlow::Break;
+                    }
+                    proxy.wakeup().into()
+                }
+            },
+            move |result| {
+                if let Err(err) = result {
+                    println!("search error: {}", err);
+                } else {
+                    info!("finished searching");
                 }
-                proxy.wakeup().into()
             },
         )?;
     }

src/search.rs

-use osmpbf::{Element, ElementReader};
+use scoped_threadpool::Pool;
+use coord::LatLon;
+use osmpbf::{Blob, BlobDecode, BlobReader};
 use regex::Regex;
 use std::path::{Path, PathBuf};
+use std::sync::mpsc::sync_channel;
 use std::thread;
-use coord::LatLon;
 
 
 #[derive(Debug, Eq, PartialEq)]
     }
 }
 
-//TODO Add callbacks for other events: search finished, on error, ...
-pub fn search_pbf<P, F>(
+enum WorkerMessage {
+    PleaseStop,
+    DoBlob(Blob),
+}
+
+pub fn par_search<P, F, G>(
     pbf_path: P,
     search_pattern: &str,
-    update_func: F,
+    found_func: F,
+    finished_func: G,
 ) -> Result<thread::JoinHandle<()>, String>
 where P: AsRef<Path>,
-      F: Fn(LatLon) -> ControlFlow + Send + 'static,
+      F: Fn(Vec<LatLon>) -> ControlFlow + Send + 'static,
+      G: Fn(Result<(), String>) + Send + 'static,
 {
-    let pathbuf = PathBuf::from(pbf_path.as_ref());
-    let re = Regex::new(search_pattern)
-        .map_err(|e| format!("{}", e))?;
-    let reader = ElementReader::from_path(&pathbuf)
-        .map_err(|e| format!("Failed to read PBF file {:?}: {}", pbf_path.as_ref(), e))?;
-
+    let pbf_path = PathBuf::from(pbf_path.as_ref());
+    let search_pattern = search_pattern.to_string();
     let handle = thread::spawn(move|| {
-        reader.for_each(|element| {
-            match element {
-                Element::Node(node) => {
-                    for (_key, val) in node.tags() {
-                        if re.is_match(val) {
-                            let pos = LatLon::new(node.lat(), node.lon());
-                            if update_func(pos) == ControlFlow::Break {
-                                return;
+        let res = par_search_blocking(pbf_path, &search_pattern, found_func);
+        finished_func(res);
+    });
+
+    Ok(handle)
+}
+
+pub fn par_search_blocking<P, F>(
+    pbf_path: P,
+    search_pattern: &str,
+    found_func: F,
+) -> Result<(), String>
+where P: AsRef<Path>,
+      F: Fn(Vec<LatLon>) -> ControlFlow + Send + 'static,
+{
+    let num_threads = ::num_cpus::get();
+    let mut pool = Pool::new(num_threads as u32);
+
+    pool.scoped(|scope| {
+        let re = Regex::new(search_pattern)
+            .map_err(|e| format!("{}", e))?;
+        let mut reader = BlobReader::from_path(&pbf_path)
+            .map_err(|e| format!("{}", e))?;
+
+        let mut chans = Vec::with_capacity(num_threads);
+        let (result_tx, result_rx) = sync_channel::<(usize, Result<Vec<LatLon>, String>)>(0);
+
+        for thread_id in 0..num_threads {
+            let re = re.clone();
+            let result_tx = result_tx.clone();
+
+            let (request_tx, request_rx) = sync_channel::<WorkerMessage>(0);
+            chans.push(request_tx);
+
+            scope.execute(move || {
+                for request in request_rx.iter() {
+                    match request {
+                        WorkerMessage::PleaseStop => return,
+                        WorkerMessage::DoBlob(blob) => {
+                            let mut matches = vec![];
+                            let block = match blob.decode() {
+                                Ok(b) => b,
+                                Err(err) => {
+                                    let _ = result_tx.send((thread_id, Err(format!("{}", err))));
+                                    return;
+                                }
+                            };
+                            if let BlobDecode::OsmData(block) = block {
+                                for node in block.groups().flat_map(|g| g.nodes()) {
+                                    for (_key, val) in node.tags() {
+                                        if re.is_match(val) {
+                                            let pos = LatLon::new(node.lat(), node.lon());
+                                            matches.push(pos);
+                                            break;
+                                        }
+                                    }
+                                }
+
+                                for node in block.groups().flat_map(|g| g.dense_nodes()) {
+                                    for (_key, val) in node.tags() {
+                                        if re.is_match(val) {
+                                            let pos = LatLon::new(node.lat(), node.lon());
+                                            matches.push(pos);
+                                            break;
+                                        }
+                                    }
+                                }
                             }
-                            break;
-                        }
-                    }
-                },
-                Element::DenseNode(node) => {
-                    for (_key, val) in node.tags() {
-                        if re.is_match(val) {
-                            let pos = LatLon::new(node.lat(), node.lon());
-                            if update_func(pos) == ControlFlow::Break {
+                            if result_tx.send((thread_id, Ok(matches))).is_err() {
                                 return;
                             }
-                            break;
                         }
-                    }
+                    };
+                }
+            });
+        }
+
+        let mut stopped_threads = 0;
+
+        // send initial message to each worker thread
+        for thread_id in 0..num_threads {
+            match reader.next() {
+                Some(Ok(blob)) => {
+                    chans[thread_id].send(WorkerMessage::DoBlob(blob))
+                        .map_err(|e| format!("{}", e))?;
+
+                },
+                Some(Err(err)) => {
+                    return Err(format!("{}", err));
+                },
+                None => {
+                    chans[thread_id].send(WorkerMessage::PleaseStop)
+                        .map_err(|e| format!("{}", e))?;
+                    stopped_threads += 1;
                 },
-                _ => {},
             }
-        }).unwrap();
-    });
+        }
 
-    Ok(handle)
+        if stopped_threads == num_threads {
+            return Ok(());
+        }
+
+        for (thread_id, matches) in result_rx.iter() {
+            let matches = matches?;
+
+            if found_func(matches) == ControlFlow::Break {
+                break;
+            }
+
+            match reader.next() {
+                Some(Ok(blob)) => {
+                    chans[thread_id].send(WorkerMessage::DoBlob(blob))
+                        .map_err(|e| format!("{}", e))?;
+                },
+                Some(Err(err)) => {
+                    return Err(format!("{}", err));
+                },
+                None => {
+                    chans[thread_id].send(WorkerMessage::PleaseStop)
+                        .map_err(|e| format!("{}", e))?;
+                    stopped_threads += 1;
+                    if stopped_threads == num_threads {
+                        break;
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    })
 }