static int sock_arg(conn_t *conn)
{
	if (conn->ctx.numargs < 1)
		return 0;

	/* CANCEL <name> [<cmd>] */
	if (!strcmp(conn->ctx.arglist[0], "CANCEL")) {

		if (conn->ctx.numargs < 3)
			cancel_timer(conn->ctx.arglist[1], NULL);
		else
			cancel_timer(conn->ctx.arglist[1], conn->ctx.arglist[2]);

		send_to_one(conn, "OK\n");
		return 1;
	}

	if (conn->ctx.numargs < 3)
		return 0;

	/* START <name> <length> */
	if (!strcmp(conn->ctx.arglist[0], "START")) {
		start_timer(conn->ctx.arglist[1], conn->ctx.arglist[2]);
		send_to_one(conn, "OK\n");
		return 1;
	}

	/* unknown */
	return 0;
}

int zmq::stream_engine_t::decode_and_push (msg_t *msg_)
{
    zmq_assert (mechanism != NULL);

    if (mechanism->decode (msg_) == -1)
        return -1;

    if(has_timeout_timer) {
        has_timeout_timer = false;
        cancel_timer(heartbeat_timeout_timer_id);
    }

    if(has_ttl_timer) {
        has_ttl_timer = false;
        cancel_timer(heartbeat_ttl_timer_id);
    }

    if(msg_->flags() & msg_t::command) {
        uint8_t cmd_id = *((uint8_t*)msg_->data());
        if(cmd_id == 4)
            process_heartbeat_message(msg_);
    }

    if (metadata)
        msg_->set_metadata (metadata);
    if (session->push_msg (msg_) == -1) {
        if (errno == EAGAIN)
            process_msg = &stream_engine_t::push_one_then_decode_and_push;
        return -1;
    }
    return 0;
}

void zmq::session_t::proceed_with_term ()
{
    if (state == terminating)
        return;

    zmq_assert (state == pending);
    state = terminating;

    //  If there's still a pending linger timer, remove it.
    if (has_linger_timer) {
        cancel_timer (linger_timer_id);
        has_linger_timer = false;
    }

    if (in_pipe) {
        register_term_acks (1);
        in_pipe->terminate ();
    }
    if (out_pipe) {
        register_term_acks (1);
        out_pipe->terminate ();
    }

    //  The session has already waited for the linger period. We don't want
    //  the child objects to linger any more thus linger is set to zero.
    own_t::process_term (0);
}

/* Start hibernating by starting a timer and waiting for it. */
void hiber_start (uint4 hiber)
{
	int4		waitover;
	int4		*waitover_addr;
	TID		tid;
	sigset_t	savemask;

	if (1 <= timer_stack_count)	/* timer services are unavailable from within a timer handler */
		GTMASSERT;
	sigprocmask(SIG_BLOCK, &blockalrm, &savemask);	/* block SIGALRM signal */
	waitover = FALSE;		/* when OUR timer pops, it will set this flag */
	waitover_addr = &waitover;
	tid = (TID)waitover_addr;	/* unique id of this timer */
	start_timer_int((TID)tid, hiber, hiber_wake, SIZEOF(waitover_addr), &waitover_addr);
	/* we will loop here until OUR timer pops and sets OUR flag */
	do
	{
		sigsuspend(&savemask);	/* unblock SIGALRM and wait for timer interrupt */
		if (outofband)
		{
                        cancel_timer(tid);
			break;
		}
	} while(FALSE == waitover);
	sigprocmask(SIG_SETMASK, &savemask, NULL);	/* reset signal handlers */
}

/*!	Sets the alarm timer for the current thread. The timer fires at the
	specified time in the future, periodically or just once, as determined
	by \a mode.
	\return the time left until a previous set alarm would have fired.
*/
bigtime_t
set_alarm(bigtime_t time, uint32 mode)
{
	struct thread *thread = thread_get_current_thread();
	bigtime_t remainingTime = 0;

	ASSERT(B_ONE_SHOT_RELATIVE_ALARM == B_ONE_SHOT_RELATIVE_TIMER);
		// just to be sure no one changes the headers some day

	TRACE(("set_alarm: thread = %p\n", thread));

	if (thread->alarm.period)
		remainingTime = (bigtime_t)thread->alarm.schedule_time - system_time();

	cancel_timer(&thread->alarm);

	if (time != B_INFINITE_TIMEOUT)
		add_timer(&thread->alarm, &alarm_event, time, mode);
	else {
		// this marks the alarm as canceled (for returning the remaining time)
		thread->alarm.period = 0;
	}

	return remainingTime;
}

void zmq::session_base_t::reconnect ()
{
    //  For delayed connect situations, terminate the pipe
    //  and reestablish later on
    if (pipe && options.immediate == 1
        && addr->protocol != "pgm" && addr->protocol != "epgm"
        && addr->protocol != "norm" && addr->protocol != "udp") {
        pipe->hiccup ();
        pipe->terminate (false);
        terminating_pipes.insert (pipe);
        pipe = NULL;

        if (has_linger_timer) {
            cancel_timer (linger_timer_id);
            has_linger_timer = false;
        }
    }

    reset ();

    //  Reconnect.
    if (options.reconnect_ivl != -1)
        start_connecting (true);

    //  For subscriber sockets we hiccup the inbound pipe, which will cause
    //  the socket object to resend all the subscriptions.
    if (pipe && (options.type == ZMQ_SUB || options.type == ZMQ_XSUB || options.type == ZMQ_DISH))
        pipe->hiccup ();
}

void RUDPSocket::reset()
{
    if(state_ != RUDP_IDLE)
    {
        RUDP_DEBUG("state = RUDP_IDLE, rudp id = " << rudp_id_);
        state_ = RUDP_IDLE; 
    }

    rudp_id_ = -1;
    local_index_ = INVALID_ADAPTER_INDEX;
    local_title_ = 0;

    remote_rudp_id_ = INVALID_RUDP_HANDLE;

    heart_ts_ = CBaseTimeValue::get_time_value().msec();
    keeplive_intnal_ = DEFAULT_KEEPLIVE;
    keeplive_count_ = 0;
    timeout_count_ = DEFAULT_TIMEOUT_COUNT;

    event_handler_ = NULL;

    error_code_ = RUDP_SUCCESS;

    send_count_ = 0;

    cancel_timer();

    ccc_.reset();

    send_buffer_.reset();
    recv_buffer_.reset();

    check_sum_ = 0;
}

    void start_request(const void* data, size_t len)
    {
        tcp::endpoint ep(boost::asio::ip::address_v4::from_string(address_), port_);
        socket_.open(boost::asio::ip::tcp::v4());

        do_timeout_check(connect_timeout_);
        socket_.async_connect(ep, [this,data,len](const boost::system::error_code& ec){

            cancel_timer();
            if(!ec)
            {
                // The connection was successful. Send the request.
                connected_ = true;
                do_timeout_check(request_timeout_);
                boost::asio::async_write(socket_, 
                    boost::asio::buffer(data, len), 
                    [this](const boost::system::error_code& ec, size_t) {
                        handle_write_request(ec);
                });
            }
            else {
                handle_error(ec);
            }
        });

        if(!service_running_)
        {
            service_running_ = true;
            boost::system::error_code ignored_ec;
            io_service_.reset();
            io_service_.run(ignored_ec);
            this->socket_.close(ignored_ec);
            service_running_ = false;
        }
    }

    void handle_read_content(const boost::system::error_code& ec, size_t bytes_transferred)
    {
        if (!ec)
        {
            // write buffer to response data
            response_buffer_->insert(response_buffer_->end(), buffer_, buffer_ + bytes_transferred);

            this->total_bytes_read_ += bytes_transferred;

            if(this->total_bytes_read_ < this->expecting_size_)
            {
                // Continue reading remaining data until EOF.
                size_t size = (std::min)(sizeof(buffer_), this->expecting_size_ - this->total_bytes_read_);
                boost::asio::async_read(socket_, 
                    boost::asio::buffer(buffer_, size ),
                    boost::asio::transfer_exactly( size ), [this](const boost::system::error_code& ec, size_t bytes_transferred){
                        handle_read_content(ec, bytes_transferred);
                });
            }
            else { // complete
                this->error_code_ = ec.value();
                cancel_timer();
            }
        }
        else
        {
            handle_error(ec);
        }
    }

    void handle_read_length(const boost::system::error_code& ec)
    {
        if (!ec)
        {
            expecting_size_ = ntohl(expecting_size_);
            if(expecting_size_ > this->max_data_len_)
            { // too long response received
                cancel_timer();
                return;
            }

            this->total_bytes_read_ = 0;

            // Start reading remaining data until EOF.
            size_t size = (std::min)(sizeof(buffer_), this->expecting_size_);
            boost::asio::async_read(socket_, 
                boost::asio::buffer(buffer_, size ),
                boost::asio::transfer_exactly( size ), [this](const boost::system::error_code& ec, size_t bytes_transferred){
                    handle_read_content(ec, bytes_transferred);
            });
        }
        else
        {
            handle_error(ec);
        }
    }

/*
 * Check if and when lwIP has its next timeout, and set or cancel our timer
 * accordingly.
 */
static void
set_lwip_timer(void)
{
	uint32_t next_timeout;
	clock_t ticks;

	/* Ask lwIP when the next alarm is supposed to go off, if any. */
	next_timeout = sys_timeouts_sleeptime();

	/*
	 * Set or update the lwIP timer.  We rely on set_timer() asking the
	 * kernel for an alarm only if the timeout is different from the one we
	 * gave it last time (if at all).  However, due to conversions between
	 * absolute and relative times, and the fact that we cannot guarantee
	 * that the uptime itself does not change while executing these
	 * routines, set_timer() will sometimes be issuing a kernel call even
	 * if the alarm has not changed.  Not a huge deal, but fixing this will
	 * require a different interface to lwIP and/or the timers library.
	 */
	if (next_timeout != (uint32_t)-1) {
		/*
		 * Round up the next timeout (which is in milliseconds) to the
		 * number of clock ticks to add to the current time.  Avoid any
		 * potential for overflows, no matter how unrealistic..
		 */
		if (next_timeout > TMRDIFF_MAX / sys_hz())
			ticks = TMRDIFF_MAX;
		else
			ticks = (next_timeout * sys_hz() + 999) / 1000;

		set_timer(&lwip_timer, ticks, expire_lwip_timer, 0 /*unused*/);
	} else
		cancel_timer(&lwip_timer);	/* not really needed.. */
}

void CConnection::reset()
{
	//清空到初始化状态
	sock_stream_.close();

	reactor_index_ = 0;

	state_ = CONN_IDLE;
	CORE_DEBUG("CConnection, state = CONN_IDLE, this = " << this);

	//取消定时器
	cancel_timer();

	timer_id_ = 0;

	sbuffer_.reset();
	rbuffer_.reset();

	index_ = 0;
	server_id_ = 0;
	server_type_ = 0;

	timer_count_ = 0;

	conn_ptr_ = NULL;

	send_flag_ = false;

	//释放太多的内存占用
	istrm_.rewind(true);
	istrm_.reduce();

	remote_addr_.set_ip(INADDR_ANY);
	remote_addr_.set_port(0);
}

/*!
*タイマ割り込みハンドラ
*/
void tmrdriver_intr(void)
{
	static int outp = 0;
	CYCCB *cycb = (CYCCB *)mg_timerque.tmrhead->rqobjp; /* 周期タイマ以外の場合は0となる */
	TMRCB *tbf;
	
	/* 差分のキューによるソフトタイマ */
  if (is_check_timer_expired(mg_timerque.index)) { /* タイマは満了したか */
		DEBUG_OUTMSG("softtimer is expire 1.\n");
		DEBUG_OUTVLE(outp, 0);
		DEBUG_OUTMSG("\n");
		outp++;
    cancel_timer(mg_timerque.index); /* タイマキャンセル処理 */
		next_tmrcb_diffque(); /* 差分のキューからタイマコントロールブロックの排除 */
		
		/* 周期タイマの時はタイマブロックを作成(createを呼ぶと先頭のタイマがかけられてしまうので排除してから呼ぶ) */
    if (cycb) {
    	tbf = (TMRCB *)create_tmrcb_diffque(OTHER_MAKE_TIMER, cycb->cyctim, (TMRRQ_OBJP)cycb, cycb->func, cycb->exinf); /* 差分のキューのノードを作成 */
    	cycb->tobjp = (TMR_OBJP)tbf;
    }
  }
	/* ハードタイマ */
	if (is_check_timer_expired(HARD_TIMER_DEFAULT_DEVICE)) { /* タイマは満了したか */
		DEBUG_OUTMSG("hardtimer id expire 0.\n");
		expire_timer(HARD_TIMER_DEFAULT_DEVICE); /* タイマ満了処理 */
	}
}

zmq::zmq_connecter_t::~zmq_connecter_t ()
{
    if (wait)
        cancel_timer (reconnect_timer_id);
    if (handle_valid)
        rm_fd (handle);
}

void gtcml_lkrundown(void)
{
	cm_region_list	*reg_walk;
	unsigned char	*ptr, laflag;
	unsigned short	top,len;
	uint4		status;

	cancel_timer((TID)curr_entry);	/* Cancel any outstanding lock starvation timer */
	reg_walk = curr_entry->region_root;
	curr_entry->state = 0;
	while (reg_walk)
	{
		reg_walk->reqnode = FALSE;
		if (reg_walk->lockdata)
		{
			if (reg_walk->blkd)
			{
				mlk_unpend(reg_walk->blkd);
				reg_walk->blkd = 0;
			}
			mlk_cm_root = reg_walk->lockdata;
			cm_cmd_lk_ct = 0;
			gtcml_unlock();
			gtcml_zdeallocate();
			reg_walk->lockdata = mlk_cm_root;
		}
		reg_walk = reg_walk->next;
	}
}

status_t
device_close(void *data)
{
	struct sis_info *info;

	if (checkDeviceInfo(info = data) != B_OK)
		return EINVAL;

	info->cookieMagic = SiS_FREE_COOKIE_MAGIC;

	// cancel timer
	cancel_timer(&info->timer);

	// remove & disable interrupt
	sis900_disableInterrupts(info);
	remove_io_interrupt_handler(info->pciInfo->u.h0.interrupt_line, sis900_interrupt, info);

	// disable the transmitter's and receiver's state machine
	write32(info->registers + SiS900_MAC_COMMAND,
		SiS900_MAC_CMD_Rx_DISABLE | SiS900_MAC_CMD_Tx_DISABLE);

	delete_sem(info->rxSem);
	delete_sem(info->txSem);

#ifdef EXCESSIVE_DEBUG
	delete_sem(gIOLock);
#endif

	return B_OK;
}

int ZCE_Timer_Heap::reschedule_timer(int timer_id, uint64_t now_trigger_msec)
{
    bool contiue_trigger = false;

    //计算下一次触发的点
    calc_next_trigger(timer_id,
                      now_trigger_msec,
                      contiue_trigger);

    //如果不需要继续触发定时器了，取消
    if (!contiue_trigger)
    {
        return cancel_timer(timer_id);
    }

    //这个TIMER NODE 仍然要触发

    //先从堆上上删除这个节点
    int ret = remove_nodeid(timer_id);

    if (ret != 0)
    {
        return ret;
    }

    //然后重新放入对中间去
    ret = add_nodeid(timer_id);

    if (ret != 0)
    {
        return ret;
    }

    return 0;
}

void ShortConnection::reset()
{
	//清空到初始化状态
	sock_stream_.close();

	reactor_index_ = 0;

	state_ = SHORT_CONN_IDLE;
	CORE_DEBUG("short conn, state = SHORT_CONN_IDLE, this = " << this);

	//取消定时器
	cancel_timer();

	timer_id_ = 0;

	sbuffer_.reset();
	rbuffer_.reset();

	timer_count_ = 0;

	//释放太多的内存占用
	istrm_.rewind(true);
	istrm_.reduce();

	remote_addr_.set_ip(INADDR_ANY);
	remote_addr_.set_port(0);

	msg_proc_ = NULL;
	conn_id_ = 0;
}

void zmq::zmq_connecter_t::process_unplug ()
{
    if (wait)
        cancel_timer ();
    if (handle_valid)
        rm_fd (handle);
}

void zmq::tcp_connecter_t::out_event ()
{
    if (connect_timer_started) {
        cancel_timer (connect_timer_id);
        connect_timer_started = false;
    }

    rm_handle ();

    const fd_t fd = connect ();

    //  Handle the error condition by attempt to reconnect.
    if (fd == retired_fd || !tune_socket (fd)) {
        close ();
        add_reconnect_timer ();
        return;
    }

    //  Create the engine object for this connection.
    stream_engine_t *engine =
      new (std::nothrow) stream_engine_t (fd, options, endpoint);
    alloc_assert (engine);

    //  Attach the engine to the corresponding session object.
    send_attach (session, engine);

    //  Shut the connecter down.
    terminate ();

    socket->event_connected (endpoint, (int) fd);
}