coro.c

#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <assert.h>
#include <string.h>
#define streq(a, b) (!strcmp((a), (b)))

#ifndef __USE_GNU
#define __USE_GNU
#endif
#include <setjmp.h>
#include <sched.h>
#include <unistd.h>
#include <pthread.h>
#include "xmmintrin.h"
#include <ucontext.h>

typedef int8_t i8;
typedef int16_t i16;
typedef int32_t i32;
typedef int64_t i64;
typedef uint8_t u8;
typedef uint16_t u16;
typedef uint32_t u32;
typedef uint64_t u64;

#define STACK_SIZE 64*1024
#define FIBER (void(*)(void))

struct coro_attr {
    void *stack_buffer;
    size_t stack_size;
};

struct coro {
    int done;
    int owned;
    void *data;
    ucontext_t src;
    ucontext_t dst;
    struct coro_attr attr;
};

static void
coro_attr_init(struct coro_attr *attr)
{
    memset(attr, 0, sizeof(*attr));
    attr->stack_buffer = 0;
    attr->stack_size = STACK_SIZE;
}

static void
coro_create(struct coro *coro, struct coro_attr *attr,
    void(*callback)(struct coro*), void *data)
{
    memset(coro, 0, sizeof(*coro));
    getcontext(&coro->src);
    coro->done = 0;
    coro->data = data;
    coro->attr = *attr;
    coro->src.uc_link = 0;
    coro->src.uc_stack.ss_size = attr->stack_size;
    if (attr->stack_buffer) {
        coro->src.uc_stack.ss_sp = attr->stack_buffer;
        coro->owned = 0;
    } else {
        coro->src.uc_stack.ss_sp = calloc(attr->stack_size,1);
        coro->owned = 1;
    }
    makecontext(&coro->src, FIBER callback, 1, coro);
    swapcontext(&coro->dst, &coro->src);
}

static void
coro_destroy(struct coro *coro)
{
    if (coro->owned)
        free(coro->src.uc_stack.ss_sp);
    memset(coro, 0, sizeof(*coro));
}

static void
coro_yield(struct coro *coro)
{
    swapcontext(&coro->src, &coro->dst);
}

static void
coro_done(struct coro * coro)
{
    coro->done = 1;
    swapcontext(&coro->src, &coro->dst);
}

static void
coro_call(struct coro *coro)
{
    swapcontext(&coro->dst, &coro->src);
}

static void
coro_send(struct coro *coro, void *data)
{
    coro->data = data;
    coro_call(coro);
}

/* --------------------------------------------------------------
 *                          GENERATOR
 * --------------------------------------------------------------*/
static void
countdown(struct coro *iter)
{
    int *n = iter->data;
    while (*n >= 0) {
        coro_yield(iter);
        *n = *n - 1;
    }
    coro_done(iter);
}

/* --------------------------------------------------------------
 *                          PIPELINE
 * --------------------------------------------------------------*/
static int str_match_here(const char *regexp, const char *text);
static int str_match_star(int c, const char *regexp, const char *text);

static int
str_match_here(const char *regexp, const char *text)
{
    if (regexp[0] == '\0')
        return 1;
    if (regexp[1] == '*')
        return str_match_star(regexp[0], regexp+2, text);
    if (regexp[0] == '$' && regexp[1] == '\0')
        return *text == '\0';
    if (*text!='\0' && (regexp[0]=='.' || regexp[0]==*text))
        return str_match_here(regexp+1, text+1);
    return 0;
}

static int
str_match_star(int c, const char *regexp, const char *text)
{
    do {/* a '* matches zero or more instances */
        if (str_match_here(regexp, text))
            return 1;
    } while (*text != '\0' && (*text++ == c || c == '.'));
    return 0;
}

static int
str_match(const char *text, const char *regexp)
{
    /*
    c    matches any literal character c
    .    matches any single character
    ^    matches the beginning of the input string
    $    matches the end of the input string
    *    matches zero or more occurrences of the previous character*/
    if (regexp[0] == '^')
        return str_match_here(regexp+1, text);
    do {    /* must look even if string is empty */
        if (str_match_here(regexp, text))
            return 1;
    } while (*text++ != '\0');
    return 0;
}

static void
cat(struct coro *coro)
{
    while (1) {
        const char *line;
        coro_yield(coro);
        line = coro->data;
        printf("%s\n", line);
    }
}

struct grep_out {
    struct coro *next;
    const char *pattern;
};

static void
grep(struct coro *coro)
{
    struct grep_out ctx = *((struct grep_out*)coro->data);
    while (1) {
        const char *line;
        coro_yield(coro);
        line = coro->data;
        if (str_match(line, ctx.pattern))
            coro_send(ctx.next, line);
    }
}

struct broadcast_out {
    struct coro *list;
    int count;
};

static void
broadcast(struct coro *coro)
{
    struct broadcast_out *targets;
    targets = coro->data;
    while (1) {
        void *item;
        coro_yield(coro);
        item = coro->data;

        {int target_index = 0;
        for (target_index; target_index < targets->count; ++target_index) {
            struct coro *target = targets->list + target_index;
            coro_send(target, item);
        }}
    }
}

/* --------------------------------------------------------------
 *
 *                          TEST
 *
 * --------------------------------------------------------------*/
int main(int argc, const char **argv)
{
    /* generator*/
    {int n = 4;
    struct coro coro;
    struct coro_attr attr;
    coro_attr_init(&attr);
    coro_create(&coro, &attr, countdown, &n);
    while (!coro.done) {
        printf("%d\n", n);
        coro_call(&coro);
    }
    coro_destroy(&coro);}

    {/* pipeline */
        int coro_index;
        struct coro_attr attr;
        struct coro printer;
        struct coro cgrep[3];
        struct coro bc;

        /* setup pipline data */
        struct grep_out g[3];
        struct broadcast_out bc_out;
        bc_out.count = 3; bc_out.list = cgrep;
        g[0].next = &printer; g[0].pattern = "python";
        g[1].next = &printer; g[1].pattern = "ply";
        g[2].next = &printer; g[2].pattern = "swig";

        /* setup pipline coroutine */
        coro_attr_init(&attr);
        coro_create(&printer, &attr, cat, 0);
        coro_create(&cgrep[0], &attr, grep, &g[0]);
        coro_create(&cgrep[1], &attr, grep, &g[1]);
        coro_create(&cgrep[2], &attr, grep, &g[2]);
        coro_create(&bc, &attr, broadcast, &bc_out);

        /* use pipeline */
        coro_send(&bc, "But yes but no but yes but no");
        coro_send(&bc, "What ever you ply not no");
        coro_send(&bc, "klopapier klang swig");
        coro_send(&bc, "That python bit me!");

        /* cleanup */
        coro_destroy(&cgrep[0]);
        coro_destroy(&cgrep[1]);
        coro_destroy(&cgrep[2]);
        coro_destroy(&printer);
        coro_destroy(&bc);
    }

    return 0;
}