PaCO++_operation.cc

Go to the documentation of this file.

#include "PaCO++_operation.h"
#include "PaCO++_src.h"

#include "com/paco_com.h"
#include "thread/paco_thread.h"
#include "DistributionLibrairie.h"
#include "CommunicationScheduling.h"

#include <assert.h>

#undef DEBUG_INTERNAL

PaCO_operation::PaCO_operation(int _number_arguments)
{
  // Default case : sequential client
  _current_id = -1;
  myRank = 0;
  mytopo.total = 1;
  _mode = PaCO::ServerSide;
  typeClient = false;
  termine = false;
  termine2 = false;
  client_id = -1;
  number_arguments = _number_arguments;
  libraries_in = new DistributionLibrary*[number_arguments];
  libraries_out = new DistributionLibrary*[number_arguments];
#ifdef PACO_WARN_DISTLIB
  libraries_in_name.resize(number_arguments);
  libraries_out_name.resize(number_arguments);
#endif
  for(int i=0; i<number_arguments; i++)
    {
      libraries_in[i] = NULL;
      libraries_out[i] = NULL;
#ifdef PACO_WARN_DISTLIB
      libraries_in_name[i]=NULL;
      libraries_out_name[i]=NULL;
#endif
    }
  FabManager=paco_getFabriqueManager();
  FabThread=0;
  FabCom=0;
  FabComScheduling=0;
  my_comScheduling=0;
  return_op = 0;

  // Exception handling
  _exception = false;
  _level = 0;

  // Timing
#ifdef PADICO
  _ticks=(padico_timing_t*)malloc(sizeof(padico_timing_t)*10);
#endif

}

PaCO_operation::~PaCO_operation()
{
#ifdef PACO_WARN_DISTLIB
  for(int i=0; i<number_arguments; i++)
    {
      if (libraries_in_name[i])
   free(libraries_in_name[i]);
      if (libraries_out_name[i])
   free(libraries_out_name[i]);
    }
#endif
  delete libraries_in;
  delete libraries_out;
  delete my_mutex;
  delete my_condition;
  delete my_mutex2;
  delete my_condition2;
  delete my_mutex3;
  // Memories must be deleted
}

void 
PaCO_operation::setClientId(CORBA::Short request_uid)
{
  client_id = request_uid;
}

void 
PaCO_operation::init_context(PaCO_operation * op)
{
  myRank = op->myRank ;
  mytopo = op->mytopo ;
  number_stubs = op->number_stubs ;

  FabManager = op->FabManager;
 
  group = op->group;
  my_com = op->my_com ;
  FabCom = op->FabCom ;

  my_comScheduling = op->my_comScheduling ;
  FabComScheduling = op->FabComScheduling ;

  // Client part
  typeClient = op->typeClient ; 
  if (op->serveur_topo_aller.total > 0)
    {
      setServerTopo(op->serveur_topo_aller,"in");
    }
  if (op->serveur_topo_retour.total > 0)
    {
      setServerTopo(op->serveur_topo_retour,"out");
    }

  // Server part
  FabThread = op->FabThread ;
  my_mutex = op->my_mutex ;
  my_condition = op->my_condition ;
  my_mutex2 = op->my_mutex2 ;
  my_condition2 = op->my_condition2 ;
  my_mutex3 = op->my_mutex3 ;
}

void 
PaCO_operation::init_return_context(PaCO_operation * op)
{
  myRank = op->myRank ;
  mytopo = op->mytopo ;
  number_stubs = op->number_stubs ;

  FabManager = op->FabManager;

  group = op->group ;
  my_com = op->my_com ;
  FabCom = op->FabCom ;

  my_comScheduling = op->my_comScheduling ;
  FabComScheduling = op->FabComScheduling ;

  // Client part
  typeClient = op->typeClient ; 
  if (op->serveur_topo_aller.total > 0)
    {
      setServerTopo(op->serveur_topo_aller,"in");
    }
  // Server part
  FabThread = op->FabThread ;
  my_mutex = op->my_mutex ;
  my_condition = op->my_condition ;
  my_mutex2 = op->my_mutex2 ;
  my_condition2 = op->my_condition2 ;
  my_mutex3 = op->my_mutex3 ;
}

void 
PaCO_operation::init_context_proxy(PaCO_operation * op)
{
  number_arguments = op->number_arguments;
  libraries_in = op->libraries_in;
  libraries_out = op->libraries_out;

  for(int i=0; i<number_arguments; i++)
    {
      if (libraries_in[i])
        {
     libraries_in[i]->setCommunicator(group);
        }
      if (libraries_out[i])
        {
     libraries_out[i]->setCommunicator(group);
        }
    }
}

void 
PaCO_operation::init_context_args(PaCO_operation * op)
{
  return_op = 1;
  number_arguments = op->number_arguments;
  _mode = op->_mode;
  libraries_in = op->libraries_out;
  libraries_out = op->libraries_in;
  
  for(int i=0; i<number_arguments; i++)
    {
      if (libraries_in[i])
        {
     libraries_in[i]->setCommunicator(group);
        }
      if (libraries_out[i])
        {
     libraries_out[i]->setCommunicator(group);
        }
    }
}

void
PaCO_operation::setFabManager(paco_fabrique_manager* fm)
{
    FabManager = fm;
}

void
PaCO_operation::setComFab(paco_fabrique_com* fc)
{
    FabCom = fc;
}

void
PaCO_operation::setLibCom(const string& LibCom, void * _group)
{
  if (FabCom==0)
    {
      if (FabManager)
   FabCom = FabManager->get_com(LibCom);
      else
   {
     cerr << "Error: Unable to instantiate lib com "<<LibCom<<endl;
     abort();
   }      
    }

    group = _group;
    my_com = FabCom->paco_create(group);

    for(int i=0; i<number_arguments; i++)
      {
        if (libraries_in[i])
        {
            libraries_in[i]->setCommunicator(group);
        }
        if (libraries_out[i])
        {
            libraries_out[i]->setCommunicator(group);
        }
      }
}

// void
// PaCO_operation::setMode(PaCO::distLoc_t mode, int param, const string& way) throw(BadWayString)
// {
//   //_mode = mode;
//   if (way == "inout")
//     {
//       libraries_in[param]->setMode(mode);
//       libraries_out[param]->setMode(mode);
//     }
//   else if (way == "in")
//     {
//       libraries_in[param]->setMode(mode);
//     }
//   else if (way == "out")
//     {
//       libraries_out[param]->setMode(mode);
//     }
//   else
//     {
//       // Bad way string
//       throw BadWayString(way);
//     }
// }

void
PaCO_operation::setDisLibArg(int arg_number, const string& lib_name, const string& way) throw(BadWayString)
{
  if (way == "inout")
    {
      libraries_in [arg_number] = FabManager->get_distribution(lib_name)->create();
      libraries_out[arg_number] = FabManager->get_distribution(lib_name)->create();
#ifdef PACO_WARN_DISTLIB
      if (libraries_in_name[arg_number]) free(libraries_in_name[arg_number]);
      if (libraries_out_name[arg_number]) free(libraries_out_name[arg_number]);
      libraries_in_name [arg_number]  = strdup(lib_name.c_str());
      libraries_out_name [arg_number] = strdup(lib_name.c_str());
#endif
    }
  else if (way == "in")
    {
      libraries_in[arg_number] = FabManager->get_distribution(lib_name)->create();
#ifdef PACO_WARN_DISTLIB
      if (libraries_in_name[arg_number]) free(libraries_in_name[arg_number]);
      libraries_in_name [arg_number]  = strdup(lib_name.c_str());
#endif
    }
  else if (way == "out")
    {
      libraries_out[arg_number] = FabManager->get_distribution(lib_name)->create();
#ifdef PACO_WARN_DISTLIB
      if (libraries_out_name[arg_number]) free(libraries_out_name[arg_number]);
      libraries_out_name [arg_number] = strdup(lib_name.c_str());
#endif
    }
  else
    {
      // Bad way string
      throw BadWayString(way);
    }
}
// void
// PaCO_operation::setTypeArg(int arg_number, Pattron * temp, const string& lib_name, const string& way) throw(BadWayString)
// {
//   Fabrique * f = new Fabrique();
//   f->enregistrer(temp);
//   if (way == "inout")
//     {
//       libraries_in[arg_number] = FabDist->paco_create(lib_name,f);
//       libraries_out[arg_number] = FabDist->paco_create(lib_name,f);
//     }
//   else if (way == "in")
//     {
//       libraries_in[arg_number] = FabDist->paco_create(lib_name,f);
//     }
//   else if (way == "out")
//     {
//       libraries_out[arg_number] = FabDist->paco_create(lib_name,f);
//     }
//   else
//     {
//       // Bad way string
//       throw BadWayString(way);
//     }
// }

// void
// PaCO_operation::initArg(PaCO::PacoInitData_t data, int nb_param, const string& way) throw(BadWayString)
// {
//    if (way == "in")
//     {
//       libraries_in[nb_param]->setGlobalDataConfiguration(data.gd);
//       libraries_in[nb_param]->setLocalDataConfiguration (data.ld);
//     }
//    else if (way == "out")
//     {
//       libraries_out[nb_param]->setGlobalDataConfiguration(data.gd);
//       libraries_out[nb_param]->setLocalDataConfiguration (data.ld);
//     }
//    else
//     {
//       // Bad way string
//       throw BadWayString(way);
//     }
// }

void
PaCO_operation::init(int rank, int total)
{
    myRank = rank;
    mytopo.total = total;
}

// int 
// PaCO_operation::setDistribConfig(void * config, int number, const string& way) throw(BadWayString, BadWaySetStringDistribConfig)
// {
//   if (way == "inout")
//     {
//       int ret_in = libraries_in[number]->setConfig(config);
//       int ret_out = libraries_out[number]->setConfig(config);
//       if (ret_in == ret_out)
//    {
//      return ret_in;
//    }
//       else
//    {
//      throw BadWaySetStringDistribConfig(way, ret_in, ret_out);
//    }
//     }
//   else if (way == "in")
//     {
//       return libraries_in[number]->setConfig(config);
//     }
//   else if (way == "out")
//     {
//       return libraries_out[number]->setConfig(config);
//     }
//   else
//     {
//       // Bad way string
//       throw BadWayString(way);
//     }
// }

// void * 
// PaCO_operation::getDistribConfig(int number, const string& way)  throw(BadWayString)
// {
//   if (way == "in")
//     {
//       return libraries_in[number]->getConfig();
//     }
//   else if (way == "out")
//     {
//       return libraries_out[number]->getConfig();
//     }
//   else
//     {
//       // Bad way string
//       throw BadWayString(way);
//     }
// }

void
PaCO_operation::setTypeClient(bool type)
{
    typeClient = type;
}


void 
PaCO_operation::setClientTopo(PaCO::PacoTopology_t _client_topo, const string& way) throw(BadWayString) 
{
   if (way == "in")
     {
       serveur_topo_retour = _client_topo;
     }
   else if (way == "out")
     {
       serveur_topo_aller = _client_topo;
     }
   else
     {
       // Bad way string
       throw BadWayString(way);
     }
}

void 
PaCO_operation::setClientTopo(PaCO::PacoTopology_t _client_topo)
{
  serveur_topo_retour =  _client_topo;
}

void 
PaCO_operation::setServerTopo(PaCO::PacoTopology_t _serveur_topo, const string& way) throw(BadWayString) 
{
   if (way == "in")
     {
       serveur_topo_aller = _serveur_topo;
     }
   else if (way == "out")
     {
       serveur_topo_retour = _serveur_topo;
     }
   else
     {
       // Bad way string
       throw BadWayString(way);
     }
}

void 
PaCO_operation::configureTopo()
  {
    // DO NOT FORGET TO CLEAR MEMORIES (WHAT ABOUT COMMUNICATION MATRICES??) !!!!
    clearAllScheduleMemory();
    
    if (return_op == 0)
      {  
   for (int i = 0; i < number_arguments; i++)
     {
       if (libraries_in[i] != NULL)
         {
      libraries_in[i]->setDestTopology(serveur_topo_aller);
      libraries_in[i]->setSourceTopology(serveur_topo_retour);
      libraries_in[i]->setNodeRank(myRank);
         }
       if (libraries_out[i] != NULL)
         {
      libraries_out[i]->setDestTopology(serveur_topo_retour);
      libraries_out[i]->setSourceTopology(serveur_topo_aller);
      libraries_out[i]->setNodeRank(myRank);
         }
     }
      }
  }

void
PaCO_operation::setThreadFab(paco_fabrique_thread* ft)
{
    FabThread = ft;
}

void
PaCO_operation::setLibThread(const string& LibThread)
{
  if (FabThread==0)
    {
      if (FabManager)
   FabThread = FabManager->get_thread(LibThread);
      else
   {
     cerr << "Error: Unable to instantiate lib thread "<<LibThread<<endl;
     abort();
   }
   
    }
    
  my_mutex = FabThread->paco_create_mutex();
  my_condition = FabThread->paco_create_condition(my_mutex);
  my_mutex2 = FabThread->paco_create_mutex();
  my_condition2 = FabThread->paco_create_condition(my_mutex2);
  my_mutex3 = FabThread->paco_create_mutex();
}

void
PaCO_operation::setComSchedulingFab(paco_fabrique_comScheduling* fcs)
{
    FabComScheduling = fcs;
}

void
PaCO_operation::setLibComScheduling(const string& LibComScheduling)
{
  if (FabComScheduling==0)
    {
      if (FabManager)
   FabComScheduling = FabManager->get_comScheduling(LibComScheduling);
      else
   {
     cerr << "Error: Unable to instantiate lib communication schedule "<<LibComScheduling<<endl;
     abort();
   }
   
    }
    
  if (my_comScheduling)
    delete my_comScheduling;
  my_comScheduling = FabComScheduling->create_comScheduling();
}

PaCO_operation::_com_info_t*
PaCO_operation::getComMemoryId(long id)
{
  // lookup existing communication info
  _com_info_map_t::iterator it2 = _com_info_map.find(id);
  if (it2 != _com_info_map.end()) 
    {
#ifdef DEBUG_INTERNAL
      std::cerr <<"[INFO] found communication info for Id "<<id<<endl;
#endif
      return  _com_info_map[id];
    }
  else
    {
      return NULL;
    }
}

void
PaCO_operation::noCommunicationMemory()
{
  _current_id = -1; // _DO_NOT_USE_COMMUNICATION_MEMORY;

#ifdef DEBUG_INTERNAL
  std::cerr <<"[INFO] not using memory"<<endl;
#endif

}

void
PaCO_operation::useCommunicationMemoryId(long id) throw (InvalidArgument)
{
  if (id<0)
    throw new InvalidArgument("useCommunicationMemeoryId expect an id >0");

#ifdef DEBUG_INTERNAL
  std::cerr <<"[INFO] using memory Id "<<id<<endl;
#endif

  // set current id
  _current_id = id;

  // Beta patch
  for(int i=0; i<number_arguments; i++)
    {
      if (libraries_in[i])
        {
     bool res=libraries_in[i]->setComId(id);
     if (!res) {
#ifdef PACO_WARN_DISTLIB
       std::cerr << "warning: argument #"<<i<<" with dist lib "<<libraries_in_name[i]<<" does not support comm cache\n";
#else
       std::cerr << "warning: argument #"<<i<<" does not support comm cache\n";
#endif
     }
        }
    }
}

void 
PaCO_operation::clearCommunicationMemoryId(long id) throw (InvalidArgument)
{
  if (id<0)
    throw new InvalidArgument("clearCommunicationMemeoryId expects an id >0");
  
#ifdef DEBUG_INTERNAL
  std::cerr <<"[INFO] clearing communication memory Id "<<id<<endl;
#endif
  
  _com_info_map_t::iterator it2 = _com_info_map.find(id);
  if (it2 != _com_info_map.end())
    {
      delete _com_info_map[id];
      _com_info_map.erase(it2);
    }

  assert(my_comScheduling!=0);
  my_comScheduling->clearScheduleId(id);
}

void
PaCO_operation::clearScheduleMemoryId(long id) throw (InvalidArgument)
{
  assert(my_comScheduling!=0);
  my_comScheduling->clearScheduleId(id);
}

void
PaCO_operation::clearAllScheduleMemory() 
{
  if (my_comScheduling)
    my_comScheduling->clearAllSchedules();
}