// Copyright Sebastian Jeckel 2014.

// Distributed under the Boost Software License, Version 1.0.

// (See accompanying file LICENSE_1_0.txt or copy at

// http://www.boost.org/LICENSE_1_0.txt)

#include "react/engine/TopoSortEngine.h"

#include "tbb/parallel_for.h"

/***************************************/ REACT_IMPL_BEGIN /**************************************/

namespace toposort {

///////////////////////////////////////////////////////////////////////////////////////////////////

/// Turn

///////////////////////////////////////////////////////////////////////////////////////////////////

ExclusiveTurn::ExclusiveTurn(TurnIdT id, TurnFlagsT flags) :

TurnBase(id, flags)

{

}

///////////////////////////////////////////////////////////////////////////////////////////////////

/// EngineBase

///////////////////////////////////////////////////////////////////////////////////////////////////

template <typename TNode, typename TTurn>

void EngineBase<TNode,TTurn>::OnNodeAttach(TNode& node, TNode& parent)

{

parent.Successors.Add(node);

if (node.Level <= parent.Level)

node.Level = parent.Level + 1;

}

template <typename TNode, typename TTurn>

void EngineBase<TNode,TTurn>::OnNodeDetach(TNode& node, TNode& parent)

{

parent.Successors.Remove(node);

}

template <typename TNode, typename TTurn>

void EngineBase<TNode,TTurn>::OnTurnInputChange(TNode& node, TTurn& turn)

{

processChildren(node, turn);

}

template <typename TNode, typename TTurn>

void EngineBase<TNode,TTurn>::OnNodePulse(TNode& node, TTurn& turn)

{

processChildren(node, turn);

}

// Explicit instantiation

template class EngineBase<SeqNode,ExclusiveTurn>;

template class EngineBase<ParNode,ExclusiveTurn>;

template class EngineBase<SeqNode,DefaultQueueableTurn<ExclusiveTurn>>;

template class EngineBase<ParNode,DefaultQueueableTurn<ExclusiveTurn>>;

///////////////////////////////////////////////////////////////////////////////////////////////////

/// SeqEngineBase

///////////////////////////////////////////////////////////////////////////////////////////////////

template <typename TTurn>

void SeqEngineBase<TTurn>::OnTurnPropagate(TTurn& turn)

{

while (scheduledNodes_.FetchNext())

{

for (auto* curNode : scheduledNodes_.NextNodes())

{

if (curNode->Level < curNode->NewLevel)

{

curNode->Level = curNode->NewLevel;

invalidateSuccessors(*curNode);

scheduledNodes_.Push(curNode);

continue;

}

curNode->Queued = false;

curNode->Tick(&turn);

}

}

}

template <typename TTurn>

void SeqEngineBase<TTurn>::OnDynamicNodeAttach(SeqNode& node, SeqNode& parent, TTurn& turn)

{

OnNodeAttach(node, parent);

invalidateSuccessors(node);

// Re-schedule this node

node.Queued = true;

scheduledNodes_.Push(&node);

}

template <typename TTurn>

void SeqEngineBase<TTurn>::OnDynamicNodeDetach(SeqNode& node, SeqNode& parent, TTurn& turn)

{

OnNodeDetach(node, parent);

}

template <typename TTurn>

void SeqEngineBase<TTurn>::processChildren(SeqNode& node, TTurn& turn)

{

// Add children to queue

for (auto* succ : node.Successors)

{

if (!succ->Queued)

{

succ->Queued = true;

scheduledNodes_.Push(succ);

}

}

}

template <typename TTurn>

void SeqEngineBase<TTurn>::invalidateSuccessors(SeqNode& node)

{

for (auto* succ : node.Successors)

{

if (succ->NewLevel <= node.Level)

succ->NewLevel = node.Level + 1;

}

}

// Explicit instantiation

template class SeqEngineBase<ExclusiveTurn>;

template class SeqEngineBase<DefaultQueueableTurn<ExclusiveTurn>>;

///////////////////////////////////////////////////////////////////////////////////////////////////

/// ParEngineBase

///////////////////////////////////////////////////////////////////////////////////////////////////

template <typename TTurn>

void ParEngineBase<TTurn>::OnTurnPropagate(TTurn& turn)

{

while (topoQueue_.FetchNext())

{

//using RangeT = tbb::blocked_range<vector<ParNode*>::const_iterator>;

using RangeT = ParEngineBase::TopoQueueT::RangeT;

// Iterate all nodes of current level and start processing them in parallel

tbb::parallel_for(

topoQueue_.NextRange(),

[&] (const RangeT& range)

{

for (auto* curNode : range)

{

if (curNode->Level < curNode->NewLevel)

{

curNode->Level = curNode->NewLevel;

invalidateSuccessors(*curNode);

topoQueue_.Push(curNode);

continue;

}

curNode->Collected = false;

// Tick -> if changed: OnNodePulse -> adds child nodes to the queue

curNode->Tick(&turn);

}

}

);

if (dynRequests_.size() > 0)

{

for (auto req : dynRequests_)

{

if (req.ShouldAttach)

applyDynamicAttach(*req.Node, *req.Parent, turn);

else

applyDynamicDetach(*req.Node, *req.Parent, turn);

}

dynRequests_.clear();

}

}

}

template <typename TTurn>

void ParEngineBase<TTurn>::OnDynamicNodeAttach(ParNode& node, ParNode& parent, TTurn& turn)

{

DynRequestData data{ true, &node, &parent };

dynRequests_.push_back(data);

}

template <typename TTurn>

void ParEngineBase<TTurn>::OnDynamicNodeDetach(ParNode& node, ParNode& parent, TTurn& turn)

{

DynRequestData data{ false, &node, &parent };

dynRequests_.push_back(data);

}

template <typename TTurn>

void ParEngineBase<TTurn>::HintUpdateDuration(ParNode& node, uint dur)

{

if (dur < min_weight)

dur = min_weight;

node.Weight = dur;

}

template <typename TTurn>

void ParEngineBase<TTurn>::applyDynamicAttach(ParNode& node, ParNode& parent, TTurn& turn)

{

OnNodeAttach(node, parent);

invalidateSuccessors(node);

// Re-schedule this node

node.Collected = true;

topoQueue_.Push(&node);

}

template <typename TTurn>

void ParEngineBase<TTurn>::applyDynamicDetach(ParNode& node, ParNode& parent, TTurn& turn)

{

OnNodeDetach(node, parent);

}

template <typename TTurn>

void ParEngineBase<TTurn>::processChildren(ParNode& node, TTurn& turn)

{

// Add children to queue

for (auto* succ : node.Successors)

if (!succ->Collected.exchange(true, std::memory_order_relaxed))

topoQueue_.Push(succ);

}

template <typename TTurn>

void ParEngineBase<TTurn>::invalidateSuccessors(ParNode& node)

{

for (auto* succ : node.Successors)

{// succ->InvalidateMutex

ParNode::InvalidateMutexT::scoped_lock lock(succ->InvalidateMutex);

if (succ->NewLevel <= node.Level)

succ->NewLevel = node.Level + 1;

}// ~succ->InvalidateMutex

}

// Explicit instantiation

template class ParEngineBase<ExclusiveTurn>;

template class ParEngineBase<DefaultQueueableTurn<ExclusiveTurn>>;

///////////////////////////////////////////////////////////////////////////////////////////////////

/// PipeliningTurn

///////////////////////////////////////////////////////////////////////////////////////////////////

PipeliningTurn::PipeliningTurn(TurnIdT id, TurnFlagsT flags) :

TurnBase(id, flags),

isMergeable_{ (flags & enable_input_merging) != 0 }

{

}

bool PipeliningTurn::AdvanceLevel()

{

std::unique_lock<mutex> lock(advMutex_);

advCondition_.wait(lock, [this] {

return !(currentLevel_ + 1 > maxLevel_);

});

// Remove the intervals we're going to leave

auto it = levelIntervals_.begin();

while (it != levelIntervals_.end())

{

if (it->second <= currentLevel_)

it = levelIntervals_.erase(it);

else

++it;

}

// Add new interval for current level

if (currentLevel_ < curUpperBound_)

levelIntervals_.emplace(currentLevel_, curUpperBound_);

currentLevel_++;

curUpperBound_ = currentLevel_;

// Min level is the minimum over all interval lower bounds

int newMinLevel =

levelIntervals_.begin() != levelIntervals_.end() ?

levelIntervals_.begin()->first :

-1;

if (minLevel_ != newMinLevel)

{

minLevel_ = newMinLevel;

return true;

}

else

{

return false;

}

}

void PipeliningTurn::SetMaxLevel(int level)

{

std::lock_guard<mutex> lock(advMutex_);

maxLevel_ = level;

advCondition_.notify_all();

}

void PipeliningTurn::WaitForMaxLevel(int targetLevel)

{

std::unique_lock<mutex> lock(advMutex_);

advCondition_.wait(lock, [this, targetLevel] {

return !(targetLevel < maxLevel_);

});

}

void PipeliningTurn::Append(PipeliningTurn* turn)

{

successor_ = turn;

if (turn)

turn->predecessor_ = this;

UpdateSuccessor();

}

void PipeliningTurn::UpdateSuccessor()

{

if (successor_)

successor_->SetMaxLevel(minLevel_ - 1);

}

void PipeliningTurn::Remove()

{

if (predecessor_)

{

predecessor_->Append(successor_);

}

else if (successor_)

{

successor_->SetMaxLevel(INT_MAX);

successor_->predecessor_ = nullptr;

}

for (const auto& e : merged_)

e.second->Unblock();

}

void PipeliningTurn::AdjustUpperBound(int level)

{

if (curUpperBound_ < level)

curUpperBound_ = level;

}

void PipeliningTurn::RunMergedInputs() const

{

for (const auto& e : merged_)

e.first();

}

///////////////////////////////////////////////////////////////////////////////////////////////////

/// PipeliningEngine

///////////////////////////////////////////////////////////////////////////////////////////////////

void PipeliningEngine::OnNodeAttach(ParNode& node, ParNode& parent)

{

parent.Successors.Add(node);

if (node.Level <= parent.Level)

node.Level = parent.Level + 1;

if (node.IsDynamicNode())

{

dynamicNodes_.insert(&node);

if (maxDynamicLevel_ < node.Level)

maxDynamicLevel_ = node.Level;

}

}

void PipeliningEngine::OnNodeDetach(ParNode& node, ParNode& parent)

{

parent.Successors.Remove(node);

// Recalc maxdynamiclevel?

if (node.IsDynamicNode())

{

dynamicNodes_.erase(&node);

if (maxDynamicLevel_ == node.Level)

{

maxDynamicLevel_ = 0;

for (auto e : dynamicNodes_)

if (maxDynamicLevel_ < e->Level)

maxDynamicLevel_ = e->Level;

}

}

}

void PipeliningEngine::OnTurnAdmissionStart(PipeliningTurn& turn)

{

{// seqMutex_

SeqMutexT::scoped_lock lock(seqMutex_);

if (tail_)

tail_->Append(&turn);

tail_ = &turn;

}// ~seqMutex_

advanceTurn(turn);

}

void PipeliningEngine::OnTurnAdmissionEnd(PipeliningTurn& turn)

{

turn.RunMergedInputs();

}

void PipeliningEngine::OnTurnEnd(PipeliningTurn& turn)

{// seqMutex_ (write)

SeqMutexT::scoped_lock lock(seqMutex_, true);

turn.Remove();

if (tail_ == &turn)

tail_ = nullptr;

}// ~seqMutex_

void PipeliningEngine::OnTurnInputChange(ParNode& node, PipeliningTurn& turn)

{

processChildren(node, turn);

}

void PipeliningEngine::OnNodePulse(ParNode& node, PipeliningTurn& turn)

{

processChildren(node, turn);

}

void PipeliningEngine::OnTurnPropagate(PipeliningTurn& turn)

{

if (maxDynamicLevel_ > 0)

turn.AdjustUpperBound(maxDynamicLevel_);

while (turn.TopoQueue.FetchNext())

{

using RangeT = PipeliningTurn::TopoQueueT::RangeT;

for (const auto* node : turn.TopoQueue.NextRange())

turn.AdjustUpperBound(node->Level);

advanceTurn(turn);

// Iterate all nodes of current level and start processing them in parallel

tbb::parallel_for(

turn.TopoQueue.NextRange(),

[&] (const RangeT& range)

{

for (auto* curNode : range)

{

if (curNode->Level < curNode->NewLevel)

{

curNode->Level = curNode->NewLevel;

invalidateSuccessors(*curNode);

turn.TopoQueue.Push(curNode);

continue;

}

curNode->Collected = false;

// Tick -> if changed: OnNodePulse -> adds child nodes to the queue

curNode->Tick(&turn);

}

}

);

if (turn.DynRequests.size() > 0)

{

for (auto req : turn.DynRequests)

{

if (req.ShouldAttach)

applyDynamicAttach(*req.Node, *req.Parent, turn);

else

applyDynamicDetach(*req.Node, *req.Parent, turn);

}

turn.DynRequests.clear();

}

}

}

void PipeliningEngine::OnDynamicNodeAttach(ParNode& node, ParNode& parent, PipeliningTurn& turn)

{

DynRequestData data{ true, &node, &parent };

turn.DynRequests.push_back(data);

}

void PipeliningEngine::OnDynamicNodeDetach(ParNode& node, ParNode& parent, PipeliningTurn& turn)

{

DynRequestData data{ false, &node, &parent };

turn.DynRequests.push_back(data);

}

void PipeliningEngine::applyDynamicAttach(ParNode& node, ParNode& parent, PipeliningTurn& turn)

{

turn.WaitForMaxLevel(INT_MAX);

OnNodeAttach(node, parent);

invalidateSuccessors(node);

// Re-schedule this node

node.Collected = true;

turn.TopoQueue.Push(&node);

}

void PipeliningEngine::applyDynamicDetach(ParNode& node, ParNode& parent, PipeliningTurn& turn)

{

OnNodeDetach(node, parent);

}

void PipeliningEngine::processChildren(ParNode& node, PipeliningTurn& turn)

{

// Add children to queue

for (auto* succ : node.Successors)

if (!succ->Collected.exchange(true, std::memory_order_relaxed))

turn.TopoQueue.Push(succ);

}

void PipeliningEngine::invalidateSuccessors(ParNode& node)

{

for (auto* succ : node.Successors)

{// succ->InvalidateMutex

ParNode::InvalidateMutexT::scoped_lock lock(succ->InvalidateMutex);

if (succ->NewLevel <= node.Level)

{

auto newLevel = node.Level + 1;

succ->NewLevel = newLevel;

{// maxDynamicLevelMutex_

MaxDynamicLevelMutexT::scoped_lock lock(maxDynamicLevelMutex_);

if (succ->IsDynamicNode() && maxDynamicLevel_ < newLevel)

maxDynamicLevel_ = newLevel;

}// ~maxDynamicLevelMutex_

}

}// ~succ->InvalidateMutex

}

void PipeliningEngine::advanceTurn(PipeliningTurn& turn)

{

// No need to wake up successor if min level did not change

if (turn.AdvanceLevel())

return;

{// seqMutex_ (read)

SeqMutexT::scoped_lock lock(seqMutex_, false);

turn.UpdateSuccessor();

}// ~seqMutex_

}

} // ~namespace toposort

/****************************************/ REACT_IMPL_END /***************************************/