专栏首页原创分享线程源码分析之线程库的初始化和线程的管理(基于linuxthreads2.0.1)

线程源码分析之线程库的初始化和线程的管理(基于linuxthreads2.0.1)

初步分析一下线程的初始化和管理,有些还没理解,后续读懂了再继续分析。

线程库的初始化代码如下。

// 在main函数之前执行该函数
void __pthread_initialize(void) __attribute__((constructor));

void __pthread_initialize(void)
{
  struct sigaction sa;
  sigset_t mask;

  /* We may be called by others.  This may happen if the constructors
     are not called in the order we need.  */
  if (__pthread_initial_thread_bos != NULL)
    return;

  /* For the initial stack, reserve at least STACK_SIZE bytes of stack
     below the current stack address, and align that on a
     STACK_SIZE boundary. */
  __pthread_initial_thread_bos =
    // 按STACK_SIZE大小对齐
    (char *)(((long)CURRENT_STACK_FRAME - 2 * STACK_SIZE) & ~(STACK_SIZE - 1));
  /* Update the descriptor for the initial thread. */
  // 即main函数代表的主进程id
  __pthread_initial_thread.p_pid = getpid();
  /* If we have special thread_self processing, initialize that for the
     main thread now.  */
#ifdef INIT_THREAD_SELF
  INIT_THREAD_SELF(&__pthread_initial_thread);
#endif
  /* Setup signal handlers for the initial thread.
     Since signal handlers are shared between threads, these settings
     will be inherited by all other threads. */
  // 为两个信号注册处理函数
  sa.sa_handler = __pthread_sighandler;
  sigemptyset(&sa.sa_mask);
  sa.sa_flags = SA_RESTART; /* does not matter for regular threads, but
                               better for the thread manager */
  sigaction(PTHREAD_SIG_RESTART, &sa, NULL);
  sa.sa_handler = pthread_handle_sigcancel;
  sa.sa_flags = 0;
  sigaction(PTHREAD_SIG_CANCEL, &sa, NULL);

  /* Initially, block PTHREAD_SIG_RESTART. Will be unblocked on demand. */
  // 屏蔽restart信号
  sigemptyset(&mask);
  sigaddset(&mask, PTHREAD_SIG_RESTART);
  sigprocmask(SIG_BLOCK, &mask, NULL);
  /* Register an exit function to kill all other threads. */
  /* Do it early so that user-registered atexit functions are called
     before pthread_exit_process. */
  // 注册退出时执行的函数
  __on_exit(pthread_exit_process, NULL);
}

在执行main函数之前会先执行__pthread_initialize函数,该函数做的事情主要有

1 在栈上分配一块内存。

2 保存当前进程,进main函数对应的进程的pid。

3 注册两个信号处理函数。

4 注册退出时执行的函数

接下来我们会调用pthread_create进行线程的创建。我们来看看该函数做了什么。

int pthread_create(pthread_t *thread, const pthread_attr_t *attr,
                   void * (*start_routine)(void *), void *arg)
{
  pthread_t self = thread_self();
  struct pthread_request request;
  // 还没执行过pthread_initialize_manager则执行,用于初始化manager线程
  if (__pthread_manager_request < 0) {
    if (pthread_initialize_manager() < 0) return EAGAIN;
  }
  // 给manager发一下请求
  request.req_thread = self;
  request.req_kind = REQ_CREATE;
  request.req_args.create.attr = attr;
  request.req_args.create.fn = start_routine;
  request.req_args.create.arg = arg;
  // 获取当前线程的信号掩码
  sigprocmask(SIG_SETMASK, (const sigset_t *) NULL,
              &request.req_args.create.mask);
  // 通过管道写入,通知manager线程,新建一个线程
  __libc_write(__pthread_manager_request, (char *) &request, sizeof(request));
  // 挂起,等待manager唤醒
  suspend(self);
  // 等于0说明创建成功,否则返回失败的错误码,p_retval在pthread_handle_create中设置
  if (self->p_retcode == 0) *thread = (pthread_t) self->p_retval;
  return self->p_retcode;
}

我们发现,该函数没有做实际的事情,他通过往管道写了一些数据。这时候就要先看pthread_initialize_manager函数了。

static int pthread_initialize_manager(void)
{
  int manager_pipe[2];

  /* Setup stack for thread manager */
  // 在堆上分配一块内存用于manager线程的栈
  __pthread_manager_thread_bos = malloc(THREAD_MANAGER_STACK_SIZE);
  if (__pthread_manager_thread_bos == NULL) return -1;
  // limit
  __pthread_manager_thread_tos =
    __pthread_manager_thread_bos + THREAD_MANAGER_STACK_SIZE;
  /* Setup pipe to communicate with thread manager */
  if (pipe(manager_pipe) == -1) {
    free(__pthread_manager_thread_bos);
    return -1;
  }
  __pthread_manager_request = manager_pipe[1]; /* writing end */
  __pthread_manager_reader = manager_pipe[0]; /* reading end */
  /* Start the thread manager */
  // 新建一个manager线程,manager_pipe是__thread_manager函数的入参
  if (__clone(__pthread_manager,
	      __pthread_manager_thread_tos,
	      CLONE_VM | CLONE_FS | CLONE_FILES | CLONE_SIGHAND,
	      (void *)(long)manager_pipe[0]) == -1) {
    free(__pthread_manager_thread_bos);
    __libc_close(manager_pipe[0]);
    __libc_close(manager_pipe[1]);
    __pthread_manager_request = -1;
    return -1;
  }
  return 0;
}

该函数做了几件事情

1 在堆上申请一块内存用作manager线程的栈

2 创建了一个管道,用于manager线程和其他线程通信。

3 然后新建了一个进程,然后执行__pthread_manager函数。(具体可参考http://www.man7.org/linux/man-pages/man2/clone.2.html)

manager线程是linuxthreads线程库比较重要的存在,他是管理其他线程的线程。我们接着看_pthread_manager函数的代码。

/* The server thread managing requests for thread creation and termination */

int __pthread_manager(void *arg)
{
  // 管道的读端
  int reqfd = (long)arg;
  sigset_t mask;
  fd_set readfds;
  struct timeval timeout;
  int n;
  struct pthread_request request;

  /* If we have special thread_self processing, initialize it.  */
#ifdef INIT_THREAD_SELF
  INIT_THREAD_SELF(&__pthread_manager_thread);
#endif
  /* Block all signals except PTHREAD_SIG_RESTART */
  // 初始化为全1
  sigfillset(&mask);
  // 设置某一位为0,这里设置可以处理restart信号
  sigdelset(&mask, PTHREAD_SIG_RESTART);
  // 设置进程的信号掩码
  sigprocmask(SIG_SETMASK, &mask, NULL);
  /* Enter server loop */
  while(1) {
    // 清0
    FD_ZERO(&readfds);
    // 置某位为1,位数由reqfd算得,这里是管道读端的文件描述符
    FD_SET(reqfd, &readfds);
    // 阻塞的超时时间
    timeout.tv_sec = 2;
    timeout.tv_usec = 0;
    // 定时阻塞等待管道有数据可读
    n = __select(FD_SETSIZE, &readfds, NULL, NULL, &timeout);
    /* Check for termination of the main thread */
    // 父进程id为1说明主进程(线程)已经退出,子进程被init(pid=1)进程接管了,
    if (getppid() == 1) {
      // 0说明不需要给主线程发,因为他已经退出了
      pthread_kill_all_threads(SIGKILL, 0);
      return 0;
    }
    /* Check for dead children */
    if (terminated_children) {
      terminated_children = 0;
      pthread_reap_children();
    }
    /* Read and execute request */
    // 管道有数据可读
    if (n == 1 && FD_ISSET(reqfd, &readfds)) {
      // 读出来放到request
      n = __libc_read(reqfd, (char *)&request, sizeof(request));
      ASSERT(n == sizeof(request));
      switch(request.req_kind) {
      // 创建线程
      case REQ_CREATE:
        request.req_thread->p_retcode =
          pthread_handle_create((pthread_t *) &request.req_thread->p_retval,
                                request.req_args.create.attr,
                                request.req_args.create.fn,
                                request.req_args.create.arg,
                                request.req_args.create.mask,
                                request.req_thread->p_pid);
        // 唤醒父线程
        restart(request.req_thread);
        break;
      case REQ_FREE:
        pthread_handle_free(request.req_args.free.thread);
        break;
      case REQ_PROCESS_EXIT:
        pthread_handle_exit(request.req_thread,
                            request.req_args.exit.code);
        break;
      case REQ_MAIN_THREAD_EXIT:
        // 标记主线程退出
        main_thread_exiting = 1;
        // 其他线程已经退出了,只有主线程了,唤醒主线程,主线程也退出,见pthread_exit,如果还有子线程没退出则主线程不能退出
        if (__pthread_main_thread->p_nextlive == __pthread_main_thread) {
          restart(__pthread_main_thread);
          return 0;
        }
        break;
      }
    }
  }
}

该函数是manager线程的主要代码。他类似一个服务器一起。接收其他线程发过来的信息,然后处理。在switch那里可以看到具体的处理。这里我们只看线程创建的逻辑。函数是pthread_handle_create。

// pthread_create发送信号给manager,manager调该函数创建线程
static int pthread_handle_create(pthread_t *thread, const pthread_attr_t *attr,
                                 void * (*start_routine)(void *), void *arg,
                                 sigset_t mask, int father_pid)
{
  int sseg;
  int pid;
  pthread_t new_thread;
  int i;

  /* Find a free stack segment for the current stack */
  sseg = 0;
  while (1) {
    while (1) {
      if (sseg >= num_stack_segments) {
        if (pthread_grow_stack_segments() == -1) return EAGAIN;
      }
      if (stack_segments[sseg] == 0) break;
      sseg++;
    }
    // 标记已使用
    stack_segments[sseg] = 1;
    // 存储线程元数据的地方
    new_thread = THREAD_SEG(sseg);
    /* Allocate space for stack and thread descriptor. */
    // 给线程分配栈
    if (mmap((caddr_t)((char *)(new_thread+1) - INITIAL_STACK_SIZE),
	     INITIAL_STACK_SIZE, PROT_READ | PROT_WRITE | PROT_EXEC,
	     MAP_PRIVATE | MAP_ANONYMOUS | MAP_FIXED | MAP_GROWSDOWN, -1, 0)
        != (caddr_t) -1) break;
    /* It seems part of this segment is already mapped. Leave it marked
       as reserved (to speed up future scans) and try the next. */
    sseg++;
  }
  /* Initialize the thread descriptor */
  new_thread->p_nextwaiting = NULL;
  new_thread->p_spinlock = 0;
  new_thread->p_signal = 0;
  new_thread->p_signal_jmp = NULL;
  new_thread->p_cancel_jmp = NULL;
  new_thread->p_terminated = 0;
  new_thread->p_detached = attr == NULL ? 0 : attr->detachstate;
  new_thread->p_exited = 0;
  new_thread->p_retval = NULL;
  new_thread->p_joining = NULL;
  new_thread->p_cleanup = NULL;
  new_thread->p_cancelstate = PTHREAD_CANCEL_ENABLE;
  new_thread->p_canceltype = PTHREAD_CANCEL_DEFERRED;
  new_thread->p_canceled = 0;
  new_thread->p_errno = 0;
  new_thread->p_h_errno = 0;
  new_thread->p_initial_fn = start_routine;
  new_thread->p_initial_fn_arg = arg;
  new_thread->p_initial_mask = mask;
  for (i = 0; i < PTHREAD_KEYS_MAX; i++) new_thread->p_specific[i] = NULL;
  /* Do the cloning */
  pid = __clone(pthread_start_thread, new_thread,
		(CLONE_VM | CLONE_FS | CLONE_FILES | CLONE_SIGHAND
		 | PTHREAD_SIG_RESTART),
		new_thread);
  /* Check if cloning succeeded */
  if (pid == -1) {
    /* Free the stack */
    munmap((caddr_t)((char *)(new_thread+1) - INITIAL_STACK_SIZE),
	   INITIAL_STACK_SIZE);
    stack_segments[sseg] = 0;
    return EAGAIN;
  }
  /* Set the priority and policy for the new thread, if available. */
  if (attr != NULL && attr->schedpolicy != SCHED_OTHER) {
    switch(attr->inheritsched) {
    case PTHREAD_EXPLICIT_SCHED:
      sched_setscheduler(pid, attr->schedpolicy, &attr->schedparam);
      break;
    case PTHREAD_INHERIT_SCHED:
      { struct sched_param father_param;
        int father_policy;
        father_policy = sched_getscheduler(father_pid);
        sched_getparam(father_pid, &father_param);
        sched_setscheduler(pid, father_policy, &father_param);
      }
      break;
    }
  }
  /* Insert new thread in doubly linked list of active threads */
  // 头插法,插入主线程和其他线程之间,
  new_thread->p_prevlive = __pthread_main_thread;
  new_thread->p_nextlive = __pthread_main_thread->p_nextlive;
  __pthread_main_thread->p_nextlive->p_prevlive = new_thread;
  __pthread_main_thread->p_nextlive = new_thread;
  /* Set pid field of the new thread, in case we get there before the
     child starts. */
  new_thread->p_pid = pid;
  /* We're all set */
  *thread = new_thread;
  return 0;
}

该函数分配一个tcb结构体表示新的线程。然后分配一个线程栈,调用clone新建一个进程。最后链接到线程链表中。最后执行pthread_start_thread函数。该函数代码如下。

// 传给clone函数的参数
static int pthread_start_thread(void *arg)
{
  // 新建的线程
  pthread_t self = (pthread_t) arg;
  void * outcome;
  /* Initialize special thread_self processing, if any.  */
#ifdef INIT_THREAD_SELF
  INIT_THREAD_SELF(self);
#endif
  /* Make sure our pid field is initialized, just in case we get there
     before our father has initialized it. */
  // 记录线程对应进程的id
  self->p_pid = getpid();
  /* Initial signal mask is that of the creating thread. (Otherwise,
     we'd just inherit the mask of the thread manager.) */
  // 设置线程的信号掩码,值继承于父线程
  sigprocmask(SIG_SETMASK, &self->p_initial_mask, NULL);
  /* Run the thread code */
  // 开始执行线程的主函数
  outcome = self->p_initial_fn(self->p_initial_fn_arg);
  /* Exit with the given return value */
  // 执行完退出
  pthread_exit(outcome);
  return 0;
}

没有太多逻辑,执行用户传进来的函数。执行完后退出。

线程库的初始化和线程的管理就分析到这,后续再深入分析。

本文分享自微信公众号 - 编程杂技(theanarkh),作者:theanarkh

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2019-10-10

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 线程的栈(基于linuxthreads-2.0.1)

    线程本质上是进程中的一个执行流,我们知道,进程有代码段,线程其实就是进程代码段中的其中一段代码。线程的一种实现是作为进程来实现的。通过调用clone,新建一个进...

    theanarkh
  • js引擎v8源码解析之平台相关(上篇)(基于v8 0.1.5)

    PlatformData 是管理线程中,不同系统中的数据。这里只看linux系统。只保存了线程id。

    theanarkh
  • linux进程通信之共享内存原理(基于linux 1.2.13)

    1 有一个全局的结构体数据,每次需要一块共享的内存时(shmget),从里面取一个结构体,记录相关的信息。

    theanarkh
  • pthread_join函数

    函数pthread_join用来等待一个线程的结束,线程间同步的操作。头文件 : #include <pthread.h> 函数定义: in...

    心跳包
  • 【前端技巧】html页面引入公共的头部和底部

    我们在做项目的时候经常会遇到一样的头部和底部,如果每个页面都复制一遍,不仅工作量大而且万一需要修改一下文章就更麻烦了,这时候就需要我们把公共部分提取出来,等需要...

    用户5997198
  • 0x02 Emacs常见报错处理

    古语有说:工欲善其事,必先利其器; Emacs无疑是编程的神器。通过这一系列的小文章,让我们一起记录熟练使用和打造这一神兵利器。

    上善若水.夏
  • 使用 Dubbo 搭建一个简单的分布式系统

    随着阿里巴巴开源的高性能分布式 RPC 框架 Dubbo 正式进入 Apache 孵化器,Dubbo 又火了一把。本文作为 Dubbo 系列开端,先教大家使用 ...

    CSDN技术头条
  • idea。git撤销merge操作。并取消

    该命令仅仅在合并后导致冲突时才使用。git merge --abort将会抛弃合并过程并且尝试重建合并前的状态。但是,当合并开始时如果存在未commit的文件,...

    DencyCheng
  • Linux 命令(109)—— ping 命令

    ping(Packet Internet Groper 命令是因特网包探索器,用于测试网络连通性,是常用的网络命令之一。

    Dabelv
  • 【原创】Java并发编程系列25 | 交换器Exchanger

    很尴尬,发了并发编程的26和27,漏了本篇25。这下子我是真的没存货了哈哈。那下面我们来补上25先,25比较短小...勿喷。

    java进阶架构师

扫码关注云+社区

领取腾讯云代金券