WIP. memory leak in jemalloc when do reload

Author: yongman

src/bio.c

@@ -98,22 +98,19 @@ void bioRestore(pthread_t *threads) {
  memcpy(bio_threads, threads, sizeof(bio_threads));
 
  for (j = 0; j < BIO_NUM_OPS; j++) {
- pthread_mutex_init(&bio_mutex[j],NULL);
- pthread_cond_init(&bio_newjob_cond[j],NULL);
- pthread_cond_init(&bio_step_cond[j],NULL);
  bio_jobs[j] = listCreate();
  bio_pending[j] = 0;
  }
 }
 
-static bool bioEmpty() {
+static unsigned bioEmpty() {
  int i;
  for (i = 0; i < BIO_NUM_OPS;i++) {
  if (bio_pending[i] > 0) {
- return false;
+ return 0;
  }
  }
- return true;
+ return 1;
 }
 
 void bioSave(pthread_t *threads) {

src/redis-server-wrap.c

@@ -19,9 +19,10 @@ int main(int argc, char ** argv)
  RELOAD reload = NULL;
  int ret;
 
- ctx.reload = false;
+ ctx.reload = 0;
 
  strncpy(ctx.hashseed, "1234567890123456", 16);
+ ctx.reloadTimes = 0;
 
  for(;;) {
  printf("load dynamic library\n");
@@ -50,7 +51,8 @@ int main(int argc, char ** argv)
  fprintf(stderr, "dlclose failed\n");
  exit(EXIT_FAILURE);
  }
- ctx.reload = true;
+ ctx.reload = 1;
+ ctx.reloadTimes++;
  }
  exit(EXIT_SUCCESS);
 }

src/server.c

@@ -55,6 +55,7 @@
 #include <sys/utsname.h>
 #include <locale.h>
 #include <sys/socket.h>
+#include <mcheck.h>
 
 /* Our shared "common" objects */
 
@@ -1525,6 +1526,26 @@ void initServerConfig(void) {
  server.watchdog_period = 0;
 }
 
+void reloadServerCommands() {
+ /* release old dict */
+ dictRelease(server.commands);
+ dictRelease(server.orig_commands);
+
+ /* re-add command to table */
+ 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");
+ server.sremCommand = lookupCommandByCString("srem");
+ server.execCommand = lookupCommandByCString("exec");
+ server.expireCommand = lookupCommandByCString("expire");
+ server.pexpireCommand = lookupCommandByCString("pexpire");
+}
+
 extern char **environ;
 
 /* Restart the server, executing the same executable that started this
@@ -3975,6 +3996,7 @@ int serverRun(int argc, char **argv, struct wrapperContext *ctx) {
  serverRestore(&ctx->server, &ctx->shared);
  bioRestore(ctx->bioThreads);
  evictionPoolRestore(ctx->evictionpool);
+ zmalloc_used_memory_restore(ctx->usedmemory);
  }
 
 #ifdef REDIS_TEST
@@ -4022,6 +4044,8 @@ int serverRun(int argc, char **argv, struct wrapperContext *ctx) {
  for (j = 0; j < argc; j++) server.exec_argv[j] = zstrdup(argv[j]);
  } else {
  moduleReloadModulesSystem();
+ /* command adress may be changed, so reload it */
+ reloadServerCommands();
  }
 
 
@@ -4041,7 +4065,7 @@ int serverRun(int argc, char **argv, struct wrapperContext *ctx) {
  else if (strstr(argv[0],"redis-check-aof") != NULL)
  redis_check_aof_main(argc,argv);
 
- if (argc >= 2) {
+ if (!ctx->reload && argc >= 2) {
  j = 1; /* First option to parse in argv[] */
  sds options = sdsempty();
  char *configfile = NULL;
@@ -4125,15 +4149,14 @@ int serverRun(int argc, char **argv, struct wrapperContext *ctx) {
 
  server.supervised = redisIsSupervised(server.supervised_mode);
  int background = server.daemonize && !server.supervised;
- if (background) daemonize();
-
  if (!ctx->reload) {
+ if (background) daemonize();
  initServer();
+ redisSetProcTitle(argv[0]);
  } else {
  reloadServer();
  }
  if (background || server.pidfile) createPidFile();
- redisSetProcTitle(argv[0]);
  redisAsciiArt();
  checkTcpBacklogSettings();
 
@@ -4144,7 +4167,7 @@ int serverRun(int argc, char **argv, struct wrapperContext *ctx) {
  linuxMemoryWarnings();
  #endif
  moduleLoadFromQueue();
- if (ctx->reload == 0) {
+ if (!ctx->reload) {
  loadDataFromDisk();
  }
  if (server.cluster_enabled) {
@@ -4172,10 +4195,18 @@ int serverRun(int argc, char **argv, struct wrapperContext *ctx) {
  aeSetAfterSleepProc(server.el,afterSleep);
  aeMain(server.el);
 
+ if (ctx->reloadTimes == 0) {
+ mtrace();
+ }
+ if (ctx->reloadTimes == 200) {
+ muntrace();
+ }
+
  /* save server ctx to s */
  serverSave(&ctx->server, &ctx->shared);
  bioSave(ctx->bioThreads);
  evictionPoolSave(&ctx->evictionpool);
+ zmalloc_used_memory_save(&ctx->usedmemory);
 
  return 0;
 }

src/server.h

@@ -1289,8 +1289,10 @@ struct wrapperContext {
  struct sharedObjectsStruct shared;
  pthread_t bioThreads[BIO_NUM_OPS];
  void *evictionpool;
- bool reload;
+ unsigned reload:1;
  char hashseed[16];
+ unsigned int reloadTimes;
+ size_t usedmemory;
 };
 /*-----------------------------------------------------------------------------
  * Extern declarations

src/zmalloc.c

@@ -215,6 +215,15 @@ size_t zmalloc_used_memory(void) {
  return um;
 }
 
+void zmalloc_used_memory_save(size_t *m) {
+ size_t um = zmalloc_used_memory();
+ memcpy(m, &um, sizeof(size_t));
+}
+
+void zmalloc_used_memory_restore(size_t m) {
+ used_memory = m;
+}
+
 void zmalloc_set_oom_handler(void (*oom_handler)(size_t)) {
  zmalloc_oom_handler = oom_handler;
 }

src/zmalloc.h

@@ -86,6 +86,9 @@ size_t zmalloc_get_smap_bytes_by_field(char *field, long pid);
 size_t zmalloc_get_memory_size(void);
 void zlibc_free(void *ptr);
 
+void zmalloc_used_memory_restore(size_t m);
+void zmalloc_used_memory_save(size_t *m);
+
 #ifdef HAVE_DEFRAG
 void zfree_no_tcache(void *ptr);
 void *zmalloc_no_tcache(size_t size);