异步ReadDirectoryChangesW调用阻止线程退出

Question

简介：

我正在编写一个小应用程序，用于监视某个目录中新添加的文件。

我想把监控代码放在一个单独的线程中，这样我就可以让主线程空闲于其他内容，并在需要时取消监视线程。

有关资料：

• 我正在使用ReadDirectoryChangesW进行监视
• 我使用原始的WIN32 API来创建/同步线程。
• 我试图继续支持Windows。

问题：

我能够正确地编码每件事，除了一件事：

我无法正确地退出监视线程，因此这篇文章。

我正在给主线程中的事件对象发送信号，等待线程退出，然后进行清理。

问题在于我对ReadDirectoryChangesW的使用，因为在我注释掉这段代码之后，一切都很好。

事件句柄发出信号后，ReadDirectoryChangesW阻塞线程，从而阻止它“捕获”事件并退出。如果我在目录中添加了一个新文件，它将“取消”ReadDirectoryChangesW，则线程“捕获”该事件并退出。

为了进一步的帮助，我在下面做了一个小的MVCE，这说明了我到目前为止所说的。

MVCE：

#include <iostream>
#include <Windows.h>
#include <map>

struct SThreadParams
{
    HANDLE hEvent;
    HANDLE hDir;
    int processDirectoryChanges(const char *buffer)
    {
        if (NULL == buffer) return -1;

        DWORD offset = 0;
        char fileName[MAX_PATH] = "";
        FILE_NOTIFY_INFORMATION *fni = NULL;

        do
        {
            fni = (FILE_NOTIFY_INFORMATION*)(&buffer[offset]);
            // since we do not use UNICODE, 
            // we must convert fni->FileName from UNICODE to multibyte
            int ret = ::WideCharToMultiByte(CP_ACP, 0, fni->FileName,
                fni->FileNameLength / sizeof(WCHAR),
                fileName, sizeof(fileName), NULL, NULL);

            switch (fni->Action)
            {
            case FILE_ACTION_ADDED:     
            {
                std::cout << "FILE_ACTION_ADDED " << fileName << std::endl;
            }
            break;
            case FILE_ACTION_REMOVED:
            {
                std::cout << "FILE_ACTION_REMOVED " << fileName << std::endl;
            }
            break;
            case FILE_ACTION_MODIFIED:
            {
                std::cout << "FILE_ACTION_MODIFIED " << fileName << std::endl;
            }
            break;
            case FILE_ACTION_RENAMED_OLD_NAME:
            {
                std::cout << "FILE_ACTION_RENAMED_OLD_NAME " << fileName << std::endl;
            }
            break;
            case FILE_ACTION_RENAMED_NEW_NAME:
            {
                std::cout << "FILE_ACTION_RENAMED_NEW_NAME " << fileName << std::endl;
            }
            break;
            default:
                break;
            }
            // clear string so we can reuse it
            ::memset(fileName, '\0', sizeof(fileName));
            // advance to next entry
            offset += fni->NextEntryOffset;

        } while (fni->NextEntryOffset != 0);

        return 0;
    }
};

DWORD WINAPI thread(LPVOID arg)
{
    SThreadParams p = *((SThreadParams *)arg);
    OVERLAPPED ovl = { 0 };
    DWORD bytesTransferred = 0, error = 0;
    char buffer[1024];

    if (NULL == (ovl.hEvent = ::CreateEvent(NULL, TRUE, FALSE, NULL)))
    {
        std::cout << "CreateEvent error = " << ::GetLastError() << std::endl;
        return ::GetLastError();
    };

    do {

        if (::ReadDirectoryChangesW(p.hDir, buffer, sizeof(buffer), FALSE,
            FILE_NOTIFY_CHANGE_FILE_NAME,
            NULL, &ovl, NULL))
        {
            if (::GetOverlappedResult(p.hDir, &ovl, &bytesTransferred, TRUE))
            {
                for (int i = 0; i < 5; ++i) std::cout << '=';
                std::cout << std::endl;

                if (-1 == p.processDirectoryChanges(buffer))
                    std::cout << "processDirectoryChanges error = " << std::endl;
            }
            else
            { 
                bytesTransferred = 0;
                std::cout << "GetOverlappedResult error = " << ::GetLastError() << std::endl;
            }

            if (0 == ::ResetEvent(ovl.hEvent))
            {
                std::cout << "ResetEvent error = " << ::GetLastError() << std::endl;
                ::CloseHandle(ovl.hEvent);
                return ::GetLastError();
            }
        }
        else
        {
            // we shall just output the error, and try again...
            std::cout << "ReadDirectoryChangesW error =  " << ::GetLastError() << std::endl;
        }

        error = ::WaitForSingleObject(p.hEvent, 2000);

    } while (WAIT_TIMEOUT == error);

    ::CloseHandle(ovl.hEvent);

    return 0;
}

int main()
{
    SThreadParams s;
    
    s.hDir = ::CreateFile(SOME_DIRECTORY,
            FILE_LIST_DIRECTORY, FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
            NULL, OPEN_EXISTING, FILE_FLAG_BACKUP_SEMANTICS, NULL);
    
    if (INVALID_HANDLE_VALUE == s.hDir)
    {
        std::cout << "CreateFile error = " << ::GetLastError() << std::endl;
        return 1;
    }

    s.hEvent = ::CreateEvent(NULL, TRUE, FALSE, NULL);
    
    if (NULL == s.hEvent)
    {
        std::cout << "CreateEvent error = " << ::GetLastError() << std::endl;
        ::CloseHandle(s.hDir);
        return 1;
    }
    
    HANDLE hThread = ::CreateThread(NULL, 0, thread, (LPVOID)&s, 0, NULL);
    
    if (NULL == hThread)
    {
        std::cout << "CreateThread error = " << ::GetLastError() << std::endl;
        ::CloseHandle(s.hDir);
        ::CloseHandle(s.hEvent);
        return 1;
    }

    std::cout << "press any key to close program..." << std::endl;
    std::cin.get();
    
    if (0 == ::CancelIoEx(s.hDir, NULL))
    {
        std::cout << "CancelIoEx error = " << ::GetLastError() << std::endl;
        ::CloseHandle(s.hDir);
        ::CloseHandle(s.hEvent);
        return 1;
    }

    if (0 == ::SetEvent(s.hEvent))
    {
        std::cout << "SetEvent error = " << ::GetLastError() << std::endl;
        ::CloseHandle(s.hDir);
        ::CloseHandle(s.hEvent);
        return 1;
    }

    // wait for thread to exit
    DWORD error = ::WaitForSingleObject(hThread, INFINITE);
    std::cout << "Thread exited with error code = " << error << std::endl;

    ::CloseHandle(s.hEvent);
    ::CloseHandle(s.hDir);
    ::CloseHandle(hThread);

    return 0;
}

我努力解决：

• 我已经将OVERLAPPED结构从线程移出到传递给线程的结构中。然后，我将OVERLAPPED.hEvent设置为强制“解除阻塞”ReadDirectoryChangesW。这似乎有效，但我感到害怕，因为我认为它不安全/容易出错，因为它是无文档的。
• 我尝试使用完成例程，但没有获得成功，因为我是新的这一切。我能够接收通知，但缓冲区(填充ReadDirectoryChangesW的缓冲区)的内容在第一次传递后没有正确读取。我仍在努力让这件事靠我自己来完成，但需要帮助。
• 我可以使用I/o完成端口，但是由于我只监视一个目录，我认为这有点过分。如果我弄错了，请指导我如何使用I/o完成端口来处理我的情况，我很想尝试一下。

问题：

考虑到上面提到的MVCE，您能告诉我如何修改线程过程中的代码吗?这样它就能正确地退出(没有ReadDirectoryChangesW阻塞)。

我有一种感觉，我将不得不使用完成例程。在这种情况下，我会谦逊地要求一些伪代码或书面指令，因为这将是我第一次使用它们。

每次我取得进展，我都会用相关数据更新这篇文章。

Answer 1

使用文件执行异步操作的exist 3种方法：

• 使用ApcRoutine
• 使用IoCompletionPort
• 使用事件-最坏的

你选择了最坏的变体。我在你家用IoCompletionPort。在这种情况下，您不需要创建事件、线程、调用GetOverlappedResult，不需要任何循环。

所有需要调用的文件上的BindIoCompletionCallback (或RtlSetIoCompletionCallback)！

关于取消-- XP中不存在CancelIoEx (“我试图支持Windows”)，但是您可以简单地关闭目录句柄--在这种情况下，IO将被STATUS_NOTIFY_CLEANUP取消。所以代码可以如下所示：

RUNDOWN_REF_EVENT g_rundown; // Run-Down Protection

class SPYDATA : 
#ifdef _USE_NT_VERSION_
    IO_STATUS_BLOCK
#else
    OVERLAPPED 
#endif
{
    HANDLE _hFile;
    LONG _dwRef;
    union {
        FILE_NOTIFY_INFORMATION _fni;
        UCHAR _buf[PAGE_SIZE];
    };

    void DumpDirectoryChanges()
    {
        union {
            PVOID buf;
            PBYTE pb;
            PFILE_NOTIFY_INFORMATION pfni;
        };

        buf = _buf;

        for (;;)
        {
            DbgPrint("%x <%.*S>\n", pfni->Action, pfni->FileNameLength >> 1, pfni->FileName);

            ULONG NextEntryOffset = pfni->NextEntryOffset;

            if (!NextEntryOffset)
            {
                break;
            }

            pb += NextEntryOffset;
        }
    }

#ifdef _USE_NT_VERSION_
    static VOID WINAPI _OvCompRoutine(
        _In_    NTSTATUS dwErrorCode,
        _In_    ULONG_PTR dwNumberOfBytesTransfered,
        _Inout_ PIO_STATUS_BLOCK Iosb
        )
    {
        static_cast<SPYDATA*>(Iosb)->OvCompRoutine(dwErrorCode, (ULONG)dwNumberOfBytesTransfered);
    }
#else
    static VOID WINAPI _OvCompRoutine(
        _In_    DWORD dwErrorCode, // really this is NTSTATUS
        _In_    DWORD dwNumberOfBytesTransfered,
        _Inout_ LPOVERLAPPED lpOverlapped
        )
    {
        static_cast<SPYDATA*>(lpOverlapped)->OvCompRoutine(dwErrorCode, dwNumberOfBytesTransfered);
    }
#endif

    VOID OvCompRoutine(NTSTATUS status, DWORD dwNumberOfBytesTransfered)
    {
        DbgPrint("[%x,%x]\n", status, dwNumberOfBytesTransfered);

        if (0 <= status) 
        {
            if (status != STATUS_NOTIFY_CLEANUP)
            {
                if (dwNumberOfBytesTransfered) DumpDirectoryChanges();
                DoRead();
            }
            else
            {
                DbgPrint("\n---- NOTIFY_CLEANUP -----\n");
            }
        }

        Release();
        g_rundown.ReleaseRundownProtection();
    }

    ~SPYDATA()
    {
        Cancel();
    }

public:

    void DoRead()
    {
        if (g_rundown.AcquireRundownProtection())
        {
            AddRef();
#ifdef _USE_NT_VERSION_
            NTSTATUS status = ZwNotifyChangeDirectoryFile(_hFile, 0, 0, this, this, &_fni, sizeof(_buf), FILE_NOTIFY_VALID_MASK, TRUE);
            if (NT_ERROR(status))
            {
                OvCompRoutine(status, 0);
            }
#else
            if (!ReadDirectoryChangesW(_hFile, _buf, sizeof(_buf), TRUE, FILE_NOTIFY_VALID_MASK, (PDWORD)&InternalHigh, this, 0))
            {
                OvCompRoutine(RtlGetLastNtStatus(), 0);
            }
#endif
        }
    }

    SPYDATA()
    {
        _hFile = 0;// ! not INVALID_HANDLE_VALUE because use ntapi for open file
        _dwRef = 1;
#ifndef _USE_NT_VERSION_
        RtlZeroMemory(static_cast<OVERLAPPED*>(this), sizeof(OVERLAPPED));
#endif
    }

    void AddRef()
    {
        InterlockedIncrement(&_dwRef);
    }

    void Release()
    {
        if (!InterlockedDecrement(&_dwRef))
        {
            delete this;
        }
    }

    BOOL Create(POBJECT_ATTRIBUTES poa)
    {
        IO_STATUS_BLOCK iosb;
        NTSTATUS status = ZwOpenFile(&_hFile, FILE_GENERIC_READ, poa, &iosb, FILE_SHARE_VALID_FLAGS, FILE_DIRECTORY_FILE);
        if (0 <= status)
        {
            return
#ifdef _USE_NT_VERSION_
            0 <= RtlSetIoCompletionCallback(_hFile, _OvCompRoutine, 0);
#else
            BindIoCompletionCallback(_hFile, _OvCompRoutine, 0);
#endif
        }
        return FALSE;
    }

    void Cancel()
    {
        if (HANDLE hFile = InterlockedExchangePointer(&_hFile, 0))
        {
            NtClose(hFile);
        }
    }
};

void DemoF()
{
    if (g_rundown.Create())
    {
        STATIC_OBJECT_ATTRIBUTES(oa, "\\systemroot\\tmp");//SOME_DIRECTORY

        if (SPYDATA* p = new SPYDATA)
        {
            if (p->Create(&oa))
            {
                p->DoRead();
            }

            MessageBoxW(0, L"wait close program...", L"", MB_OK);

            p->Cancel();

            p->Release();
        }

        g_rundown.ReleaseRundownProtection();
        g_rundown.WaitForRundown();
    }
}

等待所有IO完成时，我使用断电保护。不幸的是，这不是在用户模式下实现的，但也不是很难自己实现这个非常有用的特性。我的实施：

class __declspec(novtable) RUNDOWN_REF
{
    LONG _LockCount;

protected:

    virtual void RundownCompleted() = 0;

public:

    RUNDOWN_REF()
    {
        _LockCount = 1;
    }

    BOOL AcquireRundownProtection()
    {
        LONG LockCount = _LockCount, prevLockCount;

        do 
        {
            if (!LockCount)
            {
                return FALSE;
            }

            LockCount = InterlockedCompareExchange(&_LockCount, LockCount + 1, prevLockCount = LockCount);

        } while (LockCount != prevLockCount);

        return TRUE;
    }

    void ReleaseRundownProtection()
    {
        if (!InterlockedDecrement(&_LockCount))
        {
            RundownCompleted();
        }
    }
};

class RUNDOWN_REF_EVENT : public RUNDOWN_REF
{
    HANDLE _hEvent;

    virtual void RundownCompleted()
    {
        SetEvent(_hEvent);
    }

public:

    BOOL Create()
    {
        return (_hEvent = CreateEvent(0, TRUE, FALSE, 0)) != 0;
    }

    RUNDOWN_REF_EVENT()
    {
        _hEvent = 0;
    }

    ~RUNDOWN_REF_EVENT()
    {
        if (_hEvent) CloseHandle(_hEvent);
    }

    void WaitForRundown()
    {
        if (WaitForSingleObject(_hEvent, INFINITE) != WAIT_OBJECT_0) __debugbreak();
    }
};