/* The main test function. */
int main(int argc, char *argv[])
{
	int ret, i;
	sem_t *sems;
	sem_t sem_last;

	long max;

	/* Initialize output */
	output_init();

	max = sysconf(_SC_SEM_NSEMS_MAX);

	if (max <= 0) {
		output("sysconf(_SC_SEM_NSEMS_MAX) = %ld\n", max);
		UNTESTED("There is no constraint on SEM_NSEMS_MAX");
	}

	sems = (sem_t *) calloc(max, sizeof(sem_t));

	if (sems == NULL) {
		UNRESOLVED(errno, "Failed to alloc space");
	}

	for (i = 0; i < max; i++) {
		ret = sem_init(&sems[i], 0, 0);

		if (ret != 0) {
			output
			    ("sem_init failed to initialize the %d nth semaphore.\n",
			     i);
			output("Tryed to initialize %ld.\n", max);
			output("Error is %d: %s\n", errno, strerror(errno));

			for (; i > 0; i--)
				sem_destroy(&sems[i - 1]);

			free(sems);

			PASSED;
		}
	}

	ret = sem_init(&sem_last, 0, 1);

	if (ret == 0) {
		FAILED
		    ("We were able to sem_init more than SEM_NSEMS_MAX semaphores");
	}

	if (errno != ENOSPC) {
		output("Error is %d: %s\n", errno, strerror(errno));
	}

	for (i = 0; i < max; i++)
		sem_destroy(&sems[i]);

	free(sems);

	/* Test passed */
#if VERBOSE > 0

	output("Test passed\n");

#endif

	PASSED;
}

// destroy the cache
// return 0 on success
// return -EINVAL if the cache is still running
int md_cache_destroy( struct md_syndicate_cache* cache ) {
   
   if( cache->running ) {
      // have to stop it first
      return -EINVAL;
   }
   
   cache->pending = NULL;
   cache->completed = NULL;
   
   md_cache_block_buffer_t* pendings[] = {
      cache->pending_1,
      cache->pending_2,
      NULL
   };
   
   for( int i = 0; pendings[i] != NULL; i++ ) {
      for( md_cache_block_buffer_t::iterator itr = pendings[i]->begin(); itr != pendings[i]->end(); itr++ ) {
         if( *itr != NULL ) {
            SG_safe_free( *itr );
         }
      }
      
      SG_safe_delete( pendings[i] );
   }
   
   md_cache_completion_buffer_t* completeds[] = {
      cache->completed_1,
      cache->completed_2,
      NULL
   };
   
   for( int i = 0; completeds[i] != NULL; i++ ) {
      for( md_cache_completion_buffer_t::iterator itr = completeds[i]->begin(); itr != completeds[i]->end(); itr++ ) {
         struct md_cache_block_future* f = *itr;
         md_cache_block_future_free( f );
      }
      
      SG_safe_delete( completeds[i] );
   }
   
   md_cache_lru_t* lrus[] = {
      cache->cache_lru,
      cache->promotes_1,
      cache->promotes_2,
      cache->evicts_1,
      cache->evicts_2,
      NULL
   };
   
   for( int i = 0; lrus[i] != NULL; i++ ) {
      SG_safe_delete( lrus[i] );
   }
   
   SG_safe_delete( cache->ongoing_writes );
   
   pthread_rwlock_t* locks[] = {
      &cache->pending_lock,
      &cache->completed_lock,
      &cache->cache_lru_lock,
      &cache->promotes_lock,
      &cache->ongoing_writes_lock,
      NULL
   };
   
   for( int i = 0; locks[i] != NULL; i++ ) {
      pthread_rwlock_destroy( locks[i] );
   }
   
   sem_destroy( &cache->sem_blocks_writing );
   sem_destroy( &cache->sem_write_hard_limit );
   
   return 0;
}

int
pthread_barrier_destroy (pthread_barrier_t * barrier)
{
  int result = 0;
  pthread_barrier_t b;
  ptw32_mcs_local_node_t node;

  if (barrier == NULL || *barrier == (pthread_barrier_t) PTW32_OBJECT_INVALID)
    {
      return EINVAL;
    }

  if (0 != ptw32_mcs_lock_try_acquire(&(*barrier)->lock, &node))
    {
      return EBUSY;
    }

  b = *barrier;

  if (b->nCurrentBarrierHeight < b->nInitialBarrierHeight)
    {
      result = EBUSY;
    }
  else
	{
      if (0 == (result = sem_destroy (&(b->semBarrierBreeched))))
        {
          *barrier = (pthread_barrier_t) PTW32_OBJECT_INVALID;
          /*
           * Release the lock before freeing b.
           *
           * FIXME: There may be successors which, when we release the lock,
           * will be linked into b->lock, which will be corrupted at some
           * point with undefined results for the application. To fix this
           * will require changing pthread_barrier_t from a pointer to
           * pthread_barrier_t_ to an instance. This is a change to the ABI
           * and will require a major version number increment.
           */
          ptw32_mcs_lock_release(&node);
	  (void) free (b);
	  return 0;
	}
      else
        {
          /*
           * This should not ever be reached.
           * Restore the barrier to working condition before returning.
           */
          (void) sem_init (&(b->semBarrierBreeched), b->pshared, 0);
        }

      if (result != 0)
        {
          /*
           * The barrier still exists and is valid
           * in the event of any error above.
           */
          result = EBUSY;
        }
    }

  ptw32_mcs_lock_release(&node);
  return (result);
}

//.........这里部分代码省略.........
#ifdef CONFIG_NET_ARP_SEND
  else
#endif
    {
      FAR const struct sockaddr_in6 *into;

      /* Make sure that the IP address mapping is in the Neighbor Table */

      into = (FAR const struct sockaddr_in6 *)to;
      ret = icmpv6_neighbor(into->sin6_addr.s6_addr16);
    }
#endif /* CONFIG_NET_ICMPv6_NEIGHBOR */

  /* Did we successfully get the address mapping? */

  if (ret < 0)
    {
      ndbg("ERROR: Not reachable\n");
      return -ENETUNREACH;
    }
#endif /* CONFIG_NET_ARP_SEND || CONFIG_NET_ICMPv6_NEIGHBOR */

  /* Set the socket state to sending */

  psock->s_flags = _SS_SETSTATE(psock->s_flags, _SF_SEND);

  /* Initialize the state structure.  This is done with interrupts
   * disabled because we don't want anything to happen until we
   * are ready.
   */

  save = net_lock();
  memset(&state, 0, sizeof(struct sendto_s));
  sem_init(&state.st_sem, 0, 0);
  state.st_buflen = len;
  state.st_buffer = buf;

#if defined(CONFIG_NET_SENDTO_TIMEOUT) || defined(NEED_IPDOMAIN_SUPPORT)
  /* Save the reference to the socket structure if it will be needed for
   * asynchronous processing.
   */

  state.st_sock = psock;
#endif

#ifdef CONFIG_NET_SENDTO_TIMEOUT
  /* Set the initial time for calculating timeouts */

  state.st_time = clock_systimer();
#endif

  /* Setup the UDP socket */

  conn = (FAR struct udp_conn_s *)psock->s_conn;
  DEBUGASSERT(conn);

  ret = udp_connect(conn, to);
  if (ret < 0)
    {
      net_unlock(save);
      return ret;
    }

  /* Set up the callback in the connection */

  state.st_cb = udp_callback_alloc(conn);
  if (state.st_cb)
    {
      state.st_cb->flags   = UDP_POLL;
      state.st_cb->priv    = (void*)&state;
      state.st_cb->event   = sendto_interrupt;

      /* Notify the device driver of the availability of TX data */

      sendto_txnotify(psock, conn);

      /* Wait for either the receive to complete or for an error/timeout to occur.
       * NOTES:  (1) net_lockedwait will also terminate if a signal is received, (2)
       * interrupts may be disabled!  They will be re-enabled while the task sleeps
       * and automatically re-enabled when the task restarts.
       */

      net_lockedwait(&state.st_sem);

      /* Make sure that no further interrupts are processed */

      udp_callback_free(conn, state.st_cb);
    }

  net_unlock(save);
  sem_destroy(&state.st_sem);

  /* Set the socket state to idle */

  psock->s_flags = _SS_SETSTATE(psock->s_flags, _SF_IDLE);

  /* Return the result of the sendto() operation */

  return state.st_sndlen;
}

Device::~Device()
{
	sem_destroy(&_lock);
}

int fclose(FAR FILE *stream)
{
  int err = EINVAL;
  int ret = ERROR;
  int status;

  /* Verify that a stream was provided. */

  if (stream)
    {
      /* Check that the underlying file descriptor corresponds to an an open
       * file.
       */

      ret = OK;
      if (stream->fs_fd >= 0)
        {
          /* If the stream was opened for writing, then flush the stream */

          if ((stream->fs_oflags & O_WROK) != 0)
            {
              ret = lib_fflush(stream, true);
              err = errno;
            }

          /* Close the underlying file descriptor and save the return status */

          status = close(stream->fs_fd);

          /* If close() returns an error but flush() did not then make sure
           * that we return the close() error condition.
           */

          if (ret == OK)
            {
              ret = status;
              err = errno;
            }
        }

#if CONFIG_STDIO_BUFFER_SIZE > 0
      /* Destroy the semaphore */

      sem_destroy(&stream->fs_sem);

      /* Release the buffer */

      if (stream->fs_bufstart)
        {
          lib_free(stream->fs_bufstart);
        }

      /* Clear the whole structure */

      memset(stream, 0, sizeof(FILE));
#else
#if CONFIG_NUNGET_CHARS > 0
      /* Reset the number of ungetc characters */

      stream->fs_nungotten = 0;
#endif
      /* Reset the flags */

      stream->fs_oflags = 0;
#endif
      /* Setting the file descriptor to -1 makes the stream available for reuse */

      stream->fs_fd = -1;
    }

  /* On an error, reset the errno to the first error encountered and return
   * EOF.
   */

  if (ret != OK)
    {
      set_errno(err);
      return EOF;
    }

  /* Return success */

  return OK;
}

static inline int tcp_connect(FAR struct socket *psock, const struct sockaddr_in *inaddr)
#endif
{
  struct tcp_connect_s state;
  uip_lock_t           flags;
  int                  ret = OK;

  /* Interrupts must be disabled through all of the following because
   * we cannot allow the network callback to occur until we are completely
   * setup.
   */

  flags = uip_lock();

  /* Get the connection reference from the socket */

  if (!psock->s_conn) /* Should always be non-NULL */
    {
      ret = -EINVAL;
    }
  else
    {
      /* Perform the uIP connection operation */

      ret = uip_tcpconnect(psock->s_conn, inaddr);
    }

  if (ret >= 0)
    {
      /* Set up the callbacks in the connection */

      ret = tcp_setup_callbacks(psock, &state);
      if (ret >= 0)
        {
          /* Wait for either the connect to complete or for an error/timeout
           * to occur. NOTES:  (1) uip_lockedwait will also terminate if a signal
           * is received, (2) interrupts may be disabled!  They will be re-
           * enabled while the task sleeps and automatically re-disabled
           * when the task restarts.
           */

          ret = uip_lockedwait(&state.tc_sem);

          /* Uninitialize the state structure */

          (void)sem_destroy(&state.tc_sem);

          /* If uip_lockedwait failed, recover the negated error (probably -EINTR) */

          if (ret < 0)
            {
              ret = -errno;
            }
          else
            {
              /* If the wait succeeded, then get the new error value from
               * the state structure
               */

              ret = state.tc_result;
            }

          /* Make sure that no further interrupts are processed */

          tcp_teardown_callbacks(&state, ret);
        }

      /* Mark the connection bound and connected */

      if (ret >= 0)
        {
          psock->s_flags |= (_SF_BOUND|_SF_CONNECTED);
        }
    }

    uip_unlock(flags);
    return ret;
}

void callbacks_cleanup() {
  for (size_t i = 0; i < ARRAY_SIZE(callback_data); ++i) {
    sem_destroy(&callback_data[i].semaphore);
  }
}

int main(int argc, char *argv[]){
	int num_producers;
	int num_consumers;
	int num_total_produced;
	int num_threads;
	int produced_per_thread;
	int consumed_per_thread;
	int i;
	pthread_t *producers;
	pthread_t *consumers;
	struct p_data *producer_data;
	
	if (argc != 4) {
	fprintf(stderr, "usage: a.out p c i\n");
	return 1;
	}
	
	buffer.elements = (int *) malloc(BUFFERLENGTH * sizeof(int));
	buffer.capacity = BUFFERLENGTH;
	buffer.size = 0;
	
	sem_init(&full, 0, 0);
	sem_init(&empty, 0, BUFFERLENGTH);
	
	if (pthread_mutex_init(&lock, NULL) != 0) {
	fprintf(stderr, "could not initialize mutex\n");
	return 1;
	}
	
	num_producers = pow(2, atoi(argv[1]));
	num_consumers = pow(2, atoi(argv[2]));
	num_total_produced = pow(2, atoi(argv[3]));
	num_threads = num_producers + num_consumers;
	
	printf("Number of producer threads: %i\nNumber of consumer threads: %i\nItems to produce: %i\n",
	num_producers, num_consumers, num_total_produced);
	
	produced_per_thread = num_total_produced / num_producers;
	consumed_per_thread = num_total_produced / num_consumers;
	
	producers = (pthread_t *) malloc(num_producers * sizeof(pthread_t));
	consumers = (pthread_t *) malloc(num_consumers * sizeof(pthread_t));
	producer_data = (struct p_data *) malloc(num_producers * sizeof(struct p_data));
	
	i = 0;
	while (i < num_producers) {
	producer_data[i].thread_number = i;
	producer_data[i].num_produced = produced_per_thread;
	if (pthread_create(&producers[i], NULL, &produce, &producer_data[i]) != 0) {
	fprintf(stderr, "error creating producer %i\n", i);
	return 1;
	}
	i++;
	}
	
	i = 0;
	while (i < num_consumers) {
	if (pthread_create(&consumers[i], NULL, &consume, &consumed_per_thread) != 0) {
	fprintf(stderr, "error creating consumer %i\n", i);
	return 1;
	}
	i++;
	}
	
	i = 0;
	while (i < num_producers) {
	pthread_join(producers[i], NULL);
	i++;
	}
	
	i = 0;
	while (i < num_consumers) {
	pthread_join(consumers[i], NULL);
	i++;
	}
	
	free(producers);
	free(consumers);
	free(producer_data);
	free(buffer.elements);
	pthread_mutex_destroy(&lock);
	sem_destroy(&full);
	sem_destroy(&empty);
	
	printf("Finished Successfully!\n");
	return 0;
	}

void AlienThread_destroy(Alien_Context *ac,
                         AlienThread   *thread)
{
    AlienThread *threadList = NULL;

    if( ( ac == NULL ) || ( thread == NULL ) )
    {
        return;
    }
    DBUGF(("Destroy thread %ld\n", thread->wthread));

    /* Remove from the thread list */
    threadList = ac->threadList;
    if( threadList != NULL )
    {
        if( threadList == thread )
        {
            /* Want to remove start of list */
            ac->threadList = ac->threadList->next;
        }
        else
        {
            /* Search for the thread we want to remove */
            while( ( threadList != NULL ) &&
                   ( threadList->next != thread ) )
            {
                threadList = threadList->next;
            }

             /* Should be in the list! */
            assert( threadList != NULL );
            assert( threadList->next != NULL );

            if( ( threadList != NULL ) &&
                ( threadList->next != NULL ) )
            {
                threadList->next = threadList->next->next;
            }
        }
    }

    if (thread != NULL)
    {
        if (thread == ac->mainThread)
        {
            (void)sem_destroy(&thread->semaphore);
            ac->mainThread = NULL;
        }
        else
        {
            /* Signal this thread's semaphore so it can
             * run to completion */
            thread->shuttingDown = true;
            sem_post(&thread->semaphore);

            /* Wait for the thread to finish */
            pthread_join(thread->wthread, NULL);
        }
    }

    free( thread );
}

AlienThread *AlienThread_create(Alien_Context         *ac,
                                AlienThread_StartFunc  startFn,
                                void                  *startData,
                                unsigned long          stackSize)
{
    AlienThread      *thread;
    AlienThreadSetup *setupData;
    sigset_t          newmask;
    sigset_t          oldmask;
    int               reterr;

    stackSize = stackSize; /* Unused, shush compiler */

    /* Allocate the AlienThread structure */
    thread = malloc(sizeof(*thread));
    if( thread == NULL )
    {
        return NULL;
    }
    /* Semaphore to control thread's execution */
    if (sem_init(&thread->semaphore, 0, 0) != 0)
    {
        DBUGF(("sem_init failed: %s\n", strerror(errno)));
        free(thread);
        return NULL;
    }

    /* This flag is set to tell the thread to shut itself down */
    thread->shuttingDown = false;

    /* Add to thread list */
    thread->next = ac->threadList;
    ac->threadList = thread;

    if (startFn != NULL)
    {
        /* Create a wrapper around the startFn/startData.  This allows us
         * to have our own start function (threadStart) and to keep
         * the thread restrained by a semaphore until it's required. */
        setupData = malloc(sizeof(*setupData));
        if (setupData == NULL)
        {
            free(thread);
            (void)sem_destroy(&thread->semaphore);
            return NULL;
        }

        setupData->thread    = thread;
        setupData->startData = startData;
        setupData->startFn   = startFn;

        /* Create the thread. We don't use the stack size - linux allocates
         * stack on demand as it is used. */

        /* block the SIGALRM signal from sending to the newly created thread,
         * signal mask will be inherited by child thread */
        reterr = sigemptyset(&newmask);
        assert (0 == reterr);
        reterr = sigaddset(&newmask, SIGALRM);
        assert (0 == reterr);
        reterr = pthread_sigmask(SIG_BLOCK, &newmask, &oldmask);
        if (reterr != 0)
        {
            DBUGF(("pthread_sigmask failed: %s\n", strerror(reterr)));
            assert("pthread_sigmask failed" == NULL);
            AlienThread_destroy(ac, thread);
            free(setupData);
            return NULL;
        }

        if(pthread_create(&thread->wthread, NULL, threadStart, setupData)
           != 0)
        {
            DBUGF(("pthread_create failed: %s\n", strerror(errno)));
            AlienThread_destroy(ac, thread);
            free(setupData);
            return NULL;
        }

        /* restore the old signal mask of this thread */
        reterr = pthread_sigmask(SIG_SETMASK, &oldmask, NULL);
        if (reterr != 0)
        {
            DBUGF(("pthread_create failed: %s\n", strerror(reterr)));
            assert("pthread_sigmask failed" == NULL);
        }
        /* Continue to run until told to switch in AlienThread_switch */
    }
    else
    {
        /* Setting up AlienThread based on current existing thread */
        thread->wthread = pthread_self();
    }

    DBUGF(("Created thread %ld\n", thread->wthread));

    return thread;
}

int vp8cx_create_encoder_threads(VP8_COMP *cpi) {
  const VP8_COMMON *cm = &cpi->common;

  cpi->b_multi_threaded = 0;
  cpi->encoding_thread_count = 0;
  cpi->b_lpf_running = 0;

  pthread_mutex_init(&cpi->mt_mutex, NULL);

  if (cm->processor_core_count > 1 && cpi->oxcf.multi_threaded > 1) {
    int ithread;
    int th_count = cpi->oxcf.multi_threaded - 1;
    int rc = 0;

    /* don't allocate more threads than cores available */
    if (cpi->oxcf.multi_threaded > cm->processor_core_count) {
      th_count = cm->processor_core_count - 1;
    }

    /* we have th_count + 1 (main) threads processing one row each */
    /* no point to have more threads than the sync range allows */
    if (th_count > ((cm->mb_cols / cpi->mt_sync_range) - 1)) {
      th_count = (cm->mb_cols / cpi->mt_sync_range) - 1;
    }

    if (th_count == 0) return 0;

    CHECK_MEM_ERROR(cpi->h_encoding_thread,
                    vpx_malloc(sizeof(pthread_t) * th_count));
    CHECK_MEM_ERROR(cpi->h_event_start_encoding,
                    vpx_malloc(sizeof(sem_t) * th_count));
    CHECK_MEM_ERROR(cpi->h_event_end_encoding,
                    vpx_malloc(sizeof(sem_t) * th_count));
    CHECK_MEM_ERROR(cpi->mb_row_ei,
                    vpx_memalign(32, sizeof(MB_ROW_COMP) * th_count));
    memset(cpi->mb_row_ei, 0, sizeof(MB_ROW_COMP) * th_count);
    CHECK_MEM_ERROR(cpi->en_thread_data,
                    vpx_malloc(sizeof(ENCODETHREAD_DATA) * th_count));

    cpi->b_multi_threaded = 1;
    cpi->encoding_thread_count = th_count;

    /*
    printf("[VP8:] multi_threaded encoding is enabled with %d threads\n\n",
           (cpi->encoding_thread_count +1));
    */

    for (ithread = 0; ithread < th_count; ++ithread) {
      ENCODETHREAD_DATA *ethd = &cpi->en_thread_data[ithread];

      /* Setup block ptrs and offsets */
      vp8_setup_block_ptrs(&cpi->mb_row_ei[ithread].mb);
      vp8_setup_block_dptrs(&cpi->mb_row_ei[ithread].mb.e_mbd);

      sem_init(&cpi->h_event_start_encoding[ithread], 0, 0);
      sem_init(&cpi->h_event_end_encoding[ithread], 0, 0);

      ethd->ithread = ithread;
      ethd->ptr1 = (void *)cpi;
      ethd->ptr2 = (void *)&cpi->mb_row_ei[ithread];

      rc = pthread_create(&cpi->h_encoding_thread[ithread], 0,
                          thread_encoding_proc, ethd);
      if (rc) break;
    }

    if (rc) {
      /* shutdown other threads */
      protected_write(&cpi->mt_mutex, &cpi->b_multi_threaded, 0);
      for (--ithread; ithread >= 0; ithread--) {
        pthread_join(cpi->h_encoding_thread[ithread], 0);
        sem_destroy(&cpi->h_event_start_encoding[ithread]);
        sem_destroy(&cpi->h_event_end_encoding[ithread]);
      }

      /* free thread related resources */
      vpx_free(cpi->h_event_start_encoding);
      vpx_free(cpi->h_event_end_encoding);
      vpx_free(cpi->h_encoding_thread);
      vpx_free(cpi->mb_row_ei);
      vpx_free(cpi->en_thread_data);

      pthread_mutex_destroy(&cpi->mt_mutex);

      return -1;
    }

    {
      LPFTHREAD_DATA *lpfthd = &cpi->lpf_thread_data;

      sem_init(&cpi->h_event_start_lpf, 0, 0);
      sem_init(&cpi->h_event_end_lpf, 0, 0);

      lpfthd->ptr1 = (void *)cpi;
      rc = pthread_create(&cpi->h_filter_thread, 0, thread_loopfilter, lpfthd);

      if (rc) {
        /* shutdown other threads */
        protected_write(&cpi->mt_mutex, &cpi->b_multi_threaded, 0);
        for (--ithread; ithread >= 0; ithread--) {
//.........这里部分代码省略.........

HXPthreadSemaphore::~HXPthreadSemaphore()
{
    sem_destroy( &m_semaphore );
}

static void utc_taskmanager_broadcast_p(void)
{
	int sleep_cnt = 0;
	tm_msg_t user_data;

	sem_init(&tm_broad_sem, 0, 1);
	broad_wifi_on_cnt = 0;
	broad_wifi_off_cnt = 0;
	broad_undefined_cnt = 0;
	broadcast_data_flag = -1;

	user_data.msg_size = strlen("WIFI_ON") + 1;
	user_data.msg = malloc(user_data.msg_size);
	strncpy(user_data.msg, "WIFI_ON", user_data.msg_size);

	(void)task_manager_broadcast(TM_BROADCAST_WIFI_ON, &user_data, TM_NO_RESPONSE);
	while (1) {
		sleep(1);
		if (broad_wifi_on_cnt == TM_BROAD_TASK_NUM) {
			break;
		}
		TC_ASSERT_LEQ_CLEANUP("task_manager_broadcast", sleep_cnt, 10, sem_destroy(&tm_broad_sem));
		sleep_cnt++;
	}
	TC_ASSERT_EQ_CLEANUP("task_manager_broadcast", broad_wifi_on_cnt, TM_BROAD_TASK_NUM, sem_destroy(&tm_broad_sem); free(user_data.msg));
	TC_ASSERT_EQ_CLEANUP("task_manager_broadcast", broad_wifi_off_cnt, 0, sem_destroy(&tm_broad_sem); free(user_data.msg));
	TC_ASSERT_EQ_CLEANUP("task_manager_broadcast", broad_undefined_cnt, 0, sem_destroy(&tm_broad_sem); free(user_data.msg));
	TC_ASSERT_EQ_CLEANUP("task_manager_broadcast", broadcast_data_flag, 0, sem_destroy(&tm_broad_sem); free(user_data.msg));
	free(user_data.msg);

	broad_wifi_on_cnt = 0;
	broad_wifi_off_cnt = 0;
	broad_undefined_cnt = 0;
	sleep_cnt = 0;
	(void)task_manager_broadcast(TM_BROADCAST_WIFI_OFF, NULL, TM_NO_RESPONSE);
	while (1) {
		usleep(500);
		if (broad_wifi_off_cnt == TM_BROAD_TASK_NUM) {
			break;
		}
		TC_ASSERT_LEQ_CLEANUP("task_manager_broadcast", sleep_cnt, 10, sem_destroy(&tm_broad_sem));
		sleep_cnt++;
	}
	TC_ASSERT_EQ_CLEANUP("task_manager_broadcast", broad_wifi_on_cnt, 0, sem_destroy(&tm_broad_sem));
	TC_ASSERT_EQ_CLEANUP("task_manager_broadcast", broad_wifi_off_cnt, TM_BROAD_TASK_NUM, sem_destroy(&tm_broad_sem));
	TC_ASSERT_EQ_CLEANUP("task_manager_broadcast", broad_undefined_cnt, 0, sem_destroy(&tm_broad_sem));

	broad_wifi_on_cnt = 0;
	broad_wifi_off_cnt = 0;
	broad_undefined_cnt = 0;
	sleep_cnt = 0;
	(void)task_manager_broadcast(tm_broadcast_undefined_msg, NULL, TM_NO_RESPONSE);
	while (1) {
		usleep(500);
		if (broad_undefined_cnt == TM_BROAD_UNDEF_NUM) {
			break;
		}
		TC_ASSERT_LEQ_CLEANUP("task_manager_broadcast", sleep_cnt, 10, sem_destroy(&tm_broad_sem));
		sleep_cnt++;
	}
	TC_ASSERT_EQ_CLEANUP("task_manager_broadcast", broad_wifi_on_cnt, 0, sem_destroy(&tm_broad_sem));
	TC_ASSERT_EQ_CLEANUP("task_manager_broadcast", broad_wifi_off_cnt, 0, sem_destroy(&tm_broad_sem));
	TC_ASSERT_EQ_CLEANUP("task_manager_broadcast", broad_undefined_cnt, TM_BROAD_UNDEF_NUM, sem_destroy(&tm_broad_sem));

	sem_destroy(&tm_broad_sem);
	TC_SUCCESS_RESULT();
}

/******************************************************************************
Description: function for sending back the result
Input Value.:
Return Value:
******************************************************************************/
void server_result (int sock, string userID)
{
    if (debug) printf("result thread\n\n");

    int n, fd;
    char response[] = "ok";
    sem_t *sem_match = new sem_t(); // create a new semaphore in heap
    queue<string> *imgQueue = 0;    // queue storing the file names 

    //  Init semaphore and put the address of semaphore into map
    if (sem_init(sem_match, 0, 0) != 0)
    {
        errorSocket("ERROR semaphore init failed", sock);
    }
    // grap the lock
    pthread_mutex_lock(&sem_map_lock);
    sem_map[userID] = sem_match;
    pthread_mutex_unlock(&sem_map_lock);

    // reponse to the client
    if (!orbit)
    {
        n = write(sock, response, sizeof(response));
        if (n < 0)
        {
            error("ERROR writting to socket");
        }
    }
    else
    {
        MsgD.send(sock, response, sizeof(response));
    }

    struct sockaddr_in myaddr;
    int ret;
    char buf[1024];
    int serverPort = 9879;
    
    if (storm)
    {

        if ((fd = socket(AF_INET, SOCK_DGRAM, 0)) == -1)
            printf("socket create failed\n");
        if (debug) printf("socket created\n");

        /* bind it to all local addresses and pick any port number */

        memset((char *)&myaddr, 0, sizeof(myaddr));
        myaddr.sin_family = AF_INET;
        myaddr.sin_addr.s_addr = htonl(INADDR_ANY);
        myaddr.sin_port = htons(serverPort);

        if (bind(fd, (struct sockaddr *)&myaddr, sizeof(myaddr)) < 0) {
            perror("bind failed");
            goto stop;
        }       
        if (debug) printf("socket binded\n");
    }

    while(!global_stop) 
    {
        sem_wait(sem_match);
        // get the address of image queue
        if (imgQueue == 0)
        {
            imgQueue = queue_map[userID];
        }

        // check if the queue is empty
        if (imgQueue->empty())
        {
            sem_map.erase(userID);
            queue_map.erase(userID);
            user_map.erase(userID);
            delete(sem_match);
            delete(imgQueue);
            sem_destroy(sem_match);
            // if (orbit)
            // {
            //     MsgD.close(sock, 0);                
            // }
            printf("[server] client disconnected --- result\n");
            // pthread_exit(NULL); //terminate calling thread!
            return;
        }

        if (!storm)
        {
            if (debug) printf("\n----------- start matching -------------\n");
            string file_name = imgQueue->front(); 
            if (debug) printf("file name: [%s]\n", file_name.c_str());
            imgQueue->pop();

            // create a new thread to do the image processing

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

//.........这里部分代码省略.........
  pipe_b[0] = -1;

  ret = file_detach(pipe_b[1], &devpair->pp_master.pd_sink);
  if (ret < 0)
    {
      goto errout_with_pipeb;
    }

  pipe_b[1] = -1;

  /* Register the slave device
   *
   * BSD style (deprecated): /dev/ttypN
   * SUSv1 style:  /dev/pts/N
   *
   * Where N is the minor number
   */

#ifdef CONFIG_PSEUDOTERM_BSD
  snprintf(devname, 16, "/dev/ttyp%d", minor);
#else
  snprintf(devname, 16, "/dev/pts/%d", minor);
#endif

  ret = register_driver(devname, &pty_fops, 0666, &devpair->pp_slave);
  if (ret < 0)
    {
      goto errout_with_pipeb;
    }

  /* Register the master device
   *
   * BSD style (deprecated):  /dev/ptyN
   * SUSv1 style: Master: /dev/ptmx (multiplexor, see ptmx.c)
   *
   * Where N is the minor number
   */

  snprintf(devname, 16, "/dev/pty%d", minor);

  ret = register_driver(devname, &pty_fops, 0666, &devpair->pp_master);
  if (ret < 0)
    {
      goto errout_with_slave;
    }

  return OK;

errout_with_slave:
#ifdef CONFIG_PSEUDOTERM_BSD
  snprintf(devname, 16, "/dev/ttyp%d", minor);
#else
  snprintf(devname, 16, "/dev/pts/%d", minor);
#endif
  (void)unregister_driver(devname);

errout_with_pipeb:
  if (pipe_b[0] >= 0)
    {
      close(pipe_b[0]);
    }
  else
    {
      (void)file_close_detached(&devpair->pp_master.pd_src);
    }

  if (pipe_b[1] >= 0)
    {
      close(pipe_b[1]);
    }
  else
    {
      (void)file_close_detached(&devpair->pp_slave.pd_sink);
    }

errout_with_pipea:
  if (pipe_a[0] >= 0)
    {
      close(pipe_a[0]);
    }
  else
    {
      (void)file_close_detached(&devpair->pp_slave.pd_src);
    }

  if (pipe_a[1] >= 0)
    {
      close(pipe_a[1]);
    }
  else
    {
      (void)file_close_detached(&devpair->pp_master.pd_sink);
    }

errout_with_devpair:
   sem_destroy(&devpair->pp_exclsem);
   sem_destroy(&devpair->pp_slavesem);
   kmm_free(devpair);
   return ret;
}

int ovs_db_send_request(ovs_db_t *pdb, const char *method, const char *params,
                        ovs_db_result_cb_t cb) {
  int ret = 0;
  yajl_gen_status yajl_gen_ret;
  yajl_val jparams;
  yajl_gen jgen;
  ovs_callback_t *new_cb = NULL;
  uint64_t uid;
  char uid_buff[OVS_UID_STR_SIZE];
  const char *req = NULL;
  size_t req_len = 0;
  struct timespec ts;

  /* sanity check */
  if (!pdb || !method || !params)
    return -1;

  if ((jgen = yajl_gen_alloc(NULL)) == NULL)
    return -1;

  /* try to parse params */
  if ((jparams = yajl_tree_parse(params, NULL, 0)) == NULL) {
    OVS_ERROR("params is not a JSON string");
    yajl_gen_clear(jgen);
    return -1;
  }

  /* generate method field */
  OVS_YAJL_CALL(yajl_gen_map_open, jgen);

  OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "method");
  OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, method);

  /* generate params field */
  OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "params");
  OVS_YAJL_CALL(ovs_yajl_gen_val, jgen, jparams);
  yajl_tree_free(jparams);

  /* generate id field */
  OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "id");
  uid = ovs_uid_generate();
  snprintf(uid_buff, sizeof(uid_buff), "%" PRIX64, uid);
  OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, uid_buff);

  OVS_YAJL_CALL(yajl_gen_map_close, jgen);

  if (cb) {
    /* register result callback */
    if ((new_cb = calloc(1, sizeof(ovs_callback_t))) == NULL)
      goto yajl_gen_failure;

    /* add new callback to front */
    sem_init(&new_cb->result.sync, 0, 0);
    new_cb->result.call = cb;
    new_cb->uid = uid;
    ovs_db_callback_add(pdb, new_cb);
  }

  /* send the request */
  OVS_YAJL_CALL(yajl_gen_get_buf, jgen, (const unsigned char **)&req, &req_len);
  OVS_DEBUG("%s", req);
  if (!ovs_db_data_send(pdb, req, req_len)) {
    if (cb) {
      /* wait for result */
      clock_gettime(CLOCK_REALTIME, &ts);
      ts.tv_sec += OVS_DB_SEND_REQ_TIMEOUT;
      if (sem_timedwait(&new_cb->result.sync, &ts) < 0) {
        OVS_ERROR("%s() no replay received within %d sec", __FUNCTION__,
                  OVS_DB_SEND_REQ_TIMEOUT);
        ret = (-1);
      }
    }
  } else {
    OVS_ERROR("ovs_db_data_send() failed");
    ret = (-1);
  }

yajl_gen_failure:
  if (new_cb) {
    /* destroy callback */
    sem_destroy(&new_cb->result.sync);
    ovs_db_callback_remove(pdb, new_cb);
  }

  /* release memory */
  yajl_gen_clear(jgen);
  return (yajl_gen_ret != yajl_gen_status_ok) ? (-1) : ret;
}

static void usbhost_disconnect_event(FAR void *arg)
{
  FAR struct usbhost_class_s *hubclass = (FAR struct usbhost_class_s *)arg;
  FAR struct usbhost_hubpriv_s *priv;
  FAR struct usbhost_hubport_s *hport;
  FAR struct usbhost_hubport_s *child;
  irqstate_t flags;
  int port;

  uvdbg("Disconnecting\n");

  DEBUGASSERT(hubclass != NULL && hubclass->hport != NULL);
  priv  = &((FAR struct usbhost_hubclass_s *)hubclass)->hubpriv;
  hport = hubclass->hport;

  uvdbg("Destroying hub on port  %d\n", hport->port);

  /* Set an indication to any users of the device that the device is no
   * longer available.
   */

  flags = irqsave();

  /* Cancel any pending transfers on the interrupt IN pipe */

  DRVR_CANCEL(hport->drvr, priv->intin);

  /* Cancel any pending port status change events */

  work_cancel(LPWORK, &priv->work);

  /* Disable power to all downstream ports */

  (void)usbhost_hubpwr(priv, hport, false);

  /* Free the allocated control request */

  DRVR_FREE(hport->drvr, (FAR uint8_t *)priv->ctrlreq);

  /* Free buffer for status change (INT) endpoint */

  DRVR_IOFREE(hport->drvr, priv->buffer);

  /* Destroy the interrupt IN endpoint */

  DRVR_EPFREE(hport->drvr, priv->intin);

  /* Release per-port resources */

  for (port = 0; port < USBHUB_MAX_PORTS; port++)
    {
      /* Free any devices classes connect on this hub port */

      child = &priv->hport[port];
      if (child->devclass != NULL)
        {
          CLASS_DISCONNECTED(child->devclass);
          child->devclass = NULL;
        }

      /* Free any resources used by the hub port */

      usbhost_hport_deactivate(child);
    }

  /* Deactivate the parent hub port (unless it is the root hub port) */

  usbhost_hport_deactivate(hport);

  /* Destroy the semaphores */

  sem_destroy(&priv->exclsem);

  /* Disconnect the USB host device */

  DRVR_DISCONNECT(hport->drvr, hport);

  /* Free the class instance */

  kmm_free(hubclass);
  hport->devclass = NULL;
  irqrestore(flags);
}

int __real_sem_destroy(sem_t * sem)
{
	return sem_destroy(sem);
}

void destroy_channel(channel_t *ch) {
	sem_destroy(&ch->reader_to_writer);
	sem_destroy(&ch->writer_to_reader);
	free(ch);
}