Initial version.

1 files changed, 643 insertions, 0 deletions

diff --git a/src/nxt_cache.c b/src/nxt_cache.c
new file mode 100644
index 00000000..409ba301
--- /dev/null
+++ b/src/nxt_cache.c
@@ -0,0 +1,643 @@
+
+/*
+ * Copyright (C) Igor Sysoev
+ * Copyright (C) NGINX, Inc.
+ */
+
+#include <nxt_main.h>
+
+
+/* A cache time resolution is 10ms. */
+#define                                                                       \
+nxt_cache_time(thr)                                                           \
+    (uint64_t) (nxt_thread_time(thr) * 100)
+
+
+static nxt_int_t nxt_cache_lvlhsh_test(nxt_lvlhsh_query_t *lhq, void *data);
+static nxt_work_handler_t nxt_cache_query_locked(nxt_cache_t *cache,
+    nxt_cache_query_t *q, nxt_lvlhsh_query_t *lhq);
+static nxt_work_handler_t nxt_cache_node_hold(nxt_cache_t *cache,
+    nxt_cache_query_t *q, nxt_lvlhsh_query_t *lhq);
+static nxt_work_handler_t nxt_cache_node_test(nxt_cache_t *cache,
+    nxt_cache_query_t *q);
+
+static void nxt_cache_wait_handler(nxt_thread_t *thr, void *obj, void *data);
+static void nxt_cache_timeout_handler(nxt_thread_t *thr, void *obj, void *data);
+static void nxt_cache_wake_handler(nxt_thread_t *thr, void *obj, void *data);
+static ssize_t nxt_cache_release_locked(nxt_cache_t *cache,
+    nxt_cache_query_t *q, u_char *buf, size_t size);
+
+static nxt_cache_node_t *nxt_cache_node_alloc(nxt_cache_t *cache);
+static void nxt_cache_node_free(nxt_cache_t *cache, nxt_cache_node_t *node,
+    nxt_bool_t fast);
+static nxt_cache_query_wait_t *nxt_cache_query_wait_alloc(nxt_cache_t *cache,
+    nxt_bool_t *slow);
+static void nxt_cache_query_wait_free(nxt_cache_t *cache,
+    nxt_cache_query_wait_t *qw);
+
+
+/* STUB */
+nxt_int_t nxt_cache_shm_create(nxt_mem_zone_t *pool);
+static void *nxt_cache_shm_alloc(void *data, size_t size, nxt_uint_t nalloc);
+/**/
+
+
+nxt_int_t
+nxt_cache_shm_create(nxt_mem_zone_t *mz)
+{
+    nxt_cache_t  *cache;
+
+    static const nxt_lvlhsh_proto_t  proto  nxt_aligned(64) = {
+        NXT_LVLHSH_LARGE_SLAB,
+        0,
+        nxt_cache_lvlhsh_test,
+        (nxt_lvlhsh_alloc_t) nxt_cache_shm_alloc,
+        (nxt_lvlhsh_free_t) nxt_mem_zone_free,
+    };
+
+    cache = nxt_mem_zone_zalloc(mz, sizeof(nxt_cache_t));
+
+    if (cache == NULL) {
+        return NXT_ERROR;
+    }
+
+    cache->proto = &proto;
+    cache->pool = mz;
+
+    cache->start_time = nxt_cache_time(nxt_thread());
+
+    return NXT_OK;
+}
+
+
+static void *
+nxt_cache_shm_alloc(void *data, size_t size, nxt_uint_t nalloc)
+{
+    return nxt_mem_zone_align(data, size, size);
+}
+
+
+void
+nxt_cache_init(nxt_cache_t *cache)
+{
+    static const nxt_lvlhsh_proto_t  proto  nxt_aligned(64) = {
+        NXT_LVLHSH_LARGE_MEMALIGN,
+        0,
+        nxt_cache_lvlhsh_test,
+        nxt_lvlhsh_alloc,
+        nxt_lvlhsh_free,
+    };
+
+    cache->proto = &proto;
+
+    cache->start_time = nxt_cache_time(nxt_thread());
+}
+
+
+static nxt_int_t
+nxt_cache_lvlhsh_test(nxt_lvlhsh_query_t *lhq, void *data)
+{
+    nxt_cache_node_t  *node;
+
+    node = data;
+
+    if (nxt_str_eq(&lhq->key, node->key_data, node->key_len)) {
+        return NXT_OK;
+    }
+
+    return NXT_DECLINED;
+}
+
+
+nxt_inline void
+nxt_cache_lock(nxt_cache_t *cache)
+{
+    if (cache->shared) {
+        nxt_thread_spin_lock(&cache->lock);
+    }
+}
+
+
+nxt_inline void
+nxt_cache_unlock(nxt_cache_t *cache)
+{
+    if (cache->shared) {
+        nxt_thread_spin_unlock(&cache->lock);
+    }
+}
+
+
+void
+nxt_cache_query(nxt_cache_t *cache, nxt_cache_query_t *q)
+{
+    nxt_thread_t        *thr;
+    nxt_lvlhsh_query_t  lhq;
+    nxt_work_handler_t  handler;
+
+    thr = nxt_thread();
+
+    if (cache != NULL) {
+        lhq.key_hash = nxt_murmur_hash2(q->key_data, q->key_len);
+        lhq.replace = 0;
+        lhq.key.len = q->key_len;
+        lhq.key.data = q->key_data;
+        lhq.proto = cache->proto;
+        lhq.pool = cache->pool;
+
+        q->now = nxt_cache_time(thr);
+
+        nxt_cache_lock(cache);
+
+        handler = nxt_cache_query_locked(cache, q, &lhq);
+
+        nxt_cache_unlock(cache);
+
+    } else {
+        handler = q->state->nocache_handler;
+    }
+
+    handler(thr, q, NULL);
+}
+
+
+static nxt_work_handler_t
+nxt_cache_query_locked(nxt_cache_t *cache, nxt_cache_query_t *q,
+    nxt_lvlhsh_query_t *lhq)
+{
+    nxt_int_t                ret;
+    nxt_time_t               expiry;
+    nxt_cache_node_t         *node;
+    nxt_cache_query_state_t  *state;
+
+    if (q->hold) {
+        return nxt_cache_node_hold(cache, q, lhq);
+    }
+
+    ret = nxt_lvlhsh_find(&cache->lvlhsh, lhq);
+
+    state = q->state;
+
+    if (ret != NXT_OK) {
+        /* NXT_DECLINED */
+        return state->nocache_handler;
+    }
+
+    node = lhq->value;
+    node->count++;
+    q->node = node;
+
+    expiry = cache->start_time + node->expiry;
+
+    if (q->now < expiry) {
+        return state->ready_handler;
+    }
+
+    q->stale = 1;
+
+    return state->stale_handler;
+}
+
+
+static nxt_work_handler_t
+nxt_cache_node_hold(nxt_cache_t *cache, nxt_cache_query_t *q,
+    nxt_lvlhsh_query_t *lhq)
+{
+    nxt_int_t                ret;
+    nxt_bool_t               slow;
+    nxt_cache_node_t         *node, *sentinel;
+    nxt_work_handler_t       handler;
+    nxt_cache_query_wait_t   *qw;
+    nxt_cache_query_state_t  *state;
+
+    state = q->state;
+    sentinel = nxt_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;
+    lhq->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, lhq);
+
+    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 = lhq->value;
+    node->count++;
+    q->node = node;
+
+    handler = nxt_cache_node_test(cache, q);
+    if (handler != NULL) {
+        return handler;
+    }
+
+    /* Add the node to a wait queue. */
+
+    qw = nxt_cache_query_wait_alloc(cache, &slow);
+    if (nxt_slow_path(qw == NULL)) {
+        return state->error_handler;
+    }
+
+    if (slow) {
+        /* The node state may have been 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;
+}
+
+
+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 (q->now < 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;
+            }
+        }
+
+        /* A sentinel node. */
+        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(nxt_thread_t *thr, void *obj, void *data)
+{
+    nxt_event_timer_t  *ev;
+    nxt_cache_query_t  *cq;
+
+    cq = obj;
+
+    if (cq->timeout != 0) {
+
+        ev = &cq->timer;
+
+        if (ev->state == NXT_EVENT_TIMER_DISABLED) {
+            ev->handler = nxt_cache_timeout_handler;
+            nxt_event_timer_ident(ev, -1);
+
+            nxt_event_timer_add(thr->engine, ev, cq->timeout);
+        }
+    }
+}
+
+
+static void
+nxt_cache_timeout_handler(nxt_thread_t *thr, void *obj, void *data)
+{
+    nxt_cache_query_t  *cq;
+    nxt_event_timer_t  *ev;
+
+    ev = obj;
+
+    cq = nxt_event_timer_data(ev, nxt_cache_query_t, timer);
+
+    cq->state->timeout_handler(thr, cq, NULL);
+}
+
+
+static void
+nxt_cache_wake_handler(nxt_thread_t *thr, void *obj, void *data)
+{
+    nxt_cache_t             *cache;
+    nxt_work_handler_t      handler;
+    nxt_cache_query_t       *q;
+    nxt_cache_query_wait_t  *qw;
+
+    qw = obj;
+    q = qw->query;
+    cache = qw->cache;
+
+    nxt_cache_lock(cache);
+
+    handler = nxt_cache_node_test(cache, q);
+
+    if (handler != NULL) {
+        nxt_cache_query_wait_free(cache, qw);
+
+    } else {
+        /* Wait again. */
+        qw->next = q->node->waiting;
+        q->node->waiting = qw;
+    }
+
+    nxt_cache_unlock(cache);
+
+    handler(thr, q, NULL);
+}
+
+
+nxt_int_t
+nxt_cache_update(nxt_cache_t *cache, nxt_cache_query_t *q)
+{
+    nxt_int_t           ret;
+    nxt_cache_node_t    *node;
+    nxt_lvlhsh_query_t  lhq;
+
+    node = q->node;
+
+    node->accessed = nxt_cache_time(nxt_thread()) - cache->start_time;
+
+    node->updating = 0;
+    node->count = 1;
+
+    lhq.key_hash = nxt_murmur_hash2(node->key_data, node->key_len);
+    lhq.replace = 1;
+    lhq.key.len = node->key_len;
+    lhq.key.data = node->key_data;
+    lhq.value = node;
+    lhq.proto = cache->proto;
+    lhq.pool = cache->pool;
+
+    nxt_cache_lock(cache);
+
+    ret = nxt_lvlhsh_insert(&cache->lvlhsh, &lhq);
+
+    if (nxt_fast_path(ret != NXT_OK)) {
+
+        nxt_queue_insert_head(&cache->expiry_queue, &node->link);
+
+        node = lhq.value;
+
+        if (node != NULL) {
+            /* A replaced node. */
+
+            nxt_queue_remove(&node->link);
+
+            if (node->count != 0) {
+                node->deleted = 1;
+
+            } else {
+                // delete cache node
+            }
+        }
+    }
+
+    nxt_cache_unlock(cache);
+
+    return ret;
+}
+
+
+void
+nxt_cache_release(nxt_cache_t *cache, nxt_cache_query_t *q)
+{
+    u_char        *p, *data;
+    size_t        size;
+    ssize_t       ret;
+    nxt_thread_t  *thr;
+    u_char        buf[1024];
+
+    thr = nxt_thread();
+    q->now = nxt_cache_time(thr);
+
+    p = buf;
+    size = sizeof(buf);
+
+    for ( ;; ) {
+        nxt_cache_lock(cache);
+
+        ret = nxt_cache_release_locked(cache, q, p, size);
+
+        nxt_cache_unlock(cache);
+
+        if (ret == 0) {
+            return;
+        }
+
+        size = nxt_abs(ret);
+
+        data = nxt_malloc(size);
+
+        if (data == NULL) {
+            /* TODO: retry */
+            return;
+        }
+
+        if (ret < 0) {
+            p = data;
+            continue;
+        }
+
+        if (p != data) {
+            nxt_memcpy(data, p, size);
+        }
+
+        nxt_thread_work_queue_add(thr, &thr->work_queue.main,
+                                  cache->delete_handler, data, NULL, thr->log);
+    }
+}
+
+
+static ssize_t
+nxt_cache_release_locked(nxt_cache_t *cache, nxt_cache_query_t *q,
+    u_char *buf, size_t size)
+{
+    ssize_t           ret;
+    nxt_cache_node_t  *node;
+
+    node = q->node;
+    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 = q->now - cache->start_time;
+
+        nxt_queue_remove(&node->link);
+        nxt_queue_insert_head(&cache->expiry_queue, &node->link);
+
+        return 0;
+    }
+
+    ret = 0;
+#if 0
+
+    ret = cache->delete_copy(cache, node, buf, size);
+
+    if (ret < 0) {
+        return ret;
+    }
+
+#endif
+
+    nxt_cache_node_free(cache, node, 0);
+
+    return ret;
+}
+
+
+static nxt_cache_node_t *
+nxt_cache_node_alloc(nxt_cache_t *cache)
+{
+    nxt_queue_link_t  *link;
+    nxt_cache_node_t  *node;
+
+    link = nxt_queue_first(&cache->free_nodes);
+
+    if (nxt_fast_path(link != nxt_queue_tail(&cache->free_nodes))) {
+        cache->nfree_nodes--;
+        nxt_queue_remove(link);
+
+        node = nxt_queue_link_data(link, nxt_cache_node_t, link);
+        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->link);
+        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 *slow)
+{
+    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--;
+
+        *slow = 0;
+        return qw;
+    }
+
+    nxt_cache_unlock(cache);
+
+    qw = cache->alloc(cache->data, sizeof(nxt_cache_query_wait_t));
+    *slow = 1;
+
+    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);
+}