void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
    redisClient *c = (redisClient*) privdata;
    char buf[REDIS_IOBUF_LEN];
    int nread;
    REDIS_NOTUSED(el);
    REDIS_NOTUSED(mask);

    nread = read(fd, buf, REDIS_IOBUF_LEN);
    if (nread == -1) {
        if (errno == EAGAIN) {
            nread = 0;
        } else {
            redisLog(REDIS_VERBOSE, "Reading from client: %s",strerror(errno));
            freeClient(c);
            return;
        }
    } else if (nread == 0) {
        redisLog(REDIS_VERBOSE, "Client closed connection");
        freeClient(c);
        return;
    }
    if (nread) {
        c->querybuf = sdscatlen(c->querybuf, buf, nread);
        c->lastinteraction = time(NULL);
    } else {
        return;
    }
    processInputBuffer(c);
}

/*
 * 脚本超时钩子
 */
void luaMaskCountHook(lua_State *lua, lua_Debug *ar)
{
    long long elapsed;
    REDIS_NOTUSED(ar);
    REDIS_NOTUSED(lua);

    // 计算已执行时间
    elapsed = (ustime()/1000) - server.lua_time_start;
    if (elapsed >= server.lua_time_limit && server.lua_timedout == 0)
    {
        redisLog(REDIS_WARNING,"Lua slow script detected: still in execution after %lld milliseconds. You can try killing the script using the SCRIPT KILL command.",elapsed);
        // 已超时
        server.lua_timedout = 1;
        /* Once the script timeouts we reenter the event loop to permit others
         * to call SCRIPT KILL or SHUTDOWN NOSAVE if needed. For this reason
         * we need to mask the client executing the script from the event loop.
         * If we don't do that the client may disconnect and could no longer be
         * here when the EVAL command will return. */
        // 当脚本运行超时时，将正在执行的客户端从读事件中移除
        // 并允许其他客户端执行 SCRIPT KILL 或者 SHUTDOWN NOSAVE
        aeDeleteFileEvent(server.el, server.lua_caller->fd, AE_READABLE);
    }
    if (server.lua_timedout)
        // 在脚本上下文中，启动文件事件处理（等待 SCRIPT KILL 或 SHUTDOWN NOSAVE）
        aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT);
    if (server.lua_kill)
    {
        // 杀死脚本
        redisLog(REDIS_WARNING,"Lua script killed by user with SCRIPT KILL.");
        lua_pushstring(lua,"Script killed by user with SCRIPT KILL...");
        lua_error(lua);
    }
}

// 判断超时，并作超时的处理
void luaMaskCountHook(lua_State *lua, lua_Debug *ar) {
    long long elapsed;
    REDIS_NOTUSED(ar);
    REDIS_NOTUSED(lua);

    // 计算已经超时的时间
    elapsed = (ustime()/1000) - server.lua_time_start;
    if (elapsed >= server.lua_time_limit && server.lua_timedout == 0) {
        redisLog(REDIS_WARNING,"Lua slow script detected: still in execution after %lld milliseconds. You can try killing the script using the SCRIPT KILL command.",elapsed);
        server.lua_timedout = 1;

        // 超时了，关闭监听发送 lua 脚本命令的客户端
        /* Once the script timeouts we reenter the event loop to permit others
         * to call SCRIPT KILL or SHUTDOWN NOSAVE if needed. For this reason
         * we need to mask the client executing the script from the event loop.
         * If we don't do that the client may disconnect and could no longer be
         * here when the EVAL command will return. */
         aeDeleteFileEvent(server.el, server.lua_caller->fd, AE_READABLE);
    }

    // lua 脚本执行超时，redis 会检测对否有其他客户端会发送 SCRIPT KILL 命令
    // 尝试终结这个脚本的执行
    if (server.lua_timedout)
        aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT);

    // 如果 lua 脚本被停止了，强制产生错误，结束 lua
    if (server.lua_kill) {
        redisLog(REDIS_WARNING,"Lua script killed by user with SCRIPT KILL.");
        lua_pushstring(lua,"Script killed by user with SCRIPT KILL...");
        lua_error(lua);
    }
}

/* convert a linked list encoding to a list array encoding */
robj *cowListCopy(robj *val) {
    long long sttime;
    robj *newval;
    sttime = ustime();
    if (val->encoding == REDIS_ENCODING_ZIPLIST) {
        size_t bytes;
        redisLog(REDIS_NOTICE, "cowListCopy REDIS_ENCODING_ZIPLIST");
        newval = createZiplistObject();
        /* do raw memory copy */
        bytes = ziplistBlobLen(val->ptr);
        newval->ptr = zrealloc(newval->ptr, bytes);
        memcpy(newval->ptr, val->ptr, bytes);

        return newval;
    } else if (val->encoding == REDIS_ENCODING_LINKEDLIST) {
        list *list = val->ptr;
        cowListArray *lar;

        redisLog(REDIS_NOTICE, "cowListCopy REDIS_ENCODING_LINKEDLIST");
        lar = cowConvertListToArray(list);
        newval = createObject(REDIS_LIST, lar);
        newval->encoding = REDIS_ENCODING_LINKEDLISTARRAY;

        return newval;
    } else {
        /* error. unexpected encoding */
        return NULL;
    }
}

static void
smrConnect ()
{
  redisLog (REDIS_NOTICE,
	    "initialize SMR Connection, Local Dumpfile Seq Num:%lld",
	    server.smr_seqnum);
  server.smr_conn =
    smr_connect_tcp (server.smr_lport, server.smr_seqnum, &smrCb, NULL);

  if (server.smr_conn == NULL)
    {
      redisLog (REDIS_WARNING, "Failed to connect to smr, errno(%d)", errno);
      exit (1);
    }

  server.smr_fd = smr_get_poll_fd (server.smr_conn);
  if (server.smr_fd == -1)
    {
      redisLog (REDIS_WARNING, "Failed to get poll fd from smr");
      exit (1);
    }

  if (server.smr_fd > 0
      && aeCreateFileEvent (server.el, server.smr_fd, AE_READABLE,
			    processSmrCallback, NULL) == AE_ERR)
    {
      redisLog (REDIS_WARNING,
		"Unrecoverable error creating smr.fd file event.");
      smr_disconnect (server.smr_conn);
      exit (1);
    }
}

/* convert a hash dictionary encoding to a dictionary array encoding */
robj *cowZSetCopy(robj *val) {
    robj *newval;
    if (val->encoding == REDIS_ENCODING_ZIPLIST) {
        size_t bytes;
        redisLog(REDIS_NOTICE, "cowZSetCopy REDIS_ENCODING_ZIPLIST");
        newval = createZsetZiplistObject();
        /* do raw memory copy */
        bytes = ziplistBlobLen(val->ptr);
        newval->ptr = zrealloc(newval->ptr, bytes);
        memcpy(newval->ptr, val->ptr, bytes);
        return newval;
    } else if (val->encoding == REDIS_ENCODING_SKIPLIST) {
        zset *oldzs = (zset *)val->ptr;
        cowDictZArray *dar;

        redisLog(REDIS_NOTICE, "cowZSetCopy REDIS_ENCODING_SKIPLIST");
        dar = cowConvertDictToZArray(oldzs->dict);
        newval = createObject(REDIS_ZSET, dar);
        newval->encoding = REDIS_ENCODING_HTZARRAY;

        return newval;
    } else {
        /* error. unexpected encoding */
        return NULL;
    }
    return NULL;
}

/* convert a hash dictionary encoding to a dictionary array encoding */
robj *cowHashCopy(robj *val) {
    robj *newval = createHashObject();
    if (val->encoding == REDIS_ENCODING_ZIPMAP) {
        size_t bytes;
        redisLog(REDIS_NOTICE, "cowHashCopy REDIS_ENCODING_ZIPMAP");
        /* do raw memory copy */
        bytes = zipmapBlobLen(val->ptr);
        newval->ptr = zrealloc(newval->ptr, bytes);
        memcpy(newval->ptr, val->ptr, bytes);
        return newval;
    } else if (val->encoding == REDIS_ENCODING_HT) {
        dict *olddict = (dict *)val->ptr;
        cowDictArray *dar;

        redisLog(REDIS_NOTICE, "cowHashCopy REDIS_ENCODING_HT");
        dar = cowConvertDictToArray(olddict);
        newval = createObject(REDIS_HASH, dar);
        newval->encoding = REDIS_ENCODING_HTARRAY;

        return newval;
    } else {
        /* error. unexpected encoding */
        return NULL;
    }
    return NULL;
}

void initServer(){

    server.mainthread = pthread_self();
    server.clients = listCreate();
    server.el = aeCreateEventLoop();

    if (server.port != 0) {
        server.ipfd = anetTcpServer(server.neterr,server.port,server.bindaddr);
        if (server.ipfd == ANET_ERR) {
            redisLog(REDIS_WARNING, "Opening port %d: %s",
                server.port, server.neterr);
            exit(1);
        }
    }
//    if (server.unixsocket != NULL) {
//        unlink(server.unixsocket); /* don't care if this fails */
//        server.sofd = anetUnixServer(server.neterr,server.unixsocket,server.unixsocketperm);
//        if (server.sofd == ANET_ERR) {
//            redisLog(REDIS_WARNING, "Opening socket: %s", server.neterr);
//            exit(1);
//        }
//    }
    if (server.ipfd < 0 && server.sofd < 0) {
        redisLog(REDIS_WARNING, "Configured to not listen anywhere, exiting.");
        exit(1);
    }

    aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL);
    if (server.ipfd > 0 && aeCreateFileEvent(server.el,server.ipfd,AE_READABLE,
        acceptTcpHandler,NULL) == AE_ERR) oom("creating file event");
//    if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,
//        acceptUnixHandler,NULL) == AE_ERR) oom("creating file event");
}

void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
    int cport, cfd;
    char cip[128];
    redisClient *c;
    REDIS_NOTUSED(el);
    REDIS_NOTUSED(mask);
    REDIS_NOTUSED(privdata);

    cfd = anetAccept(server.neterr, fd, cip, &cport);
    if (cfd == AE_ERR) {
        redisLog(REDIS_VERBOSE,"Accepting client connection: %s", server.neterr);
        return;
    }
    redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport);
    if ((c = createClient(cfd)) == NULL) {
        redisLog(REDIS_WARNING,"Error allocating resoures for the client");
        close(cfd); /* May be already closed, just ingore errors */
        return;
    }
    /* If maxclient directive is set and this is one client more... close the
     * connection. Note that we create the client instead to check before
     * for this condition, since now the socket is already set in nonblocking
     * mode and we can send an error for free using the Kernel I/O */
    if (server.maxclients && listLength(server.clients) > server.maxclients) {
        char *err = "-ERR max number of clients reached\r\n";

        /* That's a best effort error message, don't check write errors */
        if (write(c->fd,err,strlen(err)) == -1) {
            /* Nothing to do, Just to avoid the warning... */
        }
        freeClient(c);
        return;
    }
    server.stat_numconnections++;
}

int luatriggleCreateFunction(lua_State *lua, sds funcname, robj *body) {
    sds funcdef = sdsempty();

    funcdef = sdscat(funcdef,"function ");
    funcdef = sdscatlen(funcdef,funcname,sdslen(funcname));
    funcdef = sdscatlen(funcdef,"() ",3);
    funcdef = sdscatlen(funcdef,body->ptr,sdslen(body->ptr));
    funcdef = sdscatlen(funcdef," end",4);

    redisLog(REDIS_NOTICE,"create lua function start: %s",funcdef);
    if (luaL_loadbuffer(lua,funcdef,sdslen(funcdef),"@user_script")) {
        redisLog(REDIS_WARNING,"Error compiling script (new function): %s\n",
                lua_tostring(lua,-1));
        lua_pop(lua,1);
        sdsfree(funcdef);
        return REDIS_ERR;
    }
    sdsfree(funcdef);

    redisLog(REDIS_NOTICE,"Load buffer ok !create lua function start: %s",funcdef);
    if (lua_pcall(lua,0,0,0)) {
        redisLog(REDIS_WARNING,"Error running script (new function): %s\n",
                lua_tostring(lua,-1));
        lua_pop(lua,1);
        return REDIS_ERR;
    }

    redisLog(REDIS_NOTICE,"call ok !create lua function start: %s",funcdef);

    return REDIS_OK;
}

int do_delete_event(struct redisClient *c,sds funcname)
{

    redisSrand48(0);
    server.lua_random_dirty = 0;
    server.lua_write_dirty = 0;
    lua_getglobal(server.lua,(char *)funcname);



    if (lua_isnil(server.lua,1)) {
        addReplyError(c,"no funcname triggle_scipts in lua");
        return 0;
    }
    luaTriggleSetGlobalArray(server.lua,"KEYS",c->argv+1,c->argc-1);

    redisLog(REDIS_NOTICE,"stack: %d",lua_gettop(server.lua));


#ifdef BRIDGE_DEBUG
    for(int i=0;i<c->argc-1;i++){
        redisLog(REDIS_NOTICE,"%s",(c->argv+1)[i]->ptr);
    }
#endif


    selectDb(server.lua_client,c->db->id);


    server.lua_time_start = ustime()/1000;
    server.lua_kill = 0;

    if (server.lua_time_limit > 0) {
        lua_sethook(server.lua,luaMaskCountHook,LUA_MASKCOUNT,100000);
        server.lua_time_start = ustime()/1000;
    } else {
        lua_sethook(server.lua,luaMaskCountHook,0,0);
    }

    if (lua_pcall(server.lua,0,1,0)) {
        selectDb(c,server.lua_client->db->id); 
        addReplyErrorFormat(c,"Error running script (call to %s): %s\n",
                (char*)funcname, lua_tostring(server.lua,-1));
        lua_pop(server.lua,1);
        lua_gc(server.lua,LUA_GCCOLLECT,0);
        return -1;
    }
    selectDb(c,server.lua_client->db->id); 
    // luaReplyToRedisReply(c,server.lua);
    server.lua_timedout = 0;
    server.lua_caller = NULL;
    lua_gc(server.lua,LUA_GCSTEP,1);

    //for slaves
    //

    return 0; 
}

/* This function is called at the end of every background saving.
 * The argument bgsaveerr is REDIS_OK if the background saving succeeded
 * otherwise REDIS_ERR is passed to the function.
 *
 * The goal of this function is to handle slaves waiting for a successful
 * background saving in order to perform non-blocking synchronization. */
void updateSlavesWaitingBgsave(int bgsaveerr) {
//backgroundSaveDoneHandler在BGSAVE操作完成后，调用这里来处理可能的从库事件。
    listNode *ln;
    int startbgsave = 0;
    listIter li;

    listRewind(server.slaves,&li);
    while((ln = listNext(&li))) {//循环遍历每一个从库，查看其状态进行相应的处理。
        redisClient *slave = ln->value;

        if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
            startbgsave = 1;//刚才在做bgsave的时候，有客户端来请求sync同步，但是我没有理他，现在得给他准备了。
            slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;//修改这个状态后，新的写入操作会记录到这个连接的缓存里
        } else if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) {
        //后台保存完成，下面需要发送rdb文件了，丫的够大的
            struct redis_stat buf;

            if (bgsaveerr != REDIS_OK) {
                freeClient(slave);
                redisLog(REDIS_WARNING,"SYNC failed. BGSAVE child returned an error");
                continue;
            }
			//打开这个rdb_filename,要准备给这个slave发送数据了。
            if ((slave->repldbfd = open(server.rdb_filename,O_RDONLY)) == -1 ||
                redis_fstat(slave->repldbfd,&buf) == -1) {
                freeClient(slave);
                redisLog(REDIS_WARNING,"SYNC failed. Can't open/stat DB after BGSAVE: %s", strerror(errno));
                continue;
            }
            slave->repldboff = 0;
            slave->repldbsize = buf.st_size;
//记住此时slave->repldbfd没有关闭，可写事件的时候就不需要打开了。
            slave->replstate = REDIS_REPL_SEND_BULK;
            aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);//删掉之前的可写回调，注册为sendBulkToSlave
            if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendBulkToSlave, slave) == AE_ERR) {
                freeClient(slave);
                continue;
            }
        }
    }
    if (startbgsave) {//悲剧，又有要sync的，还得保存一次。
        if (rdbSaveBackground(server.rdb_filename) != REDIS_OK) {
            listIter li;

            listRewind(server.slaves,&li);
            redisLog(REDIS_WARNING,"SYNC failed. BGSAVE failed");
            while((ln = listNext(&li))) {
                redisClient *slave = ln->value;
				//这下面似乎有问题，replstate已经在上面被设置为了_END。https://github.com/antirez/redis/issues/1308
                if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START)
                    freeClient(slave);
            }
        }
    }
}

void *IOThreadEntryPoint(void *arg) {
    iojob *j;
    listNode *ln;
    REDIS_NOTUSED(arg);

    pthread_detach(pthread_self());
    while(1) {
        /* Get a new job to process */
        lockThreadedIO();
        if (listLength(server.io_newjobs) == 0) {
            /* No new jobs in queue, exit. */
            redisLog(REDIS_DEBUG,"Thread %ld exiting, nothing to do",
                (long) pthread_self());
            server.io_active_threads--;
            unlockThreadedIO();
            return NULL;
        }
        ln = listFirst(server.io_newjobs);
        j = ln->value;
        listDelNode(server.io_newjobs,ln);
        /* Add the job in the processing queue */
        j->thread = pthread_self();
        listAddNodeTail(server.io_processing,j);
        ln = listLast(server.io_processing); /* We use ln later to remove it */
        unlockThreadedIO();
        redisLog(REDIS_DEBUG,"Thread %ld got a new job (type %d): %p about key '%s'",
            (long) pthread_self(), j->type, (void*)j, (char*)j->key->ptr);

        /* Process the Job */
        if (j->type == REDIS_IOJOB_LOAD) {
            vmpointer *vp = (vmpointer*)j->id;
            j->val = vmReadObjectFromSwap(j->page,vp->vtype);
        } else if (j->type == REDIS_IOJOB_PREPARE_SWAP) {
            j->pages = rdbSavedObjectPages(j->val);
        } else if (j->type == REDIS_IOJOB_DO_SWAP) {
            if (vmWriteObjectOnSwap(j->val,j->page) == REDIS_ERR)
                j->canceled = 1;
        }

        /* Done: insert the job into the processed queue */
        redisLog(REDIS_DEBUG,"Thread %ld completed the job: %p (key %s)",
            (long) pthread_self(), (void*)j, (char*)j->key->ptr);

        lockThreadedIO();
        listDelNode(server.io_processing,ln);
        listAddNodeTail(server.io_processed,j);
        unlockThreadedIO();

        /* Signal the main thread there is new stuff to process */
        redisAssert(write(server.io_ready_pipe_write,"x",1) == 1);
    }
    return NULL; /* never reached */
}

void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
    redisClient *slave = privdata;
    REDIS_NOTUSED(el);
    REDIS_NOTUSED(mask);
    char buf[REDIS_IOBUF_LEN];
    ssize_t nwritten, buflen;

    if (slave->repldboff == 0) {//还没发送一个字节，那么需要将总文件大小发送给slave。
        /* Write the bulk write count before to transfer the DB. In theory here
         * we don't know how much room there is in the output buffer of the
         * socket, but in practice SO_SNDLOWAT (the minimum count for output
         * operations) will never be smaller than the few bytes we need. */
        sds bulkcount;
//设置为RDB文件的大小。上面注释的意思是，咱们这里是进行异步发送的，
//连接可写并不代表可以写这么多字节，但是实在下面的字符串太短，实际上没问题的，能发送出去的，TCP来说SO_SNDLOWAT默认为2048
//所以这里不进行部分发送的处理了。不然麻烦点。
        bulkcount = sdscatprintf(sdsempty(),"$%lld\r\n",(unsigned long long) slave->repldbsize);
        if (write(fd,bulkcount,sdslen(bulkcount)) != (signed)sdslen(bulkcount))
        {
            sdsfree(bulkcount);
            freeClient(slave);
            return;
        }
        sdsfree(bulkcount);
    }
    lseek(slave->repldbfd,slave->repldboff,SEEK_SET);
    buflen = read(slave->repldbfd,buf,REDIS_IOBUF_LEN);//跳到未发送的位置，一次读取16K字节
    if (buflen <= 0) {
        redisLog(REDIS_WARNING,"Read error sending DB to slave: %s",
            (buflen == 0) ? "premature EOF" : strerror(errno));
        freeClient(slave);//出错都不给客户端一点错误的···不过还好，之前发送过长度了的，slave会超时了的。
        return;
    }
    if ((nwritten = write(fd,buf,buflen)) == -1) {
        redisLog(REDIS_VERBOSE,"Write error sending DB to slave: %s", strerror(errno));
        freeClient(slave);
        return;
    }
    slave->repldboff += nwritten;//向后移动指针，可能没有全部发送完毕。
    if (slave->repldboff == slave->repldbsize) {//发送完毕了，可以关闭RDB快照文件了。
        close(slave->repldbfd);
        slave->repldbfd = -1;
		//删除当前的这个可写回调sendBulkToSlave，注册一个新的sendReplyToClient可写回调，这样就能将增量的写日志发送给slave了。
        aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
        slave->replstate = REDIS_REPL_ONLINE;
        if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendReplyToClient, slave) == AE_ERR) {
            freeClient(slave);
            return;
        }
        redisLog(REDIS_NOTICE,"Synchronization with slave succeeded");
    }
}

/* This is how rewriting of the append only file in background works:
 *
 * 1) The user calls BGREWRITEAOF
 * 2) Redis calls this function, that forks():
 *    2a) the child rewrite the append only file in a temp file.
 *    2b) the parent accumulates differences in server.aof_rewrite_buf.
 * 3) When the child finished '2a' exists.
 * 4) The parent will trap the exit code, if it's OK, will append the
 *    data accumulated into server.aof_rewrite_buf into the temp file, and
 *    finally will rename(2) the temp file in the actual file name.
 *    The the new file is reopened as the new append only file. Profit!
 */
int rewriteAppendOnlyFileBackground(void) {
    pid_t childpid;
    long long start;

    if (server.aof_child_pid != -1) return REDIS_ERR;
    start = ustime();
    if ((childpid = fork()) == 0) {
        char tmpfile[256];

        /* Child */
        closeListeningSockets(0);
        redisSetProcTitle("redis-aof-rewrite");
        snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
        if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {
            size_t private_dirty = zmalloc_get_private_dirty();

            if (private_dirty) {
                redisLog(REDIS_NOTICE,
                    "AOF rewrite: %zu MB of memory used by copy-on-write",
                    private_dirty/(1024*1024));
            }
            exitFromChild(0);
        } else {
            exitFromChild(1);
        }
    } else {
        /* Parent */
        server.stat_fork_time = ustime()-start;
        server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
        latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);
        if (childpid == -1) {
            redisLog(REDIS_WARNING,
                "Can't rewrite append only file in background: fork: %s",
                strerror(errno));
            return REDIS_ERR;
        }
        redisLog(REDIS_NOTICE,
            "Background append only file rewriting started by pid %d",childpid);
        server.aof_rewrite_scheduled = 0;
        server.aof_rewrite_time_start = time(NULL);
        server.aof_child_pid = childpid;
        updateDictResizePolicy();
        /* We set appendseldb to -1 in order to force the next call to the
         * feedAppendOnlyFile() to issue a SELECT command, so the differences
         * accumulated by the parent into server.aof_rewrite_buf will start
         * with a SELECT statement and it will be safe to merge. */
        server.aof_selected_db = -1;
        replicationScriptCacheFlush();
        return REDIS_OK;
    }
    return REDIS_OK; /* unreached */
}

/* A background append only file rewriting (BGREWRITEAOF) terminated its work.
 * Handle this. */
void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
    if (!bysignal && exitcode == 0) {
        int fd;
        char tmpfile[256];

        redisLog(REDIS_NOTICE,
            "Background append only file rewriting terminated with success");
        /* Now it's time to flush the differences accumulated by the parent */
        snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) server.bgrewritechildpid);
        fd = open(tmpfile,O_WRONLY|O_APPEND);
        if (fd == -1) {
            redisLog(REDIS_WARNING, "Not able to open the temp append only file produced by the child: %s", strerror(errno));
            goto cleanup;
        }
        /* Flush our data... */
        if (write(fd,server.bgrewritebuf,sdslen(server.bgrewritebuf)) !=
                (signed) sdslen(server.bgrewritebuf)) {
            redisLog(REDIS_WARNING, "Error or short write trying to flush the parent diff of the append log file in the child temp file: %s", strerror(errno));
            close(fd);
            goto cleanup;
        }
        redisLog(REDIS_NOTICE,"Parent diff flushed into the new append log file with success (%lu bytes)",sdslen(server.bgrewritebuf));
        /* Now our work is to rename the temp file into the stable file. And
         * switch the file descriptor used by the server for append only. */
        if (rename(tmpfile,server.appendfilename) == -1) {
            redisLog(REDIS_WARNING,"Can't rename the temp append only file into the stable one: %s", strerror(errno));
            close(fd);
            goto cleanup;
        }
        /* Mission completed... almost */
        redisLog(REDIS_NOTICE,"Append only file successfully rewritten.");
        if (server.appendfd != -1) {
            /* If append only is actually enabled... */
            close(server.appendfd);
            server.appendfd = fd;
            if (server.appendfsync != APPENDFSYNC_NO) aof_fsync(fd);
            server.appendseldb = -1; /* Make sure it will issue SELECT */
            redisLog(REDIS_NOTICE,"The new append only file was selected for future appends.");
        } else {
            /* If append only is disabled we just generate a dump in this
             * format. Why not? */
            close(fd);
        }
    } else if (!bysignal && exitcode != 0) {
        redisLog(REDIS_WARNING, "Background append only file rewriting error");
    } else {
        redisLog(REDIS_WARNING,
            "Background append only file rewriting terminated by signal %d",
            bysignal);
    }
cleanup:
    sdsfree(server.bgrewritebuf);
    server.bgrewritebuf = sdsempty();
    aofRemoveTempFile(server.bgrewritechildpid);
    server.bgrewritechildpid = -1;
}

int main(int argv, char *argc){
	initServerConfig();
	initServer();

    if (server.ipfd > 0)
        redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port);
    if (server.sofd > 0)
        redisLog(REDIS_NOTICE,"The server is now ready to accept connections at %s", server.unixsocket);
//    aeSetBeforeSleepProc(server.el,beforeSleep);
    aeMain(server.el);
    aeDeleteEventLoop(server.el);

	return EXIT_SUCCESS;
}

/* Read the specified amount of bytes from 'fd'. If all the bytes are read
 * within 'timeout' milliseconds the operation succeed and 'size' is returned.
 * Otherwise the operation fails, -1 is returned, and an unspecified amount of
 * data could be read from the file descriptor. */
ssize_t syncRead(int fd, char *ptr, ssize_t size, long long timeout) {
    ssize_t nread, totread = 0;
    long long start = mstime();
    long long remaining = timeout;

    if (size == 0) return 0;
    while(1) {
        long long wait = (remaining > REDIS_SYNCIO_RESOLUTION) ?
                          remaining : REDIS_SYNCIO_RESOLUTION;
        long long elapsed;

        /* Optimistically try to read before checking if the file descriptor
         * is actually readable. At worst we get EAGAIN. */
#ifdef _WIN32
        nread = recv((SOCKET)fd,ptr,size,0);
#else
        nread = read(fd,ptr,size);
#endif
        if (nread == 0) {
            redisLog(REDIS_WARNING,"syncRead returned 0");
            return -1; /* short read. */
        }
        if (nread == -1) {
#ifdef _WIN32
            errno = WSAGetLastError();
            if ((errno == ENOENT) || (errno == WSAEWOULDBLOCK)) {
                errno = EAGAIN;
            }
#endif
            if (errno != EAGAIN) {
            redisLog(REDIS_WARNING,"syncRead returned -1 : %d", errno);
                return -1;
            }
        } else {
            ptr += nread;
            size -= nread;
            totread += nread;
        }
        if (size == 0) return totread;

        /* Wait */
        aeWait(fd,AE_READABLE,wait);
        elapsed = mstime() - start;
        if (elapsed >= timeout) {
            errno = ETIMEDOUT;
            return -1;
        }
        remaining = timeout - elapsed;
    }
}

void whitelistJob() {
	if (server.whitelist_switch == WHITELIST_ON) {
		pthread_t thread;

		if(pthread_create(&thread,NULL,intervalGetWhitelist,NULL) != 0)
		{
			redisLog(REDIS_WARNING,"Fatal:Can't initialize the whitelist thread.");
			exit(1);
		}
		return;
	}
	redisLog(REDIS_NOTICE, "whitelist : off");
	return;
}

//杀死线程，目前redis只有在崩溃时才调用该函数
void bioKillThreads(void) {
    int err, j;

    for (j = 0; j < REDIS_BIO_NUM_OPS; j++) {
        if (pthread_cancel(bio_threads[j]) == 0) {
            if ((err = pthread_join(bio_threads[j],NULL)) != 0) {
                redisLog(REDIS_WARNING,
                    "Bio thread for job type #%d can be joined: %s",
                        j, strerror(err));
            } else {
                redisLog(REDIS_WARNING,
                    "Bio thread for job type #%d terminated",j);
            }
        }
    }
}