/*=========================================================================

  Program:   Visualization Toolkit
  Module:    vtkExecutionScheduler.cxx

  Copyright (c) Ken Martin, Will Schroeder, Bill Lorensen
  All rights reserved.
  See Copyright.txt or http://www.kitware.com/Copyright.htm for details.

     This software is distributed WITHOUT ANY WARRANTY; without even
     the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
     PURPOSE.  See the above copyright notice for more information.

=========================================================================*/
/*-------------------------------------------------------------------------
  Copyright (c) 2008, 2009 by SCI Institute, University of Utah.
  
  This is part of the Parallel Dataflow System originally developed by
  Huy T. Vo and Claudio T. Silva. For more information, see:

  "Parallel Dataflow Scheme for Streaming (Un)Structured Data" by Huy
  T. Vo, Daniel K. Osmari, Brian Summa, Joao L.D. Comba, Valerio
  Pascucci and Claudio T. Silva, SCI Institute, University of Utah,
  Technical Report #UUSCI-2009-004, 2009.

  "Multi-Threaded Streaming Pipeline For VTK" by Huy T. Vo and Claudio
  T. Silva, SCI Institute, University of Utah, Technical Report
  #UUSCI-2009-005, 2009.
-------------------------------------------------------------------------*/
#include "vtkExecutionScheduler.h"

#include "vtkAlgorithm.h"
#include "vtkCommand.h"
#include "vtkComputingResources.h"
#include "vtkExecutiveCollection.h"
#include "vtkInformation.h"
#include "vtkInformationVector.h"
#include "vtkInformationExecutivePortKey.h"
#include "vtkInformationExecutivePortVectorKey.h"
#include "vtkInformationIntegerKey.h"
#include "vtkMultiThreader.h"
#include "vtkMutexLock.h"
#include "vtkObjectFactory.h"
#include "vtkThreadedStreamingPipeline.h"
#include "vtkThreadMessager.h"

#include <vtkstd/set>
#include <vtksys/hash_map.hxx>
#include <vtkstd/vector>
#include <vtksys/hash_set.hxx>

//----------------------------------------------------------------------------
vtkCxxRevisionMacro(vtkExecutionScheduler, "1.19");
vtkStandardNewMacro(vtkExecutionScheduler);

vtkInformationKeyMacro(vtkExecutionScheduler, TASK_PRIORITY, Integer);

//----------------------------------------------------------------------------
class Task 
{
public:
  Task(int _priority = -1, vtkExecutive *_exec = NULL, vtkInformation *_info = NULL) 
  {
    this->priority = _priority;
    this->exec = _exec;
    this->info = _info;
  }
  
  int             priority;
  vtkExecutive   *exec;
  vtkInformation *info;
};
  
//----------------------------------------------------------------------------
class TaskWeakOrdering 
{
public:
  bool operator()(const Task& t1,
                  const Task& t2) const 
  {
    return t1.priority < t2.priority;
  }
};

//----------------------------------------------------------------------------
// Convinient definitions of vector/set of vtkExecutive
class vtkExecutiveHasher 
{
public:
  size_t operator()(const vtkExecutive* e) const 
  {
    return (size_t)e;
  }
};
typedef vtksys::hash_set<vtkExecutive*, vtkExecutiveHasher> vtkExecutiveSet;
typedef vtkstd::vector<vtkExecutive*>                       vtkExecutiveVector;

//----------------------------------------------------------------------------
class vtkExecutionScheduler::implementation
{
public:
  // The containing object
  vtkExecutionScheduler*  Scheduler;

  // Some convenient type definitions for STL containers
  typedef vtksys::hash_map<vtkExecutive*, int, 
    vtkExecutiveHasher>                                 ExecutiveIntHashMap;
  typedef vtkstd::pair<int, int>                        Edge;
  class                                                 EdgeHasher;
  typedef vtksys::hash_set<Edge, EdgeHasher>            EdgeSet;
  typedef vtkstd::multiset<Task, TaskWeakOrdering>      TaskPriorityQueue;
  typedef vtkstd::vector<vtkMutexLock*>                 MutexLockVector;
  typedef vtkstd::vector<vtkThreadMessager*>            MessagerVector;
  class EdgeHasher 
  {
  public:
    size_t operator()(const Edge &e) const 
    {
      return (size_t)((e.first << 16) +  e.second);
    }
  };

  vtkExecutiveSet           ExecutingTasks;
  TaskPriorityQueue         PrioritizedTasks;
  ExecutiveIntHashMap       DependencyNodes;
  EdgeSet                   DependencyEdges;
  MessagerVector            TaskDoneMessagers;
  MutexLockVector           InputsReleasedLocks;
  MessagerVector            InputsReleasedMessagers;
  int                       CurrentPriority;
  
  // Description:
  // Start from the exec and go all the way up to the sources (modules
  // without any inputs), then call TraverseDownToSink to update edges
  void FindAndTraverseFromSources(vtkExecutive *exec, vtkExecutiveSet &visited);
  
  // Description:
  // Actual traverse down the network, for each nodes, construct and
  // add edges connecting all of its upstream modules to itself to the
  // dependency graph
  void TraverseDownToSink(vtkExecutive *exec, vtkExecutiveSet &upstream,
                          vtkExecutiveSet &visited);
  
  // Description:
  // Actual traverse down the network, for each nodes, construct and
  // add edges connecting all of its upstream modules to itself to the
  // dependency graph
  void CollectDownToSink(vtkExecutive *exec, vtkExecutiveSet &visited,
                         vtkExecutiveVector &graph);
  
  // Description:
  // A task can be executed if none of its predecessor tasks are still
  // on the queue. This only makes sense for tasks that are currently
  // on the queue, thus, an iterator is provided instead of the task
  // itself.
  bool CanExecuteTask(TaskPriorityQueue::const_iterator ti);
  // Description:
  // Spawn a thread to execute a module
  void Execute(const Task &task);

  // Description:
  // Check if the given exec is a new module or not. If it is then
  // traverse the network to update dependency edges for its connected
  // subgraph
  void UpdateDependencyGraph(vtkExecutive *exec);

  // Description:
  // Add the module exec to the set of dependency nodes if it is not
  // already there and return its node id number
  int AddToDependencyGraph(vtkExecutive *exec);

  // Description:
  // Add the given executive to the execution queue for later
  // execution
  void AddToQueue(vtkExecutive *exec, vtkInformation *info);
  
  // Description:
  // Obtain the priority from the information object if it is given,
  // otherwise, use a priority assigned from the scheduler
  int AcquirePriority(vtkInformation *info);
};

//----------------------------------------------------------------------------
static vtkExecutionScheduler *globalScheduler = NULL;

//----------------------------------------------------------------------------
void * vtkExecutionScheduler_ScheduleThread(void *data);
void * vtkExecutionScheduler_ExecuteThread(void *data);

//----------------------------------------------------------------------------
vtkExecutionScheduler* vtkExecutionScheduler::GetGlobalScheduler()
{
  if (!globalScheduler)
    {
    globalScheduler = vtkExecutionScheduler::New();
    }
  return globalScheduler;
}

//----------------------------------------------------------------------------
vtkExecutionScheduler::vtkExecutionScheduler()
  : Implementation(new implementation)
{
  this->Resources = vtkComputingResources::New();
  this->Resources->ObtainMaximumResources();
  this->ResourceMessager = vtkThreadMessager::New();
  this->ScheduleLock = vtkMutexLock::New();
  this->ScheduleMessager = vtkThreadMessager::New();
  this->ScheduleThreader = vtkMultiThreader::New();
  this->ScheduleThreader->SetNumberOfThreads(1);
  this->ScheduleThreadId = -1;
  this->Implementation->Scheduler = this;
  this->Implementation->CurrentPriority = 0;
}

//----------------------------------------------------------------------------
vtkExecutionScheduler::~vtkExecutionScheduler()
{
  this->Resources->Delete();
  this->ResourceMessager->Delete();
  this->ScheduleLock->Delete();
  this->ScheduleMessager->Delete();
  this->ScheduleThreader->Delete();
  delete this->Implementation;
}

//----------------------------------------------------------------------------
void vtkExecutionScheduler::PrintSelf(ostream &os, vtkIndent indent)
{
  this->Superclass::PrintSelf(os, indent);
}

//----------------------------------------------------------------------------
void vtkExecutionScheduler::SchedulePropagate(vtkExecutiveCollection *execs, vtkInformation* vtkNotUsed(info))
{
  execs->InitTraversal();
  vtkExecutiveSet    visited;
  vtkExecutiveVector graph;
  for (vtkExecutive *e = execs->GetNextItem(); e != 0; e = execs->GetNextItem())
    {
    this->Implementation->CollectDownToSink(e, visited, graph);
    }
  
  for (vtkExecutiveVector::iterator vi=graph.begin();
       vi!=graph.end(); vi++)
    {
    (*vi)->Update();
    vtkAlgorithm *rep =(*vi)->GetAlgorithm();
    if (rep->IsA("vtkDataRepresentation"))
      {
      rep->InvokeEvent(vtkCommand::UpdateEvent, NULL);
      }
    }
}

//----------------------------------------------------------------------------
void vtkExecutionScheduler::Schedule(vtkExecutiveCollection *execs, vtkInformation *info)
{
  if (this->ScheduleThreadId == -1)
    {
    this->ScheduleThreadId = this->ScheduleThreader->
      SpawnThread((vtkThreadFunctionType)(vtkExecutionScheduler_ScheduleThread), this);
    }
  this->ScheduleLock->Lock();
  vtkExecutiveVector G;
  execs->InitTraversal();
  for (vtkExecutive *e = execs->GetNextItem(); e != 0; e = execs->GetNextItem()) 
    {
    if (this->Implementation->ExecutingTasks.find(e) != this->Implementation->ExecutingTasks.end())
      {
      return;
      }
    if (this->Implementation->DependencyNodes.find(e) == this->Implementation->DependencyNodes.end()) 
      {
      this->Implementation->UpdateDependencyGraph(e);
      }
    G.push_back(e);
    }

  // Create an adjacency matrix
  unsigned i, j, k, p;
  unsigned N = (unsigned)G.size();
  int *A = (int*)malloc(N*N*sizeof(int));
  int *degree = (int*)malloc(N*sizeof(int));
  memset(A, 0, N*N*sizeof(int));
  memset(degree, 0, N*sizeof(int));
  for (i = 0; i < N; i++) 
    {
    int src = (*(this->Implementation->DependencyNodes.find(G[i]))).second;
    for (j = 0; j < N; j++) 
      {
      int dst = (*(this->Implementation->DependencyNodes.find(G[j]))).second;
      if (this->Implementation->DependencyEdges.find(implementation::Edge(src, dst)) !=
          this->Implementation->DependencyEdges.end()) 
        {
        A[i*N+j] = 1;
        degree[j]++;
        }
      }
    }
  
  unsigned *S = (unsigned*)malloc(N*sizeof(unsigned));
  k = 0;
  for (j = 0; j < N; j++)
    {
    if (degree[j] == 0)
      {
      S[k++] = j;
      }
    }
  p = 0;
  while (p < k) 
    {
    i = S[p++];
    this->Implementation->AddToQueue(G[i], info);
    for (j = 0; j < N; j++)
      if (A[i*N+j]) 
        {
        degree[j]--;
        A[i*N+j] = 0;
        if (degree[j] == 0)
          {
          S[k++] = j;
          }
        }
    }
  free(S);
  free(degree);
  free(A);
        
  // Wake the scheduling thread up if it is currently waiting for tasks
  this->ScheduleMessager->SendWakeMessage();
  this->ScheduleLock->Unlock();

}

//----------------------------------------------------------------------------
void vtkExecutionScheduler::implementation::UpdateDependencyGraph(vtkExecutive *exec)
{
  vtkExecutiveSet visited;
  this->FindAndTraverseFromSources(exec, visited);
}

//----------------------------------------------------------------------------
void vtkExecutionScheduler::implementation::FindAndTraverseFromSources
(vtkExecutive *exec, vtkExecutiveSet &visited)
{
  if (visited.find(exec) != visited.end())
    {
    return;
    }
  visited.insert(exec);
  bool isSource = true;
  for(int i = 0; i < exec->GetNumberOfInputPorts(); ++i) 
    {
    int nic = exec->GetAlgorithm()->GetNumberOfInputConnections(i);
    vtkInformationVector* inVector = exec->GetInputInformation()[i];
    for(int j = 0; j < nic; ++j) 
      {
      vtkInformation* inInfo = inVector->GetInformationObject(j);
      vtkExecutive* e;
      int producerPort;
      vtkExecutive::PRODUCER()->Get(inInfo, e, producerPort);
      if (e) 
        {
        isSource = false;
        this->FindAndTraverseFromSources(e, visited);
        }
      }
    }
  if (isSource) 
    {
    vtkExecutiveSet upstream;
    vtkExecutiveSet downVisited;
    this->TraverseDownToSink(exec, upstream, downVisited);
    }
}

//----------------------------------------------------------------------------
void vtkExecutionScheduler::implementation::TraverseDownToSink
(vtkExecutive *exec, vtkExecutiveSet &upstream, vtkExecutiveSet &visited)
{
  if (visited.find(exec)!=visited.end())
    {
    return;
    }
  
  // Now mark all edges from upstream modules to exec as dependency edges
  int vId = this->AddToDependencyGraph(exec);
  for (vtkExecutiveSet::const_iterator it = upstream.begin(); it != upstream.end(); it++)
    {
    implementation::ExecutiveIntHashMap::iterator hit = this->DependencyNodes.find(*it);
    this->DependencyEdges.insert(Edge((*hit).second, vId));
    }
  
  // Mark as visited
  visited.insert(exec);

  // Then insert it to the upstream list for going down
  upstream.insert(exec);
  for(int i = 0; i < exec->GetNumberOfOutputPorts(); ++i) 
    {
    vtkInformation* info = exec->GetOutputInformation(i);
    int consumerCount = vtkExecutive::CONSUMERS()->Length(info);
    vtkExecutive** e = vtkExecutive::CONSUMERS()->GetExecutives(info);
    for (int j = 0; j < consumerCount; j++)
      if (e[j]) 
        {
        this->TraverseDownToSink(e[j], upstream, visited);
        }
    }
  
  // Take it out of the upstream and prepare for back-tracking
  upstream.erase(exec);
}

//----------------------------------------------------------------------------
void vtkExecutionScheduler::implementation::CollectDownToSink
(vtkExecutive *exec, vtkExecutiveSet &visited, vtkExecutiveVector &graph)
{
  if (visited.find(exec)!=visited.end())
    {
    return;
    }
  
  // Mark as visited
  visited.insert(exec);

  // Add it to the ordered graph
  graph.push_back(exec);

  // Then traverse down
  for(int i = 0; i < exec->GetNumberOfOutputPorts(); ++i) 
    {
    vtkInformation* info = exec->GetOutputInformation(i);
    int consumerCount = vtkExecutive::CONSUMERS()->Length(info);
    vtkExecutive** e = vtkExecutive::CONSUMERS()->GetExecutives(info);
    for (int j = 0; j < consumerCount; j++)
      {
      if (e[j]) 
        {
        this->CollectDownToSink(e[j], visited, graph);
        }
      }
    }
}

//----------------------------------------------------------------------------
int vtkExecutionScheduler::implementation::AddToDependencyGraph(vtkExecutive *exec)
{
  // We never remove vertices, it's ok to just return the size of
  // DependencyGraph as a node id
  implementation::ExecutiveIntHashMap::iterator hit = this->DependencyNodes.find(exec);
  // Check if this is a new module or an untouched sub-network
  int vId = -1;
  if (hit == this->DependencyNodes.end()) 
    {
    vId = (int)this->DependencyNodes.size();
    this->DependencyNodes[exec] = vId;
    
    // Make sure that we have enough thread messagers for this vId
    while (this->TaskDoneMessagers.size()<=(size_t)vId)
      {
      this->TaskDoneMessagers.push_back(vtkThreadMessager::New());
      }
    
    while (this->InputsReleasedMessagers.size()<=(size_t)vId) 
      {
      this->InputsReleasedMessagers.push_back(vtkThreadMessager::New());
      this->InputsReleasedLocks.push_back(vtkMutexLock::New());
      }
    }
  // We have this module in our cache before
  else 
    {
    vId = (*hit).second;
    }
  return vId;
}

//----------------------------------------------------------------------------
int vtkExecutionScheduler::implementation::AcquirePriority(vtkInformation * info)
{
  int priority;
  if (info && info->Has(TASK_PRIORITY())) 
    {
    priority = info->Get(TASK_PRIORITY());
    }
  else 
    {
    priority = this->CurrentPriority++;
    }
  return priority;
}

//----------------------------------------------------------------------------
void vtkExecutionScheduler::implementation::AddToQueue(vtkExecutive *exec, vtkInformation * info)
{
  int priority = this->AcquirePriority(info);
  this->PrioritizedTasks.insert(Task(priority, exec, info));
  vtkMutexLock *lock = this->Scheduler->GetInputsReleasedLock(exec);
  if (lock) 
    {
//     fprintf(stderr, "A %s\n", exec->GetAlgorithm()->GetClassName());
    lock->Lock();
    }
}

//----------------------------------------------------------------------------
void vtkExecutionScheduler::WaitUntilDone(vtkExecutiveCollection *execs)
{
  vtkExecutiveSet::const_iterator it;
  execs->InitTraversal();
  for (vtkExecutive *e = execs->GetNextItem(); e != 0; e = execs->GetNextItem()) 
    {
    this->WaitForTaskDone(e);
    }
}

//----------------------------------------------------------------------------
void vtkExecutionScheduler::WaitUntilAllDone()
{
  while (true) 
    {
    vtkExecutive *exec = NULL;
    this->ScheduleLock->Lock();
    if (this->Implementation->PrioritizedTasks.size() > 0)
      exec = (*(this->Implementation->PrioritizedTasks.begin())).exec;
    this->ScheduleLock->Unlock();
    if (exec)
      this->WaitForTaskDone(exec);
    else
      break;
    }
}

//----------------------------------------------------------------------------
void vtkExecutionScheduler::WaitUntilReleased(vtkExecutiveCollection *execs)
{
  vtkExecutiveSet::const_iterator it;
  execs->InitTraversal();
  for (vtkExecutive *e = execs->GetNextItem(); e != 0; e = execs->GetNextItem())
    {
    this->WaitForInputsReleased(e);
    }
}

//----------------------------------------------------------------------------
void vtkExecutionScheduler::WaitForTaskDone(vtkExecutive *exec)
{
  vtkThreadMessager *messager = this->GetTaskDoneMessager(exec);
  if (messager)
    messager->WaitForMessage();
}

//----------------------------------------------------------------------------
void vtkExecutionScheduler::WaitForInputsReleased(vtkExecutive *exec)
{
  vtkMutexLock *lock = this->GetInputsReleasedLock(exec);
  if (lock) 
    {
    lock->Lock();
    lock->Unlock();
    }
//   vtkThreadMessager *messager = this->GetInputsReleasedMessager(exec);
//   if (messager)
//     messager->WaitForMessage();
}

//----------------------------------------------------------------------------
vtkThreadMessager* vtkExecutionScheduler::GetTaskDoneMessager(vtkExecutive *exec)
{
  implementation::ExecutiveIntHashMap::iterator hit = 
    this->Implementation->DependencyNodes.find(exec);
  if (hit != this->Implementation->DependencyNodes.end()) 
    {
    implementation::TaskPriorityQueue::const_iterator ti;
    for (ti = this->Implementation->PrioritizedTasks.begin();
         ti != this->Implementation->PrioritizedTasks.end(); ti++) 
      {
      if ((*ti).exec==exec)
        {
        return this->Implementation->TaskDoneMessagers[(*hit).second];
        }
      }
    }
  return NULL;
}

//----------------------------------------------------------------------------
vtkMutexLock* vtkExecutionScheduler::GetInputsReleasedLock(vtkExecutive *exec)
{
  implementation::ExecutiveIntHashMap::iterator hit = 
    this->Implementation->DependencyNodes.find(exec);
  if (hit != this->Implementation->DependencyNodes.end()) 
    {
    implementation::TaskPriorityQueue::const_iterator ti;
    for (ti = this->Implementation->PrioritizedTasks.begin();
         ti != this->Implementation->PrioritizedTasks.end(); ti++) 
      {
      if ((*ti).exec == exec)
        {
        return this->Implementation->InputsReleasedLocks[(*hit).second];
        }
      }
    }
  return NULL;
}

//----------------------------------------------------------------------------
vtkThreadMessager* vtkExecutionScheduler::GetInputsReleasedMessager(vtkExecutive *exec)
{
  implementation::ExecutiveIntHashMap::iterator hit = 
    this->Implementation->DependencyNodes.find(exec);
  if (hit != this->Implementation->DependencyNodes.end()) 
    {
    implementation::TaskPriorityQueue::const_iterator ti;
    for (ti = this->Implementation->PrioritizedTasks.begin();
         ti != this->Implementation->PrioritizedTasks.end(); ti++) 
      {
      if ((*ti).exec==exec)
        {
        return this->Implementation->InputsReleasedMessagers[(*hit).second];
        }
      }
    }
  return NULL;
}

//----------------------------------------------------------------------------
void vtkExecutionScheduler::ReleaseResources(vtkExecutive *exec)
{
  vtkThreadedStreamingPipeline *threadedExec = vtkThreadedStreamingPipeline::
    SafeDownCast(exec);
  if (threadedExec) 
    {
    this->ScheduleLock->Lock();
    this->Resources->Collect(threadedExec->GetResources());
    this->ResourceMessager->SendWakeMessage();
    this->ScheduleLock->Unlock();
    }
}

//----------------------------------------------------------------------------
void vtkExecutionScheduler::ReacquireResources(vtkExecutive *exec)
{
  vtkThreadedStreamingPipeline *threadedExec = vtkThreadedStreamingPipeline::
    SafeDownCast(exec);
  if (threadedExec) 
    {
    while (!this->Resources->Reserve(threadedExec->GetResources())) 
      {
      this->ResourceMessager->WaitForMessage();
      }
    }
}

//----------------------------------------------------------------------------
bool vtkExecutionScheduler::implementation::CanExecuteTask
(TaskPriorityQueue::const_iterator taskIter)
{
  if (this->ExecutingTasks.find((*taskIter).exec) != this->ExecutingTasks.end())
    {
    return false;
    }
  ExecutiveIntHashMap::iterator hit = this->DependencyNodes.find((*taskIter).exec);
  if (hit == this->DependencyNodes.end())
    {
    return true;
    }
  int dst = (*hit).second;
  TaskPriorityQueue::const_iterator ti;
  for (ti = this->PrioritizedTasks.begin();
       ti != taskIter; ti++)
    {
    if ((*ti).priority > (*taskIter).priority)
      {
      break;
      }
    hit = this->DependencyNodes.find((*ti).exec);
    int src = (*hit).second;
    if (this->DependencyEdges.find(Edge(src, dst)) != this->DependencyEdges.end())
      {
      return false;
      }
    }
  return true;
}

//----------------------------------------------------------------------------
void vtkExecutionScheduler::RescheduleFrom(vtkExecutive *exec,
                                           vtkComputingResources *resources) 
{
  vtkExecutiveVector upstream;
  // Compute the total time
  float totalUpStreamTime = 0.0;
  for(int i = 0; i < exec->GetNumberOfInputPorts(); ++i) 
    {
    int nic = exec->GetAlgorithm()->GetNumberOfInputConnections(i);
    vtkInformationVector* inVector = exec->GetInputInformation()[i];
    for(int j = 0; j < nic; ++j) 
      {
      vtkInformation* inInfo = inVector->GetInformationObject(j);
      vtkExecutive* e;
      int producerPort;
      vtkExecutive::PRODUCER()->Get(inInfo, e, producerPort);
      if (e && vtkThreadedStreamingPipeline::SafeDownCast(e))
        {
        upstream.push_back(e);
        }
      }
    }
  for (size_t i = 0; i < upstream.size(); i++)
    {
    totalUpStreamTime += vtkThreadedStreamingPipeline::SafeDownCast(upstream[i])->LastDataRequestTimeFromSource;
    }

  // Then distribute the resources based on those ratios
  vtkProcessingUnitResource *totalResources[] = 
    {
      resources->GetResourceFor(vtkThreadedStreamingPipeline::PROCESSING_UNIT_CPU),
      resources->GetResourceFor(vtkThreadedStreamingPipeline::PROCESSING_UNIT_GPU),
    };
  for (size_t i = 0; i < upstream.size(); i++) 
    {
    float ratio = vtkThreadedStreamingPipeline::
      SafeDownCast(upstream[i])->LastDataRequestTimeFromSource/totalUpStreamTime;
    for (size_t j = 0; j < sizeof(totalResources)/sizeof(vtkProcessingUnitResource*); j++) 
      {
      vtkProcessingUnitResource* moduleResource = vtkThreadedStreamingPipeline::
        SafeDownCast(upstream[i])->GetResources()->
        GetResourceFor(totalResources[j]->ProcessingUnit());
      moduleResource->IncreaseByRatio(ratio, totalResources[j]);
      }
    }
  
  // Try to reserve upstream
  for (size_t i = 0; i < upstream.size(); i++)
    {
    this->RescheduleFrom(upstream[i],
                         vtkThreadedStreamingPipeline::
                         SafeDownCast(upstream[i])->GetResources());
    }
}

//----------------------------------------------------------------------------
void vtkExecutionScheduler::RescheduleNetwork(vtkExecutive *sink) 
{
  this->Resources->ObtainMaximumResources();
  this->RescheduleFrom(sink, this->Resources);
}

//----------------------------------------------------------------------------
typedef struct 
{
  vtkExecutionScheduler *scheduler;
  Task task;
} ExecutionData;
//----------------------------------------------------------------------------
void vtkExecutionScheduler::implementation::Execute(const Task &task)
{
  ExecutionData *eData = new ExecutionData();
  eData->scheduler = this->Scheduler;
  eData->task = task;
  this->Scheduler->ScheduleThreader->SpawnThread((vtkThreadFunctionType)(vtkExecutionScheduler_ExecuteThread), eData);
}

//----------------------------------------------------------------------------
void * vtkExecutionScheduler_ScheduleThread(void *data)
{
  vtkExecutionScheduler *self = static_cast<vtkExecutionScheduler*>
    (static_cast<vtkMultiThreader::ThreadInfo*>(data)->UserData);
  while (true) 
    {
    self->ScheduleLock->Lock();
    bool needToWait = true;
    vtkExecutionScheduler::implementation::TaskPriorityQueue::iterator ti;
    for (ti = self->Implementation->PrioritizedTasks.begin();
         ti != self->Implementation->PrioritizedTasks.end(); ti++) 
      {
      if (self->Implementation->CanExecuteTask(ti)) 
        {
        vtkThreadedStreamingPipeline *exec = vtkThreadedStreamingPipeline::
          SafeDownCast((*ti).exec);
        if (self->Resources->Reserve(exec->GetResources())) 
          {
          needToWait = false;
          self->Implementation->ExecutingTasks.insert(exec);
          self->ScheduleLock->Unlock();
          self->Implementation->Execute(*ti);
          break;
          }
        }
      }
    if (needToWait) 
      {
      self->ScheduleLock->Unlock();
      self->ScheduleMessager->WaitForMessage();
      }
    }
  return NULL;
}

//----------------------------------------------------------------------------
void * vtkExecutionScheduler_ExecuteThread(void *data)
{
  ExecutionData *eData = static_cast<ExecutionData*>
    (static_cast<vtkMultiThreader::ThreadInfo*>(data)->UserData);
  vtkExecutionScheduler *self = eData->scheduler;
  Task task = eData->task;
  vtkThreadedStreamingPipeline *exec = vtkThreadedStreamingPipeline::
    SafeDownCast(task.exec);
  vtkThreadMessager *messager = self->GetTaskDoneMessager(task.exec);
  vtkMutexLock *lock = self->GetInputsReleasedLock(task.exec);
  exec->GetResources()->Deploy(exec, task.info);
  self->ScheduleLock->Lock();
  self->Implementation->PrioritizedTasks.erase(task);
  self->Implementation->ExecutingTasks.erase(exec);
  self->Resources->Collect(exec->GetResources());
  self->ResourceMessager->SendWakeMessage();
  self->ScheduleLock->Unlock();
  exec->ReleaseInputs();
  self->ScheduleMessager->SendWakeMessage();
  if (task.info && task.info->Has(vtkThreadedStreamingPipeline::AUTO_PROPAGATE()))
    {
      fprintf(stderr, "Push DOWN from %s\n", exec->GetAlgorithm()->GetClassName());
    exec->Push(task.info);
    }
  if (messager)
    {
    messager->SendWakeMessage();
    }
  delete eData;
  fprintf(stderr, "Release now\n");
  lock->Unlock();
  return NULL;
}
