int
Echo_Handler::dispatch_echo_checks (int first_call)
{
  // Set ones , if this is the first call (not from handle_timeout)
  if (first_call)
    {
      for (size_t j = 0; j < this->number_remotes_; ++j)
        {
          this->success_status_[j] = 1;
        }
      this->current_attempt_ = this->max_attempts_num_;
    }

  // Send echo-checks.
  for (size_t i = 0; i < this->number_remotes_; ++i)
    {
      if (this->success_status_[i] != 0)
        {
          if (this->ping_socket ().send_echo_check (
                this->remote_addrs_[i],
                this->connect_to_remote_) == -1)
            ACE_ERROR
              ((LM_ERROR,
                ACE_TEXT ("(%P|%t) Echo_Handler::dispatch_echo_checks - ")
                ACE_TEXT ("failed for this->remote_addrs_[%d].\n"),
                i));
        }
    }

  int rval_sched = -1;
  if ((rval_sched =
         this->reactor ()->schedule_timer (this,
                                           0,
                                           ACE_Time_Value (1),
                                           this->reply_wait_)) == -1)
    ACE_ERROR_RETURN ((LM_ERROR,
                       ACE_TEXT ("(%P|%t) Echo_Handler::dispatch_echo_checks:")
                       ACE_TEXT (" %p\n"),
                       ACE_TEXT ("schedule_timer")),
                      -1);
  return 0;
}

int
ACE_TMAIN (int argc, ACE_TCHAR* argv[])
{
  try
  {
    if (argc < 2) throw args ();

    ACE_INET_Addr addr (argv[1]);

    //FUZZ: disable check_for_lack_ACE_OS
    // Turn on message loss and reordering simulation.
    //
    ACE_RMCast::Socket socket (addr, false, true);
    //FUZZ: enable check_for_lack_ACE_OS

    Message msg;
    msg.sn = 0;

    for (unsigned short i = 0; i < payload_size; i++)
    {
      msg.payload[i] = i;
    }

    for (; msg.sn < message_count; msg.sn++)
    {
      socket.send (&msg, sizeof (msg));
    }

    // Keep running in case retransmissions are needed.
    //
    ACE_OS::sleep (ACE_Time_Value (60, 0));

    return 0;
  }
  catch (args const&)
  {
    ACE_ERROR ((LM_ERROR,
                "usage: %s <IPv4 multicast address>:<port>\n", argv[0]));
  }

  return 1;
}

int
DllOrb::fini (void)
{
  try
  {
    mv_poaManager_->deactivate (1, 1);
    mv_poaManager_ = PortableServer::POAManager::_nil ();

    // attempt to protect against sporadic BAD_INV_ORDER exceptions
    ACE_OS::sleep (ACE_Time_Value (0, 500));

      mv_rootPOA_->destroy (1, 1);

      mv_rootPOA_ = PortableServer::POA::_nil ();
    mv_orb_->shutdown (1);

    ACE_DEBUG ((LM_ERROR, ACE_TEXT ("wait() ...\n")));
    // wait for our threads to finish
    wait();
    ACE_DEBUG ((LM_ERROR, ACE_TEXT ("wait() done\n")));

    ACE_auto_ptr_reset (ma_barrier_, static_cast<ACE_Thread_Barrier *> (0));
  }
  catch (...)
  {
    ACE_DEBUG ((LM_ERROR, ACE_TEXT ("ERROR: exception\n")));
    return -1;
  }

  try
  {
    mv_orb_->destroy ();
    mv_orb_ = CORBA::ORB::_nil ();
  }
  catch (const CORBA::Exception& ex)
  {
    ex._tao_print_exception ("Exception caught:");
    return -1;
  }

  return 0;
}

  void
  Connection::write(Message& message)
  {
    ACE_Time_Value av(ACE_OS::gettimeofday() + ACE_Time_Value(1));

    if (writeMutex.acquire(av) == -1)
      throw Miro::CException(errno, "Error writing can device.");

    // Roland: Is this sleep necessary for Sparrow2003 can ???

    ACE_Time_Value time = ACE_OS::gettimeofday();
    ACE_Time_Value delta = time - lastWrite;
    if (delta < canTimeOut) {
      // is this sleep necessary ???
      // well, yes

      ACE_OS::sleep(canTimeOut - delta); // this is at least 10usec thanks to linux
      time = ACE_OS::gettimeofday();
    }

    // will definitely choke if base is off
    int rc;
    if (parameters_.module == "pcan") {
       pcanmsg * msgp;
       message.canMessage((int **) &msgp);
       msgp->msg.msgtype = MSGTYPE_EXTENDED;
       for(int i = 0; i < msgp->msg.len; i++)
          std::cout << msgp->msg.data[i];
       rc = ioctl(ioBuffer.get_handle(), PCAN_WRITE_MSG, msgp);
    }
    else {
       canmsg * msg;
       message.canMessage((int **) &msg);
       rc = ioBuffer.send_n(msg, sizeof(canmsg));
    }

    lastWrite = time;

    writeMutex.release();
    if (rc == -1)
      throw Miro::CException(errno, "Error writing can device.");
  }

  virtual int handle_timeout (const ACE_Time_Value &,
                              const void *arg)
  {
    ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Test_Handler::handle_timeout\n")));
    ++this->nr_expirations_;

    void *nc_arg = const_cast<void *> (arg);
    Thread_Timer_Queue *timer_queue =
      reinterpret_cast<Thread_Timer_Queue *> (nc_arg);

    ACE_Time_Value timeout = ACE_OS::gettimeofday () + ACE_Time_Value (1, 0);

    ACE_DEBUG ((LM_DEBUG,
                ACE_TEXT ("(%t) scheduling new timer 1 sec from now\n")));
    if (timer_queue->schedule (this, timer_queue, timeout) == -1)
      ACE_ERROR ((LM_ERROR, ACE_TEXT ("(%t) %p\n"),
                  ACE_TEXT ("schedule failed")));

    return 0;
  }

int AC_Output_Handler::svc () {
  ACE_Message_Block *chunk[ACE_IOV_MAX];
  size_t message_index = 0;
  ACE_Time_Value time_of_last_send (ACE_OS::gettimeofday ());
  ACE_Time_Value timeout;
  ACE_Sig_Action no_sigpipe ((ACE_SignalHandler) SIG_IGN);
  ACE_Sig_Action original_action;
  no_sigpipe.register_action (SIGPIPE, &original_action);

  for (;;) {
    if (message_index == 0) {
      timeout = ACE_OS::gettimeofday ();
      timeout += FLUSH_TIMEOUT;
    }
    ACE_Message_Block *mblk = 0;
    if (getq (mblk, &timeout) == -1) {
      if (errno == ESHUTDOWN) {
        if (connector_->reconnect () == -1) break;
        continue;
      } else if (errno != EWOULDBLOCK) break;
      else if (message_index == 0) continue;
    } else {
      if (mblk->size () == 0
          && mblk->msg_type () == ACE_Message_Block::MB_STOP)
        { mblk->release (); break; }
      chunk[message_index] = mblk;
      ++message_index;
    }
    if (message_index >= ACE_IOV_MAX ||
        (ACE_OS::gettimeofday () - time_of_last_send
         >= ACE_Time_Value(FLUSH_TIMEOUT))) {
      if (this->send (chunk, message_index) == -1) break;
      time_of_last_send = ACE_OS::gettimeofday ();
    }
  }

  if (message_index > 0)
    this->send (chunk, message_index);
  no_sigpipe.restore_action (SIGPIPE, original_action);
  return 0;
}

 void
 Receiver_exec_i::ccm_activate (void)
 {
   ACE_DEBUG ((LM_DEBUG, "Receiver_exec_i::ccm_activate \n "));
   /// No need to implement anything
   /// Start the timer .
    if (this->reactor ()->schedule_timer(
                this->hello_generator_,  // The callback object
                0,                       // No arguments for this callback
                ACE_Time_Value (1, 0),   // Initial delay
                this->interval_) == -1)  // Time interval
      {
        ACE_ERROR ((LM_ERROR, "Receiver_exec_i::ccm_activate - "
                              "Error scheduling timer\n"));
      }
    else
      {
        ACE_DEBUG ((LM_DEBUG, "Receiver_exec_i::ccm_activate - "
                              "Hello generator is running\n"));
      }
  }

TAO_BEGIN_VERSIONED_NAMESPACE_DECL

// Constructor.
TAO_Time_Service_Clerk::TAO_Time_Service_Clerk (int timer_value,
                                                int timer_value_usecs,
                                                const IORS& servers)
  : server_ (servers),
    helper_ (this)
{
  // Schedule the helper to be invoked by the reactor
  // periodically.

  if (TAO_ORB_Core_instance ()->reactor ()->schedule_timer
      (&helper_,
       0,
       ACE_Time_Value::zero,
       ACE_Time_Value(timer_value,timer_value_usecs)) == -1)
    ORBSVCS_ERROR ((LM_ERROR,
                "%p\n",
                "schedule_timer ()"));
}

int reuseAddr_test (void)
{
# if defined ACE_WIN32 || !defined ACE_LACKS_IOCTL
  ACE_SOCK_Dgram_Bcast sock1;
  if (sock1.open(ACE_INET_Addr(UDP_PORT),PF_INET,0,1) != 0)
  {
    ACE_ERROR ((LM_ERROR, ACE_TEXT ("Could not open socket for broadcast on port %d\n"), UDP_PORT ));
    ++result;
  }
  else
  {
    // Keep the socket open for some time
    ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Opened socket on port %d and keep it open for %d sec\n"), UDP_PORT,  SEC_TO_KEEP_SOCKET_OPEN));

    ACE_OS::sleep(ACE_Time_Value(SEC_TO_KEEP_SOCKET_OPEN,0));

    sock1.close();
  }
# endif
  return result;
}

int main (int, char* [])
{
    ACE_Reactor reactor;
    Time_Handler* th = new Time_Handler;
    int timer_id[NUMBER_TIMERS];
    for (int i = 0; i < NUMBER_TIMERS; i++)
	{
        timer_id[i] = reactor.schedule_timer(th,
                                              (const void *)i, // argument sent to handle_timeout()
                                              ACE_Time_Value(2 * i + 1)); //set timer to go off with delay
	}
	
    //Cancel the fifth timer before it goes off
    reactor.cancel_timer(timer_id[5]);//Timer ID of timer to be removed
    while (!done)
	{
        reactor.handle_events();
	}

    return 0;
}

 int svc( ) 
 {
   if ( m_sleep == ACE_Time_Value( 0 ) ) {
     printf( "###### TIMEBOMB Disabled. ######\n" );
     fflush( stdout );
     return 0;
   }
   ACE_Time_Value start = ACE_OS::gettimeofday();
   ACE_Time_Value now = ACE_OS::gettimeofday();
   do {
     ACE_OS::sleep( 1 );
     now = ACE_OS::gettimeofday();
   } while( now - start < m_sleep );
   printf( "####### ERROR: TIMEBOMB WENT OFF, TEST TIMED OUT ########\n" );
   fflush( stdout );
   
   callClientCleanup();
   
   exit( -1 );
   return 0;
 }

int CLogManager::PutLog(_LogBlockInfo* pLogBlockInfo)
{
    ACE_Message_Block* mb = pLogBlockInfo->GetQueueMessage();

    //如果正在重新加载
    if(m_blIsNeedReset == true)
    {
        //回收日志块
        m_objLogBlockPool.ReturnBlockInfo(pLogBlockInfo);
        return 0;
    }

    if(mb)
    {
        int msgcount = (int)msg_queue()->message_count();

        if (msgcount >= m_nQueueMax)
        {
            OUR_DEBUG((LM_INFO,"[CLogManager::PutLog] CLogManager queue is full!\n"));
            //回收日志块
            m_objLogBlockPool.ReturnBlockInfo(pLogBlockInfo);
            return -1;
        }

        ACE_Time_Value xtime = ACE_OS::gettimeofday()+ACE_Time_Value(0, MAX_MSG_PUTTIMEOUT);

        if(this->putq(mb, &xtime) == -1)
        {
            OUR_DEBUG((LM_ERROR,"[CLogManager::PutLog] CLogManager putq error(%s)!\n", pLogBlockInfo->m_pBlock));
            //回收日志块
            m_objLogBlockPool.ReturnBlockInfo(pLogBlockInfo);
            return -1;
        }

        return 0;
    }

    OUR_DEBUG((LM_ERROR,"[CLogManager::PutLog] CLogManager new ACE_Message_Block error!\n"));
    return -1;
}

int
Time_Handler::handle_timeout (const ACE_Time_Value &tv,
                              const void *arg)
{
  long current_count = static_cast<long> (reinterpret_cast<size_t> (arg));
  if (current_count >= 0)
    ACE_TEST_ASSERT (current_count == the_count);

  ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("[%x] Timer id %d with count #%d|%d timed out at %d!\n"),
              this,
              this->timer_id (),
              the_count,
              current_count,
              tv.sec ()));

  if (current_count == long (ACE_MAX_TIMERS - 1))
    done = 1;
  else if (the_count == long (ACE_MAX_TIMERS - 1))
    {
      done = 1;
      return -1;
    }
  else if (current_count == -1)
    {
#if defined (ACE_HAS_CPP11)
      int result = ACE_Reactor::instance ()->reset_timer_interval (this->timer_id (),
                                                                   std::chrono::seconds {the_count + 1});
#else
      int result = ACE_Reactor::instance ()->reset_timer_interval (this->timer_id (),
                                                                   ACE_Time_Value (the_count + 1));
#endif
      if (result == -1)
        ACE_ERROR ((LM_ERROR,
                    ACE_TEXT ("Error resetting timer interval\n")));
    }
  the_count += (1 + odd);
  return 0;
}

static void
test_registering_all_handlers (void)
{
  ACE_Trace t (ACE_TEXT ("test_registering_all_handler"),
               __LINE__,
               ACE_TEXT_CHAR_TO_TCHAR (__FILE__));
  Time_Handler rt[ACE_MAX_TIMERS];
  long t_id[ACE_MAX_TIMERS];

  for (size_t i = 0; i < ACE_MAX_TIMERS; i++)
    {
      t_id[i] =
        ACE_Reactor::instance ()->schedule_timer (&rt[i],
                                                  (const void *) i,
                                                  ACE_Time_Value (2 * i + 1));
      ACE_TEST_ASSERT (t_id[i] != -1);
      rt[i].timer_id (t_id[i]);
    }

  while (!done)
    ACE_Reactor::instance ()->handle_events ();
}

CORBA::Long
Simple_Server_i::test_method (CORBA::Boolean do_callback)
{
  if (do_callback)
    {
      // Scheduling ourselves for doing the callbacks
      if (TAO_debug_level > 0)
      ACE_DEBUG ((LM_DEBUG,
                  ACE_TEXT ("Scheduling for callback\n")));

      if (this->orb_->orb_core ()->reactor ()->schedule_timer (this,
                                                               0,
                                                               ACE_Time_Value (1)) == -1)
        {
          ACE_ERROR ((LM_ERROR,
                      "(%P|%t) Can't schedule timers, aborting ..\n"));

          ACE_OS::abort ();
        }
    }

  return 0;
}

/* Closing the thread pool can be a tricky exercise.  I've decided to
   take an easy approach and simply enqueue a secret message for each
   thread we have active.  */
int
Thread_Pool::close (u_long flags)
{
    ACE_UNUSED_ARG(flags);

    /* Find out how many threads are currently active */
    int counter = active_threads_.value ();

    /* For each one of the active threads, enqueue a "null" event
      handler.  Below, we'll teach our svc() method that "null" means
      "shutdown".  */
    while (counter--)
        this->enqueue (0);

    /* As each svc() method exits, it will decrement the active thread
      counter.  We just wait here for it to reach zero.  Since we don't
      know how long it will take, we sleep for a quarter of a second
      between tries.  */
    while (active_threads_.value ())
        ACE_OS::sleep (ACE_Time_Value (0, 250000));

    return(0);
}

int
Purger_Thread::svc (void)
{
  ACE_DEBUG ((LM_DEBUG,
    ACE_TEXT("(%t) Purger_Thread::svc commencing\n")));

  disable_signal (SIGPIPE, SIGPIPE);

  for (; !this->reactor_.reactor_event_loop_done ();)
    {
      // Get a connection from the cache.
      Sender *sender =
        this->connection_cache_.acquire_connection ();

      // If no connection is available in the cache, sleep for a while.
      if (sender == 0)
        ACE_OS::sleep (ACE_Time_Value (0, 10 * 1000));
      else
        {
          // The reference count on the sender was increased by the
          // cache before it was returned to us.
          ACE_Event_Handler_var safe_sender (sender);

          // Actively close the connection.
          ACE_DEBUG ((LM_DEBUG,
                      ACE_TEXT ("(%t) Purger thread calling Sender::close() ")
                      ACE_TEXT ("for handle %d\n"),
                      sender->handle_));

          sender->close ();
        }
    }
  ACE_DEBUG ((LM_DEBUG,
    ACE_TEXT("(%t) Purger_Thread::svc terminating\n")));

  return 0;
}

bool
MyMain::init_server (const ACE_TCHAR* args)
{
  std::string my_args (ACE_TEXT_ALWAYS_CHAR(args));
  // main thread and extra thread for backdoor operations
  int thread_pool = 2;

  ACE_auto_ptr_reset (server_task_, new Server_Task (my_args));
  ACE_ASSERT (server_task_.get() != 0);

  server_task_->activate (THR_NEW_LWP | THR_JOINABLE |THR_INHERIT_SCHED
                          , thread_pool);

  int duration = 4; //wait 3 seconds for initialization
  ACE_Time_Value current = ACE_High_Res_Timer::gettimeofday_hr ();
  ACE_Time_Value timeout = current + ACE_Time_Value(duration);

  while (current < timeout) {
    if (server_task_->ready()) {
      break;
    }
    ACE_Time_Value sleep_time;
    sleep_time.msec (10);
    ACE_OS::sleep (sleep_time);
    current += sleep_time;
  }

  if (!server_task_->ready()) {
    server_task_->force_shutdown ();
    server_task_->wait ();
    ACE_auto_ptr_reset (server_task_, (Server_Task*)0);
    return false;
  }

  return true;
}

int KSG_Task_Queue_Pool::wait_for_queue(const std::string &key,long time_out)
{
	queue_cond_map_type::const_iterator i = _queue_cond.find(key);
	if(i == _queue_cond.end())
	{
		ACE_DEBUG((LM_ERROR,"没有对应的队列[%s]",key.c_str()));
		return -1;
	}
	Queue_List_Cond *cond = i->second;
	ACE_DEBUG((LM_DEBUG,"等待队列组[%s]...",key.c_str()));
	int ret;
	if(time_out > 0)
	{
		ACE_Time_Value tv = ACE_Time_Value(time_out/1000,time_out%1000);
		tv += ACE_OS::gettimeofday();
		ret = cond->wait(&tv);
		// maybe timeout 
		if( ret == -1 && ACE_OS::last_error() == ETIME )
			ret = 1;
	}
	else
		ret = cond->wait(NULL);
	return ret;
}

static void
test_canceling_odd_timers (void)
{
  ACE_Trace t (ACE_TEXT ("test_canceling_odd_timers"),
               __LINE__,
               ACE_TEXT_CHAR_TO_TCHAR (__FILE__));
  Time_Handler rt[ACE_MAX_TIMERS];
  long t_id[ACE_MAX_TIMERS];

  done = 0;
  the_count = 1;
  odd = 1;

  for (size_t i = 0; i < ACE_MAX_TIMERS; i++)
    {
      t_id[i] = ACE_Reactor::instance ()->schedule_timer (&rt[i],
                                                          (const void *) i,
                                                          ACE_Time_Value (2 * i + 1));
      ACE_TEST_ASSERT (t_id[i] != -1);
      rt[i].timer_id (t_id[i]);
    }

  for (size_t j = 0; (u_long) j < ACE_MAX_TIMERS; j++)
    // Cancel handlers with odd numbered timer ids.
    if (ACE_ODD (rt[j].timer_id ()))
      {
        int result =
          ACE_Reactor::instance ()->cancel_timer (rt[j].timer_id ());
        if (result == -1)
          ACE_ERROR ((LM_ERROR,
                      ACE_TEXT ("Error cancelling timer\n")));
      }

  while (!done)
    ACE_Reactor::instance ()->handle_events ();
}