Асинхронного 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) не было правильно прочитано после первого прохода. Я все еще пытаюсь сделать эту работу самостоятельно, но могу воспользоваться помощью.

• Я мог бы использовать порт завершения ввода-вывода, но поскольку я буду контролировать только один каталог, я думаю, что это немного перебор. Если я ошибаюсь, пожалуйста, сообщите мне, как использовать порт завершения ввода-вывода для моего случая, я бы хотел попробовать их.

ВОПРОС:

Учитывая MVCE выше, вы можете поручить мне о том, как изменить код в процедуре потока, так что выходит правильно (без ReadDirectoryChangesW блокировки).

У меня возникло ощущение, что мне придется использовать процедуры завершения. В этом случае я смиренно прошу ввести псевдокод или письменные инструкции, так как это будет мой первый раз, используя их.

Каждый раз, когда я чувствую, что достиг прогресса, я соответствующим образом обновляю это сообщение соответствующими данными.

Если вам нужна дополнительная информация/пояснения, оставьте комментарий, и я отвечу.

Спасибо,

С уважением.

Answer 1

существует 3 способа сделать асинхронные операции с файлами:

• использование ApcRoutine
• использование IoCompletionPort
• использование Event - худший

Вы выбираете худший вариант. Я буду использовать 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(); 
    } 
};