[clangd] Stop exposing Futures from ClangdServer operations.

[sammccall]

Summary

LSP has asynchronous semantics, being able to block on an async operation
completing is unneccesary and leads to tighter coupling with the threading.

In practice only tests depend on this, so we add a general-purpose "block until
idle" function to the scheduler which will work for all operations.

To get this working, fix a latent condition-variable bug in ASTWorker, and make
AsyncTaskRunner const-correct.

[sammccall]

Tidy up comment, and revert notify_all to notify_one - it was a red herring.

[sammccall]

rebase

[ilya-biryukov]

clangd/Threading.h

60

Maybe remove the comment or add more context (i.e. add references) on why the overflow is buggy?

66

Maybe move this helper to .cpp file?

68

Maybe keep the locking part out of this helper? It's often desirable to hold the lock after wait(). This will model how std::condition_variable::wait is defined.

83–84

Add LLVM_NODICARD here?
For that particular method maybe we could have two overload: with and without the deadline, i.e.

void waitForAll() const;
LLVM_NODISCARD bool waitForAll(Deadline D) const;

There are places (like the destructor of this class) where the first overload is used and consuming the return value is just adding noise, but clients passing the deadline (e.g. blockUntilIdle()) should definitely consume the return value.

unittests/clangd/ClangdTests.cpp

785–786

Do we need to change this test?
It was specifically designed to keep the second request from overriding the first one before it was processed.

[sammccall]

Change AsyncTaskRunner::wait() to be LLVM_NODISCARD when used with a deadline.
Restore NoConcurrentDiagnostics test to its former glory (with comments)
Other comment fixes.

clangd/Threading.h

60

Done. (Making specific reference to time_point::max() + overflow probably makes it clear why it's unusable?)

66

it's public, and it's a template - what am I missing?

68

I tried this, but find the API just way too grungy - we need to pass a unique_lock<std::mutex>&?
Generally I'd hope we're aiming for something a little higher level than the standard library!

We don't actually ever need to do anything with the lock held. If we did, what about passing a lambda?

unittests/clangd/ClangdTests.cpp

785–786

You're right. It wasn't obvious to me that this was what it was doing or why, so it seemed easy to simplify.
Added some comments and a sanity check assertion, reverted the logic changes.

(The change might make sense anyway - we rely on sleep already, a second sleep would get rid of all the synchronization without adding latency or changing behavior, but not in this patch)

[ilya-biryukov]

clangd/TUScheduler.cpp

286

Change to notify_all()? Otherwise we might wake up some thread waiting on blockUntilIdle(), but not the worker thread.

314

We should probably call notify_all() here.
Having multiple blockUntilIdle might not work otherwise.

clangd/Threading.h

66

Sorry, I missed that it's used outside Threading.h

68

The higher-level abstraction would certainly be appealing to me as well and I would welcome it.
I don't think we get it here, though. It's actually a bit confusing that the function calls into STL's versions of wait and wait_until, has a similar name, but does a slightly different thing.

unittests/clangd/ClangdTests.cpp

783

Maybe use paren initializers? Looks a bit less curly :-)

std::atomic<int> Count(0);

[sammccall]

Review comments

clangd/TUScheduler.cpp

286

Done.

In fact I had changed this to notify_all, and changed it back because it's safe:

• this class isn't threadsafe
• therefore there are only two interesting threads, the caller thread and the worker thread
• each thread can only wake up the other one, so there's no unsatisfied waiter

But as discussed offline, notify_all() is just a less surprising default, and will fail in more obvious ways when we write threading bugs.

clangd/Threading.h

68

Yeah, fair enough.

unittests/clangd/ClangdTests.cpp

783

This is a member initializer, I think I have to use this syntax :-(

[ilya-biryukov]

LGTM

unittests/clangd/ClangdTests.cpp

783

You're right, you have to do braced init here.
Sorry, my mistake.

clangd/ClangdServer.h

public:
 
   /// Set the root path of the workspace.
   void setRootPath(PathRef RootPath);
 
   /// Add a \p File to the list of tracked C++ files or update the contents if
   /// \p File is already tracked. Also schedules parsing of the AST for it on a
   /// separate thread. When the parsing is complete, DiagConsumer passed in
   /// constructor will receive onDiagnosticsReady callback.
-  /// \return A future that will become ready when the rebuild (including
-  /// diagnostics) is finished.
-  /// FIXME: don't return futures here, LSP does not require a response for this
-  /// request.
-  std::future<void> addDocument(PathRef File, StringRef Contents);
+  void addDocument(PathRef File, StringRef Contents);
 
   /// Remove \p File from list of tracked files, schedule a request to free
   /// resources associated with it.
   void removeDocument(PathRef File);
 
   /// Force \p File to be reparsed using the latest contents.
   /// Will also check if CompileCommand, provided by GlobalCompilationDatabase
   /// for \p File has changed. If it has, will remove currently stored Preamble
   /// and AST and rebuild them from scratch.
-  /// FIXME: don't return futures here, LSP does not require a response for this
-  /// request.
-  std::future<void> forceReparse(PathRef File);
+  void forceReparse(PathRef File);
 
   /// Run code completion for \p File at \p Pos.
   /// Request is processed asynchronously.
   ///
   /// If \p OverridenContents is not None, they will used only for code
   /// completion, i.e. no diagnostics update will be scheduled and a draft for
   /// \p File will not be updated. If \p OverridenContents is None, contents of
   /// the current draft for \p File will be used. If \p UsedFS is non-null, it
public:
   /// The order of results is unspecified.
   /// Overall memory usage of clangd may be significantly more than reported
   /// here, as this metric does not account (at least) for:
   ///   - memory occupied by static and dynamic index,
   ///   - memory required for in-flight requests,
   /// FIXME: those metrics might be useful too, we should add them.
   std::vector<std::pair<Path, std::size_t>> getUsedBytesPerFile() const;
 
+  // Blocks the main thread until the server is idle. Only for use in tests.
+  // Returns false if the timeout expires.
+  LLVM_NODISCARD bool
+  blockUntilIdleForTest(llvm::Optional<double> TimeoutSeconds = 10);
+
 private:
   /// FIXME: This stats several files to find a .clang-format file. I/O can be
   /// slow. Think of a way to cache this.
   llvm::Expected<tooling::Replacements>
   formatCode(llvm::StringRef Code, PathRef File,
              ArrayRef<tooling::Range> Ranges);
 
-  std::future<void>
+  void
   scheduleReparseAndDiags(PathRef File, VersionedDraft Contents,
                           Tagged<IntrusiveRefCntPtr<vfs::FileSystem>> TaggedFS);
 
   CompileArgsCache CompileArgs;
   DiagnosticsConsumer &DiagConsumer;
   FileSystemProvider &FSProvider;
   DraftStore DraftMgr;
   // The index used to look up symbols. This could be:

clangd/ClangdServer.cpp

 
 void ClangdServer::setRootPath(PathRef RootPath) {
   std::string NewRootPath = llvm::sys::path::convert_to_slash(
       RootPath, llvm::sys::path::Style::posix);
   if (llvm::sys::fs::is_directory(NewRootPath))
     this->RootPath = NewRootPath;
 }
 
-std::future<void> ClangdServer::addDocument(PathRef File, StringRef Contents) {
+void ClangdServer::addDocument(PathRef File, StringRef Contents) {
   DocVersion Version = DraftMgr.updateDraft(File, Contents);
   auto TaggedFS = FSProvider.getTaggedFileSystem(File);
-  return scheduleReparseAndDiags(File, VersionedDraft{Version, Contents.str()},
-                                 std::move(TaggedFS));
+  scheduleReparseAndDiags(File, VersionedDraft{Version, Contents.str()},
 }
 
 void ClangdServer::removeDocument(PathRef File) {
   DraftMgr.removeDraft(File);
   CompileArgs.invalidate(File);
   WorkScheduler.remove(File);
 }
 
-std::future<void> ClangdServer::forceReparse(PathRef File) {
+void ClangdServer::forceReparse(PathRef File) {
   auto FileContents = DraftMgr.getDraft(File);
   assert(FileContents.Draft &&
          "forceReparse() was called for non-added document");
 
   // forceReparse promises to request new compilation flags from CDB, so we
   // remove any cahced flags.
   CompileArgs.invalidate(File);
 
   auto TaggedFS = FSProvider.getTaggedFileSystem(File);
-  return scheduleReparseAndDiags(File, std::move(FileContents),
-                                 std::move(TaggedFS));
+  scheduleReparseAndDiags(File, std::move(FileContents), std::move(TaggedFS));
 }
 
 void ClangdServer::codeComplete(
     PathRef File, Position Pos, const clangd::CodeCompleteOptions &Opts,
     UniqueFunction<void(Tagged<CompletionList>)> Callback,
     llvm::Optional<StringRef> OverridenContents,
     IntrusiveRefCntPtr<vfs::FileSystem> *UsedFS) {
   using CallbackType = UniqueFunction<void(Tagged<CompletionList>)>;
auto Action = [=](llvm::Expected<InputsAndAST> InpAST) -> RetType {
     if (!InpAST)
       return InpAST.takeError();
     auto Result = clangd::findDocumentHighlights(InpAST->AST, Pos);
     return make_tagged(std::move(Result), TaggedFS.Tag);
   };
   return blockingRunWithAST<RetType>(WorkScheduler, File, Action);
 }
 
-std::future<void> ClangdServer::scheduleReparseAndDiags(
+void ClangdServer::scheduleReparseAndDiags(
     PathRef File, VersionedDraft Contents,
     Tagged<IntrusiveRefCntPtr<vfs::FileSystem>> TaggedFS) {
   tooling::CompileCommand Command = CompileArgs.getCompileCommand(File);
 
   using OptDiags = llvm::Optional<std::vector<DiagWithFixIts>>;
 
   DocVersion Version = Contents.Version;
   Path FileStr = File.str();
   VFSTag Tag = std::move(TaggedFS.Tag);
 
-  std::promise<void> DonePromise;
-  std::future<void> DoneFuture = DonePromise.get_future();
-
-  auto Callback = [this, Version, FileStr, Tag](std::promise<void> DonePromise,
-                                                OptDiags Diags) {
-    auto Guard = llvm::make_scope_exit([&]() { DonePromise.set_value(); });
+  auto Callback = [this, Version, FileStr, Tag](OptDiags Diags) {
     if (!Diags)
       return; // A new reparse was requested before this one completed.
 
     // We need to serialize access to resulting diagnostics to avoid calling
     // `onDiagnosticsReady` in the wrong order.
     std::lock_guard<std::mutex> DiagsLock(DiagnosticsMutex);
     DocVersion &LastReportedDiagsVersion = ReportedDiagnosticVersions[FileStr];
     // FIXME(ibiryukov): get rid of '<' comparison here. In the current
     // implementation diagnostics will not be reported after version counters'
     // overflow. This should not happen in practice, since DocVersion is a
     // 64-bit unsigned integer.
     if (Version < LastReportedDiagsVersion)
       return;
     LastReportedDiagsVersion = Version;
 
     DiagConsumer.onDiagnosticsReady(
         FileStr, make_tagged(std::move(*Diags), std::move(Tag)));
   };
 
   WorkScheduler.update(File,
                        ParseInputs{std::move(Command),
                                    std::move(TaggedFS.Value),
                                    std::move(*Contents.Draft)},
-                       BindWithForward(Callback, std::move(DonePromise)));
-  return DoneFuture;
+                       std::move(Callback));
 }
 
 void ClangdServer::onFileEvent(const DidChangeWatchedFilesParams &Params) {
   // FIXME: Do nothing for now. This will be used for indexing and potentially
   // invalidating other caches.
 }
 
 std::vector<std::pair<Path, std::size_t>>
 ClangdServer::getUsedBytesPerFile() const {
   return WorkScheduler.getUsedBytesPerFile();
 }
+
+LLVM_NODISCARD bool
+ClangdServer::blockUntilIdleForTest(llvm::Optional<double> TimeoutSeconds) {
+  return WorkScheduler.blockUntilIdle(timeoutSeconds(TimeoutSeconds));
+}

clangd/TUScheduler.h

public:
   /// be built for any version of the file, callers must not rely on it being
   /// consistent with the current version of the file.
   /// If an error occurs during processing, it is forwarded to the \p Action
   /// callback.
   void runWithPreamble(
       PathRef File,
       UniqueFunction<void(llvm::Expected<InputsAndPreamble>)> Action);
 
+  /// Wait until there are no scheduled or running tasks.
+  /// Mostly useful for synchronizing tests.
+  bool blockUntilIdle(Deadline D) const;
+
 private:
   /// This class stores per-file data in the Files map.
   struct FileData;
 
   const bool StorePreamblesInMemory;
   const std::shared_ptr<PCHContainerOperations> PCHOps;
   const ASTParsedCallback ASTCallback;
   Semaphore Barrier;
   llvm::StringMap<std::unique_ptr<FileData>> Files;
   // None when running tasks synchronously and non-None when running tasks
   // asynchronously.
-  llvm::Optional<AsyncTaskRunner> Tasks;
+  llvm::Optional<AsyncTaskRunner> PreambleTasks;
+  llvm::Optional<AsyncTaskRunner> WorkerThreads;
 };
 } // namespace clangd
 } // namespace clang
 
 #endif

clangd/TUScheduler.cpp

public:
   static ASTWorkerHandle Create(AsyncTaskRunner *Tasks, Semaphore &Barrier,
                                 CppFile AST);
   ~ASTWorker();
 
   void update(ParseInputs Inputs,
               UniqueFunction<void(llvm::Optional<std::vector<DiagWithFixIts>>)>
                   OnUpdated);
   void runWithAST(UniqueFunction<void(llvm::Expected<InputsAndAST>)> Action);
+  bool blockUntilIdle(Deadline Timeout) const;
 
   std::shared_ptr<const PreambleData> getPossiblyStalePreamble() const;
   std::size_t getUsedBytes() const;
 
 private:
   // Must be called exactly once on processing thread. Will return after
   // stop() is called on a separate thread and all pending requests are
   // processed.
private:
   // Result of getUsedBytes() after the last rebuild or read of AST.
   std::size_t LastASTSize; /* GUARDED_BY(Mutex) */
   // Set to true to signal run() to finish processing.
   bool Done;                           /* GUARDED_BY(Mutex) */
   std::queue<RequestWithCtx> Requests; /* GUARDED_BY(Mutex) */
   // Only set when last request is an update. This allows us to cancel an update
   // that was never read, if a subsequent update comes in.
   llvm::Optional<CancellationFlag> LastUpdateCF; /* GUARDED_BY(Mutex) */
-  std::condition_variable RequestsCV;
+  mutable std::condition_variable RequestsCV;
 };
 
 /// A smart-pointer-like class that points to an active ASTWorker.
 /// In destructor, signals to the underlying ASTWorker that no new requests will
 /// be sent and the processing loop may exit (after running all pending
 /// requests).
 class ASTWorkerHandle {
   friend class ASTWorker;
if (isUpdate) {
         LastUpdateCF->cancel();
       }
       LastUpdateCF = std::move(*CF);
     } else {
       LastUpdateCF = llvm::None;
     }
     Requests.emplace(std::move(Task), Context::current().clone());
   } // unlock Mutex.
   RequestsCV.notify_one();
 }
 
 void ASTWorker::run() {
   while (true) {
     RequestWithCtx Req;
     {
       std::unique_lock<std::mutex> Lock(Mutex);
       RequestsCV.wait(Lock, [&]() { return Done || !Requests.empty(); });
       if (Requests.empty()) {
         assert(Done);
         return;
       }
       // Even when Done is true, we finish processing all pending requests
       // before exiting the processing loop.
 
       Req = std::move(Requests.front());
-      Requests.pop();
+      // Leave it on the queue for now, so waiters don't see an empty queue.
     } // unlock Mutex
 
     std::lock_guard<Semaphore> BarrierLock(Barrier);
     WithContext Guard(std::move(Req.second));
     Req.first();
+
+    {
+      std::lock_guard<std::mutex> Lock(Mutex);
+      Requests.pop();
+    }
+    RequestsCV.notify_one();
   }
 }
+
+bool ASTWorker::blockUntilIdle(Deadline Timeout) const {
+  return wait(Mutex, RequestsCV, Timeout, [&] { return Requests.empty(); });
+}
+
 } // namespace
 
 unsigned getDefaultAsyncThreadsCount() {
   unsigned HardwareConcurrency = std::thread::hardware_concurrency();
   // C++ standard says that hardware_concurrency()
   // may return 0, fallback to 1 worker thread in
   // that case.
   if (HardwareConcurrency == 0)
     return 1;
   return HardwareConcurrency;
 }
 
 struct TUScheduler::FileData {
   /// Latest inputs, passed to TUScheduler::update().
   ParseInputs Inputs;
   ASTWorkerHandle Worker;
 };
 
 TUScheduler::TUScheduler(unsigned AsyncThreadsCount,
                          bool StorePreamblesInMemory,
                          ASTParsedCallback ASTCallback)
     : StorePreamblesInMemory(StorePreamblesInMemory),
       PCHOps(std::make_shared<PCHContainerOperations>()),
       ASTCallback(std::move(ASTCallback)), Barrier(AsyncThreadsCount) {
-  if (0 < AsyncThreadsCount)
-    Tasks.emplace();
+  if (0 < AsyncThreadsCount) {
+    PreambleTasks.emplace();
+    WorkerThreads.emplace();
+  }
 }
 
 TUScheduler::~TUScheduler() {
   // Notify all workers that they need to stop.
   Files.clear();
 
   // Wait for all in-flight tasks to finish.
-  if (Tasks)
-    Tasks->waitForAll();
+  if (PreambleTasks)
+    PreambleTasks->wait();
+  if (WorkerThreads)
+    WorkerThreads->wait();
+}
+
+bool TUScheduler::blockUntilIdle(Deadline D) const {
+  for (auto &File : Files)
+    if (!File.getValue()->Worker->blockUntilIdle(D))
+      return false;
+  if (PreambleTasks)
+    if (!PreambleTasks->wait(D))
+      return false;
+  return true;
 }
 
 void TUScheduler::update(
     PathRef File, ParseInputs Inputs,
     UniqueFunction<void(llvm::Optional<std::vector<DiagWithFixIts>>)>
         OnUpdated) {
   std::unique_ptr<FileData> &FD = Files[File];
   if (!FD) {
     // Create a new worker to process the AST-related tasks.
     ASTWorkerHandle Worker = ASTWorker::Create(
-        Tasks ? Tasks.getPointer() : nullptr, Barrier,
+        WorkerThreads ? WorkerThreads.getPointer() : nullptr, Barrier,
         CppFile(File, StorePreamblesInMemory, PCHOps, ASTCallback));
     FD = std::unique_ptr<FileData>(new FileData{Inputs, std::move(Worker)});
   } else {
     FD->Inputs = Inputs;
   }
   FD->Worker->update(std::move(Inputs), std::move(OnUpdated));
 }
 
void TUScheduler::runWithPreamble(
   auto It = Files.find(File);
   if (It == Files.end()) {
     Action(llvm::make_error<llvm::StringError>(
         "trying to get preamble for non-added document",
         llvm::errc::invalid_argument));
     return;
   }
 
-  if (!Tasks) {
+  if (!PreambleTasks) {
     std::shared_ptr<const PreambleData> Preamble =
         It->second->Worker->getPossiblyStalePreamble();
     Action(InputsAndPreamble{It->second->Inputs, Preamble.get()});
     return;
   }
 
   ParseInputs InputsCopy = It->second->Inputs;
   std::shared_ptr<const ASTWorker> Worker = It->second->Worker.lock();
   auto Task = [InputsCopy, Worker, this](Context Ctx,
                                          decltype(Action) Action) mutable {
     std::lock_guard<Semaphore> BarrierLock(Barrier);
     WithContext Guard(std::move(Ctx));
     std::shared_ptr<const PreambleData> Preamble =
         Worker->getPossiblyStalePreamble();
     Action(InputsAndPreamble{InputsCopy, Preamble.get()});
   };
 
-  Tasks->runAsync(
+  PreambleTasks->runAsync(
       BindWithForward(Task, Context::current().clone(), std::move(Action)));
 }
 
 std::vector<std::pair<Path, std::size_t>>
 TUScheduler::getUsedBytesPerFile() const {
   std::vector<std::pair<Path, std::size_t>> Result;
   Result.reserve(Files.size());
   for (auto &&PathAndFile : Files)
     Result.push_back(
         {PathAndFile.first(), PathAndFile.second->Worker->getUsedBytes()});
   return Result;
 }
 
 } // namespace clangd
 } // namespace clang

clangd/Threading.h

public:
   void unlock();
 
 private:
   std::mutex Mutex;
   std::condition_variable SlotsChanged;
   std::size_t FreeSlots;
 };
 
+/// A point in time we may wait for, or None to wait forever.
+/// (Not time_point::max(), because many std::chrono implementations overflow).
+using Deadline = llvm::Optional<std::chrono::steady_clock::time_point>;
+/// Makes a deadline from a timeout in seconds.
+Deadline timeoutSeconds(llvm::Optional<double> Seconds);
+/// Waits on a condition variable until F() is true or D expires.
+template <typename Func>
+LLVM_NODISCARD bool wait(std::mutex &Mutex, std::condition_variable &CV,
+                         Deadline D, Func F) {
+  std::unique_lock<std::mutex> Lock(Mutex);
+  if (D)
+    return CV.wait_until(Lock, *D, F);
+  CV.wait(Lock, F);
+  return true;
+}
+
 /// Runs tasks on separate (detached) threads and wait for all tasks to finish.
 /// Objects that need to spawn threads can own an AsyncTaskRunner to ensure they
 /// all complete on destruction.
 class AsyncTaskRunner {
 public:
   /// Destructor waits for all pending tasks to finish.
   ~AsyncTaskRunner();
 
-  void waitForAll();
+  void wait() const { (void) wait(llvm::None); }
+  LLVM_NODISCARD bool wait(Deadline D) const;
   void runAsync(UniqueFunction<void()> Action);
 
 private:
-  std::mutex Mutex;
-  std::condition_variable TasksReachedZero;
+  mutable std::mutex Mutex;
+  mutable std::condition_variable TasksReachedZero;
   std::size_t InFlightTasks = 0;
 };
 } // namespace clangd
 } // namespace clang
 #endif

clangd/Threading.cpp

 void Semaphore::unlock() {
   std::unique_lock<std::mutex> Lock(Mutex);
   ++FreeSlots;
   Lock.unlock();
 
   SlotsChanged.notify_one();
 }
 
-AsyncTaskRunner::~AsyncTaskRunner() { waitForAll(); }
+AsyncTaskRunner::~AsyncTaskRunner() { wait(); }
 
-void AsyncTaskRunner::waitForAll() {
-  std::unique_lock<std::mutex> Lock(Mutex);
-  TasksReachedZero.wait(Lock, [&]() { return InFlightTasks == 0; });
+bool AsyncTaskRunner::wait(Deadline D) const {
+  return clangd::wait(Mutex, TasksReachedZero, D,
+                      [&] { return InFlightTasks == 0; });
 }
 
 void AsyncTaskRunner::runAsync(UniqueFunction<void()> Action) {
   {
-    std::unique_lock<std::mutex> Lock(Mutex);
+    std::lock_guard<std::mutex> Lock(Mutex);
     ++InFlightTasks;
   }
 
   auto CleanupTask = llvm::make_scope_exit([this]() {
     std::lock_guard<std::mutex> Lock(Mutex);
     int NewTasksCnt = --InFlightTasks;
     if (NewTasksCnt == 0) {
       // Note: we can't unlock here because we don't want the object to be
       // destroyed before we notify.
       TasksReachedZero.notify_one();
     }
   });
 
   std::thread(
       [](decltype(Action) Action, decltype(CleanupTask)) {
         Action();
         // Make sure function stored by Action is destroyed before CleanupTask
         // is run.
         Action = nullptr;
       },
       std::move(Action), std::move(CleanupTask))
       .detach();
 }
+
+Deadline timeoutSeconds(llvm::Optional<double> Seconds) {
+  using namespace std::chrono;
+  if (!Seconds)
+    return llvm::None;
+  return steady_clock::now() +
+         duration_cast<steady_clock::duration>(duration<double>(*Seconds));
+}
+
 } // namespace clangd
 } // namespace clang

unittests/clangd/ClangdTests.cpp

 using ::testing::Eq;
 using ::testing::Gt;
 using ::testing::IsEmpty;
 using ::testing::Pair;
 using ::testing::UnorderedElementsAre;
 
 namespace {
 
-// Don't wait for async ops in clangd test more than that to avoid blocking
-// indefinitely in case of bugs.
-static const std::chrono::seconds DefaultFutureTimeout =
-    std::chrono::seconds(10);
-
 static bool diagsContainErrors(ArrayRef<DiagWithFixIts> Diagnostics) {
   for (const auto &DiagAndFixIts : Diagnostics) {
     // FIXME: severities returned by clangd should have a descriptive
     // diagnostic severity enum
     const int ErrorSeverity = 1;
     if (DiagAndFixIts.Diag.severity == ErrorSeverity)
       return true;
   }
std::string parseSourceAndDumpAST(
     for (const auto &FileWithContents : ExtraFiles)
       FS.Files[getVirtualTestFilePath(FileWithContents.first)] =
           FileWithContents.second;
 
     auto SourceFilename = getVirtualTestFilePath(SourceFileRelPath);
 
     FS.ExpectedFile = SourceFilename;
 
-    // Have to sync reparses because requests are processed on the calling
-    // thread.
-    auto AddDocFuture = Server.addDocument(SourceFilename, SourceContents);
-
+    Server.addDocument(SourceFilename, SourceContents);
     auto Result = dumpASTWithoutMemoryLocs(Server, SourceFilename);
-
-    // Wait for reparse to finish before checking for errors.
-    EXPECT_EQ(AddDocFuture.wait_for(DefaultFutureTimeout),
-              std::future_status::ready);
+    EXPECT_TRUE(Server.blockUntilIdleForTest()) << "Waiting for diagnostics";
     EXPECT_EQ(ExpectErrors, DiagConsumer.hadErrorInLastDiags());
     return Result;
   }
 };
 
 TEST_F(ClangdVFSTest, Parse) {
   // FIXME: figure out a stable format for AST dumps, so that we can check the
   // output of the dump itself is equal to the expected one, not just that it's
)cpp";
 
   auto FooCpp = getVirtualTestFilePath("foo.cpp");
   auto FooH = getVirtualTestFilePath("foo.h");
 
   FS.Files[FooH] = "int a;";
   FS.Files[FooCpp] = SourceContents;
   FS.ExpectedFile = FooCpp;
 
-  // To sync reparses before checking for errors.
-  std::future<void> ParseFuture;
-
-  ParseFuture = Server.addDocument(FooCpp, SourceContents);
+  Server.addDocument(FooCpp, SourceContents);
   auto DumpParse1 = dumpASTWithoutMemoryLocs(Server, FooCpp);
-  ASSERT_EQ(ParseFuture.wait_for(DefaultFutureTimeout),
-            std::future_status::ready);
+  ASSERT_TRUE(Server.blockUntilIdleForTest()) << "Waiting for diagnostics";
   EXPECT_FALSE(DiagConsumer.hadErrorInLastDiags());
 
-  ParseFuture = Server.addDocument(FooCpp, "");
+  Server.addDocument(FooCpp, "");
   auto DumpParseEmpty = dumpASTWithoutMemoryLocs(Server, FooCpp);
-  ASSERT_EQ(ParseFuture.wait_for(DefaultFutureTimeout),
-            std::future_status::ready);
+  ASSERT_TRUE(Server.blockUntilIdleForTest()) << "Waiting for diagnostics";
   EXPECT_FALSE(DiagConsumer.hadErrorInLastDiags());
 
-  ParseFuture = Server.addDocument(FooCpp, SourceContents);
+  Server.addDocument(FooCpp, SourceContents);
   auto DumpParse2 = dumpASTWithoutMemoryLocs(Server, FooCpp);
-  ASSERT_EQ(ParseFuture.wait_for(DefaultFutureTimeout),
-            std::future_status::ready);
+  ASSERT_TRUE(Server.blockUntilIdleForTest()) << "Waiting for diagnostics";
   EXPECT_FALSE(DiagConsumer.hadErrorInLastDiags());
 
   EXPECT_EQ(DumpParse1, DumpParse2);
   EXPECT_NE(DumpParse1, DumpParseEmpty);
 }
 
 TEST_F(ClangdVFSTest, ReparseOnHeaderChange) {
   MockFSProvider FS;
)cpp";
 
   auto FooCpp = getVirtualTestFilePath("foo.cpp");
   auto FooH = getVirtualTestFilePath("foo.h");
 
   FS.Files[FooH] = "int a;";
   FS.Files[FooCpp] = SourceContents;
   FS.ExpectedFile = FooCpp;
 
-  // To sync reparses before checking for errors.
-  std::future<void> ParseFuture;
-
-  ParseFuture = Server.addDocument(FooCpp, SourceContents);
+  Server.addDocument(FooCpp, SourceContents);
   auto DumpParse1 = dumpASTWithoutMemoryLocs(Server, FooCpp);
-  ASSERT_EQ(ParseFuture.wait_for(DefaultFutureTimeout),
-            std::future_status::ready);
+  ASSERT_TRUE(Server.blockUntilIdleForTest()) << "Waiting for diagnostics";
   EXPECT_FALSE(DiagConsumer.hadErrorInLastDiags());
 
   FS.Files[FooH] = "";
-  ParseFuture = Server.forceReparse(FooCpp);
+  Server.forceReparse(FooCpp);
   auto DumpParseDifferent = dumpASTWithoutMemoryLocs(Server, FooCpp);
-  ASSERT_EQ(ParseFuture.wait_for(DefaultFutureTimeout),
-            std::future_status::ready);
+  ASSERT_TRUE(Server.blockUntilIdleForTest()) << "Waiting for diagnostics";
   EXPECT_TRUE(DiagConsumer.hadErrorInLastDiags());
 
   FS.Files[FooH] = "int a;";
-  ParseFuture = Server.forceReparse(FooCpp);
+  Server.forceReparse(FooCpp);
   auto DumpParse2 = dumpASTWithoutMemoryLocs(Server, FooCpp);
-  EXPECT_EQ(ParseFuture.wait_for(DefaultFutureTimeout),
-            std::future_status::ready);
+  ASSERT_TRUE(Server.blockUntilIdleForTest()) << "Waiting for diagnostics";
   EXPECT_FALSE(DiagConsumer.hadErrorInLastDiags());
 
   EXPECT_EQ(DumpParse1, DumpParse2);
   EXPECT_NE(DumpParse1, DumpParseDifferent);
 }
 
 TEST_F(ClangdVFSTest, CheckVersions) {
   MockFSProvider FS;
TEST_F(ClangdVFSTest, CheckSourceHeaderSwitch) {
 
   // Test if asking for a corresponding file that doesn't exist returns an empty
   // string.
   PathResult = Server.switchSourceHeader(Invalid);
   EXPECT_FALSE(PathResult.hasValue());
 }
 
 TEST_F(ClangdThreadingTest, NoConcurrentDiagnostics) {
-  class NoConcurrentAccessDiagConsumer : public DiagnosticsConsumer {
+  struct NoConcurrentAccessDiagConsumer : public DiagnosticsConsumer {
   public:
+    std::atomic<int> Count = {0};
+
      NoConcurrentAccessDiagConsumer(std::promise<void> StartSecondReparse)
          : StartSecondReparse(std::move(StartSecondReparse)) {}
 
-    void onDiagnosticsReady(
-        PathRef File,
-        Tagged<std::vector<DiagWithFixIts>> Diagnostics) override {
-
+    void onDiagnosticsReady(PathRef,
+                            Tagged<std::vector<DiagWithFixIts>>) override {
+      ++Count;
       std::unique_lock<std::mutex> Lock(Mutex, std::try_to_lock_t());
       ASSERT_TRUE(Lock.owns_lock())
           << "Detected concurrent onDiagnosticsReady calls for the same file.";
+
+      // If we started the second parse immediately, it might cancel the first.
+      // So we don't allow it to start until the first has delivered diags...
       if (FirstRequest) {
         FirstRequest = false;
         StartSecondReparse.set_value();
-        // Sleep long enough for the second request to be processed.
+        // ... but then we wait long enough that the callbacks would overlap.
         std::this_thread::sleep_for(std::chrono::milliseconds(50));
       }
     }
 
   private:
     std::mutex Mutex;
     bool FirstRequest = true;
     std::promise<void> StartSecondReparse;
)cpp";
   const auto SourceContentsWithErrors = R"cpp(
 int a = x;
 int b;
 int c;
 int d;
 )cpp";
 
   auto FooCpp = getVirtualTestFilePath("foo.cpp");
-  llvm::StringMap<std::string> FileContents;
-  FileContents[FooCpp] = "";
-  ConstantFSProvider FS(buildTestFS(FileContents));
-
-  std::promise<void> StartSecondReparsePromise;
-  std::future<void> StartSecondReparse = StartSecondReparsePromise.get_future();
+  MockFSProvider FS;
+  FS.Files[FooCpp] = "";
 
-  NoConcurrentAccessDiagConsumer DiagConsumer(
-      std::move(StartSecondReparsePromise));
+  std::promise<void> StartSecondPromise;
+  std::future<void> StartSecond = StartSecondPromise.get_future();
 
+  NoConcurrentAccessDiagConsumer DiagConsumer(std::move(StartSecondPromise));
   MockCompilationDatabase CDB;
-  ClangdServer Server(CDB, DiagConsumer, FS, 4,
+  ClangdServer Server(CDB, DiagConsumer, FS, /*AsyncThreadsCount=*/4,
                       /*StorePreamblesInMemory=*/true);
   Server.addDocument(FooCpp, SourceContentsWithErrors);
-  StartSecondReparse.wait();
-
-  auto Future = Server.addDocument(FooCpp, SourceContentsWithoutErrors);
-  Future.wait();
+  StartSecond.wait();
+  Server.addDocument(FooCpp, SourceContentsWithoutErrors);
+  ASSERT_TRUE(Server.blockUntilIdleForTest()) << "Waiting for diagnostics";
+  ASSERT_EQ(DiagConsumer.Count, 2); // Sanity check - we actually ran both?
 }
 
 } // namespace clangd
 } // namespace clang

unittests/clangd/CodeCompleteTests.cpp

CompletionList completions(StringRef Text,
 
   MockFSProvider FS;
   MockCompilationDatabase CDB;
   IgnoreDiagnostics DiagConsumer;
   ClangdServer Server(CDB, DiagConsumer, FS, getDefaultAsyncThreadsCount(),
                       /*StorePreamblesInMemory=*/true);
   auto File = getVirtualTestFilePath("foo.cpp");
   Annotations Test(Text);
-  Server.addDocument(File, Test.code()).wait();
+  Server.addDocument(File, Test.code());
+  EXPECT_TRUE(Server.blockUntilIdleForTest()) << "Waiting for preamble";
   auto CompletionList = runCodeComplete(Server, File, Test.point(), Opts).Value;
   // Sanity-check that filterText is valid.
   EXPECT_THAT(CompletionList.items, Each(NameContainsFilter()));
   return CompletionList;
 }
 
 std::string replace(StringRef Haystack, StringRef Needle, StringRef Repl) {
   std::string Result;
FS.Files[getVirtualTestFilePath("bar.h")] =
       R"cpp(namespace ns { struct preamble { int member; }; })cpp";
   auto File = getVirtualTestFilePath("foo.cpp");
   Annotations Test(R"cpp(
       #include "bar.h"
       namespace ns { int local; }
       void f() { ns::^; }
       void f() { ns::preamble().$2^; }
   )cpp");
-  Server.addDocument(File, Test.code()).wait();
+  Server.addDocument(File, Test.code());
+  ASSERT_TRUE(Server.blockUntilIdleForTest()) << "Waiting for preamble";
   clangd::CodeCompleteOptions Opts = {};
 
-  auto WithoutIndex = runCodeComplete(Server, File, Test.point(), Opts).Value;
-  EXPECT_THAT(WithoutIndex.items,
-              UnorderedElementsAre(Named("local"), Named("preamble")));
-
   auto I = memIndex({var("ns::index")});
   Opts.Index = I.get();
   auto WithIndex = runCodeComplete(Server, File, Test.point(), Opts).Value;
   EXPECT_THAT(WithIndex.items,
               UnorderedElementsAre(Named("local"), Named("index")));
   auto ClassFromPreamble =
       runCodeComplete(Server, File, Test.point("2"), Opts).Value;
   EXPECT_THAT(ClassFromPreamble.items, Contains(Named("member")));
+
+  Opts.Index = nullptr;
+  auto WithoutIndex = runCodeComplete(Server, File, Test.point(), Opts).Value;
+  EXPECT_THAT(WithoutIndex.items,
+              UnorderedElementsAre(Named("local"), Named("preamble")));
+
 }
 
 TEST(CompletionTest, DynamicIndexMultiFile) {
   MockFSProvider FS;
   MockCompilationDatabase CDB;
   IgnoreDiagnostics DiagConsumer;
   ClangdServer Server(CDB, DiagConsumer, FS, getDefaultAsyncThreadsCount(),
                       /*StorePreamblesInMemory=*/true,
                       /*BuildDynamicSymbolIndex=*/true);
 
-  Server
-      .addDocument(getVirtualTestFilePath("foo.cpp"), R"cpp(
+  Server.addDocument(getVirtualTestFilePath("foo.cpp"), R"cpp(
       namespace ns { class XYZ {}; void foo(int x) {} }
-  )cpp")
-      .wait();
+  )cpp");
+  ASSERT_TRUE(Server.blockUntilIdleForTest()) << "Waiting for preamble";
 
   auto File = getVirtualTestFilePath("bar.cpp");
   Annotations Test(R"cpp(
       namespace ns {
       class XXX {};
       /// Doooc
       void fooooo() {}
       }
       void f() { ns::^ }
   )cpp");
-  Server.addDocument(File, Test.code()).wait();
+  Server.addDocument(File, Test.code());
+  ASSERT_TRUE(Server.blockUntilIdleForTest()) << "Waiting for preamble";
 
   auto Results = runCodeComplete(Server, File, Test.point(), {}).Value;
   // "XYZ" and "foo" are not included in the file being completed but are still
   // visible through the index.
   EXPECT_THAT(Results.items, Has("XYZ", CompletionItemKind::Class));
   EXPECT_THAT(Results.items, Has("foo", CompletionItemKind::Function));
   EXPECT_THAT(Results.items, Has("XXX", CompletionItemKind::Class));
   EXPECT_THAT(Results.items, Contains(AllOf(Named("fooooo"), Filter("fooooo"),
SignatureHelp signatures(StringRef Text) {
   MockFSProvider FS;
   MockCompilationDatabase CDB;
   IgnoreDiagnostics DiagConsumer;
   ClangdServer Server(CDB, DiagConsumer, FS, getDefaultAsyncThreadsCount(),
                       /*StorePreamblesInMemory=*/true);
   auto File = getVirtualTestFilePath("foo.cpp");
   Annotations Test(Text);
   Server.addDocument(File, Test.code());
+  EXPECT_TRUE(Server.blockUntilIdleForTest()) << "Waiting for preamble";
   auto R = Server.signatureHelp(File, Test.point());
   assert(R);
   return R.get().Value;
 }
 
 MATCHER_P(ParamsAre, P, "") {
   if (P.size() != arg.parameters.size())
     return false;

unittests/clangd/ThreadingTests.cpp

int Counter(0); /* GUARDED_BY(Mutex) */
 
     {
       // Make sure runAsync is not running tasks synchronously on the same
       // thread by locking the Mutex used for increments.
       std::lock_guard<std::mutex> Lock(Mutex);
       scheduleIncrements();
     }
 
-    Tasks.waitForAll();
+    Tasks.wait();
     {
       std::lock_guard<std::mutex> Lock(Mutex);
       ASSERT_EQ(Counter, TasksCnt * IncrementsPerTask);
     }
 
     {
       std::lock_guard<std::mutex> Lock(Mutex);
       Counter = 0;