看门狗兼容性：解决“CancelIoEx”的变通办法
我在使用Python的watchdog库来监控文件系统事件时发现，在Windows Server 2003上，它进入了“轮询模式”，这意味着它停止使用异步的操作系统通知，从而在文件变化很多的情况下，系统性能大幅下降。
我追踪到这个问题出现在watchdog/observers/winapi.py文件中，这里使用了CancelIoEx这个系统调用来停止ReadDirectoryChangesW的调用锁定，当用户想要停止监控某个目录或文件时：
(winapi.py)
CancelIoEx = ctypes.windll.kernel32.CancelIoEx
CancelIoEx.restype = ctypes.wintypes.BOOL
CancelIoEx.errcheck = _errcheck_bool
CancelIoEx.argtypes = (
ctypes.wintypes.HANDLE, # hObject
ctypes.POINTER(OVERLAPPED) # lpOverlapped
)
...
...
...
def close_directory_handle(handle):
try:
CancelIoEx(handle, None) # force ReadDirectoryChangesW to return
except WindowsError:
return
问题在于CancelIoEx这个调用在Windows Server 2008之前是不可用的：http://msdn.microsoft.com/en-us/library/windows/desktop/aa363792(v=vs.85).aspx
一个可能的替代方案是修改close_directory_handle，让它在被监控的目录中创建一个虚拟文件，从而解锁等待ReadDirectoryChangesW返回的线程。
不过，我注意到在Windows Server 2003中，CancelIo这个系统调用实际上是可用的：
它会取消调用线程为指定文件发出的所有待处理的输入和输出（I/O）操作。这个函数不会取消其他线程为文件句柄发出的I/O操作。要取消来自其他线程的I/O操作，可以使用CancelIoEx函数。
但是调用CancelIo并不会影响正在等待的线程。
你有没有什么想法可以解决这个问题？也许可以使用threading.enumerate()来发出一个信号，让每个线程在这些处理程序中调用CancelIo？
1 个回答
自然的做法是实现一个完成例程，并使用重叠模式调用 ReadDirectoryChangesW。下面的例子展示了如何做到这一点：
RDCW_CALLBACK_F = ctypes.WINFUNCTYPE(None, ctypes.wintypes.DWORD, ctypes.wintypes.DWORD, ctypes.POINTER(OVERLAPPED))
首先，创建一个 WINFUNCTYPE 工厂，这个工厂会用来从 Python 方法生成可以被 Windows API 调用的 C 风格函数。在这个例子中，没有返回值，并且有三个参数，分别对应于
VOID CALLBACK FileIOCompletionRoutine(
_In_ DWORD dwErrorCode,
_In_ DWORD dwNumberOfBytesTransfered,
_Inout_ LPOVERLAPPED lpOverlapped
);
FileIOCompletionRoutine 的头文件。
回调引用以及重叠结构需要添加到 ReadDirectoryChangesW 的参数列表中：
ReadDirectoryChangesW = ctypes.windll.kernel32.ReadDirectoryChangesW
ReadDirectoryChangesW.restype = ctypes.wintypes.BOOL
ReadDirectoryChangesW.errcheck = _errcheck_bool
ReadDirectoryChangesW.argtypes = (
ctypes.wintypes.HANDLE, # hDirectory
LPVOID, # lpBuffer
ctypes.wintypes.DWORD, # nBufferLength
ctypes.wintypes.BOOL, # bWatchSubtree
ctypes.wintypes.DWORD, # dwNotifyFilter
ctypes.POINTER(ctypes.wintypes.DWORD), # lpBytesReturned
ctypes.POINTER(OVERLAPPED), # lpOverlapped
RDCW_CALLBACK_F # FileIOCompletionRoutine # lpCompletionRoutine
)
从这里开始，我们准备进行重叠系统调用。这是一个简单的回调，主要用来测试一切是否正常工作：
def dir_change_callback(dwErrorCode,dwNumberOfBytesTransfered,p):
print("dir_change_callback! PID:" + str(os.getpid()))
print("CALLBACK THREAD: " + str(threading.currentThread()))
准备并执行调用：
event_buffer = ctypes.create_string_buffer(BUFFER_SIZE)
nbytes = ctypes.wintypes.DWORD()
overlapped_read_dir = OVERLAPPED()
call2pass = RDCW_CALLBACK_F(dir_change_callback)
hand = get_directory_handle(os.path.abspath("/test/"))
def docall():
ReadDirectoryChangesW(hand, ctypes.byref(event_buffer),
len(event_buffer), False,
WATCHDOG_FILE_NOTIFY_FLAGS,
ctypes.byref(nbytes),
ctypes.byref(overlapped_read_dir), call2pass)
print("Waiting!")
docall()
如果你将所有这些代码加载并执行到一个 DreamPie 交互式环境中，你可以检查系统调用是否完成，并且回调是否执行，从而在 c:\test 目录下第一次更改后打印出线程和进程 ID。此外，你会注意到这些与主线程和主进程是相同的：尽管事件是由一个独立线程触发的，但回调是在与我们的主程序相同的进程和线程中运行的，这样就导致了不希望出现的行为：
lck = threading.Lock()
def dir_change_callback(dwErrorCode,dwNumberOfBytesTransfered,p):
print("dir_change_callback! PID:" + str(os.getpid()))
print("CALLBACK THREAD: " + str(threading.currentThread()))
...
...
...
lck.acquire()
print("Waiting!")
docall()
lck.acquire()
这个程序会锁住主线程，回调将永远不会执行。我尝试了很多同步工具，甚至 Windows API 的信号量，结果总是得到相同的行为，所以最后我决定使用 multiprocessing Python 库在一个单独的进程中实现异步调用，并使用同步配置来调用 ReadDirectoryChangesW：
对 get_directory_handle 的调用不会返回 Windows API 给出的句柄编号，而是一个由 winapi 库管理的句柄，为此我实现了一个句柄生成器：
class FakeHandleFactory():
_hl = threading.Lock()
_next = 0
@staticmethod
def next():
FakeHandleFactory._hl.acquire()
ret = FakeHandleFactory._next
FakeHandleFactory._next += 1
FakeHandleFactory._hl.release()
return ret
每个生成的句柄必须与一个文件系统路径全局关联：
handle2file = {}
现在每次调用 read_directory_changes 都会生成一个 ReadDirectoryRequest（继承自 multiprocessing.Process）对象：
class ReadDirectoryRequest(multiprocessing.Process):
def _perform_and_wait4request(self, path, recursive, event_buffer, nbytes):
hdl = CreateFileW(path, FILE_LIST_DIRECTORY, WATCHDOG_FILE_SHARE_FLAGS,
None, OPEN_EXISTING, WATCHDOG_FILE_FLAGS, None)
#print("path: " + path)
aux_buffer = ctypes.create_string_buffer(BUFFER_SIZE)
aux_n = ctypes.wintypes.DWORD()
#print("_perform_and_wait4request! PID:" + str(os.getpid()))
#print("CALLBACK THREAD: " + str(threading.currentThread()) + "\n----------")
try:
ReadDirectoryChangesW(hdl, ctypes.byref(aux_buffer),
len(event_buffer), recursive,
WATCHDOG_FILE_NOTIFY_FLAGS,
ctypes.byref(aux_n), None, None)
except WindowsError as e:
print("!" + str(e))
if e.winerror == ERROR_OPERATION_ABORTED:
nbytes = 0
event_buffer = []
else:
nbytes = 0
event_buffer = []
# Python 2/3 compat
nbytes.value = aux_n.value
for i in xrange(self.int_class(aux_n.value)):
event_buffer[i] = aux_buffer[i]
CloseHandle(hdl)
try:
self.lck.release()
except:
pass
def __init__(self, handle, recursive):
buffer = ctypes.create_string_buffer(BUFFER_SIZE)
self.event_buffer = multiprocessing.Array(ctypes.c_char, buffer)
self.nbytes = multiprocessing.Value(ctypes.wintypes.DWORD, 0)
targetPath = handle2file.get(handle, None)
super(ReadDirectoryRequest, self).__init__(target=self._perform_and_wait4request, args=(targetPath, recursive, self.event_buffer, self.nbytes))
self.daemon = True
self.lck = multiprocessing.Lock()
self.result = None
try:
self.int_class = long
except NameError:
self.int_class = int
if targetPath is None:
self.result = ([], -1)
def CancelIo(self):
try:
self.result = ([], 0)
self.lck.release()
except:
pass
def read_changes(self):
#print("read_changes! PID:" + str(os.getpid()))
#print("CALLBACK THREAD: " + str(threading.currentThread()) + "\n----------")
if self.result is not None:
raise Exception("ReadDirectoryRequest object can be used only once!")
self.lck.acquire()
self.start()
self.lck.acquire()
self.result = (self.event_buffer, self.int_class(self.nbytes.value))
return self.result
这个类指定了 Process，提供一个执行系统调用的进程，并等待直到（或）：
- 发生了更改事件。
- 主线程通过调用
ReadDirectoryRequest对象的CancelIo方法取消请求。
注意：
- get_directory_handle
- close_directory_handle
- read_directory_changes
这些角色现在是管理请求。为此，需要线程锁和辅助数据结构：
rqIndexLck = threading.Lock() # Protects the access to `rqIndex`
rqIndex = {} # Maps handles to request objects sets.
get_directory_handle
def get_directory_handle(path):
rqIndexLck.acquire()
ret = FakeHandleFactory.next()
handle2file[ret] = path
rqIndexLck.release()
return ret
close_directory_handle
def close_directory_handle(handle):
rqIndexLck.acquire()
rqset4handle = rqIndex.get(handle, None)
if rqset4handle is not None:
for rq in rqset4handle:
rq.CancelIo()
del rqIndex[handle]
if handle in handle2file:
del handle2file[handle]
rqIndexLck.release()
最后但同样重要的是： read_directory_changes
def read_directory_changes(handle, recursive):
rqIndexLck.acquire()
rq = ReadDirectoryRequest(handle, recursive)
set4handle = None
if handle in rqIndex:
set4handle = rqIndex[handle]
else:
set4handle = set()
rqIndex[handle] = set4handle
set4handle.add(rq)
rqIndexLck.release()
ret = rq.read_changes()
rqIndexLck.acquire()
if rq in set4handle:
set4handle.remove(rq)
rqIndexLck.release()
return ret