rn_xtcxyczjh-5 并发[线程1 线程执行模式-汇编指令层]

2015.10.02

读xtcxyczjh（系统程序员成长计划）—- 学习程序设计方法。

需求简述。

了解linux下的多线程编程的基本方法，以双向链表为载体实现传统的生产者-消费者模型（《CSAPP》2E 670页）：一个线程往双向链表中加东西，另外一个线程从这里面取。

准备。

阅读《xtcxyczjh》38 – 42页；阅读《CSAPP》2E 657-660页；阅读《CSAPP》2E 660-673页；正确理解POSIX thread函数库中所使用到的每个函数的功能。

[验证多线程执行模式的程序的保存地址：y15m10d02 ]

1. 验证（理解多线程）

Linux 下的多线程编程使用pthread(POSIX Thread)函数库，使用时包含头文件pthread.h，链接共享库libpthread.so（gcc … -lpthread）。

从书中变种一个例子过来。在linux终端新建cpthread.c/cpthread.h（cpthread–call posix thread）文件。

cpthread.c

/* cpthread.c */
/* 调用POSIX thread库中的函数定义线程线程相关函数 */
#include <pthread.h>
#include <stdio.h>

void *start_routine(void *param);

/* --------函数定义区域-------- */
//在主线程中创建线程
pthread_t create_test_thread(void)
{
pthread_t       tid = 0;
unsigned int    idx = 0;
unsigned int    imx = 10000;

pthread_create(&tid, NULL, start_routine, NULL);
for (idx = 0; idx < imx; ++idx)
fprintf(stdout, "main thread:%d\n", idx);
return tid;
}

//子线程程序（函数）
void *start_routine(void *param)
{
unsigned int    idx = 0;
unsigned int    imx = 10;

for (idx = 0; idx < imx; ++idx)
fprintf(stdout, "child thread:%d\n", idx);
return NULL;
}

在linux终端使用“man pthread_create”命令查看pthread_create函数的手册。

cpthread.h

/* cpthread.h */
/* 定义线程相关，或声明定义在cpthread.c中的调用POSIX thread库的函数 */

/* --------函数声明区域-------- */
pthread_t create_test_thread(void);

main.c

/* main.c */
/* 包含C程序入口 */
#include <stdio.h>
#include <pthread.h>
#include "cpthread.h"

//#include "dlist.h"
//#include "tk.h"
//#include "autotest.h"

int main(void)
{
void *ret = NULL;
pthread_t tid = create_test_thread();
pthread_join(tid, &ret);

return 0;
}

在linux终端使用“man pthread_join”命令查看pthread_join函数的手册。

Makefile

#Makefile
#make命令默认执行make all或者第一条规则
all:        main.o dlist.o tk.o autotest.o cpthread.o
$(CC) $^ $(CFLAGS) $@ $(CFLAG_PTHREAD_LIB)

#......
cpthread.o: cpthread.c cpthread.h
$(CC) $(CFLAG_OBJ) $< $(CFLAGS_POSTFIX)

clean:
-rm *.o

#定义变量
#......
CFLAGS_POSTFIX=-Wall
CFLAG_PTHREAD_LIB=-lpthread

[这是测试（熟悉）pthread库函数的程序，到时候还要利用双向链表接口为载体来学习并发，所以不必将Makefile中关于其它文件的编译命令删除或者屏蔽。]

在与这些程序同一级目录下使用make命令，再运行all程序，两个线程的输出信息被输出到linux终端。由于终端能显示的信息过少，不能清晰的看出线程执行的“并发”性，可以将线程的输出信息输出到文本中观察。新建cfile.c/cfile.h文件。

cfile.c

/* cfile.c */
/* 定义与文件操作相关的函数 */
#include <stdio.h>
#include "tk.h"

/* --------函数定义区域-------- */
//打开或创建一个文件
FILE *open_file(const char *fname)
{
FILE    *fp = NULL;

return_val_if_p_invalid(fname, NULL);
if (NULL == (fp = fopen(fname, "w+")))
cstd_error("fopen");
return fp;
}

//关闭有fopen打开的文件
void close_file(FILE *fp)
{
return_if_p_invalid(fp);
fclose(fp);
}

cfile.h

/* cfile.h */
/* 定义文件操作相关或声明cfile.c中定义的函数 */

/* --------宏定义区域------- */
#define PTHREAD_TEST_FILENAME   "pthread_out.txt"

/* --------函数定义区域-------- */
FILE *open_file(const char *fname);
void close_file(FILE *fp);

tk.c

/* tk.c */
/* 定义工具函数以及其它函数的错误封包函数 */
//......
#include <string.h>
#include <stdio.h>
#include <errno.h>

//……

//输出发生错误时将错误设置在errno内的错误并终止当前进程
void cstd_error(char *msg)
{
fprintf(stderr, "%s:%s\n", "fopen", strerror(errno));
exit(-1);
}

tk.h

/* tk.h */
//......
/* --------函数声明区-------- */
char  lchar2uchar(unsigned char ch);
void cstd_error(char *msg);

cpthread.c

/* cpthread.c */
/* 调用POSIX thread库中的函数定义线程线程相关函数 */
#include <pthread.h>
#include <stdio.h>

void *start_routine(void *param);

/* --------函数定义区域-------- */
//在主线程中创建线程
pthread_t create_test_thread(FILE *fp)
{
pthread_t       tid = 0;
unsigned int    idx = 0;
unsigned int    imx = 1000000;

pthread_create(&tid, NULL, start_routine, fp);
for (idx = 0; idx < imx; ++idx)
fprintf(fp, "main thread:%d\n", idx);
return tid;
}

//子线程程序（函数）
void *start_routine(void *param)
{
unsigned int    idx = 0;
unsigned int    imx = 1000000;

for (idx = 0; idx < imx; ++idx)
fprintf(param, "child thread:%d\n", idx);
return NULL;
}

cpthread.h

/* cpthread.h */
//……
/* --------函数声明区域-------- */
pthread_t create_test_thread(FILE *fp);

main.c

/* main.c */
/* 包含C程序入口 */
//......
int main(void)
{
void    *ret = NULL;
FILE    *fp = NULL;

fp  = open_file(PTHREAD_TEST_FILENAME);
pthread_t tid = create_test_thread(fp);
pthread_join(tid, &ret);
close_file(fp);

return 0;
}

所有运行在一个进程里的线程共享该进程的整个虚拟地址空间，多个线程运行在单一进程的上下文中，因此共享这个进程虚拟地址空间的整个内容，包括它的代码、数据、堆、共享库和打开的文件。

Makefile

#Makefile
#make命令默认执行make all或者第一条规则
all:        main.o dlist.o tk.o autotest.o cpthread.o cfile.o
$(CC) $^ $(CFLAGS) $@ $(CFLAG_PTHREAD_LIB)
#......
cfile.o:    cfile.c
$(CC) $(CFLAG_OBJ) $< $(CFLAGS_POSTFIX)
clean:
-rm *.o

#......

在linux终端与程序所在的同一级目录下使用make命令编译程序，运行all程序得到pthread_out.txt文件。打开pthread_out.txt，可看到：



……



……



……

线程执行模型-I。

每个进程开始生命周期时都是单一线程，这个线程称为主线程。在主线程中使用pthread_create函数创建一个对等线程，从这个时间点开始，两个线程并发地运行（直到主线程的pthread_join语句处，pthread_join一直等待它指定的线程运行完毕后返回）。和一个进程相关的线程组成一个对等（线程）池，独立于其他线程创建的线程。主线程和其他线程的区别仅在于它总是进程中第一个运行的线程。主线程创建线程后，因为主线程执行一个慢速系统调用（如read或sleep），或者因为它被系统的间隔计时器中断，控制就会通过上下文切换传递到对等线程。对等线程执行一段时间，然后控制传回主线程，依次类推。

对等线程。

对等（线程）池概念的主要影响是，一个线程可以杀死它的任何对等线程，或者等待它的对等线程终止。

2. 主线程和对等线程随机访问双向链表的程序

修改create_test_thread和start_routine程序，实现主线程往双线链表中添加东西，子线程将主线程添加的东西读出来。

cpthread.c

/* cpthread.c */
/* 调用POSIX thread库中的函数定义线程线程相关函数 */
//……
#include "dlist.h"
#include "tk.h"
#include <assert.h>
#include <stdlib.h>

void *start_routine(void *param);

/* --------函数定义区域-------- */
//在主线程中创建线程
int create_test_thread(DlistManageT *pdl)
{
int             rv = 0;
char            *p = NULL;
void            *ret = NULL;
pthread_t       tid = 0;
unsigned int    idx = 0;
unsigned int    len = 0;

return_val_if_p_invalid(pdl, -1);

//创建子线程成功后，主线程与子线程start_routine并发执行
rv  = pthread_create(&tid, NULL, start_routine, pdl);
if (0 != rv) {
fprintf(stderr, "%s\n", "create thread failed");
return -1;
}
len = get_dlist_length(pdl);
p   = alloc_nbyte_mem(len);
if (NULL == p) {
fprintf(stderr, "%s\n", "allocate mm failed");
return -1;
}
for (idx = 1; idx <= len; ++idx)
assert(-1 != (rv = init_dlist_any_node_data(pdl, idx, &p[idx - 1])));

pthread_join(tid, &ret);
free(p);    //主/子线程均结束后方才释放堆内存与双向链表
free_dlist(pdl);

return 0;
}

//子线程程序（函数）
void *start_routine(void *param)
{
int             i = 0;
char            *data = NULL;
unsigned int    len = 0;
DlistManageT    *pdl = NULL;

return_val_if_p_invalid(param, NULL);
pdl = (DlistManageT *)param;
len = get_dlist_length(pdl);

for (i = 1; i <= len; ++i) {
data    = get_dlist_node_data(pdl, i);
if (NULL != data)
fprintf(stdout, "child thread:%d\n", *((char *)data));
}

return NULL;
}

cpthread.h

/* cpthread.h */
/* 定义线程相关，或声明定义在cpthread.c中的调用POSIX thread库的函数 */
#include "dlist.h"

/* --------函数声明区域-------- */
pthread_t create_test_thread(DlistManageT *pdl);

tk.c

//分配n字节内存，第i字节内存用i初始化
char *alloc_nbyte_mem(unsigned int n)
{
int     i = 0;
char    *p = NULL;

p   = (char *)malloc(n);
if (NULL != p) {
for (i = 0; i < n; ++i)
p[i]    = i;
}

return p;
}

tk.h

/* tk.h */
/* 定义宏以及声明tk.c中的内容 */

//......

/* --------函数声明区-------- */
//……
char *alloc_nbyte_mem(unsigned int n);

对于以上的create_test_thread()函数，双向链表中的数据可以直接是create_test_thread()函数内的局部数据，因为在对等线程start_routine访问双向链表期间，create_test_thread()函数在用“ pthread_join(tid, &ret);”语句等待对等线程的结束，局部变量不会被释放。将双向链表的数据准备在堆中，就更不用担心主线程等待对等线程结束的语句在何处了。

main.c

/* main.c */
/* 包含C程序入口 */
#include <stdio.h>
#include <pthread.h>
#include "cpthread.h"

#include "dlist.h"
#include "tk.h"
//#include "autotest.h"

#include <assert.h>

int main(void)
{
DlistManageT    *pdl = NULL;
unsigned int    dlen = 10;

assert(NULL != (pdl = create_dlist(dlen)));
create_test_thread(pdl);

return 0;
}

main()函数调用的create_test_thread()函数包含了“创建子线程”、“主线程程序内容”、“等待子线程运行完毕”、”释放资源“工作。

在与程序同目录的linux终端用make命令编译各程序得到可执行程序all并运行几次得到以下结果。



多线程程序执行模型-II：主线程mian()先运行，然后让对等线程运行。主线程的create_test_pthread函数中的pthread_join语句会等待对等线程执行完毕后再返回。所以，主线程跟对等线程是交替执行的，不能保证主线程初始化完双向链表后对等线程才开始访问双向链表。如果遇到未初始化的双向链表元素，对等线程将不会输出这个双向链表元素的内容。而线程的运行并非在C语言层面就能将其解释清楚，线程的交替运行至少是以汇编指令为单位的（《CSAPP》 2E 664-666页）。

对于各线程共享的数据，要保证两点：各线程对共享数据的互斥访问；对共享数据访问的顺序（调度各线程对共享数据的访问顺序）[《CSAPP》 2E 664-670页]。即要用加锁的方式包装对数据访问额的串行化。线程加锁的方式有很多种，像互斥锁(mutex= mutual exclusive lock)，信号量(semaphore)和自旋锁(spin lock)等都是常用的。[在线程中的加锁/解锁要成对出现即不要存在遗漏解锁的路径（尤其是在返回值较多的函数中）；多锁情况下注意加锁顺序以避免死锁]

2015.10.15

3. 生产者-消费者问题的程序-I

(1) 让各线程互斥访问双向链表元素

在这个程序中，最多只能有一个线程在访问临界区。通过信号量来实现这个目标。

tk.c

/* tk.c */
/* 定义工具函数以及其它函数的错误封包函数 */
//……
#include <semaphore.h>

//……
//对sem_init函数进行错误包装
void Wsem_init(sem_t *sem, int pshared, unsigned int value)
{
if (sem_init(sem, pshared, value) < 0)
cstd_error("sem_init error");
}

//对sem_wait函数进行错误包装
void P(sem_t *s)
{
if (sem_wait(s) < 0)
cstd_error("sem_wait error");
}

//对sem_post函数进行错误包装
void V(sem_t *s)
{
if (sem_post(s) < 0)
cstd_error("sem_post error");
}

tk.h

/* tk.h */
/* 定义宏以及声明tk.c中的内容 */

#include <semaphore.h>
//……
void Wsem_init(sem_t *sem, int pshared, unsigned int value);
void P(sem_t *s);
void V(sem_t *s);

cpthread.c

/* cpthread.c */
/* 调用POSIX thread库中的函数定义线程线程相关函数 */

//……
/******** 全局变量定义区域 ******** */
sem_t           mutex;

//在主线程中创建线程
int create_test_thread(DlistManageT *pdl)
{
//……
return_val_if_p_invalid(pdl, -1);
Wsem_init(&mutex, 0, 1);

//……
for (idx = 1; idx <= len; ++idx) {
P(&mutex);
assert(-1 != (rv = init_dlist_any_node_data(pdl, idx, &p[idx - 1])));
V(&mutex);
}
//……
}

//子线程程序（函数）
void *start_routine(void *param)
{
//……
for (i = 1; i <= len; ++i) {
P(&mutex);
data    = get_dlist_node_data(pdl, i);
V(&mutex);
if (NULL != data)
fprintf(stdout, "child thread:%d\n", *((char *)data));
}
//......
}

从调用dlist.c中的函数的角度（即在create_test_thread和start_routine函数中）来看，对于双向链表中的同一个元素，只有一个线程在访问（这里指init_dlist_any_node_data或get_dlist_node_data函数执行完毕，含有一些访问双向链表元素临界区之外的一些操作，信号量的这种用法不是最佳的）。而且，对于双向链表中的同一个元素，主线程和start_routine对其访问的顺序是不定的（虽然运行结果是主线程先）。而我们的目的是先初始化双向链表（需要保证“写者优先”以及“临界区最多只有一个写者”），再将初始化内容读出来。这就还需要办法来控制各线程对共享数据的访问顺序。

(2) 控制各线程访问双向链表元素的顺序

首先可以用信号量保证用来初始化链表的数据的程序先运行。

cpthread.c

//在主线程中创建线程
int create_test_thread(DlistManageT *pdl)
{
//……
P(&mutex);
len = get_dlist_length(pdl);
p   = alloc_nbyte_mem(len);
V(&mutex);
//......
}

接下来需要保证读一个双向链表元素在写同一个元素之后（如用一个变量记录双向链表当前被写的位置，在读线程中，如果读位置小于被写位置则可读，否则读线程等待）。将形如记录双向链表被写位置等变量设置在描述双向链表的数据结构中似乎会减轻调用者编写程序的负担，那就直接进入将线程、线程相关的控制变量都揉到描述双向链表的数据结构中，即以双向链表为载体解决生产者-消费者问题。

[将多线程及其相关的东西包含描述进双向链表的数据结构的程序的保存地址：y15m10d15 ]

4. 生产者-消费者模式的程序-II

3中实现生产者-消费者模式的程序是站在用户的调度来编写的，此时站在提供接口的角度来实现生产者-消费者模型。

dlist.c

/* dlist.c */
/* 定义描述/管理（循环）双向链表的数据，定义（循环）双向链表的接口给 */
//……
#include <semaphore.h>

//……

//管理双向链表的结构体
struct _DlistManage {
DlistNodeT      *first; //指向双向链表中的第一个节点
unsigned int    num;    //保存当前双向链表的节点个数
sem_t           mutex;  //用于实现各线程之间的互斥访问共享数据
unsigned int    wl;     //用于记录已经初始化完毕的双向链表中的节点位置
};

//……

//创建含n个节点的双向链表
DlistManageT *create_dlist(unsigned int len)
{
//......
pdl->first->prev    = pndc;
pndc->next          = pdl->first;

//初始化双向链表中线程相关的元素
Wsem_init(&pdl->mutex, 0, 1);
pdl->wl= 0;

return pdl;
}

//给双向链表中第i个节点中指向数据（即data）的指针赋值
int init_dlist_any_node_data(DlistManageT *pdl, unsigned int i, void *data)
{
//……
if (NULL != (pnd = get_dlist_any_node(pdl, i))) {
P(&pdl->mutex);
pnd->data   = data;
pdl->wl++;
V(&pdl->mutex);
return 0;
}

return -1;
}

//获取双向链表中第i个节点的值
void *get_dlist_node_data(DlistManageT *pdl, unsigned int i)
{
//……
wl  = pdl->wl;
while(i > wl) { //为生产者-消费者模型提供的while代码
;
}

pnd = pdl->first;
for (k = 1; k < i; ++k)
pnd = pnd->next;
p(&pdl->mutex);
data    = pnd->data;
V(&pdl->mutex);
return data;
}

在程序中以汇编指令（微操作）为单位为程序加信号量，尽量避免破坏多线程的并发性。

将cpthread.c中涉及到的线程相关的程序删掉。在与程序和Makefile同目录下的linux终端使用make命令编译程序后可运行all程序。

读《xtcxyczjh》-Part-V pnote over.

[2015.10.15-14:03]