PaCO++_src.cc

Go to the documentation of this file.

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

PaCO::PacoTopology_t PacoTopologySeq = { 1 };

static ostream & 
operator<<(ostream & os, const CORBA::Exception & e)
{
  CORBA::Any tmp;
  tmp <<=e ;
  CORBA::TypeCode_var tc = tmp.type();
  const char * p = tc->name ();
  if (*p != '\0')
    os << p;
  else
    os << tc->id();
  return os;
}

void* paco_orb_run(void* arg)
{
  CORBA::ORB_ptr orb = (CORBA::ORB_ptr) arg;
  orb->run();
  return (void*) 0;
}

paco_fabrique_manager* paco_getFabriqueManager()
{
  static paco_fabrique_manager* pfm=NULL;

  if (!pfm)
    pfm = new paco_fabrique_manager();

  return pfm;
}


// class InterfaceManager_impl

InterfaceManager_impl::InterfaceManager_impl(CORBA::ORB_ptr current_orb)
{
  _orb = current_orb;
  _current_uid = 0;
  _totalNode = -1;
  _workNode = -1;
  _totalNodeReceived = 0;
  _rtn = 0; // normal
}

InterfaceManager_impl::~InterfaceManager_impl() {}

void
InterfaceManager_impl::setReturn(int rtn)
{
  _rtn = rtn;
}

void 
InterfaceManager_impl::setTotalNode(CORBA::Short totalNode)
{
  if (totalNode > 0)
    {
      _totalNode = totalNode;
      // Init de _informations_temp;
      _informations_temp.nodes.length(_totalNode);
      _informations_temp.serveur_topo.total = _totalNode;
      _informations.nodes.length(_totalNode);
      _informations.serveur_topo.total= _totalNode;
    }
  updateContexts();
}

void
InterfaceManager_impl::setTopo(const PaCO::PacoTopology_t& topo)
{
  serveur_topo = topo;
  setTotalNode(topo.total);
}

void 
InterfaceManager_impl::setWorkNode(CORBA::Short totalNode)
{
  _workNode = totalNode;
}

void 
InterfaceManager_impl::setNewNode(const char * node, const CORBA::Short rank)
{

try  
  {
    //CORBA::Object_var object = _orb->string_to_object(node);
    //PaCO::InterfaceParallel_var interface = PaCO::InterfaceParallel::_narrow(object);
    //  CORBA::Short rank = interface->getDeployRank();

  if (_rtn == 1)
    cerr << "Return Proxy : Adding node : " << rank << endl;
  else
    cerr << "Adding node : " << rank << endl;
  //  _informations_temp.nodes[_totalNodeReceived] = node;
  _informations_temp.nodes[rank] = node;
  _totalNodeReceived++;
  if (_totalNodeReceived == _totalNode)
    {
       if (_rtn == 1)
    cerr << "Return Proxy : All the node are registered" << endl;
       else
    cerr << "All the node are registered" << endl;

      _nodes = _informations_temp.nodes;
      _informations.nodes = _nodes;
      _informations.serveur_topo = _informations_temp.serveur_topo;
      if (_rtn == 1)
   {
     // connection for the server side
     this->finish();
   }
      _totalNodeReceived = 0;
      setConnectionInfos();
    }
  }
 catch (const CORBA::Exception & e)
    {
      cerr << " Node : " << node << endl;
      cerr << " InterfaceManager_impl::setNewNode : Exception : " << e << endl;
    }

}

void 
InterfaceManager_impl::setReturn(const char * ref)
{
  return_str = (char *) ref;
  // Now we have to give this reference to all the nodes
  CORBA::ULong i;
  CORBA::Object_var object;
  PaCO::InterfaceParallel_var interface;
  CORBA::ULong lgth;
  if (_workNode == -1) { lgth = _nodes.length();}
  else { lgth = _workNode;}
  for (i = 0; i<lgth; i++)
    {
      object = _orb->string_to_object(_nodes[i]);
      interface = PaCO::InterfaceParallel::_narrow(object);
      interface->refReturnObject(return_str.c_str());
    }
}

char *
InterfaceManager_impl::getReturn()
{
  //  cerr << " Return string : " << return_str << endl;
  char * p = new char[return_str.length()+1];
  return_str.copy(p,string::npos);
  p[return_str.length()]=0;
  //  return (char *) return_str.c_str();
  return p;
}

// CORBA::Short
// InterfaceManager_impl::getTotalNode()
// {
//   return _totalNode;
// }

void
InterfaceManager_impl::return_ok(const CORBA::Short client_id)
{
  CORBA::ULong i;
  CORBA::Object_var object;
  PaCO::InterfaceParallel_var interface;
  CORBA::ULong lgth;
  if (_workNode == -1) { lgth = _nodes.length();}
  else { lgth = _workNode;}
  cerr << "Adding a new client !!!!" << endl;
  //  cerr << "WARNING : in this version PaCO++ only plays with the last client !" << endl;
  for (i = 0; i<lgth; i++)
    {
      //cerr << "Init work node : " << i << endl;
      object = _orb->string_to_object(_nodes[i]);
      interface = PaCO::InterfaceParallel::_narrow(object);
      interface->connect_return_object(client_id);
    }
}

void
InterfaceManager_impl::setConnectionInfos()
{
  if (_rtn == 1)
    cerr << "Return Proxy : Deploy the parallel CORBA object" << endl;
  else 
    cerr << "Deploy the parallel CORBA object" << endl;

  // Initialize the nodes
  CORBA::Object_var object;
  PaCO::InterfaceParallel_var interface;
  if (_workNode == -1)
    {
      if (_rtn == 1)
   cerr << "Return Proxy : Deploying all the nodes" << endl;
      else
   cerr << "Deploying all the nodes" << endl;

      for (  CORBA::ULong i = 0; i<_nodes.length(); i++)
   {
     if (_rtn == 1)
       cerr << "Return Proxy : Deploying node : " << i << endl;
     else
       cerr << "Deploying node : " << i << endl;

     object = _orb->string_to_object(_nodes[i]);
     interface = PaCO::InterfaceParallel::_narrow(object);
     interface->init_InterfaceParallel(i,_nodes.length(), _informations.serveur_topo);

     if (_rtn == 0)
       {
         cerr << "Exception handling part..." << endl;
         PaCO::InterfaceParallel::ref ref_nodes;
         ref_nodes.length(_nodes.length());
         for (CORBA::ULong j = 0; j<_nodes.length(); j++)
      {
        ref_nodes[j] = _nodes[j];
      }
         interface->setRef(ref_nodes);
         cerr << "Exception handling part...done" << endl;
       }
   }
    }
  else
    {
      cerr << "Deploying " << _workNode << " nodes" << endl;
      for (  CORBA::Long i = 0; i<_workNode; i++)
   {
     cerr << "Deploying node : " << i << endl;
     object = _orb->string_to_object(_nodes[i]);
     interface = PaCO::InterfaceParallel::_narrow(object);
     interface->init_InterfaceParallel(i,_nodes.length(), _informations.serveur_topo);
   }
    }
  if (_rtn == 0)
    {
      // connecting the client to the interface manager
      cerr << "Finishing...\n";
      this->finish();
      cerr << "Finishing...done\n";
    }
}

PaCO::ConnectionInfos*
InterfaceManager_impl::getConnectionInfos()
{
//  cerr << "In "<<__FUNCTION__ << ":"<<(void*)this<<endl;
  _current_uid++;
  _informations.request_uid = _current_uid;
  PaCO::ConnectionInfos* retour = new PaCO::ConnectionInfos(_informations);
 // cerr << "Out "<<__FUNCTION__ << ":"<<(void*)this<<endl;
  return retour;
}

// class InterfaceParallel_impl

InterfaceParallel_impl::InterfaceParallel_impl(CORBA::ORB_ptr current_orb, char * ior)
{
  _myRank = -1;
  _totalNode = -1;
  _orb = current_orb;
  _ior = ior;
}

InterfaceParallel_impl::~InterfaceParallel_impl() 
{
}

void
InterfaceParallel_impl::deploy(CORBA::Short rank)
{
  cerr << "Debut de deploy()"<< endl;
  _deployRank = rank;
  CORBA::Object_var object;
  object = _orb->string_to_object(_ior.c_str());
  PaCO::InterfaceManager_var interface;
  interface = PaCO::InterfaceManager::_narrow(object);
  char * obj = _orb->object_to_string(_this());
  try
    {
      interface->setNewNode(obj, rank);
    }
  catch (const CORBA::Exception & e)
    {
      cerr << "Argument : " << obj << endl;
      cerr << "InterfaceParallel_impl::deploy : Exception : " << e << endl;
    }
  cerr << "Fin de deploy()"<< endl;
}

CORBA::Short
InterfaceParallel_impl::getMyRank()
{
  return _myRank;
}

CORBA::Short
InterfaceParallel_impl::getDeployRank()
{
  return _deployRank;
}

CORBA::Short
InterfaceParallel_impl::getTotalNode()
{
  return _totalNode;
}

void
InterfaceParallel_impl::init_InterfaceParallel(CORBA::Short myRank, CORBA::Short totalNode)
{
  _myRank = myRank;
  _totalNode = totalNode;
  cerr << "Servant : " << myRank << " initialized" << endl;
}

PaCO::ConnectionInfos*
InterfaceParallel_impl::getConnectionInfos()
{
  return NULL;
}

void 
InterfaceParallel_impl::refReturnObject(const char * ref) 
{
  return_object = (char *) ref;
}

void 
InterfaceParallel_impl::setRef(const PaCO::InterfaceParallel::ref & ref_nodes)
{
  _ref_nodes.length(ref_nodes.length());
  for (CORBA::ULong i = 0; i<ref_nodes.length(); i++)
    _ref_nodes[i] = ref_nodes[i];
}