Everything before initial synced list

Signed-off-by: Andrew Opalach <andrew@akon.city>

6 files changed, 268 insertions, 135 deletions

diff --git a/src/cache/handlers/cdio.c b/src/cache/handlers/cdio.c
new file mode 100644
index 0000000..23f87b6
--- /dev/null
+++ b/src/cache/handlers/cdio.c
@@ -0,0 +1,206 @@
+#include <al/log.h>
+
+#include "../backings/file.h"
+#include "../backings/memory.h"
+#include "../threaded_waits.h"
+#include "../range.h"
+
+#include "cdio.h"
+
+// 1 sector at a time to minimize skips
+#define SECTORS_PER_STEP 1
+#define BYTES_PER_STEP (SECTORS_PER_STEP * CDIO_CD_FRAMESIZE_RAW)
+
+static void cdio_log_messages(struct cch_handler_cdio *cdio)
+{
+    char *error_messages = cdio_cddap_errors(cdio->drive);
+    char *messages = cdio_cddap_messages(cdio->drive);
+    if (error_messages) {
+        al_log_error("cdio", "\n%s", error_messages);
+        cdio_cddap_free_messages(error_messages);
+    }
+    if (messages) {
+        al_log_info("cdio", "\n%s", messages);
+        cdio_cddap_free_messages(messages);
+    }
+}
+
+static bool handler_cdio_can_seek(struct cch_handler *handler)
+{
+    (void)handler;
+    return true;
+}
+
+static aki_thread_result AKI_THREADCALL cd_read_thread(void *userdata)
+{
+    struct cch_handler_cdio *cdio = (struct cch_handler_cdio *)userdata;
+    while (1) {
+        if (!al_atomic_load(s32)(&cdio->running, AL_ATOMIC_RELAXED)) {
+            // We don't care about how complete the disc read was.
+            return 0;
+        }
+        // cdio_read() here can block for a very long time so we buffer each read
+        // before writing it to the backing.
+        long ret = cdio_cddap_read(cdio->drive, aki_buffer_get_ptr(&cdio->buffer, 0), cdio->sector, SECTORS_PER_STEP);
+        cdio_log_messages(cdio);
+        al_assert(ret <= SECTORS_PER_STEP);
+        off_t pointer = cdio->sector * CDIO_CD_FRAMESIZE_RAW;
+        size_t size;
+        u8 *ptr = cdio->backing->get_ptr(cdio->backing, pointer, &size);
+        if (size < BYTES_PER_STEP) {
+            al_log_error("cdio", "Tried to write over expected size during normal operation.");
+            cdio->backing->unlock(cdio->backing);
+            break;
+        }
+        if (ret == SECTORS_PER_STEP) {
+            al_memcpy(ptr, aki_buffer_get_ptr(&cdio->buffer, 0), BYTES_PER_STEP);
+        } else if (ret == -10) {
+            al_log_error("cdio", "I/O error, skipping sector.");
+            al_memset(ptr, 0, BYTES_PER_STEP);
+            ret = SECTORS_PER_STEP;
+        } else if (ret < 0) {
+            al_log_error("cdio", "Fatal error (%d).", ret);
+            cdio->backing->unlock(cdio->backing);
+            break;
+        }
+        cch_backing_fill_range(cdio->backing, pointer, BYTES_PER_STEP);
+        cdio->sector += ret;
+        cdio->backing->unlock(cdio->backing);
+        cch_threaded_waits_evaluate(&cdio->handler, cdio->backing);
+        if (ret < SECTORS_PER_STEP || cdio->sector > cdio->end) {
+            al_log_debug("cdio", "Normal CD EOF.");
+            break;
+        }
+    }
+    if (cdio->sector <= cdio->end) {
+        al_log_warn("cdio", "Disc read cut short (sectors: %zd, end: %zd).", cdio->sector, cdio->end);
+        // We didn't read the full disc, disable waiters.
+        cch_threaded_waits_disable_all(&cdio->handler);
+    }
+    return 0;
+}
+
+static void handler_cdio_maybe_spawn_worker(struct cch_handler *handler, size_t index)
+{
+    struct cch_handler_cdio *cdio = (struct cch_handler_cdio *)handler;
+    if (cdio->start == (lsn_t)index) return;
+    if (al_atomic_load(s32)(&cdio->running, AL_ATOMIC_ACQUIRE)) {
+        al_atomic_store(s32)(&cdio->running, 0, AL_ATOMIC_RELEASE);
+        aki_thread_join(&cdio->thread);
+    }
+    cdio->start = index;
+    cdio->sector = cdio->start;
+    al_atomic_store(s32)(&cdio->running, 1, AL_ATOMIC_RELEASE);
+    aki_thread_create(&cdio->thread, cd_read_thread, cdio);
+}
+
+static bool handler_cdio_wait_for_range(struct cch_handler *handler, struct cch_handler_wait *wait)
+{
+    struct cch_handler_cdio *cdio = (struct cch_handler_cdio *)handler;
+    return cch_threaded_wait_for_range(&cdio->handler, cdio->backing, wait);
+}
+
+static void handler_cdio_free(struct cch_handler **handler)
+{
+    struct cch_handler_cdio *cdio = (struct cch_handler_cdio *)*handler;
+    cch_threaded_waits_disable_all(&cdio->handler);
+    if (al_atomic_load(s32)(&cdio->running, AL_ATOMIC_RELAXED)) {
+        al_atomic_store(s32)(&cdio->running, 0, AL_ATOMIC_RELAXED);
+        aki_thread_join(&cdio->thread);
+    }
+    if (cdio->drive) {
+        cdio_log_messages(cdio);
+        cdio_cddap_close(cdio->drive);
+    }
+    cch_threaded_waits_close(&cdio->handler);
+    al_str_free(&cdio->handler.liana);
+    al_free(cdio);
+    *handler = NULL;
+}
+
+static bool open_cd_drive(struct cch_handler_cdio *cdio)
+{
+    cdrom_drive_t *drive = NULL;
+    char **cd_drives;
+    CdIo_t *cdio_handle = NULL;
+
+    cd_drives = cdio_get_devices_with_cap(NULL, CDIO_FS_AUDIO, false);
+    if (cd_drives) {
+        // Use first drive.
+        cdio_handle = cdio_open(*cd_drives, DRIVER_UNKNOWN);
+        drive = cdio_cddap_identify_cdio(cdio_handle, CDDA_MESSAGE_LOGIT, NULL);
+    } else {
+        al_log_error("cdio", "No compatible disc found.");
+        return false;
+    }
+    cdio_free_device_list(cd_drives);
+
+    cdio_cddap_verbose_set(drive, CDDA_MESSAGE_LOGIT, CDDA_MESSAGE_LOGIT);
+
+    s32 ret = cdio_cddap_open(drive);
+    if (ret != 0) {
+        al_log_error("cdio", "Unable to open disc (%d).", ret);
+        return false;
+    }
+
+    ret = cdio_cddap_speed_set(drive, 4);
+    if (ret != 0) {
+        al_log_warn("cdio", "Failed to set drive speed (%d).", ret);
+    }
+
+    // Not sure what this is based on, but it can seemingly just be wrong.
+    track_t tracks = cdio_cddap_tracks(drive);
+    al_log_info("cdio", "CD with %d tracks loaded.", tracks);
+    al_array_reserve(cdio->handler.entry->chapters, tracks);
+
+    lsn_t start = cdio_cddap_disc_firstsector(drive);
+    lsn_t end = cdio_cddap_disc_lastsector(drive);
+    lsn_t offset = start;
+    s32 track = CDIO_INVALID_TRACK;
+    while (offset < end) {
+        track = cdio_cddap_sector_gettrack(drive, offset);
+        al_assert(track != CDIO_INVALID_TRACK);
+        lsn_t track_start = cdio_cddap_track_firstsector(drive, track);
+        lsn_t track_end = cdio_cddap_track_lastsector(drive, track);
+        al_array_push(cdio->handler.entry->chapters, ((struct cch_chapter){ track_start, track_end }));
+        offset = track_end + 1;
+    }
+    // end is a valid index so add one framesize.
+    cch_entry_set_size(cdio->handler.entry, ((end - start) + 1) * CDIO_CD_FRAMESIZE_RAW);
+    cdio->start = -1;
+    cdio->end = end;
+
+    cdio->drive = drive;
+
+    cdio_log_messages(cdio);
+
+    return true;
+}
+
+struct cch_entry *cch_handler_cdio_create(void)
+{
+    struct cch_backing *backing = cch_backing_file_create(al_str_c("/tmp/camu_cd_data"), 4096);
+    if (!backing) return NULL;
+    struct cch_handler_cdio *cdio = al_alloc_object(struct cch_handler_cdio);
+    cdio->backing = backing;
+    struct cch_entry *entry = al_alloc_object(struct cch_entry);
+    entry->backing = backing;
+    entry->handler = (struct cch_handler *)cdio;
+    entry->ref_count = 0;
+    entry->unknown_size = false;
+    entry->hash = 0;
+    aki_mutex_init(&entry->mutex);
+    entry->handler->can_seek = handler_cdio_can_seek;
+    entry->handler->maybe_spawn_worker = handler_cdio_maybe_spawn_worker;
+    entry->handler->wait_for_range = handler_cdio_wait_for_range;
+    entry->handler->free = handler_cdio_free;
+    entry->handler->entry = entry;
+    al_str_from(&entry->handler->liana, "cdio");
+    if (!open_cd_drive(cdio)) return NULL;
+    al_atomic_store(s32)(&cdio->running, 0, AL_ATOMIC_RELAXED);
+    aki_buffer_init(&cdio->buffer);
+    aki_buffer_ensure_space(&cdio->buffer, BYTES_PER_STEP);
+    aki_buffer_set_size(&cdio->buffer, BYTES_PER_STEP);
+    cch_threaded_waits_init(&cdio->handler);
+    return entry;
+}
diff --git a/src/cache/handlers/cdio.h b/src/cache/handlers/cdio.h
new file mode 100644
index 0000000..953c988
--- /dev/null
+++ b/src/cache/handlers/cdio.h
@@ -0,0 +1,21 @@
+#include <al/atomic.h>
+#include <aki/common.h>
+#include <cdio/paranoia/cdda.h>
+#include <cdio/cd_types.h>
+
+#include "../handler.h"
+#include "../entry.h"
+
+struct cch_handler_cdio {
+    struct cch_handler handler;
+    struct cch_backing *backing;
+    cdrom_drive_t *drive;
+    lsn_t start;
+    lsn_t end;
+    lsn_t sector;
+    atomic(s32) running;
+    struct aki_thread thread;
+    struct aki_buffer buffer;
+};
+
+struct cch_entry *cch_handler_cdio_create(void);
diff --git a/src/cache/handlers/file.c b/src/cache/handlers/file.c
index ec6a146..03e2814 100644
--- a/src/cache/handlers/file.c
+++ b/src/cache/handlers/file.c
@@ -25,6 +25,7 @@ static bool handler_file_wait_for_range(struct cch_handler *handler, struct cch_
 static void handler_file_free(struct cch_handler **handler)
 {
     struct cch_handler_file *file = (struct cch_handler_file *)*handler;
+    al_str_free(&file->handler.liana);
     al_free(file);
     *handler = NULL;
 }
@@ -33,17 +34,20 @@ 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_handler_file *file = al_alloc_object(struct cch_handler_file);
+    file->backing = backing;
     struct cch_entry *entry = al_alloc_object(struct cch_entry);
     entry->backing = backing;
+    entry->handler = (struct cch_handler *)file;
     entry->ref_count = 0;
+    entry->unknown_size = false;
     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;
+    al_str_from(&entry->handler->liana, "codec");
     return entry;
 }
diff --git a/src/cache/handlers/file.h b/src/cache/handlers/file.h
index 52e5c94..85e1c7f 100644
--- a/src/cache/handlers/file.h
+++ b/src/cache/handlers/file.h
@@ -7,6 +7,7 @@
 
 struct cch_handler_file {
     struct cch_handler handler;
+    struct cch_backing *backing;
 };
 
 struct cch_entry *cch_handler_file_create(str *path);
diff --git a/src/cache/handlers/http.c b/src/cache/handlers/http.c
index 789b247..71ae186 100644
--- a/src/cache/handlers/http.c
+++ b/src/cache/handlers/http.c
@@ -1,8 +1,10 @@
 #include <al/random.h>
 #include <al/log.h>
 
-#include "../backings/file.h"
 #include "../backings/memory.h"
+#include "../backings/file.h"
+#include "../threaded_waits.h"
+#include "../wait.h"
 
 #include "http.h"
 
@@ -14,21 +16,6 @@ static bool handler_http_can_seek(struct cch_handler *handler)
     return false;
 }
 
-static void disable_waiters_internal(struct cch_handler_http *http)
-{
-    aki_mutex_lock(&http->mutex);
-    // Don't allow any further waits.
-    http->disabled = true;
-    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);
-}
-
 static size_t http_callback(void *userdata, u8 op, u8 *buf, s64 int0)
 {
     struct cch_handler_http *http = (struct cch_handler_http *)userdata;
@@ -39,8 +26,7 @@ static size_t http_callback(void *userdata, u8 op, u8 *buf, s64 int0)
         break;
     }
     case AKI_HTTP_ERROR: {
-        al_log_debug("cache_handler_http", "Error.");
-        disable_waiters_internal(http);
+        cch_threaded_waits_disable_all(&http->handler);
         break;
     }
     case AKI_HTTP_RESPONSE_CODE: {
@@ -54,37 +40,18 @@ static size_t http_callback(void *userdata, u8 op, u8 *buf, s64 int0)
     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);
+        cch_threaded_waits_signal_any(&http->handler);
         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);
+        // If we got data before a Content-Length, assume the size is unknown.
+        if (cch_entry_get_size(http->handler.entry) < 0) {
+            cch_entry_set_size(http->handler.entry, 0);
         }
-        aki_mutex_unlock(&http->mutex);
+        http->backing->write(http->backing, buf, http->pointer, &ret);
+        http->pointer += ret;
+        cch_threaded_waits_evaluate(&http->handler, http->backing);
         break;
     }
     default:
@@ -93,123 +60,63 @@ static size_t http_callback(void *userdata, u8 op, u8 *buf, s64 int0)
     return ret;
 }
 
-static void signal_callback(void *userdata)
-{
-    struct cch_handler_http *http = (struct cch_handler_http *)userdata;
-    u32 size;
-    s32 index;
-    do {
-        camu_queue_try_pop(http->queue, size, index);
-        if (size == 0) break;
-        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);
+    (void)index;
+    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);
 }
 
 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 = http->disabled;
-    if (!canceled && (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;
+    return cch_threaded_wait_for_range(&http->handler, http->backing, wait);
 }
 
 static void handler_http_free(struct cch_handler **handler)
 {
     struct cch_handler_http *http = (struct cch_handler_http *)*handler;
-    disable_waiters_internal(http);
-    camu_queue_push(http->queue, -1);
-    aki_signal_send(&http->signal);
-    aki_thread_join(&http->thread);
-    aki_mutex_destroy(&http->mutex);
+    cch_threaded_waits_disable_all(&http->handler);
     struct aki_http *request;
     al_array_foreach_ptr(http->requests, i, request) {
         aki_http_close(request);
     }
+    cch_threaded_waits_close(&http->handler);
     al_array_free(http->requests);
-    camu_queue_free(http->queue);
     al_str_free(&http->url);
-    al_array_free(http->handler.waits);
+    al_str_free(&http->handler.liana);
     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_entry *cch_handler_http_create(str *url, struct aki_event_loop *loop)
 {
-    //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_handler_http *http = al_alloc_object(struct cch_handler_http);
+    http->backing = backing;
     struct cch_entry *entry = al_alloc_object(struct cch_entry);
     entry->backing = backing;
+    entry->handler = (struct cch_handler *)http;
     entry->ref_count = 0;
+    entry->unknown_size = false;
     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_from(&entry->handler->liana, "codec");
+    http->loop = loop;
     al_str_clone(&http->url, url);
     http->pointer = 0;
-    http->disabled = false;
     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);
+    cch_threaded_waits_init(&http->handler);
     return entry;
 }
diff --git a/src/cache/handlers/http.h b/src/cache/handlers/http.h
index e8fb2e0..5f9df86 100644
--- a/src/cache/handlers/http.h
+++ b/src/cache/handlers/http.h
@@ -4,22 +4,16 @@
 #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;
+    struct cch_backing *backing;
     str url;
-    bool disabled;
+    off_t pointer;
     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 aki_event_loop *loop;
 };
 
-struct cch_entry *cch_handler_http_create(str *url);
+struct cch_entry *cch_handler_http_create(str *url, struct aki_event_loop *loop);