int LsapiConn::processRespHeader()
{
    register HttpExtConnector * pHEC = getConnector();
    int ret;
    int len = 0;
    if ( !pHEC )
        return -1;
    int &respState = pHEC->getRespState();
    if ( !(respState & 0xff) )
    {        
        while( m_iPacketLeft > 0 )
        {
            len = ExtConn::read( m_pRespHeader, m_pRespHeaderBufEnd - m_pRespHeader );
            if ( D_ENABLED( DL_MEDIUM ) )
                LOG_D(( getLogger(), "[%s] process response header %d bytes",
                    getLogId(), len ));
            if ( len > 0 )
            {
                m_iPacketLeft -= len;
                ret = processRespHeader( m_pRespHeader + len, respState ); 
                switch( ret )
                {
                case -2:
                    LOG_WARN(( getLogger(), "[%s] Invalid Http response header, retry!",
                            getLogId() ));
                    //debug code
                    //::write( 1, pBuf, len );
                    errno = ECONNRESET;
                case -1:
                    return -1;
                }
                
                if ( m_iPacketLeft > 0 )
                {
                    m_pRespHeader += len;
                    if (( m_pRespHeader > m_pRespHeaderProcess )&&
                        ( m_pRespHeader != &m_respBuf[ m_respInfo.m_cntHeaders * sizeof(short) ] ))
                    {
                        len = m_pRespHeader - m_pRespHeaderProcess;
                        memmove( &m_respBuf[ m_respInfo.m_cntHeaders * sizeof(short) ],
                                m_pRespHeaderProcess, m_pRespHeader - m_pRespHeaderProcess );
                        m_pRespHeaderProcess = &m_respBuf[ m_respInfo.m_cntHeaders * sizeof(short) ];
                        m_pRespHeader = m_pRespHeaderProcess + len;
                    }    
                    else
                        m_pRespHeader = m_pRespHeaderProcess = 
                                    &m_respBuf[ m_respInfo.m_cntHeaders * sizeof(short) ];
                    setRespBuf( m_pRespHeader );
                }
                
            }
            else
                return len;
        }
        if ( m_iPacketLeft == 0 )
        {
            m_iPacketHeaderLeft = LSAPI_PACKET_HEADER_LEN;
            len = 1;
        }
        return len;        
    }
    else
    {
        //error: protocol error, header received already.
		errno = EIO;
        return -1;
    }    
}

void _pa_open(void) {
	PaStreamParameters outputParameters;
	PaError err = paNoError;
	int device_id;

	if (pa.stream) {
		if ((err = Pa_CloseStream(pa.stream)) != paNoError) {
			LOG_WARN("error closing stream: %s", Pa_GetErrorText(err));
		}
	}

	if (output.state == OUTPUT_OFF) {
		// we get called when transitioning to OUTPUT_OFF to create the probe thread
		// set err to avoid opening device and logging messages
		err = 1;

	} else if ((device_id = pa_device_id(output.device)) == -1) {
		LOG_INFO("device %s not found", output.device);
		err = 1;

	} else {

		outputParameters.device = device_id;
		outputParameters.channelCount = 2;
		outputParameters.sampleFormat = paInt32;
#ifndef PA18API
		outputParameters.suggestedLatency =
			output.latency ? (double)output.latency/(double)1000 : Pa_GetDeviceInfo(outputParameters.device)->defaultHighOutputLatency;
		outputParameters.hostApiSpecificStreamInfo = NULL;
		
#endif
#if OSX && !defined(OSXPPC)
		// enable pro mode which aims to avoid resampling if possible
		// see http://code.google.com/p/squeezelite/issues/detail?id=11 & http://code.google.com/p/squeezelite/issues/detail?id=37
		// command line controls osx_playnice which is -1 if not specified, 0 or 1 - choose playnice if -1 or 1
		PaMacCoreStreamInfo macInfo;
		unsigned long streamInfoFlags;
	 	if (output.osx_playnice) {
			LOG_INFO("opening device in PlayNice mode");
			streamInfoFlags = paMacCorePlayNice;
		} else {
			LOG_INFO("opening device in Pro mode");
			streamInfoFlags = paMacCorePro;
		}
		PaMacCore_SetupStreamInfo(&macInfo, streamInfoFlags);
		outputParameters.hostApiSpecificStreamInfo = &macInfo;
#endif
	}

	if (!err &&
#ifndef PA18API
		(err = Pa_OpenStream(&pa.stream, NULL, &outputParameters, (double)output.current_sample_rate, paFramesPerBufferUnspecified,
							 paPrimeOutputBuffersUsingStreamCallback | paDitherOff, pa_callback, NULL)) != paNoError) {
#else
		(err = Pa_OpenStream(&pa.stream, paNoDevice, 0, 0, NULL, outputParameters.device, outputParameters.channelCount,
							outputParameters.sampleFormat, NULL, (double)output.current_sample_rate, paFramesPerBuffer,
							paNumberOfBuffers, paDitherOff, pa_callback, NULL)) != paNoError) {

#endif
		LOG_WARN("error opening device %i - %s : %s", outputParameters.device, Pa_GetDeviceInfo(outputParameters.device)->name, 
				 Pa_GetErrorText(err));
	}

	if (!err) {
#ifndef PA18API
		LOG_INFO("opened device %i - %s at %u latency %u ms", outputParameters.device, Pa_GetDeviceInfo(outputParameters.device)->name,
				 (unsigned int)Pa_GetStreamInfo(pa.stream)->sampleRate, (unsigned int)(Pa_GetStreamInfo(pa.stream)->outputLatency * 1000));
#else
		LOG_INFO("opened device %i - %s at %u fpb %u nbf %u", outputParameters.device, Pa_GetDeviceInfo(outputParameters.device)->name,
				 (unsigned int)output.current_sample_rate, paFramesPerBuffer, paNumberOfBuffers);

#endif
		pa.rate = output.current_sample_rate;

#ifndef PA18API
		if ((err = Pa_SetStreamFinishedCallback(pa.stream, pa_stream_finished)) != paNoError) {
			LOG_WARN("error setting finish callback: %s", Pa_GetErrorText(err));
		}
	
		UNLOCK; // StartStream can call pa_callback in a sychronised thread on freebsd, remove lock while it is called

#endif
		if ((err = Pa_StartStream(pa.stream)) != paNoError) {
			LOG_WARN("error starting stream: %s", Pa_GetErrorText(err));
		}

#ifndef PA18API
		LOCK;
#endif
	}

	if (err && !monitor_thread_running) {
		vis_stop();

		// create a thread to check for output state change or device return
#if LINUX || OSX || FREEBSD
		pthread_create(&monitor_thread, NULL, pa_monitor, NULL);
#endif
#if WIN
		monitor_thread = CreateThread(NULL, OUTPUT_THREAD_STACK_SIZE, (LPTHREAD_START_ROUTINE)&pa_monitor, NULL, 0, NULL);
//.........这里部分代码省略.........

template < typename IN_PORT_TYPE > int tagged_file_sink_b_base::_analyzerServiceFunction( typename  std::vector< gr_istream< IN_PORT_TYPE > > &istreams )
{
    typedef typename std::vector< gr_istream< IN_PORT_TYPE > > _IStreamList;

    boost::mutex::scoped_lock lock(serviceThreadLock);

    if ( validGRBlock() == false ) {
        // create our processing block
        createBlock();

        LOG_DEBUG( tagged_file_sink_b_base, " FINISHED BUILDING  GNU RADIO BLOCK");
    }
   
    // process any Stream ID changes this could affect number of io streams
    processStreamIdChanges();
    
    if ( !validGRBlock() || istreams.size() == 0 ) {
        LOG_WARN( tagged_file_sink_b_base, "NO STREAMS ATTACHED TO BLOCK..." );
        return NOOP;
    }

    // resize data vectors for passing data to GR_BLOCK object
    _input_ready.resize( istreams.size() );
    _ninput_items_required.resize( istreams.size());
    _ninput_items.resize( istreams.size());
    _input_items.resize(istreams.size());
    _output_items.resize(0);
  
    //
    // RESOLVE: need to look at forecast strategy, 
    //    1)  see how many read items are necessary for N number of outputs
    //    2)  read input data and see how much output we can produce
    //
  
    //
    // Grab available data from input streams
    //
    typename _IStreamList::iterator istream = istreams.begin();
    int nitems=0;
    for ( int idx=0 ; istream != istreams.end() && serviceThread->threadRunning() ; idx++, istream++ ) {
        // note this a blocking read that can cause deadlocks
        nitems = istream->read();

        if ( istream->overrun() ) {
            LOG_WARN( tagged_file_sink_b_base, " NOT KEEPING UP WITH STREAM ID:" << istream->streamID );
        }
    
        // RESOLVE issue when SRI changes that could affect the GNU Radio BLOCK
        if ( istream->sriChanged() ) {
            LOG_DEBUG( tagged_file_sink_b_base, "SRI CHANGED, STREAMD IDX/ID: " 
                   << idx << "/" << istream->pkt->streamID );
        }
    }

    LOG_TRACE( tagged_file_sink_b_base, "READ NITEMS: "  << nitems );
    if ( nitems <= 0 && !_istreams[0].eos() ) return NOOP;

    bool eos = false;
    int  nout = 0;
    while ( nout > -1 && serviceThread->threadRunning() ) {
        eos = false;
        nout = _forecastAndProcess( eos, istreams );
        if ( nout > -1  ) {
            // we chunked on data so move read pointer..
            istream = istreams.begin();
            for ( ; istream != istreams.end(); istream++ ) {

                int idx=std::distance( istreams.begin(), istream );
                // if we processed data for this stream
                if ( _input_ready[idx] ) {
                    size_t nitems = 0;
                    try {
                        nitems = gr_sptr->nitems_read( idx );
                    } catch(...){}
      
                    if ( nitems > istream->nitems() ) {
                        LOG_WARN( tagged_file_sink_b_base,  "WORK CONSUMED MORE DATA THAN AVAILABLE,  READ/AVAILABLE " 
                                 << nitems << "/" << istream->nitems() );
                        nitems = istream->nitems();
                    }
                    istream->consume( nitems );
                    LOG_TRACE( tagged_file_sink_b_base, " CONSUME READ DATA  ITEMS/REMAIN " << nitems << "/" << istream->nitems());
                }
            }
            gr_sptr->reset_read_index();
        }

        // check for not enough data return
        if ( nout == -1 ) {

            // check for  end of stream
            istream = istreams.begin();
            for ( ; istream != istreams.end() ; istream++) {
                if ( istream->eos() ) {
                    eos=true;
                }
            }

            if ( eos ) {
                LOG_TRACE( tagged_file_sink_b_base, " DATA NOT READY, EOS:" << eos );
//.........这里部分代码省略.........

int deinterleave_bb_2o_base::_transformerServiceFunction( std::vector< gr_istream_base * > &istreams ,
        std::vector< gr_ostream_base * > &ostreams  )
{
    typedef std::vector< gr_istream_base * >   _IStreamList;
    typedef std::vector< gr_ostream_base * >  _OStreamList;

    boost::mutex::scoped_lock lock(serviceThreadLock);

    if ( validGRBlock() == false ) {

        // create our processing block, and setup  property notifiers
        createBlock();

        LOG_DEBUG( deinterleave_bb_2o_base, " FINISHED BUILDING  GNU RADIO BLOCK");
    }

    //process any Stream ID changes this could affect number of io streams
    processStreamIdChanges();

    if ( !validGRBlock() || istreams.size() == 0 || ostreams.size() == 0  ) {
        LOG_WARN( deinterleave_bb_2o_base, "NO STREAMS ATTACHED TO BLOCK..." );
        return NOOP;
    }

    _input_ready.resize( istreams.size() );
    _ninput_items_required.resize( istreams.size() );
    _ninput_items.resize( istreams.size() );
    _input_items.resize( istreams.size() );
    _output_items.resize( ostreams.size() );

    //
    // RESOLVE: need to look at forecast strategy,
    //    1)  see how many read items are necessary for N number of outputs
    //    2)  read input data and see how much output we can produce
    //

    //
    // Grab available data from input streams
    //
    _OStreamList::iterator ostream;
    _IStreamList::iterator istream = istreams.begin();
    int nitems=0;
    for ( int idx=0 ; istream != istreams.end() && serviceThread->threadRunning() ; idx++, istream++ ) {
        // note this a blocking read that can cause deadlocks
        nitems = (*istream)->read();

        if ( (*istream)->overrun() ) {
            LOG_WARN( deinterleave_bb_2o_base, " NOT KEEPING UP WITH STREAM ID:" << (*istream)->streamID );
        }

        if ( (*istream)->sriChanged() ) {
            // RESOLVE - need to look at how SRI changes can affect Gnu Radio BLOCK state
            LOG_DEBUG( deinterleave_bb_2o_base, "SRI CHANGED, STREAMD IDX/ID: "
                       << idx << "/" << (*istream)->getPktStreamId() );
            setOutputStreamSRI( idx, (*istream)->getPktSri() );
        }
    }

    LOG_TRACE( deinterleave_bb_2o_base, "READ NITEMS: "  << nitems );
    if ( nitems <= 0 && !_istreams[0]->eos() ) {
        return NOOP;
    }

    bool eos = false;
    int  nout = 0;
    bool workDone = false;

    while ( nout > -1 && serviceThread->threadRunning() ) {
        eos = false;
        nout = _forecastAndProcess( eos, istreams, ostreams );
        if ( nout > -1  ) {
            workDone = true;

            // we chunked on data so move read pointer..
            istream = istreams.begin();
            for ( ; istream != istreams.end(); istream++ ) {
                int idx=std::distance( istreams.begin(), istream );
                // if we processed data for this stream
                if ( _input_ready[idx] ) {
                    size_t nitems = 0;
                    try {
                        nitems = gr_sptr->nitems_read( idx );
                    } catch(...) {}

                    if ( nitems > (*istream)->nitems() ) {
                        LOG_WARN( deinterleave_bb_2o_base,  "WORK CONSUMED MORE DATA THAN AVAILABLE,  READ/AVAILABLE "
                                  << nitems << "/" << (*istream)->nitems() );
                        nitems = (*istream)->nitems();
                    }
                    (*istream)->consume( nitems );
                    LOG_TRACE( deinterleave_bb_2o_base, " CONSUME READ DATA  ITEMS/REMAIN " << nitems << "/" << (*istream)->nitems());
                }
            }
            gr_sptr->reset_read_index();
        }

        // check for not enough data return
        if ( nout == -1 ) {

            // check for  end of stream
//.........这里部分代码省略.........

void RedChannel::run()
{
    for (;;) {
        Lock lock(_action_lock);
        if (_action == WAIT_ACTION) {
            _action_cond.wait(lock);
        }
        int action = _action;
        _action = WAIT_ACTION;
        lock.unlock();
        switch (action) {
        case CONNECT_ACTION:
            try {
                get_client().get_sync_info(get_type(), get_id(), _sync_info);
                on_connecting();
                set_state(CONNECTING_STATE);
                ConnectionOptions con_options(_client.get_connection_options(get_type()),
                                              _client.get_port(),
                                              _client.get_sport(),
                                              _client.get_protocol(),
                                              _client.get_host_auth_options(),
                                              _client.get_connection_ciphers());
                RedChannelBase::connect(con_options, _client.get_connection_id(),
                                        _client.get_host().c_str(),
                                        _client.get_password().c_str());
                /* If automatic protocol, remember the first connect protocol type */
                if (_client.get_protocol() == 0) {
                    if (get_peer_major() == 1) {
                        _client.set_protocol(1);
                    } else {
                        /* Major is 2 or unstable high value, use 2 */
                        _client.set_protocol(2);
                    }
                }
                /* Initialize when we know the remote major version */
                if (_client.get_peer_major() == 1) {
                    _marshallers = spice_message_marshallers_get1();
                } else {
                    _marshallers = spice_message_marshallers_get();
                }
                on_connect();
                set_state(CONNECTED_STATE);
                _loop.add_socket(*this);
                _socket_in_loop = true;
                on_event();
                _loop.run();
            } catch (RedPeer::DisconnectedException&) {
                _error = SPICEC_ERROR_CODE_SUCCESS;
            } catch (Exception& e) {
                LOG_WARN("%s", e.what());
                _error = e.get_error_code();
            } catch (std::exception& e) {
                LOG_WARN("%s", e.what());
                _error = SPICEC_ERROR_CODE_ERROR;
            }
            if (_socket_in_loop) {
                _socket_in_loop = false;
                _loop.remove_socket(*this);
            }
            if (_outgoing_message) {
                _outgoing_message->release();
                _outgoing_message = NULL;
            }
            _incomming_header_pos = 0;
            if (_incomming_message) {
                _incomming_message->unref();
                _incomming_message = NULL;
            }
        case DISCONNECT_ACTION:
            close();
            on_disconnect();
            set_state(DISCONNECTED_STATE);
            _client.on_channel_disconnected(*this);
            continue;
        case QUIT_ACTION:
            set_state(TERMINATED_STATE);
            return;
        }
    }
}

//.........这里部分代码省略.........
        case PCMSG_CHAT:
            handleChatMessage(computer, message);
            break;

        case PCMSG_ANNOUNCE:
            handleAnnounceMessage(computer, message);
            break;

        case PCMSG_PRIVMSG:
            handlePrivMsgMessage(computer, message);
            break;

        case PCMSG_WHO:
            handleWhoMessage(computer);
            break;

        case PCMSG_ENTER_CHANNEL:
            handleEnterChannelMessage(computer, message);
            break;

        case PCMSG_USER_MODE:
            handleModeChangeMessage(computer, message);
            break;

        case PCMSG_KICK_USER:
            handleKickUserMessage(computer, message);

        case PCMSG_QUIT_CHANNEL:
            handleQuitChannelMessage(computer, message);
            break;

        case PCMSG_LIST_CHANNELS:
            handleListChannelsMessage(computer, message);
            break;

        case PCMSG_LIST_CHANNELUSERS:
            handleListChannelUsersMessage(computer, message);
            break;

        case PCMSG_TOPIC_CHANGE:
            handleTopicChange(computer, message);
            break;

        case PCMSG_DISCONNECT:
            handleDisconnectMessage(computer, message);
            break;

        case PCMSG_GUILD_CREATE:
            handleGuildCreation(computer, message);
            break;

        case PCMSG_GUILD_INVITE:
            handleGuildInvitation(computer, message);
            break;

        case PCMSG_GUILD_ACCEPT:
            handleGuildAcceptInvite(computer, message);
            break;

        case PCMSG_GUILD_GET_MEMBERS:
            handleGuildRetrieveMembers(computer, message);
            break;

        case PCMSG_GUILD_PROMOTE_MEMBER:
            handleGuildMemberLevelChange(computer, message);
            break;

        case PCMSG_GUILD_KICK_MEMBER:
            handleGuildMemberKick(computer, message);

        case PCMSG_GUILD_QUIT:
            handleGuildQuit(computer, message);
            break;

        case PCMSG_PARTY_INVITE:
            handlePartyInvite(computer, message);
            break;

        case PCMSG_PARTY_ACCEPT_INVITE:
            handlePartyAcceptInvite(computer, message);
            break;

        case PCMSG_PARTY_QUIT:
            handlePartyQuit(computer);
            break;

        case PCMSG_PARTY_REJECT_INVITE:
            handlePartyRejection(computer, message);
            break;

        default:
            LOG_WARN("ChatHandler::processMessage, Invalid message type"
                     << message.getId());
            result.writeShort(XXMSG_INVALID);
            break;
    }

    if (result.getLength() > 0)
        computer.send(result);
}

    void
    readClusters(int n,
                 uint32_t pattern,
                 const ClusterLocation *loc,
                 bool verify = true,
                 int retries = 7)
    {
        uint64_t bufsz = dStore_->getClusterSize() * n;
        std::vector<uint8_t> buf(bufsz);
        //        uint64_t off;

        std::vector<ClusterReadDescriptor> descs;
        descs.reserve(n);

        // DataStoreNG::readClusters (more precisely, the partial backend read path)
        // asserts that it is not passed duplicate ClusterLocations. Intercept these
        // here and fail the test instead of taking down the whole process.
        std::set<std::string> locs;

        for (int i = 0; i < n; ++i)
        {
            const ClusterSize csize(dStore_->getClusterSize());
            ClusterLocationAndHash loc_and_hash(loc[i],
                                                &buf[0] + (i * csize),
                                                csize);
            BackendInterfacePtr bi = vol_->getBackendInterface(loc[i].cloneID())->clone();
            VERIFY(bi);
            descs.push_back(ClusterReadDescriptor(loc_and_hash,
                                                  i * csize,
                                                  &buf[i * csize],
                                                  std::move(bi)));

            const ClusterLocation& loc = loc_and_hash.clusterLocation;
            ASSERT_TRUE(locs.emplace(boost::lexical_cast<std::string>(loc)).second) <<
                "duplicate detected: " << loc_and_hash;
        }

        while (true)
        {
            try
            {
                dStore_->readClusters(descs);
            }
            catch (TransientException &)
            {
                LOG_WARN("backend congestion detected, retrying");
                if (--retries < 0)
                    throw;
                sleep(1);
                continue;
            }
            break;
        }

        if (verify)
        {
            for (uint32_t* p = (uint32_t *) buf.data(), off = 0; off < bufsz;
                 off += sizeof(*p), ++p)
            {
                ASSERT_EQ(pattern, *p) << "p: " << p <<
                    " off: " << off;
            }
        }
    }

//.........这里部分代码省略.........
        called each time the property value changes.  This is done by calling 
        setPropertyChangeListener(<property name>, this, &fastfilter_i::<callback method>)
        in the constructor.
            
        Example:
            // This example makes use of the following Properties:
            //  - A float value called scaleValue
            
        //Add to fastfilter.cpp
        fastfilter_i::fastfilter_i(const char *uuid, const char *label) :
            fastfilter_base(uuid, label)
        {
            setPropertyChangeListener("scaleValue", this, &fastfilter_i::scaleChanged);
        }

        void fastfilter_i::scaleChanged(const std::string& id){
            std::cout << "scaleChanged scaleValue " << scaleValue << std::endl;
        }
            
        //Add to fastfilter.h
        void scaleChanged(const std::string&);
        
        
************************************************************************************************/
int fastfilter_i::serviceFunction()
{
    bulkio::InFloatPort::dataTransfer *tmp = dataFloat_in->getPacket(bulkio::Const::BLOCKING);
	if (not tmp) { // No data is available
		return NOOP;
	}

	if (tmp->inputQueueFlushed)
	{
		LOG_WARN(fastfilter_i, "input queue flushed - data has been thrown on the floor.  flushing internal buffers");
		//flush all our processor states if the queue flushed
		boost::mutex::scoped_lock lock(filterLock_);
		for (map_type::iterator i = filters_.begin(); i!=filters_.end(); i++)
			i->second.filter->flush();
	}
	bool updateSRI = tmp->sriChanged;
    float fs = 1.0/tmp->SRI.xdelta;
	{
		boost::mutex::scoped_lock lock(filterLock_);
		map_type::iterator i = filters_.find(tmp->streamID);
		firfilter* filter;
		if (i==filters_.end())
		{
			//this is a new stream - need to create a new filter & wrapper
			LOG_DEBUG(fastfilter_i, "creating new filter for streamID "<<tmp->streamID);
			if (manualTaps_)
			{
				LOG_DEBUG(fastfilter_i, "using manual taps ");
				bool real, complex;
				getManualTaps(real,complex);
				if (real)
					filter = new firfilter(fftSize, realOut, complexOut, realTaps_);
				else if(complex)
					filter = new firfilter(fftSize, realOut, complexOut, complexTaps_);
				else
				{
					LOG_WARN(fastfilter_i, "state error - using manual taps with no filter provided.  This shouldn't really happen");
					if (updateSRI)
						dataFloat_out->pushSRI(tmp->SRI);
					dataFloat_out->pushPacket(tmp->dataBuffer, tmp->T, tmp->EOS, tmp->streamID);
					delete tmp;
					return NORMAL;

void *reader_thread(void *_arg)
{
	orc_t *orc = (orc_t*) _arg;

	setup_thread();

	orc->fd = -1;

	int reconnectcount = 0;

reconnect:
	
	// reconnect, if necessary.
	while (orc->fd < 0) {
		LOG_INFO("Trying to connect to orcboard...(%i)", reconnectcount++);
		orc->fd = orc->impl->connect(orc->impl);
		
		if (orc->fd < 0)
			sleep(1);
	}

	// read for a while
	while (1) {

		// read a packet
		uint8_t buf[3];
		int res = read_fully(orc->fd, buf, 1);
		if (res <= 0)
			goto disconnected;
		if (buf[0] != 0xED) {
			LOG_DEBUG("Recovering sync [%02X]", buf[0]);
			continue;
		}

		res = read_fully(orc->fd, &buf[1], 2);
		if (res <= 0)
			goto disconnected;

		int id = buf[1];
		int datalen = buf[2];

		transaction_t *t = &orc->transactions[id];
		memcpy(t->response, buf, 3);

		res = read_fully(orc->fd, &t->response[PACKET_DATA], datalen + 1);
		if (res <= 0)
			goto disconnected;
		if (!packet_test_checksum(t->response)) {
			LOG_WARN("Bad checksum received from Orc");
			continue;
		}

		if (t->response == garbage && t->response[1]>=orc->idLow && t->response[1]<=orc->idHigh)
			LOG_VERBOSE("Unsolicited ack, id = %02X", t->response[1]);

		// is this a message from orcd, assigning us a client id?
		if (t->response[1] == 0xf7) {
			orc->idLow = t->response[4];
			orc->idHigh = t->response[5];
			orc->idLast = orc->idLow;
			LOG_INFO("Got client transaction range: %02x-%02x", orc->idLow, orc->idHigh);
		}

		if (t->response[1] == 0xfe)
			handle_pad_packet(orc, t->response);

		if (t->response[1] == PACKET_ID_ORCBOARD_BROADCAST &&
		    packet_16u(t->response, 1) == MSG_ASYNC_HEARTBEAT)
			handle_heartbeat_packet(orc, t->response);

		pthread_mutex_lock(&t->mutex);
		pthread_cond_signal(&t->cond);
		pthread_mutex_unlock(&t->mutex);

	}
	
disconnected:
	orc->impl->disconnect(orc->impl, orc->fd);
	orc->fd = -1;
	goto reconnect;

	// silence compiler
	return NULL;
}

int HttpSvc::Sn::Handler_mediafile::parseRequest()
{
    int err = 0;

    // URI looks like /mediafile/tag/<deviceId>/<objectId>
    const std::string &uri = hs->GetUri();

    VPLHttp_SplitUri(uri, uri_tokens);
    if (uri_tokens.size() != 4) {
        LOG_ERROR("Handler_mediafile[%p]: Unexpected number of segments; uri %s", this, uri.c_str());
        std::ostringstream oss;
        oss << "{\"errMsg\":\"Unexpected number of segments; uri " << uri << "\"}";
        HttpStream_Helper::SetCompleteResponse(hs, 400, oss.str(), "application/json");
        return CCD_ERROR_PARAMETER;
    }
    objectId.assign(uri_tokens[3]);

    char buf[4096];
    char *reqBody = NULL;
    {
        ssize_t bytes = hs->Read(buf, sizeof(buf) - 1);  // subtract 1 for EOL
        if (bytes < 0) {
            LOG_ERROR("Handler_mediafile[%p]: Failed to read from HttpStream[%p]: err "FMT_ssize_t, this, hs, bytes);
            std::ostringstream oss;
            oss << "{\"errMsg\":\"Failed to read from HttpStream\"}";
            HttpStream_Helper::SetCompleteResponse(hs, 500, oss.str(), "application/json");
            return 0;
        }
        buf[bytes] = '\0';
        char *boundary = strstr(buf, "\r\n\r\n");  // find header-body boundary
        if (!boundary) {
            LOG_ERROR("Handler_mediafile[%p]: Failed to find header-body boundary in request", this);
            std::ostringstream oss;
            oss << "{\"errMsg\":\"Failed to find header-body boundary in request\"}";
            HttpStream_Helper::SetCompleteResponse(hs, 400, oss.str(), "application/json");
            return 0;
        }
        reqBody = boundary + 4;
    }

    cJSON2 *json = cJSON2_Parse(reqBody);
    if (!json) {
        LOG_ERROR("Handler_mediafile[%p]: Failed to parse JSON in request body %s", this, reqBody);
        std::ostringstream oss;
        oss << "{\"errMsg\":\"Failed to parse JSON in request body\"}";
        HttpStream_Helper::SetCompleteResponse(hs, 400, oss.str(), "application/json");
        return 0;
    }
    ON_BLOCK_EXIT(cJSON2_Delete, json);

    for (int i = 0; i < cJSON2_GetArraySize(json); i++) {
        cJSON2 *item = cJSON2_GetArrayItem(json, i);
        if (item->type != cJSON2_String) {
            LOG_WARN("Handler_mediafile[%p]: Ignored non-string value", this);
            continue;
        }
        tags.push_back(std::make_pair(item->string, item->valuestring));
    }

    return err;
}

// checks for parsing overflows
void CheckOverflow(Packet *pkt, size_t size) {
  if (pkt->ptr + size - 1 >= pkt->len) {
    // overflow case, throw error
    LOG_WARN("Parsing error: pointer overflow for int");
  }
}

int32_t ofs_open_nolock(const char *ct_name, container_handle_t **ct)
{
    container_handle_t *tmp_ct = NULL;
    int32_t ret = 0;
    container_handle_t *hnd = NULL;
    avl_index_t where = 0;
    ofs_super_block_t *sb = NULL;

    if ((ct == NULL) || (ct_name == NULL))
    {
        LOG_ERROR("Invalid parameter. ct(%p) ct_name(%p)\n", ct, ct_name);
        return -INDEX_ERR_PARAMETER;
    }

    if (strlen(ct_name) >= OFS_NAME_SIZE)
    {
        LOG_ERROR("file name size must < %d bytes.\n", OFS_NAME_SIZE);
        return -INDEX_ERR_PARAMETER;
    }

    LOG_INFO("Open the ct. ct_name(%s)\n", ct_name);

    tmp_ct = avl_find(g_container_list, (avl_find_fn_t)compare_container2, ct_name, &where);
    if (tmp_ct)
    {
        tmp_ct->ref_cnt++;
        *ct = tmp_ct;
        LOG_WARN("File ref_cnt inc. ct_name(%s) ref_cnt(%d)\n", ct_name, tmp_ct->ref_cnt);
        return 0;
    }

    ret = init_container_resource(&tmp_ct, ct_name);
    if (ret < 0)
    {
        LOG_ERROR("Init ct resource failed. ct_name(%s) ret(%d)\n", ct_name, ret);
        return ret;
    }

    ret = os_disk_open(&tmp_ct->disk_hnd, ct_name);
    if (ret < 0)
    {
        LOG_ERROR("Open disk failed. ct_name(%s) ret(%d)\n", ct_name, ret);
        (void)close_container(tmp_ct);
        return ret;
    }

    sb = &tmp_ct->sb;

    sb->sectors_per_block = SECTORS_PER_BLOCK;
    sb->block_size = BYTES_PER_BLOCK;

    ret = ofs_read_super_block(tmp_ct);
    if (ret < 0)
    {
        LOG_ERROR("Read block failed. ct_name(%s) vbn(%lld) ret(%d)\n", ct_name, SUPER_BLOCK_VBN, ret);
        (void)close_container(tmp_ct);
        return ret;
    }

    ret = check_super_block(sb);
    if (ret < 0)
    {
        LOG_ERROR("Check super block failed. ct_name(%s) ret(%d)\n", ct_name, ret);
        (void)close_container(tmp_ct);
        return ret;
    }

    /* open system object */
    ret = open_system_objects(tmp_ct);
    if (ret < 0)
    {
        LOG_ERROR("Open system object failed. ct_name(%s) ret(%d)\n", ct_name, ret);
        close_container(tmp_ct);
        return ret;
    }

    *ct = tmp_ct;
    
    LOG_INFO("Open the ct success. ct_name(%s) ct(%p)\n", ct_name, ct);

    return 0;
}     

int32_t ofs_create_container_nolock(const char *ct_name, uint64_t total_sectors, container_handle_t **ct)
{
    container_handle_t *tmp_ct = NULL;
    int32_t ret = 0;
    avl_index_t where = 0;

    if ((ct == NULL) || (total_sectors == 0) || (ct_name == NULL))
    {
        LOG_ERROR("Invalid parameter. ct(%p) total_sectors(%lld) ct_name(%p)\n", ct, total_sectors, ct_name);
        return -INDEX_ERR_PARAMETER;
    }
    
    if (strlen(ct_name) >= OFS_NAME_SIZE)
    {
        LOG_ERROR("file name size must < %d bytes.\n", OFS_NAME_SIZE);
        return -INDEX_ERR_PARAMETER;
    }
    
    LOG_INFO("Create the ct. ct_name(%s) total_sectors(%lld)\n", ct_name, total_sectors);

    /* already opened */
    tmp_ct = avl_find(g_container_list, (avl_find_fn_t)compare_container2, ct_name, &where);
    if (tmp_ct)
    {
        *ct = tmp_ct;
        LOG_WARN("The ct is opened already. ct_name(%s) start_lba(%lld)\n", ct_name);
        return -INDEX_ERR_IS_OPENED;
    }

    /* allocate resource */
    ret = init_container_resource(&tmp_ct, ct_name);
    if (ret < 0)
    {
        LOG_ERROR("Init ct resource failed. ct_name(%s) ret(%d)", ct_name, ret);
        return ret;
    }
    
    /* init super block */
    ret = init_super_block(&tmp_ct->sb, total_sectors, BYTES_PER_BLOCK_SHIFT);
    if (ret < 0)
    {
        LOG_ERROR("init super block failed. name(%s)\n", ct_name);
        close_container(tmp_ct);
        return ret;
    }

    ret = os_disk_create(&tmp_ct->disk_hnd, ct_name);
    if (ret < 0)
    {
        LOG_ERROR("init disk failed. ret(%d)\n", ret);
        close_container(tmp_ct);
        return ret;
    }

    ret = ofs_init_super_block(tmp_ct);
    if (ret < 0)
    {
        LOG_ERROR("Update super block failed. ct_name(%s) vbn(%lld) ret(%d)\n", ct_name, SUPER_BLOCK_VBN, ret);
        close_container(tmp_ct);
        return ret;
    }
    
    ret = create_system_objects(tmp_ct);
    if (ret < 0)
    {
        LOG_ERROR("create system objects failed. ct_name(%s) ret(%d)\n", ct_name, ret);
        close_container(tmp_ct);
        return ret;
    }
    
    ret = ofs_update_super_block(tmp_ct);
    if (ret < 0)
    {
        LOG_ERROR("Update super block failed. hnd(%p) ret(%d)\n", tmp_ct, ret);
        close_container(tmp_ct);
        return ret;
    }

    *ct = tmp_ct;

    LOG_INFO("Create the ct success. ct_name(%s) total_sectors(%lld) ct(%p)\n", ct_name, total_sectors, tmp_ct);
    
    return 0;
}     

void ScriptTest::runTest()
{
  // before we start remove any temporary files that we generated last time.
  _removeFile(_script + ".stdout.failed");
  _removeFile(_script + ".stderr.failed");
  _removeFile(_script + ".stdout.failed.stripped");
  _removeFile(_script + ".stderr.failed.stripped");
  _removeFile(_script + ".stdout.first");
  _removeFile(_script + ".stderr.first");
  _removeFile(_script + ".stdout.first.stripped");
  _removeFile(_script + ".stderr.first.stripped");
  _removeFile(_script + ".stdout.stripped");
  _removeFile(_script + ".stderr.stripped");

  _runProcess();

  if (QFile(_script + ".stdout").exists() == false || QFile(_script + ".stderr").exists() == false)
  {
    LOG_WARN("STDOUT or STDERR doesn't exist for " + _script +
             "\n*************************\n"
             "  This can be resolved by reviewing the output for correctness and then \n"
             "  creating a new baseline. E.g.\n"
             "  verify: \n"
             "    less " + _script + ".stdout.first\n"
             "    less " + _script + ".stderr.first\n"
             "  ### NOTE: If the test is comparing against a known good file (e.g. .osm\n"
             "  ### then it may be better to update that file. You'll have to look at\n"
             "  ### the source files to be sure.\n"
             "  Make a new baseline:\n"
             "    mv " + _script + ".stdout.first " + _script + ".stdout\n"
             "    mv " + _script + ".stderr.first " + _script + ".stderr\n"
             "*************************\n"
             );

    _baseStderr = "<invalid/>";
    _baseStdout = "<invalid/>";
    _writeFile(_script + ".stdout.first", _stdout);
    _writeFile(_script + ".stderr.first", _stderr);
    _writeFile(_script + ".stdout.first.stripped", _removeIgnoredSubstrings(_stdout));
    _writeFile(_script + ".stderr.first.stripped", _removeIgnoredSubstrings(_stderr));
    CPPUNIT_ASSERT_MESSAGE(QString("STDOUT or STDERR does not exist").toStdString(), false);
  }

  _baseStderr = _readFile(_script + ".stderr");
  _baseStdout = _readFile(_script + ".stdout");

  bool failed = false;

  if (_removeIgnoredSubstrings(_baseStdout) != _removeIgnoredSubstrings(_stdout))
  {
    _writeFile(_script + ".stdout.failed", _stdout);
    _writeFile(_script + ".stdout.failed.stripped", _removeIgnoredSubstrings(_stdout));
    _writeFile(_script + ".stdout.stripped", _removeIgnoredSubstrings(_baseStdout));

    if (_printDiff)
    {
      LOG_WARN("STDOUT does not match for:\n" +
        _script + ".stdout" + " " + _script + ".stdout.failed");
      _runDiff(_script + ".stdout.stripped", _script + ".stdout.failed.stripped");
    }
    else
    {
      LOG_WARN("STDOUT does not match for:\n" +
               _script + ".stdout.failed" + " " + _script + ".stdout\n"
               "\n*************************\n"
               "  This can be resolved by reviewing the output for correctness and then \n"
               "  creating a new baseline. E.g.\n"
               "  verify: \n"
               "    diff " + _script + ".stdout.stripped " + _script + ".stdout.failed.stripped\n"
               "  ### NOTE: If the test is comparing against a known good file (e.g. .osm\n"
               "  ### then it may be better to update that file. You'll have to look at\n"
               "  ### the source files to be sure.\n"
               "  Make a new baseline:\n"
               "    mv " + _script + ".stdout.failed " + _script + ".stdout\n"
               "*************************\n"
               );
    }

    failed = true;
  }

  if (_removeIgnoredSubstrings(_baseStderr) != _removeIgnoredSubstrings(_stderr))
  {
    _writeFile(_script + ".stderr.failed", _stderr);
    _writeFile(_script + ".stderr.failed.stripped", _removeIgnoredSubstrings(_stderr));
    _writeFile(_script + ".stderr.stripped", _removeIgnoredSubstrings(_baseStderr));

    if (_printDiff)
    {
      LOG_WARN("STDERR does not match for:\n" +
        _script + ".stderr" + " " + _script + ".stderr.failed");
      _runDiff(_script + ".stderr.stripped", _script + ".stderr.failed.stripped");
    }
    else
    {
      LOG_WARN("STDERR does not match for:\n" +
               _script + ".stderr.failed" + " " + _script + ".stderr\n"
               "\n*************************\n"
               "  This can be resolved by reviewing the output for correctness and then \n"
               "  creating a new baseline. E.g.\n"
//.........这里部分代码省略.........

void thread_mutex_unlock_c(mutex_t *mutex, int line, char *file)
{
#ifdef DEBUG_MUTEXES
    thread_type *th = thread_self();

    if (!th) {
        LOG_ERROR3("No record for %u in unlock [%s:%d]", thread_self(), file, line);
    }

    LOG_DEBUG5("Unlocking %p (%s) on line %d in file %s by thread %d", mutex, mutex->name, line, file, th ? th->thread_id : -1);

    mutex->line = line;

# ifdef CHECK_MUTEXES
    if (th) {
        int locks = 0;
        avl_node *node;
        mutex_t *tmutex;

        _mutex_lock(&_mutextree_mutex);

        while (node) {
            tmutex = (mutex_t *)node->key;

            if (tmutex->mutex_id == mutex->mutex_id) {
                if (tmutex->thread_id != th->thread_id) {
                    LOG_ERROR7("ILLEGAL UNLOCK (%d != %d) on mutex [%s] in file %s line %d by thread %d [%s]", tmutex->thread_id, th->thread_id, 
                         mutex->name ? mutex->name : "undefined", file, line, th->thread_id, th->name);
                    _mutex_unlock(&_mutextree_mutex);
                    return;
                }
            } else if (tmutex->thread_id == th->thread_id) {
                locks++;
            }

            node = avl_get_next (node);
        }

        if ((locks > 0) && (_multi_mutex.thread_id != th->thread_id)) {
            /* Don't have double mutex, has more than this mutex left */
        
            LOG_WARN("(%d != %d) Thread %d [%s] tries to unlock a mutex [%s] in file %s line %d, without owning double mutex!",
                 _multi_mutex.thread_id, th->thread_id, th->thread_id, th->name, mutex->name ? mutex->name : "undefined", file, line);
        }

        _mutex_unlock(&_mutextree_mutex);
    }
# endif  /* CHECK_MUTEXES */

    _mutex_unlock(mutex);

    _mutex_lock(&_mutextree_mutex);

    LOG_DEBUG2("Unlocked %p by thread %d", mutex