return status_;
  }

  Status GetStatus() { return status_; }

  void SetStatus(const Status& status) { status_ = status; }

  bool IsRunning() const override { return !stopped_; };

  void Stop() { stopped_ = true; }

 private:
  void CountThread(std::atomic<int>* counter, int until, Notification* start) {
    if (start != nullptr) start->WaitForNotification();

  /// Returns the latest status.
  Status GetStatus();

  // Returns the stored cost model.
  Status ExportCostGraph(CostGraphDef* cost_graph) const override;

 private:
  QueueRunner() : coord_(nullptr), stopped_(false), cg_mu_(nullptr) {}

  // Initializes the instance with the QueueRunnerDef proto.
  Status Init(const QueueRunnerDef& queue_runner_def);

  // The Run function for each thread.

}

void QueueRunner::Stop(Session* sess) {
  if (coord_ != nullptr) {
    coord_->WaitForStop();
  }
  if (!cancel_op_name_.empty()) {
    UpdateStatus(RealRun(sess, cancel_op_name_, false));
  }
  stopped_ = true;
}

Status QueueRunner::Join() {
  thread_pool_.reset();
  mutex_lock l(mu_);
  return status_;
}

void QueueRunner::UpdateStatus(const Status& status) {
  {
    mutex_lock l(mu_);

			continue
		}
		diskStats[disk.PoolIndex].AvailableSpace += disk.AvailableSpace
		diskStats[disk.PoolIndex].TotalSpace += disk.TotalSpace
	}

	stopTime := meta.StoppedAt
	r = rebalanceAdminStatus{
		ID:        meta.ID,
		StoppedAt: meta.StoppedAt,
		Pools:     make([]rebalancePoolStatus, len(meta.PoolStats)),
	}
	for i, ps := range meta.PoolStats {
		r.Pools[i] = rebalancePoolStatus{
			ID:     i,

// The actual server loop (stoppable by closing stopCh) runs in a go routine, i.e. Serve does not block.
// It returns a stoppedCh that is closed when all non-hijacked active requests have been processed.
// It returns a listenerStoppedCh that is closed when the underlying http Server has stopped listening.

	"context"
	"fmt"
	"sync"
)

type status int

const (
	Pending status = iota
	Ready
	Stopped
)

// ready is a three state condition variable that blocks until is Ready if is not Stopped.
// Its initial state is Pending and its state machine diagram is as follow.
//
// Pending <------> Ready -----> Stopped
//
//	|                           ^
//	└---------------------------┘
type ready struct {

	f.Lock()
	defer f.Unlock()
	if !f.stopped {
		klog.V(4).Infof("Stopping fake watcher.")
		close(f.result)
		f.stopped = true
	}
}

func (f *FakeWatcher) IsStopped() bool {
	f.Lock()
	defer f.Unlock()
	return f.stopped
}

// Reset prepares the watcher to be reused.
func (f *FakeWatcher) Reset() {
	f.Lock()
	defer f.Unlock()
	f.stopped = false
	f.result = make(chan Event)
}

            executor.execute(action1)
            executor.execute(action2)
            executor.stop()
            instant.stopped
        }

        then:
        instant.stopped > instant.completed1
        instant.stopped > instant.completed2
    }

    def factoryStopBlocksUntilAllJobsAreComplete() {
        given:
        def action1 = {
            thread.block()

                            break;
                        case Stopped:
                            LOGGER.debug("daemon has stopped.");
                            return DaemonStopState.Clean;
                        case ForceStopped:
                            LOGGER.debug("daemon has been force stopped.");
                            return DaemonStopState.Forced;
                    }

        out =~ /\n\s*\d+\s+IDLE\s+([\w\.\+\-]+)/
        out =~ /\n\s*\d+\s+BUSY\s+([\w\.\+\-]+)/
        out =~ /\n\s*12346\s+STOPPED\s+\(GRACEFUL_EXPIRE_REASON\)/

        cleanup:
        block?.releaseAll()
        build?.waitForFinish()
    }

    def "reports stopped status of recently stopped daemons"() {
        given: