Removed race condition by joining create and cancel timer queues. (#137)

* Removed race condition by joining create and cancel timer queues.

* Removed race condition by joining create and cancel timer queues.

* Removed debug prints

* Changed formatting to please cargo fmt

* Re-added semicolon that had been deleted by cargo fmt

Author: garin1000

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "message-io"
-version = "0.14.7"
+version = "0.14.8"
 authors = ["lemunozm <[email protected]>"]
 edition = "2018"
 readme = "README.md"

Cargo.toml

@@ -26,15 +26,20 @@ pub fn split<E: Send + 'static>() -> (EventSender<E>, EventReceiver<E>) {
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
 pub struct TimerId(Instant);
 
+// Internal enum to enqueue different timer commands in a single queue
+enum TimerCommand<E> {
+    Create(E),
+    Cancel,
+}
+
 /// A generic and synchronized queue where the user can send and receive events.
 /// See [`EventSender`] to see how send events.
 /// This entity can be used as an utility for the [`crate::network`] module redirecting the
 /// network events to process them later from here.
 pub struct EventReceiver<E> {
     event_sender: EventSender<E>, // Should be before receiver in order to drop first.
     receiver: Receiver<E>,
-    timer_receiver: Receiver<(Instant, E)>,
-    remove_timer_receiver: Receiver<Instant>,
+    timer_receiver: Receiver<(Instant, TimerCommand<E>)>,
     priority_receiver: Receiver<E>,
     timers: BTreeMap<Instant, E>,
 }
@@ -46,18 +51,11 @@ where E: Send + 'static
     fn default() -> Self {
         let (sender, receiver) = crossbeam_channel::unbounded();
         let (timer_sender, timer_receiver) = crossbeam_channel::unbounded();
-        let (remove_timer_sender, remove_timer_receiver) = crossbeam_channel::unbounded();
         let (priority_sender, priority_receiver) = crossbeam_channel::unbounded();
         EventReceiver {
-            event_sender: EventSender::new(
-                sender,
-                timer_sender,
-                remove_timer_sender,
-                priority_sender,
-            ),
+            event_sender: EventSender::new(sender, timer_sender, priority_sender),
             receiver,
             timer_receiver,
-            remove_timer_receiver,
             priority_receiver,
             timers: BTreeMap::new(),
         }
@@ -76,11 +74,10 @@ where E: Send + 'static
 
     fn enque_timers(&mut self) {
         for timer in self.timer_receiver.try_iter() {
-            self.timers.insert(timer.0, timer.1);
-        }
-
-        for timer_instant in self.remove_timer_receiver.try_iter() {
-            self.timers.remove(&timer_instant);
+            match timer.1 {
+                TimerCommand::Create(e) => self.timers.insert(timer.0, e),
+                TimerCommand::Cancel => self.timers.remove(&timer.0),
+            };
         }
     }
 
@@ -175,8 +172,7 @@ where E: Send + 'static
 /// This type can only be generated by the receiver `EventReceiver`.
 pub struct EventSender<E> {
     sender: Sender<E>,
-    timer_sender: Sender<(Instant, E)>,
-    remove_timer_sender: Sender<Instant>,
+    timer_sender: Sender<(Instant, TimerCommand<E>)>,
     priority_sender: Sender<E>,
 }
 
@@ -185,11 +181,10 @@ where E: Send + 'static
 {
     fn new(
         sender: Sender<E>,
-        timer_sender: Sender<(Instant, E)>,
-        remove_timer_sender: Sender<Instant>,
+        timer_sender: Sender<(Instant, TimerCommand<E>)>,
         priority_sender: Sender<E>,
     ) -> EventSender<E> {
-        EventSender { sender, timer_sender, remove_timer_sender, priority_sender }
+        EventSender { sender, timer_sender, priority_sender }
     }
 
     /// Send instantly an event to the event queue.
@@ -210,14 +205,14 @@ where E: Send + 'static
     /// [`EventSender::cancel_timer()`] be called.
     pub fn send_with_timer(&self, event: E, duration: Duration) -> TimerId {
         let when = Instant::now() + duration;
-        self.timer_sender.send((when, event)).ok();
+        self.timer_sender.send((when, TimerCommand::Create(event))).ok();
         TimerId(when)
     }
 
     /// Remove a timer previously sent by [`EventSender::send_with_timer()`].
     /// The timer will not be receive by the [`EventReceiver`].
     pub fn cancel_timer(&self, timer_id: TimerId) {
-        self.remove_timer_sender.send(timer_id.0).ok();
+        self.timer_sender.send((timer_id.0, TimerCommand::Cancel)).ok();
     }
 }
 
@@ -228,7 +223,6 @@ where E: Send + 'static
         EventSender::new(
             self.sender.clone(),
             self.timer_sender.clone(),
-            self.remove_timer_sender.clone(),
             self.priority_sender.clone(),
         )
     }