pth

• pth: 开启pthread https://www.iteye.com/blog/lobin-2517973

 

pth安装

pth安装选项

--enable-shared[=PKGS]

编译动态库，默认编译

 

--enable-static[=PKGS]

编译静态库，默认编译

 

--enable-debug 开启debug安装debug版本，用于开发测试。指定这个选项安装，默认情况下，如果没有显式指定--enable-shared，将只会安装静态链接库。

 

--enable-pthread 支持pthread，默认不支持。

 

--with-mctx-mth=ID

指定上下文切换方式。

安装

./configure --prefix=/usr/local/pth-2.0.7

make

make install

写道

├─bin
│ pth-config
│
├─include
│ pth.h
│
├─lib
│ libpth.a
│ libpth.la
│
├─man
│ ├─man1
│ │ pth-config.1
│ │
│ └─man3
│ pth.3
│
└─share
└─aclocal
pth.m4

 

线程

pth通过pth_spawn创建一个线程，如果创建成功，返回一个pth_t类型的新创建的线程标识。

typedef struct pth_st *pth_t;

pth_t定义为struct pth_st的指针类型。

 

struct pth_st

TCB，也就是线程控制块，其定义为：

    /* thread control block */
struct pth_st {
    /* priority queue handling */
    pth_t          q_next;               /* next thread in pool                         */
    pth_t          q_prev;               /* previous thread in pool                     */
    int            q_prio;               /* (relative) priority of thread when queued   */

    /* standard thread control block ingredients */
    int            prio;                 /* base priority of thread                     */
    char           name[PTH_TCB_NAMELEN];/* name of thread (mainly for debugging)       */
    int            dispatches;           /* total number of thread dispatches           */
    pth_state_t    state;                /* current state indicator for thread          */

    /* timing */
    pth_time_t     spawned;              /* time point at which thread was spawned      */
    pth_time_t     lastran;              /* time point at which thread was last running */
    pth_time_t     running;              /* time range the thread was already running   */

    /* event handling */
    pth_event_t    events;               /* events the tread is waiting for             */

    /* per-thread signal handling */
    sigset_t       sigpending;           /* set    of pending signals                   */
    int            sigpendcnt;           /* number of pending signals                   */

    /* machine context */
    pth_mctx_t     mctx;                 /* last saved machine state of thread          */
    char          *stack;                /* pointer to thread stack                     */
    unsigned int   stacksize;            /* size of thread stack                        */
    long          *stackguard;           /* stack overflow guard                        */
    int            stackloan;            /* stack type                                  */
    void        *(*start_func)(void *);  /* start routine                               */
    void          *start_arg;            /* start argument                              */

    /* thread joining */
    int            joinable;             /* whether thread is joinable                  */
    void          *join_arg;             /* joining argument                            */

    /* per-thread specific storage */
    const void   **data_value;           /* thread specific  values                     */
    int            data_count;           /* number of stored values                     */

    /* cancellation support */
    int            cancelreq;            /* cancellation request is pending             */
    unsigned int   cancelstate;          /* cancellation state of thread                */
    pth_cleanup_t *cleanups;             /* stack of thread cleanup handlers            */

    /* mutex ring */
    pth_ring_t     mutexring;            /* ring of aquired mutex structures            */

#ifdef PTH_EX
    /* per-thread exception handling */
    ex_ctx_t       ex_ctx;               /* exception handling context                  */
#endif
};

线程属性

 

struct pth_attr_st {
    pth_t        a_tid;
    int          a_prio;
    int          a_dispatches;
    char         a_name[PTH_TCB_NAMELEN];
    int          a_joinable;
    unsigned int a_cancelstate;
    unsigned int a_stacksize;
    char        *a_stackaddr;
};

 

 

 

线程调度

优先级调度

 

优先级

分为3个优先级。标准优先级，最大优先级以及最小优先级。

    /* thread priority values */
#define PTH_PRIO_MAX                 +5
#define PTH_PRIO_STD                  0
#define PTH_PRIO_MIN                 -5

优先级队列

pth根据线程的状态各自创建一个优先级队列。

intern pth_pqueue_t pth_NQ;         /* queue of new threads                  */
intern pth_pqueue_t pth_RQ;         /* queue of threads ready to run         */
intern pth_pqueue_t pth_WQ;         /* queue of threads waiting for an event */
intern pth_pqueue_t pth_SQ;         /* queue of suspended threads            */
intern pth_pqueue_t pth_DQ;         /* queue of terminated threads           */

pth_NQ

新创建的线程进入此队列。

pth_RQ

线程就绪进入此队列

pth_WQ

线程等待事件进入此队列

pth_SQ

线程挂起进入此队列

pth_DQ

线程终止进入此队列。

 

优先级队列实现

在实现优先级队列的时候，我们可能会通过如下方式实现。 

写道

优先级队列
https://www.iteye.com/blog/lobin-2325712

 

pth通过双链表实现优先级队列。在这个双链表中，队头的优先级越高，往后优先级越低，队尾的优先级最低。

/* thread priority queue */
struct pth_pqueue_st {
    pth_t q_head;
    int   q_num;
};
typedef struct pth_pqueue_st pth_pqueue_t;

结构体struct pth_pqueue_st只定义了队头及队列中节点的数目。主要结构在struct pth_st中定义，包括优先级等。

 

调度器

 

 

上下文切换

上下文切换方式

支持两种上下文切换方式：mcsc和sjlj。mcsc是通过ucontext方式来实现上下文切换，这是linux下协程的实现方式。sjlj是通过pth_siglongjmp（siglongjmp/longjmp/_longjmp）的方式来实现上下文切换。

 

上下文切换方式在编译的时候确定，其实是在编译前配置的时候确定，在./configure的时候指定--with-mctx-mth，可以指定这两种切换方式。

写道

--with-mctx-mth=ID force mctx method (mcsc,sjlj)

 

在pth_acdef.h.in配置头文件中，有一个PTH_MCTX_MTH_use的定义。指定./configure之后，会根据这个配置头文件生成头文件pth_acdef.h。

/* define for machine context method */
#undef PTH_MCTX_MTH_use

在./configure的时候指定--with-mctx-mth，可以指定--with-mctx-mth=mcsc或者--with-mctx-mth=sjlj，默认不指定的话就是mcsc，如果采用的是mcsc，配置之后的定义如下：

/* define for machine context method */
#define PTH_MCTX_MTH_use PTH_MCTX_MTH_mcsc

 

初始化

pth在使用之前需要init

pth_init();

使用完时候需要kill

pth_kill();

 

int pth_init(void)
{
    pth_attr_t t_attr;

    /* support for implicit initialization calls
       and to prevent multiple explict initialization, too */
    if (pth_initialized)
        return pth_error(FALSE, EPERM);
    else
        pth_initialized = TRUE;

    pth_debug1("pth_init: enter");

    /* initialize syscall wrapping */
    pth_syscall_init();

    /* initialize the scheduler */
    if (!pth_scheduler_init()) {
        pth_shield { pth_syscall_kill(); }
        return pth_error(FALSE, EAGAIN);
    }

#ifdef PTH_EX
    /* optional support for exceptional handling */
    __ex_ctx       = pth_ex_ctx;
    __ex_terminate = pth_ex_terminate;
#endif

    /* spawn the scheduler thread */
    t_attr = pth_attr_new();
    pth_attr_set(t_attr, PTH_ATTR_PRIO,         PTH_PRIO_MAX);
    pth_attr_set(t_attr, PTH_ATTR_NAME,         "**SCHEDULER**");
    pth_attr_set(t_attr, PTH_ATTR_JOINABLE,     FALSE);
    pth_attr_set(t_attr, PTH_ATTR_CANCEL_STATE, PTH_CANCEL_DISABLE);
    pth_attr_set(t_attr, PTH_ATTR_STACK_SIZE,   64*1024);
    pth_attr_set(t_attr, PTH_ATTR_STACK_ADDR,   NULL);
    pth_sched = pth_spawn(t_attr, pth_scheduler, NULL);
    if (pth_sched == NULL) {
        pth_shield {
            pth_attr_destroy(t_attr);
            pth_scheduler_kill();
            pth_syscall_kill();
        }
        return FALSE;
    }

    /* spawn a thread for the main program */
    pth_attr_set(t_attr, PTH_ATTR_PRIO,         PTH_PRIO_STD);
    pth_attr_set(t_attr, PTH_ATTR_NAME,         "main");
    pth_attr_set(t_attr, PTH_ATTR_JOINABLE,     TRUE);
    pth_attr_set(t_attr, PTH_ATTR_CANCEL_STATE, PTH_CANCEL_ENABLE|PTH_CANCEL_DEFERRED);
    pth_attr_set(t_attr, PTH_ATTR_STACK_SIZE,   0 /* special */);
    pth_attr_set(t_attr, PTH_ATTR_STACK_ADDR,   NULL);
    pth_main = pth_spawn(t_attr, (void *(*)(void *))(-1), NULL);
    if (pth_main == NULL) {
        pth_shield {
            pth_attr_destroy(t_attr);
            pth_scheduler_kill();
            pth_syscall_kill();
        }
        return FALSE;
    }
    pth_attr_destroy(t_attr);

    /*
     * The first time we've to manually switch into the scheduler to start
     * threading. Because at this time the only non-scheduler thread is the
     * "main thread" we will come back immediately. We've to also initialize
     * the pth_current variable here to allow the pth_spawn_trampoline
     * function to find the scheduler.
     */
    pth_current = pth_sched;
    pth_mctx_switch(&pth_main->mctx, &pth_sched->mctx);

    /* came back, so let's go home... */
    pth_debug1("pth_init: leave");
    return TRUE;
}

 

 

创建线程

 

pth_t pth_spawn(pth_attr_t attr, void *(*func)(void *), void *arg)
{
    pth_t t;
    unsigned int stacksize;
    void *stackaddr;
    pth_time_t ts;

    pth_debug1("pth_spawn: enter");

    /* consistency */
    if (func == NULL)
        return pth_error((pth_t)NULL, EINVAL);

    /* support the special case of main() */
    if (func == (void *(*)(void *))(-1))
        func = NULL;

    /* allocate a new thread control block */
    stacksize = (attr == PTH_ATTR_DEFAULT ? 64*1024 : attr->a_stacksize);
    stackaddr = (attr == PTH_ATTR_DEFAULT ? NULL    : attr->a_stackaddr);
    if ((t = pth_tcb_alloc(stacksize, stackaddr)) == NULL)
        return pth_error((pth_t)NULL, errno);

    /* configure remaining attributes */
    if (attr != PTH_ATTR_DEFAULT) {
        /* overtake fields from the attribute structure */
        t->prio        = attr->a_prio;
        t->joinable    = attr->a_joinable;
        t->cancelstate = attr->a_cancelstate;
        t->dispatches  = attr->a_dispatches;
        pth_util_cpystrn(t->name, attr->a_name, PTH_TCB_NAMELEN);
    }
    else if (pth_current != NULL) {
        /* overtake some fields from the parent thread */
        t->prio        = pth_current->prio;
        t->joinable    = pth_current->joinable;
        t->cancelstate = pth_current->cancelstate;
        t->dispatches  = 0;
        pth_snprintf(t->name, PTH_TCB_NAMELEN, "%s.child@%d=0x%lx",
                     pth_current->name, (unsigned int)time(NULL),
                     (unsigned long)pth_current);
    }
    else {
        /* defaults */
        t->prio        = PTH_PRIO_STD;
        t->joinable    = TRUE;
        t->cancelstate = PTH_CANCEL_DEFAULT;
        t->dispatches  = 0;
        pth_snprintf(t->name, PTH_TCB_NAMELEN,
                     "user/%x", (unsigned int)time(NULL));
    }

    /* initialize the time points and ranges */
    pth_time_set(&ts, PTH_TIME_NOW);
    pth_time_set(&t->spawned, &ts);
    pth_time_set(&t->lastran, &ts);
    pth_time_set(&t->running, PTH_TIME_ZERO);

    /* initialize events */
    t->events = NULL;

    /* clear raised signals */
    sigemptyset(&t->sigpending);
    t->sigpendcnt = 0;

    /* remember the start routine and arguments for our trampoline */
    t->start_func = func;
    t->start_arg  = arg;

    /* initialize join argument */
    t->join_arg = NULL;

    /* initialize thread specific storage */
    t->data_value = NULL;
    t->data_count = 0;

    /* initialize cancellation stuff */
    t->cancelreq   = FALSE;
    t->cleanups    = NULL;

    /* initialize mutex stuff */
    pth_ring_init(&t->mutexring);

#ifdef PTH_EX
    /* initialize exception handling context */
    EX_CTX_INITIALIZE(&t->ex_ctx);
#endif

    /* initialize the machine context of this new thread */
    if (t->stacksize > 0) { /* the "main thread" (indicated by == 0) is special! */
        if (!pth_mctx_set(&t->mctx, pth_spawn_trampoline,
                          t->stack, ((char *)t->stack+t->stacksize))) {
            pth_shield { pth_tcb_free(t); }
            return pth_error((pth_t)NULL, errno);
        }
    }

    /* finally insert it into the "new queue" where
       the scheduler will pick it up for dispatching */
    if (func != pth_scheduler) {
        t->state = PTH_STATE_NEW;
        pth_pqueue_insert(&pth_NQ, t->prio, t);
    }

    pth_debug1("pth_spawn: leave");

    /* the returned thread id is just the pointer
       to the thread control block... */
    return t;
}

创建线程的时候需要传入线程属性，线程执行的函数例程，以及传给这个函数例程的参数。其中函数例程原型定义为void *(*func)(void *)，和pthread一样，这个必须指定。

 

如果不想自己指定线程属性，可以指定默认线程属性PTH_ATTR_DEFAULT

 

#define PTH_ATTR_DEFAULT (pth_attr_t)(0)

 

关于这个函数例程，有个特殊的情况是，传入-1的时候，表示主线程，这种情况会将函数例程设置为NULL。

 

接着为新创建的线程分配一个TCB，即线程控制块，同时初始化TCB，在初始化TCB的时候：

 

主要初始化线程名，线程栈大小，栈空间，栈保护空间，优先级，是否joinable，是否可取消，

 

如果没有指定线程名，如果pth_current（指向当前线程的TCB）为NULL，线程名为user/%x，其中%x为当前时间，如果pth_current（指向当前线程的TCB）不为NULL，线程名为%s.child@%d=0x%lx，其中%s为当前线程名，%d为当前时间，%lx为当前线程的TCB地址。

 

线程栈大小默认为64k，如果指定的线程栈大小大于0小于SIGSTKSZ，会将线程栈大小设置为SIGSTKSZ。

这里有个特殊情况，如果是主线程的话，在为主线程分配TCB的时候，指定的线程栈大小为0。因为不需要给主线程分配栈空间。同时在为主线程分配TCB的时候，线程栈地址指定为NULL。

 

线程栈空间如果没有指定的话，会分配一块空间给这个新创建的线程。

 

初始化后还会在栈的高地址或低地址写入0xDEAD（4个字节），DEAD？不清楚什么意图。

 

如果没有指定线程属性，将从pth_current（指向当前线程的TCB）继承线程优先级，是否joinable，是否可取消状态。

 

还会设置线程执行的函数例程以及传入的参数。

 

初始化的时候不会为线程分配TLS。

 

新创建的线程状态为PTH_STATE_NEW。

 

新线程创建后将会放入pth_NQ优先级队列中。

 

 

 

创建线程

 

#include <stdio.h>
#include <pth.h>

void* start_routine(void *arg) 
{
  char* name = (char *) arg;
  printf("thread %s called.\n", name);
  return NULL;
}

int main()
{
  pth_t pth;
  void *pth_exit;

  pth_init();
  pth = pth_spawn(PTH_ATTR_DEFAULT, start_routine, (void *) "#1");

  pth_join(pth, &pth_exit);
  pth_kill();
  return 0;
}

 

 

指定线程属性

 

#include <stdio.h>
#include <pth.h>

void* start_routine(void *arg) 
{
  char* name = (char *) arg;
  printf("thread %s called.\n", name);
  return NULL;
}

int main()
{
  pth_attr_t pth_attr;
  pth_t pth;
  void *pth_exit;

  if (! pth_init())
  {
    printf("pth init err\n");
    return 1;
  }
  pth_attr = pth_attr_new();

  pth = pth_spawn(pth_attr, start_routine, (void *) "#1");

  pth_join(pth, &pth_exit);
  if (! pth_kill())
  {
    printf("pth kill err\n");
    return 1;
  }
  return 0;
}

 

创建多个线程

#include <stdio.h>
#include <unistd.h>
#include <pth.h>

void* start_routine(void *arg) 
{
  unsigned i = 0;
  char* name = (char *) arg;
  printf("thread %s called.\n", name);
  while (++i && i <= 10)
  {
    sleep(1); // sleep 1 seconds.
    printf("%s: %d\n", name, i);
  }
  return NULL;
}

int main()
{
  pth_t pth1, pth2;
  void *pth_exit;
  if (! pth_init())
  {
    printf("pth init err\n");
    return 1;
  }
  pth1 = pth_spawn(PTH_ATTR_DEFAULT, start_routine, (void *) "#1");
  pth2 = pth_spawn(PTH_ATTR_DEFAULT, start_routine, (void *) "#2");

  pth_join(pth1, &pth_exit);
  pth_join(pth2, &pth_exit);
  if (! pth_kill())
  {
    printf("pth kill err\n");
    return 1;
  }
  return 0;
}

编译

# gcc -I /usr/local/pth-2.0.7/include/ -L /usr/local/pth-2.0.7/lib pth_multi_test2.c -o pth_multi_test2 -lpth

# ldd pth_multi_test2

linux-gate.so.1 =>  (0x00cdd000)

libpth.so.20 => /usr/lib/libpth.so.20 (0x00c35000)

libc.so.6 => /lib/libc.so.6 (0x00356000)

/lib/ld-linux.so.2 (0x800ec000)

运行

# ./pth_multi_test2 

thread #1 called.

#1: 1

#1: 2

#1: 3

#1: 4

#1: 5

#1: 6

#1: 7

#1: 8

#1: 9

#1: 10

thread #2 called.

#2: 1

#2: 2

#2: 3

#2: 4

#2: 5

#2: 6

#2: 7

#2: 8

#2: 9

#2: 10

从运行结果看线程1和2没有一点交替运行。结果感觉不正常，多次运行结果都是这样。