| 2484 | } |
| 2485 | |
| 2486 | void removeLog(TLogData* self, Reference<LogData> logData) { |
| 2487 | TraceEvent("TLogRemoved", self->dbgid) |
| 2488 | .detail("LogId", logData->logId) |
| 2489 | .detail("Input", logData->bytesInput.getValue()) |
| 2490 | .detail("Durable", logData->bytesDurable.getValue()); |
| 2491 | logData->stopped = true; |
| 2492 | if (!logData->recoveryComplete.isSet()) { |
| 2493 | logData->recoveryComplete.sendError(end_of_stream()); |
| 2494 | } |
| 2495 | |
| 2496 | logData->addActor = PromiseStream<Future<Void>>(); // there could be items still in the promise stream if one of the |
| 2497 | // actors threw an error immediately |
| 2498 | self->id_data.erase(logData->logId); |
| 2499 | |
| 2500 | while (self->popOrder.size() && !self->id_data.count(self->popOrder.front())) { |
| 2501 | self->popOrder.pop_front(); |
| 2502 | } |
| 2503 | |
| 2504 | if (self->id_data.size()) { |
| 2505 | return; |
| 2506 | } else { |
| 2507 | throw worker_removed(); |
| 2508 | } |
| 2509 | if (logData->queueCommittingVersion == 0) { |
| 2510 | // If the removed tlog never attempted a queue commit, the update storage loop could become stuck waiting for |
| 2511 | // queueCommittedVersion to advance. |
| 2512 | logData->queueCommittedVersion.set(std::numeric_limits<Version>::max()); |
| 2513 | } |
| 2514 | } |
| 2515 | |
| 2516 | // copy data from old gene to new gene without deserializing |
| 2517 | ACTOR Future<Void> pullAsyncData(TLogData* self, |