void msg_process(char * (*handler_request)(char *req_buf, int fd))
{
	LOCALTSS(tss);
	int resp_size;
	RPCREQ* req;
	char* resp;

	int fd;
	MSG_DATA *req_msg;
	MSG_DATA *resp_msg;


	req_msg = NULL;
	resp_msg = NULL;
	
	while(TRUE)
	{
		
		pthread_mutex_lock(&mutex);

		while (msg_list_head == NULL)
			pthread_cond_wait(&cond, &mutex);

	
		req_msg = msg_list_head;
		msg_list_head = msg_list_head->next;
		msg_list_len--;
		pthread_mutex_unlock(&mutex);

		if(!strncasecmp(RPC_RBD_MAGIC, req_msg->data, STRLEN(RPC_RBD_MAGIC)))
		{
			req = conn_build_req(req_msg->block_buffer, req_msg->n_size);
		}
		else
		{
			req = conn_build_req(req_msg->data, req_msg->n_size);
		}
		
		fd = req_msg->fd;

		if(req_msg->n_size)
		{
			resp = handler_request(req->data, fd);

			if (resp == NULL)
			{
				if (tss->tstat & TSS_PARSER_ERR)
				{
					resp = conn_build_resp_byte(RPC_PARSER_ERR, 0, NULL);
				}
				else
				{
					resp = conn_build_resp_byte(RPC_FAIL, 0, NULL);					
				}
			}
		}
		else
		{	resp = conn_build_resp_byte(RPC_FAIL, 0, NULL);
		}

		/* If it's the select where/range session, skip the success operation. */
		if((fd < 0) || (((RPCRESP *)resp)->status_code & RPC_SKIP_SEND))
		{
			//local msg, such as recovery task msg, so no need to response
			goto finish;
		}

		resp_size = conn_get_resp_size((RPCRESP *)resp);

		resp_msg = NULL;
		resp_msg = (MSG_DATA *)msg_mem_alloc();
		resp_msg->n_size = resp_size;
		resp_msg->block_buffer = NULL;
		Assert(resp_size < MSG_SIZE);
		MEMCPY(resp_msg->data, resp, resp_size);
		resp_msg->fd = fd;	
		
		struct epoll_event ev;
		ev.data.ptr = resp_msg;
	
		ev.events = EPOLLOUT | EPOLLET;

		epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev);

finish:

		if (req_msg->block_buffer != NULL)
		{
			free(req_msg->block_buffer);
		}
		
		msg_mem_free(req_msg);
  
		conn_destroy_req(req);
		conn_destroy_resp_byte(resp);		
		
		tss_init(tss);
	}
}

// Called once per step from python code. Puts the representedInputValues in the proper
// form for processing, then tells each GPU thread to take a step. Once they've finished
// the step, this function puts the representedOutputValues and spikes in the appropriate
// python arrays so that they can be read on the python side when this call returns
void step(float* input, float* output, float* probes, float* spikes, float start, float end, int n_steps)
{
  start_time = start;
  end_time = end;

  NengoGPUData* current_data;

  int i, j, k, l;

  for( i = 0; i < num_devices; i++)
  {
    current_data = nengo_data_array[i];
    memcpy(current_data->input_host->array, input,
           current_data->dimension * current_data->num_steps * sizeof(float));
  }

  // Tell the runner threads to run and then wait for them to finish.
  // The last of them to finish running will wake this thread up.
  pthread_mutex_lock(mutex);
  myCVsignal = 1;
  pthread_cond_broadcast(cv_GPUThreads);
  pthread_cond_wait(cv_JNI, mutex);
  pthread_mutex_unlock(mutex);

  current_data = nengo_data_array[0];
  memset(output, 0.0, current_data->dimension * current_data->num_steps * sizeof(float));

  for(i = 0; i < num_devices; i++)
  {
      current_data = nengo_data_array[i];

      for(j = 0; j < current_data->num_steps * current_data->dimension; j++)
      {
           output[j] += current_data->output_host->array[j];
      }
  }

  int probe_index = 0, spike_index = 0, index = 0;
  for(i = 0; i < nengo_data_array[0]->num_steps; i++)
  {
      for(j = 0; j < num_devices; j++)
      {
          current_data = nengo_data_array[j];

          for(k = 0; k < current_data->num_probes; k++)
          {
              probes[probe_index] =
                  current_data->probes_host->array[i * current_data->num_probes + k];

              if(current_data->num_spikes > 0)
              {
                  index = i * current_data->num_spikes
                          + k * current_data->neurons_per_item;

                  for(l = 0; l < current_data->neurons_per_item; l++)
                  {
                      spikes[spike_index] = current_data->spikes_host->array[index + l];
                      spike_index++;
                  }
              }

              probe_index++;
          }
      }
  }
}

void Conduction::wait(){
    pthread_cond_wait(&sig, &(Mutex &mutex));
}

static void * run_process(void * data)
{
    vx_world_t * world = data;


    while (world->process_running) {
        vx_world_listener_t * listener = NULL;
        char * buffer_name = NULL;

        // 1) Wait until there's data
        pthread_mutex_lock(&world->queue_mutex);
        while (zarray_size(world->buffer_queue) == 0 && zarray_size(world->listener_queue) == 0 && world->process_running) {
            pthread_cond_wait(&world->queue_cond, &world->queue_mutex);
        }

        if (!world->process_running) { // XXX cleaning out the queue?
            pthread_mutex_unlock(&world->queue_mutex);
            break;
        }

        // Processing new listeners takes priority
        if ( zarray_size(world->listener_queue) > 0) {
            zarray_get(world->listener_queue, 0, &listener);
            zarray_remove_index(world->listener_queue, 0, 0);
        } else {
            assert(zarray_size(world->buffer_queue) > 0);
            zarray_get(world->buffer_queue, 0, &buffer_name);
            zarray_remove_index(world->buffer_queue, 0, 0);
        }
        pthread_mutex_unlock(&world->queue_mutex);

        // Operation A: New listener
        if (listener != NULL) {
            // re-transmit each buffer that has already been serialized
            pthread_mutex_lock(&world->buffer_mutex);
            zhash_iterator_t itr;
            zhash_iterator_init(world->buffer_map, &itr);
            char * name = NULL;
            vx_buffer_t * buffer = NULL;
            while(zhash_iterator_next(&itr, &name, &buffer)) {
                if (buffer->front_codes->pos != 0) {
                    vx_code_output_stream_t * bresc_codes = make_buffer_resource_codes(buffer, buffer->front_resc);

                    // Doing a swap in 3 steps (instead of 4) like this
                    // is only safe if the listener is brand new, which we are
                    // guaranteed is the case in this chunk of code
                    listener->send_codes(listener, bresc_codes->data, bresc_codes->pos);
                    listener->send_resources(listener, buffer->front_resc);
                    listener->send_codes(listener, buffer->front_codes->data, buffer->front_codes->pos);

                    vx_code_output_stream_destroy(bresc_codes);
                }
            }

            pthread_mutex_unlock(&world->buffer_mutex);
        }

        // Operation B: buffer swap
        if (buffer_name != NULL) {
            vx_buffer_t * buffer = NULL;
            pthread_mutex_lock(&world->buffer_mutex);
            zhash_get(world->buffer_map, &buffer_name, &buffer);
            pthread_mutex_unlock(&world->buffer_mutex);

            delayed_swap(buffer);
        }

    }

    pthread_exit(NULL);
}

void sys_cond_wait(sys_cond_t *cond, sys_mutex_t *held_mutex)
{
    int result = pthread_cond_wait(&(cond->cond), &(held_mutex->mutex));
    assert(result == 0);
}

void __po_hi_gqueue_wait_for_incoming_event (__po_hi_task_id id,
                                             __po_hi_local_port_t* port)
{
#ifdef RTEMS_PURE
  rtems_status_code ret;
#endif

#ifdef _WIN32
  DWORD ret;
#endif

#if defined (POSIX) || defined (RTEMS_POSIX) || defined (XENO_POSIX)
  int error = pthread_mutex_lock (&__po_hi_gqueues_mutexes[id]);
  __DEBUGMSG("*** Locking (%d) %d\n", id, error);
#elif defined (XENO_NATIVE)
  rt_mutex_acquire (&__po_hi_gqueues_mutexes[id], TM_INFINITE);
#elif defined (RTEMS_PURE)
  ret = rtems_semaphore_obtain (__po_hi_gqueues_semaphores[id], RTEMS_WAIT, RTEMS_NO_TIMEOUT);
  if (ret != RTEMS_SUCCESSFUL)
    {
      __DEBUGMSG ("[GQUEUE] Cannot obtain semaphore in __po_hi_gqueue_store_in()\n");
    }
#elif defined (_WIN32)
  EnterCriticalSection(&__po_hi_gqueues_cs[id]);
#endif

  while(__po_hi_gqueues_queue_is_empty[id] == 1)
    {
      __PO_HI_INSTRUMENTATION_VCD_WRITE("0t%d\n", id);

#if defined (POSIX) || defined (RTEMS_POSIX) || defined (XENO_POSIX)
      __DEBUGMSG("*** Waiting (%d)\n", id);
      int error = pthread_cond_wait (&__po_hi_gqueues_conds[id],
                                     &__po_hi_gqueues_mutexes[id]);
      __DEBUGMSG("*** Done Waiting (%d) %d\n", id, error);
#elif defined (XENO_NATIVE)
      rt_cond_wait (&__po_hi_gqueues_conds[id], &__po_hi_gqueues_mutexes[id], TM_INFINITE);
#elif defined (RTEMS_PURE)
      ret = rtems_semaphore_release (__po_hi_gqueues_semaphores[id]);
      if (ret != RTEMS_SUCCESSFUL)
        {
          __DEBUGMSG ("[GQUEUE] Cannot obtain semaphore in __po_hi_gqueue_store_in()\n");
        }
      rtems_task_wake_after (1);
      ret = rtems_semaphore_obtain (__po_hi_gqueues_semaphores[id], RTEMS_WAIT, RTEMS_NO_TIMEOUT);
      if (ret != RTEMS_SUCCESSFUL)
        {
          __DEBUGMSG ("[GQUEUE] Cannot obtain semaphore in __po_hi_gqueue_store_in()\n");
        }
      else
        {
          __PO_HI_DEBUG_CRITICAL ("[GQUEUE] semaphore %d obtained\n", id);
        }
#elif defined (_WIN32)
      LeaveCriticalSection(&__po_hi_gqueues_cs[id]);

      ret = WaitForSingleObject (__po_hi_gqueues_events[id], INFINITE);
      if (ret == WAIT_FAILED)
        {
          __PO_HI_DEBUG_DEBUG ("[GQUEUE] Wait failed\n");
        }
      EnterCriticalSection(&__po_hi_gqueues_cs[id]);
#endif

      __PO_HI_INSTRUMENTATION_VCD_WRITE("1t%d\n", id);
    }

  __DEBUGMSG ("[GQUEUE] Gogo kiki\n");
  *port = __po_hi_gqueues_global_history[id][__po_hi_gqueues_global_history_offset[id]];


#if defined (POSIX) || defined (RTEMS_POSIX) || defined (XENO_POSIX)
  pthread_mutex_unlock (&__po_hi_gqueues_mutexes[id]);
#elif defined (XENO_NATIVE)
  rt_mutex_release (&__po_hi_gqueues_mutexes[id]);
#elif defined (_WIN32)
  LeaveCriticalSection(&__po_hi_gqueues_cs[id]);
#elif defined (RTEMS_PURE)
  ret = rtems_semaphore_release (__po_hi_gqueues_semaphores[id]);
  if (ret != RTEMS_SUCCESSFUL)
    {
      __DEBUGMSG ("[GQUEUE] Cannot release semaphore in __po_hi_gqueue_store_in()\n");
    }

  __PO_HI_DEBUG_CRITICAL ("[GQUEUE] semaphore %d released\n", id);
#endif

}

//.........这里部分代码省略.........
		}
#endif
		ep->stat.waits++;
		ep->waiting = TRUE;
		if (timeout > 0) {
			struct timespec deadline;

			clock_gettime(CLOCK_REALTIME, &deadline);
			if (timeout > 1000) {
				int sec;
				sec = timeout / 1000;
				deadline.tv_sec += sec;
				timeout -= sec * 1000;
			}

			if (deadline.tv_nsec >= 1000000000) {
				deadline.tv_sec++;
				deadline.tv_nsec -= 1000000000;
			}

			//deadline.tv_sec = mtcp->cur_tv.tv_sec;
			//deadline.tv_nsec = (mtcp->cur_tv.tv_usec + timeout * 1000) * 1000;
			ret = pthread_cond_timedwait(&ep->epoll_cond, 
					&ep->epoll_lock, &deadline);
			if (ret && ret != ETIMEDOUT) {
				/* errno set by pthread_cond_timedwait() */
				pthread_mutex_unlock(&ep->epoll_lock);
				TRACE_ERROR("pthread_cond_timedwait failed. ret: %d, error: %s\n", 
						ret, strerror(errno));
				return -1;
			}
			timeout = 0;
		} else if (timeout < 0) {
			ret = pthread_cond_wait(&ep->epoll_cond, &ep->epoll_lock);
			if (ret) {
				/* errno set by pthread_cond_wait() */
				pthread_mutex_unlock(&ep->epoll_lock);
				TRACE_ERROR("pthread_cond_wait failed. ret: %d, error: %s\n", 
						ret, strerror(errno));
				return -1;
			}
		}
		ep->waiting = FALSE;

		if (mtcp->ctx->done || mtcp->ctx->exit || mtcp->ctx->interrupt) {
			mtcp->ctx->interrupt = FALSE;
			//ret = pthread_cond_signal(&ep->epoll_cond);
			pthread_mutex_unlock(&ep->epoll_lock);
			errno = EINTR;
			return -1;
		}
	
	}
	
	/* fetch events from the user event queue */
	cnt = 0;
	num_events = eq->num_events;
	for (i = 0; i < num_events && cnt < maxevents; i++) {
		event_socket = &mtcp->smap[eq->events[eq->start].sockid];
		validity = TRUE;
		if (event_socket->socktype == MTCP_SOCK_UNUSED)
			validity = FALSE;
		if (!(event_socket->epoll & eq->events[eq->start].ev.events))
			validity = FALSE;
		if (!(event_socket->events & eq->events[eq->start].ev.events))
			validity = FALSE;

/** RX_PDSCH Decoding Thread */
static void * rx_pdsch_thread(void *param)
{

  //unsigned long cpuid;
  uint8_t dlsch_thread_index = 0;
  uint8_t pilot2,harq_pid,subframe;
  //  uint8_t last_slot;

  uint8_t dual_stream_UE = 0;
  uint8_t i_mod = 0;


#ifdef RTAI
  RT_TASK *task;
#endif

  int m,eNB_id = 0;
  int eNB_id_i = 1;
  PHY_VARS_UE *UE = PHY_vars_UE_g[0][0];

#ifdef RTAI
  task = rt_task_init_schmod(nam2num("RX_PDSCH_THREAD"), 0, 0, 0, SCHED_FIFO, 0xF);

  if (task==NULL) {
    LOG_E(PHY,"[SCHED][RX_PDSCH] Problem starting rx_pdsch thread!!!!\n");
    return 0;
  } else {
    LOG_I(PHY,"[SCHED][RX_PDSCH] rx_pdsch_thread started for with id %p\n",task);
  }

#endif

  mlockall(MCL_CURRENT | MCL_FUTURE);

  //rt_set_runnable_on_cpuid(task,1);
  //cpuid = rtai_cpuid();

#ifdef HARD_RT
  rt_make_hard_real_time();
#endif

  if (UE->lte_frame_parms.Ncp == NORMAL) {  // normal prefix
    pilot2 = 7;
  } else { // extended prefix
    pilot2 = 6;
  }


  while (!oai_exit) {
    VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_PDSCH_THREAD, 0);

    if (pthread_mutex_lock(&rx_pdsch_mutex) != 0) {
      LOG_E(PHY,"[SCHED][RX_PDSCH] error locking mutex.\n");
    } else {
      while (rx_pdsch_instance_cnt < 0) {
        pthread_cond_wait(&rx_pdsch_cond,&rx_pdsch_mutex);
      }

      if (pthread_mutex_unlock(&rx_pdsch_mutex) != 0) {
        LOG_E(PHY,"[SCHED][RX_PDSCH] error unlocking mutex.\n");
      }
    }

    VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_PDSCH_THREAD, 1);

    //      last_slot = rx_pdsch_slot;
    subframe = UE->slot_rx>>1;
    // Important! assumption that PDCCH procedure of next SF is not called yet
    harq_pid = UE->dlsch_ue[eNB_id][0]->current_harq_pid;
    UE->dlsch_ue[eNB_id][0]->harq_processes[harq_pid]->G = get_G(&UE->lte_frame_parms,
        UE->dlsch_ue[eNB_id][0]->harq_processes[harq_pid]->nb_rb,
        UE->dlsch_ue[eNB_id][0]->harq_processes[harq_pid]->rb_alloc,
        get_Qm(UE->dlsch_ue[eNB_id][0]->harq_processes[harq_pid]->mcs),
        UE->dlsch_ue[eNB_id][0]->harq_processes[harq_pid]->Nl,
        UE->lte_ue_pdcch_vars[eNB_id]->num_pdcch_symbols,
        UE->frame_rx,subframe);

    if ((UE->transmission_mode[eNB_id] == 5) &&
        (UE->dlsch_ue[eNB_id][0]->harq_processes[harq_pid]->dl_power_off==0) &&
        (openair_daq_vars.use_ia_receiver > 0)) {
      dual_stream_UE = 1;
      eNB_id_i = UE->n_connected_eNB;

      if (openair_daq_vars.use_ia_receiver == 2) {
        i_mod =  get_Qm(((UE->frame_rx%1024)/3)%28);
      } else {
        i_mod =  get_Qm(UE->dlsch_ue[eNB_id][0]->harq_processes[harq_pid]->mcs);
      }
    } else {
      dual_stream_UE = 0;
      eNB_id_i = eNB_id+1;
      i_mod = 0;
    }

    if (oai_exit) break;

    LOG_D(PHY,"[SCHED][RX_PDSCH] Frame %d, slot %d: Calling rx_pdsch_decoding with harq_pid %d\n",UE->frame_rx,UE->slot_rx,harq_pid);


//.........这里部分代码省略.........

 /**
  * Waits until this object is signalled with signal().
  *
  * @param mutex a mutex which is already locked, which is unlocked
  * atomically while we wait
  */
 void Wait(PosixMutex &mutex) {
   pthread_cond_wait(&cond, &mutex.mutex);
 }

void get_audio(float *lb, float *rb, int samples)
{
	PlayerCtx *ctx = g_ctx;

	if (ctx->audio_idx == -1) {
		memset(lb, 0, samples * sizeof(*lb));
		memset(rb, 0, samples * sizeof(*rb));
		return;
	}

	pthread_mutex_lock(&ctx->display_mode_mutex);
	DisplayMode display_mode = ctx->display_mode;
	pthread_mutex_unlock(&ctx->display_mode_mutex);

	if (display_mode != PLAY)
	{
		if (display_mode == PAUSE) {
			printf("get_audio: paused\n");
			if (!ctx->cur_frame) {
				next_video_frame(ctx);
			}
			while (ctx->cur_frame->seekid != ctx->cur_seekid &&
				   ctx->cur_frame->seekid != -ctx->cur_seekid) {
				printf("Drop audio due to seek\n");
				drop_audio(ctx, 1);
				next_video_frame(ctx);
				drop_audio(ctx, 1);
			}
			if (ctx->skip_frame) {
				ctx->skip_frame = 0;
				drop_audio(ctx, 1);
				next_video_frame(ctx);
				drop_audio(ctx, 1);
			}
			printf("get_audio: pause complete\n");
		}
		memset(lb, 0, samples * sizeof(*lb));
		memset(rb, 0, samples * sizeof(*rb));
		return;
	}
	pthread_mutex_lock(&ctx->settings_mutex);
	double volume = ctx->settings.volume / 100.0;
	pthread_mutex_unlock(&ctx->settings_mutex);

	while (samples) {
		double pts = -1;

		pthread_mutex_lock(&ctx->a_buf_mutex);
		int have_samples = ctx->a_buf_put - ctx->a_buf_get;
		if (!have_samples) {
			printf("Wait for audio\n");
			pthread_cond_wait(&ctx->a_buf_not_empty, &ctx->a_buf_mutex);
			pthread_mutex_unlock(&ctx->a_buf_mutex);
			continue;
		}
		if (have_samples < 0)
			have_samples += ctx->a_buf_len;
		pthread_mutex_unlock(&ctx->a_buf_mutex);

		int get = ctx->a_buf_get;
		int played = 0;
		while (samples && have_samples--) {
			if (ctx->a_buf[get].seekid == -ctx->cur_seekid) {
				memset(lb, 0, sizeof(*lb) * samples);
				memset(rb, 0, sizeof(*rb) * samples);
				samples = 0;
				break;
			}
			if (ctx->a_buf[get].seekid == ctx->cur_seekid) {
				pts = ctx->a_buf[get].pts;
				*lb++ = ctx->a_buf[get].l / 32768.0 * volume;
				*rb++ = ctx->a_buf[get].r / 32768.0 * volume;
				samples--;
				played++;
			}
			if (++get >= ctx->a_buf_len)
				get = 0;
		}

		pthread_mutex_lock(&ctx->a_buf_mutex);
		ctx->a_buf_get = get;
		pthread_cond_signal(&ctx->a_buf_not_full);
		pthread_mutex_unlock(&ctx->a_buf_mutex);

		printf("Played %d samples, next pts %f\n", played, pts);

		while (1) {
			if (!ctx->cur_frame) {
				next_video_frame(ctx);
				continue;
			}
			if (ctx->cur_frame->seekid == -ctx->cur_seekid)
				break;
			double next_pts = ctx->cur_frame->pts;
			if (ctx->last_frame_pts != -1)
				next_pts = 2 * ctx->cur_frame->pts - ctx->last_frame_pts;
			if (pts > next_pts || ctx->cur_frame->seekid != ctx->cur_seekid) {
				if (next_video_frame(ctx))
					continue;
			}
//.........这里部分代码省略.........

/*
 * Hook called by EGL to acquire a buffer. This call may block if no
 * buffers are available.
 *
 * The window holds a reference to the buffer between dequeueBuffer and
 * either queueBuffer or cancelBuffer, so clients only need their own
 * reference if they might use the buffer after queueing or canceling it.
 * Holding a reference to a buffer after queueing or canceling it is only
 * allowed if a specific buffer count has been set.
 *
 * The libsync fence file descriptor returned in the int pointed to by the
 * fenceFd argument will refer to the fence that must signal before the
 * dequeued buffer may be written to.  A value of -1 indicates that the
 * caller may access the buffer immediately without waiting on a fence.  If
 * a valid file descriptor is returned (i.e. any value except -1) then the
 * caller is responsible for closing the file descriptor.
 *
 * Returns 0 on success or -errno on error.
 */
int FbDevNativeWindow::dequeueBuffer(BaseNativeWindowBuffer** buffer, int *fenceFd)
{
    HYBRIS_TRACE_BEGIN("fbdev-platform", "dequeueBuffer", "");
    FbDevNativeWindowBuffer* fbnb=NULL;

    pthread_mutex_lock(&_mutex);

    if (m_allocateBuffers)
        reallocateBuffers();

    HYBRIS_TRACE_BEGIN("fbdev-platform", "dequeueBuffer-wait", "");
#if defined(DEBUG)

    if (m_frontBuf)
        TRACE("Status: Has front buf %p", m_frontBuf);

    std::list<FbDevNativeWindowBuffer*>::iterator cit = m_bufList.begin();
    for (; cit != m_bufList.end(); ++cit)
    {
        TRACE("Status: Buffer %p with busy %i\n", (*cit), (*cit)->busy);
    }
#endif

    while (m_freeBufs==0)
    {
        pthread_cond_wait(&_cond, &_mutex);
    }

    while (1)
    {
        std::list<FbDevNativeWindowBuffer*>::iterator it = m_bufList.begin();
        for (; it != m_bufList.end(); ++it)
        {
            if (*it==m_frontBuf)
                continue;
            if ((*it)->busy==0)
            {
                TRACE("Found a free non-front buffer");
                break;
            }
        }

        if (it == m_bufList.end())
        {
#if ANDROID_VERSION_MAJOR<=4 && ANDROID_VERSION_MINOR<2
            /*
             * This is acceptable in case you are on a stack that calls lock() before starting to render into buffer
             * When you are using fences (>= 2) you'll be waiting on the fence to signal instead. 
             * 
             * This optimization allows eglSwapBuffers to return and you can begin to utilize the GPU for rendering. 
             * The actual lock() probably first comes at glFlush/eglSwapBuffers
            */
            if (m_frontBuf && m_frontBuf->busy == 0)
            {
                TRACE("Used front buffer as buffer");
                fbnb = m_frontBuf;
                break;
            }
#endif
            // have to wait once again
            pthread_cond_wait(&_cond, &_mutex);
            continue;
        }

        fbnb = *it;
        break;
    }

    HYBRIS_TRACE_END("fbdev-platform", "dequeueBuffer-wait", "");
    assert(fbnb!=NULL);
    fbnb->busy = 1;
    m_freeBufs--;

    *buffer = fbnb;
    *fenceFd = -1;

    TRACE("%lu DONE --> %p", pthread_self(), fbnb);
    pthread_mutex_unlock(&_mutex);
    HYBRIS_TRACE_END("fbdev-platform", "dequeueBuffer", "");
    return 0;
}

size_t decode_video(PlayerCtx *ctx, AVPacket *packet, int new_packet, int32_t seekid)
{
	int decoded;
	int got_frame;

	if (!new_packet)
		fprintf(stderr, "warn: multi-frame video packets, pts might be inaccurate\n");

	ctx->v_pkt_pts = packet->pts;

	decoded = avcodec_decode_video2(ctx->v_codec_ctx, ctx->v_frame, &got_frame, packet);
	if (decoded < 0) {
		fprintf(stderr, "Error while decoding video frame\n");
		return packet->size;
	}
	if (!got_frame)
		return decoded;

	// The pts magic guesswork
	int64_t pts = AV_NOPTS_VALUE;
	int64_t *p_pts = (int64_t*)ctx->v_frame->opaque;
	int64_t frame_pts = AV_NOPTS_VALUE;
	if (p_pts)
		frame_pts = *p_pts;
	if (packet->dts != AV_NOPTS_VALUE) {
		ctx->v_faulty_dts += packet->dts <= ctx->v_last_dts;
		ctx->v_last_dts = packet->dts;
	}
	if (frame_pts != AV_NOPTS_VALUE) {
		ctx->v_faulty_pts += frame_pts <= ctx->v_last_pts;
		ctx->v_last_pts = frame_pts;
	}
	if ((ctx->v_faulty_pts <= ctx->v_faulty_dts || packet->dts == AV_NOPTS_VALUE)
		&& frame_pts != AV_NOPTS_VALUE)
		pts = frame_pts;
	else
		pts = packet->dts;

	if (pts == AV_NOPTS_VALUE) {
		if (ctx->v_last_pts != AV_NOPTS_VALUE) {
			pts = ctx->v_last_pts++;
		} else if (ctx->v_last_dts != AV_NOPTS_VALUE) {
			pts = ctx->v_last_dts++;
		}
	}

	if (pts == AV_NOPTS_VALUE) {
		if (ctx->v_last_pts != AV_NOPTS_VALUE) {
			pts = ctx->v_last_pts++;
		} else if (ctx->v_last_dts != AV_NOPTS_VALUE) {
			pts = ctx->v_last_dts++;
		} else {
			pts = 0;
		}
	}

	pthread_mutex_lock(&ctx->v_buf_mutex);
	while (((ctx->v_buf_put + 1) % ctx->v_buf_len) == ctx->v_buf_get) {
		printf("Wait for space in video buffer\n");
		pthread_cond_wait(&ctx->v_buf_not_full, &ctx->v_buf_mutex);
	}
	pthread_mutex_unlock(&ctx->v_buf_mutex);

	VideoFrame *frame = ctx->v_bufs[ctx->v_buf_put];
	if (!frame) {
		frame = malloc(sizeof(VideoFrame));
		frame->stride = ctx->v_frame->linesize[0];
		frame->data_size = frame->stride * ctx->height;
		frame->data = malloc(frame->data_size);
		ctx->v_bufs[ctx->v_buf_put] = frame;
	}

	if (frame->stride != ctx->v_frame->linesize[0]) {
		fprintf(stderr, "stride mismatch: %d != %d\n", (int)frame->stride, ctx->v_frame->linesize[0]);
		return decoded;
	}

	if (!ctx->v_sws_ctx) {
		ctx->v_sws_ctx = sws_getContext(ctx->width, ctx->height, ctx->v_codec_ctx->pix_fmt,
										ctx->width, ctx->height, PIX_FMT_GRAY8, SWS_BICUBIC,
										NULL, NULL, NULL);
	}

	AVPicture pict;
	pict.data[0] = frame->data;
	pict.linesize[0] = frame->stride;

	sws_scale(ctx->v_sws_ctx, ctx->v_frame->data, ctx->v_frame->linesize, 0, ctx->height, pict.data, pict.linesize);

	frame->pts = av_q2d(ctx->v_stream->time_base) * pts;
	frame->seekid = seekid;

	printf("Put frame %d (pts:%f seekid:%d)\n", ctx->v_buf_put, frame->pts, seekid);
	pthread_mutex_lock(&ctx->v_buf_mutex);
	if (++ctx->v_buf_put == ctx->v_buf_len)
		ctx->v_buf_put = 0;
	pthread_cond_signal(&ctx->v_buf_not_empty);
	pthread_mutex_unlock(&ctx->v_buf_mutex);

	return decoded;
//.........这里部分代码省略.........

size_t decode_audio(PlayerCtx *ctx, AVPacket *packet, int new_packet, int32_t seekid)
{
	int bytes = ctx->a_stride * AVCODEC_MAX_AUDIO_FRAME_SIZE;
	int decoded;

	decoded = avcodec_decode_audio3(ctx->a_codec_ctx, ctx->a_ibuf, &bytes, packet);
	if (decoded < 0) {
		fprintf(stderr, "Error while decoding audio frame\n");
		return packet->size;
	}
	int in_samples = bytes / ctx->a_stride;
	if (!in_samples)
		return decoded;


	int out_samples;
	out_samples = audio_resample(ctx->a_resampler, ctx->a_rbuf, ctx->a_ibuf, in_samples);

	pthread_mutex_lock(&ctx->a_buf_mutex);

	int free_samples;
	while (1) {
		free_samples = ctx->a_buf_get - ctx->a_buf_put;
		if (free_samples <= 0)
			free_samples += ctx->a_buf_len;

		if (free_samples <= out_samples) {
			printf("Wait for space in audio buffer\n");
			pthread_cond_wait(&ctx->a_buf_not_full, &ctx->a_buf_mutex);
		} else {
			break;
		}
	}
	pthread_mutex_unlock(&ctx->a_buf_mutex);

	if (new_packet && packet->pts != AV_NOPTS_VALUE)
		ctx->a_cur_pts = av_q2d(ctx->a_stream->time_base) * packet->pts;

	int put = ctx->a_buf_put;
	short *rbuf = ctx->a_rbuf;
	int i;
	for (i = 0; i < out_samples; i++) {
		ctx->a_buf[put].l = *rbuf++;
		ctx->a_buf[put].r = *rbuf++;
		rbuf += ctx->a_ch - 2;
		ctx->a_buf[put].seekid = seekid;
		ctx->a_buf[put].pts = ctx->a_cur_pts;
		put++;
		ctx->a_cur_pts += 1.0/SAMPLE_RATE;
		if (put == ctx->a_buf_len)
			put = 0;
	}

	printf("Put %d audio samples at pts %f\n", out_samples, ctx->a_cur_pts);

	pthread_mutex_lock(&ctx->a_buf_mutex);
	ctx->a_buf_put = put;
	pthread_cond_signal(&ctx->a_buf_not_empty);
	pthread_mutex_unlock(&ctx->a_buf_mutex);

	return decoded;
}

int main(void)
{
	struct sigaction sigact;
	int r = 1;

	r = libusb_init(NULL);
	if (r < 0) {
		fprintf(stderr, "failed to initialise libusb\n");
		exit(1);
	}

	r = find_dpfp_device();
	if (r < 0) {
		fprintf(stderr, "Could not find/open device\n");
		goto out;
	}

	r = libusb_claim_interface(devh, 0);
	if (r < 0) {
		fprintf(stderr, "usb_claim_interface error %d %s\n", r, strerror(-r));
		goto out;
	}
	printf("claimed interface\n");

	r = print_f0_data();
	if (r < 0)
		goto out_release;

	r = do_init();
	if (r < 0)
		goto out_deinit;

	/* async from here onwards */

	sigact.sa_handler = sighandler;
	sigemptyset(&sigact.sa_mask);
	sigact.sa_flags = 0;
	sigaction(SIGINT, &sigact, NULL);
	sigaction(SIGTERM, &sigact, NULL);
	sigaction(SIGQUIT, &sigact, NULL);

	r = pthread_create(&poll_thread, NULL, poll_thread_main, NULL);
	if (r)
		goto out_deinit;

	r = alloc_transfers();
	if (r < 0) {
		request_exit(1);
		pthread_join(poll_thread, NULL);
		goto out_deinit;
	}

	r = init_capture();
	if (r < 0) {
		request_exit(1);
		pthread_join(poll_thread, NULL);
		goto out_deinit;
	}

	while (!do_exit) {
		pthread_mutex_lock(&exit_cond_lock);
		pthread_cond_wait(&exit_cond, &exit_cond_lock);
		pthread_mutex_unlock(&exit_cond_lock);
	}

	printf("shutting down...\n");
	pthread_join(poll_thread, NULL);

	r = libusb_cancel_transfer(irq_transfer);
	if (r < 0) {
		request_exit(1);
		goto out_deinit;
	}

	r = libusb_cancel_transfer(img_transfer);
	if (r < 0) {
		request_exit(1);
		goto out_deinit;
	}

	while (img_transfer || irq_transfer)
		if (libusb_handle_events(NULL) < 0)
			break;

	if (do_exit == 1)
		r = 0;
	else
		r = 1;

out_deinit:
	libusb_free_transfer(img_transfer);
	libusb_free_transfer(irq_transfer);
	set_mode(0);
	set_hwstat(0x80);
out_release:
	libusb_release_interface(devh, 0);
out:
	libusb_close(devh);
	libusb_exit(NULL);
	return r >= 0 ? r : -r;
//.........这里部分代码省略.........

void* alarm_thread(void* arg)
{
	alarm_t* alarm;
	struct timespec cond_time;
	time_t now;
	int status,expired;
	status=pthread_mutex_lock(&alarm_mutex);
	if(status!=0)
	{
		err_abort(status,"Lock mutex");
	}
	while(1)
	{
		current_alarm=0;
		while(alarm_list == NULL)
	  	{
			status=pthread_cond_wait(&alarm_cond,&alarm_mutex);
			if(status!=0)
			{
				err_abort(status,"Wait on cond");
			}
		}
		alarm=alarm_list;
		alarm_list=alarm->link;
		now=time(NULL);
		expired=0;
		if(alarm->time > now)
		{
#ifdef DEBUG
			printf("[waiting:%d(%d)\"%s\"]\n",alarm->time,
					alarm->time-time(NULL),alarm->message);
#endif				
			cond_time.tv_sec=alarm->time;
			cond_time.tv_nsec=0;
			current_alarm=alarm->time;
			while(current_alarm==alarm->time)
			{
				status=pthread_cond_timedwait(
						&alarm_cond,&alarm_mutex,&cond_time);
				if(status==ETIMEDOUT)
				{
					expired=1;
					break;
				}
				if(status!=0)
				{
					err_abort(status,"Cond timedwait");
				}
			}
			if(!expired)
			{
				alarm_insert(alarm);
			}
		}			
		else
		{
			expired=1;
		}
		if(expired)
		{
			printf("(%d) %s\n",alarm->seconds,alarm->message);
			free(alarm);
		}
	}
}

static int at_send_command_full_nolock (const char *command, ATCommandType type,
                    const char *responsePrefix, const char *smspdu,
                    long long timeoutMsec, ATResponse **pp_outResponse)
{
    int err = 0;
#ifndef USE_NP
    struct timespec ts;
#endif /*USE_NP*/

    if(sp_response != NULL) {
        err = AT_ERROR_COMMAND_PENDING;
        goto error;
    }

    err = writeline (command);

    if (err < 0) {
        goto error;
    }

    s_type = type;
    s_responsePrefix = responsePrefix;
    s_smsPDU = smspdu;
    sp_response = at_response_new();

#ifndef USE_NP
    if (timeoutMsec != 0) {
        setTimespecRelative(&ts, timeoutMsec);
    }
#endif /*USE_NP*/

    while (sp_response->finalResponse == NULL && s_readerClosed == 0) {

        
        if (timeoutMsec != 0) {
#ifdef USE_NP
            err = pthread_cond_timeout_np(&s_commandcond, &s_commandmutex, timeoutMsec);
#else
            err = pthread_cond_timedwait(&s_commandcond, &s_commandmutex, &ts);
#endif /*USE_NP*/
        } else {
            err = pthread_cond_wait(&s_commandcond, &s_commandmutex);
        }

        if (err == ETIMEDOUT) {
            err = AT_ERROR_TIMEOUT;
            goto error;
        }
    }

    if (pp_outResponse == NULL) {
        at_response_free(sp_response);
    } else {
        /* line reader stores intermediate responses in reverse order */
        reverseIntermediates(sp_response);
        *pp_outResponse = sp_response;
    }

    sp_response = NULL;

    if(s_readerClosed > 0) {
        err = AT_ERROR_CHANNEL_CLOSED;
        goto error;
    }

    err = 0;
error:
    clearPendingCommand();

    return err;
}

//.........这里部分代码省略.........
}

void fio_verify_init(struct thread_data *td)
{
	if (td->o.verify == VERIFY_CRC32C_INTEL ||
	    td->o.verify == VERIFY_CRC32C) {
		crc32c_intel_probe();
	}
}

static void *verify_async_thread(void *data)
{
	struct thread_data *td = data;
	struct io_u *io_u;
	int ret = 0;

	if (td->o.verify_cpumask_set &&
	    fio_setaffinity(td->pid, td->o.verify_cpumask)) {
		log_err("fio: failed setting verify thread affinity\n");
		goto done;
	}

	do {
		FLIST_HEAD(list);

		read_barrier();
		if (td->verify_thread_exit)
			break;

		pthread_mutex_lock(&td->io_u_lock);

		while (flist_empty(&td->verify_list) &&
		       !td->verify_thread_exit) {
			ret = pthread_cond_wait(&td->verify_cond,
							&td->io_u_lock);
			if (ret) {
				pthread_mutex_unlock(&td->io_u_lock);
				break;
			}
		}

		flist_splice_init(&td->verify_list, &list);
		pthread_mutex_unlock(&td->io_u_lock);

		if (flist_empty(&list))
			continue;

		while (!flist_empty(&list)) {
			io_u = flist_first_entry(&list, struct io_u, verify_list);
			flist_del_init(&io_u->verify_list);

			io_u->flags |= IO_U_F_NO_FILE_PUT;
			ret = verify_io_u(td, &io_u);

			put_io_u(td, io_u);
			if (!ret)
				continue;
			if (td_non_fatal_error(td, ERROR_TYPE_VERIFY_BIT, ret)) {
				update_error_count(td, ret);
				td_clear_error(td);
				ret = 0;
			}
		}
	} while (!ret);

	if (ret) {

static void *item_crawler_thread(void *arg) {
    int i;

    pthread_mutex_lock(&lru_crawler_lock);
    if (settings.verbose > 2)
        fprintf(stderr, "Starting LRU crawler background thread\n");
    while (do_run_lru_crawler_thread) {
    pthread_cond_wait(&lru_crawler_cond, &lru_crawler_lock);

    while (crawler_count) {
        item *search = NULL;
        void *hold_lock = NULL;

        for (i = 0; i < LARGEST_ID; i++) {
            if (crawlers[i].it_flags != 1) {
                continue;
            }
            pthread_mutex_lock(&cache_lock);
            search = crawler_crawl_q((item *)&crawlers[i]);
            if (search == NULL ||
                (crawlers[i].remaining && --crawlers[i].remaining < 1)) {
                if (settings.verbose > 2)
                    fprintf(stderr, "Nothing left to crawl for %d\n", i);
                crawlers[i].it_flags = 0;
                crawler_count--;
                crawler_unlink_q((item *)&crawlers[i]);
                pthread_mutex_unlock(&cache_lock);
                continue;
            }
            uint32_t hv = hash(ITEM_key(search), search->nkey);
            /* Attempt to hash item lock the "search" item. If locked, no
             * other callers can incr the refcount
             */
            if ((hold_lock = item_trylock(hv)) == NULL) {
                pthread_mutex_unlock(&cache_lock);
                continue;
            }
            /* Now see if the item is refcount locked */
            if (refcount_incr(&search->refcount) != 2) {
                refcount_decr(&search->refcount);
                if (hold_lock)
                    item_trylock_unlock(hold_lock);
                pthread_mutex_unlock(&cache_lock);
                continue;
            }

            /* Frees the item or decrements the refcount. */
            /* Interface for this could improve: do the free/decr here
             * instead? */
            item_crawler_evaluate(search, hv, i);

            if (hold_lock)
                item_trylock_unlock(hold_lock);
            pthread_mutex_unlock(&cache_lock);

            if (settings.lru_crawler_sleep)
                usleep(settings.lru_crawler_sleep);
        }
    }
    if (settings.verbose > 2)
        fprintf(stderr, "LRU crawler thread sleeping\n");
    STATS_LOCK();
    stats.lru_crawler_running = false;
    STATS_UNLOCK();
    }
    pthread_mutex_unlock(&lru_crawler_lock);
    if (settings.verbose > 2)
        fprintf(stderr, "LRU crawler thread stopping\n");

    return NULL;
}

int __po_hi_gqueue_get_value (__po_hi_task_id      id,
                              __po_hi_local_port_t port,
                              __po_hi_request_t*   request)
{
   __po_hi_request_t* ptr;
#ifdef RTEMS_PURE
   rtems_status_code ret;
#endif

#ifdef _WIN32
   DWORD ret;
#endif


   ptr = &__po_hi_gqueues_most_recent_values[id][port];
#if defined (POSIX) || defined (RTEMS_POSIX) || defined (XENO_POSIX)
   pthread_mutex_lock (&__po_hi_gqueues_mutexes[id]);
#elif defined (XENO_NATIVE)
   rt_mutex_acquire (&__po_hi_gqueues_mutexes[id], TM_INFINITE);
#elif defined (RTEMS_PURE)
   ret = rtems_semaphore_obtain (__po_hi_gqueues_semaphores[id], RTEMS_WAIT, RTEMS_NO_TIMEOUT);
   if (ret != RTEMS_SUCCESSFUL)
   {
      __DEBUGMSG ("[GQUEUE] Cannot obtain semaphore in __po_hi_gqueue_store_in()\n");
   }
#elif defined (_WIN32)
  EnterCriticalSection(&