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

  Program:   ParaView
  Module:    $RCSfile$

  Copyright (c) Kitware, Inc.
  All rights reserved.
  See Copyright.txt or http://www.paraview.org/HTML/Copyright.html 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.

=========================================================================*/
#include "vtkStreamingParticlesPriorityQueue.h"

#include "vtkCompositeDataPipeline.h"
#include "vtkDataObjectTreeIterator.h"
#include "vtkInformation.h"
#include "vtkMultiBlockDataSet.h"
#include "vtkMultiProcessController.h"
#include "vtkObjectFactory.h"
#include "vtkStreamingDemandDrivenPipeline.h"
#include "vtkStreamingPriorityQueue.h"

#include <algorithm>
#include <assert.h>
#include <queue>
#include <vector>
#include <set>
#include <map>

namespace
{
  class vtkParticlesComparator
    {
  public:
    bool operator()(const vtkStreamingPriorityQueueItem& me, const
      vtkStreamingPriorityQueueItem& other) const
      {
      if (me.ScreenCoverage != other.ScreenCoverage)
        {
        return me.ScreenCoverage < other.ScreenCoverage;
        }
      if (me.Refinement != other.Refinement)
        {
        return me.Refinement > other.Refinement;
        }
      return me.Distance > other.Distance;
      }
    };
}
class vtkStreamingParticlesPriorityQueue::vtkInternals
{
public:
  vtkSmartPointer<vtkMultiBlockDataSet> Metadata;
  std::queue<unsigned int> BlocksToRequest;
  std::set<unsigned int> BlocksRequested;
  std::set<unsigned int> BlocksToPurge;

  double PreviousViewPlanes[24];

  vtkInternals()
    {
    this->ResetPreviousViewPlanes();
    }
  void ResetPreviousViewPlanes()
    {
    memset(this->PreviousViewPlanes, 0, sizeof(double)*24);
    }
  bool PlanesChanged(const double view_planes[24])
    {
#ifdef _MSC_VER
#pragma warning(push)
// Disable C4996 because it warns us that the pointer range here cannot be
// guaranteed by the runtime to not overflow (we're not using Microsoft's
// checked iterators). Since we know that 24 is the size of everything here,
// suppress the warning.
#pragma warning(disable:4996)
#endif
    return !std::equal(this->PreviousViewPlanes, this->PreviousViewPlanes + 24, view_planes);
#ifdef _MSC_VER
#pragma warning(pop)
#endif
    }
  void SetViewPlanes(const double view_planes[24])
    {
    std::copy(view_planes, view_planes + 24, this->PreviousViewPlanes);
    }
};

vtkStandardNewMacro(vtkStreamingParticlesPriorityQueue);
vtkCxxSetObjectMacro(vtkStreamingParticlesPriorityQueue, Controller, vtkMultiProcessController);
//----------------------------------------------------------------------------
vtkStreamingParticlesPriorityQueue::vtkStreamingParticlesPriorityQueue()
{
  this->Internals = new vtkInternals();
  this->Controller = 0;
  this->UseBlockDetailInformation = false;
  this->AnyProcessCanLoadAnyBlock = true;
  this->DetailLevelToLoad = 8.5e-5;
  this->SetController(vtkMultiProcessController::GetGlobalController());
}

//----------------------------------------------------------------------------
vtkStreamingParticlesPriorityQueue::~vtkStreamingParticlesPriorityQueue()
{
  delete this->Internals;
  this->Internals = 0;
  this->SetController(0);
}

//----------------------------------------------------------------------------
void vtkStreamingParticlesPriorityQueue::Initialize(vtkMultiBlockDataSet* metadata)
{
  delete this->Internals;
  this->Internals = new vtkInternals();
  this->Internals->Metadata = metadata;
}

//----------------------------------------------------------------------------
void vtkStreamingParticlesPriorityQueue::Reinitialize()
{
  if (this->Internals->Metadata)
    {
    std::set<unsigned int> blocksRequested;
    blocksRequested.swap(this->Internals->BlocksRequested);

    vtkSmartPointer<vtkMultiBlockDataSet> info = this->Internals->Metadata;
    this->Initialize(info);

    // restore blocks requested since data didn;t change.
    this->Internals->BlocksRequested.swap(blocksRequested);
    }
}

static inline bool hasOneLikeXButGreater(unsigned int x, unsigned int stride,
                           std::map<unsigned, unsigned>& map)
{
  std::map<unsigned, unsigned>::iterator itr = map.find(x % stride);
  if (itr == map.end())
    {
    return false;
    }
  else
    {
    return itr->second >= x;
    }
}

static inline bool hasOneLikeXButLess(unsigned int x, unsigned int stride,
                                      std::map<unsigned, unsigned>& map)
{
  std::map<unsigned,unsigned>::iterator itr = map.find(x%stride);
  if (itr == map.end())
    {
    return false;
    }
  else
    {
    return itr->second < x;
    }
}

//----------------------------------------------------------------------------
void vtkStreamingParticlesPriorityQueue::UpdatePriorities(
  const double view_planes[24])
{
  assert(this->Internals && this->Internals->Metadata);
  assert(this->Internals->BlocksToRequest.empty());

  vtkMultiBlockDataSet* metadata = this->Internals->Metadata;

  // This assumes for following structure:
  // Root
  //   Level 0
  //     DS 0 (Block Idx 0)
  //     DS 1 (Block Idx 1)
  //   Level 1
  //     DS 0 (Block Idx 2)
  //     DS 1 (Block Idx 3)
  //       .
  //       .
  //       .
  // Where "Block Idx" is the key that needs to be sent up the pipeline to the
  // reader to request a particular block and Level k is lower refinement than
  // Level (k+1).

  vtkStreamingPriorityQueue<vtkParticlesComparator> queue;

  unsigned int block_index = 0;
  unsigned int num_levels = metadata->GetNumberOfBlocks();
  unsigned int num_block_per_level = 0;
  bool all_levels_have_same_block_count = true;
  for (unsigned int level=0; level < num_levels; level++)
    {
    vtkMultiBlockDataSet* mb =
      vtkMultiBlockDataSet::SafeDownCast(metadata->GetBlock(level));
    assert(mb != NULL);

    unsigned int num_blocks = mb->GetNumberOfBlocks();
    if (num_blocks > num_block_per_level)
      {
      num_block_per_level = num_blocks;
      all_levels_have_same_block_count = level == 0;
      }
    for (unsigned int cc=0; cc < num_blocks; cc++, block_index++)
      {
      if (!mb->HasMetaData(cc) ||
          !mb->GetMetaData(cc)->Has(
            vtkStreamingDemandDrivenPipeline::BOUNDS()))
        {
        continue;
        }

      vtkStreamingPriorityQueueItem item;
      item.Identifier = block_index;
      item.Refinement = level;

      double bounds[6];
      vtkInformation* blockInfo = mb->GetMetaData(cc);
      blockInfo->Get(vtkStreamingDemandDrivenPipeline::BOUNDS(), bounds);
      item.Bounds.SetBounds(bounds);
      if (blockInfo->Has(vtkCompositeDataPipeline::BLOCK_AMOUNT_OF_DETAIL()))
        {
        item.AmountOfDetail = blockInfo->Get(
              vtkCompositeDataPipeline::BLOCK_AMOUNT_OF_DETAIL());
        }
      if (this->AnyProcessCanLoadAnyBlock ||
          (blockInfo->Has(vtkCompositeDataSet::CURRENT_PROCESS_CAN_LOAD_BLOCK()) &&
          blockInfo->Get(vtkCompositeDataSet::CURRENT_PROCESS_CAN_LOAD_BLOCK())))
        {
        queue.push(item);
        }
      }
    }
  if (!all_levels_have_same_block_count)
    {
    num_block_per_level *= num_levels;
    }
  double clamp_bounds[6];
  vtkMath::UninitializeBounds(clamp_bounds);
  queue.UpdatePriorities(view_planes, clamp_bounds);

  std::map<unsigned, unsigned> blocksRequested;
  for (std::set<unsigned>::iterator itr = this->Internals->BlocksRequested.begin();
       itr != this->Internals->BlocksRequested.end(); ++itr)
    {
    blocksRequested[*itr % num_block_per_level] = *itr;
    }
  this->Internals->BlocksRequested.clear();

  this->Internals->BlocksToPurge.clear();

  std::deque<unsigned int> toRequest;
  std::map<unsigned, unsigned> keepInRequest;
  while (!queue.empty())
    {
    vtkStreamingPriorityQueueItem item = queue.top();

    queue.pop();
//    if (item.Distance + item.Refinement <= 0 || item.Refinement < 1)
    double diagonal = item.Bounds.GetDiagonalLength();
    // avoid division by 0
    if (diagonal < 1e-10)
      {
      diagonal = 1e-10;
      }
    double factor = item.Refinement == 0 ? 0 : this->DetailLevelToLoad / (double) item.Refinement;
    bool detailMethodNeedsBlock =
        (item.Refinement <= 0 || (item.Distance / diagonal < factor && item.ScreenCoverage > 0));
//        (item.Refinement <= 0 ||
//         (item.ItemCoverage > 0 && item.ScreenCoverage / (item.AmountOfDetail * item.ItemCoverage ) > this->DetailLevelToLoad));
    bool genericMethodNeedsBlock =
        (item.Refinement <= 1 || item.ScreenCoverage >= 0.75);

    if ( (this->UseBlockDetailInformation && item.AmountOfDetail > 0) ? detailMethodNeedsBlock : genericMethodNeedsBlock)
      {
      if (hasOneLikeXButGreater(item.Identifier,num_block_per_level,keepInRequest) ||
          hasOneLikeXButGreater(item.Identifier,num_block_per_level,blocksRequested))
        {
        // we already requested a greater resolution of the block, do nothing
        }
      else if (hasOneLikeXButLess(item.Identifier,num_block_per_level,blocksRequested))
        {
        // we have loaded a lower resolution of the block:
        // 1: request the new (higher) one
        toRequest.push_back(item.Identifier);
        keepInRequest[item.Identifier % num_block_per_level] = item.Identifier;
        // 2: delete the one we currently have loaded
        this->Internals->BlocksToPurge.insert(blocksRequested[item.Identifier % num_block_per_level]);
        blocksRequested.erase(item.Identifier % num_block_per_level);
        }
      else if (hasOneLikeXButLess(item.Identifier,num_block_per_level,keepInRequest))
        {
        // we are requesting a lower resolution of the block, so change the request
        // to the highest resolution we need
        keepInRequest[item.Identifier % num_block_per_level] = item.Identifier;
        toRequest.push_back(item.Identifier);
        }
      else
        {
        // we don't have this block loaded or requested at all, request the current
        // resolution
        keepInRequest[item.Identifier % num_block_per_level] = item.Identifier;
        toRequest.push_back(item.Identifier);
        }
      }
    else
      {
      // if we don't need the block at this resolution, but we have it requested at this
      // level or higher then delete the resolution we have and request the resolution
      // below the current one we don't need.  We have to check the request list too since
      // we are adding to the request list without knowing for sure that we need the resolution
      // we are reqeusting.  Ideally if we are seeing a higher res block in this list, we have
      // already seen the lower resolution, but that is not guaranteed by the priority of the blocks.
      if (hasOneLikeXButGreater(item.Identifier,num_block_per_level,blocksRequested) ||
          hasOneLikeXButGreater(item.Identifier,num_block_per_level,keepInRequest))
        {
        // request the next one lower than this one (may not need that one, but we should show something for each)
        // use max to ensure we show the lowest one not an invalid block index
        keepInRequest[item.Identifier % num_block_per_level] =
            std::max((int)item.Identifier - (int)num_block_per_level,(int)(item.Identifier % num_block_per_level));
        toRequest.push_back(
              std::max((int)item.Identifier - (int)num_block_per_level,(int)(item.Identifier % num_block_per_level)));
        // this may be called when the block in question is merely to be requested and not loaded yet
        // if the block is loaded, delete it
        if (hasOneLikeXButGreater(item.Identifier,num_block_per_level,blocksRequested))
          {
          this->Internals->BlocksToPurge.insert(blocksRequested[item.Identifier % num_block_per_level]);
          blocksRequested.erase(item.Identifier % num_block_per_level);
          }
        }
      }
    }

  for (std::deque<unsigned int>::iterator itr = toRequest.begin(); itr != toRequest.end(); ++itr)
    {
    std::map<unsigned,unsigned>::iterator keep = keepInRequest.find(*itr % num_block_per_level);
    std::set<unsigned>::iterator purge = this->Internals->BlocksToPurge.find(*itr);
    if (keep != keepInRequest.end() && keep->second == *itr &&
        (all_levels_have_same_block_count  || purge == this->Internals->BlocksToPurge.end()))
      {
      this->Internals->BlocksToRequest.push(*itr);
      }
    }
  for (std::map<unsigned,unsigned>::iterator itr = blocksRequested.begin();
       itr != blocksRequested.end(); ++itr)
    {
    this->Internals->BlocksRequested.insert(itr->second);
    }

  cout << "Update information  : "  << endl
       << "  To request        : " << this->Internals->BlocksToRequest.size() << endl
       << "  Already requested : " << this->Internals->BlocksRequested.size() << endl
       << "  To purge          : " << this->Internals->BlocksToPurge.size() << endl;
}

//----------------------------------------------------------------------------
bool vtkStreamingParticlesPriorityQueue::IsEmpty()
{
  return this->Internals->BlocksToRequest.empty();
}

//----------------------------------------------------------------------------
unsigned int vtkStreamingParticlesPriorityQueue::Pop()
{
  if (this->IsEmpty())
    {
    return VTK_UNSIGNED_INT_MAX;
    }

  if (this->AnyProcessCanLoadAnyBlock)
    {
    int myid = this->Controller->GetLocalProcessId();
    int num_ranks = this->Controller->GetNumberOfProcesses();
    std::vector< unsigned int > items;
    items.resize(num_ranks);
    for (int i = 0; i < num_ranks; ++i)
      {
      items[i] = this->Internals->BlocksToRequest.front();
      this->Internals->BlocksToRequest.pop();
      this->Internals->BlocksRequested.insert(items[i]);
      }
    return items[myid];
    }
  else
    {
    unsigned int item;

    item = this->Internals->BlocksToRequest.front();
    this->Internals->BlocksToRequest.pop();
    this->Internals->BlocksRequested.insert(item);

    // As the queue is assumed to be of the local process's blocks now, return
    // the first block in the queue (will be different on each process
    return item;
    }
}

//----------------------------------------------------------------------------
void vtkStreamingParticlesPriorityQueue::Update(const double view_planes[24])
{
  double clamp_bounds[6];
  vtkMath::UninitializeBounds(clamp_bounds);
  this->Update(view_planes, clamp_bounds);
}

//----------------------------------------------------------------------------
void vtkStreamingParticlesPriorityQueue::Update(const double view_planes[24],
  const double vtkNotUsed(clamp_bounds)[6])
{
  this->Internals->BlocksToPurge.clear();

  if (!this->Internals->Metadata)
    {
    return;
    }

  // Check if the view has changed. If so, we update the priorities.
  if (!this->Internals->PlanesChanged(view_planes))
    {
    return;
    }

  this->Reinitialize();
  this->UpdatePriorities(view_planes);
  this->Internals->SetViewPlanes(view_planes);
}

//----------------------------------------------------------------------------
const std::set<unsigned int>& vtkStreamingParticlesPriorityQueue::GetBlocksToPurge() const
{
  return this->Internals->BlocksToPurge;
}

//----------------------------------------------------------------------------
void vtkStreamingParticlesPriorityQueue::PrintSelf(ostream& os, vtkIndent indent)
{
  this->Superclass::PrintSelf(os, indent);
  os << indent << "Controller: " << this->Controller << endl;
}
