added config option for loading callback interval keys

Former-commit-id: 9f7ed0b2327924563668105ead09f52833902d0b

src/config.cpp

@@ -2467,6 +2467,7 @@ standardConfig configs[] = {
 
     /* Long configs */
     createLongConfig("loading-process-events-interval-bytes", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, g_pserver->loading_process_events_interval_bytes, 2*1024*1024, MEMORY_CONFIG, NULL, NULL),
+    createLongConfig("loading-process-events-interval-keys", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, g_pserver->loading_process_events_interval_keys, 10000, MEMORY_CONFIG, NULL, NULL),
     /* Unsigned Long configs */
     createULongConfig("active-defrag-max-scan-fields", NULL, MODIFIABLE_CONFIG, 1, LONG_MAX, cserver.active_defrag_max_scan_fields, 1000, INTEGER_CONFIG, NULL, NULL), /* Default: keys with more than 1000 fields will be processed separately */
     createULongConfig("slowlog-max-len", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, g_pserver->slowlog_max_len, 128, INTEGER_CONFIG, NULL, NULL),

src/rdb.cpp

@@ -2165,8 +2165,11 @@ void stopSaving(int success) {
 void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) {
     if (g_pserver->rdb_checksum)
         rioGenericUpdateChecksum(r, buf, len);
-    if (g_pserver->loading_process_events_interval_bytes &&
+    
-        (r->processed_bytes + len)/g_pserver->loading_process_events_interval_bytes > r->processed_bytes/g_pserver->loading_process_events_interval_bytes)
+    if ((g_pserver->loading_process_events_interval_bytes &&
+        (r->processed_bytes + len)/g_pserver->loading_process_events_interval_bytes > r->processed_bytes/g_pserver->loading_process_events_interval_bytes) ||
+        (g_pserver->loading_process_events_interval_keys &&
+        (r->loaded_keys >= g_pserver->loading_process_events_interval_keys)))
     {
         /* The DB can take some non trivial amount of time to load. Update
          * our cached time since it is used to create and update the last
@@ -2184,6 +2187,8 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) {
         loadingProgress(r->processed_bytes);
         processEventsWhileBlocked(serverTL - g_pserver->rgthreadvar);
         processModuleLoadingProgressEvent(0);
+
+        r->loaded_keys = 0;
     }
 }
 
@@ -2490,6 +2495,8 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
         if (g_pserver->key_load_delay)
             usleep(g_pserver->key_load_delay);
 
+        rdb->loaded_keys++;
+
         /* Reset the state that is key-specified and is populated by
          * opcodes before the key, so that we start from scratch again. */
         expiretime = -1;

src/redis-check-rdb.cpp

@@ -354,6 +354,7 @@ int redis_check_rdb_main(int argc, const char **argv, FILE *fp) {
     if (shared.integers[0] == NULL)
         createSharedObjects();
     g_pserver->loading_process_events_interval_bytes = 0;
+    g_pserver->loading_process_events_interval_keys = 0;
     rdbCheckMode = 1;
     rdbCheckInfo("Checking RDB file %s", argv[1]);
     rdbCheckSetupSignals();

src/rio.cpp

@@ -94,6 +94,7 @@ static const rio rioBufferIO = {
     0,              /* current checksum */
     0,              /* flags */
     0,              /* bytes read or written */
+    0,              /* keys loaded */
     0,              /* read/write chunk size */
     { { NULL, 0 } } /* union for io-specific vars */
 };
@@ -148,6 +149,7 @@ static const rio rioFileIO = {
     0,              /* current checksum */
     0,              /* flags */
     0,              /* bytes read or written */
+    0,              /* keys loaded */
     0,              /* read/write chunk size */
     { { NULL, 0 } } /* union for io-specific vars */
 };
@@ -243,6 +245,7 @@ static const rio rioConnIO = {
     0,              /* current checksum */
     0,              /* flags */
     0,              /* bytes read or written */
+    0,              /* keys loaded */
     0,              /* read/write chunk size */
     { { NULL, 0 } } /* union for io-specific vars */
 };
@@ -361,6 +364,7 @@ static const rio rioFdIO = {
     0,              /* current checksum */
     0,              /* flags */
     0,              /* bytes read or written */
+    0,              /* keys loaded */
     0,              /* read/write chunk size */
     { { NULL, 0 } } /* union for io-specific vars */
 };

src/rio.h

@@ -62,6 +62,9 @@ struct _rio {
     /* The current checksum and flags (see RIO_FLAG_*) */
     uint64_t cksum, flags;
 
+    /* number of keys loaded in transaction */
+    size_t loaded_keys;
+
     /* number of bytes read or written */
     size_t processed_bytes;
 

src/server.h

@@ -1515,6 +1515,7 @@ struct redisServer {
     off_t loading_loaded_bytes;
     time_t loading_start_time;
     off_t loading_process_events_interval_bytes;
+    off_t loading_process_events_interval_keys;
 
     int active_expire_enabled;      /* Can be disabled for testing purposes. */