//============================================================================
//  Copyright (c) Kitware, Inc.
//  All rights reserved.
//  See LICENSE.txt 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.
//============================================================================

#ifndef vtk_m_filter_flow_internal_AdvectAlgorithm_h
#define vtk_m_filter_flow_internal_AdvectAlgorithm_h

#include <vtkm/cont/PartitionedDataSet.h>
#include <vtkm/filter/flow/Tracer.h>
#include <vtkm/filter/flow/internal/BoundsMap.h>
#include <vtkm/filter/flow/internal/DataSetIntegrator.h>
#include <vtkm/filter/flow/internal/ParticleMessenger.h>

#include <sstream>

namespace vtkm
{
namespace filter
{
namespace flow
{
namespace internal
{

template <typename DSIType, template <typename> class ResultType, typename ParticleType>
class AdvectAlgorithm
{
public:

  AdvectAlgorithm(const vtkm::filter::flow::internal::BoundsMap& bm,
                  std::vector<DSIType>& blocks,
                  const std::string& commType)
    : Blocks(blocks)
    , BoundsMap(bm)
    , NumRanks(this->Comm.size())
    , Rank(this->Comm.rank())
    , CommType(commType)
  {
  }

  AdvectAlgorithm(const vtkm::filter::flow::internal::BoundsMap& bm,
                  std::vector<DSIType>& blocks,
                  const std::string& commType,
                  vtkm::Id numberOfReceivers,
                  vtkm::Id numberOfParticleInPacket)
    : Blocks(blocks)
    , BoundsMap(bm)
    , NumRanks(this->Comm.size())
    , Rank(this->Comm.rank())
    , CommType(commType)
    , NumberOfReceivers(numberOfReceivers)
    , NumberOfParticleInPacket(numberOfParticleInPacket)
  {
  }

  void Execute(vtkm::Id maxNumSteps,
               vtkm::FloatDefault stepSize,
               const vtkm::cont::ArrayHandle<ParticleType>& seeds)
  {
    this->SetMaxNumberOfSteps(maxNumSteps);
    this->SetStepSize(stepSize);
    this->SetSeeds(seeds);
    this->Go();
  }

  vtkm::cont::PartitionedDataSet GetOutput() const
  {
    vtkm::cont::PartitionedDataSet output;

    for (const auto& b : this->Blocks)
    {
      vtkm::cont::DataSet ds;
      if (b.template GetOutput<ParticleType>(ds))
        output.AppendPartition(ds);
    }

    return output;
  }

  void SetStepSize(vtkm::FloatDefault stepSize) { this->StepSize = stepSize; }
  void SetMaxNumberOfSteps(vtkm::Id numSteps) { this->MaxNumberOfSteps = numSteps; }
  void SetSeeds(const vtkm::cont::ArrayHandle<ParticleType>& seeds)
  {
    this->ClearParticles();

    vtkm::Id n = seeds.GetNumberOfValues();
    auto portal = seeds.ReadPortal();

    std::vector<std::vector<vtkm::Id>> blockIDs;
    std::vector<ParticleType> particles;
    for (vtkm::Id i = 0; i < n; i++)
    {
      const ParticleType p = portal.Get(i);
      std::vector<vtkm::Id> ids = this->BoundsMap.FindBlocks(p.GetPosition());

      //Note: For duplicate blocks, this will give the seeds to the rank that are first in the list.
      if (!ids.empty())
      {
        auto ranks = this->BoundsMap.FindRank(ids[0]);
        if (!ranks.empty() && this->Rank == ranks[0])
        {
          particles.emplace_back(p);
          blockIDs.emplace_back(ids);
        }
      }
    }
    this->SetSeedArray(particles, blockIDs);
  }

  //Advect all the particles.
  virtual void Go()
  {
    vtkm::filter::flow::internal::ParticleMessenger<ParticleType> messenger(this->Comm, this->BoundsMap);
    if (this->CommType == "SYNC")
      messenger.SetUseSyncCommunication();
    else if (this->CommType == "ASYNC")
      messenger.SetUseAsyncCommunication(1, this->NumberOfParticleInPacket, this->NumberOfReceivers);
    else if (this->CommType == "ASYNC_PROBE")
      messenger.SetUseAsyncProbeCommunication();
    else
      throw vtkm::cont::ErrorFilterExecution("Unsupported communication type");

    if(this->Rank == 0){
      std::cout << "NumberOfParticleInPacket " << NumberOfParticleInPacket << " NumberOfRecievers " << NumberOfReceivers << std::endl;
    }

    this->ComputeTotalNumParticles();

    PTRACER(vtkm::filter::flow::GetTracer().Get()->TimeTraceToBuffer("GoStart"));
    vtkm::Id round = 0;
    //if (this->Rank == 0) std::cout<<"Total num seeds= "<<this->TotalNumParticles<<std::endl;
    while (this->TotalNumTerminatedParticles < this->TotalNumParticles)
    {
      //compute the counter information
      PTRACER(vtkm::Id TotalAdvectedSteps = 0);

      PTRACER(vtkm::filter::flow::GetTracer().Get()->TimeTraceToBuffer("AdvectStart"));

      std::vector<ParticleType> v;
      vtkm::Id numTerm = 0, blockId = -1;

      PTRACER(bool ifTracingParticle = false);
      PTRACER(vtkm::filter::flow::GetTracer().Get()->SetBegOverheadStart());
      PTRACER(vtkm::filter::flow::GetTracer().Get()->SetTracingStatus(false));

      if (this->GetActiveParticles(v, blockId))
      {
        PTRACER(vtkm::filter::flow::GetTracer().Get()->SetCurrGangSize(static_cast<vtkm::Id>(v.size())));
        PTRACER(vtkm::filter::flow::GetTracer().Get()->SetAllAdvectStart());
        //make this a pointer to avoid the copy?
        auto& block = this->GetDataSet(blockId);

        //std::cout<<"Debug Rank is " << Rank <<" blockId is "<<blockId<<std::endl;
        DSIHelperInfoType bb =
          DSIHelperInfo<ParticleType>(v, this->BoundsMap, this->ParticleBlockIDsMap);

#ifdef VTKm_INSTRUMENT_PARTICLE_ADVECTION
        //vtkm::filter::flow::GetTracer().Get()->TimeTraceToBuffer("AdvectStart");
        //caculate the advected steps for all particles
        vtkm::Id num0 = 0, num1 = 0, numSmall0 = 0, numSmall1 = 0;
        for (const auto& p : v)
        {
          num0 += p.GetNumberOfSteps();
          numSmall0 += p.GetNumSmallSteps();
        }
        vtkm::Id particleStepsBefore = 0, particleStepsAfter = 0;
        vtkm::filter::flow::GetTracer().Get()->SetTraceID(false);
        if (vtkm::filter::flow::GetTracer().Get()->GetTraceParticleId() >= 0)
        {
          //if dumping particles
          double aliveTime = vtkm::filter::flow::GetTracer().Get()->GetElapsedTime();
          for (const auto& p : v)
          {
            if (vtkm::filter::flow::GetTracer().Get()->IfTracingCustomized(p.GetID()))
            {
              vtkm::filter::flow::GetTracer().Get()->SetTraceID(true);
              //for the inline case, the rank id is same with process id
              //for the intransit case, the rank id is different with process id
              //also adding positions
              vtkm::filter::flow::GetTracer().Get()->ParticleInBlockToBuffer(vtkm::filter::flow::GetTracer().Get()->GetIterationStep(),
                                                this->Rank,
                                                p.GetID(),
                                                p.GetNumberOfSteps(),
                                                p.GetPosition(),
                                                aliveTime);
              particleStepsBefore = p.GetNumberOfSteps();

              double elapsedTime = vtkm::filter::flow::GetTracer().Get()->GetElapsedTime();
              vtkm::filter::flow::GetTracer().Get()->ParticleDetailsToBuffer(vtkm::filter::flow::GetTracer().Get()->GetIterationStep(),
                                                this->Rank,
                                                p.GetID(),
                                                "ADVECTSTART",
                                                elapsedTime,
                                                particleStepsBefore,
                                                vtkm::filter::flow::GetTracer().Get()->GetCurrGangSize(),
                                                vtkm::filter::flow::GetTracer().Get()->GetPrevGangSize());
              ifTracingParticle = true;
              vtkm::filter::flow::GetTracer().Get()->SetTracingStatus(true);
              /*
              std::cout<<"R:"<<this->Rank<<" "<<"P_"<<p.NumComm<<" #S="<<p.GetNumberOfSteps()<<" A= "<<p.A<<" W= "<<p.W<<" BO/EO "<<p.BO<<" "<<p.EO<<std::endl;
              std::cout<<"   s/r: "<<p.SendT<<" "<<p.RecvT<<" G: "<<vtkm::filter::flow::GetTracer().Get()->GetPrevGangSize()<<" "<<vtkm::filter::flow::GetTracer().Get()->GetCurrGangSize()<<std::endl;
              */
              //std::cout<<"R:"<<this->Rank<<" "<<"P_"<<p.NumComm<<" T0= "<<vtkm::filter::flow::GetTracer().Get()->GetElapsedTime()<<" p.OOAW= "<<p.BO<<" "<<p.EO<<" "<<p.A<<" "<<p.W<<std::endl;
            }
          }
        }

        vtkm::filter::flow::GetTracer().Get()->AlgorithmRecorderV("ADVECT_BEGIN_block_numP", {blockId, (int)v.size()});
#endif
        //if (vtkm::filter::flow::GetTracer().Get()->GetTraceID()) std::cout<<"  block.Advect0: "<<vtkm::filter::flow::GetTracer().Get()->GetElapsedTime()<<std::endl;
        block.Advect(bb, this->StepSize, this->MaxNumberOfSteps);
        //if (vtkm::filter::flow::GetTracer().Get()->GetTraceID()) std::cout<<"  block.Advect1: "<<vtkm::filter::flow::GetTracer().Get()->GetElapsedTime()<<std::endl;

#ifdef VTKm_INSTRUMENT_PARTICLE_ADVECTION
        if (vtkm::filter::flow::GetTracer().Get()->GetTraceParticleId() >= 0)
        {
          for (const auto& p : bb.Get<DSIHelperInfo<ParticleType>>().Particles)
          {
            if (vtkm::filter::flow::GetTracer().Get()->IfTracingCustomized(p.GetID()))
            {
              particleStepsAfter = p.GetNumberOfSteps();
              double elapsedTime = vtkm::filter::flow::GetTracer().Get()->GetElapsedTime();
              vtkm::filter::flow::GetTracer().Get()->ParticleDetailsToBuffer(vtkm::filter::flow::GetTracer().Get()->GetIterationStep(),
                                                this->Rank,
                                                p.GetID(),
                                                "ADVECTEND",
                                                elapsedTime,
                                                particleStepsAfter,
                                                vtkm::filter::flow::GetTracer().Get()->GetCurrGangSize(),
                                                vtkm::filter::flow::GetTracer().Get()->GetPrevGangSize());
            }
          }
        }
#endif
        //set this after worklet_end in the integrator
        //vtkm::filter::flow::GetTracer().Get()->SetEndOverheadStart();
        //this will be replaced by the worklet start/end
        //vtkm::filter::flow::GetTracer().Get()->TimeTraceToBuffer("ParticleAdvectEnd");
        numTerm = this->UpdateResult(bb.Get<DSIHelperInfo<ParticleType>>());

#ifdef VTKm_INSTRUMENT_PARTICLE_ADVECTION
        // the information in bb is updated
        for (const auto& p : bb.Get<DSIHelperInfo<ParticleType>>().Particles)
        {
          num1 += p.GetNumberOfSteps();
          numSmall1 += p.GetNumSmallSteps();
        }
        TotalAdvectedSteps = num1 - num0;
        std::string infoStr = "ParticleAdvectInfo_" + std::to_string((int)(vtkm::filter::flow::GetTracer().Get()->GetCurrGangSize())) + "_" +
          std::to_string(TotalAdvectedSteps) + "_" + std::to_string(numSmall1 - numSmall0) + "_"+std::to_string((int)vtkm::filter::flow::GetTracer().Get()->GetPrevGangSize()) + "_" + std::to_string(blockId);
        vtkm::filter::flow::GetTracer().Get()->TimeTraceToBuffer(infoStr);
        vtkm::filter::flow::GetTracer().Get()->SetEndOverheadEnd();
        vtkm::filter::flow::GetTracer().Get()->SetAllAdvectEnd();
        //this->UpdateParticleTimers(bb.Get<DSIHelperInfo<ParticleType>>());
        vtkm::filter::flow::GetTracer().Get()->AlgorithmRecorderV("ADVECT_END_nsteps_nterm", {(int)TotalAdvectedSteps, (int)numTerm});
#endif
      }
      else
      {
        PTRACER(vtkm::filter::flow::GetTracer().Get()->SetCurrGangSize(0));
        PTRACER(vtkm::filter::flow::GetTracer().Get()->SetBegOverheadEnd());
      }
      PTRACER(vtkm::filter::flow::GetTracer().Get()->TimeTraceToBuffer("AdvectEnd"));

      vtkm::Id numTermMessages = 0;
#ifdef VTKm_INSTRUMENT_PARTICLE_ADVECTION
      vtkm::filter::flow::GetTracer().Get()->TimeTraceToBuffer("CommStart");
      if (ifTracingParticle)
      {
        double elapsedTime = vtkm::filter::flow::GetTracer().Get()->GetElapsedTime();
        vtkm::filter::flow::GetTracer().Get()->ParticleDetailsToBuffer(
          vtkm::filter::flow::GetTracer().Get()->GetIterationStep(), this->Rank, -1, "GANG_COMM_START", elapsedTime, 0, 0, 0);
      }
#endif
      std::unordered_map<int, int> sendParticleInfo;
      //if (vtkm::filter::flow::GetTracer().Get()->GetTraceID()) std::cout<<"  Comm0: "<<vtkm::filter::flow::GetTracer().Get()->GetElapsedTime()<<std::endl;
      this->Communicate(messenger, numTerm, numTermMessages, sendParticleInfo, PTRACER(ifTracingParticle));
      //if (vtkm::filter::flow::GetTracer().Get()->GetTraceID()) std::cout<<"  Comm1: "<<vtkm::filter::flow::GetTracer().Get()->GetElapsedTime()<<std::endl;

#ifdef VTKm_INSTRUMENT_PARTICLE_ADVECTION
      vtkm::filter::flow::GetTracer().Get()->TimeTraceToBuffer("CommEnd");
      if (ifTracingParticle)
      {
        double elapsedTime = vtkm::filter::flow::GetTracer().Get()->GetElapsedTime();
        vtkm::filter::flow::GetTracer().Get()->ParticleDetailsToBuffer(
          vtkm::filter::flow::GetTracer().Get()->GetIterationStep(), this->Rank, -1, "GANG_COMM_END", elapsedTime, 0, 0, 0);
      }
#endif

      this->TotalNumTerminatedParticles += (numTerm + numTermMessages);
      if (this->TotalNumTerminatedParticles > this->TotalNumParticles)
        throw vtkm::cont::ErrorFilterExecution("Particle count error");

      round++;
      PTRACER(vtkm::filter::flow::GetTracer().Get()->CounterToBuffer(round, TotalAdvectedSteps, sendParticleInfo));
    }
  }

  virtual void ClearParticles()
  {
    this->Active.clear();
    this->Inactive.clear();
    this->ParticleBlockIDsMap.clear();
  }

  void ComputeTotalNumParticles()
  {
    vtkm::Id numLocal = static_cast<vtkm::Id>(this->Inactive.size());
    for (const auto& it : this->Active)
      numLocal += it.second.size();

#ifdef VTKM_ENABLE_MPI
    vtkmdiy::mpi::all_reduce(this->Comm, numLocal, this->TotalNumParticles, std::plus<vtkm::Id>{});
#else
    this->TotalNumParticles = numLocal;
#endif
  }

  DataSetIntegrator<DSIType>& GetDataSet(vtkm::Id id)
  {
    for (auto& it : this->Blocks)
      if (it.GetID() == id)
        return it;

    throw vtkm::cont::ErrorFilterExecution("Bad block");
  }

  virtual void SetSeedArray(const std::vector<ParticleType>& particles,
                            const std::vector<std::vector<vtkm::Id>>& blockIds)
  {
    VTKM_ASSERT(particles.size() == blockIds.size());

    auto pit = particles.begin();
    auto bit = blockIds.begin();
    while (pit != particles.end() && bit != blockIds.end())
    {
      vtkm::Id blockId0 = (*bit)[0];
      this->ParticleBlockIDsMap[pit->GetID()] = *bit;
      if (this->Active.find(blockId0) == this->Active.end())
        this->Active[blockId0] = { *pit };
      else
        this->Active[blockId0].emplace_back(*pit);
      pit++;
      bit++;
    }
  }

  virtual bool GetActiveParticles(std::vector<ParticleType>& particles, vtkm::Id& blockId)
  {
    particles.clear();
    blockId = -1;
    if (this->Active.empty())
      return false;

    PTRACER(vtkm::filter::flow::GetTracer().Get()->TimeTraceToBuffer("GetActiveParticlesStart"));
    //If only one, return it.
    if (this->Active.size() == 1)
    {
      blockId = this->Active.begin()->first;
      particles = std::move(this->Active.begin()->second);
      this->Active.clear();
    }
    else
    {
      //Find the blockId with the most particles.
      std::size_t maxNum = 0;
      auto maxIt = this->Active.end();
      for (auto it = this->Active.begin(); it != this->Active.end(); it++)
      {
        auto sz = it->second.size();
        if (sz > maxNum)
        {
          maxNum = sz;
          maxIt = it;
        }
      }

      if (maxNum == 0)
      {
        this->Active.clear();
        return false;
      }

      blockId = maxIt->first;
      particles = std::move(maxIt->second);
      this->Active.erase(maxIt);
    }

    PTRACER(vtkm::filter::flow::GetTracer().Get()->TimeTraceToBuffer("GetActiveParticlesStop"));
    return !particles.empty();
  }

  //for sendParticleInfo, key is the dest id, value is number of particles to dest
  void Communicate(vtkm::filter::flow::internal::ParticleMessenger<ParticleType>& messenger,
                   vtkm::Id numLocalTerminations,
                   vtkm::Id& numTermMessages,
                   std::unordered_map<int, int>& sendParticleInfo,
                   bool ifTracingParticle=false)
  {
    std::vector<ParticleType> outgoing;
    std::vector<vtkm::Id> outgoingRanks;
    this->GetOutgoingParticles(outgoing, outgoingRanks);

    std::vector<ParticleType> incoming;
    std::unordered_map<vtkm::Id, std::vector<vtkm::Id>> incomingBlockIDs;
    numTermMessages = 0;
    bool block = false;
#ifdef VTKM_ENABLE_MPI
    block = this->GetBlockAndWait(messenger.UsingSyncCommunication(), numLocalTerminations);
#endif
    messenger.Exchange(sendParticleInfo,
                       outgoing,
                       outgoingRanks,
                       this->ParticleBlockIDsMap,
                       numLocalTerminations,
                       incoming,
                       incomingBlockIDs,
                       numTermMessages,
                       block,
                       ifTracingParticle);

    //Cleanup what was sent.
    for (const auto& p : outgoing)
      this->ParticleBlockIDsMap.erase(p.GetID());

    this->UpdateActive(incoming, incomingBlockIDs);
  }

  void GetOutgoingParticles(std::vector<ParticleType>& outgoing,
                            std::vector<vtkm::Id>& outgoingRanks)
  {
    outgoing.clear();
    outgoingRanks.clear();

    outgoing.reserve(this->Inactive.size());
    outgoingRanks.reserve(this->Inactive.size());

    std::vector<ParticleType> particlesStaying;
    std::unordered_map<vtkm::Id, std::vector<vtkm::Id>> particlesStayingBlockIDs;

#ifdef VTKm_INSTRUMENT_PARTICLE_ADVECTION
    auto boT = vtkm::filter::flow::GetTracer().Get()->GetBegOverheadEnd() - vtkm::filter::flow::GetTracer().Get()->GetBegOverheadStart();
    auto eoT = vtkm::filter::flow::GetTracer().Get()->GetEndOverheadEnd() - vtkm::filter::flow::GetTracer().Get()->GetEndOverheadStart();
    auto advT = vtkm::filter::flow::GetTracer().Get()->GetAdvectEnd() - vtkm::filter::flow::GetTracer().Get()->GetAdvectStart();
#endif

    //Send out Everything.
    //for (const auto& p : this->Inactive)
    for (auto& p : this->Inactive)
    {
      const auto& bid = this->ParticleBlockIDsMap[p.GetID()];
      VTKM_ASSERT(!bid.empty());
      auto ranks = this->BoundsMap.FindRank(bid[0]);
      VTKM_ASSERT(!ranks.empty());

      //update BO EO and Advc
#ifdef VTKm_INSTRUMENT_PARTICLE_ADVECTION
      p.BO += boT;
      p.EO += eoT;
      p.A += advT;
      p.PrevGangSize += vtkm::filter::flow::GetTracer().Get()->GetPrevGangSize();
      p.GangSize += vtkm::filter::flow::GetTracer().Get()->GetCurrGangSize();
#endif

      if (ranks.size() == 1)
      {
        if (ranks[0] == this->Rank)
        {
          //goes to the block located in the same rank
          //do not add the communication
#ifdef VTKm_INSTRUMENT_PARTICLE_ADVECTION
          p.AddNumTraveledBlocks();
#endif
          particlesStaying.emplace_back(p);
          particlesStayingBlockIDs[p.GetID()] = this->ParticleBlockIDsMap[p.GetID()];
        }
        else
        {
#ifdef VTKm_INSTRUMENT_PARTICLE_ADVECTION
          p.AddNumTraveledBlocks();
          p.AddNumComm();
#endif
          outgoing.emplace_back(p);
          outgoingRanks.emplace_back(ranks[0]);
        }
      }
      else
      {
        //Decide where it should go...

        //Random selection:
        vtkm::Id outRank = ranks[static_cast<std::size_t>(std::rand() % ranks.size())];
        if (outRank == this->Rank)
        {
          //the dedicated block is in same rank
#ifdef VTKm_INSTRUMENT_PARTICLE_ADVECTION
          p.AddNumTraveledBlocks();
#endif
          particlesStayingBlockIDs[p.GetID()] = this->ParticleBlockIDsMap[p.GetID()];
          particlesStaying.emplace_back(p);
        }
        else
        {
          //comm num + 1
#ifdef VTKm_INSTRUMENT_PARTICLE_ADVECTION
          p.AddNumTraveledBlocks();
          p.AddNumComm();
#endif
          outgoing.emplace_back(p);
          outgoingRanks.emplace_back(outRank);
        }
      }
    }
    this->Inactive.clear();
    VTKM_ASSERT(outgoing.size() == outgoingRanks.size());
    VTKM_ASSERT(particlesStaying.size() == particlesStayingBlockIDs.size());
    if (!particlesStaying.empty())
      this->UpdateActive(particlesStaying, particlesStayingBlockIDs);
  }

  virtual void UpdateActive(const std::vector<ParticleType>& particles,
                            const std::unordered_map<vtkm::Id, std::vector<vtkm::Id>>& idsMap)
  {
    VTKM_ASSERT(particles.size() == idsMap.size());

    for (auto pit = particles.begin(); pit != particles.end(); pit++)
    {
      vtkm::Id particleID = pit->GetID();
      const auto& it = idsMap.find(particleID);
      VTKM_ASSERT(it != idsMap.end() && !it->second.empty());
      vtkm::Id blockId = it->second[0];
      this->Active[blockId].emplace_back(*pit);
    }

    for (const auto& it : idsMap)
      this->ParticleBlockIDsMap[it.first] = it.second;
  }

  virtual void UpdateInactive(const std::vector<ParticleType>& particles,
                              const std::vector<vtkm::Id>& indices,
                              const std::vector<std::vector<vtkm::Id>>& idsMap)
  {
    VTKM_ASSERT(indices.size() == idsMap.size());

    std::size_t n = indices.size();
    for (std::size_t i = 0; i < n; i++)
    {
      vtkm::Id idx = indices[i];
      const auto& p = particles[idx];
      this->Inactive.emplace_back(p);
      this->ParticleBlockIDsMap[p.GetID()] = idsMap[i];
    }
  }

  vtkm::Id UpdateResult(const DSIHelperInfo<ParticleType>& stuff)
  {
    VTKM_ASSERT(stuff.InIdx.size() == 0);
    //this->UpdateActive(stuff.Particles, stuff.InIdx, stuff.InBIDs);
    this->UpdateInactive(stuff.Particles, stuff.OutIdx, stuff.OutBIDs);

    vtkm::Id numTerm = static_cast<vtkm::Id>(stuff.TermIdx.size());
    //Update terminated particles.
    if (numTerm > 0)
    {
      for (const auto& idx : stuff.TermIdx)
        this->ParticleBlockIDsMap.erase(stuff.Particles[idx].GetID());
    }

    return numTerm;
  }


  virtual bool GetBlockAndWait(const bool& syncComm, const vtkm::Id& numLocalTerm)
  {
    bool haveNoWork = this->Active.empty() && this->Inactive.empty();

    //Using syncronous communication we should only block and wait if we have no particles
    if (syncComm)
    {
      return haveNoWork;
    }
    else
    {
      //Otherwise, for asyncronous communication, there are only two cases where blocking would deadlock.
      //1. There are active particles.
      //2. numLocalTerm + this->TotalNumberOfTerminatedParticles == this->TotalNumberOfParticles
      //So, if neither are true, we can safely block and wait for communication to come in.

      if (haveNoWork &&
          (numLocalTerm + this->TotalNumTerminatedParticles < this->TotalNumParticles))
      {
        std::stringstream sst;
        sst<<"** COMM_Block= True. (A= "<<this->Active.size()<<" I= "<<this->Inactive.size()<<" haveNoWork= "<<haveNoWork<<" nLocT= "<<numLocalTerm<<" totT= "<<this->TotalNumTerminatedParticles<<") ";
        PTRACER(vtkm::filter::flow::GetTracer().Get()->TimeTraceToBuffer(sst.str()));
        return true;
      }

      std::stringstream sst;
      sst<<"** COMM_Block= False. (A= "<<this->Active.size()<<" I= "<<this->Inactive.size()<<" haveNoWork= "<<haveNoWork<<" nLocT= "<<numLocalTerm<<" totT= "<<this->TotalNumTerminatedParticles<<") ";
      PTRACER(vtkm::filter::flow::GetTracer().Get()->TimeTraceToBuffer(sst.str()));
      return false;
    }
  }

#ifdef VTKm_INSTRUMENT_PARTICLE_ADVECTION
  void UpdateParticleTimers(DSIHelperInfo<ParticleType>& stuff) const
  {
//    auto boT = vtkm::filter::flow::GetTracer().Get()->GetBegOverheadEnd() - vtkm::filter::flow::GetTracer().Get()->GetBegOverheadStart();
//    auto eoT = vtkm::filter::flow::GetTracer().Get()->GetEndOverheadEnd() - vtkm::filter::flow::GetTracer().Get()->GetEndOverheadStart();
//    auto advT = vtkm::filter::flow::GetTracer().Get()->GetAdvectEnd() - vtkm::filter::flow::GetTracer().Get()->GetAdvectStart();
    auto allAdvT = vtkm::filter::flow::GetTracer().Get()->GetAllAdvectEnd() - vtkm::filter::flow::GetTracer().Get()->GetAllAdvectStart();

    for (auto& p : stuff.Particles)
    {
      p.AllA += allAdvT;
    }
  }
#endif
  /*
  void UpdateParticleTimers(std::vector<ParticleType>& particles) const
  {

    auto boT = vtkm::filter::flow::GetTracer().Get()->GetBegOverheadEnd() - vtkm::filter::flow::GetTracer().Get()->GetBegOverheadStart();
    auto eoT = vtkm::filter::flow::GetTracer().Get()->GetEndOverheadEnd() - vtkm::filter::flow::GetTracer().Get()->GetEndOverheadStart();
    auto advT = vtkm::filter::flow::GetTracer().Get()->GetAdvectEnd() - vtkm::filter::flow::GetTracer().Get()->GetAdvectStart();

    for (auto& p : particles)
    {
      p.BO += boT;
      p.EO += eoT;
      p.A += advT;
      //if (p.NumComm > 2) //GetID() > 80 && p.GetID() < 100)
      //  std::cout << p << " #comm=" << p.NumComm << " BO/EO=" << p.BO << " " << p.EO << " A=" << p.A
      //            << " W=" << p.W << std::endl;
    }
  }
  */

  //Member data
  // {blockId, std::vector of particles}
  std::unordered_map<vtkm::Id, std::vector<ParticleType>> Active;
  std::vector<DSIType> Blocks;
  vtkm::filter::flow::internal::BoundsMap BoundsMap;
  vtkmdiy::mpi::communicator Comm = vtkm::cont::EnvironmentTracker::GetCommunicator();
  std::vector<ParticleType> Inactive;
  vtkm::Id MaxNumberOfSteps = 0;
  vtkm::Id NumRanks;
  //{particleId : {block IDs}}
  std::unordered_map<vtkm::Id, std::vector<vtkm::Id>> ParticleBlockIDsMap;
  vtkm::Id Rank;
  vtkm::FloatDefault StepSize;
  vtkm::Id TotalNumParticles = 0;
  vtkm::Id TotalNumTerminatedParticles = 0;
  std::string CommType = "ASYNC";
  vtkm::Id NumberOfReceivers;
  vtkm::Id NumberOfParticleInPacket;
};
}
}
}
} //vtkm::filter::flow::internal

#endif //vtk_m_filter_flow_internal_AdvectAlgorithm_h
