Hawkin
/
arangodb
mirror of https://gitee.com/bigwinds/arangodb
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Some more async stuff

This commit is contained in:
Simon Grätzer 2017-03-02 15:03:35 +01:00
parent fefed8b1d6
commit b2ceda6250
1 changed files with 11 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
arangod/Pregel/Worker.cpp
View File

@ -255,7 +255,6 @@ void Worker<V, E, M>::receivedMessages(VPackSlice const& data) {
// Trigger the processing of vertices // Trigger the processing of vertices
if (_config.asynchronousMode() && _state == WorkerState::IDLE) { if (_config.asynchronousMode() && _state == WorkerState::IDLE) {
MUTEX_LOCKER(guard, _commandMutex);
_continueAsync(); _continueAsync();
} }
} else if (_config.asynchronousMode() && } else if (_config.asynchronousMode() &&
@ -548,7 +547,6 @@ void Worker<V, E, M>::_finishedProcessing() {
} }
} }
if (proceed) { if (proceed) {
MUTEX_LOCKER(guard, _commandMutex);
_continueAsync(); _continueAsync();
} }
} else { // no answer expected } else { // no answer expected
@ -561,12 +559,16 @@ void Worker<V, E, M>::_finishedProcessing() {
/// in async mode checks if there are messages to process /// in async mode checks if there are messages to process
template <typename V, typename E, typename M> template <typename V, typename E, typename M>
void Worker<V, E, M>::_continueAsync() { void Worker<V, E, M>::_continueAsync() {
uint64_t cnt = _writeCache->containedMessageCount(); MUTEX_LOCKER(guard, _commandMutex);
if (_state == WorkerState::IDLE && cnt > 0) { if (_state != WorkerState::IDLE || _writeCache->containedMessageCount() == 0) {
return;
}
// avoid calling this method accidentially // avoid calling this method accidentially
_state = WorkerState::COMPUTING; _state = WorkerState::COMPUTING;
if (cnt < _messageBatchSize) { ThreadPool *pool = PregelFeature::instance()->threadPool();
pool->enqueue([this]{
if (_writeCache->containedMessageCount() < _messageBatchSize) {
usleep(50000); usleep(50000);
} }
{ // swap these pointers atomically { // swap these pointers atomically
@ -576,12 +578,13 @@ void Worker<V, E, M>::_continueAsync() {
_requestedNextGSS = true; _requestedNextGSS = true;
} }
} }
MUTEX_LOCKER(guard, _commandMutex);
// overwrite conductor values with local values // overwrite conductor values with local values
_conductorAggregators->resetValues(); _conductorAggregators->resetValues();
_conductorAggregators->aggregateValues(*_workerAggregators.get()); _conductorAggregators->aggregateValues(*_workerAggregators.get());
_workerAggregators->resetValues(); _workerAggregators->resetValues();
_startProcessing(); _startProcessing();
} });
} }
template <typename V, typename E, typename M> template <typename V, typename E, typename M>