path: root/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer_Supplier_EC.cpp

// $Id$

#include "ace/SString.h"
#include "ace/Get_Opt.h"
#include "ace/Sched_Params.h"
#include "ace/Thread.h"
#include "ace/OS_NS_sys_time.h"
#include "ace/Select_Reactor_Base.h" //for ACE_Select_Reactor_Impl::DEFAULT_SIZE

#include "orbsvcs/Event/EC_Gateway_IIOP_Factory.h"
#include "orbsvcs/Event/EC_Gateway_Sched.h"
#include "orbsvcs/Event/EC_Kokyu_Factory.h"
#include "orbsvcs/Time_Utilities.h"
#include "orbsvcs/Event_Service_Constants.h"
#include "orbsvcs/Scheduler_Factory.h"
#include "tao/ORB_Core.h"

#include "Kokyu_EC.h"
#include "Consumer.h"
#include "Supplier.h"

#ifdef ACE_HAS_DSUI
#include <dsui.h>
#include "federated_config.h"
#include "federated_dsui_families.h"
#endif //ACE_HAS_DSUI

#include <sstream>

namespace
{
  int supp_id = 3;
  ACE_CString sched_type = "rms";
  ACE_CString ior_input_file = "file://consumer_ec.ior";
  ACE_CString ior_output_filename;
  FILE * ior_output_file;
}

class Consumer_Supplier_EC : public Kokyu_EC
{
  TAO_EC_Gateway_Sched gateway;
public:
  Consumer_Supplier_EC()
  {
  } //Consumer_Supplier_EC()

  ~Consumer_Supplier_EC(void)
  {
  } //~Consumer_Supplier_EC()

  void init_gateway(CORBA::ORB_ptr orb,
                    PortableServer::POA_ptr poa,
                    const char* consumer_ec_ior)
  {
      CORBA::Object_var obj;
      RtEventChannelAdmin::RtSchedEventChannel_var supplier_ec, consumer_ec;

      ACE_CHECK;
      obj = orb->string_to_object(consumer_ec_ior ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;

      consumer_ec = RtEventChannelAdmin::RtSchedEventChannel::_narrow(obj.in()
                                                                      ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;

      RtecEventChannelAdmin::EventChannel_var supplier_event_channel =
        this->event_channel(ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_CHECK;

      RtecEventChannelAdmin::EventChannel_var consumer_event_channel =
        consumer_ec->event_channel(ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_CHECK;

      RtecScheduler::Scheduler_var supplier_scheduler =
        this->scheduler(ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_CHECK;

      RtecScheduler::Scheduler_var consumer_scheduler =
        consumer_ec->scheduler(ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_CHECK;

      std::stringstream cons_gateway;
      cons_gateway << consumer_ec_ior << " from " << supp_id;

      ACE_DEBUG((LM_DEBUG,"Consumer_Supplier (%P|%t) init_gateway(): remote RT_Info entry_point is \"%s\"\n",cons_gateway.str().c_str()));

      std::stringstream supp_gateway;
      supp_gateway << "Gateway for " << supp_id;

      ACE_DEBUG((LM_DEBUG,"Consumer_Supplier (%P|%t) init_gateway(): local RT_Info entry_point is \"%s\"\n",supp_gateway.str().c_str()));

      gateway.init(supplier_event_channel.in(),
                   consumer_event_channel.in(),
                   supplier_scheduler.in(),
                   consumer_scheduler.in(),
                   supp_gateway.str().c_str(), cons_gateway.str().c_str()
                   ACE_ENV_ARG_PARAMETER);

      ACE_CHECK;
            PortableServer::ObjectId_var gateway_oid =
         poa->activate_object(&gateway ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;

      CORBA::Object_var gateway_obj =
         poa->id_to_reference(gateway_oid.in() ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;

       RtecEventChannelAdmin::Observer_var obs =
         RtecEventChannelAdmin::Observer::_narrow(gateway_obj.in() ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;

      RtecEventChannelAdmin::Observer_Handle local_ec_obs_handle =
         consumer_event_channel->append_observer (obs.in () ACE_ENV_ARG_PARAMETER);
      ACE_UNUSED_ARG(local_ec_obs_handle);
      ACE_CHECK;

      consumer_ec->start(ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_CHECK;

      ACE_DEBUG((LM_DEBUG,"Gateway Initialized\n"));
  }

  virtual void set_up_supp_and_cons (ACE_ENV_SINGLE_ARG_DECL)
      ACE_THROW_SPEC ((
        CORBA::SystemException
        , RtecScheduler::UNKNOWN_TASK
        , RtecScheduler::INTERNAL
        , RtecScheduler::SYNCHRONIZATION_FAILURE
      ))
  {
    ACE_Time_Value tv;

    //Supplier types
    RtecEventComm::EventType supp_normal_type;
    RtecEventComm::EventType supp_ft_type;
    //Consumer types
    RtecEventComm::EventType cons_normal_type;
    RtecEventComm::EventType cons_ft_type;
    if (supp_id == 3)
      {
        ACE_DEBUG((LM_DEBUG,"Consumer_Supplier_EC has id 3\n"));
        supp_normal_type = ACE_ES_EVENT_UNDEFINED+6;
        supp_ft_type = ACE_ES_EVENT_UNDEFINED+8;
        cons_normal_type = ACE_ES_EVENT_UNDEFINED+2;
        cons_ft_type = ACE_ES_EVENT_UNDEFINED+4;
      }
    else //supp_id == 4
      {
        ACE_DEBUG((LM_DEBUG,"Consumer_Supplier_EC has id 4\n"));
        supp_normal_type = ACE_ES_EVENT_UNDEFINED+7;
        supp_ft_type = ACE_ES_EVENT_UNDEFINED+9;
        cons_normal_type = ACE_ES_EVENT_UNDEFINED+3;
        cons_ft_type = ACE_ES_EVENT_UNDEFINED+5;
      }
    Kokyu_EC::EventType_Vector supp1_3_types;
    supp1_3_types.push_back(supp_normal_type);
    supp1_3_types.push_back(supp_ft_type);

    Kokyu_EC::EventType_Vector cons1_2_types(2);
    cons1_2_types.push_back(cons_normal_type);
    cons1_2_types.push_back(cons_ft_type);

    Supplier *supplier_impl1_3;
    ACE_NEW(supplier_impl1_3,
            Supplier(supp_id,supp_normal_type,supp_ft_type));

    Consumer * consumer_impl1_2;
    ACE_NEW(consumer_impl1_2,
            Consumer(cons_normal_type,cons_ft_type,supplier_impl1_3));

    tv.set(0,50000);
    consumer_impl1_2->setWorkTime(tv);
    tv.set(0,200000); //Period
    add_consumer_with_supplier(consumer_impl1_2, //deleted in consumer
                               "consumer1_2",
                               tv,
                               cons1_2_types[0],
                               RtecScheduler::VERY_LOW_CRITICALITY,
                               RtecScheduler::VERY_LOW_IMPORTANCE,
                               supplier_impl1_3,
                               "supplier1_3",
                               supp1_3_types
                               ACE_ENV_ARG_PARAMETER
                               );
    ACE_CHECK;

    EventType_Vector types(4);
    for(RtecEventComm::EventType type = ACE_ES_EVENT_UNDEFINED+6;
        type <= ACE_ES_EVENT_UNDEFINED+9; type++)
      {
        types.push_back(type);
      }

    add_dummy_supplier(types);
    ACE_CHECK;

    //DEBUG: print out schedule
    RtecScheduler::Scheduler_ptr scheduler = this->scheduler(ACE_ENV_SINGLE_ARG_PARAMETER);
    //RtecEventChannelAdmin::EventChannel_ptr event_channel = this->event_channel(ACE_ENV_SINGLE_ARG_DECL);

    RtecScheduler::RT_Info_Set_var infos;
    RtecScheduler::Config_Info_Set_var configs;
    RtecScheduler::Dependency_Set_var dependencies;
    RtecScheduler::Scheduling_Anomaly_Set unsafe_anomalies;
    RtecScheduler::Scheduling_Anomaly_Set_var anomalies;

    int min_os_priority =
      ACE_Sched_Params::priority_min (ACE_SCHED_FIFO,
                                      ACE_SCOPE_THREAD);
    int max_os_priority =
      ACE_Sched_Params::priority_max (ACE_SCHED_FIFO,
                                      ACE_SCOPE_THREAD);

    scheduler->compute_scheduling (min_os_priority,
                                   max_os_priority,
                                   infos.out (),
                                   dependencies.out (),
                                   configs.out (),
                                   anomalies.out ()
                                   ACE_ENV_ARG_PARAMETER);
    ACE_TRY_CHECK;

    std::stringstream sched_out;
    ACE_CString ior_prefix(ior_output_filename.c_str(),ior_output_filename.length()-4); //cut off '.ior'
    sched_out << "schedule_" << ior_prefix.c_str() << ".out";

    ACE_DEBUG((LM_DEBUG,"Supplier_EC writing schedule to %s\n",sched_out.str().c_str()));

    ACE_Scheduler_Factory::dump_schedule (infos.in (),
                                          dependencies.in (),
                                          configs.in (),
                                          anomalies.in (),
                                          sched_out.str().c_str());
    ////END DEBUG
    //Kokyu_EC::start(ACE_ENV_SINGLE_ARG_PARAMETER);
    //ACE_CHECK;
  }
};

int parse_args (int argc, char *argv[]);

int
main (int argc, char* argv[])
{
  //TAO_EC_Default_Factory::init_svcs ();
  TAO_EC_Kokyu_Factory::init_svcs ();
  TAO_EC_Gateway_IIOP_Factory::init_svcs ();

  ACE_DECLARE_NEW_CORBA_ENV;
  ACE_TRY
    {
#ifdef ACE_HAS_DSUI
      ds_control ds_cntl("Dynamic_Test_Consumer_Supplier","consumer_supplier_enabled.dsui");
#endif //ACE_HAS_DSUI

      // ORB initialization boiler plate...
      CORBA::ORB_var orb =
        CORBA::ORB_init (argc, argv, "" ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      if (parse_args (argc, argv) == -1)
        {
          return 1;
        }

      CORBA::Object_var object =
        orb->resolve_initial_references ("RootPOA" ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
      PortableServer::POA_var poa =
        PortableServer::POA::_narrow (object.in () ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
      PortableServer::POAManager_var poa_manager =
        poa->the_POAManager (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;
      poa_manager->activate (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

      //spawn thread to run the reactor event loop
      Reactor_Task rt;
      rt.initialize();

      // ****************************************************************

      Consumer_Supplier_EC cons_supp_ec;
      if (cons_supp_ec.init(sched_type.c_str(), poa.in()) == -1)
        {
          ACE_ERROR_RETURN((LM_ERROR, "Unable to initialize Kokyu_EC"), 1);
        }

      cons_supp_ec.init_gateway(orb.in(),
                               poa.in(),
                               ior_input_file.c_str() ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
      // ****************************************************************
      RtEventChannelAdmin::RtSchedEventChannel_var cons_supp_ec_ior =
        cons_supp_ec._this(ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

      CORBA::String_var ior = orb->object_to_string(cons_supp_ec_ior.in()
                                                    ACE_ENV_ARG_PARAMETER);

      ACE_OS::fprintf(ior_output_file, ior.in());
      ACE_OS::fclose(ior_output_file);


      ACE_hthread_t thr_handle;
      ACE_Thread::self (thr_handle);

      int prio = ACE_Sched_Params::priority_max (ACE_SCHED_FIFO);
      ACE_OS::thr_setprio (thr_handle, prio);

      cons_supp_ec.set_up_supp_and_cons(ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;
      //cons_supp_ec.start(ACE_ENV_SINGLE_ARG_PARAMETER);
      //ACE_TRY_CHECK;

      //@BT: Timeouts start when orb starts, similar to starting the DT worker thread
      //DSTRM_EVENT (MAIN_GROUP_FAM, WORKER_ACTIVATED, 1, 0, NULL);
      ACE_DEBUG((LM_DEBUG,"Consumer_Supplier_EC thread %t WORKER_ACTIVATED at %u\n",ACE_OS::gettimeofday().msec()));
      DSTRM_EVENT (MAIN_GROUP_FAM, WORKER_ACTIVATED, 0, 0, NULL);
      //DSTRM_EVENT (WORKER_GROUP_FAM, WORKER_STARTED, m_id, 0, NULL);
      ACE_DEBUG((LM_DEBUG,"Consumer_Supplier_EC thread %t WORKER_STARTED at %u\n",ACE_OS::gettimeofday().msec()));
      DSTRM_EVENT (WORKER_GROUP_FAM, WORKER_STARTED, 0, 0, NULL);

#ifdef ACE_HAS_DSUI
      //@BT
      ACE_Time_Value now(ACE_OS::gettimeofday());
      ACE_OS::printf("Consumer_Supplier_EC START at %isec %iusec\n",now.sec(),now.usec());
      DSTRM_EVENT(MAIN_GROUP_FAM, START,0,0,NULL);
#endif //ACE_HAS_DSUI

      rt.activate(); //need thread creation flags? or priority?
      ACE_Time_Value stop_time(305,0);
      orb->run (stop_time ACE_ENV_ARG_PARAMETER);
      //orb->run (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

#ifdef ACE_HAS_DSUI
      //@BT
      //DSTRM_EVENT(MAIN_GROUP_FAM, STOP, 1, 0, NULL);
      ACE_DEBUG((LM_DEBUG,"Consumer_Supplier_EC thread %t STOP at %u\n",ACE_OS::gettimeofday().msec()));
      DSTRM_EVENT(MAIN_GROUP_FAM, STOP, 1, 0, NULL);
#endif //ACE_HAS_DSUI

      // ****************************************************************

      // We should do a lot of cleanup (disconnect from the EC,
      // deactivate all the objects with the POA, etc.) but this is
      // just a simple demo so we are going to be lazy.

    }
  ACE_CATCHANY
    {
      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION, "Consumer_Supplier_EC");
      return 1;
    }
  ACE_ENDTRY;

  return 0;
}

// ****************************************************************

int parse_args (int argc, char *argv[])
{
  ACE_Get_Opt get_opts (argc, argv, "s:o:d:i:");
  int c;
  //these used for handline '-i':
  const char* input_file;
  size_t len;
  char *filename;

  while ((c = get_opts ()) != -1)
    switch (c)
      {
      case 'o':
        {
          ior_output_filename = get_opts.opt_arg();
          ior_output_file = ACE_OS::fopen (ior_output_filename.c_str(), "w");
          if (ior_output_file == 0)
            ACE_ERROR_RETURN ((LM_ERROR,
                               "Unable to open %s for writing: %p\n",
                               get_opts.opt_arg ()), -1);
          break;
        }
      case 's':
        {
          sched_type = ACE_TEXT_ALWAYS_CHAR(get_opts.opt_arg ());
          break;
        }
      case 'd':
        {
          supp_id = atol(get_opts.opt_arg ());
          break;
        }
      case 'i':
        input_file = get_opts.opt_arg();
        len = ACE_OS::strlen("file://")+ACE_OS::strlen(input_file)+1;
        filename = new char[len];
        sprintf(filename,"file://%s",input_file);
        ACE_DEBUG((LM_DEBUG,"Adding consumer IOR %s\n",filename));
        ior_input_file = filename;
        break;
      case '?':
      default:
        {
          ACE_ERROR_RETURN ((LM_ERROR,
                             "Usage:  %s -s <rms|muf|edf>"
                             " -d <supplier id>"
                             " [-o iorfile]"
                             " [-i server_EC_ior_file]"
                             "\n"
                             "For multiple consumers, specify -i multiple times\n",
                             argv [0]),
                            -1);
        }
      }
  if (ior_output_file == 0)
    {
      ior_output_filename = "consumer_supplier_ec.ior";
      ior_output_file = ACE_OS::fopen (ior_output_filename.c_str(), "w");
    }
  // Indicates sucessful parsing of the command line
  return 0;
}