Add DenseNode::raw_tags

src/dense.rs

@@ -53,13 +53,23 @@ impl<'a> DenseNode<'a> {
         self.timestamp * i64::from(self.block.get_date_granularity())
     }
 
-    /// Returns an iterator over the tags of this way (See [OSM wiki](http://wiki.openstreetmap.org/wiki/Tags)).
+    /// Returns an iterator over the tags of this node (See [OSM wiki](http://wiki.openstreetmap.org/wiki/Tags)).
     pub fn tags(&self) -> DenseTagIter<'a> {
         DenseTagIter {
             block: self.block,
             keys_vals_indices: self.keys_vals_indices.iter(),
         }
     }
+
+    /// Returns an iterator over the tags of this node
+    /// (See [OSM wiki](http://wiki.openstreetmap.org/wiki/Tags)).
+    /// A tag is represented as a pair of indices (key and value) to the stringtable of the current
+    /// `PrimitiveBlock`.
+    pub fn raw_tags(&self) -> DenseRawTagIter<'a> {
+        DenseRawTagIter {
+            keys_vals_indices: self.keys_vals_indices.iter(),
+        }
+    }
 }
 
 /// An iterator over dense nodes. It decodes the delta encoded values.
@@ -228,8 +238,35 @@ impl<'a> Iterator for DenseTagIter<'a> {
     }
 
     fn size_hint(&self) -> (usize, Option<usize>) {
-        self.keys_vals_indices.size_hint()
+        let len = self.keys_vals_indices.len() / 2;
+        (len, Some(len))
     }
 }
 
 impl<'a> ExactSizeIterator for DenseTagIter<'a> {}
+
+/// An iterator over the tags of a node. It returns a pair of indices (key and value) to the
+/// stringtable of the current `PrimitiveBlock`.
+#[derive(Clone, Debug)]
+pub struct DenseRawTagIter<'a> {
+    keys_vals_indices: std::slice::Iter<'a, i32>,
+}
+
+//TODO return Result
+impl<'a> Iterator for DenseRawTagIter<'a> {
+    type Item = (i32, i32);
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match (self.keys_vals_indices.next(), self.keys_vals_indices.next()) {
+            (Some(&key_index), Some(&val_index)) => Some((key_index, val_index)),
+            _ => None,
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let len = self.keys_vals_indices.len() / 2;
+        (len, Some(len))
+    }
+}
+
+impl<'a> ExactSizeIterator for DenseRawTagIter<'a> {}

src/hungry_count.rs

@@ -0,0 +1,94 @@
+extern crate osmpbf;
+extern crate crossbeam;
+
+use osmpbf::*;
+use std::fs::File;
+use std::sync::mpsc::sync_channel;
+
+
+enum WorkerMessage<'a> {
+    PleaseStop,
+    DoBlob(MmapBlob<'a>),
+}
+
+fn stats(file: &File) {
+    let mmap = Mmap::from_file(file).unwrap();
+
+    crossbeam::scope(|scope| {
+        let mut iter = MmapBlobIter::new(&mmap);
+
+        let num_threads = 4_usize;
+        let mut chans = Vec::with_capacity(num_threads);
+
+        let (result_tx, result_rx) = sync_channel::<(usize, usize)>(0);
+
+        for thread_id in 0..num_threads {
+            let result_tx = result_tx.clone();
+
+            let (request_tx, request_rx) = sync_channel::<WorkerMessage>(0);
+            chans.push(request_tx);
+
+            scope.spawn(move || {
+                loop {
+                    match request_rx.recv().unwrap() {
+                        WorkerMessage::PleaseStop => return,
+                        WorkerMessage::DoBlob(mmap_blob) => {
+                            let count = if let BlobDecode::OsmData(block) = mmap_blob.decode() {
+                                block.groups()
+                                     .flat_map(|g| g.dense_nodes())
+                                     .count()
+                            } else {
+                                0
+                            };
+                            match result_tx.send((thread_id, count)) {
+                                Ok(_) => {},
+                                Err(_) => {
+                                    println!("Thread id {}, Err", thread_id);
+                                    return;
+                                },
+                            };
+                        },
+                    };
+                }
+            });
+        }
+
+        //TODO fix dead lock when number of blobs is less than number of threads
+        for (thread_id, mmap_blob) in iter.by_ref().take(num_threads).enumerate() {
+            chans[thread_id].send(WorkerMessage::DoBlob(mmap_blob)).unwrap();
+        }
+
+        let mut stopped = 0;
+        let mut nodes = 0;
+        loop {
+            let (thread_id, count) = result_rx.recv().unwrap();
+            nodes += count;
+
+            if let Some(mmap_blob) = iter.next() {
+                match chans[thread_id].send(WorkerMessage::DoBlob(mmap_blob)) {
+                    Ok(_) => {},
+                    Err(_) => {println!("Err");},
+                };
+            } else {
+                match chans[thread_id].send(WorkerMessage::PleaseStop) {
+                    Ok(_) => {},
+                    Err(_) => {
+                    }
+                };
+                stopped += 1;
+                if stopped == num_threads {
+                    break;
+                }
+            }
+        }
+
+        println!("Number of nodes: {}", nodes);
+    });
+}
+
+fn main() {
+    let path = std::env::args_os().nth(1).unwrap();
+    let f = std::fs::File::open(path).unwrap();
+
+    stats(&f);
+}