ReadDirectoryChangesW question

[Elysia]

I thought I'd try to see if anyone is experienced with monitoring directories here...

So I have a basic system set up. ReadDirectoryChangesW to read directory changes.
The only problem I'm facing here is that it returns directly after a new file is added to a directory. If N files are moved into a directory, it will pretty much return N times.
This is problematic for me since I set off processing whatever files are dumped in asynchronously. And preferably, I want a total count of the files I'm processing, before I start processing (progress stuff).

So basically it is:
- Read directory change.
- Store file added.
- Wake processing thread.
- Read new directory chance.
- Store file added.
- And so on.

Ideally, I would like it to be just
- Read directory change.
- Store files added.
- Wake processing thread.

The only way I can think of doing this is by "peeking" to see if there are more changes waiting to be consumed by a call to ReadDirectoryChangesW since the function blocks until there's a new event.
Any ideas?

[Yarin]

Why not make an atomic FIFO with count? Should be easy. Think something like:

list changes;

changes_callback(str new_file)
{
  lock();
  changes.insert(0, new_file);
  if(changes.size == 1)
    wake_thread();
  unlock();
}

processing_thread()
{
  while(1)
  {
    str new_file;
    int num_left;
    lock();
    if(changes.size == 0)
    {
       unlock();
       sleep_thread();
    }
    new_file = changes.pop();
    num_left = changes.size;
    unlock();
    // PROCESS new_file
  }
}

my pseudocode here isn't good enough, there could be a race condition with the wake/sleep thread, but that could be fixed. It's the general idea.

[Elysia]

I don't see how it solves the problem? The problem is that when it finds a change, it wakes the thread and it prints progress 1 / 1. At the same time, the notifications push more files into the list, so at the next iteration of the processing thread, it will say something like progress 2 / 20. That's what I'd ideally like to avoid.

void SplittingThread()
{
	std::size_t i = 0;
	std::size_t VecSize;

	for (;;)
	{
		g_FileVectorLock.lock();
		VecSize = g_FileVector.size();
		g_FileVectorLock.unlock();

		for (; i < VecSize; i++)
		{
			std::wstring Title;
			std::wstring Body;
		
			g_FileVectorLock.lock();
			std::wstring File = g_FileVector.at(i);
			g_FileVectorLock.unlock();
			std::wstring ExamineDir = Stuff::Files::GetPath(File);

			{
				std::wstringstream ss;
				ss << L"Examining " << ExamineDir << L" (";
				ss << (int)(double(i) / VecSize * 100);
				ss << L"%)";
				Title = ss.str();
			}
			{
				std::wstringstream ss;

				ss << L"Processing file ";
				ss << i + 1 << L" of " << VecSize << L" (" << (int)(double(i) 
					/ VecSize * 100) << L"%):\n";
				ss << File;

				Body = ss.str();
			}

			TrayIcon.Modify(Body, Title);
			SplitFile(File); // TAKES A WHILE
		}

		g_FileVectorLock.lock();
		g_FileVector.clear();
		g_FileVectorLock.unlock();
		SuspendThread(g_pSplittingThread->native_handle());
	}
}

void MonitorDirectory(const std::wstring& DirToMonitor)
{
	//...
	while ( ReadDirectoryChangesW(hFolder, szBuffer, sizeof(szBuffer),
		TRUE, FILE_NOTIFY_CHANGE_FILE_NAME, &dwBytes, nullptr, nullptr) )
	{
		//...
		boost::lock_guard<boost::mutex> lock(g_FileVectorLock);
		g_FileVector.push_back(InFilename);
		//...
		ResumeThread(g_pSplittingThread->native_handle());
	}
}

I'm certainly open to other ideas on how to this more efficiently / better.
Basically SplitFile will take a while, and is a synchronous operation (of course). So by the first iteration, the size will be 1. The next iteration, the size will be >1 (having been populated fully by the changes).

[adeyblue]

The only way I can think of doing this is by "peeking" to see if there are more changes waiting to be consumed by a call to ReadDirectoryChangesW since the function blocks until there's a new event.

Only the synchronous usage blocks, use the async option instead and:

Wait on the overlapped event handle for x amount of time:
   If it gets signalled:
     save the filename
     increment the count
     ResetEvent the event
     call ReadDirectoryChanges again
     go back to waiting
   If it times out:
     there aren't any more operations so process those you have, if any
     go back to waiting

I'd also have the processing thread wait on an exit event and a semaphore rather than using SuspendThread, that just seems overly hacky to me.

[Elysia]

So (not being terribly good at Windows async I/O), there is a possibility of waiting for an event which will become signaled once there one or more events available?
Or is it just polling?

Re SuspendThread: I don't know, I don't find it very hacky. It seems intuitive. Process the work. Got o sleep. Then wake it when there's more work to do. Semaphores and mutexes are for protecting critical data sections. An event might do the trick, but that adds just another layer of complexity as far as boost goes, I think. Not sure on how to do that with boost yet, but initial research didn't look promising.

[Codeplug]

Don't use SuspendThread unless you're writing a debugger. The only way to use it otherwise is if the thread suspends itself. And that's hacky in my opinion.

Boost provides a lower-level set of primitives than what Windows provides. For example you can make a Win32 Event using a condition variable and a mutex. If you want to multiplex the handling of a set of these [homegrown] events, you'll have to write that code yourself.

Windows already has this built-in with WaitForMultipleObjects (WFMO). When it comes to systems level programming in Windows, trying not use the Windows paradigm (WFMO/WFSO) will only make things more difficult and less efficient. Boost doesn't really help in this area.

>> Or is it just polling?
OVERLAPPED I/O just waits on an event. But it looks like you may be able to use the easier API, FindFirstChangeNotificationW.

gg

[Elysia]

OK thanks. Good stuff.
Looks like I need to look into events using boost. Or I could just use Windows events with a custom deleter. Simple enough.
FindFirstChangeNotificationW looks like it might just do the trick! Exactly what I need. Thanks!

[Elysia]

I am confused...
The documentation says that I can wait on the handle returned by FindFirstChangeNotificationW. It should signal the handle when a change is detected. Right. So that works.
The documentation also says that I should call ReadDirectoryChangesW to get the actual events that transpired. But here is the problem. Even though the handle is signaled, ReadDirectoryChangesW blocks! Indeed, when I cause another change, it returns.
What could be wrong?

My WIP code is so far:

	hNotificationsWaiting[i].Attach( FindFirstChangeNotification(g_Dir[i].c_str(), TRUE, FILE_NOTIFY_CHANGE_FILE_NAME) );


	for (;;)
	{
		WaitForSingleObject(hNotificationsWaiting.get(), INFINITE);
		assert( FindNextChangeNotification(hNotificationsWaiting.get()) );
		while ( WaitForSingleObject(hNotificationsWaiting.get(), 0) == WAIT_OBJECT_0 )
		{
			do
			{
				ReadDirectoryChangesW(hFolder.get(), szBuffer, sizeof(szBuffer), TRUE,
					FILE_NOTIFY_CHANGE_FILE_NAME, &dwBytes, nullptr, nullptr);

				// Get a pointer to the first change record...
				pInfo = (FILE_NOTIFY_INFORMATION*)&szBuffer[dwOffset];

				if (pInfo->Action == FILE_ACTION_ADDED)
				{
					std::wstring InFilename = DirToMonitor + L"\\";
					std::copy(pInfo->FileName, pInfo->FileName +
						(pInfo->FileNameLength / sizeof(pInfo->FileName[0])),
						std::back_inserter(InFilename));
					boost::lock_guard<boost::mutex> lock(g_FileVectorLock);
					g_FileVector.push_back(InFilename);
					//WakeThread = true;
				}

				// More than one change may happen at the same time.  Load the next change and continue...
				dwOffset += pInfo->NextEntryOffset;
			}
			while (pInfo->NextEntryOffset != 0);
		}
		if (WakeThread)
			SetEvent(g_hResumeEvent.get());
			//ResumeThread(g_pSplittingThread->native_handle());
	}

[Codeplug]

Well, I was implying not calling ReadDirectoryChangesW at all. But since your monitoring a tree, trying to figure out what actually changed may be more trouble than it's worth. So stick with just ReadDirectoryChangesW using overlapped I/O.

Overlapped I/O isn't too hard. Create a manual reset event and set it in an OVERLAPPED structure. Pass that into ReadDirectoryChangesW - a non-zero return means that the event is ready to be waited on. Once WFSO/WFMO says the event is signaled, call GetOverlappedResult and szBuffer will contain your data.

In your WFxO loop, use an INFINITE timeout initially. The first time your directory change event is signaled, set a time out of 1 second (or so). During this 1 second window, your directory change event may be signaled several times - accumulate all those file names locally. When the 1 second window is up (WFxO returns WAIT_TIMEOUT), take what you've accumulated and pass it off to the other thread - and reset your timeout back to INFINITE.

In the consumer thread, call g_FileVector.swap() with an empty vector to efficiently consume its entire contents. During WAIT_TIMEOUT in the producer thread, call g_FileVector.insert() to add everything you've accumulated during that time-window.

The "accumulation window" could be implemented on the consumer side - but doing it on the producer side will reduce lock contention, and the consumer won't go in and out of kernel wait Q's as often (fewer context switches).

gg

[Elysia]

Another step forward. Another step backwards?
Test code:

	OVERLAPPED IoResult = {};
	IoResult.hEvent = CreateEvent(nullptr, TRUE, FALSE, L"");

	char szBuffer[640] = {};
	DWORD dwBytes;
	Stuff::RAII::CHandle hFolder( CreateFileW(g_Dir[1].c_str(), FILE_LIST_DIRECTORY, FILE_SHARE_READ, 
		nullptr, OPEN_EXISTING, FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OVERLAPPED, nullptr) );
	assert( ReadDirectoryChangesW(hFolder.get(), szBuffer, sizeof(szBuffer), TRUE, 
		FILE_NOTIFY_CHANGE_FILE_NAME, nullptr, &IoResult, nullptr) );

	WaitForSingleObject(IoResult.hEvent, INFINITE);

	DWORD dwOffset = 0;
	FILE_NOTIFY_INFORMATION* pInfo = nullptr;
	pInfo = (FILE_NOTIFY_INFORMATION*)&szBuffer[dwOffset];
	while (WaitForSingleObject(IoResult.hEvent, 1) == WAIT_OBJECT_0)
	{
		do
		{
			assert( GetOverlappedResult(hFolder.get(), &IoResult, &dwBytes, TRUE) );

			// Get a pointer to the first change record...
			pInfo = (FILE_NOTIFY_INFORMATION*)&szBuffer[dwOffset];

			if (pInfo->Action == FILE_ACTION_ADDED)
			{
				std::wstring InFilename = g_Dir[1] + L"\\";
				std::copy(pInfo->FileName, pInfo->FileName + (pInfo->FileNameLength /
					sizeof(pInfo->FileName[0])), std::back_inserter(InFilename));
				boost::lock_guard<boost::mutex> lock(g_FileVectorLock);
				g_FileVector.push_back(InFilename);
			}
			// More than one change may happen at the same time.
			// Load the next change and continue...
			dwOffset += pInfo->NextEntryOffset;
		}
		while (pInfo->NextEntryOffset != 0);
	}

Problem: Only the first notification will be processed. The contents of the buffer never changes between the calls to GetOverlappedResult.
Sigh. The documentation never documents this stuff.

Any idea what's wrong or how to make it work?
Thanks.

[Codeplug]

Here's how I would do it:

#include <windows.h>
#include <iostream>
#include <deque>
#include <vector>
#include <string>
#include <stdexcept>

//------------------------------------------------------------------------------

struct no_copy
{
protected:
    no_copy() {}
    ~no_copy() {}
private:
    no_copy(const no_copy&);
    const no_copy& operator=(const no_copy&);
};//no_copy

//------------------------------------------------------------------------------

class scoped_HANDLE : public no_copy
{
    HANDLE m_h;
public:
    explicit scoped_HANDLE(HANDLE h) : m_h(h) {}
    ~scoped_HANDLE() {if (m_h && m_h != INVALID_HANDLE_VALUE) ::CloseHandle(m_h);}
};//scoped_HANDLE

//------------------------------------------------------------------------------

class Event : public no_copy
{
    HANDLE m_handle;
public:
    Event() : m_handle(0) {}
    ~Event() {Close();}

    operator HANDLE() {return m_handle;}

    void Close() {if (m_handle) ::CloseHandle(m_handle); m_handle = 0;}

    bool Create(bool manualReset = false, bool initialOwner = false,
                const wchar_t *name = 0, LPSECURITY_ATTRIBUTES sa = 0)
    {
        m_handle = ::CreateEventW(sa, manualReset, initialOwner, name);
        return m_handle != 0;
    }//Create

    bool Signal() {return ::SetEvent(m_handle) != FALSE;}
    bool Reset() {return ::ResetEvent(m_handle) != FALSE;}
};//Event

//------------------------------------------------------------------------------

template<class sync_t>
class scoped_lock : public no_copy
{
    sync_t &m_sync;
public:
    explicit scoped_lock(sync_t &s) : m_sync(s) {m_sync.Acquire();}
    ~scoped_lock() {m_sync.Release();}
};//scoped_lock

//------------------------------------------------------------------------------

class CriticalSection : public no_copy
{
    CRITICAL_SECTION cs;
public:
    typedef scoped_lock<CriticalSection> scoped_lock;

    CriticalSection() {::InitializeCriticalSection(&cs);}
    ~CriticalSection() {::DeleteCriticalSection(&cs);}
    void Acquire() {::EnterCriticalSection(&cs);}
    void Release() {::LeaveCriticalSection(&cs);}
};//CriticalSection

//------------------------------------------------------------------------------

template <typename T>
class queue_MT
{
    mutable CriticalSection m_cs;
    mutable Event m_consumerEvent;

    typedef CriticalSection::scoped_lock lock;

    std::deque<T> m_queue;

public:
    queue_MT()
    {
        if (!m_consumerEvent.Create(true)) // manual reset
            throw std::runtime_error("queue_MT: failed to create event");
    }//constructor

    HANDLE consumer_event() const {return m_consumerEvent;}

    void enqueue(const T& val)
    {
        lock l(m_cs); 
        m_queue.push_front(val);
        m_consumerEvent.Signal();
    }//enqueue

    template<class It>
    void enqueue(It first, It last)
    {
        lock l(m_cs); 
        m_queue.insert(m_queue.begin(), first, last);
        m_consumerEvent.Signal();
    }//enqueue

    bool dequeue(T &val)
    {
        lock l(m_cs);
        if (m_queue.size() == 0)
            return false;

        val = m_queue.back();
        m_queue.pop_back();

        if (m_queue.size() == 0)
            m_consumerEvent.Reset();

        return true;
    }//dequeue

    void dequeue_all(std::deque<T> &q)
    {
        lock l(m_cs);
        m_queue.swap(q);
        m_queue.clear();
        m_consumerEvent.Reset();
    }//dequeue_all
};//class queue_MT

//------------------------------------------------------------------------------

bool EnablePrivilege(LPCTSTR name)
{
    HANDLE hToken; 
    TOKEN_PRIVILEGES tkp; 
    
    tkp.PrivilegeCount = 1;
    tkp.Privileges[0].Attributes = SE_PRIVILEGE_ENABLED; 
 
    if (!OpenProcessToken(GetCurrentProcess(), 
                          TOKEN_ADJUST_PRIVILEGES | TOKEN_QUERY, 
                          &hToken) ||
        !LookupPrivilegeValue(0, name, &tkp.Privileges[0].Luid) ||
        !AdjustTokenPrivileges(hToken, FALSE, &tkp, 0, 0, 0)) 
    {
        return false;
    }//if

    return true;
}//EnablePrivilege

//------------------------------------------------------------------------------

struct SharedData : public no_copy
{
    std::wstring m_rootpath; // producer only
    Event m_exit;
    queue_MT<std::wstring> m_q;

    SharedData() 
    {
        if (!m_exit.Create(true)) // manual reset
            throw std::runtime_error("SharedData: failed to create event");
    }//constructor
};//SharedData

//------------------------------------------------------------------------------
//------------------------------------------------------------------------------

using namespace std;

//------------------------------------------------------------------------------

DWORD WINAPI ConsumerThread(void *p)
{
    SharedData *psd = reinterpret_cast<SharedData*>(p);

    const DWORD num_objs = 2;
    const HANDLE wait_objs[] = {psd->m_exit, psd->m_q.consumer_event()};
    enum
    {
        WO_EXIT = WAIT_OBJECT_0 + 0,
        WO_CONSUME = WAIT_OBJECT_0 + 1,
    };

    for (;;)
    {
        DWORD status = WaitForMultipleObjects(num_objs, wait_objs, 
                                              FALSE, INFINITE);
        if (status == WO_EXIT)
            break;

        if (status == WO_CONSUME)
        {
            deque<wstring> work_q;
            psd->m_q.dequeue_all(work_q);

            deque<wstring>::const_iterator it = work_q.begin(), 
                                           it_end = work_q.end();
            for (; it != it_end; ++it)
            {
                wcout << *it << endl;
                if (*it == L"exit")
                    psd->m_exit.Signal();
            }//for
        }//if
        else
        {
            wcerr << L"Consumer: Unexpected WFMO result, " << status << endl;
            psd->m_exit.Signal();
            return 1;
        }//else
    }//for
    
    return 0;
}//ConsumerThread

//------------------------------------------------------------------------------

DWORD WINAPI ProducerThread(void *p)
{
    SharedData *psd = reinterpret_cast<SharedData*>(p);
    vector<wstring> work_q;

    HANDLE hDir = 
        CreateFileW(psd->m_rootpath.c_str(), 
                    FILE_LIST_DIRECTORY, 
                    FILE_SHARE_READ | FILE_SHARE_WRITE, 
                    0, OPEN_EXISTING, 
                    FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OVERLAPPED, 0);
    if (!hDir)
    {
        wcerr << L"CreateFile failed on " << psd->m_rootpath << L", LE = " 
              << GetLastError() << endl;
        psd->m_exit.Signal();
        return 1;
    }//if

    scoped_HANDLE sh(hDir);

    OVERLAPPED os = {0};
    Event osev;
    if (!osev.Create(true))
    {
        psd->m_exit.Signal();
        return 1;
    }//if

    os.hEvent = osev;

    union
    {
        FILE_NOTIFY_INFORMATION fni;
        char buff[1024 * 4];

        FILE_NOTIFY_INFORMATION* NextEntry(FILE_NOTIFY_INFORMATION *pfni)
        {
            if (!pfni->NextEntryOffset)
                return 0;
            char *p = reinterpret_cast<char*>(pfni) + pfni->NextEntryOffset;
            return reinterpret_cast<FILE_NOTIFY_INFORMATION*>(p);
        }//NextEntry
    } u;
    
    if (!ReadDirectoryChangesW(hDir, &u, sizeof(u), TRUE, 
                               FILE_NOTIFY_CHANGE_FILE_NAME, 0, &os, 0))
    {
        wcerr << L"ReadDirectoryChangesW failed, LE = " 
              << GetLastError() << endl;
        psd->m_exit.Signal();
        return 1;
    }//if

    const DWORD num_objs = 2;
    const HANDLE wait_objs[] = {psd->m_exit, osev};
    enum
    {
        WO_EXIT = WAIT_OBJECT_0 + 0,
        WO_DIR  = WAIT_OBJECT_0 + 1,
    };

    DWORD timeout = INFINITE;
    const DWORD window = 1000;
    DWORD nread;

    for (;;)
    {
        DWORD status = WaitForMultipleObjects(num_objs, wait_objs, 
                                              FALSE, timeout);
        if (status == WO_EXIT)
        {
            CancelIo(hDir);
            GetOverlappedResult(hDir, &os, &nread, TRUE);
            break;
        }//if

        if (status == WO_DIR)
        {
            if (!GetOverlappedResult(hDir, &os, &nread, TRUE))
            {
                wcerr << L"GetOverlappedResult failed, LE = " 
                      << GetLastError() << endl;
                psd->m_exit.Signal();
                return 1;
            }//if

            FILE_NOTIFY_INFORMATION *pfni = &u.fni;
            for (; pfni; pfni = u.NextEntry(pfni))
            {
                work_q.push_back(wstring());
                work_q.back().assign(
                        pfni->FileName,
                        pfni->FileName + (pfni->FileNameLength / 2));
            }//for

            if (!ReadDirectoryChangesW(hDir, &u, sizeof(u), TRUE, 
                                       FILE_NOTIFY_CHANGE_FILE_NAME, 0, &os, 0))
            {
                wcerr << L"ReadDirectoryChangesW failed, LE = " 
                      << GetLastError() << endl;
                psd->m_exit.Signal();
                return 1;
            }//if

            timeout = window;
        }//if
        else if (status == WAIT_TIMEOUT)
        {
            psd->m_q.enqueue(work_q.begin(), work_q.end());
            work_q.clear();
            timeout = INFINITE;
        }//else if
        else
        {
            wcerr << L"Producer: Unexpected WFMO result, " << status << endl;
            psd->m_exit.Signal();
            return 1;
        }//else
    }//for

    return 0;
}//ProducerThread

//------------------------------------------------------------------------------

int main()
{
    if (!EnablePrivilege(SE_DEBUG_NAME) ||
        !EnablePrivilege(SE_RESTORE_NAME))
        return 1;

    SharedData data;
    data.m_rootpath = L"C:\\temp";
        
    HANDLE hCon = CreateThread(0, 0, &ConsumerThread, &data, 0, 0);
    if (!hCon)
        return 1;

    HANDLE hProd = CreateThread(0, 0, &ProducerThread, &data, 0, 0);
    if (!hProd)
    {
        data.m_exit.Signal();
        WaitForSingleObject(hCon, INFINITE);
        return 1;
    }//if

    const DWORD num_objs = 2;
    const HANDLE wait_objs[num_objs] = {hCon, hProd};
    WaitForMultipleObjects(num_objs, wait_objs, TRUE, INFINITE);

    cout << "Done" << endl;
    return 0;
}//main

Need to make sure your buffer is DWORD aligned.
Need to re-issue another overlapped read before waiting again.
assert() expressions don't run in the release build.

gg

[Elysia]

I've finally gotten it to work. Though I am probably going to continue tweaking it and finally creating a framework or wrapper for it.
Anyway, I did notice something. Apparently GetOverlappedResult does nothing. In fact, to continue getting the changes that's been made, I have to keep calling ReadDirectoryChangesW.
Though, that said, if we move on, is there a good way of knowing when a file is closed or opened? I am possibly looking at handling situations where a file is copied into a watched directory. The file must be fully copied/written and closed. Since I'm splitting (media) files, the file must be whole and I must have access to it (to delete the original after splitting).
Looking at the documentation, it seems there is an option for last write, which usually occurs when the file is closed or when the buffer is flushed. I don't think I can trust this, though.
Ideas for that is welcome.

Thanks.

[Codeplug]

>> Apparently GetOverlappedResult does nothing.
It tells you the number of bytes trasfered - not needed in this case.
It tells you when the event becomes signaled (when last param is TRUE) - not needed when WFMO tells you it's signaled.
It tells you if the overlapped read failed or succeeded (event is signaled either way).

So if you want to remove error checking, don't call it

>> I have to keep calling ReadDirectoryChangesW
That's how all overlapped IO works. Once an asynchronous operation has been satisfied, you have to start another.

>> is there a good way of knowing when a file is closed or opened?
I would just try to open the file for exclusive access. Try again later if you can't.

gg

[Elysia]

Ah, of course. Always a catch, eh?
But polling? There is no other way besides polling?

[Codeplug]

I'm not aware of another method that tells you that you can have exclusive ownership of a file. Trying again a couple times a second won't hurt.

gg