<commit_before>/* ring.c
 * inspired by slaeshjag's Glouija
 */

#include <fcntl.h>
#include <sched.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/sysctl.h>
#include <sys/types.h>
#include <unistd.h>

#include "ring.h"

#define RING_SIZE (1024 * 4096)
#define ALIGN4(x) (((x) & 3) ? (((x) & ~3) + 4) : (x))

#define SPINLOCK_YIELD 2000
#define SPINLOCK_COUNT 10000
#define SPINLOCK_LOOP(cond) \
    for (int i = 0; i < SPINLOCK_COUNT && (cond); i++) { \
        if (i % SPINLOCK_YIELD == 0) sched_yield(); \
    }
#define SYNC_WHILE(cond) \
    while ((cond)) { \
        SPINLOCK_LOOP(cond); \
        while ((cond)) ring_wait(ring); \
    }

void ring_wait(ring_t *ring) {
    char c;
    if (__sync_lock_test_and_set(ring->waiting, 1)) {
        fprintf(stderr, "warning: both sides waiting?\n");
    }
    read(ring->fd, &c, 1);
}

void ring_post(ring_t *ring) {
    // TODO: check if waiting first, or use nonblocking IO?
    if (__sync_bool_compare_and_swap(ring->waiting, 1, 0)) {
        char c = 0;
        write(ring->fd, &c, 1);
    }
}

static void ring_wait_read(ring_t *ring) {
    SYNC_WHILE(*ring->dir != ring->me);
    SYNC_WHILE(*ring->mark == *ring->write && *ring->wrap == 0);
}

static void ring_wait_write(ring_t *ring, size_t size) {
    size_t avail = 0;
    uint32_t read, write, wrap;
    int i = 0;
    while (1) {
        // TODO: handle dir?
        read = *ring->read, write = *ring->write;
        wrap = *ring->wrap;
        //    [...|write|[here]|read|...]
        if (write < read) {
            avail = read - write;
        //    [[ here ]|read|.*|write|[ here ]]
        } else if (write > read || (write == read && wrap == 0)) {
            avail = ring->size - write;
            if (read > avail)
                avail = read;
        }
        if (avail >= size) {
            break;
        }
        if (i % SPINLOCK_YIELD == 0) sched_yield();
        if (i > SPINLOCK_COUNT) ring_wait(ring);
        else i++;
    }
}

void *ring_read(ring_t *ring, size_t *size_ret) {
    // TODO: write magic bytes into the mess to mark new messages and signal memory problems?
    ring_wait_read(ring);
    if (*ring->mark > 0 && *ring->mark == *ring->wrap) {
        *ring->mark = 0;
        *ring->wrap = 0;
    }
    uint32_t size = *(uint32_t *)(ring->buf + *ring->mark);
    if (size == 0 || size > ring->size) {
        fprintf(stderr, "panic: ring data size must be 0 < (%d) < %lu\n", size, ring->size);
        abort();
    }
    void *src;
    if (*ring->mark + size <= ring->size) {
        src = ring->buf + *ring->mark + sizeof(uint32_t);
        *ring->mark += size;
    } else {
        // our read wrapped around
        src = ring->buf;
        *ring->mark = size;
    }
    if (size_ret) *size_ret = size - sizeof(uint32_t);
    return src;
}

void ring_read_into(ring_t *ring, void *dst) {
    size_t size;
    void *data = ring_read(ring, &size);
    memcpy(dst, data, size);
}

void ring_advance(ring_t *ring) {
    *ring->read = *ring->mark;
    ring_post(ring);
}

void *ring_dma(ring_t *ring, size_t size) {
    // add an extra uint32_t to store the size
    size += sizeof(uint32_t);
    // make sure we have enough unmarked free space
    uint32_t read = *ring->read, mark = *ring->mark;
    uint32_t unmarked;
    if (read < mark) unmarked = read - mark;
    else             unmarked = ring->size - mark + read;
    if (size > unmarked) {
        fprintf(stderr, "panic: ring_write size %lu > unmarked %u\n", size + 4, unmarked);
        abort();
    }
    // wait for free space
    ring_wait_write(ring, size);
    size_t remain = ring->size - *ring->write;
    // pick destination and wrap if necessary
    void *dst;
    if (remain > size) {
        dst = ring->buf + *ring->write;
        ring->dma_write = *ring->write + size;
    } else {
        dst = ring->buf;
        ring->dma_wrap = *ring->write;
        ring->dma_write = size;
    }
    *(uint32_t *)dst = size;
    return dst + sizeof(uint32_t);
}

void ring_dma_done(ring_t *ring) {
    // move position
    if (ring->dma_wrap)
        *ring->wrap = ring->dma_wrap;
    *ring->write = ring->dma_write;
    ring->dma_wrap = 0;
    ring->dma_write = 0;
    // update direction
    if (*ring->dir == ring->me)
        *ring->dir = !ring->me;
    ring_post(ring);
}

int ring_write_multi(ring_t *ring, ring_val_t *vals, int count) {
    // measure the total size
    size_t size = 0;
    for (int i = 0; i < count; i++)
        size += vals[i].size;
    size = ALIGN4(size);
    void *dst = ring_dma(ring, size);
    // write values
    for (int i = 0; i < count; i++) {
        memcpy(dst, vals[i].buf, vals[i].size);
        dst += vals[i].size;
    }
    ring_dma_done(ring);
    return 0;
}

int ring_write(ring_t *ring, void *buf, size_t bufsize) {
    ring_val_t vals[] = {{buf, bufsize}};
    return ring_write_multi(ring, vals, 1);
}

const size_t cache_line_size() {<commit_msg>use magic ringbuffer for wrapping<commit_after>/* ring.c
 * inspired by slaeshjag's Glouija
 */

#include <fcntl.h>
#include <sched.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/sysctl.h>
#include <sys/types.h>
#include <unistd.h>

#include "ring.h"

#define RING_SIZE (1024 * 4096)
#define ALIGN4(x) (((x) & 3) ? (((x) & ~3) + 4) : (x))

#define SPINLOCK_YIELD 2000
#define SPINLOCK_COUNT 10000
#define SPINLOCK_LOOP(cond) \
    for (int i = 0; i < SPINLOCK_COUNT && (cond); i++) { \
        if (i % SPINLOCK_YIELD == 0) sched_yield(); \
    }
#define SYNC_WHILE(cond) \
    while ((cond)) { \
        SPINLOCK_LOOP(cond); \
        while ((cond)) ring_wait(ring); \
    }

void ring_wait(ring_t *ring) {
    char c;
    if (__sync_lock_test_and_set(ring->waiting, 1)) {
        fprintf(stderr, "warning: both sides waiting?\n");
    }
    read(ring->fd, &c, 1);
}

void ring_post(ring_t *ring) {
    // TODO: check if waiting first, or use nonblocking IO?
    if (__sync_bool_compare_and_swap(ring->waiting, 1, 0)) {
        char c = 0;
        write(ring->fd, &c, 1);
    }
}

static void ring_wait_read(ring_t *ring) {
    SYNC_WHILE(*ring->dir != ring->me);
    SYNC_WHILE(*ring->mark == *ring->write && *ring->wrap == 0);
}

static void ring_wait_write(ring_t *ring, size_t size) {
    size_t avail = 0;
    uint32_t read, write, wrap;
    int i = 0;
    while (1) {
        // TODO: handle dir?
        read = *ring->read, write = *ring->write;
        wrap = *ring->wrap;
        //    [...|write|[here]|read|...]
        if (write < read) {
            avail = read - write;
        //    [[ here ]|read|.*|write|[ here ]]
        } else if (write > read || (write == read && wrap == 0)) {
            avail = ring->size - (write - read);
        }
        if (avail >= size) {
            break;
        }
        if (i % SPINLOCK_YIELD == 0) sched_yield();
        if (i > SPINLOCK_COUNT) ring_wait(ring);
        else i++;
    }
}

void *ring_read(ring_t *ring, size_t *size_ret) {
    // TODO: write magic bytes into the mess to mark new messages and signal memory problems?
    ring_wait_read(ring);
    uint32_t size = *(uint32_t *)(ring->buf + *ring->mark);
    if (size == 0 || size > ring->size) {
        fprintf(stderr, "panic: ring data size must be 0 < (%d) < %lu\n", size, ring->size);
        abort();
    }
    void *src = ring->buf + *ring->mark + sizeof(uint32_t);
    *ring->mark += size;
    // detect wraps
    if (*ring->mark > ring->size) {
        *ring->wrap = 0;
        *ring->mark -= ring->size;
    }
    if (size_ret) *size_ret = size - sizeof(uint32_t);
    return src;
}

void ring_read_into(ring_t *ring, void *dst) {
    size_t size;
    void *data = ring_read(ring, &size);
    memcpy(dst, data, size);
}

void ring_advance(ring_t *ring) {
    *ring->read = *ring->mark;
    ring_post(ring);
}

void *ring_dma(ring_t *ring, size_t size) {
    // add an extra uint32_t to store the size
    size += sizeof(uint32_t);
    // make sure we have enough unmarked free space
    uint32_t read = *ring->read, mark = *ring->mark;
    uint32_t unmarked;
    if (read < mark) unmarked = read - mark;
    else             unmarked = ring->size - mark + read;
    if (size > unmarked) {
        fprintf(stderr, "panic: ring_write size %lu > unmarked %u\n", size + 4, unmarked);
        abort();
    }
    // wait for free space
    ring_wait_write(ring, size);
    size_t remain = ring->size - *ring->write;

    // we're mapped twice, so writes to the end will auto-wrap
    void *dst = ring->buf + *ring->write;
    // modulo here ensures the dma target wraps
    ring->dma_write = *ring->write + size;
    if (ring->dma_write > ring->size) {
        ring->dma_write -= ring->size;
        ring->dma_wrap = 1;
    }
    *(uint32_t *)dst = size;
    return dst + sizeof(uint32_t);
}

void ring_dma_done(ring_t *ring) {
    // move position
    if (ring->dma_wrap)
        *ring->wrap = 1;
    *ring->write = ring->dma_write;
    ring->dma_wrap = 0;
    ring->dma_write = 0;
    // update direction
    if (*ring->dir == ring->me)
        *ring->dir = !ring->me;
    ring_post(ring);
}

int ring_write_multi(ring_t *ring, ring_val_t *vals, int count) {
    // measure the total size
    size_t size = 0;
    for (int i = 0; i < count; i++)
        size += vals[i].size;
    size = ALIGN4(size);
    void *dst = ring_dma(ring, size);
    // write values
    for (int i = 0; i < count; i++) {
        memcpy(dst, vals[i].buf, vals[i].size);
        dst += vals[i].size;
    }
    ring_dma_done(ring);
    return 0;
}

int ring_write(ring_t *ring, void *buf, size_t bufsize) {
    ring_val_t vals[] = {{buf, bufsize}};
    return ring_write_multi(ring, vals, 1);
}

const size_t cache_line_size() {
{'fraction_non_alphanumeric': 0.16216493786587244, 'fraction_numerical': 0.014583547293827668, 'mean_word_length': 2.385952712100139, 'pattern_counts': {'":': 0, '<': 36, '<?xml version=': 0, '>': 144, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5726306', 'n_tokens_mistral': 3702, 'n_tokens_neox': 3569, 'n_words': 925}