关键词搜索

源码搜索 ×
×

漫话Redis源码之七十三

发布2022-02-13浏览566次

详情内容

加载配置文件中的信息，非重点：

  1. /* Load the cluster config from 'filename'.
  2.  *
  3.  * If the file does not exist or is zero-length (this may happen because
  4.  * when we lock the nodes.conf file, we create a zero-length one for the
  5.  * sake of locking if it does not already exist), C_ERR is returned.
  6.  * If the configuration was loaded from the file, C_OK is returned. */
  7. int clusterLoadConfig(char *filename) {
  8.     FILE *fp = fopen(filename,"r");
  9.     struct stat sb;
  10.     char *line;
  11.     int maxline, j;
  12.  
  13.     if (fp == NULL) {
  14.         if (errno == ENOENT) {
  15.             return C_ERR;
  16.         } else {
  17.             serverLog(LL_WARNING,
  18.                 "Loading the cluster node config from %s: %s",
  19.                 filename, strerror(errno));
  20.             exit(1);
  21.         }
  22.     }
  23.  
  24.     /* Check if the file is zero-length: if so return C_ERR to signal
  25.      * we have to write the config. */
  26.     if (fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) {
  27.         fclose(fp);
  28.         return C_ERR;
  29.     }
  30.  
  31.     /* Parse the file. Note that single lines of the cluster config file can
  32.      * be really long as they include all the hash slots of the node.
  33.      * This means in the worst possible case, half of the Redis slots will be
  34.      * present in a single line, possibly in importing or migrating state, so
  35.      * together with the node ID of the sender/receiver.
  36.      *
  37.      * To simplify we allocate 1024+CLUSTER_SLOTS*128 bytes per line. */
  38.     maxline = 1024+CLUSTER_SLOTS*128;
  39.     line = zmalloc(maxline);
  40.     while(fgets(line,maxline,fp) != NULL) {
  41.         int argc;
  42.         sds *argv;
  43.         clusterNode *n, *master;
  44.         char *p, *s;
  45.  
  46.         /* Skip blank lines, they can be created either by users manually
  47.          * editing nodes.conf or by the config writing process if stopped
  48.          * before the truncate() call. */
  49.         if (line[0] == '\n' || line[0] == '\0') continue;
  50.  
  51.         /* Split the line into arguments for processing. */
  52.         argv = sdssplitargs(line,&argc);
  53.         if (argv == NULL) goto fmterr;
  54.  
  55.         /* Handle the special "vars" line. Don't pretend it is the last
  56.          * line even if it actually is when generated by Redis. */
  57.         if (strcasecmp(argv[0],"vars") == 0) {
  58.             if (!(argc % 2)) goto fmterr;
  59.             for (j = 1; j < argc; j += 2) {
  60.                 if (strcasecmp(argv[j],"currentEpoch") == 0) {
  61.                     server.cluster->currentEpoch =
  62.                             strtoull(argv[j+1],NULL,10);
  63.                 } else if (strcasecmp(argv[j],"lastVoteEpoch") == 0) {
  64.                     server.cluster->lastVoteEpoch =
  65.                             strtoull(argv[j+1],NULL,10);
  66.                 } else {
  67.                     serverLog(LL_WARNING,
  68.                         "Skipping unknown cluster config variable '%s'",
  69.                         argv[j]);
  70.                 }
  71.             }
  72.             sdsfreesplitres(argv,argc);
  73.             continue;
  74.         }
  75.  
  76.         /* Regular config lines have at least eight fields */
  77.         if (argc < 8) {
  78.             sdsfreesplitres(argv,argc);
  79.             goto fmterr;
  80.         }
  81.  
  82.         /* Create this node if it does not exist */
  83.         n = clusterLookupNode(argv[0]);
  84.         if (!n) {
  85.             n = createClusterNode(argv[0],0);
  86.             clusterAddNode(n);
  87.         }
  88.         /* Address and port */
  89.         if ((p = strrchr(argv[1],':')) == NULL) {
  90.             sdsfreesplitres(argv,argc);
  91.             goto fmterr;
  92.         }
  93.         *p = '\0';
  94.         memcpy(n->ip,argv[1],strlen(argv[1])+1);
  95.         char *port = p+1;
  96.         char *busp = strchr(port,'@');
  97.         if (busp) {
  98.             *busp = '\0';
  99.             busp++;
  100.         }
  101.         n->port = atoi(port);
  102.         /* In older versions of nodes.conf the "@busport" part is missing.
  103.          * In this case we set it to the default offset of 10000 from the
  104.          * base port. */
  105.         n->cport = busp ? atoi(busp) : n->port + CLUSTER_PORT_INCR;
  106.  
  107.         /* The plaintext port for client in a TLS cluster (n->pport) is not
  108.          * stored in nodes.conf. It is received later over the bus protocol. */
  109.  
  110.         /* Parse flags */
  111.         p = s = argv[2];
  112.         while(p) {
  113.             p = strchr(s,',');
  114.             if (p) *p = '\0';
  115.             if (!strcasecmp(s,"myself")) {
  116.                 serverAssert(server.cluster->myself == NULL);
  117.                 myself = server.cluster->myself = n;
  118.                 n->flags |= CLUSTER_NODE_MYSELF;
  119.             } else if (!strcasecmp(s,"master")) {
  120.                 n->flags |= CLUSTER_NODE_MASTER;
  121.             } else if (!strcasecmp(s,"slave")) {
  122.                 n->flags |= CLUSTER_NODE_SLAVE;
  123.             } else if (!strcasecmp(s,"fail?")) {
  124.                 n->flags |= CLUSTER_NODE_PFAIL;
  125.             } else if (!strcasecmp(s,"fail")) {
  126.                 n->flags |= CLUSTER_NODE_FAIL;
  127.                 n->fail_time = mstime();
  128.             } else if (!strcasecmp(s,"handshake")) {
  129.                 n->flags |= CLUSTER_NODE_HANDSHAKE;
  130.             } else if (!strcasecmp(s,"noaddr")) {
  131.                 n->flags |= CLUSTER_NODE_NOADDR;
  132.             } else if (!strcasecmp(s,"nofailover")) {
  133.                 n->flags |= CLUSTER_NODE_NOFAILOVER;
  134.             } else if (!strcasecmp(s,"noflags")) {
  135.                 /* nothing to do */
  136.             } else {
  137.                 serverPanic("Unknown flag in redis cluster config file");
  138.             }
  139.             if (p) s = p+1;
  140.         }
  141.  
  142.         /* Get master if any. Set the master and populate master's
  143.          * slave list. */
  144.         if (argv[3][0] != '-') {
  145.             master = clusterLookupNode(argv[3]);
  146.             if (!master) {
  147.                 master = createClusterNode(argv[3],0);
  148.                 clusterAddNode(master);
  149.             }
  150.             n->slaveof = master;
  151.             clusterNodeAddSlave(master,n);
  152.         }
  153.  
  154.         /* Set ping sent / pong received timestamps */
  155.         if (atoi(argv[4])) n->ping_sent = mstime();
  156.         if (atoi(argv[5])) n->pong_received = mstime();
  157.  
  158.         /* Set configEpoch for this node. */
  159.         n->configEpoch = strtoull(argv[6],NULL,10);
  160.  
  161.         /* Populate hash slots served by this instance. */
  162.         for (j = 8; j < argc; j++) {
  163.             int start, stop;
  164.  
  165.             if (argv[j][0] == '[') {
  166.                 /* Here we handle migrating / importing slots */
  167.                 int slot;
  168.                 char direction;
  169.                 clusterNode *cn;
  170.  
  171.                 p = strchr(argv[j],'-');
  172.                 serverAssert(p != NULL);
  173.                 *p = '\0';
  174.                 direction = p[1]; /* Either '>' or '<' */
  175.                 slot = atoi(argv[j]+1);
  176.                 if (slot < 0 || slot >= CLUSTER_SLOTS) {
  177.                     sdsfreesplitres(argv,argc);
  178.                     goto fmterr;
  179.                 }
  180.                 p += 3;
  181.                 cn = clusterLookupNode(p);
  182.                 if (!cn) {
  183.                     cn = createClusterNode(p,0);
  184.                     clusterAddNode(cn);
  185.                 }
  186.                 if (direction == '>') {
  187.                     server.cluster->migrating_slots_to[slot] = cn;
  188.                 } else {
  189.                     server.cluster->importing_slots_from[slot] = cn;
  190.                 }
  191.                 continue;
  192.             } else if ((p = strchr(argv[j],'-')) != NULL) {
  193.                 *p = '\0';
  194.                 start = atoi(argv[j]);
  195.                 stop = atoi(p+1);
  196.             } else {
  197.                 start = stop = atoi(argv[j]);
  198.             }
  199.             if (start < 0 || start >= CLUSTER_SLOTS ||
  200.                 stop < 0 || stop >= CLUSTER_SLOTS)
  201.             {
  202.                 sdsfreesplitres(argv,argc);
  203.                 goto fmterr;
  204.             }
  205.             while(start <= stop) clusterAddSlot(n, start++);
  206.         }
  207.  
  208.         sdsfreesplitres(argv,argc);
  209.     }
  210.     /* Config sanity check */
  211.     if (server.cluster->myself == NULL) goto fmterr;
  212.  
  213.     zfree(line);
  214.     fclose(fp);
  215.  
  216.     serverLog(LL_NOTICE,"Node configuration loaded, I'm %.40s", myself->name);
  217.  
  218.     /* Something that should never happen: currentEpoch smaller than
  219.      * the max epoch found in the nodes configuration. However we handle this
  220.      * as some form of protection against manual editing of critical files. */
  221.     if (clusterGetMaxEpoch() > server.cluster->currentEpoch) {
  222.         server.cluster->currentEpoch = clusterGetMaxEpoch();
  223.     }
  224.     return C_OK;
  225.  
  226. fmterr:
  227.     serverLog(LL_WARNING,
  228.         "Unrecoverable error: corrupted cluster config file.");
  229.     zfree(line);
  230.     if (fp) fclose(fp);
  231.     exit(1);
  232. }
  233.  
  234. /* Cluster node configuration is exactly the same as CLUSTER NODES output.
  235.  *
  236.  * This function writes the node config and returns 0, on error -1
  237.  * is returned.
  238.  *
  239.  * Note: we need to write the file in an atomic way from the point of view
  240.  * of the POSIX filesystem semantics, so that if the server is stopped
  241.  * or crashes during the write, we'll end with either the old file or the
  242.  * new one. Since we have the full payload to write available we can use
  243.  * a single write to write the whole file. If the pre-existing file was
  244.  * bigger we pad our payload with newlines that are anyway ignored and truncate
  245.  * the file afterward. */
  246. int clusterSaveConfig(int do_fsync) {
  247.     sds ci;
  248.     size_t content_size;
  249.     struct stat sb;
  250.     int fd;
  251.  
  252.     server.cluster->todo_before_sleep &= ~CLUSTER_TODO_SAVE_CONFIG;
  253.  
  254.     /* Get the nodes description and concatenate our "vars" directive to
  255.      * save currentEpoch and lastVoteEpoch. */
  256.     ci = clusterGenNodesDescription(CLUSTER_NODE_HANDSHAKE, 0);
  257.     ci = sdscatprintf(ci,"vars currentEpoch %llu lastVoteEpoch %llu\n",
  258.         (unsigned long long) server.cluster->currentEpoch,
  259.         (unsigned long long) server.cluster->lastVoteEpoch);
  260.     content_size = sdslen(ci);
  261.  
  262.     if ((fd = open(server.cluster_configfile,O_WRONLY|O_CREAT,0644))
  263.         == -1) goto err;
  264.  
  265.     /* Pad the new payload if the existing file length is greater. */
  266.     if (fstat(fd,&sb) != -1) {
  267.         if (sb.st_size > (off_t)content_size) {
  268.             ci = sdsgrowzero(ci,sb.st_size);
  269.             memset(ci+content_size,'\n',sb.st_size-content_size);
  270.         }
  271.     }
  272.     if (write(fd,ci,sdslen(ci)) != (ssize_t)sdslen(ci)) goto err;
  273.     if (do_fsync) {
  274.         server.cluster->todo_before_sleep &= ~CLUSTER_TODO_FSYNC_CONFIG;
  275.         if (fsync(fd) == -1) goto err;
  276.     }
  277.  
  278.     /* Truncate the file if needed to remove the final \n padding that
  279.      * is just garbage. */
  280.     if (content_size != sdslen(ci) && ftruncate(fd,content_size) == -1) {
  281.         /* ftruncate() failing is not a critical error. */
  282.     }
  283.     close(fd);
  284.     sdsfree(ci);
  285.     return 0;
  286.  
  287. err:
  288.     if (fd != -1) close(fd);
  289.     sdsfree(ci);
  290.     return -1;
  291. }
  292.  
  293. void clusterSaveConfigOrDie(int do_fsync) {
  294.     if (clusterSaveConfig(do_fsync) == -1) {
  295.         serverLog(LL_WARNING,"Fatal: can't update cluster config file.");
  296.         exit(1);
  297.     }
  298. }
  299.  
  300. /* Lock the cluster config using flock(), and leaks the file descriptor used to
  301.  * acquire the lock so that the file will be locked forever.
  302.  *
  303.  * This works because we always update nodes.conf with a new version
  304.  * in-place, reopening the file, and writing to it in place (later adjusting
  305.  * the length with ftruncate()).
  306.  *
  307.  * On success C_OK is returned, otherwise an error is logged and
  308.  * the function returns C_ERR to signal a lock was not acquired. */
  309. int clusterLockConfig(char *filename) {
  310. /* flock() does not exist on Solaris
  311.  * and a fcntl-based solution won't help, as we constantly re-open that file,
  312.  * which will release _all_ locks anyway
  313.  */
  314. #if !defined(__sun)
  315.     /* To lock it, we need to open the file in a way it is created if
  316.      * it does not exist, otherwise there is a race condition with other
  317.      * processes. */
  318.     int fd = open(filename,O_WRONLY|O_CREAT|O_CLOEXEC,0644);
  319.     if (fd == -1) {
  320.         serverLog(LL_WARNING,
  321.             "Can't open %s in order to acquire a lock: %s",
  322.             filename, strerror(errno));
  323.         return C_ERR;
  324.     }
  325.  
  326.     if (flock(fd,LOCK_EX|LOCK_NB) == -1) {
  327.         if (errno == EWOULDBLOCK) {
  328.             serverLog(LL_WARNING,
  329.                  "Sorry, the cluster configuration file %s is already used "
  330.                  "by a different Redis Cluster node. Please make sure that "
  331.                  "different nodes use different cluster configuration "
  332.                  "files.", filename);
  333.         } else {
  334.             serverLog(LL_WARNING,
  335.                 "Impossible to lock %s: %s", filename, strerror(errno));
  336.         }
  337.         close(fd);
  338.         return C_ERR;
  339.     }
  340.     /* Lock acquired: leak the 'fd' by not closing it, so that we'll retain the
  341.      * lock to the file as long as the process exists.
  342.      *
  343.      * After fork, the child process will get the fd opened by the parent process,
  344.      * we need save `fd` to `cluster_config_file_lock_fd`, so that in redisFork(),
  345.      * it will be closed in the child process.
  346.      * If it is not closed, when the main process is killed -9, but the child process
  347.      * (redis-aof-rewrite) is still alive, the fd(lock) will still be held by the
  348.      * child process, and the main process will fail to get lock, means fail to start. */
  349.     server.cluster_config_file_lock_fd = fd;
  350. #else
  351.     UNUSED(filename);
  352. #endif /* __sun */
  353.  
  354.     return C_OK;
  355. }
  356.  
  357. /* Derives our ports to be announced in the cluster bus. */
  358. void deriveAnnouncedPorts(int *announced_port, int *announced_pport,
  359.                           int *announced_cport) {
  360.     int port = server.tls_cluster ? server.tls_port : server.port;
  361.     /* Default announced ports. */
  362.     *announced_port = port;
  363.     *announced_pport = server.tls_cluster ? server.port : 0;
  364.     *announced_cport = port + CLUSTER_PORT_INCR;
  365.     /* Config overriding announced ports. */
  366.     if (server.tls_cluster && server.cluster_announce_tls_port) {
  367.         *announced_port = server.cluster_announce_tls_port;
  368.         *announced_pport = server.cluster_announce_port;
  369.     } else if (server.cluster_announce_port) {
  370.         *announced_port = server.cluster_announce_port;
  371.     }
  372.     if (server.cluster_announce_bus_port) {
  373.         *announced_cport = server.cluster_announce_bus_port;
  374.     }
  375. }
  376.  
  377. /* Some flags (currently just the NOFAILOVER flag) may need to be updated
  378.  * in the "myself" node based on the current configuration of the node,
  379.  * that may change at runtime via CONFIG SET. This function changes the
  380.  * set of flags in myself->flags accordingly. */
  381. void clusterUpdateMyselfFlags(void) {
  382.     int oldflags = myself->flags;
  383.     int nofailover = server.cluster_slave_no_failover ?
  384.                      CLUSTER_NODE_NOFAILOVER : 0;
  385.     myself->flags &= ~CLUSTER_NODE_NOFAILOVER;
  386.     myself->flags |= nofailover;
  387.     if (myself->flags != oldflags) {
  388.         clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|
  389.                              CLUSTER_TODO_UPDATE_STATE);
  390.     }
  391. }
  392.  
  393. void clusterInit(void) {
  394.     int saveconf = 0;
  395.  
  396.     server.cluster = zmalloc(sizeof(clusterState));
  397.     server.cluster->myself = NULL;
  398.     server.cluster->currentEpoch = 0;
  399.     server.cluster->state = CLUSTER_FAIL;
  400.     server.cluster->size = 1;
  401.     server.cluster->todo_before_sleep = 0;
  402.     server.cluster->nodes = dictCreate(&clusterNodesDictType,NULL);
  403.     server.cluster->nodes_black_list =
  404.         dictCreate(&clusterNodesBlackListDictType,NULL);
  405.     server.cluster->failover_auth_time = 0;
  406.     server.cluster->failover_auth_count = 0;
  407.     server.cluster->failover_auth_rank = 0;
  408.     server.cluster->failover_auth_epoch = 0;
  409.     server.cluster->cant_failover_reason = CLUSTER_CANT_FAILOVER_NONE;
  410.     server.cluster->lastVoteEpoch = 0;
  411.     for (int i = 0; i < CLUSTERMSG_TYPE_COUNT; i++) {
  412.         server.cluster->stats_bus_messages_sent[i] = 0;
  413.         server.cluster->stats_bus_messages_received[i] = 0;
  414.     }
  415.     server.cluster->stats_pfail_nodes = 0;
  416.     memset(server.cluster->slots,0, sizeof(server.cluster->slots));
  417.     clusterCloseAllSlots();
  418.  
  419.     /* Lock the cluster config file to make sure every node uses
  420.      * its own nodes.conf. */
  421.     server.cluster_config_file_lock_fd = -1;
  422.     if (clusterLockConfig(server.cluster_configfile) == C_ERR)
  423.         exit(1);
  424.  
  425.     /* Load or create a new nodes configuration. */
  426.     if (clusterLoadConfig(server.cluster_configfile) == C_ERR) {
  427.         /* No configuration found. We will just use the random name provided
  428.          * by the createClusterNode() function. */
  429.         myself = server.cluster->myself =
  430.             createClusterNode(NULL,CLUSTER_NODE_MYSELF|CLUSTER_NODE_MASTER);
  431.         serverLog(LL_NOTICE,"No cluster configuration found, I'm %.40s",
  432.             myself->name);
  433.         clusterAddNode(myself);
  434.         saveconf = 1;
  435.     }
  436.     if (saveconf) clusterSaveConfigOrDie(1);
  437.  
  438.     /* We need a listening TCP port for our cluster messaging needs. */
  439.     server.cfd.count = 0;
  440.  
  441.     /* Port sanity check II
  442.      * The other handshake port check is triggered too late to stop
  443.      * us from trying to use a too-high cluster port number. */
  444.     int port = server.tls_cluster ? server.tls_port : server.port;
  445.     if (port > (65535-CLUSTER_PORT_INCR)) {
  446.         serverLog(LL_WARNING, "Redis port number too high. "
  447.                    "Cluster communication port is 10,000 port "
  448.                    "numbers higher than your Redis port. "
  449.                    "Your Redis port number must be 55535 or less.");
  450.         exit(1);
  451.     }
  452.     if (listenToPort(port+CLUSTER_PORT_INCR, &server.cfd) == C_ERR) {
  453.         exit(1);
  454.     }
  455.     if (createSocketAcceptHandler(&server.cfd, clusterAcceptHandler) != C_OK) {
  456.         serverPanic("Unrecoverable error creating Redis Cluster socket accept handler.");
  457.     }
  458.  
  459.     /* The slots -> keys map is a radix tree. Initialize it here. */
  460.     server.cluster->slots_to_keys = raxNew();
  461.     memset(server.cluster->slots_keys_count,0,
  462.            sizeof(server.cluster->slots_keys_count));
  463.  
  464.     /* Set myself->port/cport/pport to my listening ports, we'll just need to
  465.      * discover the IP address via MEET messages. */
  466.     deriveAnnouncedPorts(&myself->port, &myself->pport, &myself->cport);
  467.  
  468.     server.cluster->mf_end = 0;
  469.     resetManualFailover();
  470.     clusterUpdateMyselfFlags();
  471. }

相关技术文章

最新源码

下载排行榜
点击QQ咨询
开通会员
返回顶部
×
微信扫码支付 元
微信扫码支付
确定支付下载
请使用微信描二维码支付
×

提示信息

×

选择支付方式

  • 微信支付
  • 支付宝付款
确定支付下载