/** wrapper around assoc_find which does the lazy expiration/deletion logic */
item *do_item_get_notedeleted(const char *key, const size_t nkey, bool *delete_locked) {
    item *it = assoc_find(key, nkey);
    int was_found = 0;
    if (delete_locked) *delete_locked = false;

    if (settings.verbose > 2) {
        if (it == NULL) {
            fprintf(stderr, "> NOT FOUND %s", key);
        } else {
            fprintf(stderr, "> FOUND KEY %s", ITEM_key(it));
            was_found++;
        }
    }

    if (it != NULL && (it->it_flags & ITEM_DELETED)) {
        /* it's flagged as delete-locked.  let's see if that condition
           is past due, and the 5-second delete_timer just hasn't
           gotten to it yet... */
        if (!item_delete_lock_over(it)) {
            if (delete_locked) *delete_locked = true;
            it = NULL;
        }
    }

    if (it == NULL && was_found) {
        fprintf(stderr, " -nuked by delete lock");
        was_found--;
    }

    if (it != NULL && settings.oldest_live != 0 && settings.oldest_live <= current_time &&
        it->time <= settings.oldest_live) {
        do_item_unlink(it);           /* MTSAFE - cache_lock held */
        it = NULL;
    }

    if (it == NULL && was_found) {
        fprintf(stderr, " -nuked by flush");
        was_found--;
    }

    if (it != NULL && it->exptime != 0 && it->exptime <= current_time) {
        do_item_unlink(it);           /* MTSAFE - cache_lock held */
        it = NULL;
    }

    if (it == NULL && was_found) {
        fprintf(stderr, " -nuked by expire");
        was_found--;
    }

    if (it != NULL) {
        it->refcount++;
        DEBUG_REFCNT(it, '+');
    }

    if (settings.verbose > 2)
        fprintf(stderr, "\n");

    return it;
}

/** wrapper around assoc_find which does the lazy expiration logic */
hash_item *do_item_get(struct default_engine *engine,
                       const hash_key *key) {
    rel_time_t current_time = engine->server.core->get_current_time();
    hash_item *it = assoc_find(engine,
                               crc32c(hash_key_get_key(key),
                                      hash_key_get_key_len(key), 0),
                               key);
    int was_found = 0;

    if (engine->config.verbose > 2) {
        EXTENSION_LOGGER_DESCRIPTOR *logger;
        logger = (void*)engine->server.extension->get_extension(EXTENSION_LOGGER);
        if (it == NULL) {
            logger->log(EXTENSION_LOG_DEBUG, NULL,
                        "> NOT FOUND in bucket %d, %s",
                        hash_key_get_bucket_index(key),
                        hash_key_get_client_key(key));
        } else {
            logger->log(EXTENSION_LOG_DEBUG, NULL,
                        "> FOUND KEY in bucket %d, %s",
                        hash_key_get_bucket_index(item_get_key(it)),
                        hash_key_get_client_key(item_get_key(it)));
            was_found++;
        }
    }

    if (it != NULL && engine->config.oldest_live != 0 &&
        engine->config.oldest_live <= current_time &&
        it->time <= engine->config.oldest_live) {
        do_item_unlink(engine, it);           /* MTSAFE - items.lock held */
        it = NULL;
    }

    if (it == NULL && was_found) {
        EXTENSION_LOGGER_DESCRIPTOR *logger;
        logger = (void*)engine->server.extension->get_extension(EXTENSION_LOGGER);
        logger->log(EXTENSION_LOG_DEBUG, NULL, " -nuked by flush");
        was_found--;
    }

    if (it != NULL && it->exptime != 0 && it->exptime <= current_time) {
        do_item_unlink(engine, it);           /* MTSAFE - items.lock held */
        it = NULL;
    }

    if (it == NULL && was_found) {
        EXTENSION_LOGGER_DESCRIPTOR *logger;
        logger = (void*)engine->server.extension->get_extension(EXTENSION_LOGGER);
        logger->log(EXTENSION_LOG_DEBUG, NULL, " -nuked by expire");
        was_found--;
    }

    if (it != NULL) {
        it->refcount++;
        DEBUG_REFCNT(it, '+');
        do_item_update(engine, it);
    }

    return it;
}

/** wrapper around assoc_find which does the lazy expiration/deletion logic */
item *do_item_get_notedeleted(const char *key, const size_t nkey, bool *delete_locked) {
    item *it = assoc_find(key, nkey);
    if (delete_locked) *delete_locked = false;
    if (it != NULL && (it->it_flags & ITEM_DELETED)) {
        /* it's flagged as delete-locked.  let's see if that condition
           is past due, and the 5-second delete_timer just hasn't
           gotten to it yet... */
        if (!item_delete_lock_over(it)) {
            if (delete_locked) *delete_locked = true;
            it = NULL;
        }
    }
    if (it != NULL && settings.oldest_live != 0 && settings.oldest_live <= current_time &&
        it->time <= settings.oldest_live) {
        do_item_unlink(it);           /* MTSAFE - cache_lock held */
        it = NULL;
    }
    if (it != NULL && it->exptime != 0 && it->exptime <= current_time) {
        do_item_unlink(it);           /* MTSAFE - cache_lock held */
        it = NULL;
    }

    if (it != NULL) {
        it->refcount++;
        DEBUG_REFCNT(it, '+');
    }
    return it;
}

/** wrapper around assoc_find which does the lazy expiration logic */
item *do_item_get(const char *key, const size_t nkey, const uint32_t hv) {
    mutex_lock(&cache_lock);
    item *it = assoc_find(key, nkey, hv);
    if (it != NULL) {
        refcount_incr(&it->refcount);
        /* Optimization for slab reassignment. prevents popular items from
         * jamming in busy wait. Can only do this here to satisfy lock order
         * of item_lock, cache_lock, slabs_lock. */
        if (slab_rebalance_signal &&
            ((void *)it >= slab_rebal.slab_start && (void *)it < slab_rebal.slab_end)) {
            do_item_unlink_nolock(it, hv);
            do_item_remove(it);
            it = NULL;
        }
    }
    pthread_mutex_unlock(&cache_lock);
    int was_found = 0;

    if (settings.verbose > 2) {
        if (it == NULL) {
            fprintf(stderr, "> NOT FOUND %s", key);
        } else {
            fprintf(stderr, "> FOUND KEY %s", ITEM_key(it));
            was_found++;
        }
    }

    if (it != NULL) {
        if (settings.oldest_live != 0 && settings.oldest_live <= current_time &&
            it->time <= settings.oldest_live) {
            do_item_unlink(it, hv);
            do_item_remove(it);
            it = NULL;
            if (was_found) {
                fprintf(stderr, " -nuked by flush");
            }
        } else if (it->exptime != 0 && it->exptime <= current_time) {
            if (it->exptime + 10 < current_time) {
                do_item_unlink(it, hv);
                do_item_remove(it);
                if (was_found) {
                    fprintf(stderr, " -nuked by expire");
                }
            } else {
                /* re-active just expired items, to anti miss-storm */
                it->exptime = current_time + 10;
            }
            it = NULL;
        } else {
            it->it_flags |= ITEM_FETCHED;
            DEBUG_REFCNT(it, '+');
        }
    }

    if (settings.verbose > 2)
        fprintf(stderr, "\n");

    return it;
}

static hash_item *do_item_get(struct demo_engine *engine,
                              const char *key, const size_t nkey,
                              bool LRU_reposition)
{
    rel_time_t current_time = engine->server.core->get_current_time();
    hash_item *it = dm_assoc_find(engine, engine->server.core->hash(key, nkey, 0),
                                  key, nkey);

    if (it != NULL) {
        if (do_item_isvalid(engine, it, current_time)==false) {
            do_item_unlink(engine, it, ITEM_UNLINK_INVALID);
            it = NULL;
        }
    }
    if (it != NULL) {
        ITEM_REFCOUNT_INCR(it);
        DEBUG_REFCNT(it, '+');
    }

    if (engine->config.verbose > 2) {
        if (it == NULL) {
            logger->log(EXTENSION_LOG_INFO, NULL, "> NOT FOUND %s\n",
                        key);
        } else {
            logger->log(EXTENSION_LOG_INFO, NULL, "> FOUND KEY %s\n",
                        (const char*)dm_item_get_key(it));
        }
    }
    return it;
}

/*
 * Unlinks an item from the LRU and hashtable.
 */
void item_unlink(item *item) {
    uint32_t hv;
    hv = hash(ITEM_key(item), item->nkey, 0);
    item_lock(hv);
    do_item_unlink(item, hv);
    item_unlock(hv);
}

/*
 * adds a delta value to a numeric item.
 *
 * c     connection requesting the operation
 * it    item to adjust
 * incr  true to increment value, false to decrement
 * delta amount to adjust value by
 * @param ritem The resulting item after adding the delta. Only valid if
 *              ENGINE_SUCCESS is returned. Caller is responsible for calling
 *              do_item_release() on this when finished with it.
 *
 * returns a response code to send back to the client.
 */
static ENGINE_ERROR_CODE do_add_delta(struct default_engine *engine,
                                      hash_item *it, const bool incr,
                                      const int64_t delta, item** ritem,
                                      uint64_t *result, const void *cookie) {
    const char *ptr;
    uint64_t value;
    char buf[80];
    int res;

    if (it->nbytes >= (sizeof(buf) - 1)) {
        return ENGINE_EINVAL;
    }

    ptr = item_get_data(it);
    memcpy(buf, ptr, it->nbytes);
    buf[it->nbytes] = '\0';

    if (!safe_strtoull(buf, &value)) {
        return ENGINE_EINVAL;
    }

    if (incr) {
        value += delta;
    } else {
        if ((uint64_t)delta > value) {
            value = 0;
        } else {
            value -= delta;
        }
    }

    *result = value;

    res = snprintf(buf, sizeof(buf), "%" PRIu64, value);
    if (res < 0 || res >= sizeof(buf)) {
        return ENGINE_EINVAL;
    }

    if (it->refcount == 1 && res <= (int)it->nbytes) {
        /* we can do inline replacement */
        memcpy(item_get_data(it), buf, res);
        memset(item_get_data(it) + res, ' ', it->nbytes - res);
        item_set_cas(NULL, NULL, it, get_cas_id());
        *ritem = it;
    } else {
        hash_item *new_it = do_item_alloc(engine, item_get_key(it),
                                          it->flags,
                                          it->exptime, res,
                                          cookie, it->datatype);
        if (new_it == NULL) {
            do_item_unlink(engine, it);
            return ENGINE_ENOMEM;
        }
        memcpy(item_get_data(new_it), buf, res);
        do_item_replace(engine, it, new_it);
        *ritem = new_it;
    }

    return ENGINE_SUCCESS;
}

/*
 * Flushes expired items after a flush_all call
 */
void item_flush_expired(struct default_engine *engine) {
    cb_mutex_enter(&engine->items.lock);

    rel_time_t now = engine->server.core->get_current_time();
    if (now > engine->config.oldest_live) {
        engine->config.oldest_live = now - 1;
    }

    for (int ii = 0; ii < POWER_LARGEST; ii++) {
        hash_item *iter, *next;
        /*
         * The LRU is sorted in decreasing time order, and an item's
         * timestamp is never newer than its last access time, so we
         * only need to walk back until we hit an item older than the
         * oldest_live time.
         * The oldest_live checking will auto-expire the remaining items.
         */
        for (iter = engine->items.heads[ii]; iter != NULL; iter = next) {
            if (iter->time >= engine->config.oldest_live) {
                next = iter->next;
                if ((iter->iflag & ITEM_SLABBED) == 0) {
                    do_item_unlink(engine, iter);
                }
            } else {
                /* We've hit the first old item. Continue to the next queue. */
                break;
            }
        }
    }
    cb_mutex_exit(&engine->items.lock);
}

static ENGINE_ERROR_CODE item_scrub(struct default_engine *engine,
                                    hash_item *item,
                                    void *cookie) {
    rel_time_t current_time = engine->server.core->get_current_time();
    (void)cookie;
    engine->scrubber.visited++;
    /*
        scrubber is used for generic bucket deletion and scrub_cmd
        all expired or orphaned items are unlinked
    */
    if (engine->scrubber.force_delete && item->refcount > 0) {
        // warn that someone isn't releasing items before deleting their bucket.
        EXTENSION_LOGGER_DESCRIPTOR* logger;
        logger = (void*)engine->server.extension->get_extension(EXTENSION_LOGGER);
        logger->log(EXTENSION_LOG_WARNING, NULL,
                    "Bucket (%d) deletion is removing an item with refcount %d",
                     engine->bucket_id,
                     item->refcount);
    }

    if (engine->scrubber.force_delete || (item->refcount == 0 &&
       (item->exptime != 0 && item->exptime < current_time))) {
        do_item_unlink(engine, item);
        engine->scrubber.cleaned++;
    }
    return ENGINE_SUCCESS;
}

int do_item_replace(item *it, item *new_it) {
    MEMCACHED_ITEM_REPLACE(ITEM_key(it), it->nkey, it->nbytes,
                           ITEM_key(new_it), new_it->nkey, new_it->nbytes);
    assert((it->it_flags & ITEM_SLABBED) == 0);

    do_item_unlink(it);
    return do_item_link(new_it);
}

int do_item_replace(item *it, item *new_it, const uint32_t hv) {
    new_it->it_flags &= ~ITEM_FAKE_MISSED;
    MEMCACHED_ITEM_REPLACE(ITEM_key(it), it->nkey, it->nbytes,
                           ITEM_key(new_it), new_it->nkey, new_it->nbytes);
    assert((it->it_flags & ITEM_SLABBED) == 0);

    do_item_unlink(it, hv);
    return do_item_link(new_it, hv);
}

/** wrapper around assoc_find which does the lazy expiration logic */
item *do_item_get(const char *key, const size_t nkey) {
    item *it = assoc_find(key, nkey);
    if (it != NULL && settings.oldest_live != 0 && settings.oldest_live <= current_time &&
        it->time <= settings.oldest_live) {
        do_item_unlink(it);           /* MTSAFE - cache_lock held */
        it = NULL;
    }
    if (it != NULL && it->exptime != 0 && it->exptime <= current_time) {
        do_item_unlink(it);           /* MTSAFE - cache_lock held */
        it = NULL;
    }

    if (it != NULL) {
        it->refcount++;
        DEBUG_REFCNT(it, '+');
    }
    return it;
}

void LRU_list::item_unlink(base_item* item) {
    uint32_t hv;

    hv = HashTable::hash(item->data, item->nkey);
    //哈希局部锁
    hashtable.hash_lock(hv);
    do_item_unlink(item, hv);
    hashtable.hash_unlock(hv);
}

int do_item_replace(item *it, item *new_it, const uint32_t hv) {
		syslog(LOG_INFO, "[%s:%s:%d]", __FILE__, __func__, __LINE__);
    MEMCACHED_ITEM_REPLACE(ITEM_key(it), it->nkey, it->nbytes,
                           ITEM_key(new_it), new_it->nkey, new_it->nbytes);
    assert((it->it_flags & ITEM_SLABBED) == 0);

    do_item_unlink(it, hv);
    return do_item_link(new_it, hv);
}

int do_item_replace(struct default_engine *engine,
                    hash_item *it, hash_item *new_it) {
    MEMCACHED_ITEM_REPLACE(item_get_key(it), it->nkey, it->nbytes,
                           item_get_key(new_it), new_it->nkey, new_it->nbytes);
    assert((it->iflag & ITEM_SLABBED) == 0);

    do_item_unlink(engine, it);
    return do_item_link(engine, new_it);
}

//用新的item替换老的item
int do_item_replace(item *it, item *new_it, const uint32_t hv)
{
    MEMCACHED_ITEM_REPLACE(ITEM_key(it), it->nkey, it->nbytes,
                           ITEM_key(new_it), new_it->nkey, new_it->nbytes);
    assert((it->it_flags & ITEM_SLABBED) == 0);//判断it是已经分配过的，如果未分配，则断言失败

    do_item_unlink(it, hv);//删除原来的item
    return do_item_link(new_it, hv);//重新添加新的item
}

/** wrapper around assoc_find which does the lazy expiration logic */
item *do_item_get(const char *key, const size_t nkey) {
    item *it = assoc_find(key, nkey);
    int was_found = 0;

    if (settings.verbose > 2) {
        if (it == NULL) {
            fprintf(stderr, "> NOT FOUND %s", key);
        } else {
            fprintf(stderr, "> FOUND KEY %s", ITEM_key(it));
            was_found++;
        }
    }

    if (it != NULL && settings.oldest_live != 0 && settings.oldest_live <= current_time &&
        it->time <= settings.oldest_live) {
        do_item_unlink(it);           /* MTSAFE - cache_lock held */
        it = NULL;
    }

    if (it == NULL && was_found) {
        fprintf(stderr, " -nuked by flush");
        was_found--;
    }

    if (it != NULL && it->exptime != 0 && it->exptime <= current_time) {
        do_item_unlink(it);           /* MTSAFE - cache_lock held */
        it = NULL;
    }

    if (it == NULL && was_found) {
        fprintf(stderr, " -nuked by expire");
        was_found--;
    }

    if (it != NULL) {
        it->refcount++;
        DEBUG_REFCNT(it, '+');
    }

    if (settings.verbose > 2)
        fprintf(stderr, "\n");

    return it;
}

static ENGINE_ERROR_CODE do_item_dcp_step(struct default_engine *engine,
                                          struct dcp_connection *connection,
                                          const void *cookie,
                                          struct dcp_message_producers *producers)
{
    ENGINE_ERROR_CODE ret = ENGINE_DISCONNECT;

    while (connection->it == NULL) {
        if (!do_item_walk_cursor(engine, &connection->cursor, 1,
                                 item_dcp_iterfunc, connection, &ret)) {
            /* find next slab class to look at.. */
            bool linked = false;
            int ii;
            for (ii = connection->cursor.slabs_clsid + 1; ii < POWER_LARGEST && !linked;  ++ii) {
                if (engine->items.heads[ii] != NULL) {
                    /* add the item at the tail */
                    do_item_link_cursor(engine, &connection->cursor, ii);
                    linked = true;
                }
            }
            if (!linked) {
                break;
            }
        }
    }

    if (connection->it != NULL) {
        rel_time_t current_time = engine->server.core->get_current_time();
        rel_time_t exptime = connection->it->exptime;

        if (exptime != 0 && exptime < current_time) {
            const hash_key* key = item_get_key(connection->it);
            ret = producers->expiration(cookie, connection->opaque,
                                        hash_key_get_client_key(key),
                                        hash_key_get_client_key_len(key),
                                        item_get_cas(connection->it),
                                        0, 0, 0, NULL, 0);
            if (ret == ENGINE_SUCCESS) {
                do_item_unlink(engine, connection->it);
                do_item_release(engine, connection->it);
            }
        } else {
            ret = producers->mutation(cookie, connection->opaque,
                                      connection->it, 0, 0, 0, 0, NULL, 0, 0);
        }

        if (ret == ENGINE_SUCCESS) {
            connection->it = NULL;
        }
    } else {
        return ENGINE_DISCONNECT;
    }

    return ret;
}

static void do_item_replace(struct demo_engine *engine,
                            hash_item *it, hash_item *new_it)
{
    MEMCACHED_ITEM_REPLACE(dm_item_get_key(it), it->nkey, it->nbytes,
                           dm_item_get_key(new_it), new_it->nkey, new_it->nbytes);
    do_item_unlink(engine, it, ITEM_UNLINK_REPLACE);
    /* Cache item replacement does not drop the prefix item even if it's empty.
     * So, the below do_item_link function always return SUCCESS.
     */
    (void)do_item_link(engine, new_it);
}

static ENGINE_ERROR_CODE item_scrub(struct default_engine *engine,
                                    hash_item *item,
                                    void *cookie) {
    (void)cookie;
    engine->scrubber.visited++;
    rel_time_t current_time = engine->server.core->get_current_time();
    if (item->refcount == 0 &&
        (item->exptime != 0 && item->exptime < current_time)) {
        do_item_unlink(engine, item);
        engine->scrubber.cleaned++;
    }
    return ENGINE_SUCCESS;
}