Make stop_watch really wait for the WatchState's completion semaphore…

… to signal

src/windows.rs

@@ -19,6 +19,7 @@ use winapi::um::winnt::{self, FILE_NOTIFY_INFORMATION, HANDLE};
 
 use super::debounce::{Debounce, EventTx};
 use super::{op, DebouncedEvent, Error, Op, RawEvent, RecursiveMode, Result, Watcher};
+use std::collections::HashMap;
 use std::env;
 use std::ffi::OsString;
 use std::mem;
@@ -31,10 +32,6 @@ use std::sync::mpsc::{channel, Receiver, Sender};
 use std::sync::{Arc, Mutex};
 use std::thread;
 use std::time::Duration;
-use std::{
-    borrow::{Borrow, BorrowMut},
-    collections::HashMap,
-};
 
 const BUF_SIZE: u32 = 16384;
 
@@ -71,10 +68,6 @@ struct WatchState {
     complete_sem: HANDLE,
 }
 
-thread_local! {
-    pub static APC_QUEUE_SIZE: std::cell::RefCell<usize> = std::cell::RefCell::new(0);
-}
-
 struct ReadDirectoryChangesServer {
     rx: Receiver<Action>,
     event_tx: Arc<Mutex<EventTx>>,
@@ -145,12 +138,6 @@ impl ReadDirectoryChangesServer {
         }
 
         // we have to clean this up, since the watcher may be long gone
-        while APC_QUEUE_SIZE.with(|x| *x.borrow()) > 0 {
-            // There is somehow a leak in the APC queue - let the completion routines fire until the APC queue is empty.
-            unsafe {
-                synchapi::WaitForSingleObjectEx(self.wakeup_sem, INFINITE, TRUE);
-            }
-        }
         unsafe {
             handleapi::CloseHandle(self.wakeup_sem);
         }
@@ -248,7 +235,9 @@ fn stop_watch(ws: &WatchState, meta_tx: &Sender<MetaEvent>) {
         let ch = handleapi::CloseHandle(ws.dir_handle);
         // have to wait for it, otherwise we leak the memory allocated for there read request
         if cio != 0 && ch != 0 {
-            synchapi::WaitForSingleObjectEx(ws.complete_sem, INFINITE, TRUE);
+            while synchapi::WaitForSingleObjectEx(ws.complete_sem, INFINITE, TRUE) != WAIT_OBJECT_0
+            {
+            }
         }
         handleapi::CloseHandle(ws.complete_sem);
     }
@@ -307,7 +296,6 @@ fn start_read(rd: &ReadData, event_tx: Arc<Mutex<EventTx>>, handle: HANDLE) {
             synchapi::ReleaseSemaphore(request.data.complete_sem, 1, ptr::null_mut());
         } else {
             // read ok. forget overlapped to let the completion routine handle memory
-            APC_QUEUE_SIZE.with(|x| *x.borrow_mut() += 1);
             mem::forget(overlapped);
         }
     }
@@ -328,8 +316,6 @@ unsafe extern "system" fn handle_event(
     _bytes_written: u32,
     overlapped: LPOVERLAPPED,
 ) {
-    APC_QUEUE_SIZE.with(|x| *x.borrow_mut() -= 1);
-
     let overlapped: Box<OVERLAPPED> = Box::from_raw(overlapped);
     let request: Box<ReadDirectoryRequest> = Box::from_raw(overlapped.hEvent as *mut _);