| 2895 | } |
| 2896 | |
| 2897 | void processJob(rdbInsertJob &job) { |
| 2898 | redisObjectStack keyobj; |
| 2899 | initStaticStringObject(keyobj,job.key); |
| 2900 | |
| 2901 | bool f1024thKey = false; |
| 2902 | bool fStaleMvccKey = (this->rsi) ? mvccFromObj(job.val) < this->rsi->mvccMinThreshold : false; |
| 2903 | |
| 2904 | /* Check if the key already expired. This function is used when loading |
| 2905 | * an RDB file from disk, either at startup, or when an RDB was |
| 2906 | * received from the master. In the latter case, the master is |
| 2907 | * responsible for key expiry. If we would expire keys here, the |
| 2908 | * snapshot taken by the master may not be reflected on the replica. */ |
| 2909 | bool fExpiredKey = iAmMaster() && !(this->rdbflags&RDBFLAGS_AOF_PREAMBLE) && job.expiretime != INVALID_EXPIRE && job.expiretime < this->now; |
| 2910 | if (fStaleMvccKey || fExpiredKey) { |
| 2911 | if (fStaleMvccKey && !fExpiredKey && this->rsi != nullptr && this->rsi->mi != nullptr && this->rsi->mi->staleKeyMap != nullptr && lookupKeyRead(job.db, &keyobj) == nullptr) { |
| 2912 | // We have a key that we've already deleted and is not back in our database. |
| 2913 | // We'll need to inform the sending master of the delete if it is also a replica of us |
| 2914 | robj_sharedptr objKeyDup(createStringObject(job.key, sdslen(job.key))); |
| 2915 | this->rsi->mi->staleKeyMap->operator[](job.db->id).push_back(objKeyDup); |
| 2916 | } |
| 2917 | sdsfree(job.key); |
| 2918 | job.key = nullptr; |
| 2919 | decrRefCount(job.val); |
| 2920 | job.val = nullptr; |
| 2921 | } else { |
| 2922 | /* Add the new object in the hash table */ |
| 2923 | int fInserted = dbMerge(job.db, job.key, job.val, (this->rsi && this->rsi->fForceSetKey) || (this->rdbflags & RDBFLAGS_ALLOW_DUP)); // Note: dbMerge will incrRef |
| 2924 | |
| 2925 | if (fInserted) |
| 2926 | { |
| 2927 | auto ckeys = this->ckeysLoaded.fetch_add(1, std::memory_order_relaxed); |
| 2928 | f1024thKey = (ckeys % 1024) == 0; |
| 2929 | |
| 2930 | /* Set the expire time if needed */ |
| 2931 | if (job.expiretime != INVALID_EXPIRE) |
| 2932 | { |
| 2933 | setExpire(NULL,job.db,&keyobj,nullptr,job.expiretime); |
| 2934 | } |
| 2935 | |
| 2936 | /* Set usage information (for eviction). */ |
| 2937 | objectSetLRUOrLFU(job.val,job.lfu_freq,job.lru_idle,job.lru_clock,1000); |
| 2938 | |
| 2939 | /* call key space notification on key loaded for modules only */ |
| 2940 | moduleNotifyKeyspaceEvent(NOTIFY_LOADED, "loaded", &keyobj, job.db->id); |
| 2941 | |
| 2942 | replicationNotifyLoadedKey(job.db, &keyobj, job.val, job.expiretime); |
| 2943 | |
| 2944 | for (auto &pair : job.vecsubexpires) |
| 2945 | { |
| 2946 | setExpire(NULL, job.db, &keyobj, pair.first, pair.second); |
| 2947 | replicateSubkeyExpire(job.db, &keyobj, pair.first.get(), pair.second); |
| 2948 | } |
| 2949 | |
| 2950 | job.val = nullptr; // don't free this as we moved ownership to the DB |
| 2951 | } |
| 2952 | } |
| 2953 | |
| 2954 | /* If we have a storage provider check if we need to evict some keys to stay under our memory limit, |
no test coverage detected