From a48a68cfb04b2020737c0adfb0e7667451a22b5c Mon Sep 17 00:00:00 2001 From: Andrew Opalach Date: Mon, 6 Nov 2023 11:54:08 -0500 Subject: Add camu - Subprojects temporarily omitted Signed-off-by: Andrew Opalach --- src/cache/backing.h | 20 ++++ src/cache/backings/file.c | 75 ++++++++++++++ src/cache/backings/file.h | 22 +++++ src/cache/backings/file_common.c | 23 +++++ src/cache/backings/file_mapped.c | 81 +++++++++++++++ src/cache/backings/memory.c | 82 ++++++++++++++++ src/cache/backings/memory.h | 16 +++ src/cache/entry.c | 44 +++++++++ src/cache/entry.h | 21 ++++ src/cache/handle.c | 100 +++++++++++++++++++ src/cache/handle.h | 20 ++++ src/cache/handler.h | 23 +++++ src/cache/handlers/file.c | 49 +++++++++ src/cache/handlers/file.h | 12 +++ src/cache/handlers/http.c | 207 +++++++++++++++++++++++++++++++++++++++ src/cache/handlers/http.h | 24 +++++ src/cache/meson.build | 18 ++++ 17 files changed, 837 insertions(+) create mode 100644 src/cache/backing.h create mode 100644 src/cache/backings/file.c create mode 100644 src/cache/backings/file.h create mode 100644 src/cache/backings/file_common.c create mode 100644 src/cache/backings/file_mapped.c create mode 100644 src/cache/backings/memory.c create mode 100644 src/cache/backings/memory.h create mode 100644 src/cache/entry.c create mode 100644 src/cache/entry.h create mode 100644 src/cache/handle.c create mode 100644 src/cache/handle.h create mode 100644 src/cache/handler.h create mode 100644 src/cache/handlers/file.c create mode 100644 src/cache/handlers/file.h create mode 100644 src/cache/handlers/http.c create mode 100644 src/cache/handlers/http.h create mode 100644 src/cache/meson.build (limited to 'src/cache') diff --git a/src/cache/backing.h b/src/cache/backing.h new file mode 100644 index 0000000..24aff44 --- /dev/null +++ b/src/cache/backing.h @@ -0,0 +1,20 @@ +#pragma once + +#include + +enum { + CCH_BACKING_READ = 0, + CCH_BACKING_MAPPED +}; + +struct cch_backing { + u8 mode; + void (*write)(struct cch_backing *, u8 *, off_t, size_t *); + void (*read)(struct cch_backing *, u8 *, off_t, size_t *); + u8 *(*get_ptr)(struct cch_backing *, off_t, size_t *); + void (*unlock)(struct cch_backing *); + void *(*get_mmap)(struct cch_backing *); + off_t (*get_size_estimate)(struct cch_backing *); + void (*resize)(struct cch_backing *, size_t); + void (*free)(struct cch_backing **); +}; diff --git a/src/cache/backings/file.c b/src/cache/backings/file.c new file mode 100644 index 0000000..4c5f344 --- /dev/null +++ b/src/cache/backings/file.c @@ -0,0 +1,75 @@ +#include "file.h" +#include "file_common.c" + +static void file_backing_write(struct cch_backing *backing, u8 *buf, off_t index, size_t *size) +{ + struct cch_backing_file *file = (struct cch_backing_file *)backing; + aki_mutex_lock(&file->mutex); + if (file->u.pointer != index) { + file->u.pointer = aki_file_seek(&file->file, index, SEEK_SET); + al_assert(file->u.pointer == index); + } + if (file->size >= 0 && index + ((off_t)*size) >= file->size) { + *size = file->size - index; + } + aki_file_write(&file->file, buf, *size); + file->u.pointer += *size; + aki_mutex_unlock(&file->mutex); +} + +static void file_backing_read(struct cch_backing *backing, u8 *buf, off_t index, size_t *size) +{ + struct cch_backing_file *file = (struct cch_backing_file *)backing; + aki_mutex_lock(&file->mutex); + if (file->u.pointer != index) { + file->u.pointer = aki_file_seek(&file->file, index, SEEK_SET); + al_assert(file->u.pointer == index); + } + if (file->size >= 0 && index + ((off_t)*size) >= file->size) { + *size = file->size - index; + } + aki_file_read(&file->file, buf, *size); + file->u.pointer += *size; + aki_mutex_unlock(&file->mutex); +} + +static void file_backing_resize(struct cch_backing *backing, size_t size) +{ + struct cch_backing_file *file = (struct cch_backing_file *)backing; + aki_mutex_lock(&file->mutex); + file->size = size; + if ((off_t)size > file->filesize) { + aki_file_truncate(&file->file, size); + file->filesize = size; + } + aki_mutex_unlock(&file->mutex); +} + +static void file_backing_free(struct cch_backing **backing) +{ + struct cch_backing_file *file = (struct cch_backing_file *)*backing; + aki_file_close(&file->file); + al_free(file); + *backing = NULL; +} + +struct cch_backing *cch_backing_file_create(str *path, size_t size) +{ + struct cch_backing_file *file = al_alloc_object(struct cch_backing_file); + file->backing.mode = CCH_BACKING_READ; + file->backing.write = file_backing_write; + file->backing.read = file_backing_read; + file->backing.get_ptr = NULL; + file->backing.unlock = NULL; + file->backing.get_mmap = NULL; + file->backing.get_size_estimate = file_backing_get_size_estimate; + file->backing.resize = file_backing_resize; + file->backing.free = file_backing_free; + if (!file_open_internal(file, path, size)) { + al_free(file); + return NULL; + } + file->u.pointer = 0; + aki_mutex_init(&file->mutex); + return (struct cch_backing *)file; +} diff --git a/src/cache/backings/file.h b/src/cache/backings/file.h new file mode 100644 index 0000000..9794bbb --- /dev/null +++ b/src/cache/backings/file.h @@ -0,0 +1,22 @@ +#pragma once + +#include +#include +#include +#include + +#include "../backing.h" + +struct cch_backing_file { + struct cch_backing backing; + struct aki_file file; + off_t size; + off_t filesize; + union { + void *map; // CCH_BACKING_MAPPED + off_t pointer; // CCH_BACKING_READ + } u; + struct aki_mutex mutex; +}; + +struct cch_backing *cch_backing_file_create(str *path, size_t filesize); diff --git a/src/cache/backings/file_common.c b/src/cache/backings/file_common.c new file mode 100644 index 0000000..c0b54c0 --- /dev/null +++ b/src/cache/backings/file_common.c @@ -0,0 +1,23 @@ +static off_t file_backing_get_size_estimate(struct cch_backing *backing) +{ + struct cch_backing_file *file = (struct cch_backing_file *)backing; + aki_mutex_lock(&file->mutex); + off_t size = file->size; + aki_mutex_unlock(&file->mutex); + return size; +} + +static bool file_open_internal(struct cch_backing_file *file, str *path, size_t size) +{ + if (!aki_file_open(&file->file, path, size != 0)) return false; + file->filesize = aki_file_get_filesize(&file->file); + if (!size) { + file->size = file->filesize; + } else { + file->size = -1; + if (file->filesize < (off_t)size) { + aki_file_truncate(&file->file, size); + } + } + return true; +} diff --git a/src/cache/backings/file_mapped.c b/src/cache/backings/file_mapped.c new file mode 100644 index 0000000..1b06ad2 --- /dev/null +++ b/src/cache/backings/file_mapped.c @@ -0,0 +1,81 @@ +#include "file.h" +#include "file_common.c" + +static void file_backing_write(struct cch_backing *backing, u8 *buf, off_t index, size_t *size) +{ + struct cch_backing_file *file = (struct cch_backing_file *)backing; + aki_mutex_lock(&file->mutex); + if (file->size >= 0 && index + ((off_t)*size) >= file->size) { + *size = file->size - index; + } + al_memcpy(file->u.map + index, buf, *size); + aki_mutex_unlock(&file->mutex); +} + +static u8 *file_backing_get_ptr(struct cch_backing *backing, off_t index, size_t *size) +{ + struct cch_backing_file *file = (struct cch_backing_file *)backing; + aki_mutex_lock(&file->mutex); + if (file->size >= 0 && index + ((off_t)*size) >= file->size) { + *size = file->size - index; + } + return (u8 *)(file->u.map + index); +} + +static void file_backing_unlock(struct cch_backing *backing) +{ + struct cch_backing_file *file = (struct cch_backing_file *)backing; + aki_mutex_unlock(&file->mutex); +} + +static void *file_backing_get_mmap(struct cch_backing *backing) +{ + struct cch_backing_file *file = (struct cch_backing_file *)backing; + aki_mutex_lock(&file->mutex); + void *map = file->u.map; + aki_mutex_unlock(&file->mutex); + return map; +} + +static void file_backing_resize(struct cch_backing *backing, size_t size) +{ + struct cch_backing_file *file = (struct cch_backing_file *)backing; + aki_mutex_lock(&file->mutex); + file->size = size; + if ((off_t)size > file->filesize) { + file->u.map = aki_file_mremap(&file->file, size, file->u.map); + aki_file_truncate(&file->file, size); + file->filesize = size; + } + aki_mutex_unlock(&file->mutex); +} + +static void file_backing_free(struct cch_backing **backing) +{ + struct cch_backing_file *file = (struct cch_backing_file *)*backing; + aki_file_munmap(&file->file, file->u.map); + aki_file_close(&file->file); + al_free(file); + *backing = NULL; +} + +struct cch_backing *cch_backing_file_create(str *path, size_t size) +{ + struct cch_backing_file *file = al_alloc_object(struct cch_backing_file); + file->backing.mode = CCH_BACKING_MAPPED; + file->backing.write = file_backing_write; + file->backing.read = NULL; + file->backing.get_mmap = file_backing_get_mmap; + file->backing.get_ptr = file_backing_get_ptr; + file->backing.unlock = file_backing_unlock; + file->backing.get_size_estimate = file_backing_get_size_estimate; + file->backing.resize = file_backing_resize; + file->backing.free = file_backing_free; + if (!file_open_internal(file, path, size)) { + al_free(file); + return NULL; + } + file->u.map = aki_file_mmap(&file->file); + aki_mutex_init(&file->mutex); + return (struct cch_backing *)file; +} diff --git a/src/cache/backings/memory.c b/src/cache/backings/memory.c new file mode 100644 index 0000000..194d66c --- /dev/null +++ b/src/cache/backings/memory.c @@ -0,0 +1,82 @@ +#include "memory.h" + +static void ensure_alloced(struct cch_backing_memory *mem, size_t size) +{ + if (size > mem->alloc) { + mem->data = al_realloc(mem->data, size); + mem->alloc = size; + } +} + +static void memory_backing_write(struct cch_backing *backing, u8 *buf, off_t index, size_t *size) +{ + struct cch_backing_memory *mem = (struct cch_backing_memory *)backing; + aki_mutex_lock(&mem->mutex); + if (mem->size >= 0 && index + ((off_t)*size) >= mem->size) { + *size = mem->size - index; + } + ensure_alloced(mem, index + *size); + al_memcpy(mem->data + index, buf, *size); + aki_mutex_unlock(&mem->mutex); +} + +static u8 *memory_backing_get_ptr(struct cch_backing *backing, off_t index, size_t *size) +{ + struct cch_backing_memory *mem = (struct cch_backing_memory *)backing; + aki_mutex_lock(&mem->mutex); + if (mem->size >= 0 && index + ((off_t)*size) >= mem->size) { + *size = mem->size - index; + } + return (u8 *)(mem->data + index); +} + +static void memory_backing_unlock(struct cch_backing *backing) +{ + struct cch_backing_memory *mem = (struct cch_backing_memory *)backing; + aki_mutex_unlock(&mem->mutex); +} + +static off_t memory_backing_get_size_estimate(struct cch_backing *backing) +{ + struct cch_backing_memory *mem = (struct cch_backing_memory *)backing; + aki_mutex_lock(&mem->mutex); + off_t size = mem->size; + aki_mutex_unlock(&mem->mutex); + return size; +} + +static void memory_backing_resize(struct cch_backing *backing, size_t size) +{ + struct cch_backing_memory *mem = (struct cch_backing_memory *)backing; + aki_mutex_lock(&mem->mutex); + ensure_alloced(mem, size); + mem->size = size; + aki_mutex_unlock(&mem->mutex); +} + +static void memory_backing_free(struct cch_backing **backing) +{ + struct cch_backing_memory *mem = (struct cch_backing_memory *)*backing; + al_free(mem->data); + aki_mutex_destroy(&mem->mutex); + al_free(mem); + *backing = NULL; +} + +struct cch_backing *cch_backing_memory_create(size_t size) +{ + struct cch_backing_memory *mem = al_alloc_object(struct cch_backing_memory); + mem->backing.mode = CCH_BACKING_MAPPED; + mem->backing.write = memory_backing_write; + mem->backing.get_mmap = NULL; + mem->backing.get_ptr = memory_backing_get_ptr; + mem->backing.unlock = memory_backing_unlock; + mem->backing.get_size_estimate = memory_backing_get_size_estimate; + mem->backing.resize = memory_backing_resize; + mem->backing.free = memory_backing_free; + mem->data = al_malloc(size); + mem->alloc = size; + mem->size = -1; + aki_mutex_init(&mem->mutex); + return (struct cch_backing *)mem; +} diff --git a/src/cache/backings/memory.h b/src/cache/backings/memory.h new file mode 100644 index 0000000..107ae4b --- /dev/null +++ b/src/cache/backings/memory.h @@ -0,0 +1,16 @@ +#pragma once + +#include +#include + +#include "../backing.h" + +struct cch_backing_memory { + struct cch_backing backing; + u8 *data; + size_t alloc; + off_t size; + struct aki_mutex mutex; +}; + +struct cch_backing *cch_backing_memory_create(size_t size); diff --git a/src/cache/entry.c b/src/cache/entry.c new file mode 100644 index 0000000..04e8f92 --- /dev/null +++ b/src/cache/entry.c @@ -0,0 +1,44 @@ +#include + +#include "entry.h" + +bool cch_entry_get_handle(struct cch_entry *entry, struct cch_handle *handle) +{ + handle->entry = entry; + handle->pointer = 0; + handle->prev_pointer = 0; + aki_cond_init(&handle->wait.cond); + aki_mutex_init(&handle->wait.mutex); + handle->wait.disabled = false; + aki_mutex_lock(&entry->mutex); + entry->ref_count++; + aki_mutex_unlock(&entry->mutex); + return true; +} + +void cch_entry_set_size(struct cch_entry *entry, off_t filesize) +{ + return entry->backing->resize(entry->backing, filesize); +} + +off_t cch_entry_get_size(struct cch_entry *entry) +{ + return entry->backing->get_size_estimate(entry->backing); +} + +void cch_entry_return_handle(struct cch_entry *entry, struct cch_handle *handle) +{ + handle->entry = NULL; + entry->ref_count--; + aki_mutex_destroy(&handle->wait.mutex); + aki_cond_destroy(&handle->wait.cond); +} + +void cch_entry_free(struct cch_entry **entry) +{ + (*entry)->handler->free(&(*entry)->handler); + (*entry)->backing->free(&(*entry)->backing); + aki_mutex_destroy(&(*entry)->mutex); + al_free(*entry); + *entry = NULL; +} diff --git a/src/cache/entry.h b/src/cache/entry.h new file mode 100644 index 0000000..305fb1f --- /dev/null +++ b/src/cache/entry.h @@ -0,0 +1,21 @@ +#pragma once + +#include + +#include "handle.h" +#include "backing.h" +#include "handler.h" + +struct cch_entry { + u32 hash; + s32 ref_count; + struct cch_backing *backing; + struct cch_handler *handler; + struct aki_mutex mutex; +}; + +bool cch_entry_get_handle(struct cch_entry *entry, struct cch_handle *handle); +void cch_entry_set_size(struct cch_entry *entry, off_t filesize); +off_t cch_entry_get_size(struct cch_entry *entry); +void cch_entry_return_handle(struct cch_entry *entry, struct cch_handle *handle); +void cch_entry_free(struct cch_entry **entry); diff --git a/src/cache/handle.c b/src/cache/handle.c new file mode 100644 index 0000000..9dfaf70 --- /dev/null +++ b/src/cache/handle.c @@ -0,0 +1,100 @@ +#include +#include +#include + +#include "../codec/codec.h" + +#include "handle.h" +#include "entry.h" + +static bool wait_for_size(struct cch_handle *handle, struct cch_handler *handler) +{ + handle->wait.id = al_rand_u16(); + handle->wait.end = -1; + return handler->wait_for_range(handler, &handle->wait); +} + +s32 cch_handle_read(struct cch_handle *handle, u8 *buf, s32 size) +{ + struct cch_handler *handler = handle->entry->handler; + wait_for_size(handle, handler); + off_t filesize = cch_entry_get_size(handle->entry); + if (filesize >= 0) { + if (handle->pointer >= filesize) return CAMU_ERR_EOF; + if (handle->pointer + size >= filesize) { + size = filesize - handle->pointer; + } + } + handle->wait.id = al_rand_u16(); + handle->wait.start = handle->pointer; + handle->wait.end = handle->pointer + size; + if (!handler->wait_for_range(handler, &handle->wait)) { + return CAMU_ERR_EOF; + } + struct cch_backing *backing = handle->entry->backing; + size_t available = (size_t)size; + if (backing->mode == CCH_BACKING_MAPPED) { + u8 *ptr = backing->get_ptr(backing, handle->pointer, &available); + if (ptr && available > 0) { + al_memcpy(buf, ptr, available); + } + backing->unlock(backing); + } else if (backing->mode == CCH_BACKING_READ) { + backing->read(backing, buf, handle->pointer, &available); + } + handle->pointer += available; + return (available > 0) ? (s32)available : CAMU_ERR_EOF; +} + +off_t cch_handle_seek(struct cch_handle *handle, off_t offset, s32 whence) +{ + struct cch_handler *handler = handle->entry->handler; + wait_for_size(handle, handler); + off_t filesize = cch_entry_get_size(handle->entry); + if (filesize < 0) return -1; + if (whence == CAMU_SEEK_SIZE) { + al_log_debug("cache_handle", "seek_size"); + return filesize; + } + switch (whence) { + case SEEK_SET: + if (offset >= filesize || offset < 0) { + return -1; + } + handle->pointer = offset; + al_log_debug("cache_handle", "seek_set %li", offset); + break; + case SEEK_CUR: + if (handle->pointer + offset >= filesize || handle->pointer + offset < 0) { + return -1; + } + handle->pointer += offset; + al_log_debug("cache_handle", "seek_cur %li", offset); + break; + case SEEK_END: + handle->prev_pointer = handle->pointer; + handle->pointer = filesize + offset; + al_log_debug("cache_handle", "seek_end %li", offset); + break; + default: + return -1; + } + handle->prev_pointer = -1; + return handle->pointer; +} + +bool cch_handle_can_seek(struct cch_handle *handle) +{ + struct cch_handler *handler = handle->entry->handler; + return handler->can_seek(handler); +} + +void cch_handle_disable(struct cch_handle *handle) +{ + aki_mutex_lock(&handle->wait.mutex); + handle->wait.disabled = true; + if (aki_cond_is_waiting(&handle->wait.cond)) { + aki_cond_signal(&handle->wait.cond); + } + aki_mutex_unlock(&handle->wait.mutex); +} diff --git a/src/cache/handle.h b/src/cache/handle.h new file mode 100644 index 0000000..55223aa --- /dev/null +++ b/src/cache/handle.h @@ -0,0 +1,20 @@ +#pragma once + +#include +#include + +#include "handler.h" + +struct cch_entry; +struct cch_handle { + struct cch_entry *entry; + off_t pointer; + // Set after a seek_end and reset after any seek_cur or seek_set. + off_t prev_pointer; + struct cch_handler_wait wait; +}; + +s32 cch_handle_read(struct cch_handle *handle, u8 *buf, s32 size); +off_t cch_handle_seek(struct cch_handle *handle, off_t offset, s32 whence); +bool cch_handle_can_seek(struct cch_handle *handle); +void cch_handle_disable(struct cch_handle *handle); diff --git a/src/cache/handler.h b/src/cache/handler.h new file mode 100644 index 0000000..5a2861e --- /dev/null +++ b/src/cache/handler.h @@ -0,0 +1,23 @@ +#pragma once + +#include +#include +#include + +struct cch_handler_wait { + u16 id; + off_t start, end; + struct aki_cond cond; + struct aki_mutex mutex; + bool disabled; +}; + +struct cch_handler { + bool (*can_seek)(struct cch_handler *); + void (*maybe_spawn_worker)(struct cch_handler *, size_t); + bool (*wait_for_range)(struct cch_handler *, struct cch_handler_wait *); + void (*free)(struct cch_handler **); + struct cch_entry *entry; + struct cch_backing *backing; + array(struct cch_handler_wait *) waits; +}; diff --git a/src/cache/handlers/file.c b/src/cache/handlers/file.c new file mode 100644 index 0000000..ec6a146 --- /dev/null +++ b/src/cache/handlers/file.c @@ -0,0 +1,49 @@ +#include "../backings/file.h" + +#include "file.h" + +static bool handler_file_can_seek(struct cch_handler *handler) +{ + (void)handler; + return true; +} + +static void handler_file_maybe_spawn_worker(struct cch_handler *handler, size_t index) +{ + struct cch_handler_file *file = (struct cch_handler_file *)handler; + (void)file; + (void)index; +} + +static bool handler_file_wait_for_range(struct cch_handler *handler, struct cch_handler_wait *wait) +{ + (void)handler; + (void)wait; + return true; +} + +static void handler_file_free(struct cch_handler **handler) +{ + struct cch_handler_file *file = (struct cch_handler_file *)*handler; + al_free(file); + *handler = NULL; +} + +struct cch_entry *cch_handler_file_create(str *path) +{ + struct cch_backing *backing = cch_backing_file_create(path, 0); + if (!backing) return NULL; + struct cch_entry *entry = al_alloc_object(struct cch_entry); + entry->backing = backing; + entry->ref_count = 0; + entry->hash = 0; + entry->handler = (struct cch_handler *)al_alloc_object(struct cch_handler_file); + aki_mutex_init(&entry->mutex); + entry->handler->can_seek = handler_file_can_seek; + entry->handler->maybe_spawn_worker = handler_file_maybe_spawn_worker; + entry->handler->wait_for_range = handler_file_wait_for_range; + entry->handler->free = handler_file_free; + entry->handler->entry = entry; + entry->handler->backing = entry->backing; + return entry; +} diff --git a/src/cache/handlers/file.h b/src/cache/handlers/file.h new file mode 100644 index 0000000..52e5c94 --- /dev/null +++ b/src/cache/handlers/file.h @@ -0,0 +1,12 @@ +#pragma once + +#include + +#include "../handler.h" +#include "../entry.h" + +struct cch_handler_file { + struct cch_handler handler; +}; + +struct cch_entry *cch_handler_file_create(str *path); diff --git a/src/cache/handlers/http.c b/src/cache/handlers/http.c new file mode 100644 index 0000000..1daf231 --- /dev/null +++ b/src/cache/handlers/http.c @@ -0,0 +1,207 @@ +#include +#include + +#include "../backings/file.h" +#include "../backings/memory.h" + +#include "http.h" + +#define USER_AGENT al_str_c("Mozilla/5.0 (X11; Linux x86_64; rv:96.0) Gecko/20100101 Firefox/96.0") + +static bool handler_http_can_seek(struct cch_handler *handler) +{ + (void)handler; + return false; +} + +static size_t http_callback(void *userdata, u8 op, u8 *buf, s64 int0) +{ + struct cch_handler_http *http = (struct cch_handler_http *)userdata; + size_t ret = (size_t)int0; + switch (op) { + case AKI_HTTP_FINISHED: { + al_log_debug("cache_handler_http", "Transfer finished."); + break; + } + case AKI_HTTP_ERROR: { + al_log_debug("cache_handler_http", "Error."); + break; // Unhandled. + } + case AKI_HTTP_RESPONSE_CODE: { + al_log_debug("cache_handler_http", "HTTP %ld.", int0); + break; + } + case AKI_HTTP_REDIRECT: { + al_log_debug("cache_handler_http", "Redirect."); + break; + } + case AKI_HTTP_CONTENT_LENGTH: { + off_t content_length = (off_t)int0; + al_log_debug("cache_handler_http", "Content-Length: %lld.", content_length); + aki_mutex_lock(&http->mutex); + cch_entry_set_size(http->handler.entry, content_length); + struct cch_handler_wait *wait; + al_array_foreach(http->handler.waits, i, wait) { + if (wait->end < 0) { + aki_mutex_lock(&wait->mutex); + aki_cond_signal(&wait->cond); + al_array_remove_at_iter(http->handler.waits, i); + aki_mutex_unlock(&wait->mutex); + } + } + aki_mutex_unlock(&http->mutex); + break; + } + case AKI_HTTP_WRITE: { + http->handler.backing->write(http->handler.backing, buf, http->pointer, &ret); + aki_mutex_lock(&http->mutex); + http->pointer += ret; + struct cch_handler_wait *wait; + al_array_foreach(http->handler.waits, i, wait) { + aki_mutex_lock(&wait->mutex); + // Assume that if we get data before a length, we won't get a length. + // So, signal the waiter (wait_for_range returns false). + if (wait->end < 0) wait->disabled = true; + if (wait->disabled || http->pointer >= wait->end) { + aki_cond_signal(&wait->cond); + al_array_remove_at_iter(http->handler.waits, i); + } + aki_mutex_unlock(&wait->mutex); + } + aki_mutex_unlock(&http->mutex); + break; + } + default: + break; + } + return ret; +} + +static void signal_callback(void *userdata) +{ + struct cch_handler_http *http = (struct cch_handler_http *)userdata; + do { + u32 size; + camu_queue_size(http->queue, size); + if (size == 0) break; + s32 index; + camu_queue_pop(http->queue, index); + if (index == -1) { + aki_event_loop_break(&http->loop); + break; + } + al_array_push(http->requests, (struct aki_http){0}); + struct aki_http *request = &al_array_last(http->requests); + aki_http_init(request); + aki_http_set_url(request, &http->url); + aki_http_set_user_agent(request, USER_AGENT); + aki_http_request_stream(request, AKI_HTTP_GET, &http->loop, http_callback, http); + } while (1); +} + +static void handler_http_maybe_spawn_worker(struct cch_handler *handler, size_t index) +{ + struct cch_handler_http *http = (struct cch_handler_http *)handler; + camu_queue_push(http->queue, index); + aki_signal_send(&http->signal); +} + +static bool handler_http_wait_for_range(struct cch_handler *handler, struct cch_handler_wait *wait) +{ + struct cch_handler_http *http = (struct cch_handler_http *)handler; + aki_mutex_lock(&http->mutex); + bool canceled = false; + if (cch_entry_get_size(http->handler.entry) < 0 || http->pointer < wait->end) { + aki_mutex_lock(&wait->mutex); + if (!wait->disabled) { + al_array_push(http->handler.waits, wait); + aki_mutex_unlock(&http->mutex); + aki_cond_wait(&wait->cond, &wait->mutex); + } else { + aki_mutex_unlock(&wait->mutex); + aki_mutex_unlock(&http->mutex); + return false; + } + aki_mutex_unlock(&wait->mutex); + aki_mutex_lock(&http->mutex); + if (wait->disabled) { + u16 id = wait->id; + al_array_foreach(http->handler.waits, i, wait) { + if (wait->id == id) { + al_array_remove_at_iter(http->handler.waits, i); + break; + } + } + canceled = true; + } + } + aki_mutex_unlock(&http->mutex); + return !canceled; +} + +static void handler_http_free(struct cch_handler **handler) +{ + struct cch_handler_http *http = (struct cch_handler_http *)*handler; + aki_mutex_lock(&http->mutex); + struct cch_handler_wait *wait; + al_array_foreach(http->handler.waits, i, wait) { + aki_mutex_lock(&wait->mutex); + wait->disabled = true; + aki_cond_signal(&wait->cond); + aki_mutex_unlock(&wait->mutex); + } + aki_mutex_unlock(&http->mutex); + camu_queue_push(http->queue, -1); + aki_signal_send(&http->signal); + aki_thread_join(&http->thread); + aki_mutex_destroy(&http->mutex); + struct aki_http *request; + al_array_foreach_ptr(http->requests, i, request) { + aki_http_close(request); + } + al_array_free(http->requests); + camu_queue_free(http->queue); + al_str_free(&http->url); + al_array_free(http->handler.waits); + al_free(http); + *handler = NULL; +} + +static aki_thread_result AKI_THREADCALL event_loop_thread(void *userdata) +{ + struct cch_handler_http *http = (struct cch_handler_http *)userdata; + aki_event_loop_run(&http->loop); + aki_event_loop_destroy(&http->loop); + return 0; +} + +struct cch_entry *cch_handler_http_create(str *url) +{ + //struct cch_backing *backing = cch_backing_file_create(al_str_c("/tmp/camu_http_data"), 1024 * 128); + struct cch_backing *backing = cch_backing_memory_create(1024 * 128); + if (!backing) return NULL; + struct cch_entry *entry = al_alloc_object(struct cch_entry); + entry->backing = backing; + entry->ref_count = 0; + entry->hash = 0; + entry->handler = (struct cch_handler *)al_alloc_object(struct cch_handler_http); + aki_mutex_init(&entry->mutex); + entry->handler->can_seek = handler_http_can_seek; + entry->handler->maybe_spawn_worker = handler_http_maybe_spawn_worker; + entry->handler->wait_for_range = handler_http_wait_for_range; + entry->handler->free = handler_http_free; + entry->handler->entry = entry; + entry->handler->backing = entry->backing; + al_array_init(entry->handler->waits); + struct cch_handler_http *http = (struct cch_handler_http *)entry->handler; + al_str_clone(&http->url, url); + http->pointer = 0; + al_array_init(http->requests); + camu_queue_init(http->queue); + aki_event_loop_init(&http->loop); + aki_signal_init(&http->signal, signal_callback, http); + aki_signal_start(&http->signal, &http->loop); + aki_mutex_init(&http->mutex); + aki_thread_create(&http->thread, event_loop_thread, http); + return entry; +} diff --git a/src/cache/handlers/http.h b/src/cache/handlers/http.h new file mode 100644 index 0000000..f79a39a --- /dev/null +++ b/src/cache/handlers/http.h @@ -0,0 +1,24 @@ +#pragma once + +#include +#include +#include + +#include "../../util/queue.h" + +#include "../handler.h" +#include "../entry.h" + +struct cch_handler_http { + struct cch_handler handler; + off_t pointer; + str url; + array(struct aki_http) requests; + queue(s32) queue; + struct aki_event_loop loop; + struct aki_signal signal; + struct aki_mutex mutex; + struct aki_thread thread; +}; + +struct cch_entry *cch_handler_http_create(str *url); diff --git a/src/cache/meson.build b/src/cache/meson.build new file mode 100644 index 0000000..099b2a6 --- /dev/null +++ b/src/cache/meson.build @@ -0,0 +1,18 @@ +cache_src = [ + 'entry.c', + 'handle.c', + 'handlers/file.c', + 'backings/memory.c' +] + +if akiyo_has_mmap + cache_src += ['backings/file_mapped.c'] +else + cache_src += ['backings/file.c'] +endif + +if akiyo_has_curl + cache_src += ['handlers/http.c'] +endif + +cache = declare_dependency(sources: cache_src) -- cgit v1.2.3-101-g0448