Make sure parking_lot::Condvar is signaled under a mutex

ethcore/src/block_queue.rs

@@ -316,7 +316,8 @@ impl BlockQueue {
 		match verify_block_basic(&header, &bytes, self.engine.deref().deref()) {
 			Ok(()) => {
 				self.processing.write().insert(h.clone());
-				self.verification.unverified.lock().push_back(UnverifiedBlock { header: header, bytes: bytes });
+				let mut unverified = self.verification.unverified.lock();
+				unverified.push_back(UnverifiedBlock { header: header, bytes: bytes });
 				self.more_to_verify.notify_all();
 				Ok(h)
 			},
@@ -432,8 +433,11 @@ impl Drop for BlockQueue {
 	fn drop(&mut self) {
 		trace!(target: "shutdown", "[BlockQueue] Closing...");
 		self.clear();
-		self.deleting.store(true, AtomicOrdering::Release);
-		self.more_to_verify.notify_all();
+		{
+			let _ = self.verification.unverified.lock();
+			self.deleting.store(true, AtomicOrdering::Release);
+			self.more_to_verify.notify_all();
+        }
 		for t in self.verifiers.drain(..) {
 			t.join().unwrap();
 		}

util/src/io/service.rs

@@ -170,14 +170,15 @@ pub struct IoManager<Message> where Message: Send + Sync {
 	workers: Vec<Worker>,
 	worker_channel: chase_lev::Worker<Work<Message>>,
 	work_ready: Arc<Condvar>,
+	work_mutex: Arc<Mutex<()>>,
 }
 
 impl<Message> IoManager<Message> where Message: Send + Sync + Clone + 'static {
 	/// Creates a new instance and registers it with the event loop.
 	pub fn start(panic_handler: Arc<PanicHandler>, event_loop: &mut EventLoop<IoManager<Message>>) -> Result<(), UtilError> {
 		let (worker, stealer) = chase_lev::deque();
 		let num_workers = 4;
-		let work_ready_mutex =  Arc::new(Mutex::new(()));
+		let work_ready_mutex = Arc::new(Mutex::new(()));
 		let work_ready = Arc::new(Condvar::new());
 		let workers = (0..num_workers).map(|i|
 			Worker::new(
@@ -196,6 +197,7 @@ impl<Message> IoManager<Message> where Message: Send + Sync + Clone + 'static {
 			worker_channel: worker,
 			workers: workers,
 			work_ready: work_ready,
+			work_mutex: work_ready_mutex,
 		};
 		try!(event_loop.run(&mut io));
 		Ok(())
@@ -221,6 +223,7 @@ impl<Message> Handler for IoManager<Message> where Message: Send + Clone + Sync
 					self.worker_channel.push(Work { work_type: WorkType::Writable, token: token_id, handler: handler.clone(), handler_id: handler_index });
 				}
 			}
+			let _ = self.work_mutex.lock();
 			self.work_ready.notify_all();
 		}
 	}

util/src/io/worker.rs

@@ -124,8 +124,11 @@ impl Drop for Worker {
 	fn drop(&mut self) {
 		trace!(target: "shutdown", "[IoWorker] Closing...");
 		let _ = self.wait_mutex.lock();
-		self.deleting.store(true, AtomicOrdering::Release);
-		self.wait.notify_all();
+		{
+			let _ = self.wait_mutex.lock();
+			self.deleting.store(true, AtomicOrdering::Release);
+			self.wait.notify_all();
+        }
 		let thread = mem::replace(&mut self.thread, None).unwrap();
 		thread.join().ok();
 		trace!(target: "shutdown", "[IoWorker] Closed");