看门狗兼容性："CancelIoEx“的解决方案
使用python 看门狗文件系统事件监视库，我注意到在Windows 2003下使用时，它进入了“轮询模式”，从而停止使用异步OS通知，因此，在大量文件更改的情况下，严重降低了系统性能。
我将问题追溯到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:
returnCancelIoEx调用的问题是直到Windows 2008：http://msdn.microsoft.com/en-us/library/windows/desktop/aa363792(v=vs.85).aspx才可用
一种可能的选择是更改close_directory_handle，使其在受监视的目录中创建一个模拟文件，从而打开等待ReadDirectoryChangesW返回的线程。
但是，我注意到Windows 2003中的CancelIo系统调用是事实上是可用的：
取消由调用线程为指定文件发出的所有挂起的输入和输出(I/O)操作。该函数不会取消其他线程为文件句柄发出的I/O操作。要取消来自另一个线程的I/O操作，请使用CancelIoEx函数。
但是调用CancelIo不会影响等待线程。
你知道如何解决这个问题吗？可以使用threading.enumerate()，发出一个信号，由每个被这些处理程序调用的CancelIo处理的线程处理？
回答 1
Stack Overflow用户
发布于 2014-09-15 07:47:04
自然的方法是实现一个完成例程，并使用其重叠模式调用ReadDirectoryChangesW。下面的示例演示了实现此操作的方法：
RDCW_CALLBACK_F = ctypes.WINFUNCTYPE(None, ctypes.wintypes.DWORD, ctypes.wintypes.DWORD, ctypes.POINTER(OVERLAPPED))首先，创建一个WINFUNCTYPE工厂，它将用于从python方法生成类似于C的函数(可从Windows调用)。在这种情况下，没有返回值和对应于
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
)从这里开始，我们准备执行重叠的系统调用。这是一个简单的调用-- bacl只是用来测试一切是否正常：
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交互式shell中，则可以检查系统调用是否完成，回调是否执行，从而在c:\test目录下完成第一次更改后打印线程和pid编号。此外，您还会注意到它们与主线程和进程相同:尽管事件是由分离的线程引发的，回调运行在与我们的主程序相同的进程和线程中，从而提供了一种不想要的行为：
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信号量总是具有相同的行为，因此，最后，我决定在使用ReadDirectoryChangesW库管理和同步的单独进程中使用multiprocessing的同步配置来实现异步调用：
对get_directory_handle的调用不会返回由windows提供的句柄号，而是返回一个由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 retclose_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 rethttps://stackoverflow.com/questions/25669139
复制相似问题
腾讯云开发者
