longjmp和多线程：读写线程实例

因为多线程的问题比较复杂。这里再给个实际内容的例子。

这个例子中，读写线程通过一个bx缓冲区合作。如果缓冲区为空，读线程调用yield原语让出处理机让写线程来写。如果缓冲区写满了，写线程也是调用yield原语让读线程来读。

写线程由主程序启动。读线程由写线程启动。程序的功能是从当前目录读取一个文本文件并显示出来。读线程比写线程先运行。因为一开始缓冲区为空，读线程第一次读就需要让出处理机，让写线程开始写 buffer。

#include <stdio.h>
#include <string.h>
#include <setjmp.h>
#include <stdlib.h>

typedef int BOOL;
#define TRUE    1
#define FALSE   0


typedef struct _Context_ {
    jmp_buf mainBuf;
    int finish;
    struct {
        jmp_buf buf;
        int state;
    } thread[2];
} Context;


Context gCtx;
static int next_thread;
int run;

#define resume() \
    if (gCtx.thread[run].state==0) \
    { \
        longjmp(gCtx.thread[run].buf, 1); \
    }


#define yield() \
    if (0 == setjmp(gCtx.thread[run].buf)) \
    { \
        longjmp(gCtx.mainBuf, 1); \
    }

#define MAXBUFS 10
struct buffer {
        char buf[MAXBUFS][128];
        int pos;
        int bc;
        int eobf;
};
struct buffer bx;

void bx_read(char *s, int n)
{
restart:
        if (bx.bc) {
                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();
        }
        else {
                yield();
                goto restart;
        }
}

void bx_write(char *s, int n)
{
        int wpos;
        if(bx.bc>=MAXBUFS)  {
                yield();
        }
        wpos = bx.pos+bx.bc;
        if(wpos>=MAXBUFS) wpos-=MAXBUFS;

        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;
        fp = fopen("b.c", "r");

        if( !fp)  {
                printf("can not open `b.c'\n");
                exit(0);
        }
        if (setjmp(gCtx.thread[run].buf)!=0) goto next;
        start_coroutine(coroutine_read, NULL);
next:
        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[run].state=1;
        fclose(fp);
        yield();
}

void coroutine_read(void *arg)
{
        char buf[128];
        while(!bx.eobf|| bx.bc) {
                bx_read(buf, 128);
                printf("%s\n", buf);
        }
        gCtx.thread[run].state=1;
        yield();
}


BOOL start_coroutine(pf func, void *arg)
{
    int reserve[256];
    int thread_id;

    thread_id = next_thread++;
    run = thread_id;
    func(arg);
    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;

    if (setjmp(gCtx.mainBuf)!=0) goto resched;
    start_coroutine(coroutine_write, NULL);
    while (TRUE) {
      resched:
        i = sched();
        printf("sched thread(%d) to run\n", i);
        if (gCtx.thread[i].state == 0) {
            resume();
        } else break;
    }
    printf("done\n");
    return 0;
}