/* Copyright (C) 2003 MySQL AB

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 2 of the License, or
   (at your option) any later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */


#include "SimulatedBlock.hpp"
#include "SafeCounter.hpp"
#include <signaldata/NodeFailRep.hpp>

SafeCounterManager::SafeCounterManager(class SimulatedBlock & block)
  : m_block(block),
    m_activeCounters(m_counterPool)
{}
  
bool
SafeCounterManager::setSize(Uint32 maxNoOfActiveMutexes) {
  return m_counterPool.setSize(maxNoOfActiveMutexes);
}

Uint32
SafeCounterManager::getSize() const {
  return m_counterPool.getSize();
}

bool
SafeCounterManager::seize(ActiveCounterPtr& ptr){
  return m_activeCounters.seize(ptr);
}

void
SafeCounterManager::release(ActiveCounterPtr& ptr){
  m_activeCounters.release(ptr);
}

void
SafeCounterManager::getPtr(ActiveCounterPtr& ptr, Uint32 ptrI){
  m_activeCounters.getPtr(ptr, ptrI);
}


void
SafeCounterManager::printNODE_FAILREP(){
  ActiveCounterPtr ptr;

  NodeBitmask nodes;
  nodes.clear();
  //  nodes.bitORC(nodes);

  for(m_activeCounters.first(ptr); !ptr.isNull(); m_activeCounters.next(ptr)){
    ActiveCounter::SignalDesc desc = ptr.p->m_signalDesc;
    ndbout_c("theData[desc.m_senderDataOffset=%u] = %u",
	     desc.m_senderDataOffset, ptr.p->m_senderData);
    ndbout_c("theData[desc.m_errorCodeOffset=%u] = %u",
	     desc.m_errorCodeOffset, desc.m_nodeFailErrorCode);
    Uint32 len = MAX(MAX(desc.m_senderDataOffset, desc.m_errorCodeOffset),
		     desc.m_senderRefOffset);
    
    NodeBitmask overlapping = ptr.p->m_nodes;
    Uint32 i = 0;
    while((i = overlapping.find(i)) != NodeBitmask::NotFound){
      ndbout_c("  theData[desc.m_senderRefOffset=%u] = %x",
	       desc.m_senderRefOffset, numberToRef(desc.m_block, i));
      ndbout_c("  sendSignal(%x,%u,signal,%u,JBB",
	       m_block.reference(), desc.m_gsn, len+1);
      i++;
    }
  }
}

void
SafeCounterManager::execNODE_FAILREP(Signal* signal){
  Uint32 * theData = signal->getDataPtrSend();
  ActiveCounterPtr ptr;
  NodeBitmask nodes;
  nodes.assign(NodeBitmask::Size, 
	       ((const NodeFailRep*)signal->getDataPtr())->theNodes);

  for(m_activeCounters.first(ptr); !ptr.isNull(); m_activeCounters.next(ptr)){
    if(nodes.overlaps(ptr.p->m_nodes)){
      ActiveCounter::SignalDesc desc = ptr.p->m_signalDesc;
      theData[desc.m_senderDataOffset] = ptr.p->m_senderData;
      theData[desc.m_errorCodeOffset] = desc.m_nodeFailErrorCode;
      Uint32 len = MAX(MAX(desc.m_senderDataOffset, desc.m_errorCodeOffset),
		       desc.m_senderRefOffset);
      
      NodeBitmask overlapping = ptr.p->m_nodes;
      overlapping.bitAND(nodes);
      Uint32 i = 0;
      while((i = overlapping.find(i)) != NodeBitmask::NotFound){
	theData[desc.m_senderRefOffset] = numberToRef(desc.m_block, i);
	m_block.sendSignal(m_block.reference(), desc.m_gsn, signal, len+1,JBB);
	i++;
      }
    }
  }
}

BlockReference
SafeCounterManager::reference() const {
  return m_block.reference();
}

void
SafeCounterManager::progError(int line, int err_code, const char* extra){
  m_block.progError(line, err_code, extra);
}

bool
SafeCounterHandle::clearWaitingFor(SafeCounterManager& mgr, Uint32 nodeId)
{
  SafeCounterManager::ActiveCounterPtr ptr;
  mgr.getPtr(ptr, m_activeCounterPtrI);
  ptr.p->m_nodes.clear(nodeId);
  
  if (ptr.p->m_nodes.isclear()){
    mgr.release(ptr);
    m_activeCounterPtrI = RNIL;
    return true;
  }
  return false;
}

SafeCounter::~SafeCounter(){
  bool clear = m_count == 0;
  bool isnull = m_ptr.i == RNIL;

  m_activeCounterPtrI = m_ptr.i;

  if(clear && isnull)
    return;

  if(clear && !isnull){
    m_mgr.release(m_ptr);
    m_activeCounterPtrI = RNIL;
    return;
  }

  /**
   * !clear && !isnull
   */
  if(!isnull){
    m_ptr.p->m_nodes = m_nodes;
    return;
  }

  ErrorReporter::handleAssert("~SafeCounter:: wo/ init", __FILE__, __LINE__);
}