/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */

// FIXME: config.h?

#include <stdint.h>

#include <stdbool.h>

// end FIXME

#include <stdlib.h>

#include <limits.h>

#include <pthread.h>

#include <sys/types.h>

#include <sys/stat.h>

#include <sys/uio.h>

#include <fcntl.h>

#include <unistd.h>

#include <stdio.h>

#include <string.h>

#include <assert.h>

#include "extstore.h"

// TODO: better if an init option turns this on/off.

#ifdef EXTSTORE_DEBUG

#define E_DEBUG(...) \

    do { \

        fprintf(stderr, __VA_ARGS__); \

    } while (0)

#else

#define E_DEBUG(...)

#endif

#define STAT_L(e) pthread_mutex_lock(&e->stats_mutex);

#define STAT_UL(e) pthread_mutex_unlock(&e->stats_mutex);

#define STAT_INCR(e, stat, amount) { \

    pthread_mutex_lock(&e->stats_mutex); \

    e->stats.stat += amount; \

    pthread_mutex_unlock(&e->stats_mutex); \

}

#define STAT_DECR(e, stat, amount) { \

    pthread_mutex_lock(&e->stats_mutex); \

    e->stats.stat -= amount; \

    pthread_mutex_unlock(&e->stats_mutex); \

}

typedef struct __store_wbuf {

    struct __store_wbuf *next;

    char *buf;

    char *buf_pos;

    unsigned int free;

    unsigned int size;

    unsigned int offset; /* offset into page this write starts at */

    bool full; /* done writing to this page */

    bool flushed; /* whether wbuf has been flushed to disk */

} _store_wbuf;

typedef struct _store_page {

    pthread_mutex_t mutex; /* Need to be held for most operations */

    uint64_t obj_count; /* _delete can decrease post-closing */

    uint64_t bytes_used; /* _delete can decrease post-closing */

    uint64_t offset; /* starting address of page within fd */

    unsigned int version;

    unsigned int refcount;

    unsigned int allocated;

    unsigned int written; /* item offsets can be past written if wbuf not flushed */

    unsigned int bucket; /* which bucket the page is linked into */

    unsigned int free_bucket; /* which bucket this page returns to when freed */

    int fd;

    unsigned short id;

    bool active; /* actively being written to */

    bool closed; /* closed and draining before free */

    bool free; /* on freelist */

    _store_wbuf *wbuf; /* currently active wbuf from the stack */

    struct _store_page *next;

} store_page;

typedef struct store_engine store_engine;

typedef struct {

    pthread_mutex_t mutex;

    pthread_cond_t cond;

    obj_io *queue;

    store_engine *e;

    unsigned int depth; // queue depth

} store_io_thread;

typedef struct {

    pthread_mutex_t mutex;

    pthread_cond_t cond;

    store_engine *e;

} store_maint_thread;

struct store_engine {

    pthread_mutex_t mutex; /* covers internal stacks and variables */

    store_page *pages; /* directly addressable page list */

    _store_wbuf *wbuf_stack; /* wbuf freelist */

    obj_io *io_stack; /* IO's to use with submitting wbuf's */

    store_io_thread *io_threads;

    store_maint_thread *maint_thread;

    store_page *page_freelist;

    store_page **page_buckets; /* stack of pages currently allocated to each bucket */

    store_page **free_page_buckets; /* stack of use-case isolated free pages */

    size_t page_size;

    unsigned int version; /* global version counter */

    unsigned int last_io_thread; /* round robin the IO threads */

    unsigned int io_threadcount; /* count of IO threads */

    unsigned int page_count;

    unsigned int page_free; /* unallocated pages */

    unsigned int page_bucketcount; /* count of potential page buckets */

    unsigned int free_page_bucketcount; /* count of free page buckets */

    unsigned int io_depth; /* FIXME: Might cache into thr struct */

    pthread_mutex_t stats_mutex;

    struct extstore_stats stats;

};

static _store_wbuf *wbuf_new(size_t size) {

    _store_wbuf *b = calloc(1, sizeof(_store_wbuf));

    if (b == NULL)

        return NULL;

    b->buf = malloc(size);

    if (b->buf == NULL) {

        free(b);

        return NULL;

    }

    b->buf_pos = b->buf;

    b->free = size;

    b->size = size;

    return b;

}

static store_io_thread *_get_io_thread(store_engine *e) {

    int tid = -1;

    long long int low = LLONG_MAX;

    pthread_mutex_lock(&e->mutex);

    // find smallest queue. ignoring lock since being wrong isn't fatal.

    // TODO: if average queue depth can be quickly tracked, can break as soon

    // as we see a thread that's less than average, and start from last_io_thread

    for (int x = 0; x < e->io_threadcount; x++) {

        if (e->io_threads[x].depth == 0) {

            tid = x;

            break;

        } else if (e->io_threads[x].depth < low) {

                tid = x;

            low = e->io_threads[x].depth;

        }

    }

    pthread_mutex_unlock(&e->mutex);

    return &e->io_threads[tid];

}

static uint64_t _next_version(store_engine *e) {

    return e->version++;

}

static void *extstore_io_thread(void *arg);

static void *extstore_maint_thread(void *arg);

/* Copies stats internal to engine and computes any derived values */

void extstore_get_stats(void *ptr, struct extstore_stats *st) {

    store_engine *e = (store_engine *)ptr;

    STAT_L(e);

    memcpy(st, &e->stats, sizeof(struct extstore_stats));

    STAT_UL(e);

    // grab pages_free/pages_used

    pthread_mutex_lock(&e->mutex);

    st->pages_free = e->page_free;

    st->pages_used = e->page_count - e->page_free;

    pthread_mutex_unlock(&e->mutex);

    st->io_queue = 0;

    for (int x = 0; x < e->io_threadcount; x++) {

        pthread_mutex_lock(&e->io_threads[x].mutex);

        st->io_queue += e->io_threads[x].depth;

        pthread_mutex_unlock(&e->io_threads[x].mutex);

    }

    // calculate bytes_fragmented.

    // note that open and yet-filled pages count against fragmentation.

    st->bytes_fragmented = st->pages_used * e->page_size -

        st->bytes_used;

}

void extstore_get_page_data(void *ptr, struct extstore_stats *st) {

    store_engine *e = (store_engine *)ptr;

    STAT_L(e);

    memcpy(st->page_data, e->stats.page_data,

            sizeof(struct extstore_page_data) * e->page_count);

    STAT_UL(e);

}

const char *extstore_err(enum extstore_res res) {

    const char *rv = "unknown error";

    switch (res) {

        case EXTSTORE_INIT_BAD_WBUF_SIZE:

            rv = "page_size must be divisible by wbuf_size";

            break;

        case EXTSTORE_INIT_NEED_MORE_WBUF:

            rv = "wbuf_count must be >= page_buckets";

            break;

        case EXTSTORE_INIT_NEED_MORE_BUCKETS:

            rv = "page_buckets must be > 0";

            break;

        case EXTSTORE_INIT_PAGE_WBUF_ALIGNMENT:

            rv = "page_size and wbuf_size must be divisible by 1024*1024*2";

            break;

        case EXTSTORE_INIT_TOO_MANY_PAGES:

            rv = "page_count must total to < 65536. Increase page_size or lower path sizes";

            break;

        case EXTSTORE_INIT_OOM:

            rv = "failed calloc for engine";

            break;

        case EXTSTORE_INIT_OPEN_FAIL:

            rv = "failed to open file";

            break;

        case EXTSTORE_INIT_THREAD_FAIL:

            break;

    }

    return rv;

}

// TODO: #define's for DEFAULT_BUCKET, FREE_VERSION, etc

void *extstore_init(struct extstore_conf_file *fh, struct extstore_conf *cf,

        enum extstore_res *res) {

    int i;

    struct extstore_conf_file *f = NULL;

    pthread_t thread;

    if (cf->page_size % cf->wbuf_size != 0) {

        *res = EXTSTORE_INIT_BAD_WBUF_SIZE;

        return NULL;

    }

    // Should ensure at least one write buffer per potential page

    if (cf->page_buckets > cf->wbuf_count) {

        *res = EXTSTORE_INIT_NEED_MORE_WBUF;

        return NULL;

    }

    if (cf->page_buckets < 1) {

        *res = EXTSTORE_INIT_NEED_MORE_BUCKETS;

        return NULL;

    }

    // TODO: More intelligence around alignment of flash erasure block sizes

    if (cf->page_size % (1024 * 1024 * 2) != 0 ||

        cf->wbuf_size % (1024 * 1024 * 2) != 0) {

        *res = EXTSTORE_INIT_PAGE_WBUF_ALIGNMENT;

        return NULL;

    }

    store_engine *e = calloc(1, sizeof(store_engine));

    if (e == NULL) {

        *res = EXTSTORE_INIT_OOM;

        return NULL;

    }

    e->page_size = cf->page_size;

    uint64_t temp_page_count = 0;

    for (f = fh; f != NULL; f = f->next) {

        f->fd = open(f->file, O_RDWR | O_CREAT | O_TRUNC, 0644);

        if (f->fd < 0) {

            *res = EXTSTORE_INIT_OPEN_FAIL;

#ifdef EXTSTORE_DEBUG

            perror("open");

#endif

            free(e);

            return NULL;

        }

        temp_page_count += f->page_count;

        f->offset = 0;

    }

    if (temp_page_count >= UINT16_MAX) {

        *res = EXTSTORE_INIT_TOO_MANY_PAGES;

        free(e);

        return NULL;

    }

    e->page_count = temp_page_count;

    e->pages = calloc(e->page_count, sizeof(store_page));

    if (e->pages == NULL) {

        *res = EXTSTORE_INIT_OOM;

        // FIXME: loop-close. make error label

        free(e);

        return NULL;

    }

    // interleave the pages between devices

    f = NULL; // start at the first device.

    for (i = 0; i < e->page_count; i++) {

        // find next device with available pages

        while (1) {

            // restart the loop

            if (f == NULL || f->next == NULL) {

                f = fh;

            } else {

                f = f->next;

            }

            if (f->page_count) {

                f->page_count--;

                break;

            }

        }

        pthread_mutex_init(&e->pages[i].mutex, NULL);

        e->pages[i].id = i;

        e->pages[i].fd = f->fd;

        e->pages[i].free_bucket = f->free_bucket;

        e->pages[i].offset = f->offset;

        e->pages[i].free = true;

        f->offset += e->page_size;

    }

    // free page buckets allows the app to organize devices by use case

    e->free_page_buckets = calloc(cf->page_buckets, sizeof(store_page *));

    e->page_bucketcount = cf->page_buckets;

    for (i = e->page_count-1; i > 0; i--) {

        e->page_free++;

        if (e->pages[i].free_bucket == 0) {

            e->pages[i].next = e->page_freelist;

            e->page_freelist = &e->pages[i];

        } else {

            int fb = e->pages[i].free_bucket;

            e->pages[i].next = e->free_page_buckets[fb];

            e->free_page_buckets[fb] = &e->pages[i];

        }

    }

    // 0 is magic "page is freed" version

    e->version = 1;

    // scratch data for stats. TODO: malloc failure handle

    e->stats.page_data =

        calloc(e->page_count, sizeof(struct extstore_page_data));

    e->stats.page_count = e->page_count;

    e->stats.page_size = e->page_size;

    // page buckets lazily have pages assigned into them

    e->page_buckets = calloc(cf->page_buckets, sizeof(store_page *));

    e->page_bucketcount = cf->page_buckets;

    // allocate write buffers

    // also IO's to use for shipping to IO thread

    for (i = 0; i < cf->wbuf_count; i++) {

        _store_wbuf *w = wbuf_new(cf->wbuf_size);

        obj_io *io = calloc(1, sizeof(obj_io));

        /* TODO: on error, loop again and free stack. */

        w->next = e->wbuf_stack;

        e->wbuf_stack = w;

        io->next = e->io_stack;

        e->io_stack = io;

    }

    pthread_mutex_init(&e->mutex, NULL);

    pthread_mutex_init(&e->stats_mutex, NULL);

    e->io_depth = cf->io_depth;

    // spawn threads

    e->io_threads = calloc(cf->io_threadcount, sizeof(store_io_thread));

    for (i = 0; i < cf->io_threadcount; i++) {

        pthread_mutex_init(&e->io_threads[i].mutex, NULL);

        pthread_cond_init(&e->io_threads[i].cond, NULL);

        e->io_threads[i].e = e;

        // FIXME: error handling

        pthread_create(&thread, NULL, extstore_io_thread, &e->io_threads[i]);

    }

    e->io_threadcount = cf->io_threadcount;

    e->maint_thread = calloc(1, sizeof(store_maint_thread));

    e->maint_thread->e = e;

    // FIXME: error handling

    pthread_mutex_init(&e->maint_thread->mutex, NULL);

    pthread_cond_init(&e->maint_thread->cond, NULL);

    pthread_create(&thread, NULL, extstore_maint_thread, e->maint_thread);

    extstore_run_maint(e);

    return (void *)e;

}

void extstore_run_maint(void *ptr) {

    store_engine *e = (store_engine *)ptr;

    pthread_cond_signal(&e->maint_thread->cond);

}

// call with *e locked

static store_page *_allocate_page(store_engine *e, unsigned int bucket,

        unsigned int free_bucket) {

    assert(!e->page_buckets[bucket] || e->page_buckets[bucket]->allocated == e->page_size);

    store_page *tmp = NULL;

    // if a specific free bucket was requested, check there first

    if (free_bucket != 0 && e->free_page_buckets[free_bucket] != NULL) {

        assert(e->page_free > 0);

        tmp = e->free_page_buckets[free_bucket];

        e->free_page_buckets[free_bucket] = tmp->next;

    }

    // failing that, try the global list.

    if (tmp == NULL && e->page_freelist != NULL) {

        tmp = e->page_freelist;

        e->page_freelist = tmp->next;

    }

    E_DEBUG("EXTSTORE: allocating new page\n");

    // page_freelist can be empty if the only free pages are specialized and

    // we didn't just request one.

    if (e->page_free > 0 && tmp != NULL) {

        tmp->next = e->page_buckets[bucket];

        e->page_buckets[bucket] = tmp;

        tmp->active = true;

        tmp->free = false;

        tmp->closed = false;

        tmp->version = _next_version(e);

        tmp->bucket = bucket;

        e->page_free--;

        STAT_INCR(e, page_allocs, 1);

    } else {

        extstore_run_maint(e);

    }

    if (tmp)

        E_DEBUG("EXTSTORE: got page %u\n", tmp->id);

    return tmp;

}

// call with *p locked. locks *e

static void _allocate_wbuf(store_engine *e, store_page *p) {

    _store_wbuf *wbuf = NULL;

    assert(!p->wbuf);

    pthread_mutex_lock(&e->mutex);

    if (e->wbuf_stack) {

        wbuf = e->wbuf_stack;

        e->wbuf_stack = wbuf->next;

        wbuf->next = 0;

    }

    pthread_mutex_unlock(&e->mutex);

    if (wbuf) {

        wbuf->offset = p->allocated;

        p->allocated += wbuf->size;

        wbuf->free = wbuf->size;

        wbuf->buf_pos = wbuf->buf;

        wbuf->full = false;

        wbuf->flushed = false;

        p->wbuf = wbuf;

    }

}

/* callback after wbuf is flushed. can only remove wbuf's from the head onward

 * if successfully flushed, which complicates this routine. each callback

 * attempts to free the wbuf stack, which is finally done when the head wbuf's

 * callback happens.

 * It's rare flushes would happen out of order.

 */

static void _wbuf_cb(void *ep, obj_io *io, int ret) {

    store_engine *e = (store_engine *)ep;

    store_page *p = &e->pages[io->page_id];

    _store_wbuf *w = (_store_wbuf *) io->data;

    // TODO: Examine return code. Not entirely sure how to handle errors.

    // Naive first-pass should probably cause the page to close/free.

    w->flushed = true;

    pthread_mutex_lock(&p->mutex);

    assert(p->wbuf != NULL && p->wbuf == w);

    assert(p->written == w->offset);

    p->written += w->size;

    p->wbuf = NULL;

    if (p->written == e->page_size)

        p->active = false;

    // return the wbuf

    pthread_mutex_lock(&e->mutex);

    w->next = e->wbuf_stack;

    e->wbuf_stack = w;

    // also return the IO we just used.

    io->next = e->io_stack;

    e->io_stack = io;

    pthread_mutex_unlock(&e->mutex);

    pthread_mutex_unlock(&p->mutex);

}

/* Wraps pages current wbuf in an io and submits to IO thread.

 * Called with p locked, locks e.

 */

static void _submit_wbuf(store_engine *e, store_page *p) {

    _store_wbuf *w;

    pthread_mutex_lock(&e->mutex);

    obj_io *io = e->io_stack;

    e->io_stack = io->next;

    pthread_mutex_unlock(&e->mutex);

    w = p->wbuf;

    // zero out the end of the wbuf to allow blind readback of data.

    memset(w->buf + (w->size - w->free), 0, w->free);

    io->next = NULL;

    io->mode = OBJ_IO_WRITE;

    io->page_id = p->id;

    io->data = w;

    io->offset = w->offset;

    io->len = w->size;

    io->buf = w->buf;

    io->cb = _wbuf_cb;

    extstore_submit(e, io);

}

/* engine write function; takes engine, item_io.

 * fast fail if no available write buffer (flushing)

 * lock engine context, find active page, unlock

 * if page full, submit page/buffer to io thread.

 *

 * write is designed to be flaky; if page full, caller must try again to get

 * new page. best if used from a background thread that can harmlessly retry.

 */

int extstore_write_request(void *ptr, unsigned int bucket,

        unsigned int free_bucket, obj_io *io) {

    store_engine *e = (store_engine *)ptr;

    store_page *p;

    int ret = -1;

    if (bucket >= e->page_bucketcount)

        return ret;

    pthread_mutex_lock(&e->mutex);

    p = e->page_buckets[bucket];

    if (!p) {

        p = _allocate_page(e, bucket, free_bucket);

    }

    pthread_mutex_unlock(&e->mutex);

    if (!p)

        return ret;

    pthread_mutex_lock(&p->mutex);

    // FIXME: can't null out page_buckets!!!

    // page is full, clear bucket and retry later.

    if (!p->active ||

            ((!p->wbuf || p->wbuf->full) && p->allocated >= e->page_size)) {

        pthread_mutex_unlock(&p->mutex);

        pthread_mutex_lock(&e->mutex);

        _allocate_page(e, bucket, free_bucket);

        pthread_mutex_unlock(&e->mutex);

        return ret;

    }

    // if io won't fit, submit IO for wbuf and find new one.

    if (p->wbuf && p->wbuf->free < io->len && !p->wbuf->full) {

        _submit_wbuf(e, p);

        p->wbuf->full = true;

    }

    if (!p->wbuf && p->allocated < e->page_size) {

        _allocate_wbuf(e, p);

    }

    // hand over buffer for caller to copy into

    // leaves p locked.

    if (p->wbuf && !p->wbuf->full && p->wbuf->free >= io->len) {

        io->buf = p->wbuf->buf_pos;

        io->page_id = p->id;

        return 0;

    }

    pthread_mutex_unlock(&p->mutex);

    // p->written is incremented post-wbuf flush

    return ret;

}

/* _must_ be called after a successful write_request.

 * fills the rest of io structure.

 */

void extstore_write(void *ptr, obj_io *io) {

    store_engine *e = (store_engine *)ptr;

    store_page *p = &e->pages[io->page_id];

    io->offset = p->wbuf->offset + (p->wbuf->size - p->wbuf->free);

    io->page_version = p->version;

    p->wbuf->buf_pos += io->len;

    p->wbuf->free -= io->len;

    p->bytes_used += io->len;

    p->obj_count++;

    STAT_L(e);

    e->stats.bytes_written += io->len;

    e->stats.bytes_used += io->len;

    e->stats.objects_written++;

    e->stats.objects_used++;

    STAT_UL(e);

    pthread_mutex_unlock(&p->mutex);

}

/* engine submit function; takes engine, item_io stack.

 * lock io_thread context and add stack?

 * signal io thread to wake.

 * return success.

 */

int extstore_submit(void *ptr, obj_io *io) {

    store_engine *e = (store_engine *)ptr;

    store_io_thread *t = _get_io_thread(e);

    pthread_mutex_lock(&t->mutex);

    if (t->queue == NULL) {

        t->queue = io;

    } else {

        /* Have to put the *io stack at the end of current queue.

         * FIXME: Optimize by tracking tail.

         */

        obj_io *tmp = t->queue;

        while (tmp->next != NULL) {

            tmp = tmp->next;

            assert(tmp != t->queue);

        }

        tmp->next = io;

    }

    // TODO: extstore_submit(ptr, io, count)

    obj_io *tio = io;

    while (tio != NULL) {

        t->depth++;

        tio = tio->next;

    }

    pthread_mutex_unlock(&t->mutex);

    //pthread_mutex_lock(&t->mutex);

    pthread_cond_signal(&t->cond);

    //pthread_mutex_unlock(&t->mutex);

    return 0;

}

/* engine note delete function: takes engine, page id, size?

 * note that an item in this page is no longer valid

 */

int extstore_delete(void *ptr, unsigned int page_id, uint64_t page_version,

        unsigned int count, unsigned int bytes) {

    store_engine *e = (store_engine *)ptr;

    // FIXME: validate page_id in bounds

    store_page *p = &e->pages[page_id];

    int ret = 0;

    pthread_mutex_lock(&p->mutex);

    if (!p->closed && p->version == page_version) {

        if (p->bytes_used >= bytes) {

            p->bytes_used -= bytes;

        } else {

            p->bytes_used = 0;

        }

        if (p->obj_count >= count) {

            p->obj_count -= count;

        } else {

            p->obj_count = 0; // caller has bad accounting?

        }

        STAT_L(e);

        e->stats.bytes_used -= bytes;

        e->stats.objects_used -= count;

        STAT_UL(e);

        if (p->obj_count == 0) {

            extstore_run_maint(e);

        }

    } else {

        ret = -1;

    }

    pthread_mutex_unlock(&p->mutex);

    return ret;

}

int extstore_check(void *ptr, unsigned int page_id, uint64_t page_version) {

    store_engine *e = (store_engine *)ptr;

    store_page *p = &e->pages[page_id];

    int ret = 0;

    pthread_mutex_lock(&p->mutex);

    if (p->version != page_version)

        ret = -1;

    pthread_mutex_unlock(&p->mutex);

    return ret;

}

/* allows a compactor to say "we're done with this page, kill it. */

void extstore_close_page(void *ptr, unsigned int page_id, uint64_t page_version) {

    store_engine *e = (store_engine *)ptr;

    store_page *p = &e->pages[page_id];

    pthread_mutex_lock(&p->mutex);

    if (!p->closed && p->version == page_version) {

        p->closed = true;

        extstore_run_maint(e);

    }

    pthread_mutex_unlock(&p->mutex);

}

/* Finds an attached wbuf that can satisfy the read.

 * Since wbufs can potentially be flushed to disk out of order, they are only

 * removed as the head of the list successfully flushes to disk.

 */

// call with *p locked

// FIXME: protect from reading past wbuf

static inline int _read_from_wbuf(store_page *p, obj_io *io) {

    _store_wbuf *wbuf = p->wbuf;

    assert(wbuf != NULL);

    assert(io->offset < p->written + wbuf->size);

    if (io->iov == NULL) {

        memcpy(io->buf, wbuf->buf + (io->offset - wbuf->offset), io->len);

    } else {

        int x;

        unsigned int off = io->offset - wbuf->offset;

        // need to loop fill iovecs

        for (x = 0; x < io->iovcnt; x++) {

            struct iovec *iov = &io->iov[x];

            memcpy(iov->iov_base, wbuf->buf + off, iov->iov_len);

            off += iov->iov_len;

        }

    }

    return io->len;

}

/* engine IO thread; takes engine context

 * manage writes/reads

 * runs IO callbacks inline after each IO

 */

// FIXME: protect from reading past page

static void *extstore_io_thread(void *arg) {

    store_io_thread *me = (store_io_thread *)arg;

    store_engine *e = me->e;

    while (1) {

        obj_io *io_stack = NULL;

        pthread_mutex_lock(&me->mutex);

        if (me->queue == NULL) {

            pthread_cond_wait(&me->cond, &me->mutex);

        }

        // Pull and disconnect a batch from the queue

        if (me->queue != NULL) {

            int i;

            obj_io *end = NULL;

            io_stack = me->queue;

            end = io_stack;

            for (i = 1; i < e->io_depth; i++) {

                if (end->next) {

                    end = end->next;

                } else {

                    break;

                }

            }

            me->depth -= i;

            me->queue = end->next;

            end->next = NULL;

        }

        pthread_mutex_unlock(&me->mutex);

        obj_io *cur_io = io_stack;

        while (cur_io) {

            // We need to note next before the callback in case the obj_io

            // gets reused.

            obj_io *next = cur_io->next;

            int ret = 0;

            int do_op = 1;

            store_page *p = &e->pages[cur_io->page_id];

            // TODO: loop if not enough bytes were read/written.

            switch (cur_io->mode) {

                case OBJ_IO_READ:

                    // Page is currently open. deal if read is past the end.

                    pthread_mutex_lock(&p->mutex);

                    if (!p->free && !p->closed && p->version == cur_io->page_version) {

                        if (p->active && cur_io->offset >= p->written) {

                            ret = _read_from_wbuf(p, cur_io);

                            do_op = 0;

                        } else {

                            p->refcount++;

                        }

                        STAT_L(e);

                        e->stats.bytes_read += cur_io->len;

                        e->stats.objects_read++;

                        STAT_UL(e);

                    } else {

                        do_op = 0;

                        ret = -2; // TODO: enum in IO for status?

                    }

                    pthread_mutex_unlock(&p->mutex);

                    if (do_op) {

#ifdef __APPLE__

                        ret = lseek(p->fd, SEEK_SET, p->offset + cur_io->offset);

                        if (ret >= 0) {

                            if (cur_io->iov == NULL) {

                                ret = read(p->fd, cur_io->buf, cur_io->len);

                            } else {

                                ret = readv(p->fd, cur_io->iov, cur_io->iovcnt);

                            }

                        }

#else

                        if (cur_io->iov == NULL) {

                            ret = pread(p->fd, cur_io->buf, cur_io->len, p->offset + cur_io->offset);

                        } else {

                            ret = preadv(p->fd, cur_io->iov, cur_io->iovcnt, p->offset + cur_io->offset);

                        }

#endif

                    }

                    break;

                case OBJ_IO_WRITE:

                    do_op = 0;

                    // FIXME: Should hold refcount during write. doesn't

                    // currently matter since page can't free while active.

                    ret = pwrite(p->fd, cur_io->buf, cur_io->len, p->offset + cur_io->offset);

                    break;

            }

            if (ret == 0) {

                E_DEBUG("read returned nothing\n");

            }

#ifdef EXTSTORE_DEBUG

            if (ret == -1) {

                perror("read/write op failed");

            }

#endif

            cur_io->cb(e, cur_io, ret);

            if (do_op) {

                pthread_mutex_lock(&p->mutex);

                p->refcount--;

                pthread_mutex_unlock(&p->mutex);

            }

            cur_io = next;

        }

    }

    return NULL;

}

// call with *p locked.

static void _free_page(store_engine *e, store_page *p) {

    store_page *tmp = NULL;

    store_page *prev = NULL;

    E_DEBUG("EXTSTORE: freeing page %u\n", p->id);

    STAT_L(e);

    e->stats.objects_used -= p->obj_count;

    e->stats.bytes_used -= p->bytes_used;

    e->stats.page_reclaims++;

    STAT_UL(e);

    pthread_mutex_lock(&e->mutex);

    // unlink page from bucket list

    tmp = e->page_buckets[p->bucket];

    while (tmp) {

        if (tmp == p) {

            if (prev) {

                prev->next = tmp->next;

            } else {

                e->page_buckets[p->bucket] = tmp->next;

            }

            tmp->next = NULL;

            break;

        }

        prev = tmp;

        tmp = tmp->next;

    }

    // reset most values

    p->version = 0;

    p->obj_count = 0;

    p->bytes_used = 0;

    p->allocated = 0;

    p->written = 0;

    p->bucket = 0;

    p->active = false;

    p->closed = false;

    p->free = true;

    // add to page stack

    // TODO: free_page_buckets first class and remove redundancy?

    if (p->free_bucket != 0) {

        p->next = e->free_page_buckets[p->free_bucket];

        e->free_page_buckets[p->free_bucket] = p;

    } else {

        p->next = e->page_freelist;

        e->page_freelist = p;

    }

    e->page_free++;

    pthread_mutex_unlock(&e->mutex);

}

/* engine maint thread; takes engine context.

 * Uses version to ensure oldest possible objects are being evicted.

 * Needs interface to inform owner of pages with fewer objects or most space

 * free, which can then be actively compacted to avoid eviction.

 *

 * This gets called asynchronously after every page allocation. Could run less

 * often if more pages are free.

 *

 * Another allocation call is required if an attempted free didn't happen