| 294 | } |
| 295 | |
| 296 | void RegionManager::doFlush(RegionId rid, bool async) { |
| 297 | // We're wasting the remaining bytes of a region, so track it for stats |
| 298 | externalFragmentation_.add(getRegion(rid).getFragmentationSize()); |
| 299 | |
| 300 | getRegion(rid).setPendingFlush(); |
| 301 | numInMemBufWaitingFlush_.inc(); |
| 302 | |
| 303 | if (!async) { |
| 304 | doFlushInternal(rid); |
| 305 | } else { |
| 306 | if (isOnWorker()) { |
| 307 | // If configured to flush async, schedule the flush job. |
| 308 | // It has to be scheduled to the same worker thread. Otherwise if we are |
| 309 | // draining, this flush job might be scheduled to an already drained |
| 310 | // worker thread. |
| 311 | if (workerFlushAsync_) { |
| 312 | getCurrentNavyThread()->addTaskRemote( |
| 313 | [this, rid]() { doFlushInternal(rid); }); |
| 314 | } else { |
| 315 | doFlushInternal(rid); |
| 316 | } |
| 317 | |
| 318 | } else { |
| 319 | getNextWorker().addTaskRemote([this, rid]() { doFlushInternal(rid); }); |
| 320 | } |
| 321 | } |
| 322 | } |
| 323 | |
| 324 | void RegionManager::doFlushInternal(RegionId rid) { |
| 325 | INJECT_PAUSE(pause_flush_begin); |