Redis偶发连接失败案例实战记录

clientsCron(server.h):
#define CLIENTS_CRON_MIN_ITERATIONS 5
void clientsCron(void) {
 /* Make sure to process at least numclients/server.hz of clients
  * per call. Since this function is called server.hz times per second
  * we are sure that in the worst case we process all the clients in 1
  * second. */
 int numclients = listLength(server.clients);
 int iterations = numclients/server.hz;
 mstime_t now = mstime();

 /* Process at least a few clients while we are at it, even if we need
  * to process less than CLIENTS_CRON_MIN_ITERATIONS to meet our contract
  * of processing each client once per second. */
 if (iterations  CLIENTS_CRON_MIN_ITERATIONS)
  iterations = (numclients  CLIENTS_CRON_MIN_ITERATIONS) ?
      numclients : CLIENTS_CRON_MIN_ITERATIONS;

 while(listLength(server.clients)  iterations--) {
  client *c;
  listNode *head;

  /* Rotate the list, take the current head, process.
   * This way if the client must be removed from the list it's the
   * first element and we don't incur into O(N) computation. */
  listRotate(server.clients);
  head = listFirst(server.clients);
  c = listNodeValue(head);
  /* The following functions do different service checks on the client.
   * The protocol is that they return non-zero if the client was
   * terminated. */
  if (clientsCronHandleTimeout(c,now)) continue;
  if (clientsCronResizeQueryBuffer(c)) continue;
 }
}

clientsCronResizeQueryBuffer(server.h):

/* The client query buffer is an sds.c string that can end with a lot of
 * free space not used, this function reclaims space if needed.
 *
 * The function always returns 0 as it never terminates the client. */
int clientsCronResizeQueryBuffer(client *c) {
 size_t querybuf_size = sdsAllocSize(c->querybuf);
 time_t idletime = server.unixtime - c->lastinteraction;

 /* 只在以下两种情况下会Resize query buffer:
  * 1) Query buffer > BIG_ARG(在server.h 中定义#define PROTO_MBULK_BIG_ARG  (1024*32)) 
   且这个Buffer的小于一段时间的客户端使用的峰值.
  * 2) 客户端空闲超过2s且Buffer size大于1k. */
 if (((querybuf_size > PROTO_MBULK_BIG_ARG) 
   (querybuf_size/(c->querybuf_peak+1)) > 2) ||
   (querybuf_size > 1024  idletime > 2))
 {
  /* Only resize the query buffer if it is actually wasting space. */
  if (sdsavail(c->querybuf) > 1024) {
   c->querybuf = sdsRemoveFreeSpace(c->querybuf);
  }
 }
 /* Reset the peak again to capture the peak memory usage in the next
  * cycle. */
 c->querybuf_peak = 0;
 return 0;
}

readQueryFromClient（networking.c）:
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;
  }
 }

 selectDb(c,0);
 c->id = server.next_client_id++;
 c->fd = fd;
 c->name = NULL;
 c->bufpos = 0;
 c->querybuf = sdsempty(); 初始化是0

readQueryFromClient(networking.c):
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); 在这里会被扩大

if (((querybuf_size > REDIS_MBULK_BIG_ARG) 
   (querybuf_size/(c->querybuf_peak+1)) > 2) ||
   (querybuf_size > 1024*32  idletime > 2))
 {
  /* Only resize the query buffer if it is actually wasting space. */
  if (sdsavail(c->querybuf) > 1024*32) {
   c->querybuf = sdsRemoveFreeSpace(c->querybuf);
  }
 }

arena_purge(arena.c)
static void
arena_purge(arena_t *arena, bool all)
{
 arena_chunk_t *chunk;
 size_t npurgatory;
 if (config_debug) {
  size_t ndirty = 0;

  arena_chunk_dirty_iter(arena->chunks_dirty, NULL,
   chunks_dirty_iter_cb, (void *)ndirty);
  assert(ndirty == arena->ndirty);
 }
 assert(arena->ndirty > arena->npurgatory || all);
 assert((arena->nactive >> opt_lg_dirty_mult)  (arena->ndirty -
  arena->npurgatory) || all);

 if (config_stats)
  arena->stats.npurge++;
 npurgatory = arena_compute_npurgatory(arena, all);
 arena->npurgatory += npurgatory;

 while (npurgatory > 0) {
  size_t npurgeable, npurged, nunpurged;

  /* Get next chunk with dirty pages. */
  chunk = arena_chunk_dirty_first(arena->chunks_dirty);
  if (chunk == NULL) {
   arena->npurgatory -= npurgatory;
   return;
  }
  npurgeable = chunk->ndirty;
  assert(npurgeable != 0);

  if (npurgeable > npurgatory  chunk->nruns_adjac == 0) {
 
   arena->npurgatory += npurgeable - npurgatory;
   npurgatory = npurgeable;
  }
  arena->npurgatory -= npurgeable;
  npurgatory -= npurgeable;
  npurged = arena_chunk_purge(arena, chunk, all);
  nunpurged = npurgeable - npurged;
  arena->npurgatory += nunpurged;
  npurgatory += nunpurged;
 }
}

arena_purge(arena.c)
static void
arena_purge(arena_t *arena, bool all)
{
 chunk_hooks_t chunk_hooks = chunk_hooks_get(arena);
 size_t npurge, npurgeable, npurged;
 arena_runs_dirty_link_t purge_runs_sentinel;
 extent_node_t purge_chunks_sentinel;

 arena->purging = true;

 /*
  * Calls to arena_dirty_count() are disabled even for debug builds
  * because overhead grows nonlinearly as memory usage increases.
  */
 if (false  config_debug) {
  size_t ndirty = arena_dirty_count(arena);
  assert(ndirty == arena->ndirty);
 }
 assert((arena->nactive >> arena->lg_dirty_mult)  arena->ndirty || all);

 if (config_stats)
  arena->stats.npurge++;

 npurge = arena_compute_npurge(arena, all);
 qr_new(purge_runs_sentinel, rd_link);
 extent_node_dirty_linkage_init(purge_chunks_sentinel);

 npurgeable = arena_stash_dirty(arena, chunk_hooks, all, npurge,
  purge_runs_sentinel, purge_chunks_sentinel);
 assert(npurgeable >= npurge);
 npurged = arena_purge_stashed(arena, chunk_hooks, purge_runs_sentinel,
  purge_chunks_sentinel);
 assert(npurged == npurgeable);
 arena_unstash_purged(arena, chunk_hooks, purge_runs_sentinel,
  purge_chunks_sentinel);

 arena->purging = false;
}