关键词搜索

源码搜索 ×
×

漫话Redis源码之三十一

发布2021-12-12浏览626次

详情内容

这里没什么特别的，考大家一个问题，为什么在mstime中需要加long long转化？是为了溢出，这里做得非常专业，我曾经踩过这个坑：

  1. typedef struct _client {
  2.     redisContext *context;
  3.     sds obuf;
  4.     char **randptr;         /* Pointers to :rand: strings inside the command buf */
  5.     size_t randlen;         /* Number of pointers in client->randptr */
  6.     size_t randfree;        /* Number of unused pointers in client->randptr */
  7.     char **stagptr;         /* Pointers to slot hashtags (cluster mode only) */
  8.     size_t staglen;         /* Number of pointers in client->stagptr */
  9.     size_t stagfree;        /* Number of unused pointers in client->stagptr */
  10.     size_t written;         /* Bytes of 'obuf' already written */
  11.     long long start;        /* Start time of a request */
  12.     long long latency;      /* Request latency */
  13.     int pending;            /* Number of pending requests (replies to consume) */
  14.     int prefix_pending;     /* If non-zero, number of pending prefix commands. Commands
  15.                                such as auth and select are prefixed to the pipeline of
  16.                                benchmark commands and discarded after the first send. */
  17.     int prefixlen;          /* Size in bytes of the pending prefix commands */
  18.     int thread_id;
  19.     struct clusterNode *cluster_node;
  20.     int slots_last_update;
  21. } *client;
  22.  
  23. /* Threads. */
  24.  
  25. typedef struct benchmarkThread {
  26.     int index;
  27.     pthread_t thread;
  28.     aeEventLoop *el;
  29. } benchmarkThread;
  30.  
  31. /* Cluster. */
  32. typedef struct clusterNode {
  33.     char *ip;
  34.     int port;
  35.     sds name;
  36.     int flags;
  37.     sds replicate;  /* Master ID if node is a slave */
  38.     int *slots;
  39.     int slots_count;
  40.     int current_slot_index;
  41.     int *updated_slots;         /* Used by updateClusterSlotsConfiguration */
  42.     int updated_slots_count;    /* Used by updateClusterSlotsConfiguration */
  43.     int replicas_count;
  44.     sds *migrating; /* An array of sds where even strings are slots and odd
  45.                      * strings are the destination node IDs. */
  46.     sds *importing; /* An array of sds where even strings are slots and odd
  47.                      * strings are the source node IDs. */
  48.     int migrating_count; /* Length of the migrating array (migrating slots*2) */
  49.     int importing_count; /* Length of the importing array (importing slots*2) */
  50.     struct redisConfig *redis_config;
  51. } clusterNode;
  52.  
  53. typedef struct redisConfig {
  54.     sds save;
  55.     sds appendonly;
  56. } redisConfig;
  57.  
  58. /* Prototypes */
  59. char *redisGitSHA1(void);
  60. char *redisGitDirty(void);
  61. static void writeHandler(aeEventLoop *el, int fd, void *privdata, int mask);
  62. static void createMissingClients(client c);
  63. static benchmarkThread *createBenchmarkThread(int index);
  64. static void freeBenchmarkThread(benchmarkThread *thread);
  65. static void freeBenchmarkThreads();
  66. static void *execBenchmarkThread(void *ptr);
  67. static clusterNode *createClusterNode(char *ip, int port);
  68. static redisConfig *getRedisConfig(const char *ip, int port,
  69.                                    const char *hostsocket);
  70. static redisContext *getRedisContext(const char *ip, int port,
  71.                                      const char *hostsocket);
  72. static void freeRedisConfig(redisConfig *cfg);
  73. static int fetchClusterSlotsConfiguration(client c);
  74. static void updateClusterSlotsConfiguration();
  75. int showThroughput(struct aeEventLoop *eventLoop, long long id,
  76.                    void *clientData);
  77.  
  78. static sds benchmarkVersion(void) {
  79.     sds version;
  80.     version = sdscatprintf(sdsempty(), "%s", REDIS_VERSION);
  81.  
  82.     /* Add git commit and working tree status when available */
  83.     if (strtoll(redisGitSHA1(),NULL,16)) {
  84.         version = sdscatprintf(version, " (git:%s", redisGitSHA1());
  85.         if (strtoll(redisGitDirty(),NULL,10))
  86.             version = sdscatprintf(version, "-dirty");
  87.         version = sdscat(version, ")");
  88.     }
  89.     return version;
  90. }
  91.  
  92. /* Dict callbacks */
  93. static uint64_t dictSdsHash(const void *key);
  94. static int dictSdsKeyCompare(void *privdata, const void *key1,
  95.     const void *key2);
  96.  
  97. /* Implementation */
  98. static long long ustime(void) {
  99.     struct timeval tv;
  100.     long long ust;
  101.  
  102.     gettimeofday(&tv, NULL);
  103.     ust = ((long)tv.tv_sec)*1000000;
  104.     ust += tv.tv_usec;
  105.     return ust;
  106. }
  107.  
  108. static long long mstime(void) {
  109.     struct timeval tv;
  110.     long long mst;
  111.  
  112.     gettimeofday(&tv, NULL);
  113.     mst = ((long long)tv.tv_sec)*1000;
  114.     mst += tv.tv_usec/1000;
  115.     return mst;
  116. }
  117.  
  118. static uint64_t dictSdsHash(const void *key) {
  119.     return dictGenHashFunction((unsigned char*)key, sdslen((char*)key));
  120. }
  121.  
  122. static int dictSdsKeyCompare(void *privdata, const void *key1,
  123.         const void *key2)
  124. {
  125.     int l1,l2;
  126.     DICT_NOTUSED(privdata);
  127.  
  128.     l1 = sdslen((sds)key1);
  129.     l2 = sdslen((sds)key2);
  130.     if (l1 != l2) return 0;
  131.     return memcmp(key1, key2, l1) == 0;
  132. }
  133.  
  134. /* _serverAssert is needed by dict */
  135. void _serverAssert(const char *estr, const char *file, int line) {
  136.     fprintf(stderr, "=== ASSERTION FAILED ===");
  137.     fprintf(stderr, "==> %s:%d '%s' is not true",file,line,estr);
  138.     *((char*)-1) = 'x';
  139. }
  140.  
  141. static redisContext *getRedisContext(const char *ip, int port,
  142.                                      const char *hostsocket)
  143. {
  144.     redisContext *ctx = NULL;
  145.     redisReply *reply =  NULL;
  146.     if (hostsocket == NULL)
  147.         ctx = redisConnect(ip, port);
  148.     else
  149.         ctx = redisConnectUnix(hostsocket);
  150.     if (ctx == NULL || ctx->err) {
  151.         fprintf(stderr,"Could not connect to Redis at ");
  152.         char *err = (ctx != NULL ? ctx->errstr : "");
  153.         if (hostsocket == NULL)
  154.             fprintf(stderr,"%s:%d: %s\n",ip,port,err);
  155.         else
  156.             fprintf(stderr,"%s: %s\n",hostsocket,err);
  157.         goto cleanup;
  158.     }
  159.     if (config.tls==1) {
  160.         const char *err = NULL;
  161.         if (cliSecureConnection(ctx, config.sslconfig, &err) == REDIS_ERR && err) {
  162.             fprintf(stderr, "Could not negotiate a TLS connection: %s\n", err);
  163.             goto cleanup;
  164.         }
  165.     }
  166.     if (config.auth == NULL)
  167.         return ctx;
  168.     if (config.user == NULL)
  169.         reply = redisCommand(ctx,"AUTH %s", config.auth);
  170.     else
  171.         reply = redisCommand(ctx,"AUTH %s %s", config.user, config.auth);
  172.     if (reply != NULL) {
  173.         if (reply->type == REDIS_REPLY_ERROR) {
  174.             if (hostsocket == NULL)
  175.                 fprintf(stderr, "Node %s:%d replied with error:\n%s\n", ip, port, reply->str);
  176.             else
  177.                 fprintf(stderr, "Node %s replied with error:\n%s\n", hostsocket, reply->str);
  178.             freeReplyObject(reply);
  179.             redisFree(ctx);
  180.             exit(1);
  181.         }
  182.         freeReplyObject(reply);
  183.         return ctx;
  184.     }
  185.     fprintf(stderr, "ERROR: failed to fetch reply from ");
  186.     if (hostsocket == NULL)
  187.         fprintf(stderr, "%s:%d\n", ip, port);
  188.     else
  189.         fprintf(stderr, "%s\n", hostsocket);
  190. cleanup:
  191.     freeReplyObject(reply);
  192.     redisFree(ctx);
  193.     return NULL;
  194. }
  195.  
  196.  
  197.  
  198. static redisConfig *getRedisConfig(const char *ip, int port,
  199.                                    const char *hostsocket)
  200. {
  201.     redisConfig *cfg = zcalloc(sizeof(*cfg));
  202.     if (!cfg) return NULL;
  203.     redisContext *c = NULL;
  204.     redisReply *reply = NULL, *sub_reply = NULL;
  205.     c = getRedisContext(ip, port, hostsocket);
  206.     if (c == NULL) {
  207.         freeRedisConfig(cfg);
  208.         return NULL;
  209.     }
  210.     redisAppendCommand(c, "CONFIG GET %s", "save");
  211.     redisAppendCommand(c, "CONFIG GET %s", "appendonly");
  212.     int i = 0;
  213.     void *r = NULL;
  214.     for (; i < 2; i++) {
  215.         int res = redisGetReply(c, &r);
  216.         if (reply) freeReplyObject(reply);
  217.         reply = res == REDIS_OK ? ((redisReply *) r) : NULL;
  218.         if (res != REDIS_OK || !r) goto fail;
  219.         if (reply->type == REDIS_REPLY_ERROR) {
  220.             fprintf(stderr, "ERROR: %s\n", reply->str);
  221.             goto fail;
  222.         }
  223.         if (reply->type != REDIS_REPLY_ARRAY || reply->elements < 2) goto fail;
  224.         sub_reply = reply->element[1];
  225.         char *value = sub_reply->str;
  226.         if (!value) value = "";
  227.         switch (i) {
  228.         case 0: cfg->save = sdsnew(value); break;
  229.         case 1: cfg->appendonly = sdsnew(value); break;
  230.         }
  231.     }
  232.     freeReplyObject(reply);
  233.     redisFree(c);
  234.     return cfg;
  235. fail:
  236.     fprintf(stderr, "ERROR: failed to fetch CONFIG from ");
  237.     if (hostsocket == NULL) fprintf(stderr, "%s:%d\n", ip, port);
  238.     else fprintf(stderr, "%s\n", hostsocket);
  239.     int abort_test = 0;
  240.     if (reply && reply->type == REDIS_REPLY_ERROR &&
  241.         (!strncmp(reply->str,"NOAUTH",5) ||
  242.          !strncmp(reply->str,"WRONGPASS",9) ||
  243.          !strncmp(reply->str,"NOPERM",5)))
  244.         abort_test = 1;
  245.     freeReplyObject(reply);
  246.     redisFree(c);
  247.     freeRedisConfig(cfg);
  248.     if (abort_test) exit(1);
  249.     return NULL;
  250. }
  251. static void freeRedisConfig(redisConfig *cfg) {
  252.     if (cfg->save) sdsfree(cfg->save);
  253.     if (cfg->appendonly) sdsfree(cfg->appendonly);
  254.     zfree(cfg);
  255. }
  256.  
  257. static void freeClient(client c) {
  258.     aeEventLoop *el = CLIENT_GET_EVENTLOOP(c);
  259.     listNode *ln;
  260.     aeDeleteFileEvent(el,c->context->fd,AE_WRITABLE);
  261.     aeDeleteFileEvent(el,c->context->fd,AE_READABLE);
  262.     if (c->thread_id >= 0) {
  263.         int requests_finished = 0;
  264.         atomicGet(config.requests_finished, requests_finished);
  265.         if (requests_finished >= config.requests) {
  266.             aeStop(el);
  267.         }
  268.     }
  269.     redisFree(c->context);
  270.     sdsfree(c->obuf);
  271.     zfree(c->randptr);
  272.     zfree(c->stagptr);
  273.     zfree(c);
  274.     if (config.num_threads) pthread_mutex_lock(&(config.liveclients_mutex));
  275.     config.liveclients--;
  276.     ln = listSearchKey(config.clients,c);
  277.     assert(ln != NULL);
  278.     listDelNode(config.clients,ln);
  279.     if (config.num_threads) pthread_mutex_unlock(&(config.liveclients_mutex));
  280. }
  281.  
  282. static void freeAllClients(void) {
  283.     listNode *ln = config.clients->head, *next;
  284.  
  285.     while(ln) {
  286.         next = ln->next;
  287.         freeClient(ln->value);
  288.         ln = next;
  289.     }
  290. }

相关技术文章

最新源码

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

提示信息

×

选择支付方式

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