Controlled.cc

Go to the documentation of this file.

#include "Controlled.h"
#include <stdio.h>
#include <iostream>

#define SRC_TOPOLOGY 1
#define DST_TOPOLOGY 2
#define ELEMENT_SIZE 4
#define LOCAL_SIZE   8

#define CORRECT_CONFIG 15

#define DEBUG_INTERNAL
#define DEBUG_INTERNAL2


Controlled::Controlled()
{
#ifdef DEBUG_INTERNAL
  std::cerr<<"[new] Controlled"<<std::endl;
#endif  

  // Common
  _config = 0;

  // By default 1->1 :)
  _sTopo.total=1;
  _dTopo.total=1;
  _nodeRank=0;
  
  // Client side
  _clientBuffer=NULL;
  _cm = new CommMatrix(_sTopo.total,_dTopo.total);
  _maxllen=0;
  _remaining_cm = NULL;
  _clientDescr.lens.length(0);

  // Server side
  _serverDescr=NULL;
  _serverBuffer=NULL;

#ifdef DEBUG_INTERNAL
  std::cerr<<"[new] Controlled done"<<std::endl;
#endif  

}

Controlled::~Controlled() 
{
#ifdef DEBUG_INTERNAL
  std::cerr<<"[del] Controlled"<<std::endl;
#endif  
  delete _cm;
}


void
Controlled::_setSourceTopology(unsigned long total)
{
  PaCO::PacoTopology_t topo;
  topo.total = total;
  setSourceTopology(topo);
}

void
Controlled::setSourceTopology(PaCO::PacoTopology_t topo)
{
#ifdef DEBUG_INTERNAL
  std::cerr << "-- setSourceTopology: " << topo.total << std::endl;
#endif
  _config |= SRC_TOPOLOGY;

  if (_sTopo.total != topo.total) {
    _sTopo = topo;
    delete _cm;
    _cm = new CommMatrix(_sTopo.total,_dTopo.total);

  }
}

PaCO::PacoTopology_t
Controlled::getSourceTopology()
{
  std::cerr << "****** Why is the method " << __FUNCTION__ << "called?\n";
  abort();
  return _sTopo;
}


void
Controlled::_setDestTopology(unsigned long total)
{
  PaCO::PacoTopology_t topo;
  topo.total = total;
  setDestTopology(topo);
}

void
Controlled::setDestTopology(PaCO::PacoTopology_t topo)
{
#ifdef DEBUG_INTERNAL
  std::cerr << "-- setDestTopology: " << topo.total << std::endl;
#endif
  _config |= DST_TOPOLOGY;

  if (_dTopo.total != topo.total) {
    _dTopo = topo;
    delete _cm;
    _cm = new CommMatrix(_sTopo.total,_dTopo.total);
  }

}

PaCO::PacoTopology_t
Controlled::getDestTopology()
{
  std::cerr << "****** Why is the method " << __FUNCTION__ << "called?\n";
  abort();
  return _dTopo;
}

void
Controlled::setNodeRank(long Rank)
{
#ifdef DEBUG_INTERNAL
  std::cerr << "-- setNodeRank: " << Rank << std::endl;
#endif
  _nodeRank = Rank;
}

long
Controlled::getNodeRank()
{
  return _nodeRank;
}

void
Controlled::setEltSize(unsigned long size)
{
#ifdef DEBUG_INTERNAL
  std::cerr << "-- setEltSize: " << size << std::endl;
#endif
  _config |= ELEMENT_SIZE;
  _unitsize = size;
}

void
Controlled::_setLocalLen(unsigned long elt_nb)
{
#ifdef DEBUG_INTERNAL
  std::cerr << "-- setLotalNbElt: " << elt_nb << std::endl;
#endif
  _llen = elt_nb;
}

void
Controlled::setMaxLocalLen(unsigned long elt_nb)
{
#ifdef DEBUG_INTERNAL
  std::cerr << "-- setMaxLocalNbElt: " << elt_nb << std::endl;
#endif
  _config |= LOCAL_SIZE;
  _maxllen = elt_nb;
}

long
Controlled::getMaxLen()
{
  return _maxllen;
}

void
Controlled::readFromFile(char *filename)
{
#ifdef DEBUG_INTERNAL
  std::cerr << "readFromFile-------------------- done\n";
#endif

  //open comms file
  FILE *comms_file = fopen(filename, "r");
  if (comms_file == NULL) {
    perror("unable to open comms file");
    throw -1;
  }

  unsigned int senders, receivers;
  fscanf(comms_file, "%dx%d\n", &senders, &receivers);

  std::cerr<<" #s: "<<senders<<"  #r:"<<receivers<<std::endl;

  _setSourceTopology(senders);
  _setDestTopology(receivers);
   
  unsigned maxs[senders];
  for(unsigned int i = 0 ; i < senders ; i++)
    maxs[i]=0;

  //read edges in file, and add edges
  for(unsigned int i = 0 ; i < senders * receivers ; i++) {
    unsigned int value;
    fscanf(comms_file, "%d", &value); 
    unsigned int node2 = i % receivers;
    unsigned int node1 = (i - node2) / receivers;
    if (value != 0) {
      _cm->set(node1, node2, value);
#ifdef DEBUG_INTERNAL
      std::cerr << " "<<node1<<" -> "<<node2<<" : "<<value<<endl;
#endif

      maxs[node1]+=value;
    }
  }  
    
  fclose(comms_file);
  
  unsigned long max=0;
  for(unsigned int i = 0 ; i < senders ; i++)
    if (maxs[i]>max)
      max=maxs[i];

  if (_maxllen && (max>_maxllen))
    {
      std::cerr << "MaxLocalLen was set to small: "<<_maxllen<<" vs "<<max<<" (0: not set)"<<std::endl;
      abort();
    }

  setMaxLocalLen(max);

  _cm->dump();

}  

PieceToSend* 
Controlled::computePiecesToSend(unsigned& size_out)
{

  PieceToSend * sched;

#ifdef DEBUG_INTERNAL
  std::cerr << "computePiecesToSend-------------------- in\n";
  std::cerr << " config = "<<_config<<endl;
  std::cerr << " #sender: "<<_cm->getSenderSize()<<std::endl;
  std::cerr << " #recv: "<<_cm->getReceiverSize()<<std::endl;  
#endif

  if (_config != CORRECT_CONFIG) {
    std::cerr << "Controlled: incorrect configuration state "<< _config<<"/"<<CORRECT_CONFIG<<endl;
    abort();
  }

  sched = new PieceToSend[_cm->getSenderSize()*_cm->getReceiverSize()];

  unsigned count=0;
  for(unsigned j=0;j<_cm->getReceiverSize();j++) {  
    if (_cm->get(_nodeRank,j)>0) {
#ifdef DEBUG_INTERNAL
      std::cerr << " "<<_nodeRank<<" -> "<<j<<" : "<<_cm->get(_nodeRank,j)<<std::endl;
#endif
      sched[count].sourceNode = _nodeRank; 
      sched[count].destNode   = j;
      sched[count].size       = _cm->get(_nodeRank,j)*_unitsize;
      sched[count].id         = (void*) -1; /* not used */
      count++;
    }
  }
  
  size_out = count;
  
#ifdef DEBUG_INTERNAL
  std::cerr << "computePiecesToSend-------------------- done\n";
#endif

  return sched;
}

void
Controlled::setDataPtr(void* dataPtr)
{
  _clientBuffer = (char *) dataPtr;
#ifdef DEBUG_INTERNAL
  cerr << "-- _clientBuffer set to "<<dataPtr<<endl;
#endif
}

void *
Controlled::getClientData(void *pid, int dnode, long & remaining_size_octet, long & returned_length_element, bool & end)  
{
  if (!_remaining_cm)
    _remaining_cm = new CommMatrix(_cm);

  long rgsz = _remaining_cm->get(_nodeRank, dnode)*_unitsize;
  unsigned rglen;

  if (rgsz > remaining_size_octet) {
    rglen = remaining_size_octet/_unitsize;
    end=false;
  } else {
    rglen = _remaining_cm->get(_nodeRank, dnode);
    end=true;
  }    

  if (rglen>0) {

    returned_length_element=rglen;
    remaining_size_octet-=(returned_length_element*_unitsize);
    _remaining_cm->sub(_nodeRank, dnode, rglen);
  
    unsigned descr_len  = _clientDescr.lens.length();
    _clientDescr.lens.length(descr_len+1);
    _clientDescr.lens[descr_len]=rglen;

    unsigned total=0;
    for(unsigned i=0;i<_cm->getSenderSize();i++) {
      total+=_cm->get(i, dnode);
    }
    _clientDescr.llen = total;

  }
#ifdef DEBUG_INTERNAL
  std::cerr << __FUNCTION__ << ": getClientData returns #elt: "<<rglen<<endl;
#endif

  //  int dummy;
  //  fscanf(stdin, "%d", &dummy);  


  return _clientBuffer;
}

void
Controlled::clientFree() // always called at the end of an invocation (reset some variables)
{

#ifdef DEBUG_INTERNAL
  std::cerr << "-- clientFree\n";
#endif
  delete _remaining_cm;
  _remaining_cm=NULL;
}

void *
Controlled::getDescr()
{
#ifdef DEBUG_INTERNAL
  std::cerr << "-- descr:llen: "<<_clientDescr.llen<<std::endl;
#endif

  _clientDescr.lens.length(0);

  return &_clientDescr;
}

bool
Controlled::insertData(void* rcptBuf, unsigned long element_nb)
{
  // cp from recpBuf to _serverBuffer
#ifdef DEBUG_INTERNAL
  std::cerr << "-- insertData:"<<element_nb<<std::endl;
#endif
  
  unsigned long llen = element_nb*_unitsize;
  _serverToReceived -= llen;

  memcpy(_serverBuffer+_serverToReceived, rcptBuf, llen);

#ifdef DEBUG_INTERNAL
  std::cerr << "-- insertData: "<<_serverToReceived<<" octects"<<std::endl;
#endif
  

  return (_serverToReceived==0);
}

void *
Controlled::getServerData(long & length) // in element !
{
  length= _llen;
#ifdef DEBUG_INTERNAL
  std::cerr << "-- getServerData: ptr : "<<(void*)_serverBuffer<<" #element:"<<length<<std::endl;
#endif  
  return (void*) _serverBuffer;
}

void
Controlled::setDescr(void * descr)
{
  std::cerr << "-- setDescr "<<endl;

  _serverDescr = (ControlledLib::ControlledDescr*) descr;

  // Descr already received

  if ( _serverBuffer == NULL)
    {
      // 1st time -> allocate memory & init
      //this->setEltSize(_descr->usz); BUG: usz can be different in client and server (32bit vs 64bit)
      unsigned long len = _serverDescr->llen*_unitsize;
      this->_setLocalLen(_serverDescr->llen);
      
#ifdef DEBUG_INTERNAL
      fprintf(stderr, "-- setDescr: stopo: %ld\tdtopo: %ld\n",_sTopo.total, _dTopo.total);
      fprintf(stderr, "   setDescr: local len: %ld\n", _serverDescr->llen);
#endif  
      _serverBuffer = (char*) malloc(len*_unitsize);
#ifdef DEBUG_INTERNAL
      std::cerr << "-- serverMalloc: "<<(void*)_serverBuffer<<" - "<<_llen<<endl;
#endif
      _serverToReceived = len;
    }
}

void
Controlled::serverFree() 
{ 
#ifdef DEBUG_INTERNAL
  std::cerr << "-- serverFree: "<<(void*)_serverBuffer<<endl;
#endif
  free(_serverBuffer);
  _serverBuffer=NULL; // the orb (or the user) need to free the memory!
}