/* 
	fd : socket file descriptor
	mask : type of event, READABLE or WRITABLE
	proc : handler of file event
	 */
	int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
	        aeFileProc *proc, void *clientData)
	{
	    if (fd >= eventLoop->setsize) {
	        errno = ERANGE;
	        return AE_ERR;
	    }
	    aeFileEvent *fe = &eventLoop->events[fd];
	
		/* add file event, attach fd to event 
		   if type is READABLE, add fd to fd_set rfds, else if type is WRITABLE, add to wfds
		 */
	    if (aeApiAddEvent(eventLoop, fd, mask) == -1)	
	        return AE_ERR;
	    fe->mask |= mask;
	    if (mask & AE_READABLE) fe->rfileProc = proc;	//now fd attach to file event
	    if (mask & AE_WRITABLE) fe->wfileProc = proc;
	    fe->clientData = clientData;
	    if (fd > eventLoop->maxfd)
	        eventLoop->maxfd = fd;
	    return AE_OK;
	}

	static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
	    aeApiState *state = eventLoop->apidata;
	    int retval, j, numevents = 0;
	
	    memcpy(&state->_rfds,&state->rfds,sizeof(fd_set));
	    memcpy(&state->_wfds,&state->wfds,sizeof(fd_set));
	
		/* allow a program to monitor multiple file descriptors, waiting until 
		   one or more of the file descriptors become "ready" */
	    retval = select(eventLoop->maxfd+1,
	                &state->_rfds,&state->_wfds,NULL,tvp);
	    if (retval > 0) {
	        for (j = 0; j <= eventLoop->maxfd; j++) {
	            int mask = 0;
	            aeFileEvent *fe = &eventLoop->events[j];
	
	            if (fe->mask == AE_NONE) continue;
	            if (fe->mask & AE_READABLE && FD_ISSET(j,&state->_rfds))
	                mask |= AE_READABLE;
	            if (fe->mask & AE_WRITABLE && FD_ISSET(j,&state->_wfds))
	                mask |= AE_WRITABLE;
	            eventLoop->fired[numevents].fd = j;
	            eventLoop->fired[numevents].mask = mask;
	            numevents++;
	        }
	    }
	    return numevents;
	}

	/* Process every pending time event, then every pending file event
	 * (that may be registered by time event callbacks just processed).
	 * Without special flags the function sleeps until some file event
	 * fires, or when the next time event occurs (if any).
	 *
	 * If flags is 0, the function does nothing and returns.
	 * if flags has AE_ALL_EVENTS set, all the kind of events are processed.
	 * if flags has AE_FILE_EVENTS set, file events are processed.
	 * if flags has AE_TIME_EVENTS set, time events are processed.
	 * if flags has AE_DONT_WAIT set the function returns ASAP until all
	 * the events that's possible to process without to wait are processed.
	 *
	 * The function returns the number of events processed. */
	 /* the event dispatcher, make a choice to select an event handler */
	int aeProcessEvents(aeEventLoop *eventLoop, int flags)
	{
	    int processed = 0, numevents;
	
	    /* Nothing to do? return ASAP */
	    if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0;
	
	    /* Note that we want call select() even if there are no
	     * file events to process as long as we want to process time
	     * events, in order to sleep until the next time event is ready
	     * to fire. */
	    if (eventLoop->maxfd != -1 ||
	        ((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
	        int j;
	        aeTimeEvent *shortest = NULL;
	        struct timeval tv, *tvp;
	
	        if (flags & AE_TIME_EVENTS && !(flags & AE_DONT_WAIT))
	            shortest = aeSearchNearestTimer(eventLoop);
	        if (shortest) {
	            long now_sec, now_ms;
	
	            /* Calculate the time missing for the nearest
	             * timer to fire. */
	            aeGetTime(&now_sec, &now_ms);
	            tvp = &tv;
	            tvp->tv_sec = shortest->when_sec - now_sec;
	            if (shortest->when_ms < now_ms) {
	                tvp->tv_usec = ((shortest->when_ms+1000) - now_ms)*1000;
	                tvp->tv_sec --;
	            } else {
	                tvp->tv_usec = (shortest->when_ms - now_ms)*1000;
	            }
	            if (tvp->tv_sec < 0) tvp->tv_sec = 0;
	            if (tvp->tv_usec < 0) tvp->tv_usec = 0;
	        } else {
	            /* If we have to check for events but need to return
	             * ASAP because of AE_DONT_WAIT we need to set the timeout
	             * to zero */
	            if (flags & AE_DONT_WAIT) {
	                tv.tv_sec = tv.tv_usec = 0;
	                tvp = &tv;
	            } else {
	                /* Otherwise we can block */
	                tvp = NULL; /* wait forever */
	            }
	        }
	
	        numevents = aeApiPoll(eventLoop, tvp);	//how many events fired
	        for (j = 0; j < numevents; j++) {
	            aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
	            int mask = eventLoop->fired[j].mask;
	            int fd = eventLoop->fired[j].fd;
	            int rfired = 0;
	
		    /* note the fe->mask & mask & ... code: maybe an already processed
	             * event removed an element that fired and we still didn't
	             * processed, so we check if the event is still valid. */
	            if (fe->mask & mask & AE_READABLE) {
	                rfired = 1;
	                fe->rfileProc(eventLoop,fd,fe->clientData,mask);
	            }
	            if (fe->mask & mask & AE_WRITABLE) {
	                if (!rfired || fe->wfileProc != fe->rfileProc)
	                    fe->wfileProc(eventLoop,fd,fe->clientData,mask);
	            }
	            processed++;
	        }
	    }
	    /* Check time events */
	    if (flags & AE_TIME_EVENTS)
	        processed += processTimeEvents(eventLoop);
	
	    return processed; /* return the number of processed file/time events */
	}

	void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
	    int cport, cfd, max = MAX_ACCEPTS_PER_CALL;
	    char cip[REDIS_IP_STR_LEN];
	    REDIS_NOTUSED(el);
	    REDIS_NOTUSED(mask);
	    REDIS_NOTUSED(privdata);
	
	    while(max--) {
	        cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
	        if (cfd == ANET_ERR) {
	            if (errno != EWOULDBLOCK)
	                redisLog(REDIS_WARNING,
	                    "Accepting client connection: %s", server.neterr);
	            return;
	        }
	        redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport);
	        acceptCommonHandler(cfd,0);
	    }
	}

	static void acceptCommonHandler(int fd, int flags) {
	    redisClient *c;
	    if ((c = createClient(fd)) == NULL) {
	        redisLog(REDIS_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;
	}

	redisClient *createClient(int fd) {
	    redisClient *c = zmalloc(sizeof(redisClient));
	
	    /* passing -1 as fd it is possible to create a non connected client.
	     * This is useful since all the Redis commands needs to be executed
	     * in the context of a client. When commands are executed in other
	     * contexts (for instance a Lua script) we need a non connected client. */
	    if (fd != -1) {
	        anetNonBlock(NULL,fd);
	        anetEnableTcpNoDelay(NULL,fd);
	        if (server.tcpkeepalive)
	            anetKeepAlive(NULL,fd,server.tcpkeepalive);
	        if (aeCreateFileEvent(server.el,fd,AE_READABLE,
	            readQueryFromClient, c) == AE_ERR)
	        {
	            close(fd);
	            zfree(c);
	            return NULL;
	        }
		}
		/* 对客户端其他信息的初始化 */
	    ...
	}

	void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
	    redisClient *c = (redisClient*) privdata;
	    int nread, readlen;
	    size_t qblen;
	    REDIS_NOTUSED(el);
	    REDIS_NOTUSED(mask);
	
	    server.current_client = c;
	    readlen = REDIS_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 == REDIS_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1
	        && c->bulklen >= REDIS_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) {
	            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) {
	        sdsIncrLen(c->querybuf,nread);
	        c->lastinteraction = server.unixtime;
	        if (c->flags & REDIS_MASTER) c->reploff += nread;
	        server.stat_net_input_bytes += nread;
	    } else {
	        server.current_client = NULL;
	        return;
	    }
		/* 如果缓冲区超出最大范围，关闭该客户端 */
	    if (sdslen(c->querybuf) > server.client_max_querybuf_len) {	//max query buf is 1GB
	        sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty();
	
	        bytes = sdscatrepr(bytes,c->querybuf,64);
	        redisLog(REDIS_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);	//解析c->querybuf 中的参数，以redisStringObject的方式放入 c->argv 数组中
	    server.current_client = NULL;
	}

	/* This function is called every time we are going to transmit new data
	 * to the client. The behavior is the following:
	 *
	 * If the client should receive new data (normal clients will) the function
	 * returns REDIS_OK, and make sure to install the write handler in our event
	 * loop so that when the socket is writable new data gets written.
	 *
	 * If the client should not receive new data, because it is a fake client
	 * (used to load AOF in memory), a master or because the setup of the write
	 * handler failed, the function returns REDIS_ERR.
	 *
	 * The function may return REDIS_OK without actually installing the write
	 * event handler in the following cases:
	 *
	 * 1) The event handler should already be installed since the output buffer
	 *    already contained something.
	 * 2) The client is a slave but not yet online, so we want to just accumulate
	 *    writes in the buffer but not actually sending them yet.
	 *
	 * Typically gets called every time a reply is built, before adding more
	 * data to the clients output buffers. If the function returns REDIS_ERR no
	 * data should be appended to the output buffers. */
	int prepareClientToWrite(redisClient *c) {
	    /* If it's the Lua client we always return ok without installing any
	     * handler since there is no socket at all. */
	    if (c->flags & REDIS_LUA_CLIENT) return REDIS_OK;
	
	    /* Masters don't receive replies, unless REDIS_MASTER_FORCE_REPLY flag
	     * is set. */
	    if ((c->flags & REDIS_MASTER) &&
	        !(c->flags & REDIS_MASTER_FORCE_REPLY)) return REDIS_ERR;
	
	    if (c->fd <= 0) return REDIS_ERR; /* Fake client for AOF loading. */
	
	    /* Only install the handler if not already installed and, in case of
	     * slaves, if the client can actually receive writes. */
	    if (c->bufpos == 0 && listLength(c->reply) == 0 &&
	        (c->replstate == REDIS_REPL_NONE ||
	         (c->replstate == REDIS_REPL_ONLINE && !c->repl_put_online_on_ack)))
	    {
	        /* Try to install the write handler. */
	        if (aeCreateFileEvent(server.el, c->fd, AE_WRITABLE,
	                sendReplyToClient, c) == AE_ERR)
	        {
	            freeClientAsync(c);
	            return REDIS_ERR;
	        }
	    }
	
	    /* Authorize the caller to queue in the output buffer of this client. */
	    return REDIS_OK;
	}

	void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
	    redisClient *c = privdata;
	    int nwritten = 0, totwritten = 0, objlen;
	    size_t objmem;
	    robj *o;
	    REDIS_NOTUSED(el);
	    REDIS_NOTUSED(mask);
	
	    while(c->bufpos > 0 || listLength(c->reply)) {
	        if (c->bufpos > 0) {
	            nwritten = write(fd,c->buf+c->sentlen,c->bufpos-c->sentlen);
	            if (nwritten <= 0) break;
	            c->sentlen += nwritten;
	            totwritten += nwritten;
	
	            /* If the buffer was sent, set bufpos to zero to continue with
	             * the remainder of the reply. */
	            if (c->sentlen == c->bufpos) {
	                c->bufpos = 0;
	                c->sentlen = 0;
	            }
	        } else {
	            o = listNodeValue(listFirst(c->reply));
	            objlen = sdslen(o->ptr);
	            objmem = getStringObjectSdsUsedMemory(o);
	
	            if (objlen == 0) {
	                listDelNode(c->reply,listFirst(c->reply));
	                c->reply_bytes -= objmem;
	                continue;
	            }
	
	            nwritten = write(fd, ((char*)o->ptr)+c->sentlen,objlen-c->sentlen);
	            if (nwritten <= 0) break;
	            c->sentlen += nwritten;
	            totwritten += nwritten;
	
	            /* If we fully sent the object on head go to the next one */
	            if (c->sentlen == objlen) {
	                listDelNode(c->reply,listFirst(c->reply));
	                c->sentlen = 0;
	                c->reply_bytes -= objmem;
	            }
	        }
	        /* Note that we avoid to send more than REDIS_MAX_WRITE_PER_EVENT
	         * bytes, in a single threaded server it's a good idea to serve
	         * other clients as well, even if a very large request comes from
	         * super fast link that is always able to accept data (in real world
	         * scenario think about 'KEYS *' against the loopback interface).
	         *
	         * However if we are over the maxmemory limit we ignore that and
	         * just deliver as much data as it is possible to deliver. */
	        server.stat_net_output_bytes += totwritten;
	        if (totwritten > REDIS_MAX_WRITE_PER_EVENT &&
	            (server.maxmemory == 0 ||
	             zmalloc_used_memory() < server.maxmemory)) break;
	    }
	    if (nwritten == -1) {
	        if (errno == EAGAIN) {
	            nwritten = 0;
	        } else {
	            redisLog(REDIS_VERBOSE,
	                "Error writing to client: %s", strerror(errno));
	            freeClient(c);
	            return;
	        }
	    }
	    if (totwritten > 0) {
	        /* For clients representing masters we don't count sending data
	         * as an interaction, since we always send REPLCONF ACK commands
	         * that take some time to just fill the socket output buffer.
	         * We just rely on data / pings received for timeout detection. */
	        if (!(c->flags & REDIS_MASTER)) c->lastinteraction = server.unixtime;
	    }
	    if (c->bufpos == 0 && listLength(c->reply) == 0) {
	        c->sentlen = 0;
	        aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
	
	        /* Close connection after entire reply has been sent. */
	        if (c->flags & REDIS_CLOSE_AFTER_REPLY) freeClient(c);
	    }
	}

	/* Process time events */
	static int processTimeEvents(aeEventLoop *eventLoop) {
	    int processed = 0;
	    aeTimeEvent *te;
	    long long maxId;
	    time_t now = time(NULL);
	
	    /* If the system clock is moved to the future, and then set back to the
	     * right value, time events may be delayed in a random way. Often this
	     * means that scheduled operations will not be performed soon enough.
	     *
	     * Here we try to detect system clock skews, and force all the time
	     * events to be processed ASAP when this happens: the idea is that
	     * processing events earlier is less dangerous than delaying them
	     * indefinitely, and practice suggests it is. */
	    if (now < eventLoop->lastTime) {
	        te = eventLoop->timeEventHead;
	        while(te) {
	            te->when_sec = 0;
	            te = te->next;
	        }
	    }
	    eventLoop->lastTime = now;
	
	    te = eventLoop->timeEventHead;
	    maxId = eventLoop->timeEventNextId-1;
	    while(te) {
	        long now_sec, now_ms;
	        long long id;
	
	        if (te->id > maxId) {
	            te = te->next;
	            continue;
	        }
	        aeGetTime(&now_sec, &now_ms);
			/* 判断时间事件是否到达 */
	        if (now_sec > te->when_sec ||
	            (now_sec == te->when_sec && now_ms >= te->when_ms))
	        {
	            int retval;
	
				/* 调用时间事件处理函数 */
	            id = te->id;
	            retval = te->timeProc(eventLoop, id, te->clientData);
	            processed++;
	            /* After an event is processed our time event list may
	             * no longer be the same, so we restart from head.
	             * Still we make sure to don't process events registered
	             * by event handlers itself in order to don't loop forever.
	             * To do so we saved the max ID we want to handle.
	             *
	             * FUTURE OPTIMIZATIONS:
	             * Note that this is NOT great algorithmically. Redis uses
	             * a single time event so it's not a problem but the right
	             * way to do this is to add the new elements on head, and
	             * to flag deleted elements in a special way for later
	             * deletion (putting references to the nodes to delete into
	             * another linked list). */
	            if (retval != AE_NOMORE) {
	                aeAddMillisecondsToNow(retval,&te->when_sec,&te->when_ms);
	            } else {
	                aeDeleteTimeEvent(eventLoop, id);
	            }
	            te = eventLoop->timeEventHead;
	        } else {
	            te = te->next;
	        }
	    }
	    return processed;
	}

	/* With multiplexing we need to take per-client state.
 	 * Clients are taken in a linked list. */
	typedef struct redisClient {
	    uint64_t id;            /* Client incremental unique ID. */
	    int fd;					/* socket file descriptor */
	    redisDb *db;
	    int dictid;
	    robj *name;             /* As set by CLIENT SETNAME */
	    sds querybuf;
	    size_t querybuf_peak;   /* Recent (100ms or more) peak of querybuf size */
	    int argc;
	    robj **argv;
	    struct redisCommand *cmd, *lastcmd;
	    int reqtype;
	    int multibulklen;       /* number of multi bulk arguments left to read */
	    long bulklen;           /* length of bulk argument in multi bulk request */
	    list *reply;
	    unsigned long reply_bytes; /* Tot bytes of objects in reply list */
	    int sentlen;            /* Amount of bytes already sent in the current
	                               buffer or object being sent. */
	    time_t ctime;           /* Client creation time */
	    time_t lastinteraction; /* time of the last interaction, used for timeout */
	    time_t obuf_soft_limit_reached_time;
	    int flags;              /* REDIS_SLAVE | REDIS_MONITOR | REDIS_MULTI ... */
	    int authenticated;      /* when requirepass is non-NULL */
	    int replstate;          /* replication state if this is a slave */
	    int repl_put_online_on_ack; /* Install slave write handler on ACK. */
	    int repldbfd;           /* replication DB file descriptor */
	    off_t repldboff;        /* replication DB file offset */
	    off_t repldbsize;       /* replication DB file size */
	    sds replpreamble;       /* replication DB preamble. */
	    long long reploff;      /* replication offset if this is our master */
	    long long repl_ack_off; /* replication ack offset, if this is a slave */
	    long long repl_ack_time;/* replication ack time, if this is a slave */
	    long long psync_initial_offset; /* FULLRESYNC reply offset other slaves
	                                       copying this slave output buffer
	                                       should use. */
	    char replrunid[REDIS_RUN_ID_SIZE+1]; /* master run id if this is a master */
	    int slave_listening_port; /* As configured with: SLAVECONF listening-port */
	    int slave_capa;         /* Slave capabilities: SLAVE_CAPA_* bitwise OR. */
	    multiState mstate;      /* MULTI/EXEC state */
	    int btype;              /* Type of blocking op if REDIS_BLOCKED. */
	    blockingState bpop;     /* blocking state */
	    long long woff;         /* Last write global replication offset. */
	    list *watched_keys;     /* Keys WATCHED for MULTI/EXEC CAS */
	    dict *pubsub_channels;  /* channels a client is interested in (SUBSCRIBE) */
	    list *pubsub_patterns;  /* patterns a client is interested in (SUBSCRIBE) */
	    sds peerid;             /* Cached peer ID. */
	
	    /* Response buffer */
	    int bufpos;
	    char buf[REDIS_REPLY_CHUNK_BYTES];	/* 16K output buffer, soft limit */
	} redisClient;

	/* Load the server configuration from the specified filename.
	 * The function appends the additional configuration directives stored
	 * in the 'options' string to the config file before loading.
	 *
	 * Both filename and options can be NULL, in such a case are considered
	 * empty. This way loadServerConfig can be used to just load a file or
	 * just load a string. */
	void loadServerConfig(char *filename, char *options) {
	    sds config = sdsempty();
	    char buf[REDIS_CONFIGLINE_MAX+1];
	
	    /* Load the file content */
	    if (filename) {
	        FILE *fp;
	
	        if (filename[0] == '-' && filename[1] == '\0') {
	            fp = stdin;
	        } else {
	            if ((fp = fopen(filename,"r")) == NULL) {
	                redisLog(REDIS_WARNING,
	                    "Fatal error, can't open config file '%s'", filename);
	                exit(1);
	            }
	        }
	        while(fgets(buf,REDIS_CONFIGLINE_MAX+1,fp) != NULL)
	            config = sdscat(config,buf);
	        if (fp != stdin) fclose(fp);
	    }
	    /* Append the additional options */
	    if (options) {
	        config = sdscat(config,"\n");
	        config = sdscat(config,options);
	    }
	    loadServerConfigFromString(config);
	    sdsfree(config);
	}

	void loadServerConfigFromString(char *config) {
	    char *err = NULL;
	    int linenum = 0, totlines, i;
	    int slaveof_linenum = 0;
	    sds *lines;
	
	    lines = sdssplitlen(config,strlen(config),"\n",1,&totlines);
	
	    for (i = 0; i < totlines; i++) {
	        sds *argv;
	        int argc;
	
	        linenum = i+1;
	        lines[i] = sdstrim(lines[i]," \t\r\n");
	
	        /* Skip comments and blank lines */
	        if (lines[i][0] == '#' || lines[i][0] == '\0') continue;
	
	        /* Split into arguments */
	        argv = sdssplitargs(lines[i],&argc);
	        if (argv == NULL) {
	            err = "Unbalanced quotes in configuration line";
	            goto loaderr;
	        }
	
	        /* Skip this line if the resulting command vector is empty. */
	        if (argc == 0) {
	            sdsfreesplitres(argv,argc);
	            continue;
	        }
	        sdstolower(argv[0]);
	
	        /* Execute config directives */
	        if (!strcasecmp(argv[0],"timeout") && argc == 2) {
	            server.maxidletime = atoi(argv[1]);
	            if (server.maxidletime < 0) {
	                err = "Invalid timeout value"; goto loaderr;
	            }
	        } else if (!strcasecmp(argv[0],"tcp-keepalive") && argc == 2) {
	            server.tcpkeepalive = atoi(argv[1]);
	            if (server.tcpkeepalive < 0) {
	                err = "Invalid tcp-keepalive value"; goto loaderr;
	            }
			...
	}

    /* Open the AOF file if needed. */
    if (server.aof_state == REDIS_AOF_ON) {
        server.aof_fd = open(server.aof_filename,
                               O_WRONLY|O_APPEND|O_CREAT,0644);
        if (server.aof_fd == -1) {
            redisLog(REDIS_WARNING, "Can't open the append-only file: %s",
                strerror(errno));
            exit(1);
        }
    }

	void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
		...
		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) {
	        sdsIncrLen(c->querybuf,nread);
	        c->lastinteraction = server.unixtime;
	        if (c->flags & REDIS_MASTER) c->reploff += nread;
	        server.stat_net_input_bytes += nread;
	    }
		...
	}

	/* Command table -- we initiialize it here as it is part of the
	     * initial configuration, since command names may be changed via
	     * redis.conf using the rename-command directive. */
	    server.commands = dictCreate(&commandTableDictType,NULL);
	    server.orig_commands = dictCreate(&commandTableDictType,NULL);
	    populateCommandTable();
	    server.delCommand = lookupCommandByCString("del");
	    server.multiCommand = lookupCommandByCString("multi");
	    server.lpushCommand = lookupCommandByCString("lpush");
	    server.lpopCommand = lookupCommandByCString("lpop");
	    server.rpopCommand = lookupCommandByCString("rpop");

	/* Populates the Redis Command Table starting from the hard coded list
	 * we have on top of redis.c file. */
	void populateCommandTable(void) {
	    int j;
	    int numcommands = sizeof(redisCommandTable)/sizeof(struct redisCommand);
	
	    for (j = 0; j < numcommands; j++) {
	        struct redisCommand *c = redisCommandTable+j;
	        char *f = c->sflags;
	        int retval1, retval2;
	
	        while(*f != '\0') {
	            switch(*f) {
	            case 'w': c->flags |= REDIS_CMD_WRITE; break;
	            case 'r': c->flags |= REDIS_CMD_READONLY; break;
	            case 'm': c->flags |= REDIS_CMD_DENYOOM; break;
	            case 'a': c->flags |= REDIS_CMD_ADMIN; break;
	            case 'p': c->flags |= REDIS_CMD_PUBSUB; break;
	            case 's': c->flags |= REDIS_CMD_NOSCRIPT; break;
	            case 'R': c->flags |= REDIS_CMD_RANDOM; break;
	            case 'S': c->flags |= REDIS_CMD_SORT_FOR_SCRIPT; break;
	            case 'l': c->flags |= REDIS_CMD_LOADING; break;
	            case 't': c->flags |= REDIS_CMD_STALE; break;
	            case 'M': c->flags |= REDIS_CMD_SKIP_MONITOR; break;
	            case 'k': c->flags |= REDIS_CMD_ASKING; break;
	            case 'F': c->flags |= REDIS_CMD_FAST; break;
	            default: redisPanic("Unsupported command flag"); break;
	            }
	            f++;
	        }
	
	        retval1 = dictAdd(server.commands, sdsnew(c->name), c);
	        /* Populate an additional dictionary that will be unaffected
	         * by rename-command statements in redis.conf. */
	        retval2 = dictAdd(server.orig_commands, sdsnew(c->name), c);
	        redisAssert(retval1 == DICT_OK && retval2 == DICT_OK);
	    }
	}

	void call(redisClient *c, int flags) {
		...
		c->cmd->proc(c);
		...
		/* Log the command into the Slow log if needed, and populate the
	     * per-command statistics that we show in INFO commandstats. */
	    if (flags & REDIS_CALL_SLOWLOG && c->cmd->proc != execCommand) {
	        char *latency_event = (c->cmd->flags & REDIS_CMD_FAST) ?
	                              "fast-command" : "command";
	        latencyAddSampleIfNeeded(latency_event,duration/1000);
	        slowlogPushEntryIfNeeded(c->argv,c->argc,duration);
	    }
	    if (flags & REDIS_CALL_STATS) {
	        c->cmd->microseconds += duration;
	        c->cmd->calls++;
	    }
	
	    /* Propagate the command into the AOF and replication link */
	    if (flags & REDIS_CALL_PROPAGATE) {
	        int flags = REDIS_PROPAGATE_NONE;
	
	        if (c->flags & REDIS_FORCE_REPL) flags |= REDIS_PROPAGATE_REPL;
	        if (c->flags & REDIS_FORCE_AOF) flags |= REDIS_PROPAGATE_AOF;
	        if (dirty)
	            flags |= (REDIS_PROPAGATE_REPL | REDIS_PROPAGATE_AOF);
	        if (flags != REDIS_PROPAGATE_NONE)
	            propagate(c->cmd,c->db->id,c->argv,c->argc,flags);
	    }
		...
	}