@@ -238,7 +238,7 @@ struct SourceNode : ExecNode, public TracedNode {
238238
239239 void PauseProducing (ExecNode* output, int32_t counter) override {
240240 std::lock_guard<std::mutex> lg (mutex_);
241- if (counter <= backpressure_counter_) {
241+ if (counter <= backpressure_counter_ || stop_requested_ ) {
242242 return ;
243243 }
244244 backpressure_counter_ = counter;
@@ -266,11 +266,11 @@ struct SourceNode : ExecNode, public TracedNode {
266266 to_finish.MarkFinished ();
267267 }
268268
269- Status StopProducing () override {
270- // GH-35837: ensure node is not paused
269+ Status StopProducingImpl () override {
271270 Future<> to_finish;
272271 {
273272 std::lock_guard<std::mutex> lg (mutex_);
273+ stop_requested_ = true ;
274274 if (!backpressure_future_.is_finished ()) {
275275 to_finish = backpressure_future_;
276276 backpressure_future_ = Future<>::MakeFinished ();
@@ -279,13 +279,6 @@ struct SourceNode : ExecNode, public TracedNode {
279279 if (to_finish.is_valid ()) {
280280 to_finish.MarkFinished ();
281281 }
282- // only then stop
283- return ExecNode::StopProducing ();
284- }
285-
286- Status StopProducingImpl () override {
287- std::unique_lock<std::mutex> lock (mutex_);
288- stop_requested_ = true ;
289282 return Status::OK ();
290283 }
291284
0 commit comments