void *Consume (void *arg)
/* Consumer thread function. */
{
	msg_block_t *pmb;
	statistics_t * ps;
	int my_number, tstatus;
	struct timespec timeout, delta;
	
	delta.tv_sec = 2;
	delta.tv_nsec = 0;
		
	/* Create thread-specific storage key */
	tstatus = pthread_once (&once_control, once_init_function);
	if (tstatus != 0) err_abort (tstatus, "One time init failed");

	pmb = (msg_block_t *)arg;

	/* Allocate storage for thread-specific statistics */
	ps = calloc (sizeof(statistics_t), 1);
	if (ps == NULL) errno_abort ("Cannot allocate memory");
	tstatus = pthread_setspecific (ts_key, ps);
	if (tstatus != 0) err_abort (tstatus, "Error setting ts storage");
	ps->pmblock = pmb;
	/* Give this thread a unique number */
	/* Note that the mutex is "overloaded" to protect data	*/
	/* outside the message block structure			*/
	tstatus = pthread_mutex_lock (&pmb->nGuard);
	if (tstatus != 0) err_abort (tstatus, "Lock error");
	ps->th_number = thread_number++;
	tstatus = pthread_mutex_unlock (&pmb->nGuard);
	if (tstatus != 0) err_abort (tstatus, "Unlock error");
	
	/* Consume the NEXT message when prompted by the user */
	while (!pmb->fStop) { /* This is the only thread accessing stdin, stdout */
		tstatus = pthread_mutex_lock (&pmb->nGuard);
		if (tstatus != 0) err_abort (tstatus, "Lock error");
		/* Get the next message. Use a timed wait so as to be able	*/
		/* to sample the stop flag peridically.				*/
		do { 
			pthread_get_expiration_np (&delta, &timeout);
			tstatus = pthread_cond_timedwait 
				(&pmb->mReady, &pmb->nGuard, &timeout);
			if (tstatus != 0 && tstatus != ETIMEDOUT) 
				err_abort (tstatus, "CV wait error");
		} while (!pmb->f_ready && !pmb->fStop);
		if (!pmb->fStop) {
/*			printf ("Message received\n"); */
			accumulate_statistics ();
			pmb->f_consumed = 1;
			pmb->f_ready = 0;
		}
		tstatus = pthread_cond_signal (&pmb->mconsumed);
		if (tstatus != 0) err_abort (tstatus, "Signal error");
		tstatus = pthread_mutex_unlock (&pmb->nGuard);
		if (tstatus != 0) err_abort (tstatus, "Unlock error");
	}
	
	/* Shutdown. Report the statistics */
	tstatus = pthread_mutex_lock (&pmb->nGuard);
	if (tstatus != 0) err_abort (tstatus, "Lock error");
	report_statistics ();	
	tstatus = pthread_mutex_unlock (&pmb->nGuard);
	if (tstatus != 0) err_abort (tstatus, "Unlock error");

	/* Terminate the consumer thread. The destructor will 	*/
	/* free the memory allocated for the statistics		*/
	return NULL;		
}

void setGlThreadSpecific(gl_hooks_t const *value) {
    pthread_setspecific(gGLWrapperKey, value);
}

static void * s_signalListener( void * my_stack )
{
   static HB_BOOL s_bFirst = HB_TRUE;

   sigset_t   passall;
   HB_STACK * pStack = ( HB_STACK * ) my_stack;

#if defined( HB_OS_BSD )
   int sig;
#else
   siginfo_t sinfo;
#endif

#ifdef HB_THREAD_TLS_KEYWORD
   hb_thread_stack = my_stack;
#else
   pthread_setspecific( hb_pkCurrentStack, my_stack );
#endif

   pStack->th_id = HB_CURRENT_THREAD();
   hb_threadLinkStack( pStack );
   HB_STACK_LOCK;

   /* and now accepts all signals */
   sigemptyset( &passall );

   /* and wait for all signals */
   sigaddset( &passall, SIGHUP );
   sigaddset( &passall, SIGQUIT );
   sigaddset( &passall, SIGILL );
   sigaddset( &passall, SIGABRT );
   sigaddset( &passall, SIGFPE );
   sigaddset( &passall, SIGSEGV );
   sigaddset( &passall, SIGTERM );
   sigaddset( &passall, SIGUSR1 );
   sigaddset( &passall, SIGUSR2 );
   sigaddset( &passall, SIGHUP );

   pthread_cleanup_push( hb_threadTerminator, my_stack );
   pthread_setcanceltype( PTHREAD_CANCEL_DEFERRED, NULL );
   pthread_setcancelstate( PTHREAD_CANCEL_ENABLE, NULL );

   for( ;; )
   {
      /* allow safe cancelation */
      HB_STACK_UNLOCK;

      /* reset signal handling; this is done here (so I don't
         mangle with pthread_ calls, and I don't hold mutexes),
         and just once (doing it twice would be useless). */
      if( s_bFirst )
      {
         pthread_sigmask( SIG_SETMASK, &passall, NULL );
         s_bFirst = HB_FALSE;
      }

      /* This is also a cancelation point. When the main thread
         is done, it will kill all the threads having a stack
         including this one.
         ATM we don't care very much about signal handling during
         termination: no handler is set for them, so the DFL
         action is taken (and that should be fine). */
#if defined( HB_OS_BSD )
      sigwait( &passall, &sig );
#else
      sigwaitinfo( &passall, &sinfo );
#endif

      /* lock stack before passing the ball to VM. */
      HB_STACK_LOCK;
#if defined( HB_OS_BSD )
      s_signalHandler( sig, NULL, NULL );
#else
      s_signalHandler( sinfo.si_signo, &sinfo, NULL );
#endif
   }

   pthread_cleanup_pop( 1 );
   return 0;
}

void ThreadIdentifierData::initialize(ThreadIdentifier id)
{
    ASSERT(!identifier());
    pthread_setspecific(m_key, new ThreadIdentifierData(id));
}

_X_HIDDEN void
__glXSetCurrentContext(__GLXcontext * c)
{
   pthread_once(&once_control, init_thread_data);
   pthread_setspecific(ContextTSD, c);
}

static CVMBool
POSIXthreadSetSelf(const CVMThreadID *self)
{
    return (pthread_setspecific(key, self) == 0);
}

static int async_die_is_recursing(void)
{
	void *ret = pthread_getspecific(async_die_counter);
	pthread_setspecific(async_die_counter, (void *)1);
	return ret != NULL;
}

static DThreadData* DThreadData_New()
{
	DThreadData *self = (DThreadData*) dao_calloc( 1, sizeof(DThreadData) );
	pthread_setspecific( thdSpecKey, self );
	return self;
}

void * entry (void *arg)
{
  pthread_setspecific(key,
                      (void *)GC_HIDE_POINTER(GC_STRDUP("hello, world")));
  return arg;
}

void ThreadKey::setKey(void *ptr)
{
    if(key != KEY_INVALID)
        pthread_setspecific(key, ptr);
}

void DThread_Destroy( DThread *self )
{
	DMutex_Destroy( & self->mutex );
	DCondVar_Destroy( & self->condv );
	pthread_setspecific( thdSpecKey, NULL );
}

void tls_set(TLS_DATA *tls_data) {
    pthread_setspecific(pthread_key, tls_data);
}

/**
 * Sets a thread-specific variable.
 * @param key thread-local variable key (created with vlc_threadvar_create())
 * @param value new value for the variable for the calling thread
 * @return 0 on success, a system error code otherwise.
 */
int vlc_threadvar_set (vlc_threadvar_t key, void *value)
{
    return pthread_setspecific (key, value);
}

static void strerr_storage_init(void) {
	zassert(pthread_key_create(&strerrstorage,strerr_storage_free));
	zassert(pthread_setspecific(strerrstorage,NULL));
}

static inline void
birdloop_set_current(struct birdloop *loop)
{
  pthread_setspecific(current_loop_key, loop);
}

static void gil_asyncio_get() {
	pthread_setspecific(up.upt_gil_key, (void *) PyGILState_Ensure());
}

static void CmiStartThreads(char **argv)
{
  pthread_t pid;
  size_t i;
  int ok, tocreate;
  pthread_attr_t attr;
  int start, end;

  MACHSTATE(4,"CmiStartThreads")
  CmiMemLock_lock=CmiCreateLock();
  comm_mutex=CmiCreateLock();
  _smp_mutex = CmiCreateLock();
#if defined(CMK_NO_ASM_AVAILABLE) && CMK_PCQUEUE_LOCK
  cmiMemoryLock = CmiCreateLock();
  if (CmiMyNode()==0) printf("Charm++ warning> fences and atomic operations not available in native assembly\n");
#endif

#if ! (CMK_HAS_TLS_VARIABLES && !CMK_NOT_USE_TLS_THREAD)
  pthread_key_create(&Cmi_state_key, 0);
  Cmi_state_vector =
    (CmiState)calloc(_Cmi_mynodesize+1, sizeof(struct CmiStateStruct));
  for (i=0; i<_Cmi_mynodesize; i++)
    CmiStateInit(i+Cmi_nodestart, i, CmiGetStateN(i));
  /*Create a fake state structure for the comm. thread*/
/*  CmiStateInit(-1,_Cmi_mynodesize,CmiGetStateN(_Cmi_mynodesize)); */
  CmiStateInit(_Cmi_mynode+CmiNumPes(),_Cmi_mynodesize,CmiGetStateN(_Cmi_mynodesize));
#else
    /* for main thread */
  Cmi_state_vector = (CmiState *)calloc(_Cmi_mynodesize+1, sizeof(CmiState));
#if CMK_CONVERSE_MPI
      /* main thread is communication thread */
  if(!CharmLibInterOperate) {
    CmiStateInit(_Cmi_mynode+CmiNumPes(), _Cmi_mynodesize, &Cmi_mystate);
    Cmi_state_vector[_Cmi_mynodesize] = &Cmi_mystate;
  } else 
#endif
  {
    /* main thread is of rank 0 */
    CmiStateInit(Cmi_nodestart, 0, &Cmi_mystate);
    Cmi_state_vector[0] = &Cmi_mystate;
  }
#endif

#if CMK_MULTICORE || CMK_SMP_NO_COMMTHD
  if (!Cmi_commthread)
    tocreate = _Cmi_mynodesize-1;
  else
#endif
  tocreate = _Cmi_mynodesize;
#if CMK_CONVERSE_MPI
  if(!CharmLibInterOperate) {
    start = 0;
    end = tocreate - 1;                    /* skip comm thread */
  } else 
#endif
  {
    start = 1;
    end = tocreate;                       /* skip rank 0 main thread */
  }
  for (i=start; i<=end; i++) {        
    pthread_attr_init(&attr);
    pthread_attr_setscope(&attr, PTHREAD_SCOPE_SYSTEM);
    ok = pthread_create(&pid, &attr, call_startfn, (void *)i);
    if (ok<0) PerrorExit("pthread_create"); 
    pthread_attr_destroy(&attr);
  }
#if ! (CMK_HAS_TLS_VARIABLES && !CMK_NOT_USE_TLS_THREAD)
#if CMK_CONVERSE_MPI
  if(!CharmLibInterOperate)
    pthread_setspecific(Cmi_state_key, Cmi_state_vector+_Cmi_mynodesize);
  else 
#endif
    pthread_setspecific(Cmi_state_key, Cmi_state_vector);
#endif

  MACHSTATE(4,"CmiStartThreads done")
}

void uwsgi_proto_zeromq_setup(struct uwsgi_socket *uwsgi_sock) {

	           char *responder = strchr(uwsgi_sock->name, ',');
                        if (!responder) {
                                uwsgi_log("invalid zeromq address\n");
                                exit(1);
                        }
                        uwsgi_sock->receiver = uwsgi_concat2n(uwsgi_sock->name, responder - uwsgi_sock->name, "", 0);
                        responder++;

                        uwsgi_sock->pub = zmq_socket(uwsgi.zmq_context, ZMQ_PUB);
                        if (uwsgi_sock->pub == NULL) {
                                uwsgi_error("zmq_socket()");
                                exit(1);
                        }


                        // generate uuid
                        uuid_t uuid_zmq;
                        uuid_generate(uuid_zmq);
                        uuid_unparse(uuid_zmq, uwsgi_sock->uuid);

                        if (zmq_setsockopt(uwsgi_sock->pub, ZMQ_IDENTITY, uwsgi_sock->uuid, 36) < 0) {
                                uwsgi_error("zmq_setsockopt()");
                                exit(1);
                        }

                        if (zmq_connect(uwsgi_sock->pub, responder) < 0) {
                                uwsgi_error("zmq_connect()");
                                exit(1);
                        }

                        uwsgi_log("zeromq UUID for responder %s on worker %d: %.*s\n", responder, uwsgi.mywid, 36, uwsgi_sock->uuid);

                        uwsgi_sock->proto = uwsgi_proto_zeromq_parser;
                        uwsgi_sock->proto_accept = uwsgi_proto_zeromq_accept;
                        uwsgi_sock->proto_close = uwsgi_proto_zeromq_close;
                        uwsgi_sock->proto_write = uwsgi_proto_zeromq_write;
                        uwsgi_sock->proto_writev = uwsgi_proto_zeromq_writev;
                        uwsgi_sock->proto_write_header = uwsgi_proto_zeromq_write_header;
                        uwsgi_sock->proto_writev_header = uwsgi_proto_zeromq_writev_header;
                        uwsgi_sock->proto_sendfile = uwsgi_proto_zeromq_sendfile;

                        uwsgi_sock->proto_thread_fixup = uwsgi_proto_zeromq_thread_fixup;

                        uwsgi_sock->edge_trigger = 1;
                        uwsgi_sock->retry = uwsgi_malloc(sizeof(int) * uwsgi.threads);
			uwsgi_sock->retry[0] = 1;

                        // inform loop engine about edge trigger status
                        uwsgi.is_et = 1;


                        // initialize a lock for multithread usage
                        if (uwsgi.threads > 1) {
                                pthread_mutex_init(&uwsgi_sock->lock, NULL);
                        }

                        // one pull per-thread
                        if (pthread_key_create(&uwsgi_sock->key, NULL)) {
                                uwsgi_error("pthread_key_create()");
                                exit(1);
                        }

                        void *tmp_zmq_pull = zmq_socket(uwsgi.zmq_context, ZMQ_PULL);
                        if (tmp_zmq_pull == NULL) {
                                uwsgi_error("zmq_socket()");
                                exit(1);
                        }
                        if (zmq_connect(tmp_zmq_pull, uwsgi_sock->receiver) < 0) {
                                uwsgi_error("zmq_connect()");
                                exit(1);
                        }

                        pthread_setspecific(uwsgi_sock->key, tmp_zmq_pull);

#ifdef ZMQ_FD
                        size_t zmq_socket_len = sizeof(int);
                        if (zmq_getsockopt(pthread_getspecific(uwsgi_sock->key), ZMQ_FD, &uwsgi_sock->fd, &zmq_socket_len) < 0) {
                                uwsgi_error("zmq_getsockopt()");
                                exit(1);
                        }
			if (uwsgi.threads > 1) {
				uwsgi_sock->fd_threads = uwsgi_malloc(sizeof(int) * uwsgi.threads);
				uwsgi_sock->fd_threads[0] = uwsgi_sock->fd;
			}
#else
                        uwsgi_sock->fd = -1;
#endif

                        uwsgi_sock->bound = 1;
#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(3,0,0)
                        uwsgi_sock->recv_flag = ZMQ_DONTWAIT;
#else
                        uwsgi_sock->recv_flag = ZMQ_NOBLOCK;
#endif
}

void *wxThreadInternal::PthreadStart(wxThread *thread)
{
    wxThreadInternal *pthread = thread->m_internal;

    wxLogTrace(TRACE_THREADS, _T("Thread %ld started."), THR_ID(pthread));

    // associate the thread pointer with the newly created thread so that
    // wxThread::This() will work
    int rc = pthread_setspecific(gs_keySelf, thread);
    if ( rc != 0 )
    {
        wxLogSysError(rc, _("Cannot start thread: error writing TLS"));

        return (void *)-1;
    }

    // have to declare this before pthread_cleanup_push() which defines a
    // block!
    bool dontRunAtAll;

#ifdef wxHAVE_PTHREAD_CLEANUP
    // install the cleanup handler which will be called if the thread is
    // cancelled
    pthread_cleanup_push(wxPthreadCleanup, thread);
#endif // wxHAVE_PTHREAD_CLEANUP

    // wait for the semaphore to be posted from Run()
    pthread->m_semRun.Wait();

    // test whether we should run the run at all - may be it was deleted
    // before it started to Run()?
    {
        wxCriticalSectionLocker lock(thread->m_critsect);

        dontRunAtAll = pthread->GetState() == STATE_NEW &&
                       pthread->WasCancelled();
    }

    if ( !dontRunAtAll )
    {
        // call the main entry
        wxLogTrace(TRACE_THREADS,
                   _T("Thread %ld about to enter its Entry()."),
                   THR_ID(pthread));

        pthread->m_exitcode = thread->Entry();

        wxLogTrace(TRACE_THREADS,
                   _T("Thread %ld Entry() returned %lu."),
                   THR_ID(pthread), (unsigned long)pthread->m_exitcode);

        {
            wxCriticalSectionLocker lock(thread->m_critsect);

            // change the state of the thread to "exited" so that
            // wxPthreadCleanup handler won't do anything from now (if it's
            // called before we do pthread_cleanup_pop below)
            pthread->SetState(STATE_EXITED);
        }
    }

    // NB: at least under Linux, pthread_cleanup_push/pop are macros and pop
    //     contains the matching '}' for the '{' in push, so they must be used
    //     in the same block!
#ifdef wxHAVE_PTHREAD_CLEANUP
    // remove the cleanup handler without executing it
    pthread_cleanup_pop(FALSE);
#endif // wxHAVE_PTHREAD_CLEANUP

    if ( dontRunAtAll )
    {
        // FIXME: deleting a possibly joinable thread here???
        delete thread;

        return EXITCODE_CANCELLED;
    }
    else
    {
        // terminate the thread
        thread->Exit(pthread->m_exitcode);

        wxFAIL_MSG(wxT("wxThread::Exit() can't return."));

        return NULL;
    }
}

void *azbox_main_thread(void *cli)
{
	struct s_client *client = (struct s_client *) cli;
	client->thread = pthread_self();
	pthread_setspecific(getclient, cli);
	dvbapi_client = cli;

	struct s_auth *account;
	int32_t ok = 0;
	for(account = cfg.account; account; account = account->next)
	{
		if((ok = streq(cfg.dvbapi_usr, account->usr)))
			{ break; }
	}
	cs_auth_client(client, ok ? account : (struct s_auth *)(-1), "dvbapi");

	dvbapi_read_priority();

	openxcas_msg_t msg;
	int32_t ret;
	while((ret = openxcas_get_message(&msg, 0)) >= 0)
	{
		cs_sleepms(10);

		if(ret)
		{
			openxcas_stream_id = msg.stream_id;
			openxcas_seq = msg.sequence;

			switch(msg.cmd)
			{
			case OPENXCAS_SELECT_CHANNEL:
				cs_debug_mask(D_DVBAPI, LOG_PREFIX_MSG "OPENXCAS_SELECT_CHANNEL");

				// parse channel info
				struct stOpenXCASChannel chan;
				memcpy(&chan, msg.buf, msg.buf_len);

				cs_log(LOG_PREFIX "channel change: sid = %x, vpid = %x. apid = %x", chan.service_id, chan.v_pid, chan.a_pid);

				openxcas_video_pid = chan.v_pid;
				openxcas_audio_pid = chan.a_pid;
				openxcas_data_pid = chan.d_pid;
				break;
			case OPENXCAS_START_PMT_ECM:
				cs_debug_mask(D_DVBAPI, LOG_PREFIX_MSG "OPENXCAS_START_PMT_ECM");

				// parse pmt
				uchar *dest;
				if(!cs_malloc(&dest, msg.buf_len + 7 - 12 - 4))
					{ break; }

				memcpy(dest, "\x00\xFF\xFF\x00\x00\x13\x00", 7);

				dest[1] = msg.buf[3];
				dest[2] = msg.buf[4];
				dest[5] = msg.buf[11] + 1;

				memcpy(dest + 7, msg.buf + 12, msg.buf_len - 12 - 4);

				dvbapi_parse_capmt(dest, 7 + msg.buf_len - 12 - 4, -1, NULL);
				free(dest);

				unsigned char mask[12];
				unsigned char comp[12];
				memset(&mask, 0x00, sizeof(mask));
				memset(&comp, 0x00, sizeof(comp));

				mask[0] = 0xfe;
				comp[0] = 0x80;

				if((ret = openxcas_add_filter(msg.stream_id, OPENXCAS_FILTER_ECM, 0, 0xffff, openxcas_ecm_pid, mask, comp, (void *)azbox_openxcas_ecm_callback)) < 0)
					{ cs_log(LOG_PREFIX "unable to add ecm filter"); }
				else
					{ cs_debug_mask(D_DVBAPI, LOG_PREFIX "ecm filter added, pid = %x, caid = %x", openxcas_ecm_pid, 0); }

				if(openxcas_start_filter(msg.stream_id, msg.sequence, OPENXCAS_FILTER_ECM) < 0)
					{ cs_log(LOG_PREFIX "unable to start ecm filter"); }
				else
					{ cs_debug_mask(D_DVBAPI, LOG_PREFIX "ecm filter started"); }

				if(!openxcas_create_cipher_ex(msg.stream_id, openxcas_seq, 0, openxcas_ecm_pid, openxcas_video_pid, 0xffff, openxcas_audio_pid, 0xffff, 0xffff, 0xffff))
					{ cs_log(LOG_PREFIX "failed to create cipher ex"); }
				else
					{ cs_debug_mask(D_DVBAPI, LOG_PREFIX "cipher created"); }
				break;
			case OPENXCAS_STOP_PMT_ECM:
				cs_debug_mask(D_DVBAPI, LOG_PREFIX_MSG "OPENXCAS_STOP_PMT_ECM");
				openxcas_stop_filter(msg.stream_id, OPENXCAS_FILTER_ECM);
				openxcas_remove_filter(msg.stream_id, OPENXCAS_FILTER_ECM);
				openxcas_stop_filter_ex(msg.stream_id, msg.sequence, openxcas_filter_idx);
				openxcas_destory_cipher_ex(msg.stream_id, msg.sequence);
				memset(&demux, 0, sizeof(demux));
				break;
			case OPENXCAS_ECM_CALLBACK:
				cs_debug_mask(D_DVBAPI, LOG_PREFIX_MSG "OPENXCAS_ECM_CALLBACK");
				struct stOpenXCAS_Data data;
				memcpy(&data, msg.buf, msg.buf_len);
				if(!openxcas_busy)
					{ openxcas_filter_callback(msg.stream_id, msg.sequence, OPENXCAS_FILTER_ECM, &data); }
//.........这里部分代码省略.........