void execCommand(redisClient *c) {
    int j;
    robj **orig_argv;
    int orig_argc;

    if (!(c->flags & REDIS_MULTI)) {
        addReplySds(c,sdsnew("-ERR EXEC without MULTI\r\n"));
        return;
    }

    /* Check if we need to abort the EXEC if some WATCHed key was touched.
     * A failed EXEC will return a multi bulk nil object. */
    if (c->flags & REDIS_DIRTY_CAS) {
        freeClientMultiState(c);
        initClientMultiState(c);
        c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS);
        unwatchAllKeys(c);
        addReply(c,shared.nullmultibulk);
        return;
    }

    /* Replicate a MULTI request now that we are sure the block is executed.
     * This way we'll deliver the MULTI/..../EXEC block as a whole and
     * both the AOF and the replication link will have the same consistency
     * and atomicity guarantees. */
    execCommandReplicateMulti(c);

    /* Exec all the queued commands */
    unwatchAllKeys(c); /* Unwatch ASAP otherwise we'll waste CPU cycles */
    orig_argv = c->argv;
    orig_argc = c->argc;
    addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",c->mstate.count));
    for (j = 0; j < c->mstate.count; j++) {
        c->argc = c->mstate.commands[j].argc;
        c->argv = c->mstate.commands[j].argv;
        call(c,c->mstate.commands[j].cmd);
    }
    c->argv = orig_argv;
    c->argc = orig_argc;
    freeClientMultiState(c);
    initClientMultiState(c);
    c->flags &= ~(REDIS_MULTI|REDIS_DIRTY_CAS);
    /* Make sure the EXEC command is always replicated / AOF, since we
     * always send the MULTI command (we can't know beforehand if the
     * next operations will contain at least a modification to the DB). */
    server.dirty++;
}

/*
 * cetcd_cluster_request tries to request the whole cluster. It round-robin to next server if the request failed
 * */
cetcd_response *cetcd_cluster_request(cetcd_client *cli, cetcd_request *req) {
    int i;
    size_t count;
    cetcd_string url;
    cetcd_error *err;
    cetcd_response *resp;

    err = NULL;
    resp = NULL;
    count = cetcd_array_size(cli->addresses);
    
    for(i = 0; i < count; ++i) {
        url = sdscatprintf(sdsempty(), "http://%s/%s", (cetcd_string)cetcd_array_get(cli->addresses, cli->picked), req->uri);
        req->url = url;
        req->cli = cli;
        resp = cetcd_send_request(cli->curl, req);
        sdsfree(url);

        if(resp && resp->err && resp->err->ecode == error_send_request_failed) {
            if (i == count-1) {
                break;
            }
            /*try next*/
            cli->picked = (cli->picked + 1) % count;
            cetcd_response_release(resp);
            resp = NULL;
        } else {
            /*got response, return*/
            return resp;
        }
    }
    /*the whole cluster failed*/
    if (resp) {
        if(resp->err) {
            err = resp->err; /*remember last error*/
        }
        resp->err = calloc(1, sizeof(cetcd_error));
        resp->err->ecode = error_cluster_failed;
        resp->err->message = sdsnew("etcd_do_request: all cluster servers failed.");
        if (err) {
           resp->err->message = sdscatprintf(resp->err->message, " last error: %s", err->message);
           cetcd_error_release(err);
        }
        resp->err->cause = sdsdup(req->uri);
    }
    return resp;
}

void make_trace(char* path) {
	init_jcr(path);

	sds trace_file = sdsnew(path);

	char *p = trace_file + sdslen(trace_file) - 1;
	while (*p == '/')
		--p;
	*(p + 1) = 0;
	sdsupdatelen(trace_file);

	trace_file = sdscat(trace_file, ".trace");
	NOTICE("output to %s", trace_file);

	start_read_phase();
	start_chunk_phase();
	start_hash_phase();

	unsigned char code[41];

	FILE *fp = fopen(trace_file, "w");
	while (1) {
		struct chunk *c = sync_queue_pop(hash_queue);

		if (c == NULL) {
			break;
		}

		if (CHECK_CHUNK(c, CHUNK_FILE_START)) {
			destor_log(DESTOR_NOTICE, c->data);
			fprintf(fp, "file start %zd\n", strlen(c->data));
			fprintf(fp, "%s\n", c->data);

		} else if (CHECK_CHUNK(c, CHUNK_FILE_END)) {
			fprintf(fp, "file end\n");
		} else {
			hash2code(c->fp, code);
			code[40] = 0;
			fprintf(fp, "%s %d\n", code, c->size);
		}
		free_chunk(c);
	}

	fprintf(fp, "stream end");
	fclose(fp);

}

/* Mass-unblock clients because something changed in the instance that makes
 * blocking no longer safe. For example clients blocked in list operations
 * in an instance which turns from master to slave is unsafe, so this function
 * is called when a master turns into a slave.
 *
 * The semantics is to send an -UNBLOCKED error to the client, disconnecting
 * it at the same time. */
void disconnectAllBlockedClients(void) {
    listNode *ln;
    listIter li;

    listRewind(server.clients,&li);
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);

        if (c->flags & CLIENT_BLOCKED) {
            addReplySds(c,sdsnew(
                "-UNBLOCKED force unblock from blocking operation, "
                "instance state changed (master -> slave?)\r\n"));
            unblockClient(c);
            c->flags |= CLIENT_CLOSE_AFTER_REPLY;
        }
    }
}

/* This function does exactly the revese of the function above: it gets
 * as input an integer with the xored flags and returns a string representing
 * the selected classes. The string returned is an sds string that needs to
 * be released with sdsfree(). */
sds keyspaceEventsFlagsToString(int flags) {
    sds res;

    if ((flags & REDIS_NOTIFY_ALL) == REDIS_NOTIFY_ALL)
        return sdsnew("A");
    res = sdsempty();
    if (flags & REDIS_NOTIFY_GENERIC) res = sdscatlen(res,"g",1);
    if (flags & REDIS_NOTIFY_STRING) res = sdscatlen(res,"$",1);
    if (flags & REDIS_NOTIFY_LIST) res = sdscatlen(res,"l",1);
    if (flags & REDIS_NOTIFY_SET) res = sdscatlen(res,"s",1);
    if (flags & REDIS_NOTIFY_HASH) res = sdscatlen(res,"h",1);
    if (flags & REDIS_NOTIFY_ZSET) res = sdscatlen(res,"z",1);
    if (flags & REDIS_NOTIFY_EXPIRED) res = sdscatlen(res,"x",1);
    if (flags & REDIS_NOTIFY_EVICTED) res = sdscatlen(res,"e",1);
    if (flags & REDIS_NOTIFY_KEYSPACE) res = sdscatlen(res,"K",1);
    if (flags & REDIS_NOTIFY_KEYEVENT) res = sdscatlen(res,"E",1);
    return res;
}

static bool checkPurge() {
    listNode *ln; listIter li;
    uint32    num_ok = 0;
    sds       ds     = sdsnew("DIRTY\r\n");
    listRewind(server.slaves, &li);
    while((ln = listNext(&li))) {
        cli *slave = ln->value;
        redisReply *reply;
        int fd = remoteMessage(slave->bindaddr, slave->bindport, ds, 1, &reply);
        if (fd == -1) close(fd);
        if (reply) {
            assert(reply->type == REDIS_REPLY_INTEGER);
            if (reply->integer == server.alc.stat_num_dirty_commands) num_ok++;
        }
    }
    sdsfree(ds);
    return (server.slaves->len == num_ok);
}

S triefort_destroy(const char * const path) {
  S s = triefort_ok;

  sds spath = sdsnew(path);
  spath = sdscat(spath, "/" CONFIG_FILE_NAME);

  if (!file_exists(spath)) {
    s = triefort_err_not_a_triefort;
  } else {
    if (0 != recursive_remove(path)) {
      s = triefort_err_path_could_not_be_destroyed;
    }
  }

  sdsfree(spath);

  return s;
}

cetcd_response *cetcd_send_request(CURL *curl, cetcd_request *req) {
    CURLcode res;
    cetcd_response_parser parser;
    cetcd_response *resp;

    resp = calloc(1, sizeof(cetcd_response));
    parser.resp = resp;
    parser.st = 0; /*0 should be the start state of the state machine*/
    parser.buf = sdsempty();

    curl_easy_setopt(curl, CURLOPT_URL, req->url);
    curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, http_method[req->method]);
    if (req->method == ETCD_HTTP_PUT || req->method == ETCD_HTTP_POST) {
        curl_easy_setopt(curl, CURLOPT_POSTFIELDS, req->data);
    } else {
        /* We must clear post fields here:
         * We reuse the curl handle for all HTTP methods.
         * CURLOPT_POSTFIELDS would be set when issue a PUT request.
         * The field  pointed to the freed req->data. It would be 
         * reused by next request.
         * */
        curl_easy_setopt(curl, CURLOPT_POSTFIELDS, "");
    }
    curl_easy_setopt(curl, CURLOPT_HEADER, 1L);
    curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L);
    curl_easy_setopt(curl, CURLOPT_WRITEDATA, &parser);
    curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, cetcd_parse_response);
    curl_easy_setopt(curl, CURLOPT_VERBOSE, req->cli->settings.verbose); 
    curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, req->cli->settings.connect_timeout);

    res = curl_easy_perform(curl);

    sdsfree(parser.buf);
    if (res != CURLE_OK) {
        if (resp->err == NULL) {
            resp->err = calloc(1, sizeof(cetcd_error));
            resp->err->ecode = error_send_request_failed;
            resp->err->message = sdsnew(curl_easy_strerror(res));
            resp->err->cause = sdsdup(req->url);
        }
        return resp;
    }
    return resp;
}

vuiClient * createClient(int fd)
{
    if (server.client != NULL)
    {
        close(fd);
        return NULL;
    }
    LogInfo("creat client fd:%d", fd);
    vuiClient *c = zmalloc(sizeof(vuiClient));
    memset(c, 0, sizeof(vuiClient));

    c->querybuf = sdsempty();
    c->querymsg = sdsempty();

    c->prot.method = NULL;
    c->prot.version = NULL;
    c->prot.body = NULL;
    c->prot.lenght = 0;
    c->prot.waiting = 0;

    c->res.version = "VPC/1.0";
    c->res.code = 200;
    c->res.reason = sdsnew("OK");
    c->res.body = sdsempty();
    c->res.buf = sdsempty();

    c->jsons = listCreate();

    c->fd = fd;


    anetNonBlock(NULL,fd);
    anetEnableTcpNoDelay(NULL,fd);
    if (aeCreateFileEvent(server.el, fd, AE_READABLE,
                readQueryFromClient, c) == AE_ERR)
    {
        close(fd);
        zfree(c);
        return NULL;
    }
    server.client = c;
    return c;

}

/*
 * parse master info comman and get oplog.last
 */
static void
test_repl_parse_master_info()
{
    sds buf = sdsnew("#repl\r\n"            \
                     "oplog.first:3\r\n"    \
                     "oplog.last:5\r\n"  );

    sds value = repl_parse_master_info(buf, "oplog.last");
    TEST_ASSERT("parse master info",
              strcmp(value, "5") == 0);
    sdsfree(value);

    value = repl_parse_master_info(buf, "oplog.last2");
    TEST_ASSERT("parse master info",
              value == NULL);
    sdsfree(value);

    sdsfree(buf);
}

/* This should be called from any function PUSHing into lists.
 * 'c' is the "pushing client", 'key' is the key it is pushing data against,
 * 'ele' is the element pushed.
 *
 * If the function returns 0 there was no client waiting for a list push
 * against this key.
 *
 * If the function returns 1 there was a client waiting for a list push
 * against this key, the element was passed to this client thus it's not
 * needed to actually add it to the list and the caller should return asap. */
int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele) {
    struct dictEntry *de;
    redisClient *receiver;
    list *l;
    listNode *ln;

    de = dictFind(c->db->blocking_keys,key);
    if (de == NULL) return 0;
    l = dictGetEntryVal(de);
    ln = listFirst(l);
    redisAssert(ln != NULL);
    receiver = ln->value;

    addReplySds(receiver,sdsnew("*2\r\n"));
    addReplyBulk(receiver,key);
    addReplyBulk(receiver,ele);
    unblockClientWaitingData(receiver);
    return 1;
}

/* Populates the Redis Command Table starting from the hard coded list
 * we have in the rct_command.h file. */
void populateCommandTable(dict *commands) {
    
    int ret;
    int j;
    int numcommands;

    if(commands == NULL)
    {
        return;
    }

    numcommands = sizeof(rctCommandTable)/sizeof(RCTCommand);

    for (j = 0; j < numcommands; j++) {
        RCTCommand *c = rctCommandTable+j;

        ret = dictAdd(commands, sdsnew(c->name), c);
    }
}

sds _masterGetStatus() {
    /*TODO: calculate cache increase speed,
     * then adopt a suitable stale-cache freeing strategy
     * Three involved params:
     * (1) master sleep,
     * (2) ae loop wait,
     * and (3) number of stale entries in one ae loop
     */

    sds status = sdsempty();//sdsfromlonglong(master_total_mem);
    status = sdscatprintf(status,"TOL RAM: %-6.2lfMB\tUSED RAM: %-6.2lf\n",
                          BYTES_TO_MEGABYTES(MASTER_MAX_AVAIL_MEM),
                          BYTES_TO_MEGABYTES(master_total_mem));
#if (CCACHE_LOG_LEVEL == CCACHE_DEBUG)
    status = sdscatprintf(status,"Detail:\n");
    status = sdscatprintf(status,"%-3s %-32s: %-6s\n"," ","KEY","MEM");
    dictIterator *di = dictGetIterator(master_cache);
    dictEntry *de;
    int idx = 1;
    while((de = dictNext(di)) != NULL) {
        objSds *value = (objSds*)dictGetEntryVal(de);
        if(value) {
            if(value->ptr) {
                status = sdscatprintf(status,"%-3d %-32s: %-6ld\n",
                                        idx++,
                                        (char*)dictGetEntryKey(de),
                                        sdslen(value->ptr));
            }
            else {
                status = sdscatprintf(status,"%-3d %-32s: %-6s\n",
                                        idx++,
                                        (char*)dictGetEntryKey(de),
                                        "WAITING");
            }
        }
    }
    dictReleaseIterator(di);
#endif
    sds status_reply = sdsnew("HTTP/1.1 200 OK\r\n");
    status_reply = sdscatprintf(status_reply,"Content-Length: %ld\r\n\r\n%s",sdslen(status),status);
    sdsfree(status);
    return status_reply;
}

void slaveofCommand(redisClient *c) {
    if (!strcasecmp(c->argv[1]->ptr,"no") &&
        !strcasecmp(c->argv[2]->ptr,"one")) {
        if (server.masterhost) {
            sdsfree(server.masterhost);
            server.masterhost = NULL;
            if (server.master) freeClient(server.master);
            if (server.repl_state == REDIS_REPL_TRANSFER)
                replicationAbortSyncTransfer();
            else if (server.repl_state == REDIS_REPL_CONNECTING ||
                     server.repl_state == REDIS_REPL_RECEIVE_PONG)
                undoConnectWithMaster();
            server.repl_state = REDIS_REPL_NONE;
            redisLog(REDIS_NOTICE,"MASTER MODE enabled (user request)");
        }
    } else {
        long port;

        if ((getLongFromObjectOrReply(c, c->argv[2], &port, NULL) != REDIS_OK))
            return;

        /* Check if we are already attached to the specified slave */
        if (server.masterhost && !strcasecmp(server.masterhost,c->argv[1]->ptr)
            && server.masterport == port) {
            redisLog(REDIS_NOTICE,"SLAVE OF would result into synchronization with the master we are already connected with. No operation performed.");
            addReplySds(c,sdsnew("+OK Already connected to specified master\r\n"));
            return;
        }
        /* There was no previous master or the user specified a different one,
         * we can continue. */
        sdsfree(server.masterhost);
        server.masterhost = sdsdup(c->argv[1]->ptr);
        server.masterport = port;
        if (server.master) freeClient(server.master);
        disconnectSlaves(); /* Force our slaves to resync with us as well. */
        if (server.repl_state == REDIS_REPL_TRANSFER)
            replicationAbortSyncTransfer();
        server.repl_state = REDIS_REPL_CONNECT;
        redisLog(REDIS_NOTICE,"SLAVE OF %s:%d enabled (user request)",
            server.masterhost, server.masterport);
    }
    addReply(c,shared.ok);
}

/* Use this function if the caller has already read the data. It will
 * feed bytes to the reply parser.
 *
 * After this function is called, you may use redisContextReadReply to
 * see if there is a reply available. */
int redisBufferReadDone(redisContext *c, char *buf, int nread) {
    if (nread == -1) {
        if (errno == EAGAIN && !(c->flags & REDIS_BLOCK)) {
            /* Try again later */
        } else {
            __redisSetError(c,REDIS_ERR_IO,NULL);
            return REDIS_ERR;
        }
    } else if (nread == 0) {
        __redisSetError(c,REDIS_ERR_EOF, sdsnew("Server closed the connection"));
        return REDIS_ERR;
    } else {
        if (redisReaderFeed(c->reader,buf,nread) != REDIS_OK) {
            __redisSetError(c,c->reader->err,c->reader->errstr);
            return REDIS_ERR;
        }
    }
    return REDIS_OK;
}

static int
parse_options(int argc, char **argv)
{
	int i;

	if (argc == 1)
		usage();

	for (i = 1; i < argc; i++) {
		int lastarg = i==argc-1;

		if (!strcmp(argv[i],"-h") && !lastarg) {
			sdsfree(config.redd_ip);
			config.redd_ip = sdsnew(argv[++i]);
		} else if (!strcmp(argv[i],"-h") && lastarg) {
			usage();
		} else if (!strcmp(argv[i],"--help")) {
			usage();
		} else if (!strcmp(argv[i],"-p") && !lastarg) {
			config.redd_port = atoi(argv[++i]);
		} else if (!strcmp(argv[i],"-v") || !strcmp(argv[i], "--version")) {
			printf("red %s\n", RED_VERSION);
			exit(0);
		} else if (!strcmp(argv[i],"-y") || !strcmp(argv[i], "--yaml")) {
			config.yaml_out = 1;
		} else if (!strcmp(argv[i],"-n") || !strcmp(argv[i], "--nooutput")) {
			config.no_output = 1;
		} else {
			if (argv[i][0] == '-') {
				fprintf(stderr,
					"Unrecognized option or bad number of args for: '%s'\n",
					argv[i]);
				exit(1);
			} else {
				/* Likely the command name, stop here. */
				break;
			}
		}
	}

	return i;
}

void typeCommand(redisClient *c) {
    robj *o;
    char *type;

    o = lookupKeyRead(c->db,c->argv[1]);
    if (o == NULL) {
        type = "+none";
    } else {
        switch(o->type) {
        case REDIS_STRING: type = "+string"; break;
        case REDIS_LIST: type = "+list"; break;
        case REDIS_SET: type = "+set"; break;
        case REDIS_ZSET: type = "+zset"; break;
        case REDIS_HASH: type = "+hash"; break;
        default: type = "+unknown"; break;
        }
    }
    addReplySds(c,sdsnew(type));
    addReply(c,shared.crlf);
}

/* Use this function to handle a read event on the descriptor. It will try
 * and read some bytes from the socket and feed them to the reply parser.
 *
 * After this function is called, you may use redisContextReadReply to
 * see if there is a reply available. */
int redisBufferRead(redisContext *c) {
    char buf[2048];
    int nread = read(c->fd,buf,sizeof(buf));
    if (nread == -1) {
        if (errno == EAGAIN && !(c->flags & REDIS_BLOCK)) {
            /* Try again later */
        } else {
            __redisSetError(c,REDIS_ERR_IO,NULL);
            return REDIS_ERR;
        }
    } else if (nread == 0) {
        __redisSetError(c,REDIS_ERR_EOF,
            sdsnew("Server closed the connection"));
        return REDIS_ERR;
    } else {
        __redisCreateReplyReader(c);
        redisReplyReaderFeed(c->reader,buf,nread);
    }
    return REDIS_OK;
}

static response *resolve_stream(const char *url)
{
    request *request = http_get(url);
    response *response = http_send(request);

    if (http_read_body(response) < 0 && response->status != 302) {
        fprintf(stderr, "request failed %s", url); 
        return NULL;
    }

    sds stream_url = sdsnew(http_header(response, "Location"));

    free_response(response);

    request = http_get(stream_url);
    response = http_send(request);
    
    sdsfree(stream_url);

    return response;
}

void replicationFeedMonitors(redisClient *c, list *monitors, int dictid, robj **argv, int argc) {
    listNode *ln;
    listIter li;
    int j, port;
    sds cmdrepr = sdsnew("+");
    robj *cmdobj;
    char ip[32];
    struct timeval tv;

    gettimeofday(&tv,NULL);
    cmdrepr = sdscatprintf(cmdrepr,"%ld.%06ld ",(long)tv.tv_sec,(long)tv.tv_usec);
    if (c->flags & REDIS_LUA_CLIENT) {
        cmdrepr = sdscatprintf(cmdrepr,"[%d lua] ",dictid);
    } else if (c->flags & REDIS_UNIX_SOCKET) {
        cmdrepr = sdscatprintf(cmdrepr,"[%d unix:%s] ",dictid,server.unixsocket);
    } else {
        anetPeerToString(c->fd,ip,&port);
        cmdrepr = sdscatprintf(cmdrepr,"[%d %s:%d] ",dictid,ip,port);
    }

    for (j = 0; j < argc; j++) {
        if (argv[j]->encoding == REDIS_ENCODING_INT) {
            cmdrepr = sdscatprintf(cmdrepr, "\"%ld\"", (long)argv[j]->ptr);
        } else {
            cmdrepr = sdscatrepr(cmdrepr,(char*)argv[j]->ptr,
                        sdslen(argv[j]->ptr));
        }
        if (j != argc-1)
            cmdrepr = sdscatlen(cmdrepr," ",1);
    }
    cmdrepr = sdscatlen(cmdrepr,"\r\n",2);
    cmdobj = createObject(REDIS_STRING,cmdrepr);

    listRewind(monitors,&li);
    while((ln = listNext(&li))) {
        redisClient *monitor = ln->value;
        addReply(monitor,cmdobj);
    }
    decrRefCount(cmdobj);
}