《Redis读书笔记》服务器客户端设计

灰太狼 2023-07-24 05:48 20阅读 0赞

### 文章目录 ###

*   *  服务器
     *   *  服务器初始化
         *   *  初始化服务器状态结构
             *  载入配置项
             *  初始化服务器
             *  事后检查
             *  加载持久化数据
             *  执行事件循环
         *  服务器处理客户端请求
         *  定时处理
         *  数据结构
     *  客户端
     *   *  数据结构

## 服务器 ##

Redis 服务器是典型的一对多服务器程序：一个服务器可以与多个客户端建立网络连接，每个客户端可以向服务器发送命令请求,而服务器接收并处理客户端发送的命令请求，并向客户端命令回复。

*  那么问题来了，每个服务端可以维护多少个客户端连接？
 *  服务器是如何维护客户端连接的？
 *  每个客户端可以发送多大命令？
 *  客户端可以存放的value最大有多大？

### 服务器初始化 ###

redis.c/main函数

#### 初始化服务器状态结构 ####

#### 载入配置项 ####

void initServerConfig(void) {
        int j;
    
        updateCachedTime(1);
        getRandomHexChars(server.runid,CONFIG_RUN_ID_SIZE);
        server.runid[CONFIG_RUN_ID_SIZE] = '\0';
        changeReplicationId();
        clearReplicationId2();
        server.timezone = getTimeZone(); /* Initialized by tzset(). */
        server.configfile = NULL;
        server.executable = NULL;
        server.arch_bits = (sizeof(long) == 8) ? 64 : 32;
        server.bindaddr_count = 0;
        server.unixsocketperm = CONFIG_DEFAULT_UNIX_SOCKET_PERM;
        server.ipfd_count = 0;
        server.tlsfd_count = 0;
        server.sofd = -1;
        server.active_expire_enabled = 1;
        server.client_max_querybuf_len = PROTO_MAX_QUERYBUF_LEN;
        server.saveparams = NULL;
        server.loading = 0;
        server.logfile = zstrdup(CONFIG_DEFAULT_LOGFILE);
        server.aof_state = AOF_OFF;
        server.aof_rewrite_base_size = 0;
        server.aof_rewrite_scheduled = 0;
        server.aof_flush_sleep = 0;
        server.aof_last_fsync = time(NULL);
        server.aof_rewrite_time_last = -1;
        server.aof_rewrite_time_start = -1;
        server.aof_lastbgrewrite_status = C_OK;
        server.aof_delayed_fsync = 0;
        server.aof_fd = -1;
        server.aof_selected_db = -1; /* Make sure the first time will not match */
        server.aof_flush_postponed_start = 0;
    
        // 省略...
    }

#### 初始化服务器 ####

1.  为服务器设置进程信号处理器

signal(SIGHUP, SIG_IGN);
        signal(SIGPIPE, SIG_IGN);
        setupSignalHandlers();

1.  创建 clients/slaves/等

/* Initialization after setting defaults from the config system. */
        server.aof_state = server.aof_enabled ? AOF_ON : AOF_OFF;
        server.hz = server.config_hz;
        server.pid = getpid();
        server.current_client = NULL;
        server.fixed_time_expire = 0;
        server.clients = listCreate();
        server.clients_index = raxNew();
        server.clients_to_close = listCreate();
        server.slaves = listCreate();
        server.monitors = listCreate();
        server.clients_pending_write = listCreate();
        server.clients_pending_read = listCreate();
        server.slaveseldb = -1; /* Force to emit the first SELECT command. */
        server.unblocked_clients = listCreate();
        server.ready_keys = listCreate();
        server.clients_waiting_acks = listCreate();
        server.get_ack_from_slaves = 0;
        server.clients_paused = 0;
        server.system_memory_size = zmalloc_get_memory_size();

1.  创建共享对象

createSharedObjects();

1.  打开服务器的监听端口

/* Open the TCP listening socket for the user commands. */
        if (server.port != 0 &&
            listenToPort(server.port,server.ipfd,&server.ipfd_count) == C_ERR)
            exit(1);
        if (server.tls_port != 0 &&
            listenToPort(server.tls_port,server.tlsfd,&server.tlsfd_count) == C_ERR)
            exit(1);
    /* Open the listening Unix domain socket. */
        if (server.unixsocket != NULL) {
            unlink(server.unixsocket); /* don't care if this fails */
            server.sofd = anetUnixServer(server.neterr,server.unixsocket,
                server.unixsocketperm, server.tcp_backlog);
            if (server.sofd == ANET_ERR) {
                serverLog(LL_WARNING, "Opening Unix socket: %s", server.neterr);
                exit(1);
            }
            anetNonBlock(NULL,server.sofd);
        }
    
        /* Abort if there are no listening sockets at all. */
        if (server.ipfd_count == 0 && server.tlsfd_count == 0 && server.sofd < 0) {
            serverLog(LL_WARNING, "Configured to not listen anywhere, exiting.");
            exit(1);
        }

1.  创建数据库

/* Create the Redis databases, and initialize other internal state. */
        for (j = 0; j < server.dbnum; j++) { 
            server.db[j].dict = dictCreate(&dbDictType,NULL);
            server.db[j].expires = dictCreate(&keyptrDictType,NULL);
            server.db[j].expires_cursor = 0;
            server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL);
            server.db[j].ready_keys = dictCreate(&objectKeyPointerValueDictType,NULL);
            server.db[j].watched_keys = dictCreate(&keylistDictType,NULL);
            server.db[j].id = j;
            server.db[j].avg_ttl = 0;
            server.db[j].defrag_later = listCreate();
            listSetFreeMethod(server.db[j].defrag_later,(void (*)(void*))sdsfree);
        }
        evictionPoolAlloc(); /* Initialize the LRU keys pool. */

1.  为serverCron函数创建时间事件

/* Create the timer callback, this is our way to process many background * operations incrementally, like clients timeout, eviction of unaccessed * expired keys and so forth. */
        if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) { 
            serverPanic("Can't create event loop timers.");
            exit(1);
        }

1.  创建一个事件处理程序以接受TCP和Unix域套接字中的新连接。

/* Create an event handler for accepting new connections in TCP and Unix * domain sockets. */
        for (j = 0; j < server.ipfd_count; j++) { 
            if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,
                acceptTcpHandler,NULL) == AE_ERR)
                { 
                    serverPanic(
                        "Unrecoverable error creating server.ipfd file event.");
                }
        }
        for (j = 0; j < server.tlsfd_count; j++) { 
            if (aeCreateFileEvent(server.el, server.tlsfd[j], AE_READABLE,
                acceptTLSHandler,NULL) == AE_ERR)
                { 
                    serverPanic(
                        "Unrecoverable error creating server.tlsfd file event.");
                }
        }
        if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,
            acceptUnixHandler,NULL) == AE_ERR) serverPanic("Unrecoverable error creating server.sofd file event.");

1.  如果AOF持久化功能已经打开， 那么打开AOF文件，如果文件不存在，那么就创建一个新的AOF文件，为AOF写入做好准备。

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

1.  创建保存发布/订阅功能的结构

server.pubsub_channels = dictCreate(&keylistDictType,NULL);
        server.pubsub_patterns = listCreate();
        listSetFreeMethod(server.pubsub_patterns,freePubsubPattern);
        listSetMatchMethod(server.pubsub_patterns,listMatchPubsubPattern);

10 .others

replicationScriptCacheInit();
        scriptingInit(1);
        slowlogInit();
        latencyMonitorInit();

#### 事后检查 ####

redisSetProcTitle(argv[0]);
        redisAsciiArt();
        checkTcpBacklogSettings();

#### 加载持久化数据 ####

如果不是新的redis服务器，需要加载持久化的数据

loadDataFromDisk();
    /* Function called at startup to load RDB or AOF file in memory. */
    void loadDataFromDisk(void) {
        long long start = ustime();
        if (server.aof_state == AOF_ON) {
            if (loadAppendOnlyFile(server.aof_filename) == C_OK)
                serverLog(LL_NOTICE,"DB loaded from append only file: %.3f seconds",(float)(ustime()-start)/1000000);
        } else {
            rdbSaveInfo rsi = RDB_SAVE_INFO_INIT;
            if (rdbLoad(server.rdb_filename,&rsi,RDBFLAGS_NONE) == C_OK) {
                serverLog(LL_NOTICE,"DB loaded from disk: %.3f seconds",
                    (float)(ustime()-start)/1000000);
    
                /* Restore the replication ID / offset from the RDB file. */
                if ((server.masterhost ||
                    (server.cluster_enabled &&
                    nodeIsSlave(server.cluster->myself))) &&
                    rsi.repl_id_is_set &&
                    rsi.repl_offset != -1 &&
                    /* Note that older implementations may save a repl_stream_db
                     * of -1 inside the RDB file in a wrong way, see more
                     * information in function rdbPopulateSaveInfo. */
                    rsi.repl_stream_db != -1)
                {
                    memcpy(server.replid,rsi.repl_id,sizeof(server.replid));
                    server.master_repl_offset = rsi.repl_offset;
                    /* If we are a slave, create a cached master from this
                     * information, in order to allow partial resynchronizations
                     * with masters. */
                    replicationCacheMasterUsingMyself();
                    selectDb(server.cached_master,rsi.repl_stream_db);
                }
            } else if (errno != ENOENT) {
                serverLog(LL_WARNING,"Fatal error loading the DB: %s. Exiting.",strerror(errno));
                exit(1);
            }
        }
    }

优先使用AOF

#### 执行事件循环 ####

void aeMain(aeEventLoop *eventLoop) { 
        eventLoop->stop = 0;
        while (!eventLoop->stop) { 
            if (eventLoop->beforesleep != NULL)
                eventLoop->beforesleep(eventLoop);
            aeProcessEvents(eventLoop, AE_ALL_EVENTS|AE_CALL_AFTER_SLEEP);
        }
    }

### 服务器处理客户端请求 ###

参考《Redis读书笔记》事件设计

### 定时处理 ###

参考《Redis读书笔记》事件设计

### 数据结构 ###

1.  RedisServer

struct redisServer {
        /* General */
        pid_t pid;                  /* Main process pid. */
        char *configfile;           /* Absolute config file path, or NULL */
        char *executable;           /* Absolute executable file path. */
        char **exec_argv;           /* Executable argv vector (copy). */
        int dynamic_hz;             /* Change hz value depending on # of clients. */
        int config_hz;              /* Configured HZ value. May be different than
                                       the actual 'hz' field value if dynamic-hz
                                       is enabled. */
        int hz;                     /* serverCron() calls frequency in hertz */
        redisDb *db;
        dict *commands;             /* Command table */
        dict *orig_commands;        /* Command table before command renaming. */
        aeEventLoop *el;
        _Atomic unsigned int lruclock; /* Clock for LRU eviction */
        int shutdown_asap;          /* SHUTDOWN needed ASAP */
        int activerehashing;        /* Incremental rehash in serverCron() */
        int active_defrag_running;  /* Active defragmentation running (holds current scan aggressiveness) */
        char *pidfile;              /* PID file path */
        int arch_bits;              /* 32 or 64 depending on sizeof(long) */
        int cronloops;              /* Number of times the cron function run */
        char runid[CONFIG_RUN_ID_SIZE+1];  /* ID always different at every exec. */
        int sentinel_mode;          /* True if this instance is a Sentinel. */
        size_t initial_memory_usage; /* Bytes used after initialization. */
        int always_show_logo;       /* Show logo even for non-stdout logging. */
        /* Modules */
        dict *moduleapi;            /* Exported core APIs dictionary for modules. */
        dict *sharedapi;            /* Like moduleapi but containing the APIs that
                                       modules share with each other. */
        list *loadmodule_queue;     /* List of modules to load at startup. */
        int module_blocked_pipe[2]; /* Pipe used to awake the event loop if a
                                       client blocked on a module command needs
                                       to be processed. */
        pid_t module_child_pid;     /* PID of module child */
        /* Networking */
        int port;                   /* TCP listening port */
        int tls_port;               /* TLS listening port */
        int tcp_backlog;            /* TCP listen() backlog */
        char *bindaddr[CONFIG_BINDADDR_MAX]; /* Addresses we should bind to */
        int bindaddr_count;         /* Number of addresses in server.bindaddr[] */
        char *unixsocket;           /* UNIX socket path */
        mode_t unixsocketperm;      /* UNIX socket permission */
        int ipfd[CONFIG_BINDADDR_MAX]; /* TCP socket file descriptors */
        int ipfd_count;             /* Used slots in ipfd[] */
        int tlsfd[CONFIG_BINDADDR_MAX]; /* TLS socket file descriptors */
        int tlsfd_count;            /* Used slots in tlsfd[] */
        int sofd;                   /* Unix socket file descriptor */
        int cfd[CONFIG_BINDADDR_MAX];/* Cluster bus listening socket */
        int cfd_count;              /* Used slots in cfd[] */
        list *clients;              /* List of active clients */
        list *clients_to_close;     /* Clients to close asynchronously */
        list *clients_pending_write; /* There is to write or install handler. */
        list *clients_pending_read;  /* Client has pending read socket buffers. */
        list *slaves, *monitors;    /* List of slaves and MONITORs */
        client *current_client;     /* Current client executing the command. */
        long fixed_time_expire;     /* If > 0, expire keys against server.mstime. */
        rax *clients_index;         /* Active clients dictionary by client ID. */
        int clients_paused;         /* True if clients are currently paused */
        mstime_t clients_pause_end_time; /* Time when we undo clients_paused */
        char neterr[ANET_ERR_LEN];   /* Error buffer for anet.c */
        dict *migrate_cached_sockets;/* MIGRATE cached sockets */
        _Atomic uint64_t next_client_id; /* Next client unique ID. Incremental. */
        int protected_mode;         /* Don't accept external connections. */
        int gopher_enabled;         /* If true the server will reply to gopher
                                       queries. Will still serve RESP2 queries. */
        int io_threads_num;         /* Number of IO threads to use. */
        int io_threads_do_reads;    /* Read and parse from IO threads? */
    
        /* RDB / AOF loading information */
        int loading;                /* We are loading data from disk if true */
        off_t loading_total_bytes;
        off_t loading_loaded_bytes;
        time_t loading_start_time;
        off_t loading_process_events_interval_bytes;
        /* Fast pointers to often looked up command */
        struct redisCommand *delCommand, *multiCommand, *lpushCommand,
                            *lpopCommand, *rpopCommand, *zpopminCommand,
                            *zpopmaxCommand, *sremCommand, *execCommand,
                            *expireCommand, *pexpireCommand, *xclaimCommand,
                            *xgroupCommand;
        /* Fields used only for stats */
        time_t stat_starttime;          /* Server start time */
        long long stat_numcommands;     /* Number of processed commands */
        long long stat_numconnections;  /* Number of connections received */
        long long stat_expiredkeys;     /* Number of expired keys */
        double stat_expired_stale_perc; /* Percentage of keys probably expired */
        long long stat_expired_time_cap_reached_count; /* Early expire cylce stops.*/
        long long stat_expire_cycle_time_used; /* Cumulative microseconds used. */
        long long stat_evictedkeys;     /* Number of evicted keys (maxmemory) */
        long long stat_keyspace_hits;   /* Number of successful lookups of keys */
        long long stat_keyspace_misses; /* Number of failed lookups of keys */
        long long stat_active_defrag_hits;      /* number of allocations moved */
        long long stat_active_defrag_misses;    /* number of allocations scanned but not moved */
        long long stat_active_defrag_key_hits;  /* number of keys with moved allocations */
        long long stat_active_defrag_key_misses;/* number of keys scanned and not moved */
        long long stat_active_defrag_scanned;   /* number of dictEntries scanned */
        size_t stat_peak_memory;        /* Max used memory record */
        long long stat_fork_time;       /* Time needed to perform latest fork() */
        double stat_fork_rate;          /* Fork rate in GB/sec. */
        long long stat_rejected_conn;   /* Clients rejected because of maxclients */
        long long stat_sync_full;       /* Number of full resyncs with slaves. */
        long long stat_sync_partial_ok; /* Number of accepted PSYNC requests. */
        long long stat_sync_partial_err;/* Number of unaccepted PSYNC requests. */
        list *slowlog;                  /* SLOWLOG list of commands */
        long long slowlog_entry_id;     /* SLOWLOG current entry ID */
        long long slowlog_log_slower_than; /* SLOWLOG time limit (to get logged) */
        unsigned long slowlog_max_len;     /* SLOWLOG max number of items logged */
        struct malloc_stats cron_malloc_stats; /* sampled in serverCron(). */
        _Atomic long long stat_net_input_bytes; /* Bytes read from network. */
        _Atomic long long stat_net_output_bytes; /* Bytes written to network. */
        size_t stat_rdb_cow_bytes;      /* Copy on write bytes during RDB saving. */
        size_t stat_aof_cow_bytes;      /* Copy on write bytes during AOF rewrite. */
        size_t stat_module_cow_bytes;   /* Copy on write bytes during module fork. */
        /* The following two are used to track instantaneous metrics, like
         * number of operations per second, network traffic. */
        struct {
            long long last_sample_time; /* Timestamp of last sample in ms */
            long long last_sample_count;/* Count in last sample */
            long long samples[STATS_METRIC_SAMPLES];
            int idx;
        } inst_metric[STATS_METRIC_COUNT];
        /* Configuration */
        int verbosity;                  /* Loglevel in redis.conf */
        int maxidletime;                /* Client timeout in seconds */
        int tcpkeepalive;               /* Set SO_KEEPALIVE if non-zero. */
        int active_expire_enabled;      /* Can be disabled for testing purposes. */
        int active_expire_effort;       /* From 1 (default) to 10, active effort. */
        int active_defrag_enabled;
        int jemalloc_bg_thread;         /* Enable jemalloc background thread */
        size_t active_defrag_ignore_bytes; /* minimum amount of fragmentation waste to start active defrag */
        int active_defrag_threshold_lower; /* minimum percentage of fragmentation to start active defrag */
        int active_defrag_threshold_upper; /* maximum percentage of fragmentation at which we use maximum effort */
        int active_defrag_cycle_min;       /* minimal effort for defrag in CPU percentage */
        int active_defrag_cycle_max;       /* maximal effort for defrag in CPU percentage */
        unsigned long active_defrag_max_scan_fields; /* maximum number of fields of set/hash/zset/list to process from within the main dict scan */
        _Atomic size_t client_max_querybuf_len; /* Limit for client query buffer length */
        int dbnum;                      /* Total number of configured DBs */
        int supervised;                 /* 1 if supervised, 0 otherwise. */
        int supervised_mode;            /* See SUPERVISED_* */
        int daemonize;                  /* True if running as a daemon */
        clientBufferLimitsConfig client_obuf_limits[CLIENT_TYPE_OBUF_COUNT];
        /* AOF persistence */
        int aof_enabled;                /* AOF configuration */
        int aof_state;                  /* AOF_(ON|OFF|WAIT_REWRITE) */
        int aof_fsync;                  /* Kind of fsync() policy */
        char *aof_filename;             /* Name of the AOF file */
        int aof_no_fsync_on_rewrite;    /* Don't fsync if a rewrite is in prog. */
        int aof_rewrite_perc;           /* Rewrite AOF if % growth is > M and... */
        off_t aof_rewrite_min_size;     /* the AOF file is at least N bytes. */
        off_t aof_rewrite_base_size;    /* AOF size on latest startup or rewrite. */
        off_t aof_current_size;         /* AOF current size. */
        off_t aof_fsync_offset;         /* AOF offset which is already synced to disk. */
        int aof_flush_sleep;            /* Micros to sleep before flush. (used by tests) */
        int aof_rewrite_scheduled;      /* Rewrite once BGSAVE terminates. */
        pid_t aof_child_pid;            /* PID if rewriting process */
        list *aof_rewrite_buf_blocks;   /* Hold changes during an AOF rewrite. */
        sds aof_buf;      /* AOF buffer, written before entering the event loop */
        int aof_fd;       /* File descriptor of currently selected AOF file */
        int aof_selected_db; /* Currently selected DB in AOF */
        time_t aof_flush_postponed_start; /* UNIX time of postponed AOF flush */
        time_t aof_last_fsync;            /* UNIX time of last fsync() */
        time_t aof_rewrite_time_last;   /* Time used by last AOF rewrite run. */
        time_t aof_rewrite_time_start;  /* Current AOF rewrite start time. */
        int aof_lastbgrewrite_status;   /* C_OK or C_ERR */
        unsigned long aof_delayed_fsync;  /* delayed AOF fsync() counter */
        int aof_rewrite_incremental_fsync;/* fsync incrementally while aof rewriting? */
        int rdb_save_incremental_fsync;   /* fsync incrementally while rdb saving? */
        int aof_last_write_status;      /* C_OK or C_ERR */
        int aof_last_write_errno;       /* Valid if aof_last_write_status is ERR */
        int aof_load_truncated;         /* Don't stop on unexpected AOF EOF. */
        int aof_use_rdb_preamble;       /* Use RDB preamble on AOF rewrites. */
        /* AOF pipes used to communicate between parent and child during rewrite. */
        int aof_pipe_write_data_to_child;
        int aof_pipe_read_data_from_parent;
        int aof_pipe_write_ack_to_parent;
        int aof_pipe_read_ack_from_child;
        int aof_pipe_write_ack_to_child;
        int aof_pipe_read_ack_from_parent;
        int aof_stop_sending_diff;     /* If true stop sending accumulated diffs
                                          to child process. */
        sds aof_child_diff;             /* AOF diff accumulator child side. */
        /* RDB persistence */
        long long dirty;                /* Changes to DB from the last save */
        long long dirty_before_bgsave;  /* Used to restore dirty on failed BGSAVE */
        pid_t rdb_child_pid;            /* PID of RDB saving child */
        struct saveparam *saveparams;   /* Save points array for RDB */
        int saveparamslen;              /* Number of saving points */
        char *rdb_filename;             /* Name of RDB file */
        int rdb_compression;            /* Use compression in RDB? */
        int rdb_checksum;               /* Use RDB checksum? */
        time_t lastsave;                /* Unix time of last successful save */
        time_t lastbgsave_try;          /* Unix time of last attempted bgsave */
        time_t rdb_save_time_last;      /* Time used by last RDB save run. */
        time_t rdb_save_time_start;     /* Current RDB save start time. */
        int rdb_bgsave_scheduled;       /* BGSAVE when possible if true. */
        int rdb_child_type;             /* Type of save by active child. */
        int lastbgsave_status;          /* C_OK or C_ERR */
        int stop_writes_on_bgsave_err;  /* Don't allow writes if can't BGSAVE */
        int rdb_pipe_write;             /* RDB pipes used to transfer the rdb */
        int rdb_pipe_read;              /* data to the parent process in diskless repl. */
        connection **rdb_pipe_conns;    /* Connections which are currently the */
        int rdb_pipe_numconns;          /* target of diskless rdb fork child. */
        int rdb_pipe_numconns_writing;  /* Number of rdb conns with pending writes. */
        char *rdb_pipe_buff;            /* In diskless replication, this buffer holds data */
        int rdb_pipe_bufflen;           /* that was read from the the rdb pipe. */
        int rdb_key_save_delay;         /* Delay in microseconds between keys while
                                         * writing the RDB. (for testings) */
        int key_load_delay;             /* Delay in microseconds between keys while
                                         * loading aof or rdb. (for testings) */
        /* Pipe and data structures for child -> parent info sharing. */
        int child_info_pipe[2];         /* Pipe used to write the child_info_data. */
        struct {
            int process_type;           /* AOF or RDB child? */
            size_t cow_size;            /* Copy on write size. */
            unsigned long long magic;   /* Magic value to make sure data is valid. */
        } child_info_data;
        /* Propagation of commands in AOF / replication */
        redisOpArray also_propagate;    /* Additional command to propagate. */
        /* Logging */
        char *logfile;                  /* Path of log file */
        int syslog_enabled;             /* Is syslog enabled? */
        char *syslog_ident;             /* Syslog ident */
        int syslog_facility;            /* Syslog facility */
        /* Replication (master) */
        char replid[CONFIG_RUN_ID_SIZE+1];  /* My current replication ID. */
        char replid2[CONFIG_RUN_ID_SIZE+1]; /* replid inherited from master*/
        long long master_repl_offset;   /* My current replication offset */
        long long second_replid_offset; /* Accept offsets up to this for replid2. */
        int slaveseldb;                 /* Last SELECTed DB in replication output */
        int repl_ping_slave_period;     /* Master pings the slave every N seconds */
        char *repl_backlog;             /* Replication backlog for partial syncs */
        long long repl_backlog_size;    /* Backlog circular buffer size */
        long long repl_backlog_histlen; /* Backlog actual data length */
        long long repl_backlog_idx;     /* Backlog circular buffer current offset,
                                           that is the next byte will'll write to.*/
        long long repl_backlog_off;     /* Replication "master offset" of first
                                           byte in the replication backlog buffer.*/
        time_t repl_backlog_time_limit; /* Time without slaves after the backlog
                                           gets released. */
        time_t repl_no_slaves_since;    /* We have no slaves since that time.
                                           Only valid if server.slaves len is 0. */
        int repl_min_slaves_to_write;   /* Min number of slaves to write. */
        int repl_min_slaves_max_lag;    /* Max lag of <count> slaves to write. */
        int repl_good_slaves_count;     /* Number of slaves with lag <= max_lag. */
        int repl_diskless_sync;         /* Master send RDB to slaves sockets directly. */
        int repl_diskless_load;         /* Slave parse RDB directly from the socket.
                                         * see REPL_DISKLESS_LOAD_* enum */
        int repl_diskless_sync_delay;   /* Delay to start a diskless repl BGSAVE. */
        /* Replication (slave) */
        char *masteruser;               /* AUTH with this user and masterauth with master */
        char *masterauth;               /* AUTH with this password with master */
        char *masterhost;               /* Hostname of master */
        int masterport;                 /* Port of master */
        int repl_timeout;               /* Timeout after N seconds of master idle */
        client *master;     /* Client that is master for this slave */
        client *cached_master; /* Cached master to be reused for PSYNC. */
        int repl_syncio_timeout; /* Timeout for synchronous I/O calls */
        int repl_state;          /* Replication status if the instance is a slave */
        off_t repl_transfer_size; /* Size of RDB to read from master during sync. */
        off_t repl_transfer_read; /* Amount of RDB read from master during sync. */
        off_t repl_transfer_last_fsync_off; /* Offset when we fsync-ed last time. */
        connection *repl_transfer_s;     /* Slave -> Master SYNC connection */
        int repl_transfer_fd;    /* Slave -> Master SYNC temp file descriptor */
        char *repl_transfer_tmpfile; /* Slave-> master SYNC temp file name */
        time_t repl_transfer_lastio; /* Unix time of the latest read, for timeout */
        int repl_serve_stale_data; /* Serve stale data when link is down? */
        int repl_slave_ro;          /* Slave is read only? */
        int repl_slave_ignore_maxmemory;    /* If true slaves do not evict. */
        time_t repl_down_since; /* Unix time at which link with master went down */
        int repl_disable_tcp_nodelay;   /* Disable TCP_NODELAY after SYNC? */
        int slave_priority;             /* Reported in INFO and used by Sentinel. */
        int slave_announce_port;        /* Give the master this listening port. */
        char *slave_announce_ip;        /* Give the master this ip address. */
        /* The following two fields is where we store master PSYNC replid/offset
         * while the PSYNC is in progress. At the end we'll copy the fields into
         * the server->master client structure. */
        char master_replid[CONFIG_RUN_ID_SIZE+1];  /* Master PSYNC runid. */
        long long master_initial_offset;           /* Master PSYNC offset. */
        int repl_slave_lazy_flush;          /* Lazy FLUSHALL before loading DB? */
        /* Replication script cache. */
        dict *repl_scriptcache_dict;        /* SHA1 all slaves are aware of. */
        list *repl_scriptcache_fifo;        /* First in, first out LRU eviction. */
        unsigned int repl_scriptcache_size; /* Max number of elements. */
        /* Synchronous replication. */
        list *clients_waiting_acks;         /* Clients waiting in WAIT command. */
        int get_ack_from_slaves;            /* If true we send REPLCONF GETACK. */
        /* Limits */
        unsigned int maxclients;            /* Max number of simultaneous clients */
        unsigned long long maxmemory;   /* Max number of memory bytes to use */
        int maxmemory_policy;           /* Policy for key eviction */
        int maxmemory_samples;          /* Pricision of random sampling */
        int lfu_log_factor;             /* LFU logarithmic counter factor. */
        int lfu_decay_time;             /* LFU counter decay factor. */
        long long proto_max_bulk_len;   /* Protocol bulk length maximum size. */
        /* Blocked clients */
        unsigned int blocked_clients;   /* # of clients executing a blocking cmd.*/
        unsigned int blocked_clients_by_type[BLOCKED_NUM];
        list *unblocked_clients; /* list of clients to unblock before next loop */
        list *ready_keys;        /* List of readyList structures for BLPOP & co */
        /* Client side caching. */
        unsigned int tracking_clients;  /* # of clients with tracking enabled.*/
        int tracking_table_max_fill;    /* Max fill percentage. */
        /* Sort parameters - qsort_r() is only available under BSD so we
         * have to take this state global, in order to pass it to sortCompare() */
        int sort_desc;
        int sort_alpha;
        int sort_bypattern;
        int sort_store;
        /* Zip structure config, see redis.conf for more information  */
        size_t hash_max_ziplist_entries;
        size_t hash_max_ziplist_value;
        size_t set_max_intset_entries;
        size_t zset_max_ziplist_entries;
        size_t zset_max_ziplist_value;
        size_t hll_sparse_max_bytes;
        size_t stream_node_max_bytes;
        long long stream_node_max_entries;
        /* List parameters */
        int list_max_ziplist_size;
        int list_compress_depth;
        /* time cache */
        _Atomic time_t unixtime;    /* Unix time sampled every cron cycle. */
        time_t timezone;            /* Cached timezone. As set by tzset(). */
        int daylight_active;        /* Currently in daylight saving time. */
        mstime_t mstime;            /* 'unixtime' in milliseconds. */
        ustime_t ustime;            /* 'unixtime' in microseconds. */
        /* Pubsub */
        dict *pubsub_channels;  /* Map channels to list of subscribed clients */
        list *pubsub_patterns;  /* A list of pubsub_patterns */
        int notify_keyspace_events; /* Events to propagate via Pub/Sub. This is an
                                       xor of NOTIFY_... flags. */
        /* Cluster */
        int cluster_enabled;      /* Is cluster enabled? */
        mstime_t cluster_node_timeout; /* Cluster node timeout. */
        char *cluster_configfile; /* Cluster auto-generated config file name. */
        struct clusterState *cluster;  /* State of the cluster */
        int cluster_migration_barrier; /* Cluster replicas migration barrier. */
        int cluster_slave_validity_factor; /* Slave max data age for failover. */
        int cluster_require_full_coverage; /* If true, put the cluster down if
                                              there is at least an uncovered slot.*/
        int cluster_slave_no_failover;  /* Prevent slave from starting a failover
                                           if the master is in failure state. */
        char *cluster_announce_ip;  /* IP address to announce on cluster bus. */
        int cluster_announce_port;     /* base port to announce on cluster bus. */
        int cluster_announce_bus_port; /* bus port to announce on cluster bus. */
        int cluster_module_flags;      /* Set of flags that Redis modules are able
                                          to set in order to suppress certain
                                          native Redis Cluster features. Check the
                                          REDISMODULE_CLUSTER_FLAG_*. */
        /* Scripting */
        lua_State *lua; /* The Lua interpreter. We use just one for all clients */
        client *lua_client;   /* The "fake client" to query Redis from Lua */
        client *lua_caller;   /* The client running EVAL right now, or NULL */
        char* lua_cur_script; /* SHA1 of the script currently running, or NULL */
        dict *lua_scripts;         /* A dictionary of SHA1 -> Lua scripts */
        unsigned long long lua_scripts_mem;  /* Cached scripts' memory + oh */
        mstime_t lua_time_limit;  /* Script timeout in milliseconds */
        mstime_t lua_time_start;  /* Start time of script, milliseconds time */
        int lua_write_dirty;  /* True if a write command was called during the
                                 execution of the current script. */
        int lua_random_dirty; /* True if a random command was called during the
                                 execution of the current script. */
        int lua_replicate_commands; /* True if we are doing single commands repl. */
        int lua_multi_emitted;/* True if we already proagated MULTI. */
        int lua_repl;         /* Script replication flags for redis.set_repl(). */
        int lua_timedout;     /* True if we reached the time limit for script
                                 execution. */
        int lua_kill;         /* Kill the script if true. */
        int lua_always_replicate_commands; /* Default replication type. */
        /* Lazy free */
        int lazyfree_lazy_eviction;
        int lazyfree_lazy_expire;
        int lazyfree_lazy_server_del;
        /* Latency monitor */
        long long latency_monitor_threshold;
        dict *latency_events;
        /* ACLs */
        char *acl_filename;     /* ACL Users file. NULL if not configured. */
        /* Assert & bug reporting */
        const char *assert_failed;
        const char *assert_file;
        int assert_line;
        int bug_report_start; /* True if bug report header was already logged. */
        int watchdog_period;  /* Software watchdog period in ms. 0 = off */
        /* System hardware info */
        size_t system_memory_size;  /* Total memory in system as reported by OS */
        /* TLS Configuration */
        int tls_cluster;
        int tls_replication;
        int tls_auth_clients;
        redisTLSContextConfig tls_ctx_config;
    };

## 客户端 ##

### 数据结构 ###

typedef struct client {
        uint64_t id;            /* Client incremental unique ID. */
        connection *conn;
        int resp;               /* RESP protocol version. Can be 2 or 3. */
        redisDb *db;            /* Pointer to currently SELECTed DB. */
        robj *name;             /* As set by CLIENT SETNAME. */
        sds querybuf;           /* Buffer we use to accumulate client queries. */
        size_t qb_pos;          /* The position we have read in querybuf. */
        sds pending_querybuf;   /* If this client is flagged as master, this buffer
                                   represents the yet not applied portion of the
                                   replication stream that we are receiving from
                                   the master. */
        size_t querybuf_peak;   /* Recent (100ms or more) peak of querybuf size. */
        int argc;               /* Num of arguments of current command. */
        robj **argv;            /* Arguments of current command. */
        struct redisCommand *cmd, *lastcmd;  /* Last command executed. */
        user *user;             /* User associated with this connection. If the
                                   user is set to NULL the connection can do
                                   anything (admin). */
        int reqtype;            /* Request protocol type: PROTO_REQ_* */
        int multibulklen;       /* Number of multi bulk arguments left to read. */
        long bulklen;           /* Length of bulk argument in multi bulk request. */
        list *reply;            /* List of reply objects to send to the client. */
        unsigned long long reply_bytes; /* Tot bytes of objects in reply list. */
        size_t 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;
        uint64_t flags;         /* Client flags: CLIENT_* macros. */
        int authenticated;      /* Needed when the default user requires auth. */
        int replstate;          /* Replication state if this is a slave. */
        int repl_put_online_on_ack; /* Install slave write handler on first 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 read_reploff; /* Read replication offset if this is a master. */
        long long reploff;      /* Applied replication offset if this is a 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 replid[CONFIG_RUN_ID_SIZE+1]; /* Master replication ID (if master). */
        int slave_listening_port; /* As configured with: SLAVECONF listening-port */
        char slave_ip[NET_IP_STR_LEN]; /* Optionally given by REPLCONF ip-address */
        int slave_capa;         /* Slave capabilities: SLAVE_CAPA_* bitwise OR. */
        multiState mstate;      /* MULTI/EXEC state */
        int btype;              /* Type of blocking op if CLIENT_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. */
        listNode *client_list_node; /* list node in client list */
    
        /* If this client is in tracking mode and this field is non zero,
         * invalidation messages for keys fetched by this client will be send to
         * the specified client ID. */
        uint64_t client_tracking_redirection;
    
        /* Response buffer */
        int bufpos;
        char buf[PROTO_REPLY_CHUNK_BYTES];
    } client;