/******************************************************************************
Description.: this is the main worker thread
              it loops forever, grabs a fresh frame and stores it to file
Input Value.:
Return Value:
******************************************************************************/
void *worker_thread(void *arg)
{
    int ok = 1;
    int frame_size = 0;
    unsigned char *tmp_framebuffer = NULL;

    struct sockaddr_in addr;
    int sd;

    bzero(&addr, sizeof(addr));


    /* set cleanup handler to cleanup allocated ressources */
    pthread_cleanup_push(worker_cleanup, NULL);

    // set TCP server data structures ---------------------------
    if(port <= 0) {
        OPRINT("a valid TCP port must be provided\n");
        return NULL;
    }

    addr.sin_addr.s_addr = inet_addr(server);
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    // -----------------------------------------------------------

    struct timeval imageProcessingStart;
    struct timeval imageProcessingStop;

    struct timeval socketStart;
    struct timeval socketStop;

    while (ok >= 0 && !pglobal->stop) {
        //DBG("waiting for fresh frame\n");

        gettimeofday(&imageProcessingStart, NULL);

        pthread_mutex_lock(&pglobal->in[input_number].db);
        pthread_cond_wait(&pglobal->in[input_number].db_update, &pglobal->in[input_number].db);


        /* read buffer */
        frame_size = pglobal->in[input_number].size;

        /* check if buffer for frame is large enough, increase it if necessary */
        if(frame_size > max_frame_size) {
            DBG("increasing buffer size to %d\n", frame_size);

            max_frame_size = frame_size + (1 << 16);
            if((tmp_framebuffer = realloc(frame, max_frame_size)) == NULL) {
                pthread_mutex_unlock(&pglobal->in[input_number].db);
                LOG("not enough memory\n");
                return NULL;
            }

            frame = tmp_framebuffer;
        }

        /* copy frame to our local buffer now */
        memcpy(frame, pglobal->in[input_number].buf, frame_size);

        /* allow others to access the global buffer again */
        pthread_mutex_unlock(&pglobal->in[input_number].db);

        gettimeofday(&imageProcessingStop, NULL);

        printDuration(&imageProcessingStart, &imageProcessingStop, "Image");

        /* Send image */
        gettimeofday(&socketStart, NULL);

        DBG("Create connection to %s:%d\n", server, port);
        sd = socket(AF_INET, SOCK_STREAM, 0);

        // Test
        usleep(200 * 1000);

        if (connect(sd , (struct sockaddr*) &addr , sizeof(addr)) == 0) {
            DBG("Connection to %s:%d established\n", server, port);

            if (!send(sd, frame, frame_size, 0)) {
                perror("Image was not send");
            }

            DBG("Closing connection to %s:%d\n", server, port);
            close(sd);
        } else {
            perror("connect");
        }

        gettimeofday(&socketStop, NULL);

        printDuration(&socketStart, &socketStop, "Socket");
    }
//.........这里部分代码省略.........

void *channel_pth(void *item)
{
	uint32_t i,ret,dglen;
	uint32_t fd;
	uint32_t data_per_sec = 0;
	struct timeval vtime;
	struct msg_s2c_st *msg2c = NULL;
	struct sockaddr_in his_addr;
	struct channel_item *ptr = item;
	glob_t res;
	char patstr[PATSTRSIZE];
	sigset_t set,oset;
	/* block sigals */
	sigfillset(&set);
	pthread_sigmask(SIG_UNBLOCK,&set,&oset);
	pthread_sigmask(SIG_BLOCK,&set,NULL);
	/* set address */
	his_addr.sin_family = AF_INET;
	his_addr.sin_port = htons(args.recieve_port);
	inet_pton(AF_INET,args.multicast_group,&his_addr.sin_addr);
	/* start to send data */
	memset(patstr,0,PATSTRSIZE);
	strcat(patstr,ptr->path);
	strcat(patstr,PATTERN);
	if(glob(patstr,0,NULL,&res)!=0){
		syslog(LOG_ERR,"%s\n","no mp3 item found.");
		pthread_exit(NULL);
	}
	pthread_cleanup_push(glob_free,&res);
	if(((msg2c=malloc(sizeof(*msg2c)+SBUFSIZE-1))==NULL)){
		syslog(LOG_ERR,"%s\n","failed to allocate memory.");
		globfree(&res);
		pthread_exit(NULL);
	}
	pthread_cleanup_push(do_free,msg2c);
	for(i=0;i<res.gl_pathc;){
		if((fd=open(res.gl_pathv[i],O_RDONLY))<0){
			syslog(LOG_ERR,"failed to open %s.\n",ptr->path);
			free(msg2c);
			globfree(&res);
			pthread_exit(NULL);
		}
		pthread_cleanup_push(do_close,(void*)fd);
		while(1){
			if(data_per_sec<32768){
				memset(msg2c->data,0,SBUFSIZE);
				ret=read(fd,msg2c->data,SBUFSIZE);
				data_per_sec+=ret;
				if(ret==0){
					break;
				}
				msg2c->channel_id = htons(ptr->id);
				dglen = ret + sizeof(msg2c->channel_id);
				while(sendto(sd, msg2c, dglen, 0, (void*)&his_addr, sizeof(his_addr))<0) {
					if(errno==EINTR){
						continue;
					}
					syslog(LOG_ERR,"%s\n","failed to send message.");
					free(msg2c);
					globfree(&res);
					pthread_exit(NULL);
				}
			}else {
				/* init time */
				vtime.tv_sec = 1;
				vtime.tv_usec = 0;
				select(0,NULL,NULL,NULL,&vtime);
				data_per_sec = 0;
			}
		}
		pthread_cleanup_pop(1);
		i++;
		if(i==res.gl_pathc){
			i = 0;
		}
	}
	pthread_sigmask(SIG_SETMASK,&oset,NULL);
	pthread_cleanup_pop(1);
	pthread_cleanup_pop(1);
	syslog(LOG_ERR,"i=%d\t%s\n",i,"exception exit.");
	pthread_exit(NULL);
}

void *stapi_read_thread(void *sparam) {
	int32_t dev_index, ErrorCode, i, j, CRCValid;
	uint32_t QueryBufferHandle = 0, DataSize = 0;
	uchar buf[BUFFLEN];

	struct read_thread_param *para = sparam;
	dev_index = para->id;

	pthread_setspecific(getclient, para->cli);
	pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
	pthread_cleanup_push(stapi_cleanup_thread, (void*) dev_index);

	int32_t error_count = 0;

	while (1) {
		QueryBufferHandle = 0;
		ErrorCode = oscam_stapi_SignalWaitBuffer(dev_list[dev_index].SignalHandle, &QueryBufferHandle, 1000);

		switch (ErrorCode) {
			case 0: // NO_ERROR:
				break;
			case 852042: // ERROR_SIGNAL_ABORTED
				cs_log("Caught abort signal");
				pthread_exit(NULL);
				break;
			case 11: // ERROR_TIMEOUT:
				//cs_log("timeout %d", dev_index);
				//TODO: if pidindex == -1 try next
				continue;
				break;
			default:
				if (QueryBufferHandle != 0) {
					cs_log("SignalWaitBuffer error: %d", ErrorCode);
					oscam_stapi_BufferFlush(QueryBufferHandle);
					continue;
				}
				cs_log("SignalWaitBuffer: index %d ErrorCode: %d - QueryBuffer: %x", dev_index, ErrorCode, QueryBufferHandle);
				error_count++;
				if (error_count>10) {
					cs_log("Too many errors in reader thread %d, quitting.", dev_index);
					pthread_exit(NULL);
				}
				continue;
				break;
		}

		uint32_t NumFilterMatches = 0;
		int32_t demux_id = 0, filter_num = 0;
		DataSize = 0;
		uint32_t k;

		uint32_t MatchedFilterList[10];
		ErrorCode = oscam_stapi_BufferReadSection(QueryBufferHandle, MatchedFilterList, 10, &NumFilterMatches, &CRCValid, buf, BUFFLEN, &DataSize);

		if (ErrorCode != 0) {
			cs_log("BufferRead: index: %d ErrorCode: %d", dev_index, ErrorCode);
			cs_sleepms(1000);
			continue;
		}

		if (DataSize<=0)
			continue;

		pthread_mutex_lock(&filter_lock); // don't use cs_lock() here; multiple threads using same s_client struct
		for(k=0;k<NumFilterMatches;k++) {
			for (i=0;i<MAX_DEMUX;i++) {
				for (j=0;j<MAX_FILTER;j++) {
					if (dev_list[dev_index].demux_fd[i][j].fd == MatchedFilterList[k]) {
						demux_id=i;
						filter_num=j;

						dvbapi_process_input(demux_id, filter_num, buf, DataSize);
					}
				}
			}
		}
		pthread_mutex_unlock(&filter_lock);
	}
	pthread_cleanup_pop(0);
}

static void *thread_func_exit(void *arg) {
  pthread_cleanup_push(&cleanup1, arg);
  pthread_exit(NULL);
  pthread_cleanup_pop(0);
  return NULL;
}

void *rs_start_accept_thread(void *data) 
{
    int                     err, cli_fd;
    socklen_t               socklen;
    rs_master_info_t        *mi;
    struct                  sockaddr_in cli_addr;
    rs_request_dump_t       *rd;

    mi = (rs_master_info_t *) data;

    pthread_cleanup_push(rs_free_accept_thread, mi);

    if(mi == NULL) {
        rs_log_err(0, "accept thread can not get master info struct");
        goto free;
    }

    rs_memzero(&cli_addr, sizeof(cli_addr));

    for( ;; ) {

        cli_fd = accept(mi->svr_fd, (struct sockaddr *) &cli_addr, &socklen);

        if(cli_fd == -1) {
            if(rs_errno == EINTR) {
                continue;
            }

            rs_log_err(rs_errno, "accept() failed");

            goto free;
        }

        /* register slave */
        rd = rs_get_request_dump(mi->req_dump_info);

        if(rd == NULL) {
            rs_log_err(0, "no more free request_dump struct");
            rs_close(cli_fd);
            cli_fd = -1;
            continue;
        } 

        rd->cli_fd = cli_fd;

        /* init ring buffer */
        if(rs_init_ring_buffer2(&(rd->ring_buf), RS_RING_BUFFER_NUM) != RS_OK) 
        {
            goto free;
        }

        /* init slab */
        if(rs_init_slab(&(rd->slab), NULL, mi->slab_init_size, mi->slab_factor
                    , mi->slab_mem_size, RS_SLAB_PREALLOC) != RS_OK) 
        {
            goto free;
        }

        /* create dump thread */
        if((err = pthread_create(&(rd->dump_thread), 
                        &(mi->req_dump_info->thread_attr), 
                        rs_start_dump_thread, (void *) rd)) != 0) 
        {
            rs_log_err(err, "pthread_create() failed, req_dump thread");
            goto free;
        }
    }

free:

    pthread_cleanup_pop(1);

    return NULL;
}

/*Loops for connection->attempts */
uerr_t http_get_url_info_loop(connection_t * connection)
{

  pthread_mutex_lock(&connection->access_mutex);
  connection->running = TRUE;
  pthread_mutex_unlock(&connection->access_mutex);
  assert(connection->attempts >= 0);

  do
  {
    if (connection->attempts > 0 && connection->err != NEWLOCATION)
    {

      connection_show_message(connection,
			      _("Retrying... Attempt %d in %d seconds"),
			      connection->attempts,
			      connection->retry_delay.tv_sec);
      delay_ms(connection->retry_delay.tv_sec * 1000);
    }

    /*Push the handler which will cleanup any sockets that are left open */
    pthread_cleanup_push(cleanup_socks, (void *) connection);
    connection->err = proz_http_get_url_info(connection);
    /*pop the handler */
    pthread_cleanup_pop(0);

    connection->attempts++;

    switch (connection->err)
    {
    case HOK:
      connection_show_message(connection, _("Successfully got info"));
      pthread_mutex_lock(&connection->access_mutex);
      connection->running = FALSE;
      pthread_mutex_unlock(&connection->access_mutex);
      return connection->err;
      break;

    case NEWLOCATION:
      return connection->err;
      break;

    case HTTPNSFOD:
      connection_show_message(connection, _("File not found!"));
      pthread_mutex_lock(&connection->access_mutex);
      connection->running = FALSE;
      pthread_mutex_unlock(&connection->access_mutex);
      return connection->err;
      break;

    default:
      connection_show_message(connection, proz_strerror(connection->err));
      break;
    }

  }
  while ((connection->attempts < connection->max_attempts)
	 || connection->max_attempts == 0);


  connection_show_message(connection,
			  _
			  ("I have tried %d attempt(s) and have failed, aborting"),
			  connection->attempts);
  pthread_mutex_lock(&connection->access_mutex);
  connection->running = FALSE;
  pthread_mutex_unlock(&connection->access_mutex);
  return connection->err;
}

void vegas_pfb_thread(void *_args) {

    /* Get args */
    struct vegas_thread_args *args = (struct vegas_thread_args *)_args;
    int rv;

    /* Set cpu affinity */
    cpu_set_t cpuset, cpuset_orig;
    sched_getaffinity(0, sizeof(cpu_set_t), &cpuset_orig);
    //CPU_ZERO(&cpuset);
    CPU_CLR(13, &cpuset);
    CPU_SET(11, &cpuset);
    rv = sched_setaffinity(0, sizeof(cpu_set_t), &cpuset);
    if (rv<0) { 
        vegas_error("vegas_pfb_thread", "Error setting cpu affinity.");
        perror("sched_setaffinity");
    }

    /* Set priority */
    rv = setpriority(PRIO_PROCESS, 0, args->priority);
    if (rv<0) {
        vegas_error("vegas_pfb_thread", "Error setting priority level.");
        perror("set_priority");
    }

    /* Attach to status shared mem area */
    struct vegas_status st;
    rv = vegas_status_attach(&st);
    if (rv!=VEGAS_OK) {
        vegas_error("vegas_pfb_thread", 
                "Error attaching to status shared memory.");
        pthread_exit(NULL);
    }
    pthread_cleanup_push((void *)vegas_status_detach, &st);
    pthread_cleanup_push((void *)set_exit_status, &st);
    pthread_cleanup_push((void *)vegas_thread_set_finished, args);

    /* Init status */
    vegas_status_lock_safe(&st);
    hputs(st.buf, STATUS_KEY, "init");
    vegas_status_unlock_safe(&st);

    /* Init structs */
    struct vegas_params gp;
    struct sdfits sf;
    pthread_cleanup_push((void *)vegas_free_sdfits, &sf);

    /* Attach to databuf shared mem */
    struct vegas_databuf *db_in, *db_out;
    db_in = vegas_databuf_attach(args->input_buffer);
    if (db_in==NULL) {
        char msg[256];
        sprintf(msg, "Error attaching to databuf(%d) shared memory.",
                args->input_buffer);
        vegas_error("vegas_pfb_thread", msg);
        pthread_exit(NULL);
    }
    pthread_cleanup_push((void *)vegas_databuf_detach, db_in);
    db_out = vegas_databuf_attach(args->output_buffer);
    if (db_out==NULL) {
        char msg[256];
        sprintf(msg, "Error attaching to databuf(%d) shared memory.",
                args->output_buffer);
        vegas_error("vegas_pfb_thread", msg);
        pthread_exit(NULL);
    }
    pthread_cleanup_push((void *)vegas_databuf_detach, db_out);

    /* Loop */
    char *hdr_in = NULL;
    int curblock_in=0;
    int first=1;
    int acc_len = 0;
    int nchan = 0;
    int nsubband = 0;
    signal(SIGINT,cc);

    vegas_status_lock_safe(&st);
    if (hgeti4(st.buf, "NCHAN", &nchan)==0) {
        fprintf(stderr, "ERROR: %s not in status shm!\n", "NCHAN");
    }
    if (hgeti4(st.buf, "NSUBBAND", &nsubband)==0) {
        fprintf(stderr, "ERROR: %s not in status shm!\n", "NSUBBAND");
    }
    vegas_status_unlock_safe(&st);
    if (EXIT_SUCCESS != init_gpu(db_in->block_size,
                                 db_out->block_size,
                                 nsubband,
                                 nchan))
    {
        (void) fprintf(stderr, "ERROR: GPU initialisation failed!\n");
        run = 0;
    }

    while (run) {

        /* Note waiting status */
        vegas_status_lock_safe(&st);
        hputs(st.buf, STATUS_KEY, "waiting");
        vegas_status_unlock_safe(&st);
//.........这里部分代码省略.........

void *frag_thrd(void *destination_file)
{
	sigset_t sigmask, old_mask;
	char *data_buffer;
	int fd;

	sigemptyset(&sigmask);
	sigaddset(&sigmask, SIGINT);
	sigaddset(&sigmask, SIGTERM);
	sigaddset(&sigmask, SIGUSR1);
	pthread_sigmask(SIG_BLOCK, &sigmask, &old_mask);

	fd = open(destination_file, O_RDONLY);
	if(fd == -1)
		BAD_ERROR("frag_thrd: can't open destination for reading\n");

	data_buffer = malloc(SQUASHFS_FILE_MAX_SIZE);
	if(data_buffer == NULL)
		MEM_ERROR();

	pthread_cleanup_push((void *) pthread_mutex_unlock, &dup_mutex);

	while(1) {
		struct file_buffer *file_buffer = queue_get(to_process_frag);
		struct file_buffer *buffer;
		int sparse = checksum_sparse(file_buffer);
		struct file_info *dupl_ptr;
		long long file_size;
		unsigned short checksum;
		char flag;
		int res;

		if(sparse_files && sparse) {
			file_buffer->c_byte = 0;
			file_buffer->fragment = FALSE;
		} else
			file_buffer->c_byte = file_buffer->size;

		/*
		 * Specutively pull into the fragment cache any fragment blocks
		 * which contain fragments which *this* fragment may be
		 * be a duplicate.
		 *
		 * By ensuring the fragment block is in cache ahead of time
		 * should eliminate the parallelisation stall when the
		 * main thread needs to read the fragment block to do a
		 * duplicate check on it.
		 *
		 * If this is a fragment belonging to a larger file
		 * (with additional blocks) then ignore it.  Here we're
		 * interested in the "low hanging fruit" of files which
		 * consist of only a fragment
		 */
		if(file_buffer->file_size != file_buffer->size) {
			seq_queue_put(to_main, file_buffer);
			continue;
		}

		file_size = file_buffer->file_size;

		pthread_mutex_lock(&dup_mutex);
		dupl_ptr = dupl[DUP_HASH(file_size)];
		pthread_mutex_unlock(&dup_mutex);

		file_buffer->dupl_start = dupl_ptr;
		file_buffer->duplicate = FALSE;

		for(; dupl_ptr; dupl_ptr = dupl_ptr->next) {
			if(file_size != dupl_ptr->file_size ||
					file_size != dupl_ptr->fragment->size)
				continue;

			pthread_mutex_lock(&dup_mutex);
			flag = dupl_ptr->have_frag_checksum;
			checksum = dupl_ptr->fragment_checksum;
			pthread_mutex_unlock(&dup_mutex);

			/*
			 * If we have the checksum and it matches then
			 * read in the fragment block.
			 *
			 * If we *don't* have the checksum, then we are
			 * appending, and the fragment block is on the
			 * "old" filesystem.  Read it in and checksum
			 * the entire fragment buffer
			 */
			if(!flag) {
				buffer = get_fragment_cksum(dupl_ptr,
					data_buffer, fd, &checksum);
				if(checksum != file_buffer->checksum) {
					cache_block_put(buffer);
					continue;
				}
			} else if(checksum == file_buffer->checksum)
				buffer = get_fragment(dupl_ptr->fragment,
					data_buffer, fd);
			else
				continue;

			res = memcmp(file_buffer->data, buffer->data +
//.........这里部分代码省略.........

static struct file_buffer *get_fragment(struct fragment *fragment,
	char *data_buffer, int fd)
{
	struct squashfs_fragment_entry *disk_fragment;
	struct file_buffer *buffer, *compressed_buffer;
	long long start_block;
	int res, size, index = fragment->index;
	char locked;

	/*
	 * Lookup fragment block in cache.
	 * If the fragment block doesn't exist, then get the compressed version
	 * from the writer cache or off disk, and decompress it.
	 *
	 * This routine has two things which complicate the code:
	 *
	 *	1. Multiple threads can simultaneously lookup/create the
	 *	   same buffer.  This means a buffer needs to be "locked"
	 *	   when it is being filled in, to prevent other threads from
	 *	   using it when it is not ready.  This is because we now do
	 *	   fragment duplicate checking in parallel.
	 *	2. We have two caches which need to be checked for the
	 *	   presence of fragment blocks: the normal fragment cache
	 *	   and a "reserve" cache.  The reserve cache is used to
	 *	   prevent an unnecessary pipeline stall when the fragment cache
	 *	   is full of fragments waiting to be compressed.
	 */
	pthread_cleanup_push((void *) pthread_mutex_unlock, &dup_mutex);
	pthread_mutex_lock(&dup_mutex);

again:
	buffer = cache_lookup_nowait(fragment_buffer, index, &locked);
	if(buffer) {
		pthread_mutex_unlock(&dup_mutex);
		if(locked)
			/* got a buffer being filled in.  Wait for it */
			cache_wait_unlock(buffer);
		goto finished;
	}

	/* not in fragment cache, is it in the reserve cache? */
	buffer = cache_lookup_nowait(reserve_cache, index, &locked);
	if(buffer) {
		pthread_mutex_unlock(&dup_mutex);
		if(locked)
			/* got a buffer being filled in.  Wait for it */
			cache_wait_unlock(buffer);
		goto finished;
	}

	/* in neither cache, try to get it from the fragment cache */
	buffer = cache_get_nowait(fragment_buffer, index);
	if(!buffer) {
		/*
		 * no room, get it from the reserve cache, this is
		 * dimensioned so it will always have space (no more than
		 * processors + 1 can have an outstanding reserve buffer)
		 */
		buffer = cache_get_nowait(reserve_cache, index);
		if(!buffer) {
			/* failsafe */
			ERROR("no space in reserve cache\n");
			goto again;
		}
	}

	pthread_mutex_unlock(&dup_mutex);

	compressed_buffer = cache_lookup(fwriter_buffer, index);

	pthread_cleanup_push((void *) pthread_mutex_unlock, &fragment_mutex);
	pthread_mutex_lock(&fragment_mutex);
	disk_fragment = &fragment_table[index];
	size = SQUASHFS_COMPRESSED_SIZE_BLOCK(disk_fragment->size);
	start_block = disk_fragment->start_block;
	pthread_cleanup_pop(1);

	if(SQUASHFS_COMPRESSED_BLOCK(disk_fragment->size)) {
		int error;
		char *data;

		if(compressed_buffer)
			data = compressed_buffer->data;
		else {
			res = read_filesystem(fd, start_block, size, data_buffer);
			if(res == 0) {
				ERROR("Failed to read fragment from output"
					" filesystem\n");
				BAD_ERROR("Output filesystem corrupted?\n");
			}
			data = data_buffer;
		}

		res = compressor_uncompress(comp, buffer->data, data, size,
			block_size, &error);
        // CJH: Decompression errors are displayed elsewhere
		//if(res == -1)
		//	BAD_ERROR("%s uncompress failed with error code %d\n",
		//		comp->name, error);
	} else if(compressed_buffer)
//.........这里部分代码省略.........

/*
 * Same as http_xml_send_xmlrpc() but with user-defined URL path.
 */
int
http_xml_send_xmlrpc2(struct http_client *client,
	struct in_addr ip, u_int16_t port, int https,
	const char *username, const char *password, const char *urlpath,
	const char *methodName, u_int nparams,
	const struct structs_type **ptypes, const void **pdatas,
	const struct structs_type *rep_type, void *rep,
	struct xmlrpc_compact_fault *faultp, structs_xmllog_t *rlogger)
{
	const struct structs_type *const xreq_type
	    = &structs_type_xmlrpc_request;
	const struct structs_type *const xrep_type
	    = &structs_type_xmlrpc_response;
	const struct structs_type *const ftype
	    = &structs_type_xmlrpc_compact_fault;
	struct http_xml_send_xmlrpc_ctx *ctx;
	struct xmlrpc_compact_fault fault;
	char ebuf[128];
	int ret = -1;
	int r;

	/* Get context */
	if ((ctx = MALLOC(MEM_TYPE, sizeof(*ctx))) == NULL) {
		alogf(LOG_ERR, "%s: %m", "malloc");
		return (-1);
	}
	memset(ctx, 0, sizeof(*ctx));
	pthread_cleanup_push(http_xml_send_xmlrpc_cleanup, ctx);

	/* Build XML-RPC request and reply */
	if ((ctx->xreq = structs_xmlrpc_build_request(MEM_TYPE,
	    methodName, nparams, ptypes, pdatas)) == NULL) {
		alogf(LOG_ERR, "%s: %m", "structs_xmlrpc_build_request");
		goto fail;
	}
	if ((ctx->xrep = MALLOC(MEM_TYPE, xrep_type->size)) == NULL) {
		alogf(LOG_ERR, "%s: %m", "malloc");
		goto fail;
	}
	if (structs_init(xrep_type, NULL, ctx->xrep) == -1) {
		alogf(LOG_ERR, "%s: %m", "structs_init");
		FREE(MEM_TYPE, ctx->xrep);
		ctx->xrep = NULL;
		goto fail;
	}

#ifdef XML_RPC_DEBUG
	printf("%s: sending this XML-RPC request:\n", __FUNCTION__);
	(void)structs_xml_output(xreq_type, XML_RPC_REQUEST_TAG,
	    NULL, ctx->xreq, stdout, NULL, 0);
#endif

	/* Send request and get reply; note: we could get canceled here. */
	r = http_xml_send(client, ip, port, https, urlpath,
	    username, password, XML_RPC_REQUEST_TAG, NULL, xreq_type,
	    ctx->xreq, STRUCTS_XML_FULL, XML_RPC_REPLY_TAG, NULL, NULL,
	    xrep_type, ctx->xrep, 0, rlogger);

#ifdef XML_RPC_DEBUG
	printf("%s: got this XML-RPC reply (error=%s):\n",
	    __FUNCTION__, r == -1 ? strerror(errno) : "none");
	(void)structs_xml_output(xrep_type, XML_RPC_REPLY_TAG,
	    NULL, ctx->xrep, stdout, NULL, 0);
#endif

	/* Check error */
	if (r == -1)
		goto fail;

	/* Check for fault */
	if (structs_init(ftype, NULL, &fault) == -1) {
		alogf(LOG_ERR, "%s: %m", "structs_init");
		goto fail;
	}
	if (structs_xmlrpc2struct(xrep_type, ctx->xrep,
	    "fault.value", ftype, &fault, NULL, NULL, 0) == 0) {
		if (faultp != NULL)
			*faultp = fault;
		else
			structs_free(ftype, NULL, &fault);
		ret = -2;			/* -2 indicates fault */
		goto fail;
	}
	structs_free(ftype, NULL, &fault);

	/* Extract response (if desired) */
	if (rep != NULL) {
		if (rep_type != NULL) {		/* return compact type */
			if (structs_xmlrpc2struct(xrep_type, ctx->xrep,
			    "params.0.value", rep_type, rep, NULL, ebuf,
			    sizeof(ebuf)) == -1) {
				(*rlogger)(LOG_ERR, "error decoding XML-RPC"
				    " response: %s", ebuf);
				goto fail;
			}
		} else {			/* return exploded type */
			if (structs_get(&structs_type_xmlrpc_value,
//.........这里部分代码省略.........

/*
 * function handle_requests_loop(): infinite loop of requests handling
 * algorithm: forever, if there are requests to handle, take the first
 *            and handle it. Then wait on the given condition variable,
 *            and when it is signaled, re-do the loop.
 *            increases number of pending requests by one.
 * input:     id of thread, for printing purposes.
 * output:    none.
 */
void*
handle_requests_loop(void* thread_params)
{
    int rc;	                    /* return code of pthreads functions.  */
    struct request* a_request;      /* pointer to a request.               */
    struct handler_thread_params *data;
 				    /* hadler thread's parameters */

    /* sanity check -make sure data isn't NULL */
    data = (struct handler_thread_params*)thread_params;
    assert(data);

    printf("Starting thread '%d'\n", data->thread_id);
    fflush(stdout);

    /* set my cancel state to 'enabled', and cancel type to 'defered'. */
    pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
    pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);

    /* set thread cleanup handler */
    pthread_cleanup_push(cleanup_free_mutex, (void*)data->request_mutex);

    /* lock the mutex, to access the requests list exclusively. */
    rc = pthread_mutex_lock(data->request_mutex);

#ifdef DEBUG
    printf("thread '%d' after pthread_mutex_lock\n", data->thread_id);
    fflush(stdout);
#endif /* DEBUG */

    /* do forever.... */
    while (1) {
        int num_requests = get_requests_number(data->requests);

#ifdef DEBUG
    	printf("thread '%d', num_requests =  %d\n",
	       data->thread_id, num_requests);
    	fflush(stdout);
#endif /* DEBUG */
	if (num_requests > 0) { /* a request is pending */
	    a_request = get_request(data->requests);
	    if (a_request) { /* got a request - handle it and free it */
		handle_request(a_request, data->thread_id);
		free(a_request);
	    }
	}
	else {
	    /* the thread checks the flag before waiting            */
	    /* on the condition variable.                           */
	    /* if no new requests are going to be generated, exit.  */
	    if (done_creating_requests) {
                pthread_mutex_unlock(data->request_mutex);
		printf("thread '%d' exiting\n", data->thread_id);
		fflush(stdout);
		pthread_exit(NULL);
	    }
	    /* wait for a request to arrive. note the mutex will be */
	    /* unlocked here, thus allowing other threads access to */
	    /* requests list.                                       */
#ifdef DEBUG
    	    printf("thread '%d' before pthread_cond_wait\n", data->thread_id);
    	    fflush(stdout);
#endif /* DEBUG */
	    rc = pthread_cond_wait(data->got_request, data->request_mutex);
	    /* and after we return from pthread_cond_wait, the mutex  */
	    /* is locked again, so we don't need to lock it ourselves */
#ifdef DEBUG
    	    printf("thread '%d' after pthread_cond_wait\n", data->thread_id);
    	    fflush(stdout);
#endif /* DEBUG */
	}
    }

    /* remove thread cleanup handler. never reached, but we must use */
    /* it here, according to pthread_cleanup_push's manual page.     */
    pthread_cleanup_pop(0);
}

/*
 * Send a copy of the message, wait for a reply, and return the reply.
 *
 * The "reply" should already be initialized.
 *
 * This properly handles the calling thread's being canceled.
 */
int
http_xml_send(struct http_client *client, struct in_addr ip,
	u_int16_t port, int https, const char *urlpath, const char *username,
	const char *password, const char *ptag, const char *pattrs,
	const struct structs_type *ptype, const void *payload, int pflags,
	const char *rtag, char **rattrsp, const char *rattrs_mtype,
	const struct structs_type *rtype, void *reply, int rflags,
	structs_xmllog_t *rlogger)
{
	struct http_client_connection *cc;
	struct http_request *req;
	struct http_response *resp;
	int ret = -1;
	u_int code;
	FILE *fp;

	/* Get HTTP connection */
	if ((cc = http_client_connect(client, ip, port, https)) == NULL) {
		alogf(LOG_ERR, "can't %s for %s:%u: %m",
		    "get HTTP client" _ inet_ntoa(ip) _ port);
		return -1;
	}

	/* Push cleanup hook */
	pthread_cleanup_push(http_xml_send_cleanup, cc);

	/* Set up request */
	req = http_client_get_request(cc);
	if (http_request_set_method(req,
	    payload != NULL ? HTTP_METHOD_POST : HTTP_METHOD_GET) == -1) {
		alogf(LOG_ERR, "can't %s for %s:%u: %m",
		    "set method" _ inet_ntoa(ip) _ port);
		goto fail;
	}
	if (http_request_set_path(req, urlpath) == -1) {
		alogf(LOG_ERR, "can't %s for %s:%u: %m",
		    "set path" _ inet_ntoa(ip) _ port);
		goto fail;
	}
	if (http_request_set_header(req, 0, "Content-Type", "text/xml") == -1) {
		alogf(LOG_ERR, "can't %s for %s:%u: %m",
		    "set content-type" _ inet_ntoa(ip) _ port);
		goto fail;
	}
	if (username != NULL && password != NULL) {
		char *auth;

		if ((auth = http_request_encode_basic_auth(TYPED_MEM_TEMP,
		    username, password)) == NULL) {
			alogf(LOG_ERR, "can't %s for %s:%u: %m",
			    "encode authorization" _ inet_ntoa(ip) _ port);
			goto fail;
		}
		if (http_request_set_header(req, 0, "Authorization",
		    "Basic %s", auth) == -1) {
			alogf(LOG_ERR, "can't %s for %s:%u: %m",
			    "set authorization header" _ inet_ntoa(ip) _ port);
			FREE(TYPED_MEM_TEMP, auth);
			goto fail;
		}
		FREE(TYPED_MEM_TEMP, auth);
	}

	/* Write XML data to HTTP client output stream */
	if (payload != NULL) {
		if ((fp = http_request_get_output(req, 1)) == NULL) {
			alogf(LOG_ERR, "can't %s for %s:%u: %m",
			    "get output" _ inet_ntoa(ip) _ port);
			goto fail;
		}
		if (structs_xml_output(ptype,
		    ptag, pattrs, payload, fp, NULL, pflags) == -1) {
			alogf(LOG_ERR, "can't %s for %s:%u: %m",
			    "write XML" _ inet_ntoa(ip) _ port);
			goto fail;
		}
	}

	/* Get response */
	if ((resp = http_client_get_response(cc)) == NULL) {
		alogf(LOG_ERR, "can't %s for %s:%u: %m",
		    "get response" _ inet_ntoa(ip) _ port);
		goto fail;
	}
	if ((code = http_response_get_code(resp)) != HTTP_STATUS_OK) {
		alogf(LOG_ERR, "rec'd HTTP error code %d from"
		      "http%s://%s:%u%s: %s", code _ https ? "s" : "" _
		      inet_ntoa(ip) _ port _ urlpath _
		      http_response_status_msg(code));
		goto fail;
	}

	/* Read XML reply from client input stream */
//.........这里部分代码省略.........

/******************************************************************************
Description.: this is the main worker thread
              it loops forever, grabs a fresh frame, decompressed the JPEG
              and displays the decoded data using SDL
Input Value.:
Return Value:
******************************************************************************/
void *worker_thread(void *arg)
{
    int frame_size = 0, firstrun = 1;

    SDL_Surface *screen = NULL, *image = NULL;
    decompressed_image rgbimage;

    /* initialze the buffer for the decompressed image */
    rgbimage.buffersize = 0;
    rgbimage.buffer = NULL;

    /* initialze the SDL video subsystem */
    if(SDL_Init(SDL_INIT_VIDEO) < 0) {
        fprintf(stderr, "Couldn't initialize SDL: %s\n", SDL_GetError());
        exit(EXIT_FAILURE);
    }

    /* just allocate a large buffer for the JPEGs */
    if((frame = malloc(4096 * 1024)) == NULL) {
        OPRINT("not enough memory for worker thread\n");
        exit(EXIT_FAILURE);
    }

    /* set cleanup handler to cleanup allocated ressources */
    pthread_cleanup_push(worker_cleanup, NULL);

    while(!pglobal->stop) {
        DBG("waiting for fresh frame\n");
        pthread_cond_wait(&pglobal->in[plugin_number].db_update, &pglobal->in[plugin_number].db);

        /* read buffer */
        frame_size = pglobal->in[plugin_number].size;
        memcpy(frame, pglobal->in[plugin_number].buf, frame_size);

        pthread_mutex_unlock(&pglobal->in[plugin_number].db);

        /* decompress the JPEG and store results in memory */
        if(decompress_jpeg(frame, frame_size, &rgbimage)) {
            DBG("could not properly decompress JPEG data\n");
            continue;
        }

        if(firstrun) {
            /* create the primary surface (the visible window) */
            screen = SDL_SetVideoMode(rgbimage.width, rgbimage.height, 0, SDL_ANYFORMAT | SDL_HWSURFACE);
            SDL_WM_SetCaption("MJPG-Streamer Viewer", NULL);

            /* create a SDL surface to display the data */
            image = SDL_AllocSurface(SDL_SWSURFACE, rgbimage.width, rgbimage.height, 24,
#if SDL_BYTEORDER == SDL_LIL_ENDIAN
                                     0x0000FF, 0x00FF00, 0xFF0000,
#else
                                     0xFF0000, 0x00FF00, 0x0000FF,
#endif
                                     0);

            /* copy the decoded data across */
            memcpy(image->pixels, rgbimage.buffer, rgbimage.width * rgbimage.height * 3);
            free(rgbimage.buffer);

            /* now, that we know the dimensions, we can directly copy to the right surface */
            rgbimage.buffer = image->pixels;
            rgbimage.buffersize = rgbimage.width * rgbimage.height * 3;

            firstrun = 0;
        }

        /* copy the image to the primary surface */
        SDL_BlitSurface(image, NULL, screen, NULL);

        /* redraw the whole surface */
        SDL_Flip(screen);
    }

    pthread_cleanup_pop(1);

    /* get rid of the image */
    SDL_FreeSurface(image);

    return NULL;
}

void* __ctInitThread(void* v)//pcontech_thread_create ptc
{
    void* (*f)(void*);
    void (*g)(void*);
    void* a;
    unsigned int p;
    long long skew;
    ct_tsc_t start;
    pcontech_thread_create ptc = (pcontech_thread_create) v;
    f = ptc->func;
    a = ptc->arg;
    p = ptc->parent_ctid;
    start = rdtsc();
    
    //printf("Thread %u alive\n", ptc->child_ctid);
    
    // HACK -- REMOVE!!!
    #if 0
    printf("Hello!\n"); fflush(stdout);
    cpu_set_t cpuset;
    CPU_ZERO(&cpuset);
    CPU_SET(24 - (ptc->child_ctid % 16), &cpuset);
    int r = pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
    printf("%d - %d - %llx\n", ptc->child_ctid, r, cpuset);
    #endif
    
    //
    // Now compute the skew
    //
#ifdef CT_CLOCK_SKEW
    skew = 0;
    while (ptc->parent_skew == 0)
    {
        ptc->child_skew = rdtsc();
    }
    skew = ptc->parent_skew;
#else
    skew = 1;
#endif
    
    __ctThreadLocalNumber = ptc->child_ctid;//__sync_fetch_and_add(&__ctThreadGlobalNumber, 1);
    __ctAllocateLocalBuffer();
    
    __ctThreadInfoList = NULL;

    free(ptc);
    
    __ctStoreThreadCreate(p, skew, start);
    if (__ctIsROIEnabled == true && __ctIsROIActive == false)
    {
        __ctQueueBuffer(false);
    }
 
    g = __ctCleanupThread;
    pthread_cleanup_push(g, (void*)p);
    
    #ifdef CT_OVERHEAD_TRACK
    {
        ct_tsc_t end = rdtsc();
        //__ctTotalThreadOverhead += (end - start);
    }
    #endif
    
    a = f(a);
    pthread_cleanup_pop(1);
    return a;
}

static void *worker(void UNUSED(*arg))
{
  MYLDAP_SESSION *session;
  int csock;
  int j;
  struct sockaddr_storage addr;
  socklen_t alen;
  fd_set fds;
  struct timeval tv;
  /* create a new LDAP session */
  session = myldap_create_session();
  /* clean up the session if we're done */
  pthread_cleanup_push(worker_cleanup, session);
  /* start waiting for incoming connections */
  while (1)
  {
    /* time out connection to LDAP server if needed */
    myldap_session_check(session);
    /* set up the set of fds to wait on */
    FD_ZERO(&fds);
    FD_SET(nslcd_serversocket, &fds);
    /* set up our timeout value */
    tv.tv_sec = nslcd_cfg->idle_timelimit;
    tv.tv_usec = 0;
    /* wait for a new connection */
    j = select(nslcd_serversocket + 1, &fds, NULL, NULL,
               nslcd_cfg->idle_timelimit > 0 ? &tv : NULL);
    /* check result of select() */
    if (j < 0)
    {
      if (errno == EINTR)
        log_log(LOG_DEBUG, "select() failed (ignored): %s", strerror(errno));
      else
        log_log(LOG_ERR, "select() failed: %s", strerror(errno));
      continue;
    }
    /* see if our file descriptor is actually ready */
    if (!FD_ISSET(nslcd_serversocket, &fds))
      continue;
    /* wait for a new connection */
    alen = (socklen_t)sizeof(struct sockaddr_storage);
    csock = accept(nslcd_serversocket, (struct sockaddr *)&addr, &alen);
    if (csock < 0)
    {
      if ((errno == EINTR) || (errno == EAGAIN) || (errno == EWOULDBLOCK))
        log_log(LOG_DEBUG, "accept() failed (ignored): %s", strerror(errno));
      else
        log_log(LOG_ERR, "accept() failed: %s", strerror(errno));
      continue;
    }
    /* make sure O_NONBLOCK is not inherited */
    if ((j = fcntl(csock, F_GETFL, 0)) < 0)
    {
      log_log(LOG_ERR, "fctnl(F_GETFL) failed: %s", strerror(errno));
      if (close(csock))
        log_log(LOG_WARNING, "problem closing socket: %s", strerror(errno));
      continue;
    }
    if (fcntl(csock, F_SETFL, j & ~O_NONBLOCK) < 0)
    {
      log_log(LOG_ERR, "fctnl(F_SETFL,~O_NONBLOCK) failed: %s", strerror(errno));
      if (close(csock))
        log_log(LOG_WARNING, "problem closing socket: %s", strerror(errno));
      continue;
    }
    /* indicate new connection to logging module (generates unique id) */
    log_newsession();
    /* handle the connection */
    handleconnection(csock, session);
    /* indicate end of session in log messages */
    log_clearsession();
  }
  pthread_cleanup_pop(1);
  return NULL;
}

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

		return(WAIT_FAILED);
	}

	if (bogustype == TRUE) {
#ifdef DEBUG
		g_message ("%s: Returning due to bogus type", __func__);
#endif

		return(WAIT_FAILED);
	}

	done = test_and_own (numobjects, handles, waitall, &count, &lowest);
	if (done == TRUE) {
		return(WAIT_OBJECT_0+lowest);
	}
	
	/* Have to wait for some or all handles to become signalled
	 */

	if(timeout!=INFINITE) {
		_wapi_calc_timeout (&abstime, timeout);
	}

	if (alertable && _wapi_thread_apc_pending (current_thread)) {
		_wapi_thread_dispatch_apc_queue (current_thread);
		return WAIT_IO_COMPLETION;
	}
	
	while(1) {
		/* Prod all special-wait handles that aren't already
		 * signalled
		 */
		for (i = 0; i < numobjects; i++) {
			if (_wapi_handle_test_capabilities (handles[i], WAPI_HANDLE_CAP_SPECIAL_WAIT) == TRUE && _wapi_handle_issignalled (handles[i]) == FALSE) {
				_wapi_handle_ops_special_wait (handles[i], 0);
			}
		}

		/* Check before waiting on the condition, just in case
		 */
		done = test_and_own (numobjects, handles, waitall,
				     &count, &lowest);
		if (done == TRUE) {
			return(WAIT_OBJECT_0 + lowest);
		}
		
#ifdef DEBUG
		g_message ("%s: locking signal mutex", __func__);
#endif

		pthread_cleanup_push ((void(*)(void *))_wapi_handle_unlock_signal_mutex, NULL);
		thr_ret = _wapi_handle_lock_signal_mutex ();
		g_assert (thr_ret == 0);
		
		if (timeout == INFINITE) {
			ret = _wapi_handle_wait_signal ();
		} else {
			ret = _wapi_handle_timedwait_signal (&abstime);
		}

#ifdef DEBUG
		g_message ("%s: unlocking signal mutex", __func__);
#endif

		thr_ret = _wapi_handle_unlock_signal_mutex (NULL);
		g_assert (thr_ret == 0);
		pthread_cleanup_pop (0);
		
		if (alertable && _wapi_thread_apc_pending (current_thread)) {
			_wapi_thread_dispatch_apc_queue (current_thread);
			return WAIT_IO_COMPLETION;
		}
	
		/* Check if everything is signalled, as we can't
		 * guarantee to notice a shared signal even if the
		 * wait timed out
		 */
		done = test_and_own (numobjects, handles, waitall,
				     &count, &lowest);
		if (done == TRUE) {
			return(WAIT_OBJEC