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

Separate the polling logic of PollEventDriverWatchers into a separate class.

Browse source 
Allows to keep thread-local state separate from the mutex-protected/immutable state in PollingThread to make sure that there are no unwanted accesses.
This commit is contained in:
Sönke Ludwig 2019-05-30 11:33:04 +02:00
parent 83310fda70
commit 731936ce9e
1 changed files with 64 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -332,40 +332,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;
final static class Entry {
Entry parent;
string name;
ulong size;
long lastChange;
this(Entry parent, string name, ulong size, long lastChange)
{
this.parent = parent;
this.name = name;
this.size = size;
this.lastChange = lastChange;
}
string path()
{
import std.path : buildPath;
if (parent)
return buildPath(parent.path, name);
else return name;
}
bool isDir() const { return size == ulong.max; }
}
struct Key {
Entry parent;
string name;
}
// used only within the polling thread
Entry[Key] m_entries;
size_t m_entryCount;
} }
this(shared(Events) event_driver, EventID event, string path, bool recursive, FileChangesCallback callback) this(shared(Events) event_driver, EventID event, string path, bool recursive, FileChangesCallback callback)
@ -378,7 +344,6 @@ final class PollEventDriverWatchers(Events : EventDriverEvents) : EventDriverWat
m_basePath = path; m_basePath = path;
m_recursive = recursive; m_recursive = recursive;
m_callback = callback; m_callback = callback;
scan(false);
try super(&run); try super(&run);
catch (Exception e) assert(false, e.msg); catch (Exception e) assert(false, e.msg);
@ -407,8 +372,16 @@ final class PollEventDriverWatchers(Events : EventDriverEvents) : EventDriverWat
import core.time : MonoTime, msecs; import core.time : MonoTime, msecs;
import std.algorithm.comparison : min; import std.algorithm.comparison : min;
auto poller = new DirectoryPoller(m_basePath, m_recursive, (ch) {
try synchronized (m_changesMutex) {
m_changes ~= ch;
} catch (Exception e) assert(false, "Mutex lock failed: "~e.msg);
});
poller.scan(false);
try while (true) { try while (true) {
auto timeout = MonoTime.currTime() + min(m_entryCount, 60000).msecs + 1000.msecs; auto timeout = MonoTime.currTime() + min(poller.entryCount, 60000).msecs + 1000.msecs;
while (true) { while (true) {
try synchronized (m_changesMutex) { try synchronized (m_changesMutex) {
if (m_changesEvent == EventID.invalid) return; if (m_changesEvent == EventID.invalid) return;
@ -418,7 +391,7 @@ final class PollEventDriverWatchers(Events : EventDriverEvents) : EventDriverWat
sleep(min(1000.msecs, remaining)); sleep(min(1000.msecs, remaining));
} }
scan(true); poller.scan(true);
try synchronized (m_changesMutex) { try synchronized (m_changesMutex) {
if (m_changesEvent == EventID.invalid) return; if (m_changesEvent == EventID.invalid) return;
@ -434,11 +407,62 @@ final class PollEventDriverWatchers(Events : EventDriverEvents) : EventDriverWat
} }
} }
}
private final class DirectoryPoller {
private final static class Entry {
Entry parent;
string name;
ulong size;
long lastChange;
this(Entry parent, string name, ulong size, long lastChange)
{
this.parent = parent;
this.name = name;
this.size = size;
this.lastChange = lastChange;
}
string path()
{
import std.path : buildPath;
if (parent)
return buildPath(parent.path, name);
else return name;
}
bool isDir() const { return size == ulong.max; }
}
private struct Key {
Entry parent;
string name;
}
alias ChangeCallback = void delegate(FileChange) @safe nothrow;
private {
immutable string m_basePath;
immutable bool m_recursive;
Entry[Key] m_entries;
size_t m_entryCount;
ChangeCallback m_onChange;
}
this(string path, bool recursive, ChangeCallback on_change)
{
m_basePath = path;
m_recursive = recursive;
m_onChange = on_change;
}
@property size_t entryCount() const { return m_entryCount; }
private void addChange(FileChangeKind kind, Key key, bool is_dir) private void addChange(FileChangeKind kind, Key key, bool is_dir)
{ {
try synchronized (m_changesMutex) { m_onChange(FileChange(kind, m_basePath, key.parent ? key.parent.path : "", key.name, is_dir));
m_changes ~= FileChange(kind, m_basePath, key.parent ? key.parent.path : "", key.name, is_dir);
} catch (Exception e) assert(false, "Mutex lock failed: "~e.msg);
} }
private void scan(bool generate_changes) private void scan(bool generate_changes)