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 sendReplyToClient(aeEventLoop *el, int fd, httpClient *c) {
    int nwritten = 0;

    CCACHE_NOTUSED(el);    
    if(c->rep) {
        sds obuf = replyToBuffer(c->rep);
        int towrite = sdslen(obuf) - c->bufpos;
        nwritten = write(fd, obuf+c->bufpos,towrite);
        /* Content */
        if (nwritten == -1) {
            if (errno == EAGAIN) {
                nwritten = 0;
            } else {
                ulog(CCACHE_VERBOSE,
                     "Error writing to client: %s", strerror(errno));
                freeClient(c);
                return;
            }
        }
        c->lastinteraction = time(NULL);
        listMoveNodeToTail(el->clients,c->elNode);
        if(nwritten<towrite) {
            c->bufpos += nwritten;
        }
        else {
#ifdef AE_MAX_CLIENT_IDLE_TIME
            resetClient(c);
#else
            freeClient(c);
#endif
            aeModifyFileEvent(el,c->fd,AE_READABLE,c);
            printf("Send Reply: %.2lf\n", (double)(clock()));
        }
    }
}

//send to client,if send over , delete writable event
void onClientWritable( aeEventLoop *el, int fd, void *privdata, int mask )
{
    ssize_t nwritten;
	if( servG->connlist[fd].disable == 1 )
	{
		return;
	}

	nwritten = write( fd, servG->connlist[fd].send_buffer, sdslen(servG->connlist[fd].send_buffer));
	if (nwritten <= 0)
	{
		printf( "I/O error writing to client: %s", strerror(errno));
		freeClient( &servG->connlist[fd] );
		return;
	}
	
	//offset
    sdsrange(servG->connlist[fd].send_buffer,nwritten,-1);
	
	///if send_buffer no data need send, remove writable event
    if (sdslen(servG->connlist[fd].send_buffer) == 0)
    {
        aeDeleteFileEvent( el, fd, AE_WRITABLE);
		if( servG->connlist[fd].disable == 2 )
		{
		   freeClient( &servG->connlist[fd] );
		}
    }
}

/* 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 readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
    client *c = (client*) privdata;
    int nread, readlen;
    size_t qblen;
    UNUSED(el);
    UNUSED(mask);

    readlen = PROTO_IOBUF_LEN;
    /* If this is a multi bulk request, and we are processing a bulk reply
     * that is large enough, try to maximize the probability that the query
     * buffer contains exactly the SDS string representing the object, even
     * at the risk of requiring more read(2) calls. This way the function
     * processMultiBulkBuffer() can avoid copying buffers to create the
     * Redis Object representing the argument. */
    if (c->reqtype == PROTO_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1
        && c->bulklen >= PROTO_MBULK_BIG_ARG)
    {
        int remaining = (unsigned)(c->bulklen+2)-sdslen(c->querybuf);

        if (remaining < readlen) readlen = remaining;
    }

    qblen = sdslen(c->querybuf);
    if (c->querybuf_peak < qblen) c->querybuf_peak = qblen;
    c->querybuf = sdsMakeRoomFor(c->querybuf, readlen);
    nread = read(fd, c->querybuf+qblen, readlen);
    if (nread == -1) {
        if (errno == EAGAIN) {
            return;
        } else {
            Log(LL_VERBOSE, "Reading from client: %s",strerror(errno));
            freeClient(c);
            return;
        }
    } else if (nread == 0) {
        Log(LL_VERBOSE, "Client closed connection");
        freeClient(c);
        return;
    }

    sdsIncrLen(c->querybuf,nread);
    c->lastinteraction = server.unixtime;
    server.stat_net_input_bytes += nread;
    if (sdslen(c->querybuf) > server.client_max_querybuf_len) {
        sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty();

        bytes = sdscatrepr(bytes,c->querybuf,64);
        Log(LL_WARNING,"Closing client that reached max query buffer length: %s (qbuf initial bytes: %s)", ci, bytes);
        sdsfree(ci);
        sdsfree(bytes);
        freeClient(c);
        return;
    }
    processInputBuffer(c);
}

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");
    }
}

void readQueryFromClient(aeEventLoop *el, int fd, httpClient *c) {
    char buf[CCACHE_IOBUF_LEN];
    int nread;
    nread = read(fd, buf, CCACHE_IOBUF_LEN);
    if (nread == -1) {
        if (errno == EAGAIN) { /* try again */
            nread = 0;
        } else {
            ulog(CCACHE_VERBOSE, "Reading from client: %s",strerror(errno));
            freeClient(c);
            return;
        }
    } else if (nread == 0) {
        ulog(CCACHE_VERBOSE, "End of client request");
        freeClient(c);
        return;
    }
    if (nread>0) {
        printf("Read Request: %.2lf \n", (double)(clock()));
        c->lastinteraction = time(NULL);
        listMoveNodeToTail(el->clients,c->elNode);
        /* NOTICE: nread or nread-1 */
        switch(requestParse(c->req,buf,buf+nread)){
        case parse_not_completed:
            break;
        case parse_completed:
        {            
            int handle_result = requestHandle(c->req,c->rep,el->cache,c);
            if(handle_result == HANDLER_BLOCK){
                blockClient(el,c);
            }
            else {
                if (_installWriteEvent(el, c) != CCACHE_OK) return;
                printf("Install Write: %.2lf\n", (double)(clock()));
                /* For HANDLE_OK there is nothing to do */
                if(handle_result == HANDLER_ERR) requestHandleError(c->req,c->rep);
            }
                break;
        }
        case parse_error:
            if (_installWriteEvent(el, c) != CCACHE_OK) {
                return;
            }
            requestHandleError(c->req,c->rep);
            break;
        default:
            break;
        };
    }
}

void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
    vuiClient *c = (vuiClient *) privdata;
    int nread, readlen;
    size_t qblen;

    /* If this is a multi bulk request, and we are processing a bulk reply
     * that is large enough, try to maximize the probability that the query
     * buffer contains exactly the SDS string representing the object, even
     * at the risk of requiring more read(2) calls. This way the function
     * processMultiBulkBuffer() can avoid copying buffers to create the
     * Redis Object representing the argument. */
    if (c->prot.waiting)
    {
        readlen = c->prot.waiting;
    }
    else
    {
        readlen = 1024 * 2;
    }

    qblen = sdslen(c->querybuf);
    c->querybuf = sdsMakeRoomFor(c->querybuf, readlen);
    nread = read(fd, c->querybuf+qblen, readlen);

    if (nread == -1) {
        if (errno == EAGAIN) {
            nread = 0;
        } else {
            LogInfo("Reading from client: %s",strerror(errno));
            freeClient(c);
            return;
        }
    } else if (nread == 0) {
        LogInfo("Client closed connection");
        freeClient(c);
        return;
    }
    if (nread) {
        sdsIncrLen(c->querybuf,nread);
    } 
    //TODO
    if (sdslen(c->querybuf) > 1024 * 1024) {
        LogWarn("Closing client that reached max query buffer length");
        sdsclear(c->querybuf);
        freeClient(c);
        return;
    }
    processInputBuffer(c);
}

static void clientDone(client c) {
    if (config.requests_finished == config.requests) {
        freeClient(c);
        aeStop(config.el);
        return;
    }
    if (config.keepalive) {
        resetClient(c);
    } else {
        config.liveclients--;
        createMissingClients(c);
        config.liveclients++;
        freeClient(c);
    }
}

void onCloseByClient( aeEventLoop *el , void *privdata , appnetServer *serv ,
		appnetConnection *conn )
{
	
	if (conn == NULL)
	{
		return;
	}
	
	appnetPipeData data =
			{0};
	data.type = PIPE_EVENT_CLOSE;
	data.connfd = conn->connfd;
	data.len = 0;
	
	int worker_id = servG->connlist[conn->connfd].worker_id;
	setPipeWritable( el , privdata , conn->connfd );
	
	/* append close event message to sendbuffer will send to worker */
	pthread_mutex_lock( &servG->workers[worker_id].w_mutex );
	
	servG->workers[worker_id].send_buffer = sdscatlen(
			servG->workers[worker_id].send_buffer , &data , PIPE_DATA_HEADER_LENG );
	
	pthread_mutex_unlock( &servG->workers[worker_id].w_mutex );
	
	freeClient( conn );
}

void
dispatch_conn_exist(client *c, int tid)
{
    struct connswapunit *su = csui_new();
    char buf[1];
    vr_worker *worker;

    if (su == NULL) {
        freeClient(c);
        /* given that malloc failed this may also fail, but let's try */
        log_error("Failed to allocate memory for connection swap object\n");
        return ;
    }

    su->num = tid;
    su->data = c;
    
    unlinkClientFromEventloop(c);

    cbsul_push(su);

    worker = darray_get(&workers, (uint32_t)c->curidx);

    /* Back to master */
    buf[0] = 'b';
    if (vr_write(worker->socketpairs[1], buf, 1) != 1) {
        log_error("Notice the worker failed.");
    }
}

/********************************************************
 * handleClientsEvent
 * do					for each client in clientList check if his socket is ready for reading (and if so call handleOneClient(Sd)
 ********************************************************/
void handleFilesPortsEvents(struct clientNode** clientsList,fd_set readfds)
{
	int i=0;
	struct clientNode* currentClient;

	//start from the head of the list
	currentClient = *clientsList;
	//for each client in the client list
	while(currentClient!=NULL)
	{
		//for each fileSd in client
		for(i=0; i<currentClient->numOfFiles ; i++)
		{
			// if currentClient->filesSd[i] is ready for reading
			if(FD_ISSET(currentClient->filesSd[i], &readfds))
			{
				handleOneClient(currentClient->filesSd[i]);
				currentClient->countFiles++;

				//if we get all the files from the client free the client
				if(currentClient->countFiles == currentClient->numOfFiles)
				{
					freeClient(currentClient, clientsList);
				}
				return;
			}

		}
		//else go to the next client
		currentClient = currentClient->next;
	}


}

static void acceptCommonHandler(int fd, int flags) {
    client *c;
    if ((c = createClient(fd)) == NULL) {
        Log(WARNING,
            "Error registering fd event for the new client: %s (fd=%d)",
            strerror(errno),fd);
        close(fd); /* May be already closed, just ignore 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 non-blocking
     * mode and we can send an error for free using the Kernel I/O */
    if (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... */
        }
        server.stat_rejected_conn++;
        freeClient(c);
        return;
    }
    server.stat_numconnections++;
    c->flags |= flags;
}

static void clientsCron()
{
    long idletime;
    unsigned long numclients;
    unsigned long iteration;
    struct client *c;
    struct listNode *node;

    numclients = listLength(Clients);
    iteration = numclients < 50 ? numclients : numclients / 10;
    while (listLength(Clients) && iteration--) {
        node = listFirst(Clients);
        c = listNodeValue(node);
        ASSERT(c);

        listRotate(Clients);
        idletime = Server.cron_time.tv_sec - c->last_io.tv_sec;
        if (idletime > Server.worker_timeout) {
            wheatLog(WHEAT_VERBOSE, "Closing idle client %s timeout: %lds",
                    c->name, idletime);
            getStatItemByName("Total timeout client")->val++;
            freeClient(c);
            continue;
        }

        if (!listLength(c->conns))
            msgClean(c->req_buf);
    }
}

void closeTimedoutClients(void) {
    redisClient *c;
    listNode *ln;
    time_t now = time(NULL);
    listIter li;

    listRewind(server.clients,&li);
    while ((ln = listNext(&li)) != NULL) {
        c = listNodeValue(ln);
        if (server.maxidletime &&
            !(c->flags & REDIS_SLAVE) &&    /* no timeout for slaves */
            !(c->flags & REDIS_MASTER) &&   /* no timeout for masters */
            !(c->flags & REDIS_BLOCKED) &&  /* no timeout for BLPOP */
            dictSize(c->pubsub_channels) == 0 && /* no timeout for pubsub */
            listLength(c->pubsub_patterns) == 0 &&
            (now - c->lastinteraction > server.maxidletime))
        {
            redisLog(REDIS_VERBOSE,"Closing idle client");
            freeClient(c);
        } else if (c->flags & REDIS_BLOCKED) {
            if (c->blockingto != 0 && c->blockingto < now) {
                addReply(c,shared.nullmultibulk);
                unblockClientWaitingData(c);
            }
        }
    }
}

int closeTimedoutClients(aeEventLoop *el) {    
    if(el->myid != 0) {
        httpClient *c;
        int deletedNodes = 0;
        time_t now = time(NULL);
        listIter li;
        listNode *ln;

        listRewind(el->clients,&li);
        while ((ln = listNext(&li)) != NULL) {
            c = listNodeValue(ln);
            if (el->maxidletime &&
                    (now - c->lastinteraction > el->maxidletime))
            {                
                /* the client is waiting for reply */
                if (c->blocked) {
                    /* This situation happens when request_handler time exceeds client timeout.
                     * Client timeout is typically 30 seconds and
                     * Request_handler rarely consumes more than 1 second.
                     * This rare case has a very small role in overall performance.                    
                     */
                    listNode *ln = listSearchKey(c->ceList,c);
                    if(ln) listDelNode(c->ceList,ln);
                }
                freeClient(c);
                deletedNodes++;
            }
            else break;
        }
        return deletedNodes;
    }
    return 0;
}

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++;
}

static void handleRequest(struct evcenter *center, int fd, void *data, int mask)
{
    struct client *client;
    struct conn *conn;
    ssize_t nread, ret;
    struct timeval start, end;
    long time_use;
    struct slice slice;
    size_t parsed;

    parsed = 0;
    ret = 0;
    client = data;

    gettimeofday(&start, NULL);
    nread = WorkerProcess->worker->recvData(client);
    if (!isClientValid(client)) {
        freeClient(client);
        return ;
    }

    if (msgGetSize(client->req_buf) > getStatVal(StatBufferSize)) {
        getStatVal(StatBufferSize) = msgGetSize(client->req_buf);
    }

    while (msgCanRead(client->req_buf)) {
        conn = connGet(client);

        msgRead(client->req_buf, &slice);
        ret = client->protocol->parser(conn, &slice, &parsed);

        if (ret == WHEAT_WRONG) {
            wheatLog(WHEAT_NOTICE, "parse data failed");
            msgSetReaded(client->req_buf, 0);
            setClientUnvalid(client);
            break;
        } else if (ret == WHEAT_OK) {
            msgSetReaded(client->req_buf, parsed);
            getStatVal(StatTotalRequest)++;
            client->pending = NULL;
            ret = client->protocol->spotAppAndCall(conn);
            if (ret != WHEAT_OK) {
                getStatVal(StatFailedRequest)++;
                client->should_close = 1;
                wheatLog(WHEAT_NOTICE, "app failed");
                break;
            }
        } else if (ret == 1) {
            client->pending = conn;
            msgSetReaded(client->req_buf, parsed);
            continue;
        }
    }
    tryFreeClient(client);
    gettimeofday(&end, NULL);
    time_use = 1000000 * (end.tv_sec - start.tv_sec) + (end.tv_usec - start.tv_usec);
    getStatVal(StatRunTime) += time_use;
}

static void freeAllClients(void) {
    listNode *ln = config.clients->head, *next;

    while(ln) {
        next = ln->next;
        freeClient(ln->value);
        ln = next;
    }
}

void http_close( httpHeader* reqHeader , int force  )
{
	if ( reqHeader->keep_alive == 0 || servG->http_keep_alive == 0 || force == 1 )
	{
		aeConnection c  = servG->connlist[reqHeader->connfd];
		freeClient( &c );
	}
	free( reqHeader );
}