c++ - 异步 ReadDirectoryChangesW 调用阻止线程退出

Question

简介:

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

我想将监控代码放在一个单独的线程中，这样我就可以让主线程空闲下来处理其他事情，并在需要时取消监控线程。

相关信息:

• 我正在使用 ReadDirectoryChangesW做监控
• 我正在使用原始 WIN32 API 进行线程创建/同步
• 我正在努力支持 Windows XP。

问题:

我能够正确地编写所有代码，除了一件事:

我无法正确退出监控线程，因此出现了这篇文章。

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

问题出在我对 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

存在 3 种对文件进行异步操作的方法:

• 使用 ApcRoutine
• 使用 IoCompletionPort
• 使用事件 - 最差

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

所有需要电话BindIoCompletionCallback (或 RtlSetIoCompletionCallback)文件和所有!

关于取消 - CancelIoEx在 XP 中不存在(“我试图支持 Windows XP”),但您可以简单地关闭目录句柄 - 在这种情况下，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 完成后我使用 Run-Down Protection .不幸的是，这不是在用户模式下实现的，但你自己并不难实现这个非常有用的功能。我的实现:

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();
    }
};