Sink refactor wip

Dangling target won't stay around. I have a couple of ideas about how to massively simplify the handling of ended entries.

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

10 files changed, 109 insertions, 106 deletions

diff --git a/src/cache/backings/file.c b/src/cache/backings/file.c
index bf62190..f421623 100644
--- a/src/cache/backings/file.c
+++ b/src/cache/backings/file.c
@@ -17,12 +17,14 @@ static void file_backing_write(struct cch_backing *backing, u8 *buf, off_t index
         file->u.pointer = nn_file_seek(&file->file, index, SEEK_SET);
         al_assert(file->u.pointer == index);
     }
-    if (file->size >= 0 && index + ((off_t)*size) >= file->size) {
-        *size = MAX(file->size - index, 0L);
+    off_t n = (off_t)*size;
+    if (file->size >= 0 && index + n >= file->size) {
+        n = MAX(file->size - index, (off_t)0);
     }
-    nn_file_write(&file->file, buf, *size);
-    file->u.pointer += *size;
-    cch_backing_fill_range(&file->backing, index, *size);
+    nn_file_write(&file->file, buf, n);
+    file->u.pointer += n;
+    cch_backing_fill_range(&file->backing, index, n);
+    *size = (size_t)n;
     nn_mutex_unlock(&file->mutex);
 }
 
@@ -34,11 +36,13 @@ static void file_backing_read(struct cch_backing *backing, u8 *buf, off_t index,
         file->u.pointer = nn_file_seek(&file->file, index, SEEK_SET);
         al_assert(file->u.pointer == index);
     }
-    if (file->size >= 0 && index + ((off_t)*size) >= file->size) {
-        *size = MAX(file->size - index, 0L);
+    off_t n = (off_t)*size;
+    if (file->size >= 0 && index + n >= file->size) {
+        n = MAX(file->size - index, (off_t)0);
     }
-    nn_file_read(&file->file, buf, *size);
-    file->u.pointer += *size;
+    nn_file_read(&file->file, buf, n);
+    file->u.pointer += n;
+    *size = (size_t)n;
     nn_mutex_unlock(&file->mutex);
 }
 
@@ -53,9 +57,8 @@ static void file_backing_resize(struct cch_backing *backing, size_t size)
     struct cch_backing_file *file = (struct cch_backing_file *)backing;
     nn_mutex_lock(&file->mutex);
     file->size = size;
-    if ((off_t)size > file->filesize) {
+    if (size > file->file.size) {
         nn_file_truncate(&file->file, size);
-        file->filesize = size;
     }
     nn_mutex_unlock(&file->mutex);
 }
diff --git a/src/cache/backings/file.h b/src/cache/backings/file.h
index 8a2bf7a..9ffd650 100644
--- a/src/cache/backings/file.h
+++ b/src/cache/backings/file.h
@@ -11,7 +11,6 @@ struct cch_backing_file {
     struct cch_backing backing;
     struct nn_file file;
     off_t size;
-    off_t filesize;
     union {
         void *map; // CACHE_BACKING_MAPPED
         off_t pointer; // CACHE_BACKING_READ
diff --git a/src/cache/backings/file_common.c b/src/cache/backings/file_common.c
index 0ee2574..e8bd2ca 100644
--- a/src/cache/backings/file_common.c
+++ b/src/cache/backings/file_common.c
@@ -11,13 +11,12 @@ static bool file_open_internal(struct cch_backing_file *file, str *path, size_t
 {
     s32 flags = size != 0 ? NNWT_FILE_CREATE : NNWT_FILE_READONLY;
     if (!nn_file_open(&file->file, path, flags)) return false;
-    file->filesize = nn_file_get_filesize(&file->file);
     if (!size) {
-        file->size = file->filesize;
+        file->size = file->file.size;
         cch_backing_fill_range(&file->backing, 0, file->size);
     } else {
         file->size = -1;
-        if (file->filesize < (off_t)size) {
+        if (file->file.size < size) {
             nn_file_truncate(&file->file, size);
         }
     }
diff --git a/src/cache/backings/file_mapped.c b/src/cache/backings/file_mapped.c
index 35ab267..e13d602 100644
--- a/src/cache/backings/file_mapped.c
+++ b/src/cache/backings/file_mapped.c
@@ -13,11 +13,13 @@ static void file_backing_write(struct cch_backing *backing, u8 *buf, off_t index
 {
     struct cch_backing_file *file = (struct cch_backing_file *)backing;
     nn_mutex_lock(&file->mutex);
-    if (file->size >= 0 && index + (off_t)*size >= file->size) {
-        *size = MAX(file->size - index, 0L);
+    off_t n = (off_t)*size;
+    if (file->size >= 0 && index + n >= file->size) {
+        n = MAX(file->size - index, (off_t)0);
     }
-    al_memcpy(file->u.map + index, buf, *size);
-    cch_backing_fill_range(&file->backing, index, *size);
+    al_memcpy(file->u.map + index, buf, n);
+    cch_backing_fill_range(&file->backing, index, n);
+    *size = (size_t)n;
     nn_mutex_unlock(&file->mutex);
 }
 
@@ -25,9 +27,11 @@ static u8 *file_backing_get_ptr(struct cch_backing *backing, off_t index, size_t
 {
     struct cch_backing_file *file = (struct cch_backing_file *)backing;
     nn_mutex_lock(&file->mutex);
-    if (file->size >= 0 && index + (off_t)*size >= file->size) {
-        *size = MAX(file->size - index, 0L);
+    off_t n = (off_t)*size;
+    if (file->size >= 0 && index + n >= file->size) {
+        n = MAX(file->size - index, (off_t)0);
     }
+    *size = (size_t)n;
     return (u8 *)(file->u.map + index);
 }
 
@@ -42,11 +46,11 @@ static void file_backing_resize(struct cch_backing *backing, size_t size)
     struct cch_backing_file *file = (struct cch_backing_file *)backing;
     nn_mutex_lock(&file->mutex);
     file->size = size;
-    if ((off_t)size > file->filesize) {
+    if (size > file->file.size) {
         nn_file_munmap(&file->file, file->u.map);
         nn_file_truncate(&file->file, size);
         file->u.map = nn_file_mmap(&file->file);
-        file->filesize = size;
+        file->file.size = size;
     }
     nn_mutex_unlock(&file->mutex);
 }
diff --git a/src/cache/entry.c b/src/cache/entry.c
index 3cae147..560e5ee 100644
--- a/src/cache/entry.c
+++ b/src/cache/entry.c
@@ -8,7 +8,7 @@ bool cch_entry_get_handle(struct cch_entry *entry, struct cch_handle *handle)
     handle->pointer = 0;
     handle->wait.disabled = false;
     nn_cond_init(&handle->wait.cond);
-    nn_mutex_init(&handle->wait.mutex);
+    nn_mutex_init(&handle->wait.lock);
     nn_mutex_lock(&entry->mutex);
     entry->ref_count++;
     nn_mutex_unlock(&entry->mutex);
@@ -42,7 +42,7 @@ void cch_entry_return_handle(struct cch_entry *entry, struct cch_handle *handle)
 {
     handle->entry = NULL;
     entry->ref_count--;
-    nn_mutex_destroy(&handle->wait.mutex);
+    nn_mutex_destroy(&handle->wait.lock);
     nn_cond_destroy(&handle->wait.cond);
 }
 
diff --git a/src/cache/handle.c b/src/cache/handle.c
index 9f39f8e..1b351a2 100644
--- a/src/cache/handle.c
+++ b/src/cache/handle.c
@@ -1,4 +1,5 @@
-#include <al/lib.h>
+#define AL_LOG_SECTION "cache_handle"
+//#define AL_LOG_ENABLE_TRACE
 #include <al/log.h>
 
 #include "../codec/codec.h"
@@ -11,6 +12,7 @@ static off_t wait_for_size(struct cch_handle *handle, struct cch_handler *handle
 {
     handle->wait.end = -1;
     if (!handler->wait_for_range(handler, &handle->wait)) {
+        trace("Wait for size failed.");
         return -1;
     }
     return cch_entry_get_size(handler->entry);
@@ -19,6 +21,7 @@ static off_t wait_for_size(struct cch_handle *handle, struct cch_handler *handle
 s32 cch_handle_read(struct cch_handle *handle, u8 *buf, s32 size)
 {
     struct cch_handler *handler = handle->entry->handler;
+    // @TODO: Unknown size is unhandled in backings.
     // filesize < 0: Size is yet to be evaluated and we can wait on it.
     // filesize = 0: Size is explicitly unknown.
     off_t filesize = cch_entry_get_size(handle->entry);
@@ -26,14 +29,14 @@ s32 cch_handle_read(struct cch_handle *handle, u8 *buf, s32 size)
     if (filesize > 0) {
         if (handle->pointer >= filesize) {
             return CAMU_ERR_EOF;
-        }
-        if (handle->pointer + size >= filesize) {
+        } else if (handle->pointer + size >= filesize) {
             size = filesize - handle->pointer;
         }
     }
     handle->wait.start = handle->pointer;
     handle->wait.end = handle->pointer + size;
     if (!handler->wait_for_range(handler, &handle->wait)) {
+        trace("Wait for range (%zd-%zd) failed.", handle->wait.start, handle->wait.end);
         return CAMU_ERR_EOF;
     }
     struct cch_backing *backing = handle->entry->backing;
@@ -49,7 +52,7 @@ s32 cch_handle_read(struct cch_handle *handle, u8 *buf, s32 size)
         backing->read(backing, buf, handle->pointer, &available);
     }
     handle->pointer += available;
-    al_assert(available <= INT32_MAX);
+    trace("read(%u), pointer: %zd.", size, handle->pointer);
     return available > 0 ? (s32)available : CAMU_ERR_EOF;
 }
 
@@ -59,21 +62,27 @@ off_t cch_handle_seek(struct cch_handle *handle, off_t offset, s32 whence)
     off_t filesize = cch_entry_get_size(handle->entry);
     if (filesize < 0) filesize = wait_for_size(handle, handler);
     if (filesize <= 0) return -1;
-    if (whence == CAMU_SEEK_SIZE) return filesize;
+    if (whence == CAMU_SEEK_SIZE) {
+        trace("SEEK_SIZE(), pointer: %zd, filesize: %zd.", handle->pointer, filesize);
+        return filesize;
+    }
     switch (whence) {
     case SEEK_SET:
-        if (offset >= filesize || offset < 0) {
+        trace("SEEK_SET(%zd, %d), pointer: %zd.", offset, whence, handle->pointer);
+        if (offset > filesize || offset < 0) {
             return -1;
         }
         handle->pointer = offset;
         break;
     case SEEK_CUR:
-        if (handle->pointer + offset >= filesize || handle->pointer + offset < 0) {
+        trace("SEEK_CUR(%zd, %d), pointer: %zd.", offset, whence, handle->pointer);
+        if (handle->pointer + offset > filesize || handle->pointer + offset < 0) {
             return -1;
         }
         handle->pointer += offset;
         break;
     case SEEK_END:
+        trace("SEEK_END(%zd, %d), pointer: %zd.", offset, whence, handle->pointer);
         handle->pointer = filesize + offset;
         break;
     default:
diff --git a/src/cache/range.h b/src/cache/range.h
index 9ec1159..d6c7539 100644
--- a/src/cache/range.h
+++ b/src/cache/range.h
@@ -5,8 +5,8 @@
 AL_UNUSED_FUNCTION_PUSH
 
 // @TODO:
-//  cch_backing_remove_range().
-//  Report overlap, don't unlock in backing_write() and handle fill in handler.
+//  - cch_backing_remove_range().
+//  - Report overlap, don't unlock in backing_write() and handle fill in handler.
 
 static void cch_backing_fill_range(struct cch_backing *backing, off_t index, off_t size)
 {
diff --git a/src/cache/threaded_waits.c b/src/cache/threaded_waits.c
index 541bee6..6d8a5d8 100644
--- a/src/cache/threaded_waits.c
+++ b/src/cache/threaded_waits.c
@@ -1,151 +1,140 @@
 #include "threaded_waits.h"
 
-void cch_threaded_waits_init(struct cch_handler_waits *waits)
+void cch_threaded_waits_init(struct cch_handler_waits *thw)
 {
-    waits->disabled = false;
-    nn_mutex_init(&waits->mutex);
-    al_array_init(waits->active);
+    thw->disabled = false;
+    nn_mutex_init(&thw->lock);
+    al_array_init(thw->active);
 }
 
 static bool wait_range_satisfied(struct cch_backing *backing, struct cch_handler_wait *wait)
 {
-    if (wait->end < 0) {
-        return true;
-    }
-
+    if (wait->end < 0) return true;
     struct cch_range *range;
     al_array_foreach_ptr(backing->available, i, range) {
         if (wait->start >= range->start && range->end >= wait->end) {
             return true;
         }
     }
-
     return false;
 }
 
-bool cch_threaded_wait_for_range(struct cch_handler_waits *waits, struct cch_entry *entry,
+bool cch_threaded_wait_for_range(struct cch_handler_waits *thw, struct cch_entry *entry,
     struct cch_backing *backing, struct cch_handler_wait *wait)
 {
-    nn_mutex_lock(&waits->mutex);
-
     off_t size = cch_entry_get_size(entry);
+
+    // Lock before checking wait_range_satisfied() in case the result changes
+    // before wait is possibly added to active.
+    nn_mutex_lock(&thw->lock);
+
     backing->lock(backing);
     bool satisfied = size >= 0 && wait_range_satisfied(backing, wait);
     backing->unlock(backing);
 
-    nn_mutex_lock(&wait->mutex);
-
-    bool canceled = waits->disabled || wait->disabled;
-
+    nn_mutex_lock(&wait->lock);
+    bool canceled = thw->disabled || wait->disabled;
     if (!canceled && !satisfied) {
-        al_array_push(waits->active, wait);
-
+        al_array_push(thw->active, wait);
         // Unlock handler to wait.
-        nn_mutex_unlock(&waits->mutex);
-        nn_cond_wait(&wait->cond, &wait->mutex);
+        nn_mutex_unlock(&thw->lock);
+        nn_cond_wait(&wait->cond, &wait->lock);
         // A successful wait will have been removed before signaling.
-
         if (wait->disabled) {
             canceled = true;
             // Re-lock to remove.
-            nn_mutex_lock(&waits->mutex);
-            al_array_remove(waits->active, wait);
+            nn_mutex_lock(&thw->lock);
+            al_array_remove(thw->active, wait);
         }
     }
+    nn_mutex_unlock(&wait->lock);
 
-    nn_mutex_unlock(&wait->mutex);
     if (canceled || satisfied) {
         // Else, we already unlocked to wait.
-        nn_mutex_unlock(&waits->mutex);
+        nn_mutex_unlock(&thw->lock);
     }
 
     return !canceled;
 }
 
-void cch_threaded_waits_signal_any(struct cch_handler_waits *waits)
+void cch_threaded_waits_signal_any(struct cch_handler_waits *thw)
 {
-    nn_mutex_lock(&waits->mutex);
-
+    nn_mutex_lock(&thw->lock);
     struct cch_handler_wait *wait;
-    al_array_foreach_rev(waits->active, i, wait) {
+    al_array_foreach_rev(thw->active, i, wait) {
         if (wait->end < 0) {
             // This will be considered a successful wait.
-            al_array_remove_at(waits->active, i);
-            nn_mutex_lock(&wait->mutex);
+            al_array_remove_at(thw->active, i);
+            nn_mutex_lock(&wait->lock);
             nn_cond_signal(&wait->cond);
-            nn_mutex_unlock(&wait->mutex);
+            nn_mutex_unlock(&wait->lock);
         }
     }
-
-    nn_mutex_unlock(&waits->mutex);
+    nn_mutex_unlock(&thw->lock);
 }
 
-void cch_threaded_waits_evaluate(struct cch_handler_waits *waits, struct cch_backing *backing)
+void cch_threaded_waits_evaluate(struct cch_handler_waits *thw, struct cch_backing *backing)
 {
-    nn_mutex_lock(&waits->mutex);
+    nn_mutex_lock(&thw->lock);
     backing->lock(backing);
-
     struct cch_handler_wait *wait;
-    al_array_foreach_rev(waits->active, i, wait) {
+    al_array_foreach_rev(thw->active, i, wait) {
         if (wait_range_satisfied(backing, wait)) {
-            al_array_remove_at(waits->active, i);
-            nn_mutex_lock(&wait->mutex);
+            al_array_remove_at(thw->active, i);
+            nn_mutex_lock(&wait->lock);
             nn_cond_signal(&wait->cond);
-            nn_mutex_unlock(&wait->mutex);
+            nn_mutex_unlock(&wait->lock);
         }
     }
-
     backing->unlock(backing);
-    nn_mutex_unlock(&waits->mutex);
+    nn_mutex_unlock(&thw->lock);
 }
 
 void cch_threaded_wait_disable(struct cch_handler_wait *wait)
 {
-    nn_mutex_lock(&wait->mutex);
-
-    // If this wait is in waits->active, cond_is_waiting() will be true.
-    // This should mean we can be assured that successive calls to
-    // disable() -> enable() won't leave this wait in waits->active.
+    // If this wait is in active, cond_is_waiting() will be true.
+    // So, signaling cond while holding the lock here will ensure it's removed before
+    // unlocking. This enables synchronization of successive disable() -> enable() calls.
+    nn_mutex_lock(&wait->lock);
     wait->disabled = true;
     if (nn_cond_is_waiting(&wait->cond)) {
         nn_cond_signal(&wait->cond);
     }
-
-    nn_mutex_unlock(&wait->mutex);
+    nn_mutex_unlock(&wait->lock);
 }
 
 void cch_threaded_wait_enable(struct cch_handler_wait *wait)
 {
-    nn_mutex_lock(&wait->mutex);
+    nn_mutex_lock(&wait->lock);
     wait->disabled = false;
-    nn_mutex_unlock(&wait->mutex);
+    nn_mutex_unlock(&wait->lock);
 }
 
-void cch_threaded_waits_disable_all(struct cch_handler_waits *waits)
+void cch_threaded_waits_disable_all(struct cch_handler_waits *thw)
 {
-    nn_mutex_lock(&waits->mutex);
+    nn_mutex_lock(&thw->lock);
 
-    if (waits->disabled) {
-        nn_mutex_unlock(&waits->mutex);
+    if (thw->disabled) {
+        nn_mutex_unlock(&thw->lock);
         return;
     }
 
     // Disallow any further waits.
-    waits->disabled = true;
+    thw->disabled = true;
 
     struct cch_handler_wait *wait;
-    al_array_foreach_rev(waits->active, i, wait) {
-        nn_mutex_lock(&wait->mutex);
+    al_array_foreach_rev(thw->active, i, wait) {
+        nn_mutex_lock(&wait->lock);
         wait->disabled = true;
         nn_cond_signal(&wait->cond);
-        nn_mutex_unlock(&wait->mutex);
+        nn_mutex_unlock(&wait->lock);
     }
 
-    nn_mutex_unlock(&waits->mutex);
+    nn_mutex_unlock(&thw->lock);
 }
 
-void cch_threaded_waits_close(struct cch_handler_waits *waits)
+void cch_threaded_waits_close(struct cch_handler_waits *thw)
 {
-    nn_mutex_destroy(&waits->mutex);
-    al_array_free(waits->active);
+    nn_mutex_destroy(&thw->lock);
+    al_array_free(thw->active);
 }
diff --git a/src/cache/threaded_waits.h b/src/cache/threaded_waits.h
index 1893395..ce0f19c 100644
--- a/src/cache/threaded_waits.h
+++ b/src/cache/threaded_waits.h
@@ -8,16 +8,16 @@
 
 struct cch_handler_waits {
     bool disabled;
-    struct nn_mutex mutex;
+    struct nn_mutex lock;
     array(struct cch_handler_wait *) active;
 };
 
-void cch_threaded_waits_init(struct cch_handler_waits *waits);
-bool cch_threaded_wait_for_range(struct cch_handler_waits *waits, struct cch_entry *entry,
+void cch_threaded_waits_init(struct cch_handler_waits *thw);
+bool cch_threaded_wait_for_range(struct cch_handler_waits *thw, struct cch_entry *entry,
     struct cch_backing *backing, struct cch_handler_wait *wait);
-void cch_threaded_waits_signal_any(struct cch_handler_waits *waits);
-void cch_threaded_waits_evaluate(struct cch_handler_waits *waits, struct cch_backing *backing);
+void cch_threaded_waits_signal_any(struct cch_handler_waits *thw);
+void cch_threaded_waits_evaluate(struct cch_handler_waits *thw, struct cch_backing *backing);
 void cch_threaded_wait_disable(struct cch_handler_wait *wait);
 void cch_threaded_wait_enable(struct cch_handler_wait *wait);
-void cch_threaded_waits_disable_all(struct cch_handler_waits *waits);
-void cch_threaded_waits_close(struct cch_handler_waits *waits);
+void cch_threaded_waits_disable_all(struct cch_handler_waits *thw);
+void cch_threaded_waits_close(struct cch_handler_waits *thw);
diff --git a/src/cache/wait.h b/src/cache/wait.h
index 00c5749..143e2b9 100644
--- a/src/cache/wait.h
+++ b/src/cache/wait.h
@@ -6,5 +6,5 @@ struct cch_handler_wait {
     off_t start, end;
     bool disabled;
     struct nn_cond cond;
-    struct nn_mutex mutex;
+    struct nn_mutex lock;
 };