dt_creator.cpp

来自「这是广泛使用的通信开源项目,对于大容量,高并发的通讯要求完全能够胜任,他广泛可用」· C++ 代码 · 共 677 行 · 第 1/2 页

//DT_Creator.cpp,v 1.4 2003/07/29 05:08:14 yamuna Exp

#include "DT_Creator.h"
#include "Thread_Task.h"
#include "tao/ORB_Core.h"
#include "Task_Stats.h"
#include "ace/High_Res_Timer.h"
#include "DT_Creator.h"
#include "tao/RTScheduling/Current.h"

ACE_Atomic_Op<TAO_SYNCH_MUTEX, long> guid_counter;

int
DT_Creator::dt_task_init (ACE_Arg_Shifter& arg_shifter)
{
  static int dt_index = 0;
  int start_time = 0;
  int load = 0;
  int iter = 0;
  int importance = 0;
  char *job_name = 0;
  int dist = 0;
  const ACE_TCHAR* current_arg = 0;
  if (arg_shifter.cur_arg_strncasecmp ("-Importance") == 0)
    {
      arg_shifter.consume_arg ();
      current_arg = arg_shifter.get_current ();
      importance = ACE_OS::atoi (current_arg);
      arg_shifter.consume_arg ();
    }
  if ((current_arg = arg_shifter.get_the_parameter ("-Start_Time")))
    {
      start_time = ACE_OS::atoi (current_arg);
      arg_shifter.consume_arg ();
    }
  if ((current_arg = arg_shifter.get_the_parameter ("-Iter")))
    {
      iter = ACE_OS::atoi (current_arg);
      arg_shifter.consume_arg ();
    }
  if ((current_arg = arg_shifter.get_the_parameter ("-Load")))
    {
      load = ACE_OS::atoi (current_arg);
      arg_shifter.consume_arg ();
    }
  if ((current_arg = arg_shifter.get_the_parameter ("-JobName")))
    {
      job_name = (char *)current_arg;
      dist = 1;
      arg_shifter.consume_arg ();
    }
  dt_list_ [dt_index++] = this->create_thr_task (importance,
						 start_time,
						 load,
						 iter,
						 dist,
						 job_name);
  return 0;

}

int log_index = 0;
int
DT_Creator::init (int argc, char *argv [])
{
  gsf_ = ACE_High_Res_Timer::global_scale_factor ();
  state_lock_ = new ACE_Lock_Adapter <TAO_SYNCH_MUTEX>;
  shutdown_lock_ = new ACE_Lock_Adapter <TAO_SYNCH_MUTEX>;
  active_dt_count_ = 0;
  active_job_count_ = 0;
  ACE_NEW_RETURN (log, char*[BUFSIZ * 100],-1);

  ACE_Arg_Shifter arg_shifter (argc, argv);

  const ACE_TCHAR* current_arg = 0;

  dt_count_ = 0;
  poa_count_ = 0;
  int poa_count = 0;
  job_count_ = 0;
  int job_count = 0;
  while (arg_shifter.is_anything_left ())
    {
	if ((current_arg = arg_shifter.get_the_parameter ("-GuidSeed")))
        {
	  guid_counter = (long) ACE_OS::atoi (current_arg);
	  arg_shifter.consume_arg ();
	}
	else if ((current_arg = arg_shifter.get_the_parameter ("-DT_Count")))
	  {
	    dt_count_ = ACE_OS::atoi (current_arg);
          ACE_NEW_RETURN (dt_list_, Thread_Task*[dt_count_], -1);
	  active_dt_count_ = dt_count_;
          arg_shifter.consume_arg ();
	  }
      else if ((current_arg = arg_shifter.get_the_parameter ("-POA_Count")))
        {
          poa_count_ = ACE_OS::atoi (current_arg);
          ACE_NEW_RETURN (poa_list_, POA_Holder*[poa_count_], -1);
          arg_shifter.consume_arg ();
        }
      else if ((current_arg = arg_shifter.get_the_parameter ("-JOB_Count")))
        {
          job_count_ = ACE_OS::atoi (current_arg);
	  active_job_count_ = job_count_;
	  ACE_NEW_RETURN (job_list_, Job_i*[job_count_], -1);
          arg_shifter.consume_arg ();
        }
      else if (arg_shifter.cur_arg_strncasecmp ("-DT_Task") == 0)
	{
          arg_shifter.consume_arg ();
	  dt_task_init (arg_shifter);
	}
      else if (arg_shifter.cur_arg_strncasecmp ("-POA") == 0)
	{
          arg_shifter.consume_arg ();

          ACE_NEW_RETURN (this->poa_list_[poa_count], POA_Holder (), -1);

          if (this->poa_list_[poa_count]->init (arg_shifter) == -1)
            {
              delete this->poa_list_[poa_count];
              return -1;
            }
	  else
	    poa_count++;
	}
      else if (arg_shifter.cur_arg_strncasecmp ("-Job") == 0)
	{
          arg_shifter.consume_arg ();

          ACE_NEW_RETURN (this->job_list_[job_count], Job_i (this), -1);

          if (this->job_list_[job_count]->init (arg_shifter) == -1)
	    {
	      delete this->job_list_[job_count];
	      return -1;
	    }
	  else 
	    job_count++;
	}
      else if ((current_arg = arg_shifter.get_the_parameter ("-OutFile")))
	{
	  file_name_ = CORBA::string_dup (current_arg);
	  arg_shifter.consume_arg ();
	}
      else if ((current_arg = arg_shifter.get_the_parameter ("-LogFile")))
	{
	  log_file_name_ = CORBA::string_dup (current_arg);
	  arg_shifter.consume_arg ();
	}
      else
        {
          arg_shifter.ignore_arg ();
        }
    }

  return 0;
}

void
DT_Creator::register_synch_obj (ACE_ENV_SINGLE_ARG_DECL)
{
  CosNaming::Name name (1);
  name.length (1);

  CosNaming::NamingContext_var synch_context;

  ACE_TRY
    {
      // Try binding the sender context in the NS
      name [0].id =
        CORBA::string_dup ("Synch");

      synch_context = this->naming_->bind_new_context (name
						       ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      //
      // We reach here if there was no exception raised in
      // <bind_new_context>.  We then create a receiver context.
      //

    }
  ACE_CATCH (CosNaming::NamingContext::AlreadyBound, al_ex)
    {
      //
      // The synch context already exists, probably created by the
      // receiver(s).
      //

      // Get the synch context.
      name [0].id =
        CORBA::string_dup ("Synch");

      CORBA::Object_var object =
        this->naming_->resolve (name
				ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;

      synch_context = CosNaming::NamingContext::_narrow (object.in ());

    }
  ACE_ENDTRY;
  ACE_CHECK;

  ACE_CString synch_name ("Synch");
  ACE_Time_Value timestamp = ACE_OS::gettimeofday ();

  char buf [BUFSIZ];
  synch_name += CORBA::string_dup (ACE_OS::itoa (timestamp.sec (), buf, 10));

  name [0].id =
    CORBA::string_dup (synch_name.c_str ());

  ACE_DEBUG ((LM_DEBUG,
	      "Synch Name %s\n",
	      synch_name.c_str ()));
  
  ACE_NEW (synch_,
	   Synch_i);
  
  Synch_var synch = synch_->_this (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;

  // Register the synch object with the Synch context.
  synch_context->rebind (name,
			 synch.in ()
			 ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

}


int
DT_Creator::activate_root_poa (ACE_ENV_SINGLE_ARG_DECL)
{
  CORBA::Object_var object =
    orb_->resolve_initial_references ("RootPOA"
				      ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (-1);

  root_poa_ =
    PortableServer::POA::_narrow (object.in ()
				  ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (-1);

  PortableServer::POAManager_var poa_manager =
    root_poa_->the_POAManager (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK_RETURN (-1);

  poa_manager->activate (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK_RETURN (-1);

  return 0;
}

void
DT_Creator::activate_poa_list (ACE_ENV_SINGLE_ARG_DECL)
{
  if (TAO_debug_level > 0)
    ACE_DEBUG ((LM_DEBUG,
		"DT_Creator::activate_poa_list\n"));

  if (poa_count_ > 0)
    {
      CORBA::Object_var object =
	orb_->resolve_initial_references ("RTORB"
					  ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;

      this->rt_orb_ =
	RTCORBA::RTORB::_narrow (object.in ()
				 ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;
    }
  
  for (int i = 0; i < poa_count_; ++i)
    {
      poa_list_[i]->activate (this->rt_orb_.in(), this->root_poa_.in ()
			      ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;
    }
}

void
DT_Creator::activate_job_list (ACE_ENV_SINGLE_ARG_DECL)
{

  if (TAO_debug_level > 0)
    ACE_DEBUG ((LM_DEBUG,
		"DT_Creator::activate_job_list\n"));

  Job_i* job;

  for (int i = 0; i < job_count_; ++i)
    {
      job = job_list_[i];

      if (TAO_debug_level > 0)
        ACE_DEBUG ((LM_DEBUG, "Activating job:%s\n", job->name ().c_str ()));

      // find your poa
      PortableServer::POA_var host_poa =
        root_poa_->find_POA (job->poa ().c_str (), 0
                             ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;

      PortableServer::ServantBase_var servant_var (job);

      // Register with poa.
      PortableServer::ObjectId_var id;

      id = host_poa->activate_object (job
                                      ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;

      CORBA::Object_var server =
        host_poa->id_to_reference (id.in ()
                                   ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;

      CORBA::String_var ior =
        orb_->object_to_string (server.in ()
                                ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;

      const ACE_CString &job_name = job->name ();

      CosNaming::Name_var name =
        this->naming_->to_name (job_name.c_str ()
                                ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;

      this->naming_->rebind (name.in (),
                             server.in ()
                             ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;