//assoc_insert函数会调用本函数，当item数量到了哈希表表长的1.5被才会调用 
static void assoc_start_expand(void) {
    if (started_expanding)
        return;
    started_expanding = true;
    pthread_cond_signal(&maintenance_cond);
}

/**
 * Search worker. This method was invoked on threads launched from `main`.
 */
void *search_worker(void *arg)
{
    enum file_type t;
    worker_params *params = (worker_params *)arg;
    file_queue_node *current;
    const size_t out_len = 1024;
    int utf8_pattern_len = strlen(op.pattern);

    while (1) {
        // This worker takes out a search target file from the queue. If the queue is empty, worker
        // will be waiting until find at least one target file.
        pthread_mutex_lock(&file_mutex);
        while ((current = peek_file_for_search(params->queue)) == NULL) {
            // Break this loop if all files was searched.
            if (is_complete_scan_file()) {
                pthread_mutex_unlock(&file_mutex);
                return NULL;
            }
            pthread_cond_wait(&file_cond, &file_mutex);
        }
        pthread_mutex_unlock(&file_mutex);

        // Check file type of the target file, then if it is a binary, we skip it because the hw is
        // the software in order to search "source code", so searching binary files is not good.
        int fd = open(current->filename, O_RDONLY);
        if (fd != -1 && (t = detect_file_type(fd, current->filename)) != FILE_TYPE_BINARY) {
            char *pattern = op.pattern;

            // Convert the pattern to appropriate encoding if an encoding of the target file is
            // not UTF-8.
            char out[out_len + 1], pattern_len = utf8_pattern_len;
            if (t == FILE_TYPE_EUC_JP || t == FILE_TYPE_SHIFT_JIS) {
                if (t == FILE_TYPE_EUC_JP) {
                    to_euc(op.pattern, utf8_pattern_len, out, out_len);
                } else if (t == FILE_TYPE_SHIFT_JIS) {
                    to_sjis(op.pattern, utf8_pattern_len, out, out_len);
                }
                pattern = out;
                pattern_len = strlen(pattern);
            }

            // Searching.
            match_line_list *match_lines = create_match_line_list();
            int match_count = search(fd, pattern, pattern_len, t, match_lines, params->index);

            if (match_count > 0) {
                // Set additional data to the queue data because it will be used on print worker in
                // order to print results to the console. `match_lines` variable will be released
                // along with the file queue when it is released.
                current->matched      = true;
                current->match_lines  = match_lines;
                current->t            = t;
            } else {
                // If the pattern was not matched, the lines queue is no longer needed, so do free.
                free_match_line_list(match_lines);
            }
        }

        // Wake up print worker.
        current->searched = true;
        pthread_cond_signal(&print_cond);

        close(fd);
    }

    return NULL;
}

static void worker_thread(void *arg)
{
    uintptr_t no = (uintptr_t) arg;
    unsigned int i;
    lkey_t key;
    val_t getval;

    /*
     * increment begin_thread_num, and wait for broadcast signal from last created thread
     */
    if (system_variables.thread_num != 1) {
      pthread_mutex_lock(&begin_mtx);
      begin_thread_num++;
      if (begin_thread_num == system_variables.thread_num)
	pthread_cond_broadcast(&begin_cond);
      else {
	while (begin_thread_num < system_variables.thread_num)
	  pthread_cond_wait(&begin_cond, &begin_mtx);	
      }
      pthread_mutex_unlock(&begin_mtx);
    }

    gettimeofday(&stat_data[no].begin, NULL);
    sum[no] = 0;

    /*  main loop */
    key = no * system_variables.item_num;
    for (i = 0; i < system_variables.item_num; i++) {
      ++key;
      if (0 < system_variables.verbose)
	fprintf(stderr, "thread[%lu] add: %lu\n", (uintptr_t)no,
		(uintptr_t) key);

      if (add(list, (lkey_t) key, (val_t)key) != true)
	fprintf (stderr, "ERROR[%lu]: add %lu\n", (uintptr_t)no, (uintptr_t)key);

      if (1 < system_variables.verbose)
	show_list(list);
      
      //      usleep(no);
      //      pthread_yield(NULL);
    }

    usleep(no * 10);

    key = no * system_variables.item_num;
    for (i = 0; i < system_variables.item_num; i++) {
      ++key;
      if (delete(list, (lkey_t) key, &getval) != true) {
	printf ("ERROR[%lu]: del %lu\n", (uintptr_t)no, (uintptr_t)key);
      }

      if (1 < system_variables.verbose)
	show_list(list);
      
      if (0 < system_variables.verbose)
	fprintf(stderr, "delete: val = %ld\n", (lkey_t) getval);
      
      sum[no] += getval;
      
      check[getval]++;
    }

    /* send signal */
    gettimeofday(&stat_data[no].end, NULL);
    pthread_mutex_lock(&end_mtx);
    end_thread_num++;
    pthread_cond_signal(&end_cond);
    pthread_mutex_unlock(&end_mtx);
}

//.........这里部分代码省略.........
  tvhftrace("main", http_client_done);
  tvhftrace("main", tcp_server_done);

  // Note: the locking is obviously a bit redundant, but without
  //       we need to disable the gtimer_arm call in epg_save()
  pthread_mutex_lock(&global_lock);
  tvhftrace("main", epg_save);

#if ENABLE_TIMESHIFT
  tvhftrace("main", timeshift_term);
#endif
  pthread_mutex_unlock(&global_lock);

  tvhftrace("main", epggrab_done);
#if ENABLE_MPEGTS
  tvhftrace("main", mpegts_done);
#endif
  tvhftrace("main", descrambler_done);
  tvhftrace("main", service_mapper_done);
  tvhftrace("main", service_done);
  tvhftrace("main", channel_done);
  tvhftrace("main", bouquet_done);
  tvhftrace("main", dvr_done);
  tvhftrace("main", subscription_done);
  tvhftrace("main", access_done);
  tvhftrace("main", epg_done);
  tvhftrace("main", avahi_done);
  tvhftrace("main", bonjour_done);
  tvhftrace("main", imagecache_done);
  tvhftrace("main", lang_code_done);
  tvhftrace("main", api_done);

  tvhtrace("main", "tasklet enter");
  pthread_cond_signal(&tasklet_cond);
  pthread_join(tasklet_tid, NULL);
  tvhtrace("main", "tasklet thread end");
  tasklet_flush();
  tvhtrace("main", "tasklet leave");

  tvhftrace("main", hts_settings_done);
  tvhftrace("main", dvb_done);
  tvhftrace("main", lang_str_done);
  tvhftrace("main", esfilter_done);
  tvhftrace("main", profile_done);
  tvhftrace("main", intlconv_done);
  tvhftrace("main", urlparse_done);
  tvhftrace("main", idnode_done);
  tvhftrace("main", notify_done);
  tvhftrace("main", spawn_done);

  tvhlog(LOG_NOTICE, "STOP", "Exiting HTS Tvheadend");
  tvhlog_end();

  tvhftrace("main", config_done);

  if(opt_fork)
    unlink(opt_pidpath);
    
#if ENABLE_TSFILE
  free(opt_tsfile.str);
#endif
  free(opt_satip_xml.str);

  /* OpenSSL - welcome to the "cleanup" hell */
  ENGINE_cleanup();
  RAND_cleanup();

int OS_COND_SIGNAL(OS_COND* cond)
{
    return (pthread_cond_signal(cond));
}

void *commands_thread(void *arg)
{
  const Position *position = ((Commands_thread_arg*) arg)->position;
  Move **first_move = ((Commands_thread_arg*) arg)->first_move;
  pthread_cond_t *command_cond =  ((Commands_thread_arg*) arg)->command_cond;
  pthread_cond_t *position_cond =  ((Commands_thread_arg*) arg)->position_cond;
  pthread_cond_t *network_cond =  ((Commands_thread_arg*) arg)->network_cond;
  Serial_connection *robot = ((Commands_thread_arg*) arg)->robot;
  Commands_infos *commands_infos = ((Commands_thread_arg*) arg)->commands_infos;

  LOG_DEBUG("Commands thread started\n");

  pthread_mutex_lock(&robot->free);
  //mutex_and_send(robot, "rpid s\r", "write rpid s unsucessful\n");

  for(;;)
  {
    // Wait until we get a set of commands
    while(*first_move == NULL)
      wait_time_cond(command_cond, robot, COMMAND_CHECK_INTERVAL);
    LOG_DEBUG("Command : New set of commands\n");

    Move *current_first_move = *first_move;
    Move current_move = *current_first_move;
    LOG_DEBUG("Command : New move : %d, %d, %d : %d\n",
      (int) current_move.goal.x, (int) current_move.goal.y, (int) current_move.goal.t, current_move.speed);

    int goal_reached = 0;
    int orientation_phase = -1;
    int move_phase = -1;
    int bypass_phase = -1;
    int left = 0, right = 0, return_towards_obs = 0, is_beginning_defined = 0;
    int return_to_direction = -1;
    int count_dbg = 0;
    Position begin_bypass_position;
    char command[500];

    for(;;)
    {
      // Check if we are on desired position
      /*if ((*first_move)->type == POSITION && check_desired_pos(position, &(current_move.goal)))
      {
        //stop();
        pthread_cond_signal(position_cond);
        LOG_DEBUG("Move ended\n");
        count_dbg = 0;
        goal_reached = 1;
      }*/

      // Check if we have another move to do
      if (goal_reached && (current_move.next_move != NULL))
      {
        current_move = *current_move.next_move;
        goal_reached = 0;
        orientation_phase = -1;
        move_phase = -1;
        bypass_phase = -1;
        left = right = return_towards_obs = is_beginning_defined = 0;
        return_to_direction = -1;
        LOG_DEBUG("Command : New move : %d, %d, %d : %d\n",
          (int) current_move.goal.x, (int) current_move.goal.y, (int) current_move.goal.t, current_move.speed);
      }
      else if (return_to_direction == 1)
      {
        if (position->t > (current_move.goal.t - 4)
              && position->t < (current_move.goal.t + 4))
        {
          sprintf(command, "stop\r");
          mutex_and_send(robot, command, "write stop unsuccessful");
          orientation_phase = 0;
        }

      }
      else if (return_to_direction == 0)
      {
        if(current_move.speed == 0)
        {
          sprintf(command, "pwm 1:0 2:0\r");
        }
        else
        {
          if(current_move.speed > 0)
            sprintf(command, "mogo 1:%d 2:%d\r", (current_move.speed*24/100)+10, (current_move.speed*24/100)+10);
          else
            sprintf(command, "mogo 1:%d 2:%d\r", -((current_move.speed*24/100)+10), -((current_move.speed*24/100)+10));
        }
        mutex_and_send(robot, command, "write mogo unsuccesful\n");
      }
      else if (!goal_reached)
      {
        if(current_move.speed == 0)
        {
          LOG_DEBUG("speed 0\n");
          mutex_and_send(robot, "stop\r", "write stop unsuccessful\n");
          goal_reached = 1;
          continue;
        }
        

	    	if (current_move.type == DIRECTION && bypass_phase == -1)
//.........这里部分代码省略.........

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

  int pshared;
  ASSERT_EQ(0, pthread_condattr_getpshared(&attr, &pshared));
  ASSERT_EQ(PTHREAD_PROCESS_PRIVATE, pshared);
}

TEST(pthread, pthread_condattr_setclock) {
  pthread_condattr_t attr;
  pthread_condattr_init(&attr);

  ASSERT_EQ(0, pthread_condattr_setclock(&attr, CLOCK_REALTIME));
  clockid_t clock;
  ASSERT_EQ(0, pthread_condattr_getclock(&attr, &clock));
  ASSERT_EQ(CLOCK_REALTIME, clock);

  ASSERT_EQ(0, pthread_condattr_setclock(&attr, CLOCK_MONOTONIC));
  ASSERT_EQ(0, pthread_condattr_getclock(&attr, &clock));
  ASSERT_EQ(CLOCK_MONOTONIC, clock);

  ASSERT_EQ(EINVAL, pthread_condattr_setclock(&attr, CLOCK_PROCESS_CPUTIME_ID));
}

TEST(pthread, pthread_cond_broadcast__preserves_condattr_flags) {
#if defined(__BIONIC__)
  pthread_condattr_t attr;
  pthread_condattr_init(&attr);

  ASSERT_EQ(0, pthread_condattr_setclock(&attr, CLOCK_MONOTONIC));
  ASSERT_EQ(0, pthread_condattr_setpshared(&attr, PTHREAD_PROCESS_SHARED));

  pthread_cond_t cond_var;
  ASSERT_EQ(0, pthread_cond_init(&cond_var, &attr));

  ASSERT_EQ(0, pthread_cond_signal(&cond_var));
  ASSERT_EQ(0, pthread_cond_broadcast(&cond_var));

  attr = static_cast<pthread_condattr_t>(*reinterpret_cast<uint32_t*>(cond_var.__private));
  clockid_t clock;
  ASSERT_EQ(0, pthread_condattr_getclock(&attr, &clock));
  ASSERT_EQ(CLOCK_MONOTONIC, clock);
  int pshared;
  ASSERT_EQ(0, pthread_condattr_getpshared(&attr, &pshared));
  ASSERT_EQ(PTHREAD_PROCESS_SHARED, pshared);
#else  // !defined(__BIONIC__)
  GTEST_LOG_(INFO) << "This tests a bionic implementation detail.\n";
#endif  // !defined(__BIONIC__)
}

class pthread_CondWakeupTest : public ::testing::Test {
 protected:
  pthread_mutex_t mutex;
  pthread_cond_t cond;

  enum Progress {
    INITIALIZED,
    WAITING,
    SIGNALED,
    FINISHED,
  };
  std::atomic<Progress> progress;
  pthread_t thread;

 protected:
  virtual void SetUp() {
    ASSERT_EQ(0, pthread_mutex_init(&mutex, NULL));
    ASSERT_EQ(0, pthread_cond_init(&cond, NULL));

void ConditionImpl::signal()
{
    int rc = pthread_cond_signal( &_cond );
    if( rc != 0 )
        throw SystemError( rc, "pthread_cond_signal failed");
}

void equeue_sema_signal(equeue_sema_t *s) {
    pthread_mutex_lock(&s->mutex);
    s->signal = true;
    pthread_cond_signal(&s->cond);
    pthread_mutex_unlock(&s->mutex);
}

// --------------------------------------------------------------------------------------  FUNCTION
// --------------------------------------------------------------------------------------  FUNCTION
void Monitor::Signal() const
{
    int err = pthread_cond_signal(&mCondition);
    HandleErrno(err, "pthread_cond_signal");
}

int main(){
    int LISTEN_BACKLOG=10;
    int sfd=socket(AF_INET,SOCK_STREAM,0);	
    struct sockaddr_in my_addr, their_addr;
    if(sfd==-1){
	fprintf(stderr,"socket error\n");
	exit(1);
    }
    //struct sockaddr addr;
    my_addr.sin_family=AF_INET;                
    my_addr.sin_port = htons(9100);            
    my_addr.sin_addr.s_addr = INADDR_ANY; 
    bzero(&(my_addr.sin_zero), 8);               
    if(bind(sfd,(struct sockaddr *)&my_addr,sizeof(struct sockaddr))==-1){
	fprintf(stderr,"bind error\n");
	exit(1);
    }
    if(listen(sfd,LISTEN_BACKLOG)==-1){
	fprintf(stderr,"listen error\n");
	exit(1);
    }
	int nfds;
	m_epollfd=epoll_create(QUEUELEN);
	struct epoll_event m_epollev[QUEUELEN];
	struct epoll_event ev;
	ev.data.fd=sfd;
	ev.events=EPOLLIN|EPOLLET;
	epoll_ctl(m_epollfd,EPOLL_CTL_ADD,sfd,&ev);

	int sockfd=0;int i=0;
	int sock;
	socklen_t sin_size;
	   
	thread_make(THREAD_NUM);

	while(1){
//	epoll
	//if recv,lock sockArray[sockRear++]=sock; unlock;
	//read rear=(rear+1)%QUEUELEN	sockArray[rear]=sock;
	nfds=epoll_wait(m_epollfd,m_epollev,QUEUELEN,-1);
	for(i=0;i<nfds;i++)
	{
		sockfd=m_epollev[i].data.fd;
		struct epoll_event _event=m_epollev[i];
		if(sfd==sockfd)
		{
			//accept
			sin_size=sizeof(struct sockaddr);
			sock=accept(sfd,(void*)&their_addr,&sin_size);
			ev.data.fd=sock;
			ev.events=EPOLLIN|EPOLLET;
			epoll_ctl(m_epollfd,EPOLL_CTL_ADD,sock,&ev);
		}
		else if(_event.events&EPOLLIN)
		{
		//lock
			pthread_mutex_lock(&mlock);
			if(sockFront==((sockRear+1)%QUEUELEN))
			{
				printf("GET TO THE MAX LIMIT %d\n",QUEUELEN);
				pthread_mutex_unlock(&mlock);
				continue;
			}
			sockRear=(sockRear+1)%QUEUELEN;
			sockArray[sockRear]=sockfd;
			printf("here we are %d\n",sockfd);
			pthread_mutex_unlock(&mlock);
			pthread_cond_signal(&m_condlock);	
		//unlock
		}
		else
			printf("epoll:else\n");
	}
    }
} 

//.........这里部分代码省略.........
/* 			info("got %d * %d = %d", message_timeout, steps, fwd_msg->timeout); */
			steps++;
			fwd_msg->timeout += (start_timeout*steps);
/* 			info("now  + %d*%d = %d", start_timeout, steps, fwd_msg->timeout); */
		}

		ret_list = slurm_receive_msgs(fd, steps, fwd_msg->timeout);
		/* info("sent %d forwards got %d back", */
/* 		     fwd_msg->header.forward.cnt, list_count(ret_list)); */

		if(!ret_list || (fwd_msg->header.forward.cnt != 0
				 && list_count(ret_list) <= 1)) {
			slurm_mutex_lock(fwd_msg->forward_mutex);
			mark_as_failed_forward(&fwd_msg->ret_list, name,
					       errno);
			free(name);
			if(ret_list)
				list_destroy(ret_list);
			if (hostlist_count(hl) > 0) {
				free_buf(buffer);
				buffer = init_buf(fwd_msg->buf_len);
				slurm_mutex_unlock(fwd_msg->forward_mutex);
				slurm_close_accepted_conn(fd);
				fd = -1;
				continue;
			}
			goto cleanup;
		} else if((fwd_msg->header.forward.cnt+1)
			  != list_count(ret_list)) {
			/* this should never be called since the above
			   should catch the failed forwards and pipe
			   them back down, but this is here so we
			   never have to worry about a locked
			   mutex */
			ListIterator itr = NULL;
			char *tmp = NULL;
			int first_node_found = 0;
			hostlist_iterator_t host_itr
				= hostlist_iterator_create(hl);
			error("We shouldn't be here.  We forwarded to %d "
			      "but only got %d back",
			      (fwd_msg->header.forward.cnt+1),
			      list_count(ret_list));
			while((tmp = hostlist_next(host_itr))) {
				int node_found = 0;
				itr = list_iterator_create(ret_list);
				while((ret_data_info = list_next(itr))) {
					if(!ret_data_info->node_name) {
						first_node_found = 1;
						ret_data_info->node_name =
							xstrdup(name);
					}
					if(!strcmp(tmp,
						   ret_data_info->node_name)) {
						node_found = 1;
						break;
					}
				}
				list_iterator_destroy(itr);
				if(!node_found) {
					mark_as_failed_forward(
						&fwd_msg->ret_list,
						tmp,
						SLURM_COMMUNICATIONS_CONNECTION_ERROR);
				}
				free(tmp);
			}
			hostlist_iterator_destroy(host_itr);
			if(!first_node_found) {
				mark_as_failed_forward(&fwd_msg->ret_list,
						       name,
						       SLURM_COMMUNICATIONS_CONNECTION_ERROR);
			}
		}
		break;
	}
	slurm_mutex_lock(fwd_msg->forward_mutex);
	if(ret_list) {
		while((ret_data_info = list_pop(ret_list)) != NULL) {
			if(!ret_data_info->node_name) {
				ret_data_info->node_name = xstrdup(name);
			}
			list_push(fwd_msg->ret_list, ret_data_info);
			debug3("got response from %s",
			       ret_data_info->node_name);
		}
		list_destroy(ret_list);
	}
	free(name);
cleanup:
	if ((fd >= 0) && slurm_close_accepted_conn(fd) < 0)
		error ("close(%d): %m", fd);
	hostlist_destroy(hl);
	destroy_forward(&fwd_msg->header.forward);
	free_buf(buffer);
	pthread_cond_signal(fwd_msg->notify);
	slurm_mutex_unlock(fwd_msg->forward_mutex);

	return (NULL);
}

void *_fwd_tree_thread(void *arg)
{
	fwd_tree_t *fwd_tree = (fwd_tree_t *)arg;
	List ret_list = NULL;
	char *name = NULL;
	char *buf = NULL;
	slurm_msg_t send_msg;

	slurm_msg_t_init(&send_msg);
	send_msg.msg_type = fwd_tree->orig_msg->msg_type;
	send_msg.data = fwd_tree->orig_msg->data;

	/* repeat until we are sure the message was sent */
	while ((name = hostlist_shift(fwd_tree->tree_hl))) {
		if (slurm_conf_get_addr(name, &send_msg.address)
		    == SLURM_ERROR) {
			error("fwd_tree_thread: can't find address for host "
			      "%s, check slurm.conf", name);
			slurm_mutex_lock(fwd_tree->tree_mutex);
			mark_as_failed_forward(&fwd_tree->ret_list, name,
					       SLURM_UNKNOWN_FORWARD_ADDR);
 			pthread_cond_signal(fwd_tree->notify);
			slurm_mutex_unlock(fwd_tree->tree_mutex);
			free(name);

			continue;
		}

		send_msg.forward.timeout = fwd_tree->timeout;
		if((send_msg.forward.cnt = hostlist_count(fwd_tree->tree_hl))) {
			buf = hostlist_ranged_string_xmalloc(
					fwd_tree->tree_hl);
			send_msg.forward.nodelist = buf;
		} else
			send_msg.forward.nodelist = NULL;

		if (send_msg.forward.nodelist && send_msg.forward.nodelist[0]) {
			debug3("Tree sending to %s along with %s",
			       name, send_msg.forward.nodelist);
		} else
			debug3("Tree sending to %s", name);

		ret_list = slurm_send_addr_recv_msgs(&send_msg, name,
						     fwd_tree->timeout);

		xfree(send_msg.forward.nodelist);

		if(ret_list) {
			slurm_mutex_lock(fwd_tree->tree_mutex);
			list_transfer(fwd_tree->ret_list, ret_list);
			pthread_cond_signal(fwd_tree->notify);
			slurm_mutex_unlock(fwd_tree->tree_mutex);
			list_destroy(ret_list);
		} else {
			/* This should never happen (when this was
			   written slurm_send_addr_recv_msgs always
			   returned a list */
			error("fwd_tree_thread: no return list given from "
			      "slurm_send_addr_recv_msgs spawned for %s",
			      name);
			slurm_mutex_lock(fwd_tree->tree_mutex);
			mark_as_failed_forward(
				&fwd_tree->ret_list, name,
				SLURM_COMMUNICATIONS_CONNECTION_ERROR);
 			pthread_cond_signal(fwd_tree->notify);
			slurm_mutex_unlock(fwd_tree->tree_mutex);
			free(name);

			continue;
		}

		free(name);

		/* check for error and try again */
		if(errno == SLURM_COMMUNICATIONS_CONNECTION_ERROR)
 			continue;

		break;
	}

	_destroy_tree_fwd(fwd_tree);

	return NULL;
}

 void TwoQStrategy::release(uint64_t id) {
     lockPreserved();
     _preserved[id] = false;
     unlockPreserved();
     pthread_cond_signal(&_preserved_changed);
 }

static boolean msg_bus_post_msg(mct_bus_t *bus, mct_bus_msg_t *bus_msg)
{
  mct_bus_msg_t *local_msg;
  int post_msg = FALSE;
  unsigned int payload_size;

  if (!bus_msg) {
    ALOGE("%s:%d NULL ptr", __func__, __LINE__);
    goto error_2;
  }

  if (bus->bus_queue->length > MAX_MCT_BUS_QUEUE_LENGTH) {
      pthread_mutex_lock(&bus->bus_msg_q_lock);
      mct_bus_queue_flush(bus);
      ALOGE("%s : Discard the bus msg's that got stagnated in the queue\n",__func__);
      pthread_mutex_unlock(&bus->bus_msg_q_lock);
      return TRUE;
  }

  CDBG("%s:type=%d", __func__, bus_msg->type);

  switch (bus_msg->type) {
    case MCT_BUS_MSG_ISP_SOF:
      payload_size = sizeof(mct_bus_msg_isp_sof_t);
      if (bus->thread_run == 1) {
        pthread_mutex_lock(&bus->bus_sof_msg_lock);
        pthread_cond_signal(&bus->bus_sof_msg_cond);
        pthread_mutex_unlock(&bus->bus_sof_msg_lock);
      }
      break;
    case MCT_BUS_MSG_Q3A_AF_STATUS:
      payload_size = sizeof(mct_bus_msg_af_status_t);
      break;
    case MCT_BUS_MSG_ASD_HDR_SCENE_STATUS:
      payload_size = sizeof(mct_bus_msg_asd_hdr_status_t);
      break;
    case MCT_BUS_MSG_FACE_INFO:
      payload_size = sizeof(cam_face_detection_data_t);
      break;
    case MCT_BUS_MSG_HIST_STATS_INFO:
      payload_size = sizeof(cam_hist_stats_t);
      break;
    case MCT_BUS_MSG_PREPARE_HW_DONE:
      payload_size = sizeof(cam_prep_snapshot_state_t);
      break;
    case MCT_BUS_MSG_ZSL_TAKE_PICT_DONE:
      payload_size = sizeof(cam_frame_idx_range_t);
      break;
    case MCT_BUS_MSG_SET_SENSOR_INFO:
      payload_size = sizeof(mct_bus_msg_sensor_metadata_t);
      break;
    case MCT_BUS_MSG_SET_STATS_AEC_INFO:
      payload_size = bus_msg->size;
      break;
    case MCT_BUS_MSG_SET_ISP_GAMMA_INFO:
    case MCT_BUS_MSG_SET_ISP_STATS_AWB_INFO:
      payload_size = bus_msg->size;
      break;
   case MCT_BUS_MSG_ISP_STREAM_CROP:
      payload_size = sizeof(mct_bus_msg_stream_crop_t);
      break;
   case MCT_BUS_MSG_SET_AEC_STATE:
      payload_size = sizeof(int32_t);
      break;
   case MCT_BUS_MSG_SET_AEC_PRECAPTURE_ID:
      payload_size = sizeof(int32_t);
      break;
   case MCT_BUS_MSG_SET_AEC_RESET:
      payload_size = 0;
      break;
   case MCT_BUS_MSG_SET_AF_STATE:
      payload_size = sizeof(int32_t);
      break;
   case MCT_BUS_MSG_SET_AF_TRIGGER_ID:
      payload_size = sizeof(int32_t);
    case MCT_BUS_MSG_AUTO_SCENE_DECISION:
      payload_size = sizeof(mct_bus_msg_asd_decision_t);
      break;
   case MCT_BUS_MSG_ERROR_MESSAGE:
      payload_size = sizeof(mct_bus_msg_error_message_t);
      break;
   case MCT_BUS_MSG_AE_INFO:
      payload_size = sizeof(cam_ae_params_t);
      break;
   case MCT_BUS_MSG_AWB_INFO:
      payload_size = sizeof(cam_awb_params_t);
      break;
    case MCT_BUS_MSG_AE_EXIF_DEBUG_INFO:
       payload_size = sizeof(cam_ae_exif_debug_t);
       break;
    case MCT_BUS_MSG_AWB_EXIF_DEBUG_INFO:
       payload_size = sizeof(cam_awb_exif_debug_t);
       break;
   case MCT_BUS_MSG_AF_EXIF_DEBUG_INFO:
       payload_size = sizeof(cam_af_exif_debug_t);
       break;
   case MCT_BUS_MSG_ASD_EXIF_DEBUG_INFO:
       payload_size = sizeof(cam_asd_exif_debug_t);
       break;
   case MCT_BUS_MSG_STATS_EXIF_DEBUG_INFO:
//.........这里部分代码省略.........

//.........这里部分代码省略.........
      hot_provider = GetNextActiveProvider(p, hot_provider);
   }

   /* Spawn writer thread LAST - It will release readygo */
   /* The point here is to give te workers every chance to get started
      and wait until the very last minute to spawn the writer thread.
      Because it is the writer thread that will release the readygo
      lock and let things start. While this is not *perfect* (the
      workers *could* be still starting - but that is exceptionally
      unlikely), it is fairly solid (as witnessed in practice). */
   if ( SpawnWriterThread(tr) )
      return(1);

   /* If the user said run for x time, then set that up here */
   stop_at = time(NULL);
   stop_at--;            /* Effectively an assert(), but more about
                            making the compiler happy. */

   if ( 0 < cfg->run_for )
   {
      stop_at = time(NULL);
      stop_at += cfg->run_for;
   }


   /* This is the timer loop that activates all the other threads */
   first_iteration = 1;
   while(go)
   {
      /* Check for run_for time expiration */
      if ( 0 < cfg->run_for )
      {
         time_now = time(NULL);
         if ( time_now >= stop_at )
         {
            go = 0;
            continue;
         }
      }

      /* The first iteration is used to avoid printing data on start
         that is typically incomplete anyways. (Incomplete = the diff
         values are not properly timed) Also, the first printing is
         timed VERY CLOSE to the second iteration. So it is ok to 
         skip this. Unfortunately, the way the loop is structured the
         timer (event) has to happen at the BOTTOM so that we can exit
         out (without some kind of break) when we get our event. The
         break *may* have been the more elegant answer here....
      */
      if ( first_iteration )
         first_iteration = 0;
      else
      {
         pthread_mutex_lock(&tr->wrarg->readygo); /* Gotta take lock to go */

         tc = 0;
         while ( tc < apcnt )
         {
            pthread_cond_signal(&tr->wkalist[tc].runcond);
            tc++;
         }

         pthread_cond_signal(&tr->wrarg->writecond);
      }

      /* What is going on here:
         - There is a mutex/cond semaphore that we block on waiting
           for some sort of signal. Typically this will be a simple
           alarm timer, but it may be a Ctrl-C or similarly induced
           signal. The key difference will be what the value of the
           global "go" is. If it is a timer, then go will be 1. If
           it is an exit-inducing event, then go will be 0. Either
           way, go will be evaluated at the top of the loop.
         - No, go is not protected by a mutex. No, this is not seen
           as a problem.
         - The mutex here is only for access to the cond variable.
           It protects nothing (other than the cond).
         - The cond here is about responding to the signal induced
           event loop. And as expressed earlier... there are really
           only two events we care about - a timer tick and a quit
           signal.
       */

      /* Standard cond wait block */
      pthread_mutex_lock(&timerc->mutex);
      while( tt == 0 ) /* Use tt because timer may have already fired */
         pthread_cond_wait(&timerc->cond, &timerc->mutex);
      tt = 0; /* Immediately clear */
      pthread_mutex_unlock(&timerc->mutex);
   }

   /* Take the writer lock (to insure that it is done writing)
      and then finish up the write - whatever that may be. The
      writer_finish() method was registered at about the same time the
      write() method.
   */
   pthread_mutex_lock(&tr->wrarg->writelock);
   writer_finish(tr->wrarg->writer_data);
   return(0);
}

void cond_signal(cond_t cond) {
  pthread_cond_t *pcond = (pthread_cond_t *)cond;

  int res = pthread_cond_signal(cond);
  CHECK_EQ(res, 0);
}

/*Scheduler input function*/
void auth_mgr_input(nkn_task_id_t id)
{
    nkn_task_set_state(id, TASK_STATE_EVENT_WAIT);
    struct nkn_task     *ntask = nkn_task_get_task(id);
    int dns_type;
    int ret=0;
    char domain_name[2048];
    char *p_domain_name;

    assert(ntask);
    auth_msg_t *data = (auth_msg_t*)nkn_task_get_data(id);
    /* If its DNS lookup, the scheduler thread will just fire this request
    and leave, so no need to put it into a queue like auth tasks*/
    if (data->authtype==DNSLOOKUP)
    {
        if (adnsd_enabled) {
            adns_daemon_input(id);
            return;
        }

        auth_dns_t* pdns=(auth_dns_t*)(data->authdata);
        pdns->auth_task_id=id;
        //Returns void
        pthread_mutex_lock(&cares_mutex);
        AO_fetch_and_add1(&glob_dns_task_called);
        /* Since we will support ttl provided by nameserver
         * we will not be using gethostbyname() anymore
         * Also, I should be using nameser.h and have ns_c_in
         * and ns_t_a for the below query, but lots of woes.
                 * ares_gethostbyname(channel,(char*)pdns->domain,AF_INET,
         *		cares_callback,data);
         */

        /* TODO: use nameser.h */
        if (pdns->ip[0].family == AF_INET) {
            dns_type = 1; // ns_t_a
        } else {
            dns_type = 28; // ns_t_aaaa
        }

        if (pdns->dns_query_len == 0) {
            p_domain_name = (char *)pdns->domain;
        } else {
            memcpy(domain_name, pdns->domain, pdns->domain_len);
            domain_name[pdns->domain_len] = '.';
            memcpy(domain_name + pdns->domain_len + 1, pdns->dns_query, pdns->dns_query_len);
            domain_name[pdns->domain_len + 1 + pdns->dns_query_len] = '\0';
            p_domain_name = domain_name;
        }
        ares_query(channel,
                   p_domain_name,
                   1, /*ns_c_in = 1 Class: Internet. */
                   dns_type,
                   cares_callback_req, data);
        /*
         * Wake up the select loop.
         */
        if (channel_ready == 0) {
            ares_destroy(channel);
            //re-init the channel to get a new port
            ret = ares_init_options(&channel, &options, optmask);
            if (ret != ARES_SUCCESS) {
                DBG_LOG(SEVERE, MOD_AUTHMGR,"ares_init: %d %s", ret, ares_strerror(ret));
                assert(0);
            }
            channel_ready = 1;
            AO_fetch_and_add1(&glob_dns_channel_reinit);
            pthread_mutex_unlock(&cares_mutex);
            return;
        }
        pthread_cond_signal(&cares_cond);
        pthread_mutex_unlock(&cares_mutex);
    }
    else {
        NKN_MUTEX_LOCK(&authreq_mutex);
        if (isam_task_arrFull()) {
            nkn_task_set_action_and_state(id, TASK_ACTION_OUTPUT,
                                          TASK_STATE_RUNNABLE);
            NKN_MUTEX_UNLOCK(&authreq_mutex);
            return;
        }
        auth_task_arr[auth_task_arr_tail] = id;
        if (auth_task_arr_tail == (MAX_AM_TASK_ARR - 1))
        {
            auth_task_arr_tail = 0;
        }
        else
        {
            auth_task_arr_tail++;
        }
        pthread_cond_signal(&authreq_cond);
        NKN_MUTEX_UNLOCK(&authreq_mutex);
    }

    return;
}

 void unlock()
 {
     pdalboost::pthread::pthread_mutex_scoped_lock const local_lock(&m);
     is_locked=false;
     BOOST_VERIFY(!pthread_cond_signal(&cond));
 }

/*
 * Wakeup the kernel, if sleeping (shall not be called from a signal handler) */
void interrupt(void)
{
    pthread_cond_signal(&wfi_cond);
}