summaryrefslogtreecommitdiff
path: root/src/cache
diff options
context:
space:
mode:
Diffstat (limited to 'src/cache')
-rw-r--r--src/cache/backing.h20
-rw-r--r--src/cache/backings/file.c75
-rw-r--r--src/cache/backings/file.h22
-rw-r--r--src/cache/backings/file_common.c23
-rw-r--r--src/cache/backings/file_mapped.c81
-rw-r--r--src/cache/backings/memory.c82
-rw-r--r--src/cache/backings/memory.h16
-rw-r--r--src/cache/entry.c44
-rw-r--r--src/cache/entry.h21
-rw-r--r--src/cache/handle.c100
-rw-r--r--src/cache/handle.h20
-rw-r--r--src/cache/handler.h23
-rw-r--r--src/cache/handlers/file.c49
-rw-r--r--src/cache/handlers/file.h12
-rw-r--r--src/cache/handlers/http.c207
-rw-r--r--src/cache/handlers/http.h24
-rw-r--r--src/cache/meson.build18
17 files changed, 837 insertions, 0 deletions
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 <al/types.h>
+
+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 <al/str.h>
+#include <aki/thread.h>
+#include <aki/common.h>
+#include <aki/file.h>
+
+#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 <al/lib.h>
+#include <aki/thread.h>
+
+#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 <al/lib.h>
+
+#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 <al/types.h>
+
+#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 <al/lib.h>
+#include <al/log.h>
+#include <al/random.h>
+
+#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 <al/types.h>
+#include <aki/thread.h>
+
+#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 <al/types.h>
+#include <al/array.h>
+#include <aki/thread.h>
+
+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 <al/str.h>
+
+#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 <al/random.h>
+#include <al/log.h>
+
+#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 <aki/http.h>
+#include <aki/thread.h>
+#include <aki/signal.h>
+
+#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)
generated by cgit v1.2.3-101-g0448 (git 2.54.0) at 2026-07-25 19:24:59 +0000