Initial version.

1 files changed, 508 insertions, 0 deletions

diff --git a/src/nxt_file_cache.c b/src/nxt_file_cache.c
new file mode 100644
index 00000000..3af3c0c5
--- /dev/null
+++ b/src/nxt_file_cache.c
@@ -0,0 +1,508 @@
+
+/*
+ * Copyright (C) Igor Sysoev
+ * Copyright (C) NGINX, Inc.
+ */
+
+#include <nxt_main.h>
+
+
+static nxt_int_t nxt_file_cache_lvlhsh_test(nxt_lvlhsh_key_t *hkey, void *data);
+static nxt_work_handler_t nxt_file_cache_query_locked(nxt_file_cache_t *cache,
+    nxt_file_cache_query_t *q, nxt_lvlhsh_key_t *hkey);
+static nxt_work_handler_t nxt_file_cache_node_hold(nxt_file_cache_t *cache,
+    nxt_file_cache_query_t *q, nxt_lvlhsh_key_t *hkey);
+static nxt_work_handler_t nxt_file_cache_node_test(nxt_file_cache_t *cache,
+    nxt_file_cache_query_t *q);
+
+static void nxt_file_cache_wait_handler(void *data);
+static void nxt_file_cache_timeout_handler(nxt_event_timer_t *ev);
+static void nxt_file_cache_wake_handler(void *data);
+
+static nxt_file_cache_node_t *nxt_file_cache_node_alloc(nxt_cache_t *cache);
+static void nxt_file_cache_node_free(nxt_file_cache_t *cache,
+    nxt_file_cache_node_t *node, nxt_bool_t fast);
+static nxt_file_cache_query_wait_t *nxt_file_cache_query_wait_alloc(
+    nxt_file_cache_t *cache, nxt_bool_t *fast);
+static void nxt_file_cache_query_wait_free(nxt_file_cache_t *cache,
+    nxt_file_cache_query_wait_t *qw);
+static void nxt_file_cache_lock(nxt_file_cache_t *cache);
+static void nxt_file_cache_unlock(nxt_file_cache_t *cache);
+
+
+void
+nxt_file_cache_init(nxt_cache_t *cache)
+{
+    static const nxt_lvlhsh_ctx_t  ctx = {
+        nxt_file_cache_lvlhsh_test,
+        nxt_lvlhsh_alloc,
+        nxt_lvlhsh_free,
+        0,
+    };
+
+    /* lvlhsh with large first level. */
+    cache->lvlhsh.shift[1] = 10;
+
+    cache->lvlhsh.ctx = &ctx;
+
+    cache->start_time = nxt_thread_time();
+}
+
+
+static nxt_int_t
+nxt_file_cache_lvlhsh_test(nxt_lvlhsh_key_t *hkey, void *data)
+{
+    nxt_file_cache_node_t  *node;
+
+    node = data;
+
+    if (nxt_strmem_eq(&hkey->key, node->key_data, node->key_len)) {
+        return NXT_OK;
+    }
+
+    return NXT_DECLINED;
+}
+
+
+void
+nxt_file_cache_query(nxt_file_cache_t *cache, nxt_file_cache_query_t *q)
+{
+    nxt_lvlhsh_key_t    hkey;
+    nxt_work_handler_t  handler;
+
+    if (cache != NULL) {
+        hkey.key.len = q->key_len;
+        hkey.key.data = q->key_data;
+        hkey.key_hash = nxt_murmur_hash2(q->key_data, q->key_len);
+        hkey.replace = 0;
+
+        nxt_file_cache_lock(cache);
+
+        handler = nxt_file_cache_query_locked(cache, q, &hkey);
+
+        nxt_file_cache_unlock(cache);
+
+    } else {
+        handler = q->state->nocache_handler;
+    }
+
+    handler(q);
+}
+
+
+static nxt_work_handler_t
+nxt_file_cache_query_locked(nxt_file_cache_t *cache, nxt_file_cache_query_t *q,
+    nxt_lvlhsh_key_t *hkey)
+{
+    nxt_int_t                     ret;
+    nxt_bool_t                    fast;
+    nxt_work_handler_t            handler;
+    nxt_file_cache_node_t         *node, *sentinel;
+    nxt_file_cache_query_wait_t   *qw;
+    nxt_file_cache_query_state_t  *state;
+
+    state = q->state;
+    sentinel = nxt_file_cache_node_alloc(cache);
+
+    if (nxt_slow_path(sentinel == NULL)) {
+        return state->error_handler;
+    }
+
+    sentinel->key_data = q->key_data;
+    sentinel->key_len = q->key_len;
+    hkey->value = sentinel;
+
+    /*
+     * Try to insert an empty sentinel node to hold updating
+     * process if there is no existent cache node in cache.
+     */
+
+    ret = nxt_lvlhsh_insert(&cache->lvlhsh, hkey);
+
+    if (ret == NXT_OK) {
+        /* The sentinel node was successully added. */
+
+        q->node = sentinel;
+        sentinel->updating = 1;
+        return state->update_handler;
+    }
+
+    nxt_cache_node_free(cache, sentinel, 1);
+
+    if (ret == NXT_ERROR) {
+        return state->error_handler;
+    }
+
+    /* NXT_DECLINED: a cache node exists. */
+
+    node = hkey->value;
+    node->count++;
+    q->node = node;
+
+    handler = nxt_cache_node_test(cache, q);
+
+    if (handler == NULL) {
+        /* Add the node to a wait queue. */
+
+        qw = nxt_cache_query_wait_alloc(cache, &fast);
+        if (nxt_slow_path(qw == NULL)) {
+            return state->error_handler;
+        }
+
+        if (!fast) {
+            /* The node state may be changed during slow allocation. */
+            handler = nxt_cache_node_test(cache, q);
+
+            if (handler != NULL) {
+                nxt_cache_query_wait_free(cache, qw);
+                return handler;
+            }
+        }
+
+        qw->query = q;
+        qw->next = node->waiting;
+        qw->busy = 0;
+        qw->deleted = 0;
+        qw->pid = nxt_pid;
+        qw->engine = nxt_thread_event_engine();
+        qw->handler = nxt_cache_wake_handler;
+        qw->cache = cache;
+
+        node->waiting = qw;
+
+        return nxt_cache_wait_handler;
+    }
+
+    return handler;
+}
+
+
+static nxt_work_handler_t
+nxt_cache_node_test(nxt_cache_t *cache, nxt_cache_query_t *q)
+{
+    nxt_time_t               expiry;
+    nxt_cache_node_t         *node;
+    nxt_cache_query_state_t  *state;
+
+    q->stale = 0;
+    state = q->state;
+    node = q->node;
+
+    expiry = cache->start_time + node->expiry;
+
+    if (nxt_thread_time() < expiry) {
+        return state->ready_handler;
+    }
+
+    /*
+     * A valid stale or empty sentinel cache node.
+     * The sentinel node can be only in updating state.
+     */
+
+    if (node->updating) {
+
+        if (node->expiry != 0) {
+            /* A valid stale cache node. */
+
+            q->stale = 1;
+
+            if (q->use_stale) {
+                return state->stale_handler;
+            }
+        }
+
+        return NULL;
+    }
+
+    /* A valid stale cache node is not being updated now. */
+
+    q->stale = 1;
+
+    if (q->use_stale) {
+
+        if (q->update_stale) {
+            node->updating = 1;
+            return state->update_stale_handler;
+        }
+
+        return state->stale_handler;
+    }
+
+    node->updating = 1;
+    return state->update_handler;
+}
+
+
+static void
+nxt_cache_wait_handler(void *data)
+{
+    nxt_thread_t       *thr;
+    nxt_event_timer_t  *ev;
+    nxt_cache_query_t  *q;
+
+    q = data;
+
+    if (&q->timeout == 0) {
+        return;
+    }
+
+    ev = &q->timer;
+
+    if (!nxt_event_timer_is_set(ev)) {
+        thr = nxt_thread();
+        ev->log = thr->log;
+        ev->handler = nxt_cache_timeout_handler;
+        ev->data = q;
+        nxt_event_timer_ident(ev, -1);
+
+        nxt_event_timer_add(thr->engine, ev, q->timeout);
+    }
+}
+
+
+static void
+nxt_cache_timeout_handler(nxt_event_timer_t *ev)
+{
+    nxt_cache_query_t  *q;
+
+    q = ev->data;
+
+    q->state->timeout_handler(q);
+}
+
+
+static void
+nxt_cache_wake_handler(void *data)
+{
+    nxt_cache_t             *cache;
+    nxt_work_handler_t      handler;
+    nxt_cache_query_t       *q;
+    nxt_cache_query_wait_t  *qw;
+
+    qw = data;
+    q = qw->query;
+    cache = qw->cache;
+
+    nxt_cache_lock(cache);
+
+    handler = nxt_cache_node_test(cache, q);
+
+    if (handler == NULL) {
+        /* Wait again. */
+        qw->next = q->node->waiting;
+        q->node->waiting = qw;
+    }
+
+    nxt_cache_unlock(cache);
+
+    if (handler != NULL) {
+        nxt_cache_query_wait_free(cache, qw);
+    }
+
+    handler(q);
+}
+
+
+static nxt_cache_node_t *
+nxt_cache_node_alloc(nxt_cache_t *cache)
+{
+    nxt_queue_node_t  *qn;
+    nxt_cache_node_t  *node;
+
+    qn = nxt_queue_first(&cache->free_nodes);
+
+    if (nxt_fast_path(qn != nxt_queue_tail(&cache->free_nodes))) {
+        cache->nfree_nodes--;
+        nxt_queue_remove(qn);
+
+        node = nxt_queue_node_data(qn, nxt_cache_node_t, queue);
+        nxt_memzero(node, sizeof(nxt_cache_node_t));
+
+        return node;
+    }
+
+    nxt_cache_unlock(cache);
+
+    node = cache->alloc(cache->data, sizeof(nxt_cache_node_t));
+
+    nxt_cache_lock(cache);
+
+    return node;
+}
+
+
+static void
+nxt_cache_node_free(nxt_cache_t *cache, nxt_cache_node_t *node, nxt_bool_t fast)
+{
+    if (fast || cache->nfree_nodes < 32) {
+        nxt_queue_insert_head(&cache->free_nodes, &node->queue);
+        cache->nfree_nodes++;
+        return;
+    }
+
+    nxt_cache_unlock(cache);
+
+    cache->free(cache->data, node);
+
+    nxt_cache_lock(cache);
+}
+
+
+static nxt_cache_query_wait_t *
+nxt_cache_query_wait_alloc(nxt_cache_t *cache, nxt_bool_t *fast)
+{
+    nxt_cache_query_wait_t  *qw;
+
+    qw = cache->free_query_wait;
+
+    if (nxt_fast_path(qw != NULL)) {
+        cache->free_query_wait = qw->next;
+        cache->nfree_query_wait--;
+
+        *fast = 1;
+        return qw;
+    }
+
+    nxt_cache_unlock(cache);
+
+    qw = cache->alloc(cache->data, sizeof(nxt_cache_query_wait_t));
+    *fast = 0;
+
+    nxt_cache_lock(cache);
+
+    return qw;
+}
+
+
+static void
+nxt_cache_query_wait_free(nxt_cache_t *cache, nxt_cache_query_wait_t *qw)
+{
+    if (cache->nfree_query_wait < 32) {
+        qw->next = cache->free_query_wait;
+        cache->free_query_wait = qw;
+        cache->nfree_query_wait++;
+        return;
+    }
+
+    nxt_cache_unlock(cache);
+
+    cache->free(cache->data, qw);
+
+    nxt_cache_lock(cache);
+}
+
+
+#if 0
+
+nxt_int_t
+nxt_cache_update(nxt_cache_t *cache, nxt_cache_node_t *node)
+{
+    nxt_lvlhsh_key_t  hkey;
+
+    if (node->expiry == 0) {
+        /* An empty sentinel node. */
+        nxt_cache_release(cache, node);
+        return;
+    }
+
+    hkey.key.len = node->key_len;
+    hkey.key.data = node->key_data;
+    hkey.key_hash = nxt_murmur_hash2(node->key_data, node->key_len);
+    hkey.replace = 1;
+    hkey.value = node;
+
+    node->count = 1;
+
+    if (nxt_lvlhsh_insert(&cache->lvlhsh, &hkey) != NXT_OK) {
+        return NXT_ERROR;
+    }
+
+    node = hkey.value;
+
+    if (node != NULL) {
+        if (node->count != 0) {
+            node->delete = 1;
+
+        } else {
+            // delete cache node
+        }
+    }
+
+    return NXT_OK;
+}
+
+#endif
+
+
+void
+nxt_cache_node_release(nxt_cache_t *cache, nxt_cache_node_t *node)
+{
+    nxt_bool_t  delete;
+
+    nxt_cache_lock(cache);
+
+    delete = nxt_cache_node_release_locked(cache, node);
+
+    nxt_cache_unlock(cache);
+
+    if (delete) {
+        nxt_thread_work_queue_add(cache->delete_handler, node);
+    }
+}
+
+
+nxt_bool_t
+nxt_cache_node_release_locked(nxt_cache_t *cache, nxt_cache_node_t *node)
+{
+#if 0
+    nxt_lvlhsh_key_t  hkey;
+#endif
+
+    node->count--;
+
+    if (node->count != 0) {
+        return 0;
+    }
+
+    if (!node->deleted) {
+        /*
+         * A cache node is locked whilst its count is non zero.
+         * To minimize number of operations the node's place in expiry
+         * queue can be updated only if the node is not currently used.
+         */
+        node->accessed = nxt_thread_time() - cache->start_time;
+
+        nxt_queue_remove(&node->queue);
+        nxt_queue_insert_head(&cache->expiry_queue, &node->queue);
+
+        return 0;
+    }
+
+#if 0
+    hkey.key.len = node->key_len;
+    hkey.key.data = node->key_data;
+    hkey.key_hash = nxt_murmur_hash2(node->key_data, node->key_len);
+
+    nxt_lvlhsh_delete(&cache->lvlhsh, &hkey);
+#endif
+
+    return 1;
+}
+
+
+static void
+nxt_file_cache_lock(nxt_file_cache_t *cache)
+{
+    if (cache->shared) {
+        nxt_thread_spin_lock(&cache->lock);
+    }
+}
+
+
+static void
+nxt_file_cache_unlock(nxt_file_cache_t *cache)
+{
+    if (cache->shared) {
+        nxt_thread_spin_unlock(&cache->lock);
+    }
+}