summaryrefslogtreecommitdiffstats
path: root/src/server/PreProcessor.cpp
diff options
context:
space:
mode:
authorDavid Robillard <d@drobilla.net>2016-09-18 21:29:30 -0400
committerDavid Robillard <d@drobilla.net>2016-09-29 16:04:19 -0400
commita7f83c70733288ee9efee5c08330fbdf638446db (patch)
tree33c1d507d39a2a642f20498a52eb0c5225cfde43 /src/server/PreProcessor.cpp
parentb40326d759a9b4b2db7eb930f9a32403d4a9227f (diff)
downloadingen-a7f83c70733288ee9efee5c08330fbdf638446db.tar.gz
ingen-a7f83c70733288ee9efee5c08330fbdf638446db.tar.bz2
ingen-a7f83c70733288ee9efee5c08330fbdf638446db.zip
Atomic bundle execution
Diffstat (limited to 'src/server/PreProcessor.cpp')
-rw-r--r--src/server/PreProcessor.cpp54
1 files changed, 49 insertions, 5 deletions
diff --git a/src/server/PreProcessor.cpp b/src/server/PreProcessor.cpp
index e3e5475f..2ea4ac30 100644
--- a/src/server/PreProcessor.cpp
+++ b/src/server/PreProcessor.cpp
@@ -38,6 +38,7 @@ PreProcessor::PreProcessor(Engine& engine)
, _head(NULL)
, _prepared_back(NULL)
, _tail(NULL)
+ , _block_state(BlockState::UNBLOCKED)
, _exit_flag(false)
, _thread(&PreProcessor::run, this)
{}
@@ -87,16 +88,44 @@ PreProcessor::process(RunContext& context, PostProcessor& dest, size_t limit)
size_t n_processed = 0;
Event* ev = head;
Event* last = ev;
- while (ev && ev->is_prepared() && ev->time() < context.end()) {
- if (ev->time() < context.start()) {
- // Didn't get around to executing in time, oh well...
- ev->set_time(context.start());
+ while (ev && ev->is_prepared()) {
+ switch (_block_state) {
+ case BlockState::UNBLOCKED:
+ break;
+ case BlockState::PRE_BLOCKED:
+ if (ev->get_execution() == Event::Execution::BLOCK) {
+ _block_state = BlockState::BLOCKED;
+ }
+ break;
+ case BlockState::BLOCKED:
+ break;
+ case BlockState::PRE_UNBLOCKED:
+ assert(ev->get_execution() == Event::Execution::BLOCK);
+ _block_state = BlockState::PROCESSING;
+ break;
+ case BlockState::PROCESSING:
+ if (ev->get_execution() == Event::Execution::UNBLOCK) {
+ _block_state = BlockState::UNBLOCKED;
+ }
+ }
+
+ if (_block_state == BlockState::BLOCKED) {
+ break; // Waiting for PRE_UNBLOCKED
+ } else if (ev->time() < context.start()) {
+ ev->set_time(context.start()); // Too late, nudge to context start
+ } else if (_block_state != BlockState::PROCESSING &&
+ ev->time() >= context.end()) {
+ break; // Event is for a future cycle
}
+
+ // Execute event
ev->execute(context);
last = ev;
ev = ev->next();
++n_processed;
- if (limit && n_processed >= limit) {
+
+ if (_block_state != BlockState::PROCESSING &&
+ limit && n_processed >= limit) {
break;
}
}
@@ -158,6 +187,21 @@ PreProcessor::run()
}
assert(ev->is_prepared());
+ switch (ev->get_execution()) {
+ case Event::Execution::NORMAL:
+ break;
+ case Event::Execution::BLOCK:
+ assert(_block_state == BlockState::UNBLOCKED);
+ _block_state = BlockState::PRE_BLOCKED;
+ break;
+ case Event::Execution::UNBLOCK:
+ while (_block_state != BlockState::BLOCKED) {
+ // Wait for process thread to reach blocking event
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ }
+ _block_state = BlockState::PRE_UNBLOCKED;
+ }
+
_prepared_back = (Event*)ev->next();
}
}