码迷，mamicode.com
首页 > 其他好文 > 详细

libuv学习笔记（24）

时间：2016-07-03 19:25:37      阅读：152      评论：0      收藏：0      [点我收藏+]

标签：

libuv学习笔记（24）

线程相关数据结构与函数（2）

数据结构

typedef union {//读写锁
  struct {
    unsigned int num_readers_;
    CRITICAL_SECTION num_readers_lock_;
    HANDLE write_semaphore_;
  } state_;
  /* TODO: remove me in v2.x. */
  struct {
    SRWLOCK unused_;
  } unused1_;
  /* TODO: remove me in v2.x. */
  struct {
    uv_mutex_t unused1_;
    uv_mutex_t unused2_;
  } unused2_;
} uv_rwlock_t;
typedef HANDLE uv_sem_t;//信号量
typedef union {//线程池同步管理
  CONDITION_VARIABLE cond_var;//调用系统API实现
  struct {//libuv自己实现
    unsigned int waiters_count;
    CRITICAL_SECTION waiters_count_lock;
    HANDLE signal_event;
    HANDLE broadcast_event;
  } fallback;
} uv_cond_t;
typedef struct {
  unsigned int n;
  unsigned int count;
  uv_mutex_t mutex;
  uv_sem_t turnstile1;
  uv_sem_t turnstile2;
} uv_barrier_t;

读写锁相关函数

初始化

int uv_rwlock_init(uv_rwlock_t* rwlock) {
  //创建信号量，最大资源数和可用资源数都为1
  HANDLE handle = CreateSemaphoreW(NULL, 1, 1, NULL);
  if (handle == NULL)
    return uv_translate_sys_error(GetLastError());
  rwlock->state_.write_semaphore_ = handle;
  //初始化临界区
  InitializeCriticalSection(&rwlock->state_.num_readers_lock_);
  //初始化读请求的数量
  rwlock->state_.num_readers_ = 0;
  return 0;
}

释放读写锁

void uv_rwlock_destroy(uv_rwlock_t* rwlock) {
  DeleteCriticalSection(&rwlock->state_.num_readers_lock_);
  CloseHandle(rwlock->state_.write_semaphore_);
}

读锁定

void uv_rwlock_rdlock(uv_rwlock_t* rwlock) {
  //进入临界区
  EnterCriticalSection(&rwlock->state_.num_readers_lock_);
  //递增读请求数量
  if (++rwlock->state_.num_readers_ == 1) {
    //如果为1，说明没有其他的地方读锁定了，等待资源可用
    DWORD r = WaitForSingleObject(rwlock->state_.write_semaphore_, INFINITE);
    if (r != WAIT_OBJECT_0)
      uv_fatal_error(GetLastError(), "WaitForSingleObject");
  }
  //离开临界区
  LeaveCriticalSection(&rwlock->state_.num_readers_lock_);
}

尝试读锁定

int uv_rwlock_tryrdlock(uv_rwlock_t* rwlock) {
  int err;
  //尝试进入临界区
  if (!TryEnterCriticalSection(&rwlock->state_.num_readers_lock_))
    return UV_EBUSY;
  err = 0;
  if (rwlock->state_.num_readers_ == 0) {
    //获取资源
    DWORD r = WaitForSingleObject(rwlock->state_.write_semaphore_, 0);
    if (r == WAIT_OBJECT_0)
      rwlock->state_.num_readers_++;
    else if (r == WAIT_TIMEOUT)
      err = UV_EBUSY;
    else if (r == WAIT_FAILED)
      uv_fatal_error(GetLastError(), "WaitForSingleObject");
  } else {
    rwlock->state_.num_readers_++;
  }

  LeaveCriticalSection(&rwlock->state_.num_readers_lock_);
  return err;
}

释放读锁定

void uv_rwlock_rdunlock(uv_rwlock_t* rwlock) {
  //进入临界区
  EnterCriticalSection(&rwlock->state_.num_readers_lock_);
  if (--rwlock->state_.num_readers_ == 0) {
    //没有都请求了，释放资源
    if (!ReleaseSemaphore(rwlock->state_.write_semaphore_, 1, NULL))
      uv_fatal_error(GetLastError(), "ReleaseSemaphore");
  }
  //离开临界区
  LeaveCriticalSection(&rwlock->state_.num_readers_lock_);
}

写锁定

void uv_rwlock_wrlock(uv_rwlock_t* rwlock) {
  //等待资源可用
  DWORD r = WaitForSingleObject(rwlock->state_.write_semaphore_, INFINITE);
  if (r != WAIT_OBJECT_0)
    uv_fatal_error(GetLastError(), "WaitForSingleObject");
}

尝试写锁定

int uv_rwlock_trywrlock(uv_rwlock_t* rwlock) {
  DWORD r = WaitForSingleObject(rwlock->state_.write_semaphore_, 0);
  if (r == WAIT_OBJECT_0)
    return 0;
  else if (r == WAIT_TIMEOUT)
    return UV_EBUSY;
  else
    uv_fatal_error(GetLastError(), "WaitForSingleObject");
}

解除写锁定

void uv_rwlock_wrunlock(uv_rwlock_t* rwlock) {
  //释放信号量
  if (!ReleaseSemaphore(rwlock->state_.write_semaphore_, 1, NULL))
    uv_fatal_error(GetLastError(), "ReleaseSemaphore");
}

信号量相关API

初始化

int uv_sem_init(uv_sem_t* sem, unsigned int value) {
  *sem = CreateSemaphore(NULL, value, INT_MAX, NULL);//创建信号量
  if (*sem == NULL)
    return uv_translate_sys_error(GetLastError());
  else
    return 0;
}

释放信号量

void uv_sem_destroy(uv_sem_t* sem) {
  if (!CloseHandle(*sem))//关闭句柄
    abort();
}

发送

void uv_sem_post(uv_sem_t* sem) {
  if (!ReleaseSemaphore(*sem, 1, NULL))//释放一个资源
    abort();
}

等待

void uv_sem_wait(uv_sem_t* sem) {
  if (WaitForSingleObject(*sem, INFINITE) != WAIT_OBJECT_0)
    abort();
}

尝试等待

int uv_sem_trywait(uv_sem_t* sem) {
  DWORD r = WaitForSingleObject(*sem, 0);

  if (r == WAIT_OBJECT_0)
    return 0;

  if (r == WAIT_TIMEOUT)
    return UV_EAGAIN;

  abort();
  return -1; /* Satisfy the compiler. */
}

线程池同步相关API

初始化

int uv_cond_init(uv_cond_t* cond) {
  uv__once_init();//调用uv_init全局初始化
  if (HAVE_CONDVAR_API())//系统支持相关的API
    return uv_cond_condvar_init(cond);
  else
    return uv_cond_fallback_init(cond);
}

系统支持

static int uv_cond_condvar_init(uv_cond_t* cond) {
  //调用InitializeConditionVariable初始化
  pInitializeConditionVariable(&cond->cond_var);
  return 0;
}

系统不支持

static int uv_cond_fallback_init(uv_cond_t* cond) {
  int err;
  cond->fallback.waiters_count = 0;
  //初始化临界区
  InitializeCriticalSection(&cond->fallback.waiters_count_lock);
  //新建一个自动还原状态的事件，初始化为无信号
  cond->fallback.signal_event = CreateEvent(NULL,  /* no security */
                                            FALSE, /* auto-reset event */
                                            FALSE, /* non-signaled initially */
                                            NULL); /* unnamed */
  if (!cond->fallback.signal_event) {
    err = GetLastError();
    goto error2;
  }
  //创建一个需要手动改变状态的事件
  cond->fallback.broadcast_event = CreateEvent(NULL,  /* no security */
                                               TRUE,  /* manual-reset */
                                               FALSE, /* non-signaled */
                                               NULL); /* unnamed */
  if (!cond->fallback.broadcast_event) {
    err = GetLastError();
    goto error;
  }
  return 0;
error:
  CloseHandle(cond->fallback.signal_event);
error2:
  DeleteCriticalSection(&cond->fallback.waiters_count_lock);
  return uv_translate_sys_error(err);
}

释放

void uv_cond_destroy(uv_cond_t* cond) {
  if (HAVE_CONDVAR_API())//根据系统的支持情况释放资源
    uv_cond_condvar_destroy(cond);
  else
    uv_cond_fallback_destroy(cond);
}

设为有信号状态

void uv_cond_signal(uv_cond_t* cond) {
  if (HAVE_CONDVAR_API())
    uv_cond_condvar_signal(cond);
  else
    uv_cond_fallback_signal(cond);
}

使用系统API

static void uv_cond_condvar_signal(uv_cond_t* cond) {
  pWakeConditionVariable(&cond->cond_var);//唤醒
}

libuv自己实现

static void uv_cond_fallback_signal(uv_cond_t* cond) {
  int have_waiters;
  EnterCriticalSection(&cond->fallback.waiters_count_lock);
  have_waiters = cond->fallback.waiters_count > 0;
  LeaveCriticalSection(&cond->fallback.waiters_count_lock);
  if (have_waiters)
    //将signal_event设为有信号，这样等待该event的线程中的一个将会唤醒
    SetEvent(cond->fallback.signal_event);
}

广播通知（所有等待的线程都被唤醒）

void uv_cond_broadcast(uv_cond_t* cond) {
  if (HAVE_CONDVAR_API())
    uv_cond_condvar_broadcast(cond);
  else
    uv_cond_fallback_broadcast(cond);
}

使用系统API实现

static void uv_cond_condvar_broadcast(uv_cond_t* cond) {
  pWakeAllConditionVariable(&cond->cond_var);//唤醒所有线程
}

libuv自己实现

static void uv_cond_fallback_broadcast(uv_cond_t* cond) {
  int have_waiters;
  EnterCriticalSection(&cond->fallback.waiters_count_lock);
  have_waiters = cond->fallback.waiters_count > 0;
  LeaveCriticalSection(&cond->fallback.waiters_count_lock);
  if (have_waiters)
    //所有等待broadcast_event的线程都被唤醒
    SetEvent(cond->fallback.broadcast_event);
}

等待

void uv_cond_wait(uv_cond_t* cond, uv_mutex_t* mutex) {
  if (HAVE_CONDVAR_API())
    uv_cond_condvar_wait(cond, mutex);
  else
    uv_cond_fallback_wait(cond, mutex);
}

通过系统API实现

static void uv_cond_condvar_wait(uv_cond_t* cond, uv_mutex_t* mutex) {
  //等待cond_var，并离开临界区mutex
  if (!pSleepConditionVariableCS(&cond->cond_var, mutex, INFINITE))
    abort();
}

libuv自己实现

static int uv_cond_wait_helper(uv_cond_t* cond, uv_mutex_t* mutex,
    DWORD dwMilliseconds) {
  DWORD result;
  int last_waiter;
  HANDLE handles[2] = {
    cond->fallback.signal_event,
    cond->fallback.broadcast_event
  };
  EnterCriticalSection(&cond->fallback.waiters_count_lock);
  cond->fallback.waiters_count++;//等待的线程计数加一
  LeaveCriticalSection(&cond->fallback.waiters_count_lock);
  //离开临界区
  uv_mutex_unlock(mutex);
  //等待任意一个event为有信号状态
  result = WaitForMultipleObjects(2, handles, FALSE, dwMilliseconds);
  EnterCriticalSection(&cond->fallback.waiters_count_lock);
  cond->fallback.waiters_count--;
  //如果信号是broadcast_event并且没有等待者了，说明这是最后一个等待的线程
  last_waiter = result == WAIT_OBJECT_0 + 1
      && cond->fallback.waiters_count == 0;
  LeaveCriticalSection(&cond->fallback.waiters_count_lock);
  //对于最后一个等待线程，手动将broadcast_event设为无信号
  if (last_waiter) {
    ResetEvent(cond->fallback.broadcast_event);
  }
  uv_mutex_lock(mutex);
  if (result == WAIT_OBJECT_0 || result == WAIT_OBJECT_0 + 1)
    return 0;
  if (result == WAIT_TIMEOUT)
    return UV_ETIMEDOUT;
  abort();
  return -1; /* Satisfy the compiler. */
}

等待指定时间，与上面的函数相比，多了一个超时参数

int uv_cond_timedwait(uv_cond_t* cond, uv_mutex_t* mutex,
    uint64_t timeout) {
  if (HAVE_CONDVAR_API())
    return uv_cond_condvar_timedwait(cond, mutex, timeout);
  else
    return uv_cond_fallback_timedwait(cond, mutex, timeout);
}

barrier相关函数

初始化

int uv_barrier_init(uv_barrier_t* barrier, unsigned int count) {
  int err;
  barrier->n = count;//任务数量
  barrier->count = 0;
  err = uv_mutex_init(&barrier->mutex);
  if (err)
    return err;
  err = uv_sem_init(&barrier->turnstile1, 0);
  if (err)
    goto error2;
  err = uv_sem_init(&barrier->turnstile2, 1);
  if (err)
    goto error;
  return 0;
error:
  uv_sem_destroy(&barrier->turnstile1);
error2:
  uv_mutex_destroy(&barrier->mutex);
  return err;
}

释放

void uv_barrier_destroy(uv_barrier_t* barrier) {
  uv_sem_destroy(&barrier->turnstile2);
  uv_sem_destroy(&barrier->turnstile1);
  uv_mutex_destroy(&barrier->mutex);
}

等待

int uv_barrier_wait(uv_barrier_t* barrier) {
  int serial_thread;
  uv_mutex_lock(&barrier->mutex);//进入临界区
  if (++barrier->count == barrier->n) {//最后一个任务
    uv_sem_wait(&barrier->turnstile2);//等待第二个信号量
    uv_sem_post(&barrier->turnstile1);//释放第一个信号量
  }
  uv_mutex_unlock(&barrier->mutex);
  uv_sem_wait(&barrier->turnstile1);//等待第一个信号量
  uv_sem_post(&barrier->turnstile1);//释放第一个信号量
  uv_mutex_lock(&barrier->mutex);//进入临界区
  serial_thread = (--barrier->count == 0);
  if (serial_thread) {//最后一个任务
    uv_sem_wait(&barrier->turnstile1);/等待第一个
    uv_sem_post(&barrier->turnstile2);//释放第二个
  }
  uv_mutex_unlock(&barrier->mutex);

  uv_sem_wait(&barrier->turnstile2);//等待第二个
  uv_sem_post(&barrier->turnstile2);//释放第二个
  return serial_thread;
}

当某一线程需要等待其他一些线程任务完成之后才能继续运行时，可以使用barrier。
流程如下：
a.所有的相关线程都调用uv_barrier_wait等待同一个uv_barrier_t，此时除了最后一个，都会在等待第一个信号量的地方阻塞。
b.最后一个调用uv_barrier_wait的线程会等待第二个信号量，此时第二个信号量没有资源，然后释放第一个信号量。
c.之前阻塞在等待第一个信号量的线程中的一个获取信号量，继续运行，接着释放第一个信号量，这导致所有阻塞的进程都会因此一个一个的继续运行
d.接着除了最后一个线程，所有的线程都会阻塞在等待第二个信号量的地方。
e.最后一个线程，等待第一个信号量，此时第一个信号量重新变为没有资源，接着释放第二个信号量，激活其他等待的线程
f.其他阻塞在等待第二个信号量的线程，一个一个的唤醒，最终第二个信号量的资源为一，所有的线程继续运行

注意，最后一个线程是指逻辑上的最后一个，并不是确定的。
如果线程数比初始化时设置的任务数少，那么都会阻塞
如果线程池数比任务数多，那么等待任务数量的线程之后就会继续运行。

libuv学习笔记（24）

标签：

原文地址：http://blog.csdn.net/paohui0134/article/details/51810989
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！