static void
t_rput(queue_t *q, mblk_t *mp)
{
	int err = EOPNOTSUPP;

	trace();
	if (q->q_count && mp->b_datap->db_type < QPCTL) {
		putq(q, mp);
		return;
	}
	switch (mp->b_datap->db_type) {
	case M_DATA:
		if ((err = t_m_data(q, mp)))
			break;
		return;
	case M_CTL:
	case M_PROTO:
	case M_PCPROTO:
		if ((err = t_m_proto(q, mp)))
			break;
		return;
	}
	switch (err) {
	case EAGAIN:
		putq(q, mp);
		return;
	case EOPNOTSUPP:
		if (q->q_next) {
			putnext(q, mp);
			return;
		}
	}
	freemsg(mp);
}

/*
 * This is the main mouse input routine.  Commands and parameters
 * from upstream are sent to the mouse device immediately, unless
 * the mouse is in the process of being reset, in which case
 * commands are queued and executed later in the service procedure.
 */
static int
mouse8042_wput(queue_t *q, mblk_t *mp)
{
	struct mouse_state *state;
	state = (struct mouse_state *)q->q_ptr;

	/*
	 * Process all messages immediately, unless a reset is in
	 * progress.  If a reset is in progress, deflect processing to
	 * the service procedure.
	 */
	if (state->reset_state != MSE_RESET_IDLE)
		return (putq(q, mp));

	/*
	 * If there are still messages outstanding in the queue that
	 * the service procedure hasn't processed yet, put this
	 * message in the queue also, to ensure proper message
	 * ordering.
	 */
	if (q->q_first)
		return (putq(q, mp));

	(void) mouse8042_process_msg(q, mp, state);

	return (0);
}

void output_double(double d)
{
    long n;
    long i;

    if (d < 0.0) {
        d = 0.0 - d;
        putchar('-');
    }

    if (d > 9220000000000000000.0) {
        output_double_binary(d);
        return;
    }

    if (d < 0.0000001) {
        output_double_binary(d);
        return;
    }

    n = (long)d;
    putq(n);
    d = d - (double)n;

    putchar('.');

    for (i = 0; i < 16; i = i + 1) {
        d = d * 10.0;
        n = (long)d;
        putq(n);
        d = d - (double)n;
    }

    return;
}

/*
 *  -------------------------------------------------------------------------
 *
 *  WRITE QUEUE PUT and SRV routines
 *
 *  -------------------------------------------------------------------------
 *
 *  SSCOP WPUT - Message from above.
 *
 *  If the message is priority message we attempt to process it immediately.
 *  If the message is non-priority message, but there are no messages on the
 *  queue yet, we attempt to process it immediately.  If the message is not
 *  supported, we pass it down-queue if possible.  If the message cannot be
 *  processed immediately, we place it on the queue.
 */
STATIC int
sscop_wput(queue_t *q, mblk_t *mp)
{
	mblk_t *mp;
	int err = -EOPNOTSUPP;

	if (q->q_count && mp->b_datap->db_type < QPCTL) {
		putq(q, mp);
		/* 
		 *  NOTE:- after placing messages on the queue here, I should
		 *  check for transmit congestion.  I should check if placing
		 *  the message on the queue crosses a band threshold for
		 *  congestion onset and abatement.  When crossing congestion
		 *  thresholds, I should notify MTP3.
		 */
		return (0);
	}
	switch (mp->b_datap->db_type) {
	case M_DATA:
		if ((err = sscop_w_data(q, mp)))
			break;
		return (0);
	case M_PROTO:
		if ((err = sscop_w_proto(q, mp)))
			break;
		return (0);
	case M_PCPROTO:
		if ((err = sscop_w_pcproto(q, mp)))
			break;
		return (0);
	case M_CTL:
		if ((err = sscop_w_ctl(q, mp)))
			break;
		return (0);
	case M_IOCTL:
		if ((err = sscop_w_ioctl(q, mp)))
			break;
		return (0);
	case M_FLUSH:
		sscop_w_flush(q, mp);
		return (0);
	}
	switch (err) {
	case -EAGAIN:
		if (mp->b_datap->db_type < QPCTL) {
			putq(q, mp);
			return (0);
		}
	case -EOPNOTSUPP:
		if (q->q_next) {
			putnext(q, mp);
			return (0);
		}
	}
	freemsg(mp);
	return (err);
}

/*
 * Through message handle for read side stream
 *
 * Requires Lock (( M: Mandatory, P: Prohibited, A: Allowed ))
 *  -. uinst_t->lock   : M [RW_READER]
 *  -. uinst_t->u_lock : A
 *  -. uinst_t->l_lock : P
 *  -. uinst_t->c_lock : P
 */
int
oplmsu_rcmn_through_hndl(queue_t *q, mblk_t *mp, int pri_flag)
{
	lpath_t	*lpath;
	ctrl_t	*ctrl;
	queue_t	*dst_queue = NULL;
	int	act_flag;

	ASSERT(RW_READ_HELD(&oplmsu_uinst->lock));

	mutex_enter(&oplmsu_uinst->l_lock);
	lpath = (lpath_t *)q->q_ptr;
	if (lpath->uinst != NULL) {
		act_flag = ACTIVE_RES;
	} else {
		act_flag = NOT_ACTIVE_RES;
	}
	mutex_exit(&oplmsu_uinst->l_lock);

	mutex_enter(&oplmsu_uinst->c_lock);
	if (((ctrl = oplmsu_uinst->user_ctrl) != NULL) &&
	    (((mp->b_datap->db_type == M_IOCACK) ||
	    (mp->b_datap->db_type == M_IOCNAK)) || (act_flag == ACTIVE_RES))) {
		dst_queue = RD(ctrl->queue);
	} else {
		mutex_exit(&oplmsu_uinst->c_lock);
		freemsg(mp);
		return (SUCCESS);
	}

	if (pri_flag == MSU_HIGH) {
		putq(dst_queue, mp);
	} else {
		if (canput(dst_queue)) {
			putq(dst_queue, mp);
		} else {
			/*
			 * Place a normal priority message at the head of
			 * read queue
			 */

			ctrl = (ctrl_t *)dst_queue->q_ptr;
			ctrl->lrq_flag = 1;
			ctrl->lrq_queue = q;
			mutex_exit(&oplmsu_uinst->c_lock);
			putbq(q, mp);
			return (FAILURE);
		}
	}
	mutex_exit(&oplmsu_uinst->c_lock);
	return (SUCCESS);
}

/*
 *  =========================================================================
 *
 *  STREAMS QUEUE PUT and QUEUE SERVICE routines
 *
 *  =========================================================================
 *
 *  READ QUEUE PUT and SRV routines
 *
 *  -------------------------------------------------------------------------
 *
 *  SSCOP RPUT - Message from below.
 *
 *  If the message is a priority message we attempt to process it immediately.
 *  If the message is a non-priority message, but there are no messages on the
 *  queue yet, we attempt to process it immediately.  If the message is not
 *  supported, we pass it up-queue if possible.  If the message cannot be
 *  processed immediately we place it on the queue.
 */
STATIC int
sscop_rput(queue_t *q, mblk_t *mp)
{
	int err = -EOPNOTSUPP;

	if (mp->b_datap->db_type < QPCTL && q->q_count) {
		putq(q, mp);
		/* 
		 *  NOTE:- after placing messages on the queue here, I should
		 *  check if placing the message on the queue crosses a band
		 *  threshold for congestion accept and congestion discard.
		 *  When crossing congestion accept, I should sent busy to the
		 *  peer and notify MTP3.  When crossing congestion discard I
		 *  should notify MTP3.
		 */
		return (0);
	}
	switch (mp->b_datap->db_type) {
	case M_DATA:
		if ((err = sscop_r_data(q, mp)))
			break;
		return (0);
	case M_PROTO:
		if ((err = sscop_r_proto(q, mp)))
			break;
		return (0);
	case M_PCPROTO:
		if ((err = sscop_r_pcproto(q, mp)))
			break;
		return (0);
	case M_CTL:
		if ((err = sscop_r_ctl(q, mp)))
			break;
		return (0);
	case M_ERROR:
		sscop_r_error(q, mp);
		return (0);
	}
	switch (err) {
	case -EAGAIN:
		putq(q, mp);
		return (0);
	case -EOPNOTSUPP:
		if (q->q_next) {
			putnext(q, mp);
			return (0);
		}
	}
	freemsg(mp);
	return (err);
}

/*
 * wput(9E) is symmetric for master and slave sides, so this handles both
 * without splitting the codepath.
 *
 * zc_wput() looks at the other side; if there is no process holding that
 * side open, it frees the message.  This prevents processes from hanging
 * if no one is holding open the console.  Otherwise, it putnext's high
 * priority messages, putnext's normal messages if possible, and otherwise
 * enqueues the messages; in the case that something is enqueued, wsrv(9E)
 * will take care of eventually shuttling I/O to the other side.
 */
static void
zc_wput(queue_t *qp, mblk_t *mp)
{
	unsigned char type = mp->b_datap->db_type;

	ASSERT(qp->q_ptr);

	DBG1("entering zc_wput, %s side", zc_side(qp));

	if (zc_switch(RD(qp)) == NULL) {
		DBG1("wput to %s side (no one listening)", zc_side(qp));
		switch (type) {
		case M_FLUSH:
			handle_mflush(qp, mp);
			break;
		case M_IOCTL:
			miocnak(qp, mp, 0, 0);
			break;
		default:
			freemsg(mp);
			break;
		}
		return;
	}

	if (type >= QPCTL) {
		DBG1("(hipri) wput, %s side", zc_side(qp));
		switch (type) {
		case M_READ:		/* supposedly from ldterm? */
			DBG("zc_wput: tossing M_READ\n");
			freemsg(mp);
			break;
		case M_FLUSH:
			handle_mflush(qp, mp);
			break;
		default:
			/*
			 * Put this to the other side.
			 */
			ASSERT(zc_switch(RD(qp)) != NULL);
			putnext(zc_switch(RD(qp)), mp);
			break;
		}
		DBG1("done (hipri) wput, %s side", zc_side(qp));
		return;
	}

	/*
	 * Only putnext if there isn't already something in the queue.
	 * otherwise things would wind up out of order.
	 */
	if (qp->q_first == NULL && bcanputnext(RD(zc_switch(qp)), mp->b_band)) {
		DBG("wput: putting message to other side\n");
		putnext(RD(zc_switch(qp)), mp);
	} else {
		DBG("wput: putting msg onto queue\n");
		(void) putq(qp, mp);
	}
	DBG1("done wput, %s side", zc_side(qp));
}

int ftpc_send_msg4 (u_long type, 
                    u_long pio, 
                    u_char * arg1p, 
                    u_long arg1len,
                    u_char * arg2p, 
                    u_long arg2len,
                    u_char * arg3p, 
                    u_long arg3len)
{
   struct ftpctask_msg * msgp;
   unsigned char * startp;

   msgp = (struct ftpctask_msg *) FTPC_ALLOC (sizeof (struct ftpctask_msg) + arg1len + arg2len + arg3len);
   if (!msgp)
   {
      ++ftpc_err.alloc_fail;
      return -1;
   }
   
   msgp->type = type;
   msgp->pio = pio;
   startp = &(msgp->parms[0]);
   memcpy (startp, arg1p, arg1len);
   memcpy (startp + arg1len, arg2p, arg2len);
   memcpy (startp + arg1len + arg2len, arg3p, arg3len);

   /* send message to FTP client task */
   LOCK_NET_RESOURCE (FTPCQ_RESID);
   putq(&ftpcq, (q_elt)msgp);
   UNLOCK_NET_RESOURCE (FTPCQ_RESID);

   post_app_sem (FTPC_SEMID);

   return 0;
}

static mblk_t * simulate_latency(RtpSession *session, mblk_t *input){
	OrtpNetworkSimulatorCtx *sim=session->net_sim_ctx;
	struct timeval current;
	mblk_t *output=NULL;
	uint32_t current_ts;
	ortp_gettimeofday(&current,NULL);
	/*since we must store expiration date in reserved2(32bits) only(reserved1
	already used), we need to reduce time stamp to milliseconds only*/
	current_ts = 1000*current.tv_sec + current.tv_usec/1000;

	/*queue the packet - store expiration timestamps in reserved fields*/
	if (input){
		input->reserved2 = current_ts + sim->params.latency;
		putq(&sim->latency_q,input);
	}

	if ((output=peekq(&sim->latency_q))!=NULL){
		if (TIME_IS_NEWER_THAN(current_ts, output->reserved2)){
			output->reserved2=0;
			getq(&sim->latency_q);
			/*return the first dequeued packet*/
			return output;
		}
	}

	return NULL;
}

static streamscall int
zap_rput(queue_t *q, mblk_t *mp)
{
	if ((!pcmsg(DB_TYPE(mp)) && (q->q_first || (q->q_flag & QSVCBUSY))) || zap_r_msg(q, mp))
		putq(q, mp);
	return (0);
}

void rtp_session_add_contributing_source(RtpSession *session, uint32_t csrc,
	const char *cname, const char *name, const char *email, const char *phone,
	const char *loc, const char *tool, const char *note) {
	mblk_t *chunk = sdes_chunk_new(csrc);
	sdes_chunk_set_full_items(chunk, cname, name, email, phone, loc, tool, note);
	putq(&session->contributing_sources, chunk);
}

void Chk_User_Control(int type)
/****************************************************************/
{
	//int i= 0;
	int pcm = 0; 
	int pno = 0;
	point_info point;
	
	switch(type) {
		case USER_CONTROL_ONOFF: 	  pcm = GHP_ONOFF_PCM; 			break;	
		case USER_CONTROL_MODE:  	  pcm = GHP_MODE_PCM; 			break;				
		case USER_CONTROL_SETTEMP: 	  pcm = GHP_SET_TEMP_PCM; 		break;	
		case USER_CONTROL_SPEED: 	  pcm = GHP_WINDSPEED_PCM; 		break;	
		case USER_CONTROL_DIRECTION:  pcm = GHP_WINDDIRECTION_PCM; 	break;	
		default: return;									
	}	
		
	for(pno = 0; pno < GHP_UNIT_MAX; pno++)	{

		if(prePtbl[pcm][pno] != g_fExPtbl[pcm][pno]) {

			if(g_dbgShow) printf("Change Point = %d,%d (%f, %f)\n", 
				pcm, 
				pno, 
				prePtbl[pcm][pno], 
				g_fExPtbl[pcm][pno]);
			prePtbl[pcm][pno] = g_fExPtbl[pcm][pno];
			
			point.pcm = pcm;
			point.pno = pno;
			point.value = g_fExPtbl[pcm][pno];
			putq(&ghp_message_queue, &point); 
		}
	}
}

int HDCCUSvrHandler::handle_input(ACE_HANDLE fd)
{
    ACE_Message_Block * mb;
    ACE_NEW_RETURN(mb,ACE_Message_Block(MAX_MESBUF_LEN),0);
    // read data
    ACE_INT32 n = 0;
    ACE_INT32 m = 0;
    while( (n = peer().recv(mb->wr_ptr(),mb->size() - m)) >= 0 )
    {
        mb->wr_ptr(n);
    }
    if(mb->length() <= 0)
    {
        mb->release();
        return -1;
    }

    // 放入队列
    if(putq(mb) == -1)
    {
        ACE_DEBUG((LM_ERROR,"保存失败"));
        return -1;
    }
    _close_time = 1;
    // 线程已经启动,返回 0 表示可以继续处理事件
    REACTOR::instance()->remove_handler(this,
                                        ACE_Event_Handler::READ_MASK|ACE_Event_Handler::DONT_CALL|
                                        ACE_Event_Handler::WRITE_MASK);
    activate(THR_NEW_LWP|THR_JOINABLE,1);
    //ACE_Time_Value reschedule(_max_timeout_sec.sec()/2);
    REACTOR::instance()->schedule_timer(this,NULL,_max_timeout_sec);
    return 0;
}

/*
 * Flush handle for write side stream
 *
 * Requires Lock (( M: Mandatory, P: Prohibited, A: Allowed ))
 *  -. uinst_t->lock   : M [RW_READER or RW_WRITER]
 *  -. uinst_t->u_lock : P
 *  -. uinst_t->l_lock : P
 *  -. uinst_t->c_lock : P
 */
void
oplmsu_wcmn_flush_hndl(queue_t *q, mblk_t *mp, krw_t rw)
{
	queue_t	*dst_queue = NULL;

	ASSERT(RW_LOCK_HELD(&oplmsu_uinst->lock));

	if (*mp->b_rptr & FLUSHW) {	/* Write side */
		flushq(q, FLUSHDATA);
	}

	dst_queue = oplmsu_uinst->lower_queue;
	if (dst_queue == NULL) {
		if (*mp->b_rptr & FLUSHR) {
			flushq(RD(q), FLUSHDATA);
			*mp->b_rptr &= ~FLUSHW;

			rw_exit(&oplmsu_uinst->lock);
			OPLMSU_TRACE(q, mp, MSU_TRC_UO);
			qreply(q, mp);
			rw_enter(&oplmsu_uinst->lock, rw);
		} else {
			freemsg(mp);
		}
	} else {
		putq(WR(dst_queue), mp);
	}
}

static OSStatus au_read_cb (
							  void                        *inRefCon,
							  AudioUnitRenderActionFlags  *ioActionFlags,
							  const AudioTimeStamp        *inTimeStamp,
							  UInt32                      inBusNumber,
							  UInt32                      inNumberFrames,
							  AudioBufferList             *ioData
)
{
	AUData *d=(AUData*)inRefCon;
	if (d->readTimeStamp.mSampleTime <0) {
		d->readTimeStamp=*inTimeStamp;
	}
	OSStatus err=0;
	mblk_t * rm=NULL;
	if (d->read_started) {
		rm=allocb(ioData->mBuffers[0].mDataByteSize,0);
		ioData->mBuffers[0].mData=rm->b_wptr;
	}
	err = AudioUnitRender(d->io_unit, ioActionFlags, &d->readTimeStamp, inBusNumber,inNumberFrames, ioData);
	if (d->read_started){
		if (err == 0) {
			rm->b_wptr += ioData->mBuffers[0].mDataByteSize;
			ms_mutex_lock(&d->mutex);
			putq(&d->rq,rm);
			ms_mutex_unlock(&d->mutex);
			d->readTimeStamp.mSampleTime+=ioData->mBuffers[0].mDataByteSize/(d->bits/2);
		}else ms_warning("AudioUnitRender() failed: %i",err);
	}
	return err;
}

HRESULT ( Callback)(IMediaSample* pSample, REFERENCE_TIME* sTime, REFERENCE_TIME* eTime, BOOL changed)
{
	BYTE *byte_buf=NULL;
	mblk_t *buf;

	V4wState *s = s_callback;
	if (s==NULL)
		return S_OK;

	HRESULT hr = pSample->GetPointer(&byte_buf);
	if (FAILED(hr))
	{
		return S_OK;
	}

	int size = pSample->GetActualDataLength();
	if (size>+1000)
	{
		buf=allocb(size,0);
		memcpy(buf->b_wptr, byte_buf, size);
		buf->b_wptr+=size;  

		ms_mutex_lock(&s->mutex);
		putq(&s->rq, buf);
		ms_mutex_unlock(&s->mutex);

	}
	return S_OK;
}

static void
q_drain(struct queue * qsrc, struct queue * qdest)
{
   while(qsrc->q_len)
   {
      putq(qdest, getq(qsrc));
   }
}

static void
t_wput(queue_t *q, mblk_t *mp)
{
	trace();
	if (mp->b_datap->db_type < QPCTL && (q->q_count || !canputnext(q)))
		putq(q, mp);
	else
		putnext(q, mp);
}

static void *msv4l2_thread(void *ptr){
	V4l2State *s=(V4l2State*)ptr;
	int try=0;
	
	ms_message("msv4l2_thread starting");
	if (s->fd==-1){
		if( msv4l2_open(s)!=0){
			ms_warning("msv4l2 could not be openned");
			goto close;
		}
	}
	
	if (!s->configured && msv4l2_configure(s)!=0){
		ms_warning("msv4l2 could not be configured");		
		goto close;
	}
	if (msv4l2_do_mmap(s)!=0)
	{
		ms_warning("msv4l2 do mmap");
		goto close;
	}
	ms_message("V4L2 video capture started.");
	while(s->thread_run)
	{
		if (s->fd!=-1){
			mblk_t *m;
			m=v4lv2_grab_image(s,50);
			if (m){
				mblk_t *om=dupmsg(m);
				mblk_set_marker_info(om,(s->pix_fmt==MS_MJPEG));
				ms_mutex_lock(&s->mutex);
				putq(&s->rq,om);
				ms_mutex_unlock(&s->mutex);
			}
		}
	}
	/*dequeue pending buffers so that we can properly unref them (avoids memleak )*/
	while(s->queued && try<10){
		v4l2_dequeue_ready_buffer(s,50);
		try++;
	}
	if (try==10) ms_warning("msv4l2: buffers not dequeued at exit !");
	msv4l2_do_munmap(s);
close:
	msv4l2_close(s);
	ms_message("msv4l2_thread exited.");
	ms_thread_exit(NULL);
	return NULL;
}

static void msv4l2_preprocess(MSFilter *f){
	V4l2State *s=(V4l2State*)f->data;
	s->thread_run=TRUE;
	ms_thread_create(&s->thread,NULL,msv4l2_thread,s);
	s->th_frame_count=-1;
	s->mean_inter_frame=0;
}

static void size_conv_process(MSFilter *f){
	SizeConvState *s=(SizeConvState*)f->data;
	YuvBuf inbuf;
	mblk_t *im;
	int cur_frame;

	ms_filter_lock(f);

	if (s->frame_count==-1){
		s->start_time=(float)f->ticker->time;
		s->frame_count=0;
	}
	while((im=ms_queue_get(f->inputs[0]))!=NULL ){
		putq(&s->rq, im);
	}

	cur_frame=(int)((f->ticker->time-s->start_time)*s->fps/1000.0);
	if (cur_frame<=s->frame_count && s->fps>=0) {
		/* too much frame */
		while(s->rq.q_mcount>1){
			ms_message("MSSizeConv: extra frame removed.");
			im=getq(&s->rq);
			freemsg(im);
		}
		ms_filter_unlock(f);
		return;
	}

	if (cur_frame>s->frame_count && s->fps>=0) {
		/*keep the most recent frame if several frames have been captured */
		while(s->rq.q_mcount>1){
			ms_message("MSSizeConv: extra frame removed.");
			im=getq(&s->rq);
			freemsg(im);
		}
	}
	while((im=getq(&s->rq))!=NULL ){
		if (ms_yuv_buf_init_from_mblk(&inbuf,im)==0){
			if (inbuf.w==s->target_vsize.width &&
				inbuf.h==s->target_vsize.height){
				ms_queue_put(f->outputs[0],im);
			}else{
				struct ms_SwsContext *sws_ctx=get_resampler(s,inbuf.w,inbuf.h);
				mblk_t *om=size_conv_alloc_mblk(s);
				if (ms_sws_scale(sws_ctx,inbuf.planes,inbuf.strides, 0,
					inbuf.h, s->outbuf.planes, s->outbuf.strides)<0){
					ms_error("MSSizeConv: error in ms_sws_scale().");
				}
				ms_queue_put(f->outputs[0],om);
				freemsg(im);
			}
			s->frame_count++;
		}else freemsg(im);
	}

	ms_filter_unlock(f);
}