In Redis RDB check: initial POC.

So far we used an external program (later executed within Redis) and
parser in order to check RDB files for correctness. This forces, at each
RDB format update, to have two copies of the same format implementation
that are hard to keep in sync. Morover the former RDB checker only
checked the very high-level format of the file, without actually trying
to load things in memory. Certain corruptions can only be handled by
really loading key-value pairs.

This first commit attempts to unify the Redis RDB loadig code with the
task of checking the RDB file for correctness. More work is needed but
it looks like a sounding direction so far.

2 files changed, 189 insertions, 647 deletions

diff --git a/src/redis-check-rdb.c b/src/redis-check-rdb.c
index 7bb93b60b..c7886b0d4 100644
--- a/src/redis-check-rdb.c
+++ b/src/redis-check-rdb.c
@@ -1,6 +1,5 @@
 /*
- * Copyright (c) 2009-2012, Pieter Noordhuis <pcnoordhuis at gmail dot com>
- * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
+ * Copyright (c) 2016, Salvatore Sanfilippo <antirez at gmail dot com>
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -28,683 +27,226 @@
  * POSSIBILITY OF SUCH DAMAGE.
  */
 
-
 #include "server.h"
 #include "rdb.h"
-#include <stdlib.h>
-#include <stdio.h>
-#include <unistd.h>
-#include <fcntl.h>
-#include <sys/stat.h>
-#include <sys/mman.h>
-#include "lzf.h"
-#include "crc64.h"
 
-#define ERROR(...) { \
-    serverLog(LL_WARNING, __VA_ARGS__); \
-    exit(1); \
-}
+#include <stdarg.h>
+
+void rdbLoadProgressCallback(rio *r, const void *buf, size_t len);
+long long rdbLoadMillisecondTime(rio *rdb);
 
-/* data type to hold offset in file and size */
-typedef struct {
-    void *data;
-    size_t size;
-    size_t offset;
-} pos;
+struct {
+    rio *rio;
+    robj *key;                      /* Current key we are reading. */
+    unsigned long keys;             /* Number of keys processed. */
+    unsigned long expires;          /* Number of keys with an expire. */
+    unsigned long already_expired;  /* Number of keys already expired. */
+    int doing;                      /* The state while reading the RDB. */
+} rdbstate;
 
-static unsigned char level = 0;
-static pos positions[16];
+#define RDB_CHECK_DOING_START 0
+#define RDB_CHECK_DOING_READ_EXPIRE 1
+#define RDB_CHECK_DOING_READ_KEY 2
+#define RDB_CHECK_DOING_READ_VALUE 3
 
-#define CURR_OFFSET (positions[level].offset)
+/* Called on RDB errors. Provides details about the RDB and the offset
+ * we were when the error was detected. */
+void rdbCheckError(const char *fmt, ...) {
+    char msg[1024];
+    va_list ap;
 
-/* Hold a stack of errors */
-typedef struct {
-    char error[16][1024];
-    size_t offset[16];
-    size_t level;
-} errors_t;
-static errors_t errors;
+    va_start(ap, fmt);
+    vsnprintf(msg, sizeof(msg), fmt, ap);
+    va_end(ap);
 
-#define SHIFT_ERROR(provided_offset, ...) { \
-    sprintf(errors.error[errors.level], __VA_ARGS__); \
-    errors.offset[errors.level] = provided_offset; \
-    errors.level++; \
+    printf("*** RDB CHECK FAILED: %s ***\n", msg);
+    printf("AT RDB OFFSET: %llu\n",
+        (unsigned long long) (rdbstate.rio ?
+            rdbstate.rio->processed_bytes : 0));
+    if (rdbstate.key)
+        printf("READING KEY: %s\n", (char*)rdbstate.key->ptr);
 }
 
-/* Data type to hold opcode with optional key name an success status */
-typedef struct {
-    char* key;
-    int type;
-    char success;
-} entry;
+/* Print informations during RDB checking. */
+void rdbCheckInfo(const char *fmt, ...) {
+    char msg[1024];
+    va_list ap;
 
-#define MAX_TYPES_NUM 256
-#define MAX_TYPE_NAME_LEN 16
-/* store string types for output */
-static char types[MAX_TYPES_NUM][MAX_TYPE_NAME_LEN];
+    va_start(ap, fmt);
+    vsnprintf(msg, sizeof(msg), fmt, ap);
+    va_end(ap);
 
-/* Return true if 't' is a valid object type. */
-static int rdbCheckType(unsigned char t) {
-    /* In case a new object type is added, update the following
-     * condition as necessary. */
-    return
-        (t >= RDB_TYPE_HASH_ZIPMAP && t <= RDB_TYPE_HASH_ZIPLIST) ||
-        t <= RDB_TYPE_HASH ||
-        t >= RDB_OPCODE_EXPIRETIME_MS;
+    printf("[offset %llu] %s\n",
+        (unsigned long long) (rdbstate.rio ?
+            rdbstate.rio->processed_bytes : 0), msg);
 }
 
-/* when number of bytes to read is negative, do a peek */
-static int readBytes(void *target, long num) {
-    char peek = (num < 0) ? 1 : 0;
-    num = (num < 0) ? -num : num;
+/* During RDB check we setup a special signal handler for memory violations
+ * and similar conditions, so that we can log the offending part of the RDB
+ * if the crash is due to broken content. */
+void rdbCheckHandleCrash(int sig, siginfo_t *info, void *secret) {
+    UNUSED(sig);
+    UNUSED(info);
+    UNUSED(secret);
 
-    pos p = positions[level];
-    if (p.offset + num > p.size) {
-        return 0;
-    } else {
-        memcpy(target, (void*)((size_t)p.data + p.offset), num);
-        if (!peek) positions[level].offset += num;
-    }
-    return 1;
+    rdbCheckError("Server crash checking the specified RDB file!");
+    exit(1);
 }
 
-int processHeader(void) {
-    char buf[10] = "_________";
-    int dump_version;
-
-    if (!readBytes(buf, 9)) {
-        ERROR("Cannot read header");
-    }
-
-    /* expect the first 5 bytes to equal REDIS */
-    if (memcmp(buf,"REDIS",5) != 0) {
-        ERROR("Wrong signature in header");
-    }
+void rdbCheckSetupSignals(void) {
+    struct sigaction act;
 
-    dump_version = (int)strtol(buf + 5, NULL, 10);
-    if (dump_version < 1 || dump_version > 6) {
-        ERROR("Unknown RDB format version: %d", dump_version);
-    }
-    return dump_version;
+    sigemptyset(&act.sa_mask);
+    act.sa_flags = SA_NODEFER | SA_RESETHAND | SA_SIGINFO;
+#warning "Uncomment here"
+//    act.sa_sigaction = rdbCheckHandleCrash;
+    sigaction(SIGSEGV, &act, NULL);
+    sigaction(SIGBUS, &act, NULL);
+    sigaction(SIGFPE, &act, NULL);
+    sigaction(SIGILL, &act, NULL);
 }
 
-static int loadType(entry *e) {
-    uint32_t offset = CURR_OFFSET;
-
-    /* this byte needs to qualify as type */
-    unsigned char t;
-    if (readBytes(&t, 1)) {
-        if (rdbCheckType(t)) {
-            e->type = t;
-            return 1;
-        } else {
-            SHIFT_ERROR(offset, "Unknown type (0x%02x)", t);
-        }
-    } else {
-        SHIFT_ERROR(offset, "Could not read type");
+/* Check the specified RDB file. */
+int redis_check_rdb(char *rdbfilename) {
+    uint64_t dbid;
+    int type, rdbver;
+    char buf[1024];
+    long long expiretime, now = mstime();
+    FILE *fp;
+    rio rdb;
+
+    if ((fp = fopen(rdbfilename,"r")) == NULL) return C_ERR;
+
+    rioInitWithFile(&rdb,fp);
+    rdbstate.rio = &rdb;
+    rdb.update_cksum = rdbLoadProgressCallback;
+    if (rioRead(&rdb,buf,9) == 0) goto eoferr;
+    buf[9] = '\0';
+    if (memcmp(buf,"REDIS",5) != 0) {
+        rdbCheckError("Wrong signature trying to load DB from file");
+        return 1;
     }
-
-    /* failure */
-    return 0;
-}
-
-static int peekType() {
-    unsigned char t;
-    if (readBytes(&t, -1) && (rdbCheckType(t)))
-        return t;
-    return -1;
-}
-
-/* discard time, just consume the bytes */
-static int processTime(int type) {
-    uint32_t offset = CURR_OFFSET;
-    unsigned char t[8];
-    int timelen = (type == RDB_OPCODE_EXPIRETIME_MS) ? 8 : 4;
-
-    if (readBytes(t,timelen)) {
+    rdbver = atoi(buf+5);
+    if (rdbver < 1 || rdbver > RDB_VERSION) {
+        rdbCheckError("Can't handle RDB format version %d",rdbver);
         return 1;
-    } else {
-        SHIFT_ERROR(offset, "Could not read time");
     }
 
-    /* failure */
+    startLoading(fp);
+    while(1) {
+        robj *key, *val;
+        expiretime = -1;
+
+        /* Read type. */
+        if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
+
+        /* Handle special types. */
+        if (type == RDB_OPCODE_EXPIRETIME) {
+            /* EXPIRETIME: load an expire associated with the next key
+             * to load. Note that after loading an expire we need to
+             * load the actual type, and continue. */
+            if ((expiretime = rdbLoadTime(&rdb)) == -1) goto eoferr;
+            /* We read the time so we need to read the object type again. */
+            if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
+            /* the EXPIRETIME opcode specifies time in seconds, so convert
+             * into milliseconds. */
+            expiretime *= 1000;
+        } else if (type == RDB_OPCODE_EXPIRETIME_MS) {
+            /* EXPIRETIME_MS: milliseconds precision expire times introduced
+             * with RDB v3. Like EXPIRETIME but no with more precision. */
+            if ((expiretime = rdbLoadMillisecondTime(&rdb)) == -1) goto eoferr;
+            /* We read the time so we need to read the object type again. */
+            if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
+        } else if (type == RDB_OPCODE_EOF) {
+            /* EOF: End of file, exit the main loop. */
+            break;
+        } else if (type == RDB_OPCODE_SELECTDB) {
+            /* SELECTDB: Select the specified database. */
+            if ((dbid = rdbLoadLen(&rdb,NULL)) == RDB_LENERR)
+                goto eoferr;
+            rdbCheckInfo("Selecting DB ID %d", dbid);
+            continue; /* Read type again. */
+        } else if (type == RDB_OPCODE_RESIZEDB) {
+            /* RESIZEDB: Hint about the size of the keys in the currently
+             * selected data base, in order to avoid useless rehashing. */
+            uint64_t db_size, expires_size;
+            if ((db_size = rdbLoadLen(&rdb,NULL)) == RDB_LENERR)
+                goto eoferr;
+            if ((expires_size = rdbLoadLen(&rdb,NULL)) == RDB_LENERR)
+                goto eoferr;
+            continue; /* Read type again. */
+        } else if (type == RDB_OPCODE_AUX) {
+            /* AUX: generic string-string fields. Use to add state to RDB
+             * which is backward compatible. Implementations of RDB loading
+             * are requierd to skip AUX fields they don't understand.
+             *
+             * An AUX field is composed of two strings: key and value. */
+            robj *auxkey, *auxval;
+            if ((auxkey = rdbLoadStringObject(&rdb)) == NULL) goto eoferr;
+            if ((auxval = rdbLoadStringObject(&rdb)) == NULL) goto eoferr;
+
+            rdbCheckInfo("%s = '%s'", (char*)auxkey->ptr, (char*)auxval->ptr);
+            decrRefCount(auxkey);
+            decrRefCount(auxval);
+            continue; /* Read type again. */
+        }
+
+        /* Read key */
+        if ((key = rdbLoadStringObject(&rdb)) == NULL) goto eoferr;
+        rdbstate.key = key;
+        rdbstate.keys++;
+        /* Read value */
+        if ((val = rdbLoadObject(type,&rdb)) == NULL) goto eoferr;
+        /* Check if the key already expired. This function is used when loading
+         * an RDB file from disk, either at startup, or when an RDB was
+         * received from the master. In the latter case, the master is
+         * responsible for key expiry. If we would expire keys here, the
+         * snapshot taken by the master may not be reflected on the slave. */
+        if (server.masterhost == NULL && expiretime != -1 && expiretime < now)
+            rdbstate.already_expired++;
+        if (expiretime != -1) rdbstate.expires++;
+        rdbstate.key = NULL;
+        decrRefCount(key);
+        decrRefCount(val);
+    }
+    /* Verify the checksum if RDB version is >= 5 */
+    if (rdbver >= 5 && server.rdb_checksum) {
+        uint64_t cksum, expected = rdb.cksum;
+
+        if (rioRead(&rdb,&cksum,8) == 0) goto eoferr;
+        memrev64ifbe(&cksum);
+        if (cksum == 0) {
+            rdbCheckInfo("RDB file was saved with checksum disabled: no check performed.");
+        } else if (cksum != expected) {
+            rdbCheckError("RDB CRC error");
+        }
+    }
+
+    fclose(fp);
     return 0;
-}
-
-static uint32_t loadLength(int *isencoded) {
-    unsigned char buf[2];
-    uint32_t len;
-    int type;
-
-    if (isencoded) *isencoded = 0;
-    if (!readBytes(buf, 1)) return RDB_LENERR;
-    type = (buf[0] & 0xC0) >> 6;
-    if (type == RDB_6BITLEN) {
-        /* Read a 6 bit len */
-        return buf[0] & 0x3F;
-    } else if (type == RDB_ENCVAL) {
-        /* Read a 6 bit len encoding type */
-        if (isencoded) *isencoded = 1;
-        return buf[0] & 0x3F;
-    } else if (type == RDB_14BITLEN) {
-        /* Read a 14 bit len */
-        if (!readBytes(buf+1,1)) return RDB_LENERR;
-        return ((buf[0] & 0x3F) << 8) | buf[1];
-    } else {
-        /* Read a 32 bit len */
-        if (!readBytes(&len, 4)) return RDB_LENERR;
-        return (unsigned int)ntohl(len);
-    }
-}
-
-static char *loadIntegerObject(int enctype) {
-    uint32_t offset = CURR_OFFSET;
-    unsigned char enc[4];
-    long long val;
-
-    if (enctype == RDB_ENC_INT8) {
-        uint8_t v;
-        if (!readBytes(enc, 1)) return NULL;
-        v = enc[0];
-        val = (int8_t)v;
-    } else if (enctype == RDB_ENC_INT16) {
-        uint16_t v;
-        if (!readBytes(enc, 2)) return NULL;
-        v = enc[0]|(enc[1]<<8);
-        val = (int16_t)v;
-    } else if (enctype == RDB_ENC_INT32) {
-        uint32_t v;
-        if (!readBytes(enc, 4)) return NULL;
-        v = enc[0]|(enc[1]<<8)|(enc[2]<<16)|(enc[3]<<24);
-        val = (int32_t)v;
-    } else {
-        SHIFT_ERROR(offset, "Unknown integer encoding (0x%02x)", enctype);
-        return NULL;
-    }
-
-    /* convert val into string */
-    char *buf;
-    buf = zmalloc(sizeof(char) * 128);
-    sprintf(buf, "%lld", val);
-    return buf;
-}
-
-static char* loadLzfStringObject() {
-    unsigned int slen, clen;
-    char *c, *s;
-
-    if ((clen = loadLength(NULL)) == RDB_LENERR) return NULL;
-    if ((slen = loadLength(NULL)) == RDB_LENERR) return NULL;
-
-    c = zmalloc(clen);
-    if (!readBytes(c, clen)) {
-        zfree(c);
-        return NULL;
-    }
-
-    s = zmalloc(slen+1);
-    if (lzf_decompress(c,clen,s,slen) == 0) {
-        zfree(c); zfree(s);
-        return NULL;
-    }
 
-    zfree(c);
-    return s;
-}
-
-/* returns NULL when not processable, char* when valid */
-static char* loadStringObject() {
-    uint32_t offset = CURR_OFFSET;
-    int isencoded;
-    uint32_t len;
-
-    len = loadLength(&isencoded);
-    if (isencoded) {
-        switch(len) {
-        case RDB_ENC_INT8:
-        case RDB_ENC_INT16:
-        case RDB_ENC_INT32:
-            return loadIntegerObject(len);
-        case RDB_ENC_LZF:
-            return loadLzfStringObject();
-        default:
-            /* unknown encoding */
-            SHIFT_ERROR(offset, "Unknown string encoding (0x%02x)", len);
-            return NULL;
-        }
-    }
-
-    if (len == RDB_LENERR) return NULL;
-
-    char *buf = zmalloc(sizeof(char) * (len+1));
-    if (buf == NULL) return NULL;
-    buf[len] = '\0';
-    if (!readBytes(buf, len)) {
-        zfree(buf);
-        return NULL;
-    }
-    return buf;
-}
-
-static int processStringObject(char** store) {
-    unsigned long offset = CURR_OFFSET;
-    char *key = loadStringObject();
-    if (key == NULL) {
-        SHIFT_ERROR(offset, "Error reading string object");
-        zfree(key);
-        return 0;
-    }
-
-    if (store != NULL) {
-        *store = key;
-    } else {
-        zfree(key);
-    }
-    return 1;
-}
-
-static double* loadDoubleValue() {
-    char buf[256];
-    unsigned char len;
-    double* val;
-
-    if (!readBytes(&len,1)) return NULL;
-
-    val = zmalloc(sizeof(double));
-    switch(len) {
-    case 255: *val = R_NegInf;  return val;
-    case 254: *val = R_PosInf;  return val;
-    case 253: *val = R_Nan;     return val;
-    default:
-        if (!readBytes(buf, len)) {
-            zfree(val);
-            return NULL;
-        }
-        buf[len] = '\0';
-        sscanf(buf, "%lg", val);
-        return val;
-    }
-}
-
-static int processDoubleValue(double** store) {
-    unsigned long offset = CURR_OFFSET;
-    double *val = loadDoubleValue();
-    if (val == NULL) {
-        SHIFT_ERROR(offset, "Error reading double value");
-        zfree(val);
-        return 0;
-    }
-
-    if (store != NULL) {
-        *store = val;
-    } else {
-        zfree(val);
-    }
-    return 1;
-}
-
-static int loadPair(entry *e) {
-    uint32_t offset = CURR_OFFSET;
-    uint32_t i;
-
-    /* read key first */
-    char *key;
-    if (processStringObject(&key)) {
-        e->key = key;
-    } else {
-        SHIFT_ERROR(offset, "Error reading entry key");
-        return 0;
-    }
-
-    uint32_t length = 0;
-    if (e->type == RDB_TYPE_LIST ||
-        e->type == RDB_TYPE_SET  ||
-        e->type == RDB_TYPE_ZSET ||
-        e->type == RDB_TYPE_HASH) {
-        if ((length = loadLength(NULL)) == RDB_LENERR) {
-            SHIFT_ERROR(offset, "Error reading %s length", types[e->type]);
-            return 0;
-        }
-    }
-
-    switch(e->type) {
-    case RDB_TYPE_STRING:
-    case RDB_TYPE_HASH_ZIPMAP:
-    case RDB_TYPE_LIST_ZIPLIST:
-    case RDB_TYPE_SET_INTSET:
-    case RDB_TYPE_ZSET_ZIPLIST:
-    case RDB_TYPE_HASH_ZIPLIST:
-        if (!processStringObject(NULL)) {
-            SHIFT_ERROR(offset, "Error reading entry value");
-            return 0;
-        }
-    break;
-    case RDB_TYPE_LIST:
-    case RDB_TYPE_SET:
-        for (i = 0; i < length; i++) {
-            offset = CURR_OFFSET;
-            if (!processStringObject(NULL)) {
-                SHIFT_ERROR(offset, "Error reading element at index %d (length: %d)", i, length);
-                return 0;
-            }
-        }
-    break;
-    case RDB_TYPE_ZSET:
-        for (i = 0; i < length; i++) {
-            offset = CURR_OFFSET;
-            if (!processStringObject(NULL)) {
-                SHIFT_ERROR(offset, "Error reading element key at index %d (length: %d)", i, length);
-                return 0;
-            }
-            offset = CURR_OFFSET;
-            if (!processDoubleValue(NULL)) {
-                SHIFT_ERROR(offset, "Error reading element value at index %d (length: %d)", i, length);
-                return 0;
-            }
-        }
-    break;
-    case RDB_TYPE_HASH:
-        for (i = 0; i < length; i++) {
-            offset = CURR_OFFSET;
-            if (!processStringObject(NULL)) {
-                SHIFT_ERROR(offset, "Error reading element key at index %d (length: %d)", i, length);
-                return 0;
-            }
-            offset = CURR_OFFSET;
-            if (!processStringObject(NULL)) {
-                SHIFT_ERROR(offset, "Error reading element value at index %d (length: %d)", i, length);
-                return 0;
-            }
-        }
-    break;
-    default:
-        SHIFT_ERROR(offset, "Type not implemented");
-        return 0;
-    }
-    /* because we're done, we assume success */
-    e->success = 1;
+eoferr: /* unexpected end of file is handled here with a fatal exit */
+    rdbCheckError("Unexpected EOF reading RDB file");
     return 1;
 }
 
-static entry loadEntry() {
-    entry e = { NULL, -1, 0 };
-    uint32_t length, offset[4];
-
-    /* reset error container */
-    errors.level = 0;
-
-    offset[0] = CURR_OFFSET;
-    if (!loadType(&e)) {
-        return e;
-    }
-
-    offset[1] = CURR_OFFSET;
-    if (e.type == RDB_OPCODE_SELECTDB) {
-        if ((length = loadLength(NULL)) == RDB_LENERR) {
-            SHIFT_ERROR(offset[1], "Error reading database number");
-            return e;
-        }
-        if (length > 63) {
-            SHIFT_ERROR(offset[1], "Database number out of range (%d)", length);
-            return e;
-        }
-    } else if (e.type == RDB_OPCODE_EOF) {
-        if (positions[level].offset < positions[level].size) {
-            SHIFT_ERROR(offset[0], "Unexpected EOF");
-        } else {
-            e.success = 1;
-        }
-        return e;
-    } else {
-        /* optionally consume expire */
-        if (e.type == RDB_OPCODE_EXPIRETIME ||
-            e.type == RDB_OPCODE_EXPIRETIME_MS) {
-            if (!processTime(e.type)) return e;
-            if (!loadType(&e)) return e;
-        }
-
-        offset[1] = CURR_OFFSET;
-        if (!loadPair(&e)) {
-            SHIFT_ERROR(offset[1], "Error for type %s", types[e.type]);
-            return e;
-        }
-    }
-
-    /* all entries are followed by a valid type:
-     * e.g. a new entry, SELECTDB, EXPIRE, EOF */
-    offset[2] = CURR_OFFSET;
-    if (peekType() == -1) {
-        SHIFT_ERROR(offset[2], "Followed by invalid type");
-        SHIFT_ERROR(offset[0], "Error for type %s", types[e.type]);
-        e.success = 0;
-    } else {
-        e.success = 1;
-    }
-
-    return e;
-}
-
-static void printCentered(int indent, int width, char* body) {
-    char head[256], tail[256];
-    memset(head, '\0', 256);
-    memset(tail, '\0', 256);
-
-    memset(head, '=', indent);
-    memset(tail, '=', width - 2 - indent - strlen(body));
-    serverLog(LL_WARNING, "%s %s %s", head, body, tail);
-}
-
-static void printValid(uint64_t ops, uint64_t bytes) {
-    char body[80];
-    sprintf(body, "Processed %llu valid opcodes (in %llu bytes)",
-        (unsigned long long) ops, (unsigned long long) bytes);
-    printCentered(4, 80, body);
-}
-
-static void printSkipped(uint64_t bytes, uint64_t offset) {
-    char body[80];
-    sprintf(body, "Skipped %llu bytes (resuming at 0x%08llx)",
-        (unsigned long long) bytes, (unsigned long long) offset);
-    printCentered(4, 80, body);
-}
-
-static void printErrorStack(entry *e) {
-    unsigned int i;
-    char body[64];
-
-    if (e->type == -1) {
-        sprintf(body, "Error trace");
-    } else if (e->type >= 253) {
-        sprintf(body, "Error trace (%s)", types[e->type]);
-    } else if (!e->key) {
-        sprintf(body, "Error trace (%s: (unknown))", types[e->type]);
-    } else {
-        char tmp[41];
-        strncpy(tmp, e->key, 40);
-
-        /* display truncation at the last 3 chars */
-        if (strlen(e->key) > 40) {
-            memset(&tmp[37], '.', 3);
-        }
-
-        /* display unprintable characters as ? */
-        for (i = 0; i < strlen(tmp); i++) {
-            if (tmp[i] <= 32) tmp[i] = '?';
-        }
-        sprintf(body, "Error trace (%s: %s)", types[e->type], tmp);
-    }
-
-    printCentered(4, 80, body);
-
-    /* display error stack */
-    for (i = 0; i < errors.level; i++) {
-        serverLog(LL_WARNING, "0x%08lx - %s",
-            (unsigned long) errors.offset[i], errors.error[i]);
-    }
-}
-
-void process(void) {
-    uint64_t num_errors = 0, num_valid_ops = 0, num_valid_bytes = 0;
-    entry entry = { NULL, -1, 0 };
-    int dump_version = processHeader();
-
-    /* Exclude the final checksum for RDB >= 5. Will be checked at the end. */
-    if (dump_version >= 5) {
-        if (positions[0].size < 8) {
-            serverLog(LL_WARNING, "RDB version >= 5 but no room for checksum.");
-            exit(1);
-        }
-        positions[0].size -= 8;
-    }
-
-    level = 1;
-    while(positions[0].offset < positions[0].size) {
-        positions[1] = positions[0];
-
-        entry = loadEntry();
-        if (!entry.success) {
-            printValid(num_valid_ops, num_valid_bytes);
-            printErrorStack(&entry);
-            num_errors++;
-            num_valid_ops = 0;
-            num_valid_bytes = 0;
-
-            /* search for next valid entry */
-            uint64_t offset = positions[0].offset + 1;
-            int i = 0;
-
-            while (!entry.success && offset < positions[0].size) {
-                positions[1].offset = offset;
-
-                /* find 3 consecutive valid entries */
-                for (i = 0; i < 3; i++) {
-                    entry = loadEntry();
-                    if (!entry.success) break;
-                }
-                /* check if we found 3 consecutive valid entries */
-                if (i < 3) {
-                    offset++;
-                }
-            }
-
-            /* print how many bytes we have skipped to find a new valid opcode */
-            if (offset < positions[0].size) {
-                printSkipped(offset - positions[0].offset, offset);
-            }
-
-            positions[0].offset = offset;
-        } else {
-            num_valid_ops++;
-            num_valid_bytes += positions[1].offset - positions[0].offset;
-
-            /* advance position */
-            positions[0] = positions[1];
-        }
-        zfree(entry.key);
-    }
-
-    /* because there is another potential error,
-     * print how many valid ops we have processed */
-    printValid(num_valid_ops, num_valid_bytes);
-
-    /* expect an eof */
-    if (entry.type != RDB_OPCODE_EOF) {
-        /* last byte should be EOF, add error */
-        errors.level = 0;
-        SHIFT_ERROR(positions[0].offset, "Expected EOF, got %s", types[entry.type]);
-
-        /* this is an EOF error so reset type */
-        entry.type = -1;
-        printErrorStack(&entry);
-
-        num_errors++;
-    }
-
-    /* Verify checksum */
-    if (dump_version >= 5) {
-        uint64_t crc = crc64(0,positions[0].data,positions[0].size);
-        uint64_t crc2;
-        unsigned char *p = (unsigned char*)positions[0].data+positions[0].size;
-        crc2 = ((uint64_t)p[0] << 0) |
-               ((uint64_t)p[1] << 8) |
-               ((uint64_t)p[2] << 16) |
-               ((uint64_t)p[3] << 24) |
-               ((uint64_t)p[4] << 32) |
-               ((uint64_t)p[5] << 40) |
-               ((uint64_t)p[6] << 48) |
-               ((uint64_t)p[7] << 56);
-        if (crc != crc2) {
-            SHIFT_ERROR(positions[0].offset, "RDB CRC64 does not match.");
-        } else {
-            serverLog(LL_WARNING, "CRC64 checksum is OK");
-        }
-    }
-
-    /* print summary on errors */
-    if (num_errors) {
-        serverLog(LL_WARNING, "Total unprocessable opcodes: %llu",
-            (unsigned long long) num_errors);
-    }
-}
-
-int redis_check_rdb(char *rdbfilename) {
-    int fd;
-    off_t size;
-    struct stat stat;
-    void *data;
-
-    fd = open(rdbfilename, O_RDONLY);
-    if (fd < 1) {
-        ERROR("Cannot open file: %s", rdbfilename);
-    }
-    if (fstat(fd, &stat) == -1) {
-        ERROR("Cannot stat: %s", rdbfilename);
-    } else {
-        size = stat.st_size;
-    }
-
-    if (sizeof(size_t) == sizeof(int32_t) && size >= INT_MAX) {
-        ERROR("Cannot check dump files >2GB on a 32-bit platform");
-    }
-
-    data = mmap(NULL, size, PROT_READ, MAP_SHARED, fd, 0);
-    if (data == MAP_FAILED) {
-        ERROR("Cannot mmap: %s", rdbfilename);
-    }
-
-    /* Initialize static vars */
-    positions[0].data = data;
-    positions[0].size = size;
-    positions[0].offset = 0;
-    errors.level = 0;
-
-    /* Object types */
-    sprintf(types[RDB_TYPE_STRING], "STRING");
-    sprintf(types[RDB_TYPE_LIST], "LIST");
-    sprintf(types[RDB_TYPE_SET], "SET");
-    sprintf(types[RDB_TYPE_ZSET], "ZSET");
-    sprintf(types[RDB_TYPE_HASH], "HASH");
-
-    /* Object types only used for dumping to disk */
-    sprintf(types[RDB_OPCODE_EXPIRETIME], "EXPIRETIME");
-    sprintf(types[RDB_OPCODE_SELECTDB], "SELECTDB");
-    sprintf(types[RDB_OPCODE_EOF], "EOF");
-
-    process();
-
-    munmap(data, size);
-    close(fd);
-    return 0;
-}
-
 /* RDB check main: called form redis.c when Redis is executed with the
- * redis-check-rdb alias. */
+ * redis-check-rdb alias.
+ *
+ * The function never returns, but exits with the status code according
+ * to success (RDB is sane) or error (RDB is corrupted). */
 int redis_check_rdb_main(char **argv, int argc) {
     if (argc != 2) {
         fprintf(stderr, "Usage: %s <rdb-file-name>\n", argv[0]);
         exit(1);
     }
-    serverLog(LL_WARNING, "Checking RDB file %s", argv[1]);
-    exit(redis_check_rdb(argv[1]));
-    return 0;
+    createSharedObjects(); /* Needed for loading. */
+    server.loading_process_events_interval_bytes = 0;
+    rdbCheckInfo("Checking RDB file %s", argv[1]);
+    rdbCheckSetupSignals();
+    int retval = redis_check_rdb(argv[1]);
+    if (retval == 0) {
+        rdbCheckInfo("\\o/ RDB looks OK! \\o/");
+    }
+    exit(retval);
 }
diff --git a/src/server.c b/src/server.c
index 8ef6620b0..4199b6776 100644
--- a/src/server.c
+++ b/src/server.c
@@ -3968,7 +3968,7 @@ int main(int argc, char **argv) {
      * the program main. However the program is part of the Redis executable
      * so that we can easily execute an RDB check on loading errors. */
     if (strstr(argv[0],"redis-check-rdb") != NULL)
-        exit(redis_check_rdb_main(argv,argc));
+        redis_check_rdb_main(argv,argc);
 
     if (argc >= 2) {
         j = 1; /* First option to parse in argv[] */