pthread库是linux系统的多线程库。上次写完longjmp和多线程的例子。就想用系统的真线程库改写一下读写线程的例子来和大家分享。真线程库应该比自己写的功能更完善,代码也更容易写吧!… …
说到这里。我不知道接下来说什么了… …。也许是自己对多线程库掌握不够熟练吧!代码早就改好了。但是运行多数时候是好的,有的时候却会crash。我确信这是同步的问题。查了好多天。除了费解还是费解。都怀疑到天气热,我的机器内存数据发生突变了。直到刚才改了一处,才稳定了。目前没发现再crash。这就把代码分享上来。这次被这bug整的代码已经有些过度保守了,不像是我自己的风格,但仍不敢保证没有错误了。万一谁在调试中发现新的错误。淡定,在这基础上改。
现在的数据结构变成这样:
typedef struct _Context_ {pthread_mutex_t sched;volatile int wait_thread_lock_sched;int finish;struct {pthread_mutex_t mutex;volatile int wait_sched_lock_thread;int pend;int state;} thread[2];
} Context;#define WRITE_THREAD 0
#define READ_THREAD 1
这里用mutex做同步。mutex适合做互斥。在同一线程中上锁,解锁,互斥保护共享资源访问。而理论上是可以做同步的,做同步需要在不同线程中上锁,解锁,实际用下来感觉不理想。
使用pthread线程库时,要注意先测试一下。这一步不能省。我的版本,文档说明说的是没有attr说明的mutex默认创建为普通锁。这可不一定。测试下来发现给的是递归锁。区别在于锁定一个已经锁定mutex,如果上次是自己锁的,普通锁会休眠,而递归锁会成功。
普通用法,在同一线程中对资源上锁,解锁,操作时串行的,不会发生自己两次锁同一个资源问题,这时两者的作用是相同的,除非发生了操作中用了递归函数。可能是因为这个原因,我的系统中默认给了递归锁。
但是同步的用法,上锁和解锁在2个不同的线程中,系统调度是随机的。有可能上锁的线程被调度了2次,而解锁一次也没调度。这样,如果递归锁,它每次上锁都是成功的,它不停下了,它还要继续上锁。这就发生了复杂的问题了。
回到这里的代码,一个调度和2个线程。mutex设置的是普通锁。注意系统的调度仍然在跑,这里意思是自己写的同步的调度,以下简称调度,意思都是自己做的调度。如果线程自己的mutex被别的线程上锁,那么线程运行到这个位置的时候,再次上锁就会休眠,而别的线程解锁,也就唤醒了这个休眠的线程。这样就有了同步的效果。注意休眠线醒来之后已经获得了mutex上锁,它可能需要立即解锁,以便别的线程继续控制锁。
初始化:
pthread_mutexattr_t mat;pthread_mutexattr_init(&mat);pthread_mutexattr_settype(&mat, PTHREAD_MUTEX_NORMAL);pthread_mutex_init(&bx_mutex, &mat);pthread_mutex_init(&gCtx.sched, &mat);pthread_mutex_init(&gCtx.thread[0].mutex, &mat);pthread_mutex_init(&gCtx.thread[1].mutex, &mat);mutex_lock(&gCtx.thread[0].mutex);mutex_lock(&gCtx.thread[1].mutex);start_coroutine((pf) coroutine_write, NULL);start_coroutine((pf) coroutine_read, NULL);
start_coroutine之后,读写线程有可能已经跑起来了,有可能还没跑,这不一定。之前调度程序已经给它们mutex上了锁,跑起来之后,首先是去对自己的mutex上锁,这样可以进入休眠,让调度程序有序唤醒它们进行操作。
void coroutine_write(void *arg)
{FILE *fp;char s[128];int len;int i = (int) arg;i = WRITE_THREAD;gCtx.thread[i].wait_sched_lock_thread = 999;mutex_lock(&gCtx.thread[i].mutex);mutex_unlock(&gCtx.thread[i].mutex);lock_scheduler();while (gCtx.thread[i].wait_sched_lock_thread != 0) {sched_yield();}fp = fopen("b.c", "r");if (!fp) {printf("can not open `b.c'\n");exit(0);}mutex_lock(&bx_mutex);while (fgets(s, 128, fp) != NULL) {len = strlen(s);if (len) {--len;if (s[len] == '\n') s[len] = '\0'; else ++len;}if (len) {--len;if (s[len] == '\r') s[len] = '\0'; else ++len;}bx_write(s, len + 1);}bx.eobf = 1;gCtx.thread[WRITE_THREAD].state = 1;fclose(fp);
printf("[WRITE_THREAD FINISHED]\n");
report();yield(WRITE_THREAD);
}
void coroutine_read(void *arg)
{char buf[128];int i = (int) arg;i = READ_THREAD;gCtx.thread[i].wait_sched_lock_thread = 999;mutex_lock(&gCtx.thread[i].mutex);mutex_unlock(&gCtx.thread[i].mutex);lock_scheduler();while (gCtx.thread[i].wait_sched_lock_thread != 0) {sched_yield();}mutex_lock(&bx_mutex);while (!bx.eobf || bx.bc) {bx_read(buf, 128);printf("%s\n", buf);}gCtx.thread[READ_THREAD].state = 1;
printf("[READ_THREAD FINISHED]\n");
report();yield(READ_THREAD);
}
线程被调度程序唤醒之后,立即用lock_scheduler()对调度程序的mutex上锁,让调度程序完成调度语义之后可以在这个锁上休眠,等候被调线程完成操作后重新唤醒它。
void lock_scheduler()
{mutex_lock(&gCtx.sched);gCtx.wait_thread_lock_sched = 0;
}
调度程序用resume()唤醒读写线程,之一。唤醒之后跟读写线程交换mutex锁定。最后,调度程序的mutex被读写线程锁定,读写线程的mutex被调度程序锁定:
void resume(int th)
{int i = 0;gCtx.wait_thread_lock_sched = 1;gCtx.thread[th].wait_sched_lock_thread = 1;mutex_unlock(&gCtx.thread[th].mutex);while (gCtx.wait_thread_lock_sched != 0) {sched_yield();}i = 0;mutex_lock(&gCtx.thread[th].mutex);gCtx.thread[th].wait_sched_lock_thread = 0;
}
读写线程做完操作之后,用yield()唤醒调度,同步之后在上了锁的mutex休眠。奇怪的第一句是防止读写线程运行得太快。唤醒之后工作做完了,又一次进入yield,而调度程序上次reume的最后一个同步语句还没有完成。这是有可能发生的。只要OS切换这个调度程序之后,一直不给它CPU,调度程序的动作就会一直被搁置。
void yield(int th)
{int i = 0;if (gCtx.thread[th].wait_sched_lock_thread != 0 ) {gCtx.thread[th].pend++;do {sched_yield();} while(gCtx.thread[th].wait_sched_lock_thread != 0);}mutex_unlock(&bx_mutex);//printf("thread %d yield\n", th);mutex_unlock(&gCtx.sched);gCtx.thread[th].wait_sched_lock_thread = -1;mutex_lock(&gCtx.thread[th].mutex);mutex_unlock(&gCtx.thread[th].mutex);gCtx.thread[th].wait_sched_lock_thread = 0;
}
int main()
{int i;pthread_mutexattr_t mat;pthread_mutexattr_init(&mat);pthread_mutexattr_settype(&mat, PTHREAD_MUTEX_NORMAL);pthread_mutex_init(&bx_mutex, &mat);pthread_mutex_init(&gCtx.sched, &mat);pthread_mutex_init(&gCtx.thread[0].mutex, &mat);pthread_mutex_init(&gCtx.thread[1].mutex, &mat);mutex_lock(&gCtx.thread[0].mutex);mutex_lock(&gCtx.thread[1].mutex);start_coroutine((pf) coroutine_write, NULL);start_coroutine((pf) coroutine_read, NULL);sched_yield();while (gCtx.thread[0].wait_sched_lock_thread == 0) {sched_yield();}gCtx.thread[0].wait_sched_lock_thread = 0;while (gCtx.thread[1].wait_sched_lock_thread == 0) {sched_yield();}gCtx.thread[1].wait_sched_lock_thread = 0;run = 1;while (TRUE) {resched:i = sched();if (gCtx.thread[i].state == 0) {resume(i);mutex_lock(&gCtx.sched);mutex_unlock(&gCtx.sched);} else break;}printf("done\n");return 0;
}
现在看起来已经很严谨了。好像已经可以正常运行了。但是不是哦。多线程变成常会遇到各种意想不到的情况。到了这一步,后来就发生了莫名其妙的crash,但不是每次都错。然后是连续不断的调试。直至怀疑系统lib有bug,甚至是不是自己的机器“热坏了”。好在现在调好了,不用怀疑最后这两条了。
最后贴上完整代码,感兴趣的可以自己试下。不保证一定没有错误哦!
#include <stdio.h>
#include <string.h>
#include <setjmp.h>
#include <stdlib.h>
#include <pthread.h>int mutex_lock(pthread_mutex_t * mutex)
{return pthread_mutex_lock(mutex);
}int mutex_unlock(pthread_mutex_t * mutex)
{return pthread_mutex_unlock(mutex);
}typedef int BOOL;
#define TRUE 1
#define FALSE 0extern pthread_mutex_t bx_mutex;typedef struct _Context_ {pthread_mutex_t sched;volatile int wait_thread_lock_sched;int finish;struct {pthread_mutex_t mutex;volatile int wait_sched_lock_thread;int pend;int state;} thread[2];
} Context;#define WRITE_THREAD 0
#define READ_THREAD 1Context gCtx;
static int next_thread;
int run;void report()
{printf("thread 0: locksync %d, pend %d, state %d\n",gCtx.thread[0]. wait_sched_lock_thread,gCtx.thread[0]. pend,gCtx.thread[0]. state);printf("thread 1: locksync %d, pend %d, state %d\n",gCtx.thread[1]. wait_sched_lock_thread,gCtx.thread[1]. pend,gCtx.thread[1]. state);
}
void resume(int th)
{int i = 0;gCtx.wait_thread_lock_sched = 1;gCtx.thread[th].wait_sched_lock_thread = 1;mutex_unlock(&gCtx.thread[th].mutex);while (gCtx.wait_thread_lock_sched != 0) {sched_yield();}i = 0;mutex_lock(&gCtx.thread[th].mutex);gCtx.thread[th].wait_sched_lock_thread = 0;
}void lock_scheduler()
{mutex_lock(&gCtx.sched);gCtx.wait_thread_lock_sched = 0;
}void yield(int th)
{int i = 0;if (gCtx.thread[th].wait_sched_lock_thread != 0 ) {gCtx.thread[th].pend++;do {sched_yield();} while(gCtx.thread[th].wait_sched_lock_thread != 0);}mutex_unlock(&bx_mutex);//printf("thread %d yield\n", th);mutex_unlock(&gCtx.sched);gCtx.thread[th].wait_sched_lock_thread = -1;mutex_lock(&gCtx.thread[th].mutex);mutex_unlock(&gCtx.thread[th].mutex);gCtx.thread[th].wait_sched_lock_thread = 0;
}#define MAXBUFS 10
struct buffer {char buf[MAXBUFS][128];int pos;int bc;int eobf;
};
struct buffer bx;
pthread_mutex_t bx_mutex;void bx_read(char *s, int n)
{restart:if (bx.bc) {if (!(bx.pos >= 0 && bx.pos < MAXBUFS)) {printf("error\n");exit(0);}strncpy(s, bx.buf[bx.pos], n);s[n - 1] = '\0';bx.bc--;bx.pos++;if (bx.pos >= MAXBUFS)bx.pos = 0;return;} else if (bx.eobf) {gCtx.finish = 1;gCtx.thread[run].state = 1;yield(READ_THREAD);} else {yield(READ_THREAD);lock_scheduler();mutex_lock(&bx_mutex);goto restart;}
}void bx_write(char *s, int n)
{int wpos;while (bx.bc >= MAXBUFS) {yield(WRITE_THREAD);lock_scheduler();mutex_lock(&bx_mutex);}wpos = bx.pos + bx.bc;if (wpos >= MAXBUFS)wpos -= MAXBUFS;if (!(wpos >= 0 && wpos < MAXBUFS)) {printf("error\n");exit(0);}if (n < 128) {strcpy(bx.buf[wpos], s);}else {strncpy(bx.buf[wpos], s, 128);bx.buf[wpos][127] = '\0';}bx.bc++;
}typedef void *(*pf) (void *);
BOOL start_coroutine(pf func, void *arg);
void coroutine_read(void *arg);void coroutine_write(void *arg)
{FILE *fp;char s[128];int len;int i = (int) arg;i = WRITE_THREAD;gCtx.thread[i].wait_sched_lock_thread = 999;mutex_lock(&gCtx.thread[i].mutex);mutex_unlock(&gCtx.thread[i].mutex);lock_scheduler();while (gCtx.thread[i].wait_sched_lock_thread != 0) {sched_yield();}fp = fopen("b.c", "r");if (!fp) {printf("can not open `b.c'\n");exit(0);}mutex_lock(&bx_mutex);while (fgets(s, 128, fp) != NULL) {len = strlen(s);if (len) {--len;if (s[len] == '\n') s[len] = '\0'; else ++len;}if (len) {--len;if (s[len] == '\r') s[len] = '\0'; else ++len;}bx_write(s, len + 1);}bx.eobf = 1;gCtx.thread[WRITE_THREAD].state = 1;fclose(fp);
printf("[WRITE_THREAD FINISHED]\n");
report();yield(WRITE_THREAD);
}void coroutine_read(void *arg)
{char buf[128];int i = (int) arg;i = READ_THREAD;gCtx.thread[i].wait_sched_lock_thread = 999;mutex_lock(&gCtx.thread[i].mutex);mutex_unlock(&gCtx.thread[i].mutex);lock_scheduler();while (gCtx.thread[i].wait_sched_lock_thread != 0) {sched_yield();}mutex_lock(&bx_mutex);while (!bx.eobf || bx.bc) {bx_read(buf, 128);printf("%s\n", buf);}gCtx.thread[READ_THREAD].state = 1;
printf("[READ_THREAD FINISHED]\n");
report();yield(READ_THREAD);
}BOOL start_coroutine(pf func, void *arg)
{int thread_id;pthread_t pt_id;thread_id = next_thread++;pthread_create(&pt_id, NULL, func, (void *) thread_id);pthread_detach(pt_id);return TRUE;
}int sched()
{int s;if (run == 0) s = 1; else s = 0;if (gCtx.thread[s].state == 0) run = s;return run;
}int main()
{int i;pthread_mutexattr_t mat;pthread_mutexattr_init(&mat);pthread_mutexattr_settype(&mat, PTHREAD_MUTEX_NORMAL);pthread_mutex_init(&bx_mutex, &mat);pthread_mutex_init(&gCtx.sched, &mat);pthread_mutex_init(&gCtx.thread[0].mutex, &mat);pthread_mutex_init(&gCtx.thread[1].mutex, &mat);mutex_lock(&gCtx.thread[0].mutex);mutex_lock(&gCtx.thread[1].mutex);start_coroutine((pf) coroutine_write, NULL);start_coroutine((pf) coroutine_read, NULL);sched_yield();while (gCtx.thread[0].wait_sched_lock_thread == 0) {sched_yield();}gCtx.thread[0].wait_sched_lock_thread = 0;while (gCtx.thread[1].wait_sched_lock_thread == 0) {sched_yield();}gCtx.thread[1].wait_sched_lock_thread = 0;run = 1;while (TRUE) {resched:i = sched();if (gCtx.thread[i].state == 0) {resume(i);mutex_lock(&gCtx.sched);mutex_unlock(&gCtx.sched);} else break;}printf("done\n");return 0;
}
最后的代码里多出了 一个bx_mutex锁。用来保护bx缓冲区。程序实在调不出来才被逼加上了。但是加上了还是没用!现在如果去掉又要好复杂的调试,已经没有心情去做了。
最后解释一下,究竟出了什么bug,这么难搞。后来又是改在哪里好起来的。供大家参考。缓冲区bx_write上来有这一句:
int wpos;while (bx.bc >= MAXBUFS) {yield(WRITE_THREAD);lock_scheduler();mutex_lock(&bx_mutex);}
while 原来写的是if。查我的前文可以看到对比。为什么这么改?linux内核编程中有告诫,睡眠进程被唤醒之后,需要重新检查它等待的条件,不能直接认为它等待的条件已经成立。如果不成立,需要重新进入睡眠。