/*
 * b_query_propagate :
 * This routine is used by both search_svc_1() and b_query_svc_1() to propagate
 * the query to the peers.
 * 
 * RETURN VALUE :
 * If it relays the requests to its peers then it adds this request to a
 * local pending queue and returns a pointer to that request.
 *
 * Outline of the routine: 
 * - Checks if we already have received this query (check on the pending list
 *   for the msg_id of the received message.
 *      - If we already have processed this don't do anything.
 *      - Decrement the TTL. If TTL is zero after that don't relay it anymore but
 *        do the following :  
 *        - Look at the local cache and call b_hitquery() with the results to
 *          the uphost.
 *      - If we have not processed it we do the following :
 *          - Create a new query_req. Add it to the pending list.
 *          - Walk the list of peers and do the following :
 *              - Check if the cached index file already has the name of the
 *                peer.
 *              - If the peer name is not in the cache send it a b_query()
 *                request.
 *          - Look at the local cache and call b_hitquery() with the results to
 *            the uphost.
 */
query_node_t *
b_query_propagate(b_query_req *argp, int flag)
{
	bool_t retval = TRUE;
    query_node_t *node;
    b_query_req  req;
    peers_t my_cache;
    CLIENT *clnt;
    int i, cnt;
    enum clnt_stat stat;
    int ret;

    node = find_node(&argp->id);

    /*
     * We already have processed msg_id. Hence, we have nothing to do now.
     */
    if (node) {
        pthread_mutex_unlock(&node->node_lock);
#ifdef DEBUG
        printf("b_query_propagate: This request for file %s from %s is already processed\n", argp->fname, argp->uphost);
#endif
        return (NULL);
    }

    /*
     * We did not find a matching node. So, this is a new request and we need to
     * process it.
     */

    /*
     * Check if this request should be relayed to the peers.
     */
    if (argp->ttl <= 1) {
        /*
         * We don't need to relay this to the peers. 
         */
        return (NULL);
    }

    node = (query_node_t *)malloc(sizeof(query_node_t));
    if (node == NULL) {
        printf("Failed to allocate query_node_t. Hence not processing this request.\n");
        printf("Hoping the reaper thread will reap some memory\n");
        return (NULL);
    }

    node->req.id = argp->id;
    strcpy(node->req.uphost, argp->uphost);
    strcpy(node->req.fname, argp->fname);
    node->req.ttl = argp->ttl - 1;  /* Decrement the TTL by one */
    node->sent = 0; node->recv = 0;
    node->next = NULL;
    pthread_mutex_init(&node->node_lock, NULL);
    pthread_cond_init(&node->allhome_cv, NULL);

    /*
     * Build the request that we need to send.
     */
    req.id = argp->id;
    strcpy(req.uphost, localhostname);
    strcpy(req.fname, argp->fname);
    req.ttl = argp->ttl - 1;

    (void) insert_node(node);

    /*
     * Lock the node. This is to avoid processing of responses
     * even before we are done with the propagation of requests to all the peers.
     */
    pthread_mutex_lock(&node->node_lock);

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

/** module_cac_init:
 *
 *  Arguments:
 *  @name - name of the module
 *
 * Description: This function is used to initialize the cac
 * module
 *
 * Return values:
 *     MCTL module instance pointer
 *
 * Notes: none
 **/
mct_module_t *module_cac_init(const char *name)
{
  mct_module_t *p_mct_mod = NULL;
  module_cac_t *p_mod = NULL;
  img_core_ops_t *p_core_ops = NULL;
  mct_port_t *p_sinkport = NULL, *p_sourceport = NULL;
  int rc = 0;
  int i = 0;

  IDBG_MED("%s:%d] ", __func__, __LINE__);
  p_mct_mod = mct_module_create(name);
  if (NULL == p_mct_mod) {
    IDBG_ERROR("%s:%d cannot allocate mct module", __func__, __LINE__);
    return NULL;
  }

  p_mod = malloc(sizeof(module_cac_t));
  if (NULL == p_mod) {
    IDBG_ERROR("%s:%d failed", __func__, __LINE__);
    goto error;
  }

  p_mct_mod->module_private = (void *)p_mod;
  memset(p_mod, 0, sizeof(module_cac_t));

  pthread_mutex_init(&p_mod->mutex, NULL);
  pthread_cond_init(&p_mod->cond, NULL);
  p_core_ops = &p_mod->core_ops;

  IDBG_MED("%s:%d] ", __func__, __LINE__);
  /* check if the cac module is present */
  rc = img_core_get_comp(IMG_COMP_CAC, "qcom.cac", p_core_ops);
  if (IMG_ERROR(rc)) {
    IDBG_ERROR("%s:%d] Error rc %d", __func__, __LINE__, rc);
    goto error;
  }
 /* try to load the component */
  rc = IMG_COMP_LOAD(p_core_ops, NULL);
  if (IMG_ERROR(rc)) {
    IDBG_ERROR("%s:%d] Error rc %d", __func__, __LINE__, rc);
    goto error;
  }
  p_mod->lib_ref_count++;
  p_mod->cac_client = NULL;

  IDBG_MED("%s:%d] ", __func__, __LINE__);
  /* create static ports */
  for (i = 0; i < MAX_CAC_STATIC_PORTS; i++) {
    p_sinkport = module_cac_create_port(p_mct_mod, MCT_PORT_SINK);
    if (NULL == p_sinkport) {
      IDBG_ERROR("%s:%d] create SINK port failed", __func__, __LINE__);
      goto error;
    }
    p_sourceport = module_cac_create_port(p_mct_mod, MCT_PORT_SRC);
    if (NULL == p_sourceport) {
      IDBG_ERROR("%s:%d] create SINK port failed", __func__, __LINE__);
      goto error;
    }
  }

  p_mct_mod->start_session    = module_cac_start_session;
  p_mct_mod->stop_session     = module_cac_stop_session;
  IDBG_MED("%s:%d] %p", __func__, __LINE__, p_mct_mod);
  return p_mct_mod;

error:
  if (p_mod) {
    module_cac_deinit(p_mct_mod);
  } else if (p_mct_mod) {
    mct_module_destroy(p_mct_mod);
    p_mct_mod = NULL;
  }
  return NULL;

}

static void chpl_thread_condvar_init(chpl_thread_condvar_t* cv) {
  if (pthread_cond_init((pthread_cond_t*) cv, NULL))
    chpl_internal_error("pthread_cond_init() failed");
}

//.........这里部分代码省略.........
	struct fb_var_screeninfo fb_var;

	disp = (struct vpu_display *)calloc(1, sizeof(struct vpu_display));
	if (disp == NULL) {
		fputs("falied to allocate vpu_display\n", stderr);
		return NULL;
	}

	/* set alpha */
#ifdef BUILD_FOR_ANDROID
	disp->fd = open("/dev/graphics/fb0", O_RDWR, 0);
#else
	disp->fd = open("/dev/fb0", O_RDWR, 0);
#endif
	if (disp->fd < 0) {
		fputs("unable to open fb0\n", stderr);
		free(disp);
		return NULL;
	}
	alpha.alpha = 0;
	alpha.enable = 1;
	if (ioctl(disp->fd, MXCFB_SET_GBL_ALPHA, &alpha) < 0) {
		fputs("set alpha blending failed\n", stderr);
		close(disp->fd);
		free(disp);
		return NULL;
	}
	if ( ioctl(disp->fd, FBIOGET_VSCREENINFO, &fb_var) < 0) {
		fputs("Get FB var info failed!\n", stderr);
		close(disp->fd);
		free(disp);
		return NULL;
	}
	if (!disp_width || !disp_height) {
		disp_width = fb_var.xres;
		disp_height = fb_var.yres;
	}

	/* allocate buffers, use an extra buf for init buf */
	disp->nframes = nframes;
	disp->frame_size = width*height*3/2;
	for (i=0;i<nframes;i++) {
		err = ipu_memory_alloc(disp->frame_size, 1, &(disp->ipu_bufs[i].ipu_paddr),
				&(disp->ipu_bufs[i].ipu_vaddr), disp->fd);
		if ( err < 0) {
			fputs("ipu_memory_alloc failed\n", stderr);
			free(disp);
			return NULL;
		}
	}

	memset(&(disp->ipu_handle), 0, sizeof(ipu_lib_handle_t));
	memset(&(disp->input), 0, sizeof(ipu_lib_input_param_t));
	memset(&(disp->output), 0, sizeof(ipu_lib_output_param_t));

        disp->input.width = width;
        disp->input.height = height;
	disp->input.input_crop_win.pos.x = 0;
	disp->input.input_crop_win.pos.y = 0;
	disp->input.input_crop_win.win_w = 0;
	disp->input.input_crop_win.win_h = 0;

	/* Set VDI motion algorithm. */
	if (motion_mode) {
		if (motion_mode == 'h') {
			disp->input.motion_sel = HIGH_MOTION;
		} else if (motion_mode == 'l') {
			disp->input.motion_sel = LOW_MOTION;
		} else if (motion_mode == 'm') {
			disp->input.motion_sel = MED_MOTION;
		} else {
			disp->input.motion_sel = MED_MOTION;
			fprintf(stderr, "%c unknown motion mode, medium, the default is used\n", motion_mode);
		}
	}

	disp->input.fmt = V4L2_PIX_FMT_YUV420;
	disp->output.width = disp_width;
	disp->output.height = disp_height;
	disp->output.fmt = V4L2_PIX_FMT_UYVY;
	disp->output.fb_disp.pos.x = disp_left;
	disp->output.fb_disp.pos.y = disp_top;
	disp->output.show_to_fb = 1;
	disp->output.fb_disp.fb_num = 2;

	fprintf(stderr, "Display to %d %d, top offset %d, left offset %d\n",
			disp_width, disp_height, disp_top, disp_left);

	disp->ipu_q.tail = disp->ipu_q.head = 0;
	disp->stopping = 0;

	dec->disp = disp;
	pthread_mutex_init(&ipu_mutex, NULL);
	pthread_cond_init(&ipu_cond, NULL);

	/* start disp loop thread */
	pthread_create(&(disp->ipu_disp_loop_thread), NULL, (void*)ipu_disp_loop_thread, (void *)dec);

	return disp;
}

int 
ldap_pvt_thread_cond_init( ldap_pvt_thread_cond_t *cond )
{
	return ERRVAL( pthread_cond_init(
		cond, LDAP_INT_THREAD_CONDATTR_DEFAULT ) );
}

SC_Semaphore::SC_Semaphore(int initialCount)
{
    pthread_mutex_init (&mutex, NULL);
    pthread_cond_init (&available, NULL);
    count = initialCount;
}

int sock_cntr_open(struct fid_domain *domain, struct fi_cntr_attr *attr,
		   struct fid_cntr **cntr, void *context)
{
	int ret;
	struct sock_domain *dom;
	struct sock_cntr *_cntr;
	struct fi_wait_attr wait_attr;
	struct sock_fid_list *list_entry;
	struct sock_wait *wait;
	
	dom = container_of(domain, struct sock_domain, dom_fid);
	if (attr && sock_cntr_verify_attr(attr))
		return -FI_ENOSYS;

	_cntr = calloc(1, sizeof(*_cntr));
	if (!_cntr)
		return -FI_ENOMEM;

	ret = pthread_cond_init(&_cntr->cond, NULL);
	if (ret)
		goto err;

	if(attr == NULL)
		memcpy(&_cntr->attr, &sock_cntr_add, sizeof(sock_cntr_attr));
	else 
		memcpy(&_cntr->attr, attr, sizeof(sock_cntr_attr));
	
	switch (_cntr->attr.wait_obj) {

	case FI_WAIT_NONE:
	case FI_WAIT_UNSPEC:
	case FI_WAIT_MUTEX_COND:
		_cntr->signal = 0;
		break;

	case FI_WAIT_FD:
		wait_attr.flags = 0;
		wait_attr.wait_obj = FI_WAIT_FD;
		ret = sock_wait_open(&dom->fab->fab_fid, &wait_attr,
				     &_cntr->waitset);
		if (ret) {
			ret = FI_EINVAL;
			goto err;
		}
		_cntr->signal = 1;
		break;
		
	case FI_WAIT_SET:
		if (!attr) {
			ret = FI_EINVAL;
			goto err;
		}

		_cntr->waitset = attr->wait_set;
		_cntr->signal = 1;
		wait = container_of(attr->wait_set, struct sock_wait, wait_fid);
		list_entry = calloc(1, sizeof(*list_entry));
		dlist_init(&list_entry->entry);
		list_entry->fid = &_cntr->cntr_fid.fid;
		dlist_insert_after(&list_entry->entry, &wait->fid_list);
		break;
		
	default:
		break;
	}

	pthread_mutex_init(&_cntr->mut, NULL);
	fastlock_init(&_cntr->list_lock);

	atomic_initialize(&_cntr->ref, 0);
	atomic_initialize(&_cntr->err_cnt, 0);

	atomic_initialize(&_cntr->value, 0);
	atomic_initialize(&_cntr->threshold, ~0);

	dlist_init(&_cntr->tx_list);
	dlist_init(&_cntr->rx_list);

	_cntr->cntr_fid.fid.fclass = FI_CLASS_CNTR;
	_cntr->cntr_fid.fid.context = context;
	_cntr->cntr_fid.fid.ops = &sock_cntr_fi_ops;
	_cntr->cntr_fid.ops = &sock_cntr_ops;

	atomic_inc(&dom->ref);
	_cntr->domain = dom;
	*cntr = &_cntr->cntr_fid;
	return 0;

err:
	free(_cntr);
	return -ret;
}

//Implements the master thread
int main(int argc, char *argv[]){
    if(argc < 4){
        printf("Not enough arguments:%d\n",argc);
        exit(EXIT_FAILURE); 
    }
    //Number of disc drives (1 <= D <= 16)
    int const D = atoi(argv[1]);    
    //Number of worker threads (1 <= W <= 16)
    int const W = atoi(argv[2]);
    //Number of iterations each worker does(1 <= L)
    int const L = atoi(argv[3]);

    /* Debugging */
    /*
    printf("D:%d",D);
    printf("\nW:%d",W);
    printf("\nL:%d\n",L);
    */

    info thread_info;

    /*Initialise mutexs */
    thread_info.read_mons = emalloc(sizeof(pthread_mutex_t) * D);
    thread_info.write_mons = emalloc(sizeof(pthread_mutex_t) * D);
    thread_info.read_resp_lock = emalloc(sizeof(pthread_mutex_t) * W);
    thread_info.write_resp_lock = emalloc(sizeof(pthread_mutex_t) * W);
    /* and conditions */
    thread_info.read_resp_fin = emalloc(sizeof(pthread_cond_t) * W);
    thread_info.write_resp_fin = emalloc(sizeof(pthread_cond_t) * W);
    thread_info.read_ready = emalloc(sizeof(pthread_cond_t) * D);
    thread_info.write_ready = emalloc(sizeof(pthread_cond_t) * D);

    /* Initialise array of read/write queues for disc */
    thread_info.read_queues = emalloc(sizeof(circ_buf) * D);
    thread_info.write_queues = emalloc(sizeof(circ_buf) * D);
    /* Initialise array of read/write response monitors for workers */
    mon *r_resp = emalloc(sizeof(mon) * W);
    mon *w_resp = emalloc(sizeof(mon) * W);
    thread_info.read_response = r_resp;
    thread_info.write_response = w_resp;
    //thread_info.read_response = emalloc(sizeof(mon) * W);
    //thread_info.write_response = emalloc(sizeof(mon) * W);

    /* Initialise array of ids */
    thread_info.disc_ids = emalloc(sizeof(int) * D);
    thread_info.work_ids = emalloc(sizeof(int) * W);

    /* Initialise array of disc_kill messages */
    thread_info.disc_kill = emalloc(sizeof(int) * D);
    /* Initialise arrays for clocks */
    thread_info.disc_times = emalloc(sizeof(int) * D);
    thread_info.work_times = emalloc(sizeof(int) * W);

    /* Store number of discs, workers and iterations */
    thread_info.D = D;
    thread_info.W = W;
    thread_info.L = L;

    pthread_t disc_threads[D], worker_threads[W];
    int rc, i;

    //Set PTHREAD_MUTEX_ERRORCHECK
    pthread_mutexattr_t mutex_attr;
    pthread_mutexattr_settype(&mutex_attr, PTHREAD_MUTEX_ERRORCHECK);
    //Set up conditional attr
    pthread_condattr_t cond_attr;
    if (pthread_condattr_init(&cond_attr) != 0){
        printf("\nMaster: Cond_attr init failed\n");
    }

    //starts D disc threads
    for (i=0; i < D; ++i) {
        /* Initialise read-write locks */
        if (pthread_mutex_init(&(thread_info.read_mons[i]), &mutex_attr) != 0){
            printf("\nMaster: %d, disc read mutex init failed\n",i);
        }
        if (pthread_mutex_init(&(thread_info.write_mons[i]), &mutex_attr) != 0){
            printf("\nMaster: %d, disc write mutex init failed\n",i);
        }
        if (pthread_cond_init(&(thread_info.read_ready[i]), &cond_attr) != 0){
            printf("\nMaster: %d, worker read response cond init failed\n",i);
        }
        if (pthread_cond_init(&(thread_info.write_ready[i]), &cond_attr) != 0){
            printf("\nMaster: %d, worker write response cond init failed\n",i);
        }
        thread_info.read_queues[i].head = 0;
        thread_info.read_queues[i].tail = 0;
        thread_info.write_queues[i].head = 0;
        thread_info.write_queues[i].tail = 0;
        thread_info.disc_times[i] = 0;
        printf("Creating disc thread %d\n", i);
        rc = pthread_create(&disc_threads[i], NULL, disc_start, &thread_info);
        if(rc != 0){
            printf("Creation of disc thread:%d failed, code:%d\n",i,rc);
            exit(EXIT_FAILURE);
        }
        //printf("Creation of disc thread:%u successful\n",disc_threads[i]);
        thread_info.disc_ids[i] = disc_threads[i];
    }
//.........这里部分代码省略.........

int ipclite_client_create(ipclite **c, const char *msg, int flags)
{
    ipclite_client *client;
    int pip[2];

    if(! c)
        return IPCLITE_ERR_INV;

    if(pipe2(pip, O_NONBLOCK) == -1)  {
        return IPCLITE_ERR_PIP;
    }

    if(! (client = new_instance(ipclite_client)))  {
        close(pip[0]);
        close(pip[1]);
        return IPCLITE_ERR_OOM;
    }

    client->base.type = IPCLITE_CLIENT;
    client->base.msg = ((msg && *msg) ? strdup(msg) : NULL);

    client->base.handler = NULL;
    client->base.ud = NULL;

    client->base.master = -1;
    client->base.path[0] = '\0';

    ipclite_port_init(&client->port, -1);
    client->port_state = client->port.state;

    client->peer = -1;

    client->notify[0] = pip[0];
    client->notify[1] = pip[1];

    pthread_mutex_init(&client->lock, NULL);
    pthread_cond_init(&client->cond, NULL);

    client->worker_started = 0;
    client->quit = 0;

    FD_ZERO(&client->rd);
    FD_ZERO(&client->wr);

    FD_SET(pip[0], &client->rd);
    client->max_fd = pip[0] + 1;

    pthread_mutex_init(&client->rd_lock, NULL);
    pthread_cond_init(&client->rd_cond, NULL);
    client->rd_quit = 0;
    client->rd_started = 0;
    list_init(&client->rd_queue);
    list_init(&client->req_queue);

    pthread_mutex_init(&client->wait_lock, NULL);
    pthread_cond_init(&client->wait_cond, NULL);
    client->wait_cnt = 0;
    list_init(&client->wait_queue);

    *c = (ipclite *)client;
    return IPCLITE_ERR_OK;
}

	int main(int argc, char **argv)
	{
		//extern const char g_appkey[];
		extern const uint8_t g_appkey[];
		extern const size_t g_appkey_size;
		sp_session_config spconfig;

		memset(&spconfig, 0, sizeof(spconfig));
		spconfig.api_version = SPOTIFY_API_VERSION;

		spconfig.cache_location = "tmp";
		spconfig.settings_location = "tmp";

		spconfig.application_key = g_appkey;
		spconfig.application_key_size = g_appkey_size;

		spconfig.user_agent = "jukebewkx";

		//spconfig.callbacks = 0;
		spconfig.callbacks = &session_callbacks;

		spconfig.compress_playlists = 0;

		sp_session *sp;
		sp_error err;
		int next_timeout = 0;
		const char *username = NULL;
		const char *password = NULL;
		int opt;

		while ((opt = getopt(argc, argv, "u:p:l:d")) != EOF) {
			switch (opt) {
			case 'u':
				username = optarg;
				break;

			case 'p':
				password = optarg;
				break;

			case 'l':
				g_listname = optarg;
				break;

			case 'd':
				g_remove_tracks = 1;
				break;

			default:
				exit(1);
			}
		}

		if (!username || !password || !g_listname) {
			usage(argv[0]);
			exit(1);
		}

		//audio_init(&g_audiofifo);

		/* Create session */
		spconfig.application_key_size = g_appkey_size;

		pthread_mutex_init(&g_notify_mutex, NULL);
		pthread_cond_init(&g_notify_cond, NULL);

		for (int i = 0; i < 322; i++)
		{
			char penis = g_appkey[i];
		}

		err = sp_session_create(&spconfig, &sp);

		if (SP_ERROR_OK != err) {
			fprintf(stderr, "Unable to create session: %s\n",
				sp_error_message(err));
			exit(1);
		}

		g_sess = sp;


		sp_session_login(sp, username, password, 0, NULL);
		pthread_mutex_lock(&g_notify_mutex);

		for (;;) {
			if (next_timeout == 0) {
				while(!g_notify_do)
					pthread_cond_wait(&g_notify_cond, &g_notify_mutex);
			} else {
				struct timespec ts;

	#if _POSIX_TIMERS > 0
				clock_gettime(CLOCK_REALTIME, &ts);
	#else
				struct timeval tv;
				gettimeofday(&tv, NULL);
				TIMEVAL_TO_TIMESPEC(&tv, &ts);
	#endif
				ts.tv_sec += next_timeout / 1000;
//.........这里部分代码省略.........

DirectResult
voodoo_manager_create( int             fd,
                       VoodooClient   *client,
                       VoodooServer   *server,
                       VoodooManager **ret_manager )
{
     DirectResult     ret;
     VoodooManager   *manager;
     int              val;
     unsigned int     len;
     static const int tos = IPTOS_LOWDELAY;

     D_ASSERT( fd >= 0 );
     D_ASSERT( (client != NULL) ^ (server != NULL) );
     D_ASSERT( ret_manager != NULL );

     /* Allocate manager structure. */
     manager = D_CALLOC( 1, sizeof(VoodooManager) );
     if (!manager) {
          D_WARN( "out of memory" );
          return DR_NOLOCALMEMORY;
     }

     D_DEBUG( "Voodoo/Manager: Creating manager at %p.\n", manager );

     if (setsockopt( fd, SOL_IP, IP_TOS, &tos, sizeof(tos) ) < 0)
          D_PERROR( "Voodoo/Manager: Could not set IP_TOS!\n" );

     if (setsockopt( fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one) ) < 0)
          D_PERROR( "Voodoo/Manager: Could not set TCP_NODELAY!\n" );

     DUMP_SOCKET_OPTION( SO_SNDLOWAT );
     DUMP_SOCKET_OPTION( SO_RCVLOWAT );
     DUMP_SOCKET_OPTION( SO_SNDBUF );
     DUMP_SOCKET_OPTION( SO_RCVBUF );

     /* Create the hash table for dispatcher instances. */
     ret = direct_hash_create( 251, &manager->instances.local );
     if (ret) {
          D_FREE( manager );
          return ret;
     }

     /* Create the hash table for requestor instances. */
     ret = direct_hash_create( 251, &manager->instances.remote );
     if (ret) {
          direct_hash_destroy( manager->instances.local );
          D_FREE( manager );
          return ret;
     }

     /* Store file descriptor. */
     manager->fd = fd;

     /* Store client or server. */
     manager->client = client;
     manager->server = server;

     /* Initialize all locks. */
     direct_util_recursive_pthread_mutex_init( &manager->instances.lock );
     direct_util_recursive_pthread_mutex_init( &manager->response.lock );
     direct_util_recursive_pthread_mutex_init( &manager->input.lock );
     direct_util_recursive_pthread_mutex_init( &manager->output.lock );

     /* Initialize all wait conditions. */
     pthread_cond_init( &manager->response.wait, NULL );
     pthread_cond_init( &manager->input.wait, NULL );
     pthread_cond_init( &manager->output.wait, NULL );

     /* Set default buffer limit. */
     manager->input.max = IN_BUF_MAX;

     D_MAGIC_SET( manager, VoodooManager );

     /* Create all threads. */
     manager->dispatcher    = direct_thread_create( DTT_MESSAGING, manager_dispatch_loop,
                                                    manager, "Voodoo Dispatch" );

     manager->input.thread  = direct_thread_create( DTT_INPUT, manager_input_loop,
                                                    manager, "Voodoo Input" );

     manager->output.thread = direct_thread_create( DTT_OUTPUT, manager_output_loop,
                                                    manager, "Voodoo Output" );

     /* Return the new manager. */
     *ret_manager = manager;

     return DR_OK;
}

    int main(int argc, char ** argv){
    if(argc!=2){
      printf("Error!!! \n");
      return 0;
    }

    FILE *f = fopen(argv[1],"r");
   
    if(f==NULL){
      printf("Error!! \n");
    }
   
    char buffer[101];
    fgets(buffer, 101,f);
    int p = atoi(buffer);
   
    fgets(buffer, 101,f);
    int n = atoi(buffer);
   
   //allocating memory on heap for matrix
    g = (int*)malloc(sizeof(int)*n*n);
 
    int i=0;
    int j=0;
    for(i=0;i<n;i++){
      for(j=0;j<n;j++){   
	g[(i*n)+j]=0;       
      }
    }
  /*****   intializes the matrix while reading the file ******/
    while(fgets(buffer, 101,f)){
     if(strchr(buffer,' ')!=NULL){
      char *ic =(char*) strtok(buffer," ");
      int i = atoi(ic);
      ic =(char*) strtok(NULL," ");
      int j = atoi(ic);
     
      g[(i-1)*n+j-1] = 1;
	}
    }
   //thread variables
    pthread_t tid[p];
    pthread_mutex_init(&m,NULL);
    pthread_cond_init(&c,NULL);
    int ret[p];


    if(p>n){
		printf("Input Error!!\n");
		return 0;
    }
    int k = 0;
    int rows ;
    if(p==0)
    	rows = n;
    else
    	rows = (int)(n/p);
    int x = 0;

   count=0; 
  //time variables
   struct timeval start;
   struct timeval end;
   gettimeofday(&start, NULL);

 /** creating p threads and assigning each one what rows to do ***/
    for(x = 0; x < p-1; x++){
	//printf("main start:end %i:%i\n",x*rows,(x+1)*rows);
      myArgs *args = (myArgs*)malloc(sizeof(myArgs));
	args->n = n;
	args->start = x*rows;
	args->end = (x+1)*rows;
	args->p = p;
      
      ret[x] = pthread_create(&tid[x], NULL, trans, args);
    }
   // printf("main start:end %i:%i\n",x*rows,n);
   	myArgs *args = (myArgs*)malloc(sizeof(myArgs));
	args->n = n;
	args->start = (x)*rows;
	args->end = n;
	args->p = p;
	if(p==0){
		trans(args);
	}
	else
    	ret[p-1] = pthread_create(&tid[p-1], NULL, trans, args);
   
 /*******  main waits for all threds to finish **********/
    for(i=0; i < p; i++){
      pthread_join(tid[i], NULL);
    }
    gettimeofday(&end, NULL);
  
/****** calculating time ***********/
    int seconds = end.tv_sec - start.tv_sec;
    int suseconds = end.tv_usec - start.tv_usec;
    if(suseconds < 0){
      seconds--;
      suseconds = 1000000 + suseconds;
//.........这里部分代码省略.........

/*********************************
    线程池初始化
*********************************/
tpool_t *tpool_init(int num_worker_threads,             /*线程池线程个数    最大任务数      是否阻塞任务满的时候 */
        int max_queue_size, int do_not_block_when_full)
{
    int i, rtn;
    tpool_t *pool; 


lprintf(log, INFO, "init pool  begin ...\n");
    /* make the thread pool structure */
    if((pool = (struct tpool *)malloc(sizeof(struct tpool))) == NULL)
    {
        lprintf(log, FATAL, "Unable to malloc() thread pool!\n");
        return NULL;
    }

    /* set the desired thread pool values */
    pool->num_threads = num_worker_threads;                      /*工作线程个数*/
    pool->max_queue_size = max_queue_size;                       /*任务链表最大长度*/
    pool->do_not_block_when_full = do_not_block_when_full;       /*任务链表满时是否等待*/

    /* create an array to hold a ptr to the worker threads   生成线程池缓存 */
    if((pool->threads = (pthread_t *)malloc(sizeof(pthread_t)
                    *num_worker_threads)) == NULL)
    {
        lprintf(log, FATAL,"Unable to malloc() thread info array\n)");
        return NULL;
    }

    /* initialize the work queue  初始化任务链表 */
    pool->cur_queue_size = 0;
    pool->queue_head = NULL;
    pool->queue_tail = NULL;
    pool->queue_closed = 0;
    pool->shutdown = 0;

    /* create the mutexs and cond vars  初始化互斥变量 条件变量 用于线程之间的同步 */
    if((rtn = pthread_mutex_init(&(pool->queue_lock),NULL)) != 0) {
        lprintf(log,FATAL,"pthread_mutex_init %s",strerror(rtn));
        return NULL;
    }
    if((rtn = pthread_cond_init(&(pool->queue_not_empty),NULL)) != 0) {
        lprintf(log,FATAL,"pthread_cond_init %s",strerror(rtn));
        return NULL;
    }
    if((rtn = pthread_cond_init(&(pool->queue_not_full),NULL)) != 0) {
        lprintf(log,FATAL,"pthread_cond_init %s",strerror(rtn));
        return NULL;
    }
    if((rtn = pthread_cond_init(&(pool->queue_empty),NULL)) != 0) {
        lprintf(log,FATAL,"pthread_cond_init %s",strerror(rtn));
        return NULL;
    }

    /* 
     * from "man 3c pthread_attr_init"
     * Define the scheduling contention scope for the created thread.  The only
     * value     supported    in    the    LinuxThreads    implementation    is
     * !PTHREAD_SCOPE_SYSTEM!, meaning that the threads contend  for  CPU  time
     * with all processes running on the machine.
     *
     * so no need to explicitly set the SCOPE
     */

    /* create the individual worker threads */
    for(i = 0; i != num_worker_threads; i++)
    {
        if( (rtn=pthread_create(&(pool->threads[i]),NULL,
                        tpool_thread,(void*)pool)) != 0)
        {
            lprintf(log,FATAL,"pthread_create %s\n",strerror(rtn));
            return NULL;
        }

lprintf(log, INFO, "init pthread  %d!\n",i);
        
        
    }
lprintf(log, INFO, "init pool end!\n");
    return pool;
}

void *POSIX_Init(
  void *argument
)
{
  uint32_t            end_time;
  int                 status;
  int                 i;
  pthread_t           threadId;
  pthread_attr_t      attr;
  struct sched_param  param;
  int                 policy;

  TEST_BEGIN();

  /* Setup variables */
  status = pthread_create( &threadId, NULL, Blocker, NULL );
  rtems_test_assert( status == 0 );
  status = pthread_mutex_init(&MutexID, NULL);
  rtems_test_assert( status == 0 );
  status = pthread_cond_init(&CondID, NULL);
  rtems_test_assert( status == 0 );

  /* Setup so threads are created with a high enough priority to preempt
   * as they get created.
   */
  status = pthread_attr_init( &attr );
  rtems_test_assert( status == 0 );
  status = pthread_attr_setinheritsched( &attr, PTHREAD_EXPLICIT_SCHED );
  rtems_test_assert( status == 0 );
  status = pthread_attr_setschedpolicy( &attr, SCHED_FIFO );
  rtems_test_assert( status == 0 );
  param.sched_priority = sched_get_priority_max(SCHED_FIFO) / 2;
  status = pthread_attr_setschedparam( &attr, &param );
  rtems_test_assert( status == 0 );

  for ( i=0 ; i < N ; i++) {
    /* Threads will preempt as they are created, start up, and block */
    status = pthread_create(&threadId, &attr, Blocker, NULL);
    rtems_test_assert( status == 0 );
  }

  /* Now that all the threads have been created, adjust our priority
   * so we don't get preempted on broadcast.
   */
  status = pthread_getschedparam(pthread_self(), &policy, &param);
  rtems_test_assert( status == 0 );
  param.sched_priority = sched_get_priority_max(policy) - 1;
  status = pthread_setschedparam(pthread_self(), policy, &param);
  rtems_test_assert( status == 0 );

  benchmark_timer_initialize();
  status = pthread_cond_broadcast(&CondID);
  end_time = benchmark_timer_read();
  rtems_test_assert( status == 0 );

  put_time(
    "pthread_cond_broadcast: threads waiting no preempt",
    end_time,
    1,
    0,
    0
  );

  TEST_END();
  rtems_test_exit( 0 );

  return NULL;
}

RTDECL(int)  RTSemEventMultiCreateEx(PRTSEMEVENTMULTI phEventMultiSem, uint32_t fFlags, RTLOCKVALCLASS hClass,
                                     const char *pszNameFmt, ...)
{
    AssertReturn(!(fFlags & ~RTSEMEVENTMULTI_FLAGS_NO_LOCK_VAL), VERR_INVALID_PARAMETER);

    /*
     * Allocate semaphore handle.
     */
    int rc;
    struct RTSEMEVENTMULTIINTERNAL *pThis = (struct RTSEMEVENTMULTIINTERNAL *)RTMemAlloc(sizeof(struct RTSEMEVENTMULTIINTERNAL));
    if (pThis)
    {
        /*
         * Create the condition variable.
         */
        pthread_condattr_t CondAttr;
        rc = pthread_condattr_init(&CondAttr);
        if (!rc)
        {
#if defined(CLOCK_MONOTONIC) && defined(IPRT_HAVE_PTHREAD_CONDATTR_SETCLOCK)
            /* ASSUMES RTTimeSystemNanoTS() == RTTimeNanoTS() == clock_gettime(CLOCK_MONOTONIC). */
            rc = pthread_condattr_setclock(&CondAttr, CLOCK_MONOTONIC);
            pThis->fMonotonicClock = rc == 0;
#else
            pThis->fMonotonicClock = false;
#endif
            rc = pthread_cond_init(&pThis->Cond, &CondAttr);
            if (!rc)
            {
                /*
                 * Create the semaphore.
                 */
                pthread_mutexattr_t MutexAttr;
                rc = pthread_mutexattr_init(&MutexAttr);
                if (!rc)
                {
                    rc = pthread_mutex_init(&pThis->Mutex, &MutexAttr);
                    if (!rc)
                    {
                        pthread_mutexattr_destroy(&MutexAttr);
                        pthread_condattr_destroy(&CondAttr);

                        ASMAtomicXchgU32(&pThis->u32State, EVENTMULTI_STATE_NOT_SIGNALED);
                        ASMAtomicXchgU32(&pThis->cWaiters, 0);
#ifdef RTSEMEVENTMULTI_STRICT
                        if (!pszNameFmt)
                        {
                            static uint32_t volatile s_iSemEventMultiAnon = 0;
                            RTLockValidatorRecSharedInit(&pThis->Signallers, hClass, RTLOCKVAL_SUB_CLASS_ANY, pThis,
                                                         true /*fSignaller*/, !(fFlags & RTSEMEVENTMULTI_FLAGS_NO_LOCK_VAL),
                                                         "RTSemEventMulti-%u", ASMAtomicIncU32(&s_iSemEventMultiAnon) - 1);
                        }
                        else
                        {
                            va_list va;
                            va_start(va, pszNameFmt);
                            RTLockValidatorRecSharedInitV(&pThis->Signallers, hClass, RTLOCKVAL_SUB_CLASS_ANY, pThis,
                                                          true /*fSignaller*/, !(fFlags & RTSEMEVENTMULTI_FLAGS_NO_LOCK_VAL),
                                                          pszNameFmt, va);
                            va_end(va);
                        }
                        pThis->fEverHadSignallers = false;
#endif

                        *phEventMultiSem = pThis;
                        return VINF_SUCCESS;
                    }

                    pthread_mutexattr_destroy(&MutexAttr);
                }
                pthread_cond_destroy(&pThis->Cond);
            }
            pthread_condattr_destroy(&CondAttr);
        }

        rc = RTErrConvertFromErrno(rc);
        RTMemFree(pThis);
    }
    else
        rc = VERR_NO_MEMORY;

    return rc;

}

CFSMounter::MountRes CFSMounter::mount(const char * const ip, const char * const dir, const char * const local_dir, 
				       const FSType fstype, const char * const username, const char * const password, 
				       char * options1, char * options2)
{
	std::string cmd;
	pthread_mutex_init(&g_mut, NULL);
	pthread_cond_init(&g_cond, NULL);
	g_mntstatus=-1;

	FS_Support sup = fsSupported(fstype, true); /* keep modules if necessary */

	if (sup == CFSMounter::FS_UNSUPPORTED)
	{
		printf("[CFSMounter] FS type %d not supported\n", (int) fstype);
		return MRES_FS_NOT_SUPPORTED;
	}

	printf("[CFSMounter] Mount(%d) %s:%s -> %s\n", (int) fstype, ip, dir, local_dir);
	
	if (isMounted(local_dir))
	{
		printf("[CFSMounter] FS mount error %s already mounted\n", local_dir);
		return MRES_FS_ALREADY_MOUNTED;
	}

	if(options1[0] == '\0')
	{
		strcpy(options1,options2);
		options2[0] = '\0';
	}
	
	if((options1[0] == '\0') && (options2[0] == '\0'))
	{
		if(fstype == NFS)
		{
			strcpy(options1,"ro,soft,udp");
			strcpy(options2,"nolock,rsize=8192,wsize=8192");
		}
		else if(fstype == CIFS)
		{
			strcpy(options1,"ro");
			strcpy(options2,"");
		}
		else if(fstype == LUFS)
		{
			strcpy(options1,"");
			strcpy(options2,"");
		}
	}
	
	if(fstype == NFS)
	{
		cmd = "mount -t nfs ";
		cmd += ip;
		cmd += ':';
		cmd += dir;
		cmd += ' ';
		cmd += local_dir;
		cmd += " -o ";
		cmd += options1;
	}
	else if(fstype == CIFS)
	{
		cmd = "mount -t cifs //";
		cmd += ip;
		cmd += '/';
		cmd += dir;
		cmd += ' ';
		cmd += local_dir;
		cmd += " -o username=";
		cmd += username;
		cmd += ",password=";
		cmd += password;
		//cmd += ",unc=//"; for whats needed?
		//cmd += ip;
		//cmd += '/';
		//cmd += dir;
		//cmd += ',';
		//cmd += options1;
	}
	else
	{
		cmd = "lufsd none ";
		cmd += local_dir;
		cmd += " -o fs=ftpfs,username=";
		cmd += username;
		cmd += ",password=";
		cmd += password;
		cmd += ",host=";
		cmd += ip;
		cmd += ",root=/";
		cmd += dir;
		cmd += ',';
		cmd += options1;
	}
	
	if (options2[0] !='\0')
	{
		cmd += ',';
		cmd += options2;
//.........这里部分代码省略.........

string get_reply(string ip, string msg){
	int sockfd;
	struct addrinfo hints, *servinfo, *p;
	int rv;
	int numbytes;

	//add thread id to the end of message
	msg = msg + "_" + uint_to_str((unsigned int)pthread_self());
	memset(&hints, 0, sizeof hints);
	hints.ai_family = AF_UNSPEC;
	hints.ai_socktype = SOCK_DGRAM;

	if ((rv = getaddrinfo(ip.c_str(), SERVERPORT, &hints, &servinfo)) != 0) {
		fprintf(stderr, "getaddrinfo: %s\n", gai_strerror(rv));
		return "";
	}

	// loop through all the results and make a socket
	for(p = servinfo; p != NULL; p = p->ai_next) {
		if ((sockfd = socket(p->ai_family, p->ai_socktype,
				p->ai_protocol)) == -1) {
			perror("talker: socket");
			continue;
		}

		break;
	}

	if (p == NULL) {
		fprintf(stderr, "talker: failed to bind socket\n");
		return "";
	}

	if ((numbytes = sendto(sockfd, msg.c_str(), strlen(msg.c_str()), 0,
			 p->ai_addr, p->ai_addrlen)) == -1) {
		perror("talker: sendto");
		return "";
	}
	freeaddrinfo(servinfo);
	close(sockfd);
	cond_wait cw;
	time(&cw.time);
	if (pthread_mutex_init(cw.mutex, NULL) != 0) {
		perror("pthread_mutex_init() error");
		return "";
	}
	if (pthread_cond_init(cw.cond, NULL) != 0) {
		perror("pthread_cond_init() error");
		return "";
	}
	if (pthread_mutex_lock(cw.mutex) != 0) {
		perror("pthread_mutex_lock() error");
		return "";
	}
	cw.thread_id = (unsigned int)pthread_self();
	waiting_list.push_back(cw);
	//cout<<"TID:"<<cw.thread_id<<","<<cw.cond<<","<<cw.mutex<<endl;
	if (pthread_cond_wait(cw.cond, cw.mutex) != 0) {
	    perror("pthread_cond_timedwait() error");
	    return "";
	}
	//cout<<"out of wait & wls="<<waiting_list.size()<<endl;
	string ret;
	for(int i = 0; i < waiting_list.size(); i++){
		if(waiting_list[i].thread_id == (unsigned int)pthread_self()){
			ret = waiting_list[i].return_val;
			waiting_list.erase(waiting_list.begin() + i);
			break;
		}
	}
	//cout<<"out of for & wls="<<waiting_list.size()<<endl;
	return ret;
}

/*
 * Initializes a connection queue.
 */
static void cq_init(CQ *cq) {
    pthread_mutex_init(&cq->lock, NULL);
    pthread_cond_init(&cq->cond, NULL);
    cq->head = NULL;
    cq->tail = NULL;
}