自学教程：C++ uv_thread_create函数代码示例

int main() {  int tracklen = 10;  uv_thread_t hare_id;  uv_thread_t tortoise_id;  uv_thread_create(&hare_id, hare, &tracklen);  uv_thread_create(&tortoise_id, tortoise, &tracklen);  uv_thread_join(&hare_id);  uv_thread_join(&tortoise_id);  return 0;}

int main(){	uv_thread_t threads[2];	main_thread_id = uv_thread_self();	//获得当前线程的id 	uv_thread_create(threads + 0, check_thread, subthreads + 0);	//创建子线程	uv_thread_create(threads + 1, check_thread, subthreads + 1); 	uv_thread_join(threads + 0);	//thread join，阻塞，直到目标子线程执行完成	uv_thread_join(threads + 1); 	return 0;}

static bool gpio_set_edge(iotjs_gpio_t* gpio) {  IOTJS_VALIDATED_STRUCT_METHOD(iotjs_gpio_t, gpio);  char edge_path[GPIO_PATH_BUFFER_SIZE];  snprintf(edge_path, GPIO_PATH_BUFFER_SIZE, GPIO_PIN_FORMAT_EDGE, _this->pin);  iotjs_systemio_open_write_close(edge_path, gpio_edge_string[_this->edge]);  if (_this->direction == kGpioDirectionIn && _this->edge != kGpioEdgeNone) {    char value_path[GPIO_PATH_BUFFER_SIZE];    snprintf(value_path, GPIO_PATH_BUFFER_SIZE, GPIO_PIN_FORMAT_VALUE,             _this->pin);    if ((_this->platform_data->value_fd = open(value_path, O_RDONLY)) < 0) {      DLOG("GPIO Error in open");      return false;    }    // Create edge detection thread    // When the GPIO pin is closed, thread is terminated.    int ret = uv_thread_create(&_this->platform_data->thread,                               gpio_edge_detection_cb, (void*)gpio);    if (ret < 0) {      DLOG("GPIO Error in uv_thread_create");    }    return false;  }  return true;}

int main(int argc, char **argv) {  printf("starting...");  int r = uv_thread_create(&thread, thread_cb, NULL);  uv_thread_join(&thread);  printf("done./n");  return 0;}

static void init_once(void) {  unsigned int i;  const char* val;  nthreads = ARRAY_SIZE(default_threads);  val = getenv("UV_THREADPOOL_SIZE");  if (val != NULL) nthreads = atoi(val);  if (nthreads == 0) nthreads = 1;  if (nthreads > MAX_THREADPOOL_SIZE) nthreads = MAX_THREADPOOL_SIZE;  threads = default_threads;  if (nthreads > ARRAY_SIZE(default_threads)) {    threads = malloc(nthreads * sizeof(threads[0]));    if (threads == NULL) {      nthreads = ARRAY_SIZE(default_threads);      threads = default_threads;    }  }  if (uv_cond_init(&cond)) abort();  if (uv_mutex_init(&mutex)) abort();  QUEUE_INIT(&wq);  for (i = 0; i < nthreads; i++)    if (uv_thread_create(threads + i, worker, NULL)) abort();  initialized = 1;}

/** * Perform query operations using each session in multiple threads * * @param sessions Session container for closing sessions */void query_sessions(const SessionContainer& sessions) {  //Query session in multiple threads  unsigned int thread_count = sessions.count() * SESSION_STRESS_NUMBER_OF_SHARED_SESSION_THREADS;  BOOST_TEST_MESSAGE("/t/tThreads: " << thread_count);  std::vector<uv_thread_t> threads(thread_count);  std::vector<QuerySession> queries(thread_count);  for (unsigned int iterations = 0; iterations < SESSION_STRESS_NUMBER_OF_SHARED_SESSION_THREADS; ++iterations) {    for (unsigned int n = 0; n < sessions.count(); ++n) {      int thread_index = (sessions.count() * iterations) + n;      queries.push_back(QuerySession(sessions.sessions[n].get()));      uv_thread_create(&threads[thread_index], query_session, &queries.back());    }  }  //Ensure all threads have completed  for (unsigned int n = 0; n < thread_count; ++n) {    uv_thread_join(&threads[n]);  }  for (unsigned int n = 0; n < thread_count; ++n) {    //Timeouts are OK (especially on the minor chaos test)    CassError error_code = queries[n].error_code;    if (error_code != CASS_OK && error_code != CASS_ERROR_LIB_REQUEST_TIMED_OUT) {      BOOST_FAIL("Error occurred during query '" << std::string(cass_error_desc(error_code)));    }  }}

int pc_client_connect3(pc_client_t *client, struct sockaddr_in *addr) {  pc_connect_t *conn_req = pc_connect_req_new(addr);  if(client->enable_reconnect){    memcpy(&client->addr, addr, sizeof(struct sockaddr_in));  }  if(conn_req == NULL) {    fprintf(stderr, "Fail to malloc pc_connect_t./n");    goto error;  }  if(pc_connect(client, conn_req, NULL, pc__client_connect3_cb)) {    fprintf(stderr, "Fail to connect to server./n");    goto error;  }  uv_thread_create(&client->worker, pc__worker, client);  return 0;error:  if(conn_req) pc_connect_req_destroy(conn_req);  return -1;}

int uv_worker_init(uv_worker_t *worker, uv_loop_t *loop, int count, const char *name) {#ifdef DEBUG    worker->thread_id = uv_thread_self();#endif    worker->loop = loop;    worker->name = name;    worker->count = 0;    worker->close_cb = NULL;    worker->active_items = 0;    worker->msgr = (uv_messenger_t *)malloc(sizeof(uv_messenger_t));    int ret = uv_messenger_init(loop, worker->msgr, uv__worker_after);    if (ret < 0) {        free(worker->msgr);        return ret;    }    uv_messenger_unref(worker->msgr);    ret = uv_chan_init(&worker->chan);    if (ret < 0) return ret;    // Initialize all worker threads.    int i;    for (i = 0; i < count; i++) {        uv__worker_thread_t *worker_thread = (uv__worker_thread_t *)malloc(sizeof(uv__worker_thread_t));        worker_thread->worker = worker;        ret = uv_thread_create(&worker_thread->thread, uv__worker_thread_loop, worker_thread);        if (ret < 0) return ret;        worker->count++;    }    return 0;}

int pc_client_connect(pc_client_t *client, struct sockaddr_in *addr) {  pc_connect_t *conn_req = pc_connect_req_new(addr);  if(conn_req == NULL) {    LOGD( "Fail to malloc pc_connect_t./n");    goto error;  }  if(pc_connect(client, conn_req, NULL, pc__client_connected_cb)) {    LOGD( "Fail to connect to server./n");    goto error;  }  // 1. start work thread  // 2. wait connect result  uv_thread_create(&client->worker, pc__worker, client);  // TODO should set a timeout?  pc__cond_wait(client, 0);  pc_connect_req_destroy(conn_req);  if(PC_ST_WORKING != client->state) {    return -1;  }  return 0;error:  if(conn_req) pc_connect_req_destroy(conn_req);  return -1;}

static int run_test(int inprocess) {  uv_process_t process;  uv_thread_t tid;  int r;  if (inprocess) {    r = uv_thread_create(&tid, ipc_send_recv_helper_threadproc, (void *) 42);    ASSERT(r == 0);    uv_sleep(1000);    r = uv_pipe_init(uv_default_loop(), &ctx.channel, 1);    ASSERT(r == 0);    uv_pipe_connect(&ctx.connect_req, &ctx.channel, TEST_PIPENAME_3, connect_cb);  } else {    spawn_helper(&ctx.channel, &process, "ipc_send_recv_helper");    connect_cb(&ctx.connect_req, 0);  }  r = uv_run(uv_default_loop(), UV_RUN_DEFAULT);  ASSERT(r == 0);  ASSERT(recv_cb_count == 2);  if (inprocess) {    r = uv_thread_join(&tid);    ASSERT(r == 0);  }  return 0;}

LUALIB_API intluaopen_clientsocket(lua_State *L) {	luaL_checkversion(L);	luaL_Reg l[] = {		{ "connect", lconnect },		{ "recv", lrecv },		{ "send", lsend },		{ "close", lclose },		{ "usleep", lusleep },		{ NULL, NULL },	};	luaL_newlib(L, l);#ifdef _WINDOWS_	WORD wVersionRequested;	WSADATA wsaData;	wVersionRequested = MAKEWORD(2, 2);	WSAStartup(wVersionRequested, &wsaData);#endif	struct queue * q = lua_newuserdata(L, sizeof(*q));	memset(q, 0, sizeof(*q));	lua_pushcclosure(L, lreadstdin, 1);	lua_setfield(L, -2, "readstdin");	uv_thread_create(&pid, readline_stdin, q);	return 1;}

int main(int argc, char * argv[]){    handle_t handle;    if (argc != 5) {        printf("Usage:/n");        return -1;    }    const char * master_addr = argv[1];    int port = atoi(argv[2]);    int concurrency = atoi(argv[3]);    int repeat = atoi(argv[4]);    int ret = libuv_connect(master_addr, port, &handle);    printf("test: connect status: %d/n", ret);    if (ret < 0) {        return -1;    }    uv_thread_t ids[concurrency];    test_task_t tasks[concurrency];    for (int num = 0; num < concurrency; ++num) {        tasks[num].num = num;        tasks[num].handle = handle;        tasks[num].num_requests = repeat;        ret = uv_thread_create(&ids[num], client_test_task, (void *)&tasks[num]);        assert(ret == 0);    }    getchar();    return 0;}

/** * Close a number of sessions concurrently or sequentially * * @param sessions Session container for closing sessions * @param num_of_sessions Number of sessions to close concurrently * @param is_concurrently True if concurrent; false otherwise */void close_sessions(SessionContainer* sessions, unsigned int num_of_sessions, bool is_concurrently = true) {  //Close session threads (LIFO)  std::vector<uv_thread_t> threads(num_of_sessions);  for (unsigned int n = 0; n < num_of_sessions; ++n) {    if (is_concurrently) {      uv_thread_create(&threads[n], close_session, sessions->sessions[sessions->count() - n - 1].get());    } else {      close_session(sessions->sessions[sessions->count() - n - 1].get());    }  }  //Ensure all threads have completed  if (is_concurrently) {    for (unsigned int n = 0; n < num_of_sessions; ++n) {      uv_thread_join(&threads[n]);    }  }  sessions->sessions.clear();  //TODO: Remove sleep and create a timed wait for log messages (no boost)  for (unsigned int n = 0; n < (SESSION_STRESS_NUMBER_OF_ITERATIONS * 20); ++n) {    if (static_cast<unsigned int>(test_utils::CassLog::message_count()) != num_of_sessions) {      boost::this_thread::sleep_for(boost::chrono::milliseconds(100));    } else {      break;    }  }  BOOST_TEST_MESSAGE("/t/tClosed: " << test_utils::CassLog::message_count());}

int nub_thread_create(nub_loop_t* loop, nub_thread_t* thread) {  uv_async_t* async_handle;  int er;  async_handle = (uv_async_t*) malloc(sizeof(*async_handle));  CHECK_NE(NULL, async_handle);  er = uv_async_init(&loop->uvloop, async_handle, nub__work_signal_cb);  ASSERT(0 == er);  async_handle->data = thread;  thread->async_signal_ = async_handle;  ASSERT(uv_loop_alive(&loop->uvloop));  er = uv_sem_init(&thread->thread_lock_sem_, 0);  ASSERT(0 == er);  er = uv_sem_init(&thread->sem_wait_, 1);  ASSERT(0 == er);  fuq_init(&thread->incoming_);  thread->disposed = 0;  thread->nubloop = loop;  thread->disposed_cb_ = NULL;  thread->work.thread = thread;  thread->work.work_type = NUB_LOOP_QUEUE_NONE;  ++loop->ref_;  return uv_thread_create(&thread->uvthread, nub__thread_entry_cb, thread);}

int uv__platform_loop_init(uv_loop_t* loop, int default_loop) {  CFRunLoopSourceContext ctx;  int r;  if (uv__kqueue_init(loop))    return -1;  loop->cf_loop = NULL;  if ((r = uv_mutex_init(&loop->cf_mutex)))    return r;  if ((r = uv_sem_init(&loop->cf_sem, 0)))    return r;  QUEUE_INIT(&loop->cf_signals);  memset(&ctx, 0, sizeof(ctx));  ctx.info = loop;  ctx.perform = uv__cf_loop_cb;  loop->cf_cb = CFRunLoopSourceCreate(NULL, 0, &ctx);  if ((r = uv_thread_create(&loop->cf_thread, uv__cf_loop_runner, loop)))    return r;  /* Synchronize threads */  uv_sem_wait(&loop->cf_sem);  assert(ACCESS_ONCE(CFRunLoopRef, loop->cf_loop) != NULL);  return 0;}

/** * start UVRunner() */boolUVEventLoop::StartUVRunner(){	if (runUV) return true;	runUV = true;	uv_thread_create(&runner, Runner, this);	return true;}

static void init_threads(void) {  unsigned int i;  const char* val;  nthreads = ARRAY_SIZE(default_threads);  val = getenv("UV_THREADPOOL_SIZE");  if (val != NULL)    nthreads = atoi(val);  if (nthreads == 0)    nthreads = 1;  if (nthreads > MAX_THREADPOOL_SIZE)    nthreads = MAX_THREADPOOL_SIZE;  threads = default_threads;  if (nthreads > ARRAY_SIZE(default_threads)) {    threads = uv__malloc(nthreads * sizeof(threads[0]));    if (threads == NULL) {      nthreads = ARRAY_SIZE(default_threads);      threads = default_threads;    }  }  if (uv_cond_init(&cond))    abort();  if (uv_mutex_init(&mutex))    abort();  QUEUE_INIT(&wq);  for (i = 0; i < nthreads; i++) {#ifndef _WIN32    struct data_t *data = malloc(sizeof(struct data_t));    memset(data, '/0', sizeof(struct data_t));    data->nr = i;    if (uv_thread_create(threads + i, worker, data))      abort();#else    if (uv_thread_create(threads + i, worker, NULL))      abort();#endif	}  initialized = 1;}

void runStorageProcessor(){	uv_thread_t thread;	saveQueue = malloc(sizeof(struct threadqueue));	thread_queue_init(saveQueue);	uv_thread_create(&thread, storageProcessor, NULL);}

int main() {    uv_thread_t tid;    loop = create_loop();    async_handle = malloc(sizeof(uv_async_t));    uv_async_init(loop, async_handle, async_callback);    uv_thread_create(&tid, thread_main, NULL);    uv_async_send(async_handle);    uv_thread_join(&tid);}

bool Kinect::setup() {  if(ctx) {    printf("Error: the freenect context has been setup already./n");    return false;  }  if(freenect_init(&ctx, NULL) < 0) {    printf("Error: cannot init libfreenect./n");    return false;  }  freenect_set_log_level(ctx, FREENECT_LOG_DEBUG);  int ndevices = freenect_num_devices(ctx);  if(ndevices < 1) {    printf("Error: cannot find a kinect. @todo cleanup mem./n");    freenect_shutdown(ctx);    ctx = NULL;    return false;  }  printf("Number of found kinect devices: %d/n", ndevices);  freenect_select_subdevices(ctx, (freenect_device_flags)(FREENECT_DEVICE_MOTOR | FREENECT_DEVICE_CAMERA));  //freenect_select_subdevices(ctx, (freenect_device_flags)( FREENECT_DEVICE_CAMERA));  int devnum = 0;  if(freenect_open_device(ctx, &device, devnum) < 0) {    printf("Error: cannot open device: %d/n", devnum);    freenect_shutdown(ctx);    ctx = NULL;    return false;  }  freenect_set_user(device, this);  uint32_t w = 640;  uint32_t h = 480;  uint32_t nbytes = w * h * 3;  nbytes_rgb = nbytes;  rgb_back = new uint8_t[nbytes];  rgb_mid = new uint8_t[nbytes];  rgb_front = new uint8_t[nbytes];  depth_back = new uint8_t[nbytes];  depth_mid = new uint8_t[nbytes];  depth_front = new uint8_t[nbytes];  uv_mutex_lock(&mutex);    must_stop = false;  uv_mutex_unlock(&mutex);  uv_thread_create(&thread, kinect_thread, this);  return true;}

static void timer_cb(uv_timer_t* handle, int status) {  printf("TIMER_CB/n");  assert(handle != NULL);  assert(status == 0);  uv_thread_t thread_id;  uv_thread_create(&thread_id,thread_method,NULL);    }

int main(){	uv_thread_t threads[3];	int thread_nums[] = {1, 2, 1};	uv_barrier_init(&blocker, 4);	shared_num = 0;	uv_rwlock_init(&numlock);	uv_thread_create(&threads[0], reader, &thread_nums[0]);	uv_thread_create(&threads[1], reader, &thread_nums[1]);	uv_thread_create(&threads[2], writer, &thread_nums[2]);	uv_barrier_wait(&blocker);	uv_barrier_destroy(&blocker);	uv_rwlock_destroy(&numlock);	return 0;}

UI *ui_bridge_attach(UI *ui, ui_main_fn ui_main, event_scheduler scheduler){  UIBridgeData *rv = xcalloc(1, sizeof(UIBridgeData));  rv->ui = ui;  rv->bridge.rgb = ui->rgb;  rv->bridge.stop = ui_bridge_stop;  rv->bridge.resize = ui_bridge_resize;  rv->bridge.clear = ui_bridge_clear;  rv->bridge.eol_clear = ui_bridge_eol_clear;  rv->bridge.cursor_goto = ui_bridge_cursor_goto;  rv->bridge.mode_info_set = ui_bridge_mode_info_set;  rv->bridge.update_menu = ui_bridge_update_menu;  rv->bridge.busy_start = ui_bridge_busy_start;  rv->bridge.busy_stop = ui_bridge_busy_stop;  rv->bridge.mouse_on = ui_bridge_mouse_on;  rv->bridge.mouse_off = ui_bridge_mouse_off;  rv->bridge.mode_change = ui_bridge_mode_change;  rv->bridge.set_scroll_region = ui_bridge_set_scroll_region;  rv->bridge.scroll = ui_bridge_scroll;  rv->bridge.highlight_set = ui_bridge_highlight_set;  rv->bridge.put = ui_bridge_put;  rv->bridge.bell = ui_bridge_bell;  rv->bridge.visual_bell = ui_bridge_visual_bell;  rv->bridge.update_fg = ui_bridge_update_fg;  rv->bridge.update_bg = ui_bridge_update_bg;  rv->bridge.update_sp = ui_bridge_update_sp;  rv->bridge.flush = ui_bridge_flush;  rv->bridge.suspend = ui_bridge_suspend;  rv->bridge.set_title = ui_bridge_set_title;  rv->bridge.set_icon = ui_bridge_set_icon;  rv->scheduler = scheduler;  for (UIWidget i = 0; (int)i < UI_WIDGETS; i++) {    rv->bridge.ui_ext[i] = ui->ui_ext[i];  }  rv->ui_main = ui_main;  uv_mutex_init(&rv->mutex);  uv_cond_init(&rv->cond);  uv_mutex_lock(&rv->mutex);  rv->ready = false;  if (uv_thread_create(&rv->ui_thread, ui_thread_run, rv)) {    abort();  }  while (!rv->ready) {    uv_cond_wait(&rv->cond, &rv->mutex);  }  uv_mutex_unlock(&rv->mutex);  ui_attach_impl(&rv->bridge);  return &rv->bridge;}

void worker_mgr_t::start_worker(uint32_t ctx_num){	spin_lock(m_worker_lock);	if (!m_exiting && m_worker_num < m_max_worker_num && (ctx_num > m_worker_num * WORKER_LOAD_THRESHOLD)) {		int32_t index = m_worker_num;		m_worker[index] = new worker_t(index);		atomic_barrier();		++m_worker_num;		uv_thread_create(&m_thread[index], worker_entry, (void*)index);	}	spin_unlock(m_worker_lock);}

static int luv_new_thread(lua_State* L) {  int ret;  size_t len;  const char* buff;  luv_thread_t* thread;  int cbidx = 1;#if LUV_UV_VERSION_GEQ(1, 26, 0)  uv_thread_options_t options;  options.flags = UV_THREAD_NO_FLAGS;#endif  thread = (luv_thread_t*)lua_newuserdata(L, sizeof(*thread));  memset(thread, 0, sizeof(*thread));  luaL_getmetatable(L, "uv_thread");  lua_setmetatable(L, -2);#if LUV_UV_VERSION_GEQ(1, 26, 0)  if (lua_type(L, 1) == LUA_TTABLE)  {    cbidx++;    lua_getfield(L, 1, "stack_size");    if (!lua_isnil(L, -1))    {      options.flags |= UV_THREAD_HAS_STACK_SIZE;      if (lua_isnumber(L, -1)) {        options.stack_size = lua_tointeger(L, -1);      }      else {        return luaL_argerror(L, 1, "stack_size option must be a number if set");      }    }    lua_pop(L, 1);  }#endif  buff = luv_thread_dumped(L, cbidx, &len);  //clear in luv_thread_gc or in child threads  thread->argc = luv_thread_arg_set(L, &thread->arg, cbidx+1, lua_gettop(L) - 1, LUVF_THREAD_UHANDLE);  thread->len = len;  thread->code = (char*)malloc(thread->len);  memcpy(thread->code, buff, len);#if LUV_UV_VERSION_GEQ(1, 26, 0)  ret = uv_thread_create_ex(&thread->handle, &options, luv_thread_cb, thread);#else  ret = uv_thread_create(&thread->handle, luv_thread_cb, thread);#endif  if (ret < 0) return luv_error(L, ret);  return 1;}

// same as ./10-locks, but using try* functions when obtaining read and write locks//int main() {  int r;  const int count = 4;  fprintf(stderr, "barrier: init/n");  uv_barrier_init(&blocker, count);  shared_num = 0;  // https://github.com/thlorenz/libuv-dox/blob/master/methods.md#rwlock  fprintf(stderr, "rwlock: init/n");  r = uv_rwlock_init(&numlock);  if (r) ERROR("rwlock_init", r);  uv_thread_t threads[3];  int thread_nums[] = { 1, 2, 1 };  r = uv_thread_create(&threads[0], reader_entry, &thread_nums[0]);  if (r) ERROR("thread_create", r);  r = uv_thread_create(&threads[1], reader_entry, &thread_nums[1]);  if (r) ERROR("thread_create", r);  r = uv_thread_create(&threads[2], writer_entry, &thread_nums[2]);  if (r) ERROR("thread_create", r);  // https://github.com/thlorenz/libuv-dox/blob/master/methods.md#barrier  fprintf(stderr, "barrier: wait/n");  uv_barrier_wait(&blocker);  fprintf(stderr, "barrier: destroy/n");  uv_barrier_destroy(&blocker);  fprintf(stderr, "rwlock: destroy/n");  uv_rwlock_destroy(&numlock);  if (r) ERROR("rwlock_destroy", r);  return 0;}

static void prepare_cb(uv_prepare_t* handle, int status) {  int r;  ASSERT(handle == &prepare);  ASSERT(status == 0);  if (prepare_cb_called++)    return;  r = uv_thread_create(&thread, thread_cb, NULL);  ASSERT(r == 0);  uv_mutex_unlock(&mutex);}

bool Graph::start() {    if(!address.size()) {    printf("error: no address set; cannot connect to grapher./n");    return false;  }  uv_thread_create(&thread, graph_thread, this);  is_running = true;  return true;}

async_worker_t *async_worker_create(void) {	async_worker_t *worker = calloc(1, sizeof(struct async_worker_s));	if(!worker) goto bail;	if(uv_sem_init(&worker->sem, 0) < 0) goto bail;	worker->async.data = worker;	if(uv_async_init(loop, &worker->async, leave) < 0) goto bail;	if(uv_thread_create(&worker->thread, work, worker) < 0) goto bail;	return worker;bail:	async_worker_free(worker);	return NULL;}