/* Delete a key, value, and associated expiration entry if any, from the DB */
int dbDelete(redisDb *db, robj *key) {
    /* Deleting an entry from the expires dict will not free the sds of
     * the key, because it is shared with the main dictionary. */
    if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr);
    if (dictDelete(db->dict,key->ptr) == DICT_OK) {
        return 1;
    } else {
        return 0;
    }
}

/* Delete a key, value, and associated expiration entry if any, from the DB */
int dbDelete(redisDb *db, robj *key) {
    /* If VM is enabled make sure to awake waiting clients for this key:
     * deleting the key will kill the I/O thread bringing the key from swap
     * to memory, so the client will never be notified and unblocked if we
     * don't do it now. */
    if (server.vm_enabled) handleClientsBlockedOnSwappedKey(db,key);
    /* Deleting an entry from the expires dict will not free the sds of
     * the key, because it is shared with the main dictionary. */
    if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr);
    return dictDelete(db->dict,key->ptr) == DICT_OK;
}

/* Delete a key, value, and associated expiration entry if any, from the DB */
int dbSyncDelete(redisDb *db, robj *key) {
    /* Deleting an entry from the expires dict will not free the sds of
     * the key, because it is shared with the main dictionary. */
    if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr);
    if (dictDelete(db->dict,key->ptr) == DICT_OK) {
        if (server.cluster_enabled) slotToKeyDel(key);
        return 1;
    } else {
        return 0;
    }
}

/* Handle a response to a given request. if this is a quorum setting, choose the
 * right response. Then make sure all the requests are satisfied in a fragmented
 * request scenario and then use the post coalesce logic to cook up a combined
 * response
 */
static rstatus_t
client_handle_response(struct conn *conn, msgid_t reqid, struct msg *rsp)
{
    ASSERT_LOG(!rsp->peer, "response %lu:%lu has peer set",
               rsp->id, rsp->parent_id);

    // now the handler owns the response.
    ASSERT(conn->type == CONN_CLIENT);
    // Fetch the original request
    struct msg *req = dictFetchValue(conn->outstanding_msgs_dict, &reqid);
    if (!req) {
        log_notice("looks like we already cleanedup the request for %d", reqid);
        rsp_put(rsp);
        return DN_OK;
    }
    // we have to submit the response irrespective of the unref status.
    rstatus_t status = msg_handle_response(req, rsp);
    if (conn->waiting_to_unref) {
        // dont care about the status.
        if (req->awaiting_rsps)
            return DN_OK;
        // all responses received
        dictDelete(conn->outstanding_msgs_dict, &reqid);
        log_info("Putting req %d", req->id);
        req_put(req);
        client_unref_internal_try_put(conn);
        return DN_OK;
    }
    if (status == DN_NOOPS) {
        // by now the response is dropped
        if (!req->awaiting_rsps) {
            // if we have sent the response for this request or the connection
            // is closed and we are just waiting to drain off the messages.
            if (req->rsp_sent) {
                dictDelete(conn->outstanding_msgs_dict, &reqid);
                log_info("Putting req %d", req->id);
                req_put(req);
            }
        }
    } else if (status == DN_OK) {
        g_pre_coalesce(req->selected_rsp);
        if (req_done(conn, req)) {
            struct context *ctx = conn_to_ctx(conn);
            status = event_add_out(ctx->evb, conn);
            if (status != DN_OK) {
                conn->err = errno;
            }
        }
    }
    return status;
}

static int deleteKey(redisDb *db, robj *key) {
    int retval;

    /* We need to protect key from destruction: after the first dictDelete()
     * it may happen that 'key' is no longer valid if we don't increment
     * it's count. This may happen when we get the object reference directly
     * from the hash table with dictRandomKey() or dict iterators */
    incrRefCount(key);
    if (dictSize(db->expires)) dictDelete(db->expires,key);
    retval = dictDelete(db->dict,key);
    decrRefCount(key);

    return retval == DICT_OK;
}

/* Delete a key, value, and associated expiration entry if any, from the DB */
int dbDelete(redisDb *db, robj *key) {
    /* If diskstore is enabled make sure to awake waiting clients for this key
     * as it is not really useful to wait for a key already deleted to be
     * loaded from disk. */
    if (server.ds_enabled) {
        handleClientsBlockedOnSwappedKey(db,key);
        cacheSetKeyDoesNotExist(db,key);
    }

    /* Deleting an entry from the expires dict will not free the sds of
     * the key, because it is shared with the main dictionary. */
    if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr);
    return dictDelete(db->dict,key->ptr) == DICT_OK;
}

void
rsp_send_done(struct context *ctx, struct conn *conn, struct msg *rsp)
{

    ASSERT(conn->type == CONN_CLIENT);
    ASSERT(conn->smsg == NULL);

    if (log_loggable(LOG_VVERB)) {
       log_debug(LOG_VVERB, "send done rsp %"PRIu64" on c %d", rsp->id, conn->sd);
    }

    log_debug(LOG_VERB, "conn %p rsp %p done", conn, rsp);
    struct msg *req = rsp->peer;
    ASSERT_LOG(req, "response %d does not have a corresponding request", rsp->id);
    ASSERT_LOG(!req->rsp_sent, "request %d:%d already had a response sent",
               req->id, req->parent_id);

    ASSERT(!rsp->request && req->request);
    ASSERT(req->selected_rsp == rsp);
    req->rsp_sent = 1;

    /* dequeue request from client outq */
    conn_dequeue_outq(ctx, conn, req);

    // Remove it from the dict
    if (!req->awaiting_rsps) {
        log_debug(LOG_VERB, "conn %p removing message %d:%d", conn, req->id, req->parent_id);
        dictDelete(conn->outstanding_msgs_dict, &req->id);
        req_put(req);
    } else {
        log_info("req %d:%d still awaiting rsps %d", req->id, req->parent_id,
                  req->awaiting_rsps);
    }
}

/* Remove the 'key' from the list of blocked keys for a given client.
 *
 * The function returns 1 when there are no longer blocking keys after
 * the current one was removed (and the client can be unblocked). */
int dontWaitForSwappedKey(redisClient *c, robj *key) {
    list *l;
    listNode *ln;
    listIter li;
    struct dictEntry *de;

    /* The key object might be destroyed when deleted from the c->io_keys
     * list (and the "key" argument is physically the same object as the
     * object inside the list), so we need to protect it. */
    incrRefCount(key);

    /* Remove the key from the list of keys this client is waiting for. */
    listRewind(c->io_keys,&li);
    while ((ln = listNext(&li)) != NULL) {
        if (equalStringObjects(ln->value,key)) {
            listDelNode(c->io_keys,ln);
            break;
        }
    }
    redisAssert(ln != NULL);

    /* Remove the client form the key => waiting clients map. */
    de = dictFind(c->db->io_keys,key);
    redisAssert(de != NULL);
    l = dictGetEntryVal(de);
    ln = listSearchKey(l,c);
    redisAssert(ln != NULL);
    listDelNode(l,ln);
    if (listLength(l) == 0)
        dictDelete(c->db->io_keys,key);

    decrRefCount(key);
    return listLength(c->io_keys) == 0;
}

int main(int argc, char *argv[]) 
{
    int ret;
    dict *d = dictCreate(&testDictType, NULL);
    assert(d);
    Key_t *k = (Key_t*)malloc(sizeof(*k)); 
    k->laddr = 112;
    k->raddr = 112;
    k->lport = 1123;
    k->rport = 3306;

    Val_t *v = (Val_t*)malloc(sizeof(*v)); 
    v->v = malloc(100);
    snprintf(v->v, 100, "%s", "abcdefg");
    
    ret = dictAdd(d, k, v);
    assert(ret == DICT_OK);

    Val_t *v2 = dictFetchValue(d, k);

    assert(0 == strcmp(v2->v, v->v));

    printf("%d-%s-%s\n", ret, v->v, v2->v);

    dictPrintStats(d);

    dictDelete(d, k);

    dictPrintStats(d);

    dictRelease(d);

    return 0;
}

static rstatus_t
dnode_client_handle_response(struct conn *conn, msgid_t reqid, struct msg *rsp)
{
    // Forward the response to the caller which is client connection.
    rstatus_t status = DN_OK;
    struct context *ctx = conn_to_ctx(conn);

    ASSERT(conn->type == CONN_DNODE_PEER_CLIENT);
    // Fetch the original request
    struct msg *req = dictFetchValue(conn->outstanding_msgs_dict, &reqid);
    if (!req) {
        log_notice("looks like we already cleanedup the request for %d", reqid);
        rsp_put(rsp);
        return DN_OK;
    }

    // dnode client has no extra logic of coalescing etc like the client/coordinator.
    // Hence all work for this request is done at this time
    ASSERT_LOG(!req->peer, "req %lu:%lu has peer set", req->id, req->parent_id);
    req->selected_rsp = rsp;
    rsp->peer = req;

    // Remove the message from the hash table. 
    dictDelete(conn->outstanding_msgs_dict, &reqid);

    // If this request is first in the out queue, then the connection is ready,
    // add the connection to epoll for writing
    if (conn_is_req_first_in_outqueue(conn, req)) {
        status = event_add_out(ctx->evb, conn);
        if (status != DN_OK) {
            conn->err = errno;
        }
    }
    return status;
}

/* Unblock a client that's waiting in a blocking operation such as BLPOP */
void unblockClientWaitingData(redisClient *c) {
    dictEntry *de;
    list *l;
    int j;

    redisAssert(c->blocking_keys != NULL);
    /* The client may wait for multiple keys, so unblock it for every key. */
    for (j = 0; j < c->blocking_keys_num; j++) {
        /* Remove this client from the list of clients waiting for this key. */
        de = dictFind(c->db->blocking_keys,c->blocking_keys[j]);
        redisAssert(de != NULL);
        l = dictGetEntryVal(de);
        listDelNode(l,listSearchKey(l,c));
        /* If the list is empty we need to remove it to avoid wasting memory */
        if (listLength(l) == 0)
            dictDelete(c->db->blocking_keys,c->blocking_keys[j]);
        decrRefCount(c->blocking_keys[j]);
    }
    /* Cleanup the client structure */
    zfree(c->blocking_keys);
    c->blocking_keys = NULL;
    c->flags &= (~REDIS_BLOCKED);
    server.blpop_blocked_clients--;
    /* We want to process data if there is some command waiting
     * in the input buffer. Note that this is safe even if
     * unblockClientWaitingData() gets called from freeClient() because
     * freeClient() will be smart enough to call this function
     * *after* c->querybuf was set to NULL. */
    if (c->querybuf && sdslen(c->querybuf) > 0) processInputBuffer(c);
}

const set *dbCreate(const sds setName)
{
    set *newSet = NULL;

    if (NULL == setName || 0 == strlen(setName))
    {
        return NULL;
    }

    if (NULL == (newSet = setCreate()))
        return NULL;

    lockWrite(sets);

    if (DICT_OK != dictAdd(sets, setName, newSet))
    {
        unlockWrite(sets);
        setDestroy(newSet);
        return NULL;
    }

    if (0 != registerSyncObject(newSet) && 1 != syncObjectIsRegistered(newSet))
    {
        unlockWrite(sets);
        dictDelete(sets, setName);
        setDestroy(newSet);
        return NULL;
    }

    unlockWrite(sets);
    return newSet;
}

//解阻塞一个正在阻塞中的client
void unblockClientWaitingData(client *c) {
    dictEntry *de;
    dictIterator *di;
    list *l;

    serverAssertWithInfo(c,NULL,dictSize(c->bpop.keys) != 0);
    //创建一个字典的迭代器，指向的是造成client阻塞的键所组成的字典
    di = dictGetIterator(c->bpop.keys);
    /* The client may wait for multiple keys, so unblock it for every key. */
    //因为client可能被多个key所阻塞，所以要遍历所有的键
    while((de = dictNext(di)) != NULL) {
        robj *key = dictGetKey(de); //获得key对象

        /* Remove this client from the list of clients waiting for this key. */
        //根据key找到对应的列表类型值，值保存着被阻塞的client，从中找c->db->blocking_keys中寻找
        l = dictFetchValue(c->db->blocking_keys,key);
        serverAssertWithInfo(c,key,l != NULL);
        // 将阻塞的client从列表中移除
        listDelNode(l,listSearchKey(l,c));
        /* If the list is empty we need to remove it to avoid wasting memory */
        //如果当前列表为空了，则从c->db->blocking_keys中将key删除
        if (listLength(l) == 0)
            dictDelete(c->db->blocking_keys,key);
    }
    dictReleaseIterator(di);    //释放迭代器

    /* Cleanup the client structure */
    //清空bpop.keys的所有节点
    dictEmpty(c->bpop.keys,NULL);
    //如果保存有新添加的元素，则应该释放
    if (c->bpop.target) {
        decrRefCount(c->bpop.target);
        c->bpop.target = NULL;
    }
}

// 撤销对这个客户端的所有 WATCH
// 清除 EXEC dirty FLAG 的任务由调用者完成
void unwatchAllKeys(redisClient *c) {
    listIter li;
    listNode *ln;

    // 没有 WATCHED KEY ，直接返回
    if (listLength(c->watched_keys) == 0) return;

    listRewind(c->watched_keys,&li);
    while((ln = listNext(&li))) {
        list *clients;
        watchedKey *wk;

        /* Lookup the watched key -> clients list and remove the client
         * from the list */
        // 将当前客户端从监视 KEY 的链表中移除
        wk = listNodeValue(ln);
        clients = dictFetchValue(wk->db->watched_keys, wk->key);
        redisAssertWithInfo(c,NULL,clients != NULL);
        listDelNode(clients,listSearchKey(clients,c));

        /* Kill the entry at all if this was the only client */
        // 如果监视 KEY 的只有这个客户端
        // 那么将链表从字典中删除
        if (listLength(clients) == 0)
            dictDelete(wk->db->watched_keys, wk->key);

        /* Remove this watched key from the client->watched list */
        // 还需要将 KEY 从 client->watched_keys 链表中移除
        listDelNode(c->watched_keys,ln);
        decrRefCount(wk->key);
        zfree(wk);
    }
}

/* Unblock a client that's waiting in a blocking operation such as BLPOP */
void unblockClientWaitingData(redisClient *c) {
    dictEntry *de;
    list *l;
    int j;

    redisAssertWithInfo(c,NULL,c->bpop.keys != NULL);
    /* The client may wait for multiple keys, so unblock it for every key. */
    for (j = 0; j < c->bpop.count; j++) {
        /* Remove this client from the list of clients waiting for this key. */
        de = dictFind(c->db->blocking_keys,c->bpop.keys[j]);
        redisAssertWithInfo(c,c->bpop.keys[j],de != NULL);
        l = dictGetVal(de);
        listDelNode(l,listSearchKey(l,c));
        /* If the list is empty we need to remove it to avoid wasting memory */
        if (listLength(l) == 0)
            dictDelete(c->db->blocking_keys,c->bpop.keys[j]);
        decrRefCount(c->bpop.keys[j]);
    }

    /* Cleanup the client structure */
    zfree(c->bpop.keys);
    c->bpop.keys = NULL;
    if (c->bpop.target) decrRefCount(c->bpop.target);
    c->bpop.target = NULL;
    c->flags &= ~REDIS_BLOCKED;
    c->flags |= REDIS_UNBLOCKED;
    server.bpop_blocked_clients--;
    listAddNodeTail(server.unblocked_clients,c);
}

 int debugger_clear_breakpoint(lua_State* L)
{
	if (lua_type(L, 1) == LUA_TNUMBER && lua_type(L, 2) == LUA_TSTRING)
	{
		dict* linedefined_dictionary = NULL;

		int line = luaL_checkint(L, 1);
		const char* source = luaL_checkstring(L, 2);
		const char* file_name = trim_path(source);
		if (file_name != NULL)
		{
			linedefined_dictionary = dictFetchValue(breakpoint_dictionary, (const void *)line);
			if (linedefined_dictionary != NULL)
			{
				breakpoint_info* bp = dictFetchValue(linedefined_dictionary, file_name);
				if (bp != NULL)
				{
					dictDelete(linedefined_dictionary, file_name);
					luaL_unref(L, LUA_REGISTRYINDEX, bp->conditionFunction);
					free(bp);
					lua_pushinteger(L, 1);
					return 1;
				}
			}
		}

	}
	lua_pushinteger(L, 0);
	return 1;
}

void zremCommand(redisClient *c) {
    robj *zsetobj;
    zset *zs;
    dictEntry *de;
    double curscore;
    int deleted;

    if ((zsetobj = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
        checkType(c,zsetobj,REDIS_ZSET)) return;

    zs = zsetobj->ptr;
    c->argv[2] = tryObjectEncoding(c->argv[2]);
    de = dictFind(zs->dict,c->argv[2]);
    if (de == NULL) {
        addReply(c,shared.czero);
        return;
    }
    /* Delete from the skiplist */
    curscore = *(double*)dictGetEntryVal(de);
    deleted = zslDelete(zs->zsl,curscore,c->argv[2]);
    redisAssert(deleted != 0);

    /* Delete from the hash table */
    dictDelete(zs->dict,c->argv[2]);
    if (htNeedsResize(zs->dict)) dictResize(zs->dict);
    if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]);
    touchWatchedKey(c->db,c->argv[1]);
    server.dirty++;
    addReply(c,shared.cone);
}

/* Delete all the elements with score between min and max from the skiplist.
 * Min and mx are inclusive, so a score >= min || score <= max is deleted.
 * Note that this function takes the reference to the hash table view of the
 * sorted set, in order to remove the elements from the hash table too. */
unsigned long zslDeleteRangeByScore(zskiplist *zsl, zrangespec range, dict *dict) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned long removed = 0;
    int i;

    x = zsl->header;
    for (i = zsl->level-1; i >= 0; i--) {
        while (x->level[i].forward && (range.minex ?
            x->level[i].forward->score <= range.min :
            x->level[i].forward->score < range.min))
                x = x->level[i].forward;
        update[i] = x;
    }

    /* Current node is the last with score < or <= min. */
    x = x->level[0].forward;

    /* Delete nodes while in range. */
    while (x && (range.maxex ? x->score < range.max : x->score <= range.max)) {
        zskiplistNode *next = x->level[0].forward;
        zslDeleteNode(zsl,x,update);
        dictDelete(dict,x->obj);
        zslFreeNode(x);
        removed++;
        x = next;
    }
    return removed;
}

/* Delete all the elements with rank between start and end from the skiplist.
 * Start and end are inclusive. Note that start and end need to be 1-based */
unsigned long zslDeleteRangeByRank(zskiplist *zsl, unsigned int start, unsigned int end, dict *dict) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned long traversed = 0, removed = 0;
    int i;

    x = zsl->header;
    for (i = zsl->level-1; i >= 0; i--) {
        while (x->level[i].forward && (traversed + x->level[i].span) < start) {
            traversed += x->level[i].span;
            x = x->level[i].forward;
        }
        update[i] = x;
    }

    traversed++;
    x = x->level[0].forward;
    while (x && traversed <= end) {
        zskiplistNode *next = x->level[0].forward;
        zslDeleteNode(zsl,x,update);
        dictDelete(dict,x->obj);
        zslFreeNode(x);
        removed++;
        traversed++;
        x = next;
    }
    return removed;
}

/* Unblock a client that's waiting in a blocking operation such as BLPOP */
void unblockClientWaitingData(redisClient *c) {
    dictEntry *de;
    dictIterator *di;
    list *l;

    redisAssertWithInfo(c,NULL,dictSize(c->bpop.keys) != 0);
    di = dictGetIterator(c->bpop.keys);
    /* The client may wait for multiple keys, so unblock it for every key. */
    while((de = dictNext(di)) != NULL) {
        robj *key = dictGetKey(de);

        /* Remove this client from the list of clients waiting for this key. */
        l = dictFetchValue(c->db->blocking_keys,key);
        redisAssertWithInfo(c,key,l != NULL);
        listDelNode(l,listSearchKey(l,c));
        /* If the list is empty we need to remove it to avoid wasting memory */
        if (listLength(l) == 0)
            dictDelete(c->db->blocking_keys,key);
    }
    dictReleaseIterator(di);

    /* Cleanup the client structure */
    dictEmpty(c->bpop.keys,NULL);
    if (c->bpop.target) {
        decrRefCount(c->bpop.target);
        c->bpop.target = NULL;
    }
    c->flags &= ~REDIS_BLOCKED;
    c->flags |= REDIS_UNBLOCKED;
    server.bpop_blocked_clients--;
    listAddNodeTail(server.unblocked_clients,c);
}