redis源码分析——8、RDB持久化

save <seconds> <changes> 

save 900 1   # 900s内至少改变了一个key 
save 300 10     # 300s内至少改变了10个key 
save 60 10000   # 60s内至少改变了10000个key 
save 5 0    # 每5s保存一次 

stop-writes-on-bgsave-error yes 

rdbcompression yes 

rdbchecksum yes 

dbfilename dump.rdb 

rdb-del-sync-files no 

dir ./ 

struct saveparam { 
    time_t seconds; 
    int changes; 
}; 

struct redisServer { 
    ... 
    /* RDB persistence */ 
    long long dirty;                /* Changes to DB from the last save */ 
    long long dirty_before_bgsave;  /* Used to restore dirty on failed BGSAVE */ 
    pid_t rdb_child_pid;            /* PID of RDB saving child */ 
    struct saveparam *saveparams;   /* Save points array for RDB */ 
    int saveparamslen;              /* Number of saving points */ 
    char *rdb_filename;             /* Name of RDB file */ 
    int rdb_compression;            /* Use compression in RDB? */ 
    int rdb_checksum;               /* Use RDB checksum? */ 
    int rdb_del_sync_files;         /* Remove RDB files used only for SYNC if 
                                       the instance does not use persistence. */ 
    time_t lastsave;                /* Unix time of last successful save */ 
    time_t lastbgsave_try;          /* Unix time of last attempted bgsave */ 
    time_t rdb_save_time_last;      /* Time used by last RDB save run. */ 
    time_t rdb_save_time_start;     /* Current RDB save start time. */ 
    int rdb_bgsave_scheduled;       /* BGSAVE when possible if true. */ 
    int rdb_child_type;             /* Type of save by active child. */ 
    int lastbgsave_status;          /* C_OK or C_ERR */ 
    int stop_writes_on_bgsave_err;  /* Don't allow writes if can't BGSAVE */ 
    ... 
} 

int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { 
    ... 
    /* Check if a background saving or AOF rewrite in progress terminated. */ 
    if (hasActiveChildProcess() || ldbPendingChildren()) 
    { 
        checkChildrenDone(); 
    } else { 
        /* If there is not a background saving/rewrite in progress check if 
         * we have to save/rewrite now. */ 
        for (j = 0; j < server.saveparamslen; j++) { 
            struct saveparam *sp = server.saveparams+j; 

            /* Save if we reached the given amount of changes, 
             * the given amount of seconds, and if the latest bgsave was 
             * successful or if, in case of an error, at least 
             * CONFIG_BGSAVE_RETRY_DELAY seconds already elapsed. */ 
            if (server.dirty >= sp->changes && 
                server.unixtime-server.lastsave > sp->seconds && 
                (server.unixtime-server.lastbgsave_try > 
                 CONFIG_BGSAVE_RETRY_DELAY || 
                 server.lastbgsave_status == C_OK)) 
            { 
                serverLog(LL_NOTICE,"%d changes in %d seconds. Saving...", 
                    sp->changes, (int)sp->seconds); 
                rdbSaveInfo rsi, *rsiptr; 
                rsiptr = rdbPopulateSaveInfo(&rsi); 
                rdbSaveBackground(server.rdb_filename,rsiptr); 
                break; 
            } 
        } 
        /* Trigger an AOF rewrite if needed. */ 
        if (server.aof_state == AOF_ON && 
            !hasActiveChildProcess() && 
            server.aof_rewrite_perc && 
            server.aof_current_size > server.aof_rewrite_min_size) 
        { 
            long long base = server.aof_rewrite_base_size ? 
                server.aof_rewrite_base_size : 1; 
            long long growth = (server.aof_current_size*100/base) - 100; 
            if (growth >= server.aof_rewrite_perc) { 
                serverLog(LL_NOTICE,"Starting automatic rewriting of AOF on %lld%% growth",growth); 
                rewriteAppendOnlyFileBackground(); 
            } 
        } 
    ... 
} 

int rdbSaveRio(rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi) { 
    dictIterator *di = NULL; 
    dictEntry *de; 
    char magic[10]; 
    int j; 
    uint64_t cksum; 
    size_t processed = 0; 

    if (server.rdb_checksum) 
        rdb->update_cksum = rioGenericUpdateChecksum; 
    snprintf(magic,sizeof(magic),"REDIS%04d",RDB_VERSION); 
    if (rdbWriteRaw(rdb,magic,9) == -1) goto werr; 
    if (rdbSaveInfoAuxFields(rdb,rdbflags,rsi) == -1) goto werr; 
    if (rdbSaveModulesAux(rdb, REDISMODULE_AUX_BEFORE_RDB) == -1) goto werr; 

    for (j = 0; j < server.dbnum; j++) { 
        redisDb *db = server.db+j; 
        dict *d = db->dict; 
        if (dictSize(d) == 0) continue; 
        di = dictGetSafeIterator(d); 

        /* Write the SELECT DB opcode */ 
        if (rdbSaveType(rdb,RDB_OPCODE_SELECTDB) == -1) goto werr; 
        if (rdbSaveLen(rdb,j) == -1) goto werr; 

        /* Write the RESIZE DB opcode. */ 
        uint64_t db_size, expires_size; 
        db_size = dictSize(db->dict); 
        expires_size = dictSize(db->expires); 
        if (rdbSaveType(rdb,RDB_OPCODE_RESIZEDB) == -1) goto werr; 
        if (rdbSaveLen(rdb,db_size) == -1) goto werr; 
        if (rdbSaveLen(rdb,expires_size) == -1) goto werr; 

        /* Iterate this DB writing every entry */ 
        while((de = dictNext(di)) != NULL) { 
            sds keystr = dictGetKey(de); 
            robj key, *o = dictGetVal(de); 
            long long expire; 

            initStaticStringObject(key,keystr); 
            expire = getExpire(db,&key); 
            if (rdbSaveKeyValuePair(rdb,&key,o,expire) == -1) goto werr; 

            /* When this RDB is produced as part of an AOF rewrite, move 
             * accumulated diff from parent to child while rewriting in 
             * order to have a smaller final write. */ 
            if (rdbflags & RDBFLAGS_AOF_PREAMBLE && 
                rdb->processed_bytes > processed+AOF_READ_DIFF_INTERVAL_BYTES) 
            { 
                processed = rdb->processed_bytes; 
                aofReadDiffFromParent(); 
            } 
        } 
        dictReleaseIterator(di); 
        di = NULL; /* So that we don't release it again on error. */ 
    } 

    /* If we are storing the replication information on disk, persist 
     * the script cache as well: on successful PSYNC after a restart, we need 
     * to be able to process any EVALSHA inside the replication backlog the 
     * master will send us. */ 
    if (rsi && dictSize(server.lua_scripts)) { 
        di = dictGetIterator(server.lua_scripts); 
        while((de = dictNext(di)) != NULL) { 
            robj *body = dictGetVal(de); 
            if (rdbSaveAuxField(rdb,"lua",3,body->ptr,sdslen(body->ptr)) == -1) 
                goto werr; 
        } 
        dictReleaseIterator(di); 
        di = NULL; /* So that we don't release it again on error. */ 
    } 

    if (rdbSaveModulesAux(rdb, REDISMODULE_AUX_AFTER_RDB) == -1) goto werr; 

    /* EOF opcode */ 
    if (rdbSaveType(rdb,RDB_OPCODE_EOF) == -1) goto werr; 

    /* CRC64 checksum. It will be zero if checksum computation is disabled, the 
     * loading code skips the check in this case. */ 
    cksum = rdb->cksum; 
    memrev64ifbe(&cksum); 
    if (rioWrite(rdb,&cksum,8) == 0) goto werr; 
    return C_OK; 

werr: 
    if (error) *error = errno; 
    if (di) dictReleaseIterator(di); 
    return C_ERR; 
} 

#define RDB_OPCODE_MODULE_AUX 247   /* Module auxiliary data. */ 
#define RDB_OPCODE_IDLE       248   /* LRU idle time. */ 
#define RDB_OPCODE_FREQ       249   /* LFU frequency. */ 
#define RDB_OPCODE_AUX        250   /* RDB aux field. */ 
#define RDB_OPCODE_RESIZEDB   251   /* Hash table resize hint. */ 
#define RDB_OPCODE_EXPIRETIME_MS 252    /* Expire time in milliseconds. */ 
#define RDB_OPCODE_EXPIRETIME 253       /* Old expire time in seconds. */ 
#define RDB_OPCODE_SELECTDB   254   /* DB number of the following keys. */ 
#define RDB_OPCODE_EOF        255   /* End of the RDB file. */ 

/* Save a key-value pair, with expire time, type, key, value. 
 * On error -1 is returned. 
 * On success if the key was actually saved 1 is returned, otherwise 0 
 * is returned (the key was already expired). */ 
int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime) { 
    int savelru = server.maxmemory_policy & MAXMEMORY_FLAG_LRU; 
    int savelfu = server.maxmemory_policy & MAXMEMORY_FLAG_LFU; 

    /* Save the expire time */ 
    if (expiretime != -1) { 
        if (rdbSaveType(rdb,RDB_OPCODE_EXPIRETIME_MS) == -1) return -1; 
        if (rdbSaveMillisecondTime(rdb,expiretime) == -1) return -1; 
    } 

    /* Save the LRU info. */ 
    if (savelru) { 
        uint64_t idletime = estimateObjectIdleTime(val); 
        idletime /= 1000; /* Using seconds is enough and requires less space.*/ 
        if (rdbSaveType(rdb,RDB_OPCODE_IDLE) == -1) return -1; 
        if (rdbSaveLen(rdb,idletime) == -1) return -1; 
    } 

    /* Save the LFU info. */ 
    if (savelfu) { 
        uint8_t buf[1]; 
        buf[0] = LFUDecrAndReturn(val); 
        /* We can encode this in exactly two bytes: the opcode and an 8 
         * bit counter, since the frequency is logarithmic with a 0-255 range. 
         * Note that we do not store the halving time because to reset it 
         * a single time when loading does not affect the frequency much. */ 
        if (rdbSaveType(rdb,RDB_OPCODE_FREQ) == -1) return -1; 
        if (rdbWriteRaw(rdb,buf,1) == -1) return -1; 
    } 

    /* Save type, key, value */ 
    if (rdbSaveObjectType(rdb,val) == -1) return -1; /* 保存类型 */ 
    if (rdbSaveStringObject(rdb,key) == -1) return -1; /* 保存key */ 
    if (rdbSaveObject(rdb,val,key) == -1) return -1; /* 保存value */ 

    /* Delay return if required (for testing) */ 
    if (server.rdb_key_save_delay) 
        usleep(server.rdb_key_save_delay); 

    return 1; 
} 

int rdbSaveObjectType(rio *rdb, robj *o) { 
    switch (o->type) { 
    case OBJ_STRING: 
        return rdbSaveType(rdb,RDB_TYPE_STRING); 
    case OBJ_LIST: 
        if (o->encoding == OBJ_ENCODING_QUICKLIST) 
            return rdbSaveType(rdb,RDB_TYPE_LIST_QUICKLIST); 
        else 
            serverPanic("Unknown list encoding"); 
    case OBJ_SET: 
        if (o->encoding == OBJ_ENCODING_INTSET) 
            return rdbSaveType(rdb,RDB_TYPE_SET_INTSET); 
        else if (o->encoding == OBJ_ENCODING_HT) 
            return rdbSaveType(rdb,RDB_TYPE_SET); 
        else 
            serverPanic("Unknown set encoding"); 
    case OBJ_ZSET: 
        if (o->encoding == OBJ_ENCODING_ZIPLIST) 
            return rdbSaveType(rdb,RDB_TYPE_ZSET_ZIPLIST); 
        else if (o->encoding == OBJ_ENCODING_SKIPLIST) 
            return rdbSaveType(rdb,RDB_TYPE_ZSET_2); 
        else 
            serverPanic("Unknown sorted set encoding"); 
    case OBJ_HASH: 
        if (o->encoding == OBJ_ENCODING_ZIPLIST) 
            return rdbSaveType(rdb,RDB_TYPE_HASH_ZIPLIST); 
        else if (o->encoding == OBJ_ENCODING_HT) 
            return rdbSaveType(rdb,RDB_TYPE_HASH); 
        else 
            serverPanic("Unknown hash encoding"); 
    case OBJ_STREAM: 
        return rdbSaveType(rdb,RDB_TYPE_STREAM_LISTPACKS); 
    case OBJ_MODULE: 
        return rdbSaveType(rdb,RDB_TYPE_MODULE_2); 
    default: 
        serverPanic("Unknown object type"); 
    } 
    return -1; /* avoid warning */ 
} 

/* Saves an encoded length. The first two bits in the first byte are used to 
 * hold the encoding type. See the RDB_* definitions for more information 
 * on the types of encoding. */ 
int rdbSaveLen(rio *rdb, uint64_t len) { 
    unsigned char buf[2]; 
    size_t nwritten; 

    if (len < (1<<6)) {  /* 长度<64, 用后6位表示 */ 
        /* Save a 6 bit len */ 
        buf[0] = (len&0xFF)|(RDB_6BITLEN<<6); 
        if (rdbWriteRaw(rdb,buf,1) == -1) return -1; 
        nwritten = 1; 
    } else if (len < (1<<14)) { /* 长度<16384, 当前字节的6位+后一字节，共14位表示*/ 
        /* Save a 14 bit len */ 
        buf[0] = ((len>>8)&0xFF)|(RDB_14BITLEN<<6); 
        buf[1] = len&0xFF; 
        if (rdbWriteRaw(rdb,buf,2) == -1) return -1; 
        nwritten = 2; 
    } else if (len <= UINT32_MAX) { /* 跳过当前字节，后4字节表示*/ 
        /* Save a 32 bit len */ 
        buf[0] = RDB_32BITLEN; 
        if (rdbWriteRaw(rdb,buf,1) == -1) return -1; 
        uint32_t len32 = htonl(len); 
        if (rdbWriteRaw(rdb,&len32,4) == -1) return -1; 
        nwritten = 1+4; 
    } else { 
        /* Save a 64 bit len */ 
        buf[0] = RDB_64BITLEN; 
        if (rdbWriteRaw(rdb,buf,1) == -1) return -1; 
        len = htonu64(len); 
        if (rdbWriteRaw(rdb,&len,8) == -1) return -1; 
        nwritten = 1+8; 
    } 
    return nwritten; 
} 

ssize_t rdbSaveStringObject(rio *rdb, robj *obj) { 
    /* Avoid to decode the object, then encode it again, if the 
     * object is already integer encoded. */ 
    if (obj->encoding == OBJ_ENCODING_INT) { 
        return rdbSaveLongLongAsStringObject(rdb,(long)obj->ptr); 
    } else { 
        serverAssertWithInfo(NULL,obj,sdsEncodedObject(obj)); 
        return rdbSaveRawString(rdb,obj->ptr,sdslen(obj->ptr)); 
    } 
} 

ssize_t rdbSaveRawString(rio *rdb, unsigned char *s, size_t len) { 
    int enclen; 
    ssize_t n, nwritten = 0; 

    /* Try integer encoding */ 
    if (len <= 11) { 
        unsigned char buf[5]; 
        if ((enclen = rdbTryIntegerEncoding((char*)s,len,buf)) > 0) { 
            if (rdbWriteRaw(rdb,buf,enclen) == -1) return -1; 
            return enclen; 
        } 
    } 

    /* Try LZF compression - under 20 bytes it's unable to compress even 
     * aaaaaaaaaaaaaaaaaa so skip it */ 
    if (server.rdb_compression && len > 20) { 
        n = rdbSaveLzfStringObject(rdb,s,len); 
        if (n == -1) return -1; 
        if (n > 0) return n; 
        /* Return value of 0 means data can't be compressed, save the old way */ 
    } 

    /* Store verbatim(逐字) */ 
    if ((n = rdbSaveLen(rdb,len)) == -1) return -1; 
    nwritten += n; 
    if (len > 0) { 
        if (rdbWriteRaw(rdb,s,len) == -1) return -1; 
        nwritten += len; 
    } 
    return nwritten; 
} 

ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key) { 
    ssize_t n = 0, nwritten = 0; 

    if (o->type == OBJ_STRING) { 
        /* Save a string value */ 
        if ((n = rdbSaveStringObject(rdb,o)) == -1) return -1; 
        nwritten += n; 
    } else if (o->type == OBJ_LIST) { 
        /* Save a list value */ 
        if (o->encoding == OBJ_ENCODING_QUICKLIST) { 
            quicklist *ql = o->ptr; 
            quicklistNode *node = ql->head; 

            if ((n = rdbSaveLen(rdb,ql->len)) == -1) return -1; 
            nwritten += n; 

            while(node) { 
                if (quicklistNodeIsCompressed(node)) { 
                    void *data; 
                    size_t compress_len = quicklistGetLzf(node, &data); 
                    if ((n = rdbSaveLzfBlob(rdb,data,compress_len,node->sz)) == -1) return -1; 
                    nwritten += n; 
                } else { 
                    if ((n = rdbSaveRawString(rdb,node->zl,node->sz)) == -1) return -1; 
                    nwritten += n; 
                } 
                node = node->next; 
            } 
        } else { 
            serverPanic("Unknown list encoding"); 
        } 
    } 
    /////////////// 后面还有很多其他类型的 ///////////////////////// 
}