: m_allWorkersComplete(true)

, m_terminating(false)

{

//Create our thread pool

for(size_t i=0; i<numThreads; i++)

m_threads.push_back(make_unique<thread>(&FilterGraphExecutor::ExecutorThread, this, i));

{

//Nothing to do if we have no nodes to run

if(nodes.empty())

return;

{

lock_guard<mutex> lock(m_perfStatsMutex);

m_currentExecutionTime.clear();

}

{

lock_guard<mutex> lock(m_mutex);

if(!m_allWorkersComplete)

LogWarning("Entering RunBlocking() but not all workers are complete from previous run\n");

m_incompleteNodes = nodes;

m_incompleteNodes.erase(nullptr); //don't crash if a null filter somehow ended up in the list

m_runnableNodes.clear();

m_allWorkersComplete = false;

Filter::ClearAnalysisCache();

}

//Wake up our workers

m_workerCvar.notify_all();

//Block until they're finished

while(true)

{

unique_lock<mutex> lock(m_completionCvarMutex);

m_completionCvar.wait(lock, [this]{return m_allWorkersComplete;});

lock_guard<mutex> lock2(m_mutex);

if(m_runnableNodes.empty())

break;

}

//Update global performance stats

{

lock_guard<mutex> lock(m_perfStatsMutex);

//For now, fixed half life exponential moving average

float halflife = 8;

float decay = 1 / pow(2, 1/halflife);

//TODO: staleness or removing of some sort for old entries?

//Add the new data

for(auto& it : m_currentExecutionTime)

m_lastExecutionTime[it.first] = (m_lastExecutionTime[it.first] * decay) + (it.second * (1-decay));

}

{

//Do nothing if we already have other filters marked runnable

if(!m_runnableNodes.empty())

return;

//Look for new filters that are eligible to run

for(auto f : m_incompleteNodes)

{

//If the filter is already running, nothing we can do

if(m_runningNodes.find(f) != m_runningNodes.end())

continue;

//Not actively running.

//Is it blocked by any of our incomplete filters?

bool ok = true;

for(size_t i=0; i<f->GetInputCount(); i++)

{

auto in = f->GetInput(i).m_channel;

if(m_incompleteNodes.find(in) != m_incompleteNodes.end())

{

ok = false;

break;

}

}

//Not blocked. It's runnable.

if(ok)

m_runnableNodes.emplace(f);

}

{

#ifdef __linux__

pthread_setname_np(pthread_self(), "FilterGraph");

#endif

//Make locale handling thread safe on Windows

#ifdef _WIN32

_configthreadlocale(_ENABLE_PER_THREAD_LOCALE);

Unit::SetDefaultLocale();

#endif

pThis->DoExecutorThread(i);

{

//Create a queue and command buffer for this thread's accelerated processing

std::shared_ptr<QueueHandle> queue(g_vkQueueManager->GetComputeQueue("FilterGraphExecutor[" + to_string(i) + "].queue"));

vk::CommandPoolCreateInfo poolInfo(

vk::CommandPoolCreateFlagBits::eTransient | vk::CommandPoolCreateFlagBits::eResetCommandBuffer,

queue->m_family );

vk::raii::CommandPool pool(*g_vkComputeDevice, poolInfo);

vk::CommandBufferAllocateInfo bufinfo(*pool, vk::CommandBufferLevel::ePrimary, 1);

vk::raii::CommandBuffer cmdbuf(std::move(vk::raii::CommandBuffers(*g_vkComputeDevice, bufinfo).front()));

if(g_hasDebugUtils)

{

string prefix = string("FilterGraphExecutor[") + to_string(i) + "]";

string poolname = prefix + ".pool";

string bufname = prefix + ".cmdbuf";

g_vkComputeDevice->setDebugUtilsObjectNameEXT(

vk::DebugUtilsObjectNameInfoEXT(

vk::ObjectType::eCommandPool,

reinterpret_cast<uint64_t>(static_cast<VkCommandPool>(*pool)),

poolname.c_str()));

g_vkComputeDevice->setDebugUtilsObjectNameEXT(

vk::DebugUtilsObjectNameInfoEXT(

vk::ObjectType::eCommandBuffer,

reinterpret_cast<uint64_t>(static_cast<VkCommandBuffer>(*cmdbuf)),

bufname.c_str()));

}

//Main loop

while(true)

{

{

//Wait until the main thread starts a new round of execution, or the timeout elapses

//When we time out, check if we're shutting down

unique_lock<mutex> lock(m_workerCvarMutex);

m_workerCvar.wait_for(lock, chrono::milliseconds(50));

}

//If they woke us up because the context is being destroyed, we're done

if(m_terminating)

break;

//If we're already done, nothing to do

if(m_allWorkersComplete)

continue;

//Evaluate nodes as they become available, then stop when there's nothing left to do

FlowGraphNode* f;

while( (f = GetNextRunnableNode()) != nullptr)

{

shared_lock<shared_mutex> lock(g_vulkanActivityMutex);

//Make sure the filter's inputs are where we need them

auto loc = f->GetInputLocation();

if(loc != Filter::LOC_DONTCARE)

{

bool expectGpuInput = (loc == Filter::LOC_GPU);

bool expectCpuInput = (loc == Filter::LOC_CPU);

for(size_t j=0; j<f->GetInputCount(); j++)

{

auto data = f->GetInput(j).GetData();

if(data)

{

if(expectGpuInput)

data->PrepareForGpuAccess();

else if(expectCpuInput)

data->PrepareForCpuAccess();

}

}

}

//Actually execute the filter

double start = GetTime();

f->Refresh(cmdbuf, queue);

double dt = GetTime() - start;

{

lock_guard<mutex> slock(m_perfStatsMutex);

m_currentExecutionTime[f] = dt * FS_PER_SECOND;

}

//Filter execution has completed, remove it from the running list and mark as completed

lock_guard<mutex> lock2(m_mutex);

m_runningNodes.erase(f);

m_incompleteNodes.erase(f);

//Wake up all threads that might have been waiting on this filter to complete

m_workerCvar.notify_all();

}

//We have no more filters to run.

//If this was the last filter (nothing left incomplete), we're done - wake up the main thread

bool empty = false;

{

lock_guard<mutex> lock2(m_mutex);

empty = m_incompleteNodes.empty();

}

if(empty)

{

{

lock_guard<mutex> lock3(m_completionCvarMutex);

m_allWorkersComplete = true;

}

m_completionCvar.notify_all();

}

}