remove the usage of Receiver::recv_timeout

It will conditionally panic due to this issue: rust-lang/rust#39364
The issue is not fixed yet, so I try to use alternative ways (e.g.
try_recv & sleep).
It will cause some performance penalty in theory, but not noticable by
human(for now).

Author: lotabout

Cargo.lock

@@ -1,8 +1,11 @@
 // All the events that will be used
 
 use std::any::Any;
+use std::sync::mpsc::{Receiver, Sender};
 
 pub type EventArg = Box<Any + 'static + Send>;
+pub type EventReceiver = Receiver<(Event, EventArg)>;
+pub type EventSender = Sender<(Event, EventArg)>;
 
 #[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
 pub enum Event {

src/event.rs

@@ -10,16 +10,16 @@ use std::collections::VecDeque;
 use std::time::Duration;
 use utf8parse;
 
-use event::{Event, EventArg, parse_action};
+use event::{Event, EventSender, parse_action};
 
 pub struct Input {
-    tx_input: Sender<(Event, EventArg)>,
+    tx_input: EventSender,
     keyboard: KeyBoard,
     keymap: HashMap<Key, (Event, Option<String>)>,
 }
 
 impl Input {
-    pub fn new(tx_input: Sender<(Event, EventArg)>) -> Self {
+    pub fn new(tx_input: EventSender) -> Self {
         let f = File::open("/dev/tty").unwrap();
         let keyboard = KeyBoard::new(f);
         Input {

src/input.rs

@@ -31,7 +31,7 @@ use std::time::Duration;
 use std::env;
 use std::sync::mpsc::{sync_channel, channel, Sender, Receiver};
 use event::Event::*;
-use event::{Event, EventArg};
+use event::{EventSender, EventReceiver};
 use std::mem;
 use std::ptr;
 use libc::{sigemptyset, sigaddset, sigwait, pthread_sigmask};
@@ -184,7 +184,7 @@ fn real_main() -> i32 {
         return 0;
     }
 
-    let (tx_input, rx_input): (Sender<(Event, EventArg)>, Receiver<(Event, EventArg)>) = channel();
+    let (tx_input, rx_input): (EventSender, EventReceiver) = channel();
     //------------------------------------------------------------------------------
     // register terminal resize event, `pthread_sigmask` should be run before any thread.
     let mut sigset = unsafe {mem::uninitialized()};

src/main.rs

@@ -1,6 +1,6 @@
 use std::sync::{Arc, RwLock};
-use std::sync::mpsc::{Receiver, Sender, channel};
-use event::{Event, EventArg};
+use std::sync::mpsc::channel;
+use event::{Event, EventSender, EventReceiver};
 use item::{Item, ItemGroup, MatchedItem, MatchedItemGroup, MatchedRange};
 use std::thread;
 
@@ -33,12 +33,12 @@ enum MatcherMode {
 }
 
 pub struct Matcher {
-    tx_result: Sender<(Event, EventArg)>,
+    tx_result: EventSender,
     mode: MatcherMode,
 }
 
 impl Matcher {
-    pub fn new(tx_result: Sender<(Event, EventArg)>) -> Self {
+    pub fn new(tx_result: EventSender) -> Self {
         Matcher {
             tx_result: tx_result,
             mode: MatcherMode::Fuzzy,
@@ -78,18 +78,21 @@ impl Matcher {
 
     }
 
-    pub fn run(&self, rx_item: Receiver<(Event, EventArg)>) {
-        let (tx_matcher, rx_matcher) = channel();
+    pub fn run(&self, rx_item: EventReceiver) {
+        let (tx_matcher, rx_matcher): (EventSender, EventReceiver) = channel();
         let matcher_restart = Arc::new(AtomicBool::new(false));
         // start a new thread listening for EvMatcherRestart, that means the query had been
         // changed, so that matcher shoudl discard all previous events.
         {
             let matcher_restart = Arc::clone(&matcher_restart);
             thread::spawn(move || {
                 while let Ok((ev, arg)) = rx_item.recv() {
+                    debug!("matcher: rx_item: {:?}", ev);
                     match ev {
                         Event::EvMatcherRestart => {
                             matcher_restart.store(true, Ordering::Relaxed);
+
+                            let _ = tx_matcher.send((ev, Box::new(true)));
                             while matcher_restart.load(Ordering::Relaxed) {
                                 thread::sleep(Duration::from_millis(10));
                             }
@@ -108,77 +111,76 @@ impl Matcher {
         let mut matcher_engine: Option<Box<MatchEngine>> = None;
         let mut num_processed: usize = 0;
         let mut matcher_mode = self.mode;
-        loop {
 
+        while let Ok((ev, arg)) = rx_matcher.recv() {
+            debug!("matcher: rx_matcher: {:?}", ev);
             if matcher_restart.load(Ordering::Relaxed) {
                 while let Ok(_) = rx_matcher.try_recv() {}
                 matcher_restart.store(false, Ordering::Relaxed);
+                continue;
             }
 
-            if let Ok((ev, arg)) = rx_matcher.recv_timeout(Duration::from_millis(10)) {
-                match ev {
-                    Event::EvMatcherNewItem => {
-                        let items: ItemGroup = *arg.downcast().unwrap();
-                        num_processed += items.len();
-
-                        matcher_engine.as_ref().map(|mat| {
-                            let matched_items: MatchedItemGroup = items.into_iter()
-                                .filter_map(|item| mat.match_item(item))
-                                .collect();
-                            let _ = self.tx_result.send((Event::EvModelNewItem, Box::new(matched_items)));
-                        });
-
-                        // report the number of processed items
-                        let _ = self.tx_result.send((Event::EvModelNotifyProcessed, Box::new(num_processed)));
-                    }
+            match ev {
+                Event::EvMatcherRestart => {
+                    num_processed = 0;
+                    let query = arg.downcast::<String>().unwrap();
 
-                    Event::EvReaderStopped | Event::EvReaderStarted => {
-                        let _ = self.tx_result.send((ev, arg));
-                    }
-                    Event::EvSenderStopped => {
-                        // Since matcher is single threaded, sender stopped means all items are
-                        // processed.
-                        let _ = self.tx_result.send((Event::EvModelNotifyProcessed, Box::new(num_processed)));
-                        let _ = self.tx_result.send((Event::EvMatcherStopped, arg));
-                    }
+                    // notifiy the model that the query had been changed
+                    let _ = self.tx_result.send((Event::EvModelRestart, Box::new(true)));
 
+                    let mode_string = match matcher_mode {
+                        MatcherMode::Regex => "RE".to_string(),
+                        MatcherMode::Exact => "EX".to_string(),
+                        _ => "".to_string(),
+                    };
+                    let _ = self.tx_result.send((Event::EvModelNotifyMatcherMode, Box::new(mode_string)));
 
-                    Event::EvMatcherRestart => {
-                        num_processed = 0;
-                        let query = arg.downcast::<String>().unwrap();
+                    matcher_engine = Some(EngineFactory::build(&query, matcher_mode));
+                }
 
-                        // notifiy the model that the query had been changed
-                        let _ = self.tx_result.send((Event::EvModelRestart, Box::new(true)));
+                Event::EvMatcherNewItem => {
+                    let items: ItemGroup = *arg.downcast().unwrap();
+                    num_processed += items.len();
 
-                        let mode_string = match matcher_mode {
-                            MatcherMode::Regex => "RE".to_string(),
-                            MatcherMode::Exact => "EX".to_string(),
-                            _ => "".to_string(),
-                        };
-                        let _ = self.tx_result.send((Event::EvModelNotifyMatcherMode, Box::new(mode_string)));
+                    matcher_engine.as_ref().map(|mat| {
+                        let matched_items: MatchedItemGroup = items.into_iter()
+                            .filter_map(|item| mat.match_item(item))
+                            .collect();
+                        let _ = self.tx_result.send((Event::EvModelNewItem, Box::new(matched_items)));
+                    });
 
-                        matcher_engine = Some(EngineFactory::build(&query, matcher_mode));
-                    }
+                    // report the number of processed items
+                    let _ = self.tx_result.send((Event::EvModelNotifyProcessed, Box::new(num_processed)));
+                }
 
-                    Event::EvActRotateMode => {
-                        if self.mode == MatcherMode::Regex {
-                            // sk started with regex mode.
-                            matcher_mode = if matcher_mode == self.mode {
-                                MatcherMode::Fuzzy
-                            } else {
-                                MatcherMode::Regex
-                            };
+                Event::EvReaderStopped | Event::EvReaderStarted => {
+                    let _ = self.tx_result.send((ev, arg));
+                }
+                Event::EvSenderStopped => {
+                    // Since matcher is single threaded, sender stopped means all items are
+                    // processed.
+                    let _ = self.tx_result.send((Event::EvModelNotifyProcessed, Box::new(num_processed)));
+                    let _ = self.tx_result.send((Event::EvMatcherStopped, arg));
+                }
+
+                Event::EvActRotateMode => {
+                    if self.mode == MatcherMode::Regex {
+                        // sk started with regex mode.
+                        matcher_mode = if matcher_mode == self.mode {
+                            MatcherMode::Fuzzy
                         } else {
-                            matcher_mode = if matcher_mode == self.mode {
-                                MatcherMode::Regex
-                            } else {
-                                self.mode
-                            }
+                            MatcherMode::Regex
+                        };
+                    } else {
+                        matcher_mode = if matcher_mode == self.mode {
+                            MatcherMode::Regex
+                        } else {
+                            self.mode
                         }
                     }
-
-                    _ => {}
                 }
+
+                _ => {}
             }
         }
     }

src/matcher.rs

@@ -1,5 +1,5 @@
-use std::sync::mpsc::{Receiver, Sender};
-use event::{Event, EventArg};
+use std::sync::mpsc::Sender;
+use event::{Event, EventReceiver};
 use item::{MatchedItem, MatchedItemGroup, MatchedRange};
 use std::cmp::{max, min};
 use orderedvec::OrderedVec;
@@ -29,7 +29,7 @@ lazy_static! {
 }
 
 pub struct Model {
-    rx_cmd: Receiver<(Event, EventArg)>,
+    rx_cmd: EventReceiver,
     items: OrderedVec<Arc<MatchedItem>>, // all items
     selected: HashMap<(usize, usize), Arc<MatchedItem>>,
 
@@ -57,7 +57,7 @@ pub struct Model {
 }
 
 impl Model {
-    pub fn new(rx_cmd: Receiver<(Event, EventArg)>) -> Self {
+    pub fn new(rx_cmd: EventReceiver) -> Self {
         Model {
             rx_cmd: rx_cmd,
             items: OrderedVec::new(),

src/model.rs

@@ -4,7 +4,7 @@ use item::Item;
 use std::sync::{Arc, RwLock};
 use std::process::{Command, Stdio, Child};
 use std::io::{BufRead, BufReader};
-use event::{Event, EventArg};
+use event::{EventSender, EventReceiver, Event, EventArg};
 use std::thread::JoinHandle;
 use std::thread;
 use std::time::Duration;
@@ -66,14 +66,14 @@ impl ReaderOption {
 }
 
 pub struct Reader {
-    rx_cmd: Receiver<(Event, EventArg)>,
+    rx_cmd: EventReceiver,
     tx_item: SyncSender<(Event, EventArg)>,
     option: Arc<RwLock<ReaderOption>>,
     real_stdin: Option<File>,  // used to support piped output
 }
 
 impl Reader {
-    pub fn new(rx_cmd: Receiver<(Event, EventArg)>,
+    pub fn new(rx_cmd: EventReceiver,
                tx_item: SyncSender<(Event, EventArg)>,
                real_stdin: Option<File>) -> Self {
         Reader {
@@ -184,7 +184,7 @@ lazy_static! {
 
 fn reader(cmd: &str,
           rx_cmd: Receiver<bool>,
-          tx_sender: &Sender<(Event, EventArg)>,
+          tx_sender: &EventSender,
           option: Arc<RwLock<ReaderOption>>,
           source_file: Option<File>) {
 
@@ -210,13 +210,15 @@ fn reader(cmd: &str,
                 break;
             }
 
-            if rx_control.recv_timeout(Duration::from_millis(10)).is_ok() {
+            if rx_control.try_recv().is_ok() {
                 command.map(|mut x| {
                     let _ = x.kill();
                     let _ = x.wait();
                 });
                 break;
             }
+
+            thread::sleep(Duration::from_millis(5));
         }
     });
 

src/reader.rs

@@ -1,18 +1,18 @@
-use std::sync::mpsc::{Receiver, SyncSender};
+use std::sync::mpsc::SyncSender;
 use item::ItemGroup;
-use event::{Event, EventArg};
+use event::{Event, EventArg, EventReceiver};
 use std::thread;
 use std::time::Duration;
 
 // sender is a cache of reader
 pub struct CachedSender {
     items: Vec<ItemGroup>, // cache
-    rx_sender: Receiver<(Event, EventArg)>,
+    rx_sender: EventReceiver,
     tx_item: SyncSender<(Event, EventArg)>,
 }
 
 impl CachedSender {
-    pub fn new(rx_sender: Receiver<(Event, EventArg)>,
+    pub fn new(rx_sender: EventReceiver,
                tx_item: SyncSender<(Event, EventArg)>) -> Self {
         CachedSender{
             items: Vec::new(),
@@ -32,7 +32,7 @@ impl CachedSender {
 
         loop {
             // try to read a bunch of items first
-            if let Ok((ev, arg)) = self.rx_sender.recv_timeout(Duration::from_millis(10)) {
+            if let Ok((ev, arg)) = self.rx_sender.try_recv() {
                 match ev {
                     Event::EvReaderStarted => {
                         reader_stopped = false;
@@ -81,7 +81,7 @@ impl CachedSender {
                     am_i_runing = false;
                 }
             } else {
-                thread::sleep(Duration::from_millis(3));
+                thread::sleep(Duration::from_millis(10));
             }
         }
     }