tip/ucx/buffer.c

/*
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
 *
 * Copyright 2021 Mike Becker, Olaf Wintermann All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *   1. Redistributions of source code must retain the above copyright
 *      notice, this list of conditions and the following disclaimer.
 *
 *   2. Redistributions in binary form must reproduce the above copyright
 *      notice, this list of conditions and the following disclaimer in the
 *      documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#include "cx/buffer.h"

#include <stdio.h>
#include <string.h>
#include <errno.h>

#ifdef _WIN32
#include <Windows.h>
#include <sysinfoapi.h>
static unsigned long system_page_size() {
    static unsigned long ps = 0;
    if (ps == 0) {
        SYSTEM_INFO sysinfo;
        GetSystemInfo(&sysinfo);
        ps = sysinfo.dwPageSize;
    }
    return ps;
}
#define SYSTEM_PAGE_SIZE system_page_size()
#else
#include <unistd.h>
#define SYSTEM_PAGE_SIZE sysconf(_SC_PAGESIZE)
#endif

static int buffer_copy_on_write(CxBuffer* buffer) {
    if (0 == (buffer->flags & CX_BUFFER_COPY_ON_WRITE)) return 0;
    void *newspace = cxMalloc(buffer->allocator, buffer->capacity);
    if (NULL == newspace) return -1;
    memcpy(newspace, buffer->space, buffer->size);
    buffer->space = newspace;
    buffer->flags &= ~CX_BUFFER_COPY_ON_WRITE;
    buffer->flags |= CX_BUFFER_FREE_CONTENTS;
    return 0;
}

int cxBufferInit(
        CxBuffer *buffer,
        void *space,
        size_t capacity,
        const CxAllocator *allocator,
        int flags
) {
    if (allocator == NULL) {
        allocator = cxDefaultAllocator;
    }
    if (flags & CX_BUFFER_COPY_ON_EXTEND) {
        flags |= CX_BUFFER_AUTO_EXTEND;
    }
    buffer->allocator = allocator;
    buffer->flags = flags;
    if (!space) {
        buffer->bytes = cxMalloc(allocator, capacity);
        if (buffer->bytes == NULL) {
            return -1; // LCOV_EXCL_LINE
        }
        buffer->flags |= CX_BUFFER_FREE_CONTENTS;
    } else {
        buffer->bytes = space;
    }
    buffer->capacity = capacity;
    buffer->size = 0;
    buffer->pos = 0;

    buffer->flush = NULL;

    return 0;
}

int cxBufferEnableFlushing(
    CxBuffer *buffer,
    CxBufferFlushConfig config
) {
    buffer->flush = cxMallocDefault(sizeof(CxBufferFlushConfig));
    if (buffer->flush == NULL) return -1; // LCOV_EXCL_LINE
    memcpy(buffer->flush, &config, sizeof(CxBufferFlushConfig));
    return 0;
}

void cxBufferDestroy(CxBuffer *buffer) {
    if (buffer->flags & CX_BUFFER_FREE_CONTENTS) {
        cxFree(buffer->allocator, buffer->bytes);
    }
    cxFreeDefault(buffer->flush);
    memset(buffer, 0, sizeof(CxBuffer));
}

CxBuffer *cxBufferCreate(
        void *space,
        size_t capacity,
        const CxAllocator *allocator,
        int flags
) {
    if (allocator == NULL) {
        allocator = cxDefaultAllocator;
    }
    CxBuffer *buf = cxMalloc(allocator, sizeof(CxBuffer));
    if (buf == NULL) return NULL;
    if (0 == cxBufferInit(buf, space, capacity, allocator, flags)) {
        return buf;
    } else {
        // LCOV_EXCL_START
        cxFree(allocator, buf);
        return NULL;
        // LCOV_EXCL_STOP
    }
}

void cxBufferFree(CxBuffer *buffer) {
    if (buffer == NULL) return;
    const CxAllocator *allocator = buffer->allocator;
    cxBufferDestroy(buffer);
    cxFree(allocator, buffer);
}

int cxBufferSeek(
        CxBuffer *buffer,
        off_t offset,
        int whence
) {
    size_t npos;
    switch (whence) {
        case SEEK_CUR:
            npos = buffer->pos;
            break;
        case SEEK_END:
            npos = buffer->size;
            break;
        case SEEK_SET:
            npos = 0;
            break;
        default:
            errno = EINVAL;
            return -1;
    }

    size_t opos = npos;
    npos += offset;

    if ((offset > 0 && npos < opos) || (offset < 0 && npos > opos)) {
        // to be compliant with fseek() specification
        // we return EINVAL on underflow
        errno = EINVAL;
        return -1;
    }

    if (npos > buffer->size) {
        // not compliant with fseek() specification
        // but this is the better behavior for CxBuffer
        errno = EINVAL;
        return -1;
    } else {
        buffer->pos = npos;
        return 0;
    }

}

void cxBufferClear(CxBuffer *buffer) {
    if (0 == (buffer->flags & CX_BUFFER_COPY_ON_WRITE)) {
        memset(buffer->bytes, 0, buffer->size);
    }
    buffer->size = 0;
    buffer->pos = 0;
}

void cxBufferReset(CxBuffer *buffer) {
    buffer->size = 0;
    buffer->pos = 0;
}

bool cxBufferEof(const CxBuffer *buffer) {
    return buffer->pos >= buffer->size;
}

int cxBufferMinimumCapacity(
        CxBuffer *buffer,
        size_t newcap
) {
    if (newcap <= buffer->capacity) {
        return 0;
    }

    unsigned long pagesize = SYSTEM_PAGE_SIZE;
    // if page size is larger than 64 KB - for some reason - truncate to 64 KB
    if (pagesize > 65536) pagesize = 65536;
    if (newcap < pagesize) {
        // when smaller as one page, map to the next power of two
        newcap--;
        newcap |= newcap >> 1;
        newcap |= newcap >> 2;
        newcap |= newcap >> 4;
        // last operation only needed for pages larger 4096 bytes
        // but if/else would be more expensive than just doing this
        newcap |= newcap >> 8;
        newcap++;
    } else {
        // otherwise, map to a multiple of the page size
        newcap -= newcap % pagesize;
        newcap += pagesize;
        // note: if newcap is already page aligned,
        // this gives a full additional page (which is good)
    }


    const int force_copy_flags = CX_BUFFER_COPY_ON_WRITE | CX_BUFFER_COPY_ON_EXTEND;
    if (buffer->flags & force_copy_flags) {
        void *newspace = cxMalloc(buffer->allocator, newcap);
        if (NULL == newspace) return -1;
        memcpy(newspace, buffer->space, buffer->size);
        buffer->space = newspace;
        buffer->capacity = newcap;
        buffer->flags &= ~force_copy_flags;
        buffer->flags |= CX_BUFFER_FREE_CONTENTS;
        return 0;
    } else if (cxReallocate(buffer->allocator,
                     (void **) &buffer->bytes, newcap) == 0) {
        buffer->capacity = newcap;
        return 0;
    } else {
        return -1; // LCOV_EXCL_LINE
    }
}

void cxBufferShrink(
        CxBuffer *buffer,
        size_t reserve
) {
    // Ensure buffer is in a reallocatable state
    const int force_copy_flags = CX_BUFFER_COPY_ON_WRITE | CX_BUFFER_COPY_ON_EXTEND;
    if (buffer->flags & force_copy_flags) {
        // do nothing when we are not allowed to reallocate
        return;
    }

    // calculate new capacity
    size_t newCapacity = buffer->size + reserve;

    // If new capacity is smaller than current capacity, resize the buffer
    if (newCapacity < buffer->capacity) {
        if (0 == cxReallocate(buffer->allocator, &buffer->bytes, newCapacity)) {
            buffer->capacity = newCapacity;
        }
    }
}

static size_t cx_buffer_flush_helper(
        const CxBuffer *buffer,
        const unsigned char *src,
        size_t size,
        size_t nitems
) {
    // flush data from an arbitrary source
    // does not need to be the buffer's contents
    size_t max_items = buffer->flush->blksize / size;
    size_t fblocks = 0;
    size_t flushed_total = 0;
    while (nitems > 0 && fblocks < buffer->flush->blkmax) {
        fblocks++;
        size_t items = nitems > max_items ? max_items : nitems;
        size_t flushed = buffer->flush->wfunc(
            src, size, items, buffer->flush->target);
        if (flushed > 0) {
            flushed_total += flushed;
            src += flushed * size;
            nitems -= flushed;
        } else {
            // if no bytes can be flushed out anymore, we give up
            break;
        }
    }
    return flushed_total;
}

static size_t cx_buffer_flush_impl(CxBuffer *buffer, size_t size) {
    // flush the current contents of the buffer
    unsigned char *space = buffer->bytes;
    size_t remaining = buffer->pos / size;
    size_t flushed_total = cx_buffer_flush_helper(
        buffer, space, size, remaining);

    // shift the buffer left after flushing
    // IMPORTANT: up to this point, copy on write must have been
    // performed already, because we can't do error handling here
    cxBufferShiftLeft(buffer, flushed_total*size);

    return flushed_total;
}

size_t cxBufferFlush(CxBuffer *buffer) {
    if (buffer_copy_on_write(buffer)) return 0;
    return cx_buffer_flush_impl(buffer, 1);
}

size_t cxBufferWrite(
        const void *ptr,
        size_t size,
        size_t nitems,
        CxBuffer *buffer
) {
    // optimize for easy case
    if (size == 1 && (buffer->capacity - buffer->pos) >= nitems) {
        if (buffer_copy_on_write(buffer)) return 0;
        memcpy(buffer->bytes + buffer->pos, ptr, nitems);
        buffer->pos += nitems;
        if (buffer->pos > buffer->size) {
            buffer->size = buffer->pos;
        }
        return nitems;
    }

    size_t len, total_flushed = 0;
cx_buffer_write_retry:
    if (cx_szmul(size, nitems, &len)) {
        errno = EOVERFLOW;
        return total_flushed;
    }
    if (buffer->pos > SIZE_MAX - len) {
        errno = EOVERFLOW;
        return total_flushed;
    }

    size_t required = buffer->pos + len;
    bool perform_flush = false;
    if (required > buffer->capacity) {
        if (buffer->flags & CX_BUFFER_AUTO_EXTEND) {
            if (buffer->flush != NULL && required > buffer->flush->threshold) {
                perform_flush = true;
            } else {
                if (cxBufferMinimumCapacity(buffer, required)) {
                    return total_flushed; // LCOV_EXCL_LINE
                }
            }
        } else {
            if (buffer->flush != NULL) {
                perform_flush = true;
            } else {
                // truncate data, if we can neither extend nor flush
                len = buffer->capacity - buffer->pos;
                if (size > 1) {
                    len -= len % size;
                }
                nitems = len / size;
            }
        }
    }

    // check here and not above because of possible truncation
    if (len == 0) {
        return total_flushed;
    }

    // check if we need to copy
    if (buffer_copy_on_write(buffer)) return 0;

    // perform the operation
    if (perform_flush) {
        size_t items_flushed;
        if (buffer->pos == 0) {
            // if we don't have data in the buffer, but are instructed
            // to flush, it means that we are supposed to relay the data
            items_flushed = cx_buffer_flush_helper(buffer, ptr, size, nitems);
            if (items_flushed == 0) {
                // we needed to relay data, but could not flush anything
                // i.e. we have to give up to avoid endless trying
                return 0;
            }
            nitems -= items_flushed;
            total_flushed += items_flushed;
            if (nitems > 0) {
                ptr = ((unsigned char*)ptr) + items_flushed * size;
                goto cx_buffer_write_retry;
            }
            return total_flushed;
        } else {
            items_flushed = cx_buffer_flush_impl(buffer, size);
            if (items_flushed == 0) {
                // flush target is full, let's try to truncate
                size_t remaining_space;
                if (buffer->flags & CX_BUFFER_AUTO_EXTEND) {
                    remaining_space = buffer->flush->threshold > buffer->pos
                                          ? buffer->flush->threshold - buffer->pos
                                          : 0;
                } else {
                    remaining_space = buffer->capacity > buffer->pos
                                          ? buffer->capacity - buffer->pos
                                          : 0;
                }
                nitems = remaining_space / size;
                if (nitems == 0) {
                    return total_flushed;
                }
            }
            goto cx_buffer_write_retry;
        }
    } else {
        memcpy(buffer->bytes + buffer->pos, ptr, len);
        buffer->pos += len;
        if (buffer->pos > buffer->size) {
            buffer->size = buffer->pos;
        }
        return total_flushed + nitems;
    }
}

size_t cxBufferAppend(
        const void *ptr,
        size_t size,
        size_t nitems,
        CxBuffer *buffer
) {
    size_t pos = buffer->pos;
    size_t append_pos = buffer->size;
    buffer->pos = append_pos;    
    size_t written = cxBufferWrite(ptr, size, nitems, buffer);
    // the buffer might have been flushed
    // we must compute a possible delta for the position
    // expected: pos = append_pos + written
    // -> if this is not the case, there is a delta
    size_t delta = append_pos + written*size - buffer->pos;
    if (delta > pos) {
        buffer->pos = 0;
    } else {
        buffer->pos = pos - delta;
    }
    return written;
}

int cxBufferPut(
        CxBuffer *buffer,
        int c
) {
    c &= 0xFF;
    unsigned char const ch = c;
    if (cxBufferWrite(&ch, 1, 1, buffer) == 1) {
        return c;
    } else {
        return EOF;
    }
}

int cxBufferTerminate(CxBuffer *buffer) {
    if (0 == cxBufferPut(buffer, 0)) {
        buffer->size = buffer->pos - 1;
        return 0;
    } else {
        return -1;
    }
}

size_t cxBufferPutString(
        CxBuffer *buffer,
        const char *str
) {
    return cxBufferWrite(str, 1, strlen(str), buffer);
}

size_t cxBufferRead(
        void *ptr,
        size_t size,
        size_t nitems,
        CxBuffer *buffer
) {
    size_t len;
    if (cx_szmul(size, nitems, &len)) {
        errno = EOVERFLOW;
        return 0;
    }
    if (buffer->pos + len > buffer->size) {
        len = buffer->size - buffer->pos;
        if (size > 1) len -= len % size;
    }

    if (len <= 0) {
        return len;
    }

    memcpy(ptr, buffer->bytes + buffer->pos, len);
    buffer->pos += len;

    return len / size;
}

int cxBufferGet(CxBuffer *buffer) {
    if (cxBufferEof(buffer)) {
        return EOF;
    } else {
        int c = buffer->bytes[buffer->pos];
        buffer->pos++;
        return c;
    }
}

int cxBufferShiftLeft(
        CxBuffer *buffer,
        size_t shift
) {
    if (shift >= buffer->size) {
        buffer->pos = buffer->size = 0;
    } else {
        if (buffer_copy_on_write(buffer)) return -1;
        memmove(buffer->bytes, buffer->bytes + shift, buffer->size - shift);
        buffer->size -= shift;

        if (buffer->pos >= shift) {
            buffer->pos -= shift;
        } else {
            buffer->pos = 0;
        }
    }
    return 0;
}

int cxBufferShiftRight(
        CxBuffer *buffer,
        size_t shift
) {
    if (buffer->size > SIZE_MAX - shift) {
        errno = EOVERFLOW;
        return -1;
    }
    size_t req_capacity = buffer->size + shift;
    size_t movebytes;

    // auto extend buffer, if required and enabled
    if (buffer->capacity < req_capacity) {
        if (buffer->flags & CX_BUFFER_AUTO_EXTEND) {
            if (cxBufferMinimumCapacity(buffer, req_capacity)) {
                return -1; // LCOV_EXCL_LINE
            }
            movebytes = buffer->size;
        } else {
            movebytes = buffer->capacity - shift;
        }
    } else {
        movebytes = buffer->size;
    }

    if (movebytes > 0) {
        if (buffer_copy_on_write(buffer)) return -1;
        memmove(buffer->bytes + shift, buffer->bytes, movebytes);
        buffer->size = shift + movebytes;
    }

    buffer->pos += shift;
    if (buffer->pos > buffer->size) {
        buffer->pos = buffer->size;
    }

    return 0;
}

int cxBufferShift(
        CxBuffer *buffer,
        off_t shift
) {
    if (shift < 0) {
        return cxBufferShiftLeft(buffer, (size_t) (-shift));
    } else if (shift > 0) {
        return cxBufferShiftRight(buffer, (size_t) shift);
    } else {
        return 0;
    }
}