說明

thread封裝了pthread, 完成的功能是, 使用fixed_queue作為workquque, 將需要被執行的thread function放入其中(enqueue), 使用thread_post來enqueue,enqueue完成后semophore發送信號給dequeue, 然后使用reactor中的epoll_wait監控到dequeue semaphore變更, 就讀出queue中的數據, 而queue中的item是thread function與args, 于是運行這個function, 完成任務執行功能. 

結構體

struct thread_t {   bool is_joined;  pthread_t pthread;  pid_t tid;  char name[THREAD_NAME_MAX + 1];   reactor_t *reactor;        // 對dequeue semophore fd進行監聽  fixed_queue_t *work_queue; //存放work_item};struct start_arg {   thread_t *thread;  semaphore_t *start_sem;  int error;};typedef struct { //用于thread_post傳遞需要的執行的function以及function需要的args  thread_fn func;   void *context;} work_item_t;
函數分析
創建thread以及reactor
thread_t *thread_new_sized(const char *name, size_t work_queue_capacity) {  assert(name != NULL);  assert(work_queue_capacity != 0);  thread_t *ret = osi_calloc(sizeof(thread_t));  if (!ret)    goto error;  ret->reactor = reactor_new();   if (!ret->reactor)    goto error;  ret->work_queue = fixed_queue_new(work_queue_capacity);  if (!ret->work_queue)    goto error;  // Start is on the stack, but we use a semaphore, so it's safe  struct start_arg start;  start.start_sem = semaphore_new(0);  if (!start.start_sem)    goto error;  strncpy(ret->name, name, THREAD_NAME_MAX);   start.thread = ret;  start.error = 0;  pthread_create(&ret->pthread, NULL, run_thread, &start);  semaphore_wait(start.start_sem); // A1:等待run_thread執行后,且使用PRctl更改名字成功才返回
等待thread_post中進行喚醒才會退出,在那之前包裝pthread_create創建出來的thread function可以一直執行, 這里返回即為主線程的退出  semaphore_free(start.start_sem);  if (start.error) //error值在run_pthread中設置的,見下面的B1    goto error;  return ret;error:;  if (ret) {    fixed_queue_free(ret->work_queue, osi_free);    reactor_free(ret->reactor);  }  osi_free(ret);  return NULL;}
thread_post
完成thread的main_loop函數設置
bool thread_post(thread_t *thread, thread_fn func, void *context) {  assert(thread != NULL);  assert(func != NULL);  // TODO(sharvil): if the current thread == |thread| and we've run out  // of queue space, we should abort this Operation, otherwise we'll  // deadlock.  // Queue item is freed either when the queue itself is destroyed  // or when the item is removed from the queue for dispatch.  work_item_t *item = (work_item_t *)osi_malloc(sizeof(work_item_t));  if (!item) {    LOG_ERROR("%s unable to allocate memory: %s", __func__, strerror(errno));    return false;  }  item->func = func; //設置回調函數  item->context = context;  fixed_queue_enqueue(thread->work_queue, item);//A4: enqueue后dequeue的semophore會從epoll_wait中回來, 然后就會去dequeue.對應B4  return true;}
run_thread阻塞等待poll_wait
static void *run_thread(void *start_arg) {  assert(start_arg != NULL);  struct start_arg *start = start_arg;  thread_t *thread = start->thread;  assert(thread != NULL);  if (prctl(PR_SET_NAME, (unsigned long)thread->name) == -1) {    LOG_ERROR("%s unable to set thread name: %s", __func__, strerror(errno));    start->error = errno;  // B1:設置thread name失敗    semaphore_post(start->start_sem);//B2:設置好了error變量可以讓new_thread返回退出了.    return NULL;  }  thread->tid = gettid();  semaphore_post(start->start_sem); //B3: 同B2  int fd = fixed_queue_get_dequeue_fd(thread->work_queue);   void *context = thread->work_queue;  reactor_object_t *work_queue_object = reactor_register(thread->reactor, fd, context, work_queue_read_cb, NULL);//B4:注意里面fd是dequeue的fd,因此在enqueue后actor會執行workqueue_read_cb讀取queue的數據進行處理  reactor_start(thread->reactor); //B5: poll_wait,等待enqueue來喚醒自己,在沒有enqueue之前都是休眠,有了enqueue就是有需要執行的任務(item->func)需要執行了.  reactor_unregister(work_queue_object);  // Make sure we dispatch all queued work items before exiting the thread.  // This allows a caller to safely tear down by enqueuing a teardown  // work item and then joining the thread.  size_t count = 0;  work_item_t *item = fixed_queue_try_dequeue(thread->work_queue);  while (item && count <= fixed_queue_capacity(thread->work_queue)) {    item->func(item->context); //取出callback函數進行執行    osi_free(item);    item = fixed_queue_try_dequeue(thread->work_queue);//逐個取出    ++count;   }  if (count > fixed_queue_capacity(thread->work_queue))    LOG_DEBUG("%s growing event queue on shutdown.", __func__);  return NULL;}
使用
下面以hci_layer.c中的thread為例說明一下使用.
1. 使用thread_new創建thread
這里僅僅傳入thread name:
  thread = thread_new("hci_thread");  if (!thread) {    LOG_ERROR("%s unable to create thread.", __func__);    goto error;  }這里面靜默創建了reactor與fixed_queue. 
2. 使用thread_post注冊thread需要處理的function,即enqueue item function,然后喚醒run_thread dequeue來執行
下面這個就是event_finish_startup為需要執行的函數
  thread_post(thread, event_finish_startup, NULL);
這個注冊的函數定義如下, 這里面可以看到調用到了HAL層的open, 即libbt-vendor.so中的open, 在switch中的"case BT_VND_OP_USERIAL_OPEN".
static void event_finish_startup(UNUSED_ATTR void *context) {  LOG_INFO("%s", __func__);  hal->open();  vendor->send_async_command(VENDOR_CONFIGURE_FIRMWARE, NULL);}
對于case BT_VND_OP_USERIAL_OPEN, 可以查看文件hardware/broadcom/libbt/src/bt_vendor_brcm.c中的代碼, 這里面調用到了硬件操作, 例如:
        case BT_VND_OP_USERIAL_OPEN:            {                int (*fd_array)[] = (int (*)[]) param;                int fd, idx;                ALOGW("--------- BT_VND_OP_USERIAL_OPEN Done ------------");                fd = userial_vendor_open((tUSERIAL_CFG *) &userial_init_cfg); //Open硬件操作                if (fd != -1)                {                    for (idx=0; idx < CH_MAX; idx++)                        (*fd_array)[idx] = fd;                    retval = 1;                }                /* retval contains numbers of open fd of HCI channels */            }            break;dequeue取出后執行這個函數, 然后返回.