自学教程：C++ ACE_Reactor类代码示例

  void  Sender_exec_i::start (void)  {    ACE_Reactor* reactor = 0;    ::CORBA::Object_var ccm_object = this->ciao_context_->get_CCM_object();    if (!::CORBA::is_nil (ccm_object.in ()))      {        ::CORBA::ORB_var orb = ccm_object->_get_orb ();        if (!::CORBA::is_nil (orb.in ()))          {            reactor = orb->orb_core ()->reactor ();          }      }    if (reactor)      {        // calculate the interval time        long const usec = 1000000 / this->rate_;        if (reactor->schedule_timer (                    this->ticker_,                    0,                    ACE_Time_Value (3, usec),                    ACE_Time_Value (0, usec)) == -1)        {          ACE_ERROR ((LM_ERROR, ACE_TEXT ("Sender_exec_i::start : ")                                ACE_TEXT ("Error scheduling timer")));        }      }    else      {        throw ::CORBA::INTERNAL ();      }  }

intImR_Activator_i::handle_exit (ACE_Process * process){  if (debug_ > 0)    {      ORBSVCS_DEBUG        ((LM_DEBUG,          ACE_TEXT ("Process %d exited with exit code %d, delay = %d/n"),          process->getpid (), process->return_value (), this->induce_delay_));    }  if (this->induce_delay_ > 0 && this->active_check_pid_ == ACE_INVALID_PID)    {      ACE_Reactor *r = this->orb_->orb_core()->reactor();      ACE_Time_Value dtv (0, this->induce_delay_ * 1000);      pid_t pid = process->getpid();      Act_token_type token = static_cast<Act_token_type>(pid);      r->schedule_timer (this, reinterpret_cast<void *>(token), dtv );    }  else    {      this->handle_exit_i (process->getpid());    }  return 0;}

voiddevice_averager::deactivate(){    doit( device_state::command_off );    ACE_Reactor * reactor = acewrapper::singleton::ReactorThread::instance()->get_reactor();    reactor->cancel_timer( this );}

// If any command line arg is given, run the test with high res timer// queue. Else run it normally.intrun_main (int argc, ACE_TCHAR *[]){  ACE_START_TEST (ACE_TEXT ("Reactor_Timer_Test"));  if (argc > 1)    {      ACE_DEBUG ((LM_DEBUG,                  ACE_TEXT ("Running with high-res timer queue/n")));      ACE_Reactor *r = ACE_Reactor::instance ();      (void) ACE_High_Res_Timer::global_scale_factor ();      r->timer_queue ()->gettimeofday (&ACE_High_Res_Timer::gettimeofday_hr);    }  // Register all different handlers, i.e., one per timer.  test_registering_all_handlers ();  // Now try multiple timers for ONE event handler (should produce the  // same result).  test_registering_one_handler ();  // Try canceling handlers with odd numbered timer ids.  test_canceling_odd_timers ();  // Make sure <reset_timer_inverval> works.  test_resetting_timer_intervals ();  ACE_END_TEST;  return 0;}

intACE_TMAIN (int argc, ACE_TCHAR *argv[]){  ACE_Sig_Action sa ((ACE_SignalHandler) handler, SIGINT);  ACE_OS::signal (SIGCLD, SIG_IGN);  ACE_UNUSED_ARG (sa);  parse_args (argc, argv);  OUTPUT_FILE = ACE_OS::open (OUTPUT_FILE_NAME, O_CREAT | O_WRONLY, 0644);  if (OUTPUT_FILE == 0)    return 1;  ACE_Reactor reactor;  Handle_Events handle_events (UDP_PORT, MCAST_ADDR, INTERFACE, reactor);  // main loop  while (!done)    reactor.handle_events ();  ACE_OS::close (OUTPUT_FILE);  cout << "/nbenchd done./n";  return 0;}

// This is run at program initializationvoid CommandLineServer::run(void* args){	s_log = log4cxx::Logger::getLogger("interface.commandlineserver");	unsigned short tcpPort = (unsigned short)(unsigned long)args;	//unsigned short tcpPort = (unsigned short)*(int*)args;	CommandLineAcceptor peer_acceptor;	ACE_INET_Addr addr (tcpPort);	ACE_Reactor reactor;	CStdString tcpPortString = IntToString(tcpPort);	if (peer_acceptor.open (addr, &reactor) == -1)	{		LOG4CXX_ERROR(s_log, CStdString("Failed to start command line server on port:") + tcpPortString + CStdString(" do you have another instance of orkaudio running?"));	}	else	{		LOG4CXX_INFO(s_log, CStdString("Started command line server on port:")+tcpPortString);		for(;;)		{			reactor.handle_events();		}	}}

Dispatch_Count_Handler::Dispatch_Count_Handler (void){  ACE_Reactor *r = ACE_Reactor::instance ();  this->input_seen_ = this->notify_seen_ = 0;  this->timers_fired_ = 0;  // Initialize the pipe.  if (this->pipe_.open () == -1)    ACE_ERROR ((LM_ERROR,                ACE_TEXT ("%p/n"),                ACE_TEXT ("ACE_Pipe::open")));  // Register the "read" end of the pipe.  else if (r->register_handler (this->pipe_.read_handle (),                                this,                                ACE_Event_Handler::READ_MASK) == -1)    ACE_ERROR ((LM_ERROR,                ACE_TEXT ("%p/n"),                ACE_TEXT ("register_handler")));  // Put something in our pipe and smoke it... ;-)  else if (ACE::send (this->pipe_.write_handle (),                      "z",                      1) == -1)    ACE_ERROR ((LM_ERROR,                ACE_TEXT ("%p/n"),                ACE_TEXT ("send")));  // Call notify to prime the pump for this, as well.  else if (r->notify (this) == -1)    ACE_ERROR ((LM_ERROR,                ACE_TEXT ("%p/n"),                ACE_TEXT ("notify")));}

intACE_Process_Manager::append_proc (ACE_Process *proc){  ACE_TRACE ("ACE_Process_Manager::append_proc");  // Try to resize the array to twice its existing size if we run out  // of space...  if (this->current_count_ >= this->max_process_table_size_      && this->resize (this->max_process_table_size_ * 2) == -1)    return -1;  else    {      ACE_Process_Descriptor &proc_desc =        this->process_table_[this->current_count_];      proc_desc.process_ = proc;      proc_desc.exit_notify_ = 0;#if defined (ACE_WIN32)      // If we have a Reactor, then we're supposed to reap Processes      // automagically.  Get a handle to this new Process and tell the      // Reactor we're interested in <handling_input> on it.      ACE_Reactor *r = this->reactor ();      if (r != 0)        r->register_handler (this,                             proc->gethandle ());#endif /* ACE_WIN32 */      this->current_count_++;      return 0;    }}

intImR_Activator_i::handle_exit (ACE_Process * process){  if (debug_ > 0)    {      ORBSVCS_DEBUG        ((LM_DEBUG,        ACE_TEXT ("Process %d exited with exit code %d/n"),        process->getpid (), process->return_value ()));    }  if (this->induce_delay_ > 0)    {      ACE_Reactor *r = this->orb_->orb_core()->reactor();      ACE_Time_Value dtv (0, this->induce_delay_ * 1000);      pid_t pid = process->getpid();#if (ACE_SIZEOF_VOID_P == 8)      ACE_INT64 token = static_cast<ACE_INT64>(pid);#else      ACE_INT32 token = static_cast<ACE_INT32>(pid);#endif      r->schedule_timer (this, reinterpret_cast<void *>(token), dtv );    }  else    {      this->handle_exit_i (process->getpid());    }  return 0;}

static void *worker (void *args){  ACE_Reactor *reactor = reinterpret_cast<ACE_Reactor *> (args);  // Make this thread the owner of the Reactor's event loop.  reactor->owner (ACE_Thread::self ());  // Use a timeout to inform the Reactor when to shutdown.  ACE_Time_Value timeout (4);  for (;;)    switch (reactor->handle_events (timeout))      {      case -1:        ACE_ERROR_RETURN ((LM_ERROR,                           ACE_TEXT ("(%t) %p/n"),                           ACE_TEXT ("reactor")),                          0);        /* NOTREACHED */      case 0:        ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Reactor shutdown/n")));        return 0;      }  ACE_NOTREACHED (return 0);}

static ACE_THR_FUNC_RETURN event_loop (void *arg) {  ACE_Reactor *reactor = static_cast<ACE_Reactor *> (arg);  reactor->owner (ACE_OS::thr_self ());  reactor->run_reactor_event_loop ();  return 0;}

// server main function// uses a portable form of the "main" function along with its argumentsint ACE_TMAIN (int argc, ACE_TCHAR *argv[]){   // instantiate an object of the Proto Handler class  Proto_Handler proto;  // retrieve a reactor. Here we could have retrieved different  // implementations of reactor. For now we get the default singletom  // reactor.   ACE_Reactor *reactor = ACE_Reactor::instance ();  // initialize the proto handler object  if (proto.open (argc, argv, reactor) == -1) {      ACE_ERROR ((LM_ERROR, ACE_TEXT ("(%P|%t) %p/n"),                   ACE_TEXT ("handler open")));      return -1;  }  // now let the server handle the events  for (;;) {    if (reactor->handle_events () == -1) {      ACE_ERROR ((LM_ERROR, ACE_TEXT ("(%P|%t) %p/n"),                   ACE_TEXT ("handle events")));      return -1;    }  }  // return exit status  return 0;}

static ACE_THR_FUNC_RETURN controller (void *arg) {  ACE_Reactor *reactor = static_cast<ACE_Reactor *> (arg);  Quit_Handler *quit_handler = 0;  ACE_NEW_RETURN (quit_handler, Quit_Handler (reactor), 0);#if defined (ACE_WIN32) && (!defined (ACE_HAS_STANDARD_CPP_LIBRARY) || /                            (ACE_HAS_STANDARD_CPP_LIBRARY == 0) || /                            defined (ACE_USES_OLD_IOSTREAMS))  for (;;) {    char user_input[80];    ACE_OS::fgets (user_input, sizeof (user_input), stdin);    if (ACE_OS::strcmp (user_input, "quit") == 0) {      reactor->notify (quit_handler);      break;    }  }#else  for (;;) {    std::string user_input;    std::getline (cin, user_input, '/n');    if (user_input == "quit") {      reactor->notify (quit_handler);      break;    }  }#endif  return 0;}

static voidtest_for_spin (ACE_Reactor &reactor){  Handler handler (reactor, true);  // This should trigger a call to <handle_input>.  ssize_t result =    ACE::send_n (handler.pipe_.write_handle (),                 message,                 ACE_OS::strlen (message));  if (result != ssize_t (ACE_OS::strlen (message)))    ACE_ERROR ((LM_ERROR, ACE_TEXT ("Handler sent %b bytes; should be %B/n"),                result, ACE_OS::strlen (message)));  reactor.run_reactor_event_loop ();  if (0 != reactor.remove_handler (handler.pipe_.read_handle (),                                   ACE_Event_Handler::ALL_EVENTS_MASK |                                   ACE_Event_Handler::DONT_CALL))    ACE_ERROR ((LM_ERROR,                ACE_TEXT ("%p/n"),                ACE_TEXT ("test_for_spin, remove pipe")));  if (0 == reactor.remove_handler (handler.other_pipe_.write_handle (),                                   ACE_Event_Handler::ALL_EVENTS_MASK |                                   ACE_Event_Handler::DONT_CALL))    ACE_ERROR ((LM_ERROR,                ACE_TEXT ("test_for_spin remove other_pipe succeeded ")                ACE_TEXT ("but shouldn't/n")));}

voidTAO_Thread_Lane_Resources::shutdown_reactor (void){    TAO_Leader_Follower &leader_follower = this->leader_follower ();    ACE_GUARD (TAO_SYNCH_MUTEX,               ace_mon,               leader_follower.lock ());    ACE_Reactor *reactor = leader_follower.reactor ();    // Wakeup all the threads waiting blocked in the event loop, this    // does not guarantee that they will all go away, but reduces the    // load on the POA....    // If there are some client threads running we have to wait until    // they finish, when the last one does it will shutdown the reactor    // for us.  Meanwhile no new requests will be accepted because the    // POA will not process them.    if (!this->orb_core_.resource_factory ()->drop_replies_during_shutdown () &&            leader_follower.has_clients ())    {        reactor->wakeup_all_threads ();    }    else    {        // End the reactor if we want shutdown dropping replies along the        // way.        reactor->end_reactor_event_loop ();    }}

intServer_i::enable_multicast (const char *ior){  if (this->parse_args (this->argc_, this->argv_) != 0)    return -1;  // Get reactor instance from TAO.  ACE_Reactor *reactor =    this->orb_->orb_core ()->reactor ();  // Instantiate a handler which will handle client requests for the  // bootstrappable service, received on the multicast port.  ACE_NEW_RETURN (this->ior_multicast_,                  TAO_IOR_Multicast (),                  -1);  if (this->ior_multicast_->init (ior,                                  this->mcast_address_.in (),                                  TAO_SERVICEID_MCASTSERVER) == -1)    return -1;  // Register event handler for the ior multicast.  if (reactor->register_handler (this->ior_multicast_,                                 ACE_Event_Handler::READ_MASK) == -1)    {      if (TAO_debug_level > 0)        ACE_DEBUG ((LM_DEBUG,                    "MCast_Server: cannot register Event handler/n"));      return -1;    }  return 0;}

intrun_main (int, ACE_TCHAR *[]){  ACE_START_TEST (ACE_TEXT ("Timer_Cancellation_Test"));  ACE_Reactor reactor (new ACE_TP_Reactor,                       1);  Deadlock deadlock;  deadlock.reactor (&reactor);  Event_Handler handler (deadlock);  // Scheduler a timer to kick things off.  reactor.schedule_timer (&handler,                          0,                          ACE_Time_Value (1));  // Run the event loop for a while.  ACE_Time_Value timeout (4);  reactor.run_reactor_event_loop (timeout);  ACE_END_TEST;  return 0;}

static void *dispatch (void *arg){  // every thread must register the same stream to write to file  if (out_stream)    {      ACE_LOG_MSG->set_flags (ACE_Log_Msg::OSTREAM);      ACE_LOG_MSG->msg_ostream (out_stream);    }  ACE_DEBUG ((LM_DEBUG, ACE_TEXT (" (%t) Dispatcher Thread started!/n")));  ACE_Reactor *r = reinterpret_cast <ACE_Reactor *> (arg);  int result;  r->owner (ACE_OS::thr_self ());  while (1)    {      result = r->handle_events ();      if (result <= 0)        ACE_DEBUG ((LM_DEBUG,                    ACE_TEXT ("Dispatch: handle_events (): %d"),                    result));    }  ACE_NOTREACHED (return 0);}

/** * This is the function run by all threads in the thread pool. * * @param arg is expected to be of type (ACE_Reactor *) */ACE_THR_FUNC_RETURN threadFunc(void *arg) {    ACE_TRACE("threadFunc(void *)");    ACE_Reactor *reactor = (ACE_Reactor *) arg;    reactor->run_reactor_event_loop();    return 0;}

voidClient_Peer::crash(void){  Crasher * crasher = new Crasher;  ACE_Time_Value clk_tck (0, Clock_Ticks::get_usecs_per_tick ());  ACE_Reactor * reactor = this->orb_->orb_core()->reactor();  reactor->schedule_timer(crasher, 0, clk_tck);}

voidLocator_Repository::teardown_multicast (){  ACE_Reactor* r = ior_multicast_.reactor ();  if (r != 0) {    r->remove_handler (&ior_multicast_, ACE_Event_Handler::READ_MASK);    ior_multicast_.reactor (0);  }}

CORBA::BooleanImR_Activator_i::kill_server (const char* name, CORBA::Long lastpid, CORBA::Short signum){  if (debug_ > 1)    ORBSVCS_DEBUG((LM_DEBUG,                   "ImR Activator: Killing server <%C>, lastpid = %d/n",                   name, lastpid));  pid_t lpid = static_cast<pid_t>(lastpid);  pid_t pid = 0;  bool found = false;  int result = -1;  for (ProcessMap::iterator iter = process_map_.begin();       !found && iter != process_map_.end (); iter++)    {      if (iter->item () == name)        {          pid = iter->key ();          found = pid == lpid;        }    }  if (!found && pid == 0)    {      pid = lpid;    }#if defined (ACE_WIN32)  found = false; // sigchild apparently doesn't work on windows#endif  if (pid != 0)    {      result =#if !defined (ACE_WIN32)        (signum != 9) ? ACE_OS::kill (pid, signum) :#endif        ACE::terminate_process (pid);      if (this->running_server_list_.remove (name) == 0)        {          this->dying_server_list_.insert (name);        }      if (debug_ > 1)        ORBSVCS_DEBUG((LM_DEBUG,                       "ImR Activator: Killing server <%C> "                       "signal %d to pid %d, found %d, this->notify_imr_ %d,  result = %d/n",                       name, signum, static_cast<int> (pid), found, this->notify_imr_, result));      if (!found && result == 0 && this->notify_imr_)        {          this->process_map_.bind (pid, name);          ACE_Reactor *r = this->orb_->orb_core()->reactor();          Act_token_type token = static_cast<Act_token_type>(pid);          r->schedule_timer (this, reinterpret_cast<void *>(token), ACE_Time_Value ());        }    }  return result == 0;}

int MyTask::svc(){    ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%t) enter MyTask::svc()/n/n")));    //initialization    selector = new ACE_Select_Reactor;    ACE_Reactor* reactor = new ACE_Reactor(selector);    ACE_Reactor::instance(reactor);  //then, ACE_Reactor::instance() is reactor    this->reactor(reactor);  //this reactor_ of ACE_Event_Handler can not be set    //register socket handler    Handler handler(UDP_PORT);    if (reactor->register_handler(&handler, ACE_Event_Handler::READ_MASK) == -1)    {        ACE_ERROR((LM_ERROR, "%p/n", "cant't register with Reactor in MyTask::svc()/n"));        return -1;    }    //spawn mytask2    mytask2::instance()->open(&handler);    //handle_events in forever-loop until receive two data packets from socket, then, it will notify the MY_EXIT_HANDLER    while (exit_flag == 0)    {        int result = reactor->handle_events();  //timeout will not occur at all        if (result)        {            ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%t) handle_events() succeed, result = %d/n/n"), result));            if (RECV_COUNT == handler.get_count())            {                ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%t) MyTask::svc() notify the exit handler/n")));                ACE_Reactor::instance()->notify(&handler, ACE_Event_Handler::EXCEPT_MASK);            }            if (handler.get_flag())                exit_flag = 1;        }        else        {            ACE_ERROR((LM_ERROR, "%p/n", "handle_events() failed/n"));            return -1;        }    }    if (reactor->remove_handler(&handler, ACE_Event_Handler::READ_MASK |                                ACE_Event_Handler::DONT_CALL) == -1)    {        ACE_ERROR((LM_ERROR, "%p/n", "cant't remove handler from Reactor/n"));        return -1;    }    delete reactor;    reactor = NULL;    ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%t) exit MyTask::svc()/n/n")));    return 0;}

static void *run_reactor (void *pReactor){    ACE_Reactor *pLogReactor = (ACE_Reactor *)pReactor;    pLogReactor->owner(ACE_Thread_Manager::instance ()->thr_self());    pLogReactor->run_reactor_event_loop ();    ACE_DEBUG((LM_INFO, ACE_TEXT("(%P|%t) %M run_reactor exit[%N,%l]/n")));    return 0;}

    ~ServerInstance()        {            udpReactor.end_reactor_event_loop();            tcpReactor.end_reactor_event_loop();            if(tcp_thread >= 0)                ACE_Thread_Manager::instance ()->wait_grp(tcp_thread);            ACE_Thread_Manager::instance ()->wait_grp(udp_thread);        }