debian-packages/eventcore
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Fix shutdown procedure and rename event order for the polling watcher.

Browse source 
Renames should always be reported as removed->added instead of added->removed.
This commit is contained in:
Sönke Ludwig 2017-11-23 02:41:47 +01:00
parent 1d4fbc4fa2
commit 19879712e6
1 changed files with 37 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

50
source/eventcore/drivers/posix/watchers.d
View file

@ -216,7 +216,22 @@ final class PollEventDriverWatchers(Events : EventDriverEvents) : EventDriverWat
PollingThread[EventID] m_pollers; PollingThread[EventID] m_pollers;
} }
this(Events events) { m_events = events; } this(Events events)
{
m_events = events;
}
void dispose()
@trusted {
foreach (pt; m_pollers.byValue) {
pt.dispose();
try pt.join();
catch (Exception e) {
// not bringing down the application here, because not being
// able to join the thread here likely isn't a problem
}
}
}
final override WatcherID watchDirectory(string path, bool recursive, FileChangesCallback on_change) final override WatcherID watchDirectory(string path, bool recursive, FileChangesCallback on_change)
{ {
@ -280,6 +295,7 @@ final class PollEventDriverWatchers(Events : EventDriverEvents) : EventDriverWat
pt.m_callback(cast(WatcherID)evt, ch); pt.m_callback(cast(WatcherID)evt, ch);
} }
private final class PollingThread : Thread { private final class PollingThread : Thread {
int refCount = 1; int refCount = 1;
EventID changesEvent; EventID changesEvent;
@ -291,7 +307,6 @@ final class PollEventDriverWatchers(Events : EventDriverEvents) : EventDriverWat
immutable string m_basePath; immutable string m_basePath;
immutable bool m_recursive; immutable bool m_recursive;
immutable FileChangesCallback m_callback; immutable FileChangesCallback m_callback;
shared bool m_shutdown = false;
size_t m_entryCount; size_t m_entryCount;
struct Entry { struct Entry {
@ -337,8 +352,6 @@ final class PollEventDriverWatchers(Events : EventDriverEvents) : EventDriverWat
void dispose() void dispose()
nothrow { nothrow {
import core.atomic : atomicStore;
try synchronized (m_changesMutex) { try synchronized (m_changesMutex) {
changesEvent = EventID.invalid; changesEvent = EventID.invalid;
} catch (Exception e) assert(false, e.msg); } catch (Exception e) assert(false, e.msg);
@ -346,21 +359,25 @@ final class PollEventDriverWatchers(Events : EventDriverEvents) : EventDriverWat
private void run() private void run()
nothrow @trusted { nothrow @trusted {
import core.atomic : atomicLoad;
import core.time : msecs; import core.time : msecs;
import std.algorithm.comparison : min; import std.algorithm.comparison : min;
import std.datetime : Clock, UTC;
try while (true) { try while (true) {
() @trusted { Thread.sleep(min(m_entryCount, 60000).msecs + 1000.msecs); } (); auto timeout = Clock.currTime(UTC()) + min(m_entryCount, 60000).msecs + 1000.msecs;
while (true) {
try synchronized (m_changesMutex) { try synchronized (m_changesMutex) {
if (changesEvent == EventID.invalid) break; if (changesEvent == EventID.invalid) return;
} catch (Exception e) assert(false, "Mutex lock failed: "~e.msg); } catch (Exception e) assert(false, "Mutex lock failed: "~e.msg);
auto remaining = timeout - Clock.currTime(UTC());
if (remaining <= 0.msecs) break;
sleep(min(1000.msecs, remaining));
}
scan(true); scan(true);
try synchronized (m_changesMutex) { try synchronized (m_changesMutex) {
if (changesEvent == EventID.invalid) break; if (changesEvent == EventID.invalid) return;
if (m_changes.length) if (m_changes.length)
m_eventsDriver.trigger(changesEvent, false); m_eventsDriver.trigger(changesEvent, false);
} catch (Exception e) assert(false, "Mutex lock failed: "~e.msg); } catch (Exception e) assert(false, "Mutex lock failed: "~e.msg);
@ -385,9 +402,10 @@ final class PollEventDriverWatchers(Events : EventDriverEvents) : EventDriverWat
import std.algorithm.mutation : swap; import std.algorithm.mutation : swap;
Entry*[Key] new_entries; Entry*[Key] new_entries;
Entry*[] added;
size_t ec = 0; size_t ec = 0;
scan(null, generate_changes, new_entries, ec); scan(null, generate_changes, new_entries, added, ec);
foreach (e; m_entries.byKeyValue) { foreach (e; m_entries.byKeyValue) {
if (!e.key.parent || Key(e.key.parent.parent, e.key.parent.name) !in m_entries) { if (!e.key.parent || Key(e.key.parent.parent, e.key.parent.name) !in m_entries) {
@ -397,11 +415,14 @@ final class PollEventDriverWatchers(Events : EventDriverEvents) : EventDriverWat
delete e.value; delete e.value;
} }
foreach (e; added)
addChange(FileChangeKind.added, Key(e.parent, e.name), e.isDir);
swap(m_entries, new_entries); swap(m_entries, new_entries);
m_entryCount = ec; m_entryCount = ec;
} }
private void scan(Entry* parent, bool generate_changes, ref Entry*[Key] new_entries, ref size_t ec) private void scan(Entry* parent, bool generate_changes, ref Entry*[Key] new_entries, ref Entry*[] added, ref size_t ec)
@trusted nothrow { @trusted nothrow {
import std.file : SpanMode, dirEntries; import std.file : SpanMode, dirEntries;
import std.path : buildPath, baseName; import std.path : buildPath, baseName;
@ -413,7 +434,7 @@ final class PollEventDriverWatchers(Events : EventDriverEvents) : EventDriverWat
if (auto pe = key in m_entries) { if (auto pe = key in m_entries) {
if ((*pe).isDir) { if ((*pe).isDir) {
if (m_recursive) if (m_recursive)
scan(*pe, generate_changes, new_entries, ec); scan(*pe, generate_changes, new_entries, added, ec);
} else { } else {
if ((*pe).size != de.size || (*pe).lastChange != modified_time) { if ((*pe).size != de.size || (*pe).lastChange != modified_time) {
if (generate_changes) if (generate_changes)
@ -430,10 +451,9 @@ final class PollEventDriverWatchers(Events : EventDriverEvents) : EventDriverWat
auto e = new Entry(parent, key.name, de.isDir ? ulong.max : de.size, modified_time); auto e = new Entry(parent, key.name, de.isDir ? ulong.max : de.size, modified_time);
new_entries[key] = e; new_entries[key] = e;
ec++; ec++;
if (generate_changes) if (generate_changes) added ~= e;
addChange(FileChangeKind.added, key, e.isDir);
if (de.isDir && m_recursive) scan(e, false, new_entries, ec); if (de.isDir && m_recursive) scan(e, false, new_entries, added, ec);
} }
} catch (Exception e) {} // will result in all children being flagged as removed } catch (Exception e) {} // will result in all children being flagged as removed
} }