Local stack of Aora's dequeue vs recusion.

gistfile1.txt

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

#define FALSE 0
#define TRUE 1

typedef uint32_t half_word;
union deque_age {
    void *data;
    struct {
        half_word tag;
        half_word top;
    } fields;
};

#define DEQUE_VALUE_SIZE (16384 / 64)
#define DEQUE_ARRAY_SIZE DEQUE_VALUE_SIZE / 100
#define DEQUE_SIZE_MASK (DEQUE_VALUE_SIZE - 1)
#define DEQUE_MAX (DEQUE_VALUE_SIZE - 2)

typedef struct local_markstack {
    void *ary[64];
    size_t index;
    struct local_markstack *next;
} local_markstack_t;

typedef struct deque {
    void *datas[DEQUE_VALUE_SIZE];
    size_t bottom;
    union deque_age age;
    local_markstack_t *stack;
} deque_t;

static deque_t *deq = NULL;

void
init(void) {
    int i;
    local_markstack_t *p;

    deq = malloc(sizeof(deque_t));
    memset(deq, 0, sizeof(deque_t));

    for (i = 0; i < DEQUE_VALUE_SIZE; i++) {
        p = malloc(sizeof(local_markstack_t));
        p->index = 0;
        p->next = deq->stack;
        deq->stack = p;
    }
}

static inline size_t
deque_increment(deque_t *deque, size_t index)
{
    return (index + 1) & DEQUE_SIZE_MASK;
}

static inline size_t
deque_decrement(deque_t *deque, size_t index)
{
    return (index - 1) & DEQUE_SIZE_MASK;
}

static inline size_t
raw_size_deque(deque_t *deque, size_t bottom, size_t top)
{
    size_t size;

    size = (bottom - top) & DEQUE_SIZE_MASK;
    return size;
}

static inline size_t
size_deque(deque_t *deque, size_t bottom, size_t top)
{
    size_t size;

    size = raw_size_deque(deque, bottom, top);
    if (size == DEQUE_SIZE_MASK)
        return 0;
    return size;
}

static inline int
is_full_deque(deque_t *deque, size_t bottom, size_t top)
{
    if (size_deque(deque, bottom, top) == DEQUE_MAX) {
        return TRUE;
    }
    return FALSE;
}

static inline int
is_empty_deque(deque_t *deque, size_t bottom, size_t top)
{
    if (size_deque(deque, bottom, top) == 0) {
        return TRUE;
    }
    return FALSE;
}

static inline void *
atomic_compxchg_ptr(void **addr, void *old, void *new)
{
    return __sync_val_compare_and_swap(addr, old, new);
}

static int
pop_bottom(deque_t *deque, local_markstack_t **data)
{
    union deque_age old_age, new_age, res_age;
    size_t local_bottom, size;

    old_age = deque->age;
    local_bottom = deque->bottom;
    if (is_empty_deque(deque, local_bottom, old_age.fields.top))
        return FALSE;

    local_bottom = deque_decrement(deque, local_bottom);
    deque->bottom = local_bottom;

    /* necessary memory barrier. */
    /* order_access_memory_barrier(); */
    *data = deque->datas[local_bottom];
    /* must second read of age after local_bottom decremented.
       The local_bottom decrement is lock on. */
    old_age = deque->age;
    if (size_deque(deque, local_bottom, old_age.fields.top) > 0) {
        return TRUE;
    }

    /* only one data in deque */
    new_age.fields.top = local_bottom;
    new_age.fields.tag = old_age.fields.tag + 1;
    if (local_bottom == old_age.fields.top) {
        res_age.data = atomic_compxchg_ptr((void **)&deque->age.data,
                                           (void *)old_age.data,
                                           (void *)new_age.data);
        if (res_age.data == old_age.data) {
            /* before pop_top() */
            return TRUE;
        }
    }
    /* after pop_top() */
    deque->age = new_age;
    return FALSE;
}

static int
push_bottom(deque_t *deque, local_markstack_t *data)
{
    size_t local_bottom;
    half_word top;

    local_bottom = deque->bottom;
    top = deque->age.fields.top;
    if (!is_full_deque(deque, local_bottom, top)) {
        deque->datas[local_bottom] = data;
        deque->bottom = deque_increment(deque, local_bottom);
        return TRUE;
    }
    return FALSE;
}

typedef struct data {
    int flag;
    struct data *next;
} data_t;

static data_t *top = NULL;
static data_t *cur = NULL;

void
setup_datas(void)
{
    int i;
    data_t *p;
    for (i = 0; i < 255; i++) {
        p = (data_t *)malloc(sizeof(data_t));
        memset(p, 0, sizeof(p));
        p->next = top;
        top = p;
    }
}

void
mark_by_deqa(void)
{
    data_t *p;
    local_markstack_t *m;
    int i;

    m = deq->stack;
    m->ary[m->index++] = top;

    while (TRUE) {
        if (m->index == 0) {
            if (!pop_bottom(deq, &m))
                break;
        }

        p = m->ary[--m->index];
        p->flag = 0;

        if (p->next != NULL) {
            if (m->index >= 64) {
                push_bottom(deq, m);
                m = deq->stack = m->next;
            }
            m->ary[m->index++] = p->next;
        }
    }
    // 1. 配列内のindexをチェック
    // 2. もし空ならばpop_bottomをおこない、未マークの配列を取得。
    //    なければループ終了
    // 3. 配列内オブジェクトを一つ取り出し、その子オブジェクトを格納
    // 4. 配列がすでにいっぱいだったらlocal stackから切り離しpush_bottom
}

void
mark_by_recusion(data_t *p)
{
    if (p != NULL) {
        p->flag = 0;
        mark_by_recusion(p->next);
    }
}

void
main(void)
{
    int i;
    init();
    setup_datas();

    for (i = 0; i < 100000; i++) {
        mark_by_recusion(top);
        //mark_by_deqa();
    }
}

こうするとdequeueの方が早くなる。そりゃまぁ当然なんだけどね…。