GaBro.cc

Go to the documentation of this file.

#include "GaBro.h"
#include <iostream>

GaBro::GaBro()
{
  // Common
  _data_size = -1;
  _data_elt_nb = -1;
  _my_rank = 0;
  _server_nb_nodes = 1;

  // Client side
  _client_buffer = NULL;
  _pos = NULL;
  _elt_node_nb = -1;
  _node_pos = -1;

  this->setComId(-1);

  // Server side
  _server_buffer = NULL;
  _nb_rcp = NULL;
  _current_rcp = -1;
  _total_data_nb = -1;
  _node_data_nb = -1;
  _current_data_size = -1;
  _current_pos = -1;
}

GaBro::~GaBro() 
{
  delete _pos;
  delete _nb_rcp;
  for(info_list_map_t::iterator it=_info_list_map.begin(); it!=_info_list_map.end(); ++it)
    {
      info_t* info = it->second;
      delete info;
      _info_list_map.erase(it);
    }
  _info_list_map.clear();
}

void
GaBro::setSourceTopology(PaCO::PacoTopology_t topo)
{
  _srcTopo = topo;
  _nb_rcp = new long[_srcTopo.total];
  for (unsigned int i=0; i<_srcTopo.total; i++)
    {
      _nb_rcp[i] = -1;
    }
}

PaCO::PacoTopology_t
GaBro::getSourceTopology()
{
  return _srcTopo;
}

void
GaBro::setDestTopology(PaCO::PacoTopology_t topo)
{
  _destTopo = topo;
  _server_nb_nodes = topo.total;
  //  _cur_info->_server_nb_nodes = _server_nb_nodes;
  _pos = new int[_server_nb_nodes];
  for (int i=0; i<_server_nb_nodes; i++)
    {
      _pos[i] = 0;
    }
}

PaCO::PacoTopology_t
GaBro::getDestTopology()
{
  return _destTopo;
}

void
GaBro::setNodeRank(long Rank)
{
  _my_rank = Rank;
  _descr.sender = Rank;
}

long
GaBro::getNodeRank()
{
  return _my_rank;
}

void
GaBro::setEltSize(long size)
{
  _data_size = size;
  //  _cur_info->_data_size = _data_size;
}

void
GaBro::setTotalNbElt(long elt_nb)
{
  _data_elt_nb = elt_nb;
  _descr.total_data_nb = _data_elt_nb; 
}

void 
GaBro::setNodeNbElt(long elt_node_nb)
{
  _elt_node_nb = elt_node_nb;
  //  _cur_info->_elt_node_nb = _elt_node_nb;
  _descr.node_data_nb = _elt_node_nb;
}
  
void 
GaBro::setNodePos(long node_pos)
{
  _node_pos = node_pos;
  _descr.node_position = _node_pos;
}

PieceToSend* 
GaBro::computePiecesToSend(unsigned& size)
{
  size = _server_nb_nodes;
  // Compute total data length
  long total_length = _data_size * _elt_node_nb;
  // We send at each server node the same data
  PieceToSend * local_vector = new PieceToSend[_server_nb_nodes];
  for (int i = 0; i < _server_nb_nodes; i++)
    {
      local_vector[i].sourceNode = _my_rank;
      local_vector[i].destNode = i;
      local_vector[i].size = total_length;
    }
  return local_vector;
}

void
GaBro::setDataPtr(void* dataPtr)
{
  _client_buffer = (char *) dataPtr;
}

void *
GaBro::getClientData(void* pid, int server_node, long & size, long & length, bool & end)
{
  void * result = NULL;
  end = true;
  //std:cerr << _data_size << " " << _data_elt_nb << " " << _pos[server_node] << endl;
  long rst_data = (_data_size * _elt_node_nb) - _pos[server_node];
  //std:cerr << "Size : " << size << endl;
  //std:cerr << "Rst_data : " << rst_data << endl;
  if (size == rst_data)
    {
      length = size / _data_size;
      _descr.current_data_size = size;
      result = (void *)&_client_buffer[_pos[server_node]];
      _pos[server_node] = _pos[server_node] + size;
      size = 0;
    }
  else if (size > rst_data)
    {
      length = rst_data / _data_size;
      _descr.current_data_size = rst_data;
      result = (void *)&_client_buffer[_pos[server_node]];
      _pos[server_node] = _pos[server_node] + rst_data;
      size = size - rst_data;
    }
  else if (size < rst_data)
    {
      length = size / _data_size;
      _descr.current_data_size = size;
      result = (void *)&_client_buffer[_pos[server_node]];
      _pos[server_node] = _pos[server_node] + size;
      size = 0;
    }
  return result;
}

void
GaBro::clientFree()
{
  for (int i=0; i<_server_nb_nodes; i++)
    {
      _pos[i] = 0;
    }
}

void *
GaBro::getDescr()
{
  return &_descr;
}

bool
GaBro::insertData(void* rcptBuf, unsigned long element_nb)
{

  bool ret = true;
  if (element_nb != 0)
    {
  if (_server_buffer == NULL)
    {
      _server_buffer = (char *)malloc(_data_size * _total_data_nb);
    }
  // Copy the buffer !!!
//   cerr << "current_pos : " << _current_pos << endl;
//   cerr << "data_size : " << _data_size << endl;
//   cerr << "nb_rcp[_current_rcp] " << _nb_rcp[_current_rcp] << endl;  

  long current_sender_pos = (_current_pos * _data_size) + ((_data_size * _node_data_nb) - _nb_rcp[_current_rcp]);
//   cerr << "current_sender_pos : " << current_sender_pos << endl;
  memcpy((void *)&(_server_buffer[current_sender_pos]), rcptBuf, _current_data_size);
  _nb_rcp[_current_rcp] = _nb_rcp[_current_rcp] - _current_data_size;
    }
  for (unsigned int i=0; i<_srcTopo.total; i++)
    {
      if (_nb_rcp[i] != 0)
   {
     ret = false;
   }
    }
  return ret;
}

void *
GaBro::getServerData(long & length)
{
  length = _total_data_nb;
  return _server_buffer;
}

void
GaBro::setDescr(void * descr)
{
  _descr = *((GaBroLib::GaBroDescr*) descr);
  if (_current_rcp == -1)
    {
      // First piece reception !
      _total_data_nb = _descr.total_data_nb;
    }
  _current_rcp = _descr.sender;
  _current_pos = _descr.node_position;
  _node_data_nb = _descr.node_data_nb;
  _current_data_size = _descr.current_data_size;
  //std:cerr << _current_data_size << endl;
  if (_nb_rcp[_current_rcp] == -1)
    {
      _nb_rcp[_current_rcp] = (_data_size * _node_data_nb);
    }
}

void
GaBro::serverFree() 
{ 
  _current_rcp = -1;
  _current_pos = -1;
  _server_buffer = NULL;
  for (unsigned int i=0; i<_srcTopo.total; i++)
    {
      _nb_rcp[i] = -1;
    }
}

bool
GaBro::setComId(long id)
{ 
  info_t * _infolists;
  info_list_map_t::iterator it = _info_list_map.find(id);
  if (it == _info_list_map.end()) 
    {
      _infolists = new info_t();
      _infolists->_data_size = _data_size;
      _infolists->_elt_node_nb = _elt_node_nb;
      _infolists->_server_nb_nodes = _server_nb_nodes;
      _infolists->_data_elt_nb = _data_elt_nb;
      _infolists->_node_pos = _node_pos;
      _info_list_map[id] = _infolists;
    }
  else 
    {
      _infolists   = it->second;
    }
  _cur_info = _infolists;
  _cur_id = id;
  _data_size = _infolists->_data_size;
  _elt_node_nb = _infolists->_elt_node_nb;
  _data_elt_nb = _infolists->_data_elt_nb;
  _server_nb_nodes = _infolists->_server_nb_nodes;
  _node_pos = _infolists->_node_pos;

  _descr.node_position = _node_pos;
  _descr.total_data_nb = _data_elt_nb; 
  _descr.node_data_nb = _elt_node_nb;

  //std:cerr << _data_size << " " << _elt_node_nb << " " << _server_nb_nodes <<endl;
  return true;
}

bool
GaBro::freeComId(long id)
{ 
  if (id == -1) {
    //    cerr << "freeComId: not allowed to free com id -1 (ignoring)\n";
    return true;
  }
  if (_cur_id == id) {
    //    cerr << "freeComId: not allowed to free current com (ignoring)\n";
    return true;
  }
  //  cerr << "freeComid: communication id: "<<id<<endl;
  info_list_map_t::iterator it = _info_list_map.find(id);
  if (it != _info_list_map.end() ) 
    {
      info_t* info = it->second;
      delete info;
      _info_list_map.erase(it);
    }
  return true;
}