| 1148 | }; |
| 1149 | |
| 1150 | int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) { |
| 1151 | // TODO: This needs to be on a background thread |
| 1152 | int retval = C_OK; |
| 1153 | serverAssert(GlobalLocksAcquired()); |
| 1154 | serverLog(LL_NOTICE, "Starting fast full sync with target: %s", "disk"); |
| 1155 | |
| 1156 | std::shared_ptr<replicationBuffer> spreplBuf = std::make_shared<replicationBuffer>(); |
| 1157 | listNode *ln; |
| 1158 | listIter li; |
| 1159 | listRewind(g_pserver->slaves, &li); |
| 1160 | while ((ln = listNext(&li))) { |
| 1161 | client *replicaCur = (client*)listNodeValue(ln); |
| 1162 | if ((replicaCur->slave_capa & SLAVE_CAPA_KEYDB_FASTSYNC) && (replicaCur->replstate == SLAVE_STATE_WAIT_BGSAVE_START)) { |
| 1163 | spreplBuf->addReplica(replicaCur); |
| 1164 | replicaCur->replstate = SLAVE_STATE_FASTSYNC_TX; |
| 1165 | } |
| 1166 | } |
| 1167 | |
| 1168 | spreplBuf->addArrayLen(2); // Two sections: Metadata and databases |
| 1169 | |
| 1170 | // MetaData |
| 1171 | aeAcquireLock(); |
| 1172 | spreplBuf->addArrayLen(3 + dictSize(g_pserver->lua_scripts)); |
| 1173 | spreplBuf->addArrayLen(2); |
| 1174 | spreplBuf->addString("repl-stream-db", 14); |
| 1175 | spreplBuf->addLong(rsi->repl_stream_db); |
| 1176 | spreplBuf->addArrayLen(2); |
| 1177 | spreplBuf->addString("repl-id", 7); |
| 1178 | spreplBuf->addString(rsi->repl_id, CONFIG_RUN_ID_SIZE); |
| 1179 | spreplBuf->addArrayLen(2); |
| 1180 | spreplBuf->addString("repl-offset", 11); |
| 1181 | spreplBuf->addLongLong(rsi->master_repl_offset); |
| 1182 | |
| 1183 | if (dictSize(g_pserver->lua_scripts)) { |
| 1184 | dictEntry *de; |
| 1185 | auto di = dictGetIterator(g_pserver->lua_scripts); |
| 1186 | while((de = dictNext(di)) != NULL) { |
| 1187 | robj *body = (robj*)dictGetVal(de); |
| 1188 | |
| 1189 | spreplBuf->addArrayLen(2); |
| 1190 | spreplBuf->addString("lua", 3); |
| 1191 | spreplBuf->addString(szFromObj(body), sdslen(szFromObj(body))); |
| 1192 | } |
| 1193 | dictReleaseIterator(di); |
| 1194 | di = NULL; /* So that we don't release it again on error. */ |
| 1195 | } |
| 1196 | |
| 1197 | std::shared_ptr<std::vector<const redisDbPersistentDataSnapshot*>> spvecsnapshot = std::make_shared<std::vector<const redisDbPersistentDataSnapshot*>>(); |
| 1198 | for (int idb = 0; idb < cserver.dbnum; ++idb) { |
| 1199 | spvecsnapshot->emplace_back(g_pserver->db[idb]->createSnapshot(getMvccTstamp(), false)); |
| 1200 | } |
| 1201 | aeReleaseLock(); |
| 1202 | |
| 1203 | g_pserver->asyncworkqueue->AddWorkFunction([spreplBuf = std::move(spreplBuf), spvecsnapshot = std::move(spvecsnapshot)]{ |
| 1204 | int retval = C_OK; |
| 1205 | auto timeStart = ustime(); |
| 1206 | auto lastLogTime = timeStart; |
| 1207 | size_t cbData = 0; |
no test coverage detected