From 06464bd07cb1efbc9df4ca650978c14f4c20390a Mon Sep 17 00:00:00 2001 From: Ziv Scully Date: Thu, 12 Nov 2015 16:36:35 -0500 Subject: Fix committing multiple stores/flushes. Locking is WIP. --- include/urweb/types_cpp.h | 3 +- include/urweb/urweb_cpp.h | 4 +- src/c/urweb.c | 108 +++++++++++++++++++++++++++------------------- src/lru_cache.sml | 8 ++-- 4 files changed, 72 insertions(+), 51 deletions(-) diff --git a/include/urweb/types_cpp.h b/include/urweb/types_cpp.h index c4af2866..82f8d30a 100644 --- a/include/urweb/types_cpp.h +++ b/include/urweb/types_cpp.h @@ -121,6 +121,7 @@ typedef struct { // Caching +#include #include "uthash.h" typedef struct uw_Sqlcache_Value { @@ -132,7 +133,7 @@ typedef struct uw_Sqlcache_Value { typedef struct uw_Sqlcache_Entry uw_Sqlcache_Entry; typedef struct uw_Sqlcache_Cache { - //pthread_rwlock_t *lock; + pthread_rwlock_t lock; uw_Sqlcache_Entry *table; unsigned long timeInvalid; unsigned long timeNow; diff --git a/include/urweb/urweb_cpp.h b/include/urweb/urweb_cpp.h index 3e70b4ac..2c032e7b 100644 --- a/include/urweb/urweb_cpp.h +++ b/include/urweb/urweb_cpp.h @@ -406,8 +406,8 @@ void uw_Basis_writec(struct uw_context *, char); // Sqlcache. -uw_Sqlcache_Value *uw_Sqlcache_check(uw_Sqlcache_Cache *, char **); -void *uw_Sqlcache_store(uw_Sqlcache_Cache *, char **, uw_Sqlcache_Value *); +uw_Sqlcache_Value *uw_Sqlcache_check(struct uw_context *, uw_Sqlcache_Cache *, char **); +void *uw_Sqlcache_store(struct uw_context *, uw_Sqlcache_Cache *, char **, uw_Sqlcache_Value *); void *uw_Sqlcache_flush(struct uw_context *, uw_Sqlcache_Cache *, char **); #endif diff --git a/src/c/urweb.c b/src/c/urweb.c index 4db019fe..4afc7297 100644 --- a/src/c/urweb.c +++ b/src/c/urweb.c @@ -424,11 +424,12 @@ typedef struct { void (*free)(void*); } global; -typedef struct uw_Sqlcache_Inval { +typedef struct uw_Sqlcache_Update { uw_Sqlcache_Cache *cache; char **keys; - struct uw_Sqlcache_Inval *next; -} uw_Sqlcache_Inval; + uw_Sqlcache_Value *value; + struct uw_Sqlcache_Update *next; +} uw_Sqlcache_Update; struct uw_context { uw_app *app; @@ -497,7 +498,8 @@ struct uw_context { // Sqlcache. int numRecording; int recordingOffset; - uw_Sqlcache_Inval *inval; + uw_Sqlcache_Update *cacheUpdate; + uw_Sqlcache_Update *cacheUpdateTail; int remoteSock; }; @@ -508,6 +510,7 @@ size_t uw_heap_max = SIZE_MAX; size_t uw_script_max = SIZE_MAX; uw_context uw_init(int id, uw_loggers *lg) { + puts("Initializing"); uw_context ctx = malloc(sizeof(struct uw_context)); ctx->app = NULL; @@ -585,7 +588,8 @@ uw_context uw_init(int id, uw_loggers *lg) { ctx->numRecording = 0; ctx->recordingOffset = 0; - ctx->inval = NULL; + ctx->cacheUpdate = NULL; + ctx->cacheUpdateTail = NULL; ctx->remoteSock = -1; @@ -4629,10 +4633,9 @@ char *uw_Sqlcache_keyCopy(char *buf, char *key) { } // The NUL-terminated prefix of [key] below always looks something like "_k1_k2_k3..._kn". -// TODO: strlen(key) = buf - key? -uw_Sqlcache_Value *uw_Sqlcache_check(uw_Sqlcache_Cache *cache, char **keys) { - //pthread_rwlock_rdlock(cache->lock); +uw_Sqlcache_Value *uw_Sqlcache_check(uw_context ctx, uw_Sqlcache_Cache *cache, char **keys) { + pthread_rwlock_rdlock(&cache->lock); size_t numKeys = cache->numKeys; char *key = uw_Sqlcache_allocKeyBuffer(keys, numKeys); char *buf = key; @@ -4644,18 +4647,20 @@ uw_Sqlcache_Value *uw_Sqlcache_check(uw_Sqlcache_Cache *cache, char **keys) { entry = uw_Sqlcache_find(cache, key, len, 1); if (!entry) { free(key); + pthread_rwlock_unlock(&cache->lock); return NULL; } timeInvalid = uw_Sqlcache_timeMax(timeInvalid, entry->timeInvalid); } free(key); + // TODO: pass back copy of value and free it in the generated code... or use uw_malloc? uw_Sqlcache_Value *value = entry->value; - //pthread_rwlock_unlock(cache->lock); + pthread_rwlock_unlock(&cache->lock); return value && value->timeValid > timeInvalid ? value : NULL; } -void uw_Sqlcache_store(uw_Sqlcache_Cache *cache, char **keys, uw_Sqlcache_Value *value) { - //pthread_rwlock_wrlock(cache->lock); +void uw_Sqlcache_storeCommitOne(uw_Sqlcache_Cache *cache, char **keys, uw_Sqlcache_Value *value) { + pthread_rwlock_wrlock(&cache->lock); size_t numKeys = cache->numKeys; char *key = uw_Sqlcache_allocKeyBuffer(keys, numKeys); char *buf = key; @@ -4669,7 +4674,7 @@ void uw_Sqlcache_store(uw_Sqlcache_Cache *cache, char **keys, uw_Sqlcache_Value entry = malloc(sizeof(uw_Sqlcache_Entry)); entry->key = strdup(key); entry->value = NULL; - entry->timeInvalid = 0; // ASK: is this okay? + entry->timeInvalid = 0; uw_Sqlcache_add(cache, entry, len); } } @@ -4677,10 +4682,11 @@ void uw_Sqlcache_store(uw_Sqlcache_Cache *cache, char **keys, uw_Sqlcache_Value uw_Sqlcache_freeValue(entry->value); entry->value = value; entry->value->timeValid = timeNow; - //pthread_rwlock_unlock(cache->lock); + pthread_rwlock_unlock(&cache->lock); } void uw_Sqlcache_flushCommitOne(uw_Sqlcache_Cache *cache, char **keys) { + pthread_rwlock_wrlock(&cache->lock); size_t numKeys = cache->numKeys; char *key = uw_Sqlcache_allocKeyBuffer(keys, numKeys); char *buf = key; @@ -4709,55 +4715,69 @@ void uw_Sqlcache_flushCommitOne(uw_Sqlcache_Cache *cache, char **keys) { free(key); // All the keys were non-null and the relevant entry is present, so we delete it. uw_Sqlcache_delete(cache, entry); + pthread_rwlock_unlock(&cache->lock); } -void uw_Sqlcache_flushFree(void *data, int dontCare) { - uw_Sqlcache_Inval *inval = (uw_Sqlcache_Inval *)data; - while (inval) { - char** keys = inval->keys; - size_t numKeys = inval->cache->numKeys; +void uw_Sqlcache_freeUpdate(void *data, int dontCare) { + uw_context ctx = (uw_context)data; + uw_Sqlcache_Update *update = ctx->cacheUpdate; + while (update) { + char** keys = update->keys; + size_t numKeys = update->cache->numKeys; while (numKeys-- > 0) { free(keys[numKeys]); } free(keys); - uw_Sqlcache_Inval *nextInval = inval->next; - free(inval); - inval = nextInval; + // Don't free [update->value]: it's in the cache now! + uw_Sqlcache_Update *nextUpdate = update->next; + free(update); + update = nextUpdate; } -} - -void uw_Sqlcache_flushCommit(void *data) { - uw_Sqlcache_Inval *inval = (uw_Sqlcache_Inval *)data; - while (inval) { - uw_Sqlcache_Cache *cache = inval->cache; - char **keys = inval->keys; - uw_Sqlcache_flushCommitOne(cache, keys); - inval = inval->next; + ctx->cacheUpdate = NULL; + ctx->cacheUpdateTail = NULL; +} + +void uw_Sqlcache_commitUpdate(void *data) { + uw_context ctx = (uw_context)data; + uw_Sqlcache_Update *update = ctx->cacheUpdate; + while (update) { + uw_Sqlcache_Cache *cache = update->cache; + char **keys = update->keys; + if (update->value) { + uw_Sqlcache_storeCommitOne(cache, keys, update->value); + } else { + uw_Sqlcache_flushCommitOne(cache, keys); + } + update = update->next; } } char **uw_Sqlcache_copyKeys(char **keys, size_t numKeys) { char **copy = malloc(sizeof(char *) * numKeys); while (numKeys-- > 0) { - char * k = keys[numKeys]; + char *k = keys[numKeys]; copy[numKeys] = k ? strdup(k) : NULL; } return copy; } -void uw_Sqlcache_flush(uw_context ctx, uw_Sqlcache_Cache *cache, char **keys) { - //pthread_rwlock_wrlock(cache->lock); - uw_Sqlcache_Inval *inval = malloc(sizeof(uw_Sqlcache_Inval)); - inval->cache = cache; - inval->keys = uw_Sqlcache_copyKeys(keys, cache->numKeys); - inval->next = NULL; - if (ctx->inval) { - // An invalidation is already registered, so just extend it. - ctx->inval->next = inval; +void uw_Sqlcache_store(uw_context ctx, uw_Sqlcache_Cache *cache, char **keys, uw_Sqlcache_Value *value) { + uw_Sqlcache_Update *update = malloc(sizeof(uw_Sqlcache_Update)); + update->cache = cache; + update->keys = uw_Sqlcache_copyKeys(keys, cache->numKeys); + update->value = value; + update->next = NULL; + if (ctx->cacheUpdateTail) { + // An update is already registered, so just extend it. + ctx->cacheUpdateTail->next = update; } else { - uw_register_transactional(ctx, inval, uw_Sqlcache_flushCommit, NULL, uw_Sqlcache_flushFree); + ctx->cacheUpdate = update; + uw_register_transactional(ctx, ctx, uw_Sqlcache_commitUpdate, NULL, uw_Sqlcache_freeUpdate); } - // [ctx->inval] should always point to the last invalidation. - ctx->inval = inval; - //pthread_rwlock_unlock(cache->lock); + ctx->cacheUpdateTail = update; +} + +void uw_Sqlcache_flush(uw_context ctx, uw_Sqlcache_Cache *cache, char **keys) { + // A flush is represented in the queue as storing NULL. + uw_Sqlcache_store(ctx, cache, keys, NULL); } diff --git a/src/lru_cache.sml b/src/lru_cache.sml index 9d65420b..b6ffe700 100644 --- a/src/lru_cache.sml +++ b/src/lru_cache.sml @@ -65,8 +65,8 @@ fun setupQuery {index, params} = in Print.box [string ("static uw_Sqlcache_Cache cacheStruct" ^ i ^ " = {"), - (* newline, *) - (* string " .lock = PTHREAD_RWLOCK_INITIALIZER,", *) + newline, + string " .lock = PTHREAD_RWLOCK_INITIALIZER,", newline, string " .table = NULL,", newline, @@ -85,7 +85,7 @@ fun setupQuery {index, params} = newline, string (" char *ks[] = {" ^ revArgs ^ "};"), newline, - string (" uw_Sqlcache_Value *v = uw_Sqlcache_check(cache" ^ i ^ ", ks);"), + string (" uw_Sqlcache_Value *v = uw_Sqlcache_check(ctx, cache" ^ i ^ ", ks);"), newline, (* If the output is null, it means we had too much recursion, so it's a miss. *) string " if (v && v->output != NULL) {", @@ -123,7 +123,7 @@ fun setupQuery {index, params} = newline, string (" puts(\"SQLCACHE: stored " ^ i ^ ".\");"), newline, - string (" uw_Sqlcache_store(cache" ^ i ^ ", ks, v);"), + string (" uw_Sqlcache_store(ctx, cache" ^ i ^ ", ks, v);"), newline, string " return uw_unit_v;", newline, -- cgit v1.2.3