| 2741 | } |
| 2742 | |
| 2743 | void removeLog(TLogData* self, Reference<LogData> logData) { |
| 2744 | TraceEvent("TLogRemoved", self->dbgid) |
| 2745 | .detail("LogId", logData->logId) |
| 2746 | .detail("Input", logData->bytesInput.getValue()) |
| 2747 | .detail("Durable", logData->bytesDurable.getValue()); |
| 2748 | logData->stop(); |
| 2749 | logData->unblockWaitingPeeks(); |
| 2750 | if (!logData->recoveryComplete.isSet()) { |
| 2751 | logData->recoveryComplete.sendError(end_of_stream()); |
| 2752 | } |
| 2753 | |
| 2754 | logData->addActor = PromiseStream<Future<Void>>(); // there could be items still in the promise stream if one of the |
| 2755 | // actors threw an error immediately |
| 2756 | self->id_data.erase(logData->logId); |
| 2757 | |
| 2758 | while (self->popOrder.size() && !self->id_data.count(self->popOrder.front())) { |
| 2759 | self->popOrder.pop_front(); |
| 2760 | } |
| 2761 | |
| 2762 | if (self->id_data.size() == 0) { |
| 2763 | throw worker_removed(); |
| 2764 | } |
| 2765 | if (logData->queueCommittingVersion == 0) { |
| 2766 | // If the removed tlog never attempted a queue commit, the update storage loop could become stuck waiting for |
| 2767 | // queueCommittedVersion to advance. |
| 2768 | logData->queueCommittedVersion.set(std::numeric_limits<Version>::max()); |
| 2769 | } |
| 2770 | } |
| 2771 | |
| 2772 | // remote tLog pull data from log routers |
| 2773 | ACTOR Future<Void> pullAsyncData(TLogData* self, |