| 2324 | |
| 2325 | static std::atomic<bool> s_fFlushInProgress { false }; |
| 2326 | void flushStorageWeak() |
| 2327 | { |
| 2328 | bool fExpected = false; |
| 2329 | if (s_fFlushInProgress.compare_exchange_strong(fExpected, true /* desired */, std::memory_order_seq_cst, std::memory_order_relaxed)) |
| 2330 | { |
| 2331 | g_pserver->asyncworkqueue->AddWorkFunction([]{ |
| 2332 | aeAcquireLock(); |
| 2333 | mstime_t storage_process_latency; |
| 2334 | latencyStartMonitor(storage_process_latency); |
| 2335 | std::vector<redisDb*> vecdb; |
| 2336 | for (int idb = 0; idb < cserver.dbnum; ++idb) { |
| 2337 | if (g_pserver->db[idb]->processChanges(true)) |
| 2338 | vecdb.push_back(g_pserver->db[idb]); |
| 2339 | } |
| 2340 | latencyEndMonitor(storage_process_latency); |
| 2341 | latencyAddSampleIfNeeded("storage-process-changes", storage_process_latency); |
| 2342 | aeReleaseLock(); |
| 2343 | |
| 2344 | std::vector<const redisDbPersistentDataSnapshot*> vecsnapshotFree; |
| 2345 | vecsnapshotFree.resize(vecdb.size()); |
| 2346 | for (size_t idb = 0; idb < vecdb.size(); ++idb) |
| 2347 | vecdb[idb]->commitChanges(&vecsnapshotFree[idb]); |
| 2348 | |
| 2349 | for (size_t idb = 0; idb < vecsnapshotFree.size(); ++idb) { |
| 2350 | if (vecsnapshotFree[idb] != nullptr) |
| 2351 | vecdb[idb]->endSnapshotAsync(vecsnapshotFree[idb]); |
| 2352 | } |
| 2353 | s_fFlushInProgress = false; |
| 2354 | }, true /* fHiPri */); |
| 2355 | } |
| 2356 | else |
| 2357 | { |
| 2358 | serverLog(LOG_INFO, "Missed storage flush due to existing flush still in flight. Consider increasing storage-weak-flush-period"); |
| 2359 | } |
| 2360 | } |
| 2361 | |
| 2362 | /* This is our timer interrupt, called g_pserver->hz times per second. |
| 2363 | * Here is where we do a number of things that need to be done asynchronously. |
no test coverage detected