[clangd] Use multiple working threads in clangd.

Reviewers: bkramer, krasimir, klimek

Reviewed By: klimek

Subscribers: arphaman, cfe-commits

Differential Revision: https://reviews.llvm.org/D36261

llvm-svn: 310821

clang-tools-extra/clangd/ClangdLSPServer.cpp

@@ -220,10 +220,10 @@ void ClangdLSPServer::LSPProtocolCallbacks::onGoToDefinition(
       R"(,"result":[)" + Locations + R"(]})");
 }
 
-ClangdLSPServer::ClangdLSPServer(JSONOutput &Out, bool RunSynchronously,
+ClangdLSPServer::ClangdLSPServer(JSONOutput &Out, unsigned AsyncThreadsCount,
                                  llvm::Optional<StringRef> ResourceDir)
     : Out(Out), DiagConsumer(*this),
-      Server(CDB, DiagConsumer, FSProvider, RunSynchronously, ResourceDir) {}
+      Server(CDB, DiagConsumer, FSProvider, AsyncThreadsCount, ResourceDir) {}
 
 void ClangdLSPServer::run(std::istream &In) {
   assert(!IsDone && "Run was called before");

clang-tools-extra/clangd/ClangdLSPServer.h

@@ -26,7 +26,7 @@ class JSONOutput;
 /// dispatch and ClangdServer together.
 class ClangdLSPServer {
 public:
-  ClangdLSPServer(JSONOutput &Out, bool RunSynchronously,
+  ClangdLSPServer(JSONOutput &Out, unsigned AsyncThreadsCount,
                   llvm::Optional<StringRef> ResourceDir);
 
   /// Run LSP server loop, receiving input for it from \p In. \p In must be

clang-tools-extra/clangd/ClangdServer.cpp

@@ -78,40 +78,52 @@ RealFileSystemProvider::getTaggedFileSystem(PathRef File) {
   return make_tagged(vfs::getRealFileSystem(), VFSTag());
 }
 
-ClangdScheduler::ClangdScheduler(bool RunSynchronously)
-    : RunSynchronously(RunSynchronously) {
+unsigned clangd::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;
+}
+
+ClangdScheduler::ClangdScheduler(unsigned AsyncThreadsCount)
+    : RunSynchronously(AsyncThreadsCount == 0) {
   if (RunSynchronously) {
     // Don't start the worker thread if we're running synchronously
     return;
   }
 
-  // Initialize Worker in ctor body, rather than init list to avoid potentially
-  // using not-yet-initialized members
-  Worker = std::thread([this]() {
-    while (true) {
-      std::future<void> Request;
+  Workers.reserve(AsyncThreadsCount);
+  for (unsigned I = 0; I < AsyncThreadsCount; ++I) {
+    Workers.push_back(std::thread([this]() {
+      while (true) {
+        std::future<void> Request;
 
-      // Pick request from the queue
-      {
-        std::unique_lock<std::mutex> Lock(Mutex);
-        // Wait for more requests.
-        RequestCV.wait(Lock, [this] { return !RequestQueue.empty() || Done; });
-        if (Done)
-          return;
+        // Pick request from the queue
+        {
+          std::unique_lock<std::mutex> Lock(Mutex);
+          // Wait for more requests.
+          RequestCV.wait(Lock,
+                         [this] { return !RequestQueue.empty() || Done; });
+          if (Done)
+            return;
 
-        assert(!RequestQueue.empty() && "RequestQueue was empty");
+          assert(!RequestQueue.empty() && "RequestQueue was empty");
 
-        // We process requests starting from the front of the queue. Users of
-        // ClangdScheduler have a way to prioritise their requests by putting
-        // them to the either side of the queue (using either addToEnd or
-        // addToFront).
-        Request = std::move(RequestQueue.front());
-        RequestQueue.pop_front();
-      } // unlock Mutex
+          // We process requests starting from the front of the queue. Users of
+          // ClangdScheduler have a way to prioritise their requests by putting
+          // them to the either side of the queue (using either addToEnd or
+          // addToFront).
+          Request = std::move(RequestQueue.front());
+          RequestQueue.pop_front();
+        } // unlock Mutex
 
-      Request.get();
-    }
-  });
+        Request.get();
+      }
+    }));
+  }
 }
 
 ClangdScheduler::~ClangdScheduler() {
@@ -123,19 +135,21 @@ ClangdScheduler::~ClangdScheduler() {
     // Wake up the worker thread
     Done = true;
   } // unlock Mutex
-  RequestCV.notify_one();
-  Worker.join();
+  RequestCV.notify_all();
+
+  for (auto &Worker : Workers)
+    Worker.join();
 }
 
 ClangdServer::ClangdServer(GlobalCompilationDatabase &CDB,
                            DiagnosticsConsumer &DiagConsumer,
                            FileSystemProvider &FSProvider,
-                           bool RunSynchronously,
+                           unsigned AsyncThreadsCount,
                            llvm::Optional<StringRef> ResourceDir)
     : CDB(CDB), DiagConsumer(DiagConsumer), FSProvider(FSProvider),
       ResourceDir(ResourceDir ? ResourceDir->str() : getStandardResourceDir()),
       PCHs(std::make_shared<PCHContainerOperations>()),
-      WorkScheduler(RunSynchronously) {}
+      WorkScheduler(AsyncThreadsCount) {}
 
 std::future<void> ClangdServer::addDocument(PathRef File, StringRef Contents) {
   DocVersion Version = DraftMgr.updateDraft(File, Contents);

clang-tools-extra/clangd/ClangdServer.h

@@ -101,11 +101,20 @@ public:
 
 class ClangdServer;
 
-/// Handles running WorkerRequests of ClangdServer on a separate threads.
-/// Currently runs only one worker thread.
+/// Returns a number of a default async threads to use for ClangdScheduler.
+/// Returned value is always >= 1 (i.e. will not cause requests to be processed
+/// synchronously).
+unsigned getDefaultAsyncThreadsCount();
+
+/// Handles running WorkerRequests of ClangdServer on a number of worker
+/// threads.
 class ClangdScheduler {
 public:
-  ClangdScheduler(bool RunSynchronously);
+  /// If \p AsyncThreadsCount is 0, requests added using addToFront and addToEnd
+  /// will be processed synchronously on the calling thread.
+  // Otherwise, \p AsyncThreadsCount threads will be created to schedule the
+  // requests.
+  ClangdScheduler(unsigned AsyncThreadsCount);
   ~ClangdScheduler();
 
   /// Add a new request to run function \p F with args \p As to the start of the
@@ -146,17 +155,16 @@ public:
 private:
   bool RunSynchronously;
   std::mutex Mutex;
-  /// We run some tasks on a separate threads(parsing, CppFile cleanup).
-  /// This thread looks into RequestQueue to find requests to handle and
-  /// terminates when Done is set to true.
-  std::thread Worker;
-  /// Setting Done to true will make the worker thread terminate.
+  /// We run some tasks on separate threads(parsing, CppFile cleanup).
+  /// These threads looks into RequestQueue to find requests to handle and
+  /// terminate when Done is set to true.
+  std::vector<std::thread> Workers;
+  /// Setting Done to true will make the worker threads terminate.
   bool Done = false;
-  /// A queue of requests.
-  /// FIXME(krasimir): code completion should always have priority over parsing
-  /// for diagnostics.
+  /// A queue of requests. Elements of this vector are async computations (i.e.
+  /// results of calling std::async(std::launch::deferred, ...)).
   std::deque<std::future<void>> RequestQueue;
-  /// Condition variable to wake up the worker thread.
+  /// Condition variable to wake up worker threads.
   std::condition_variable RequestCV;
 };
 
@@ -165,22 +173,19 @@ private:
 /// diagnostics for tracked files).
 class ClangdServer {
 public:
-  /// Creates a new ClangdServer. If \p RunSynchronously is false, no worker
-  /// thread will be created and all requests will be completed synchronously on
-  /// the calling thread (this is mostly used for tests). If \p RunSynchronously
-  /// is true, a worker thread will be created to parse files in the background
-  /// and provide diagnostics results via DiagConsumer.onDiagnosticsReady
-  /// callback. File accesses for each instance of parsing will be conducted via
-  /// a vfs::FileSystem provided by \p FSProvider. Results of code
-  /// completion/diagnostics also include a tag, that \p FSProvider returns
-  /// along with the vfs::FileSystem.
-  /// When \p ResourceDir is set, it will be used to search for internal headers
+  /// Creates a new ClangdServer. To server parsing requests ClangdScheduler,
+  /// that spawns \p AsyncThreadsCount worker threads will be created (when \p
+  /// AsyncThreadsCount is 0, requests will be processed on the calling thread).
+  /// instance of parsing will be conducted via a vfs::FileSystem provided by \p
+  /// FSProvider. Results of code completion/diagnostics also include a tag,
+  /// that \p FSProvider returns along with the vfs::FileSystem. When \p
+  /// ResourceDir is set, it will be used to search for internal headers
   /// (overriding defaults and -resource-dir compiler flag, if set). If \p
   /// ResourceDir is None, ClangdServer will attempt to set it to a standard
   /// location, obtained via CompilerInvocation::GetResourcePath.
   ClangdServer(GlobalCompilationDatabase &CDB,
                DiagnosticsConsumer &DiagConsumer,
-               FileSystemProvider &FSProvider, bool RunSynchronously,
+               FileSystemProvider &FSProvider, unsigned AsyncThreadsCount,
                llvm::Optional<StringRef> ResourceDir = llvm::None);
 
   /// Add a \p File to the list of tracked C++ files or update the contents if

clang-tools-extra/clangd/tool/ClangdMain.cpp

@@ -16,14 +16,20 @@
 #include <iostream>
 #include <memory>
 #include <string>
+#include <thread>
 
 using namespace clang;
 using namespace clang::clangd;
 
-static llvm::cl::opt<bool>
-    RunSynchronously("run-synchronously",
-                     llvm::cl::desc("Parse on main thread"),
-                     llvm::cl::init(false), llvm::cl::Hidden);
+static llvm::cl::opt<unsigned>
+    WorkerThreadsCount("j",
+                       llvm::cl::desc("Number of async workers used by clangd"),
+                       llvm::cl::init(getDefaultAsyncThreadsCount()));
+
+static llvm::cl::opt<bool> RunSynchronously(
+    "run-synchronously",
+    llvm::cl::desc("Parse on main thread. If set, -j is ignored"),
+    llvm::cl::init(false), llvm::cl::Hidden);
 
 static llvm::cl::opt<std::string>
     ResourceDir("resource-dir",
@@ -33,6 +39,17 @@ static llvm::cl::opt<std::string>
 int main(int argc, char *argv[]) {
   llvm::cl::ParseCommandLineOptions(argc, argv, "clangd");
 
+  if (!RunSynchronously && WorkerThreadsCount == 0) {
+    llvm::errs() << "A number of worker threads cannot be 0. Did you mean to "
+                    "specify -run-synchronously?";
+    return 1;
+  }
+
+  // Ignore -j option if -run-synchonously is used.
+  // FIXME: a warning should be shown here.
+  if (RunSynchronously)
+    WorkerThreadsCount = 0;
+
   llvm::raw_ostream &Outs = llvm::outs();
   llvm::raw_ostream &Logs = llvm::errs();
   JSONOutput Out(Outs, Logs);
@@ -43,6 +60,7 @@ int main(int argc, char *argv[]) {
   llvm::Optional<StringRef> ResourceDirRef = None;
   if (!ResourceDir.empty())
     ResourceDirRef = ResourceDir;
-  ClangdLSPServer LSPServer(Out, RunSynchronously, ResourceDirRef);
+
+  ClangdLSPServer LSPServer(Out, WorkerThreadsCount, ResourceDirRef);
   LSPServer.run(std::cin);
 }

clang-tools-extra/unittests/clangd/ClangdTests.cpp

@@ -19,6 +19,7 @@
 #include <algorithm>
 #include <chrono>
 #include <iostream>
+#include <random>
 #include <string>
 #include <vector>
 
@@ -136,18 +137,23 @@ namespace {
 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;
+  }
+  return false;
+}
+
 class ErrorCheckingDiagConsumer : public DiagnosticsConsumer {
 public:
   void
   onDiagnosticsReady(PathRef File,
                      Tagged<std::vector<DiagWithFixIts>> Diagnostics) override {
-    bool HadError = false;
-    for (const auto &DiagAndFixIts : Diagnostics.Value) {
-      // FIXME: severities returned by clangd should have a descriptive
-      // diagnostic severity enum
-      const int ErrorSeverity = 1;
-      HadError = DiagAndFixIts.Diag.severity == ErrorSeverity;
-    }
+    bool HadError = diagsContainErrors(Diagnostics.Value);
 
     std::lock_guard<std::mutex> Lock(Mutex);
     HadErrorInLastDiags = HadError;
@@ -196,24 +202,46 @@ public:
   std::vector<std::string> ExtraClangFlags;
 };
 
+IntrusiveRefCntPtr<vfs::FileSystem>
+buildTestFS(llvm::StringMap<std::string> const &Files) {
+  IntrusiveRefCntPtr<vfs::InMemoryFileSystem> MemFS(
+      new vfs::InMemoryFileSystem);
+  for (auto &FileAndContents : Files)
+    MemFS->addFile(FileAndContents.first(), time_t(),
+                   llvm::MemoryBuffer::getMemBuffer(FileAndContents.second,
+                                                    FileAndContents.first()));
+
+  auto OverlayFS = IntrusiveRefCntPtr<vfs::OverlayFileSystem>(
+      new vfs::OverlayFileSystem(vfs::getTempOnlyFS()));
+  OverlayFS->pushOverlay(std::move(MemFS));
+  return OverlayFS;
+}
+
+class ConstantFSProvider : public FileSystemProvider {
+public:
+  ConstantFSProvider(IntrusiveRefCntPtr<vfs::FileSystem> FS,
+                     VFSTag Tag = VFSTag())
+      : FS(std::move(FS)), Tag(std::move(Tag)) {}
+
+  Tagged<IntrusiveRefCntPtr<vfs::FileSystem>>
+  getTaggedFileSystem(PathRef File) override {
+    return make_tagged(FS, Tag);
+  }
+
+private:
+  IntrusiveRefCntPtr<vfs::FileSystem> FS;
+  VFSTag Tag;
+};
+
 class MockFSProvider : public FileSystemProvider {
 public:
   Tagged<IntrusiveRefCntPtr<vfs::FileSystem>>
   getTaggedFileSystem(PathRef File) override {
-    IntrusiveRefCntPtr<vfs::InMemoryFileSystem> MemFS(
-        new vfs::InMemoryFileSystem);
     if (ExpectedFile)
       EXPECT_EQ(*ExpectedFile, File);
 
-    for (auto &FileAndContents : Files)
-      MemFS->addFile(FileAndContents.first(), time_t(),
-                     llvm::MemoryBuffer::getMemBuffer(FileAndContents.second,
-                                                      FileAndContents.first()));
-
-    auto OverlayFS = IntrusiveRefCntPtr<vfs::OverlayFileSystem>(
-        new vfs::OverlayFileSystem(vfs::getTempOnlyFS()));
-    OverlayFS->pushOverlay(std::move(MemFS));
-    return make_tagged(OverlayFS, Tag);
+    auto FS = buildTestFS(Files);
+    return make_tagged(FS, Tag);
   }
 
   llvm::Optional<SmallString<32>> ExpectedFile;
@@ -272,8 +300,7 @@ protected:
     MockFSProvider FS;
     ErrorCheckingDiagConsumer DiagConsumer;
     MockCompilationDatabase CDB(/*AddFreestandingFlag=*/true);
-    ClangdServer Server(CDB, DiagConsumer, FS,
-                        /*RunSynchronously=*/false);
+    ClangdServer Server(CDB, DiagConsumer, FS, getDefaultAsyncThreadsCount());
     for (const auto &FileWithContents : ExtraFiles)
       FS.Files[getVirtualTestFilePath(FileWithContents.first)] =
           FileWithContents.second;
@@ -282,7 +309,8 @@ protected:
 
     FS.ExpectedFile = SourceFilename;
 
-    // Have to sync reparses because RunSynchronously is false.
+    // Have to sync reparses because requests are processed on the calling
+    // thread.
     auto AddDocFuture = Server.addDocument(SourceFilename, SourceContents);
 
     auto Result = dumpASTWithoutMemoryLocs(Server, SourceFilename);
@@ -334,8 +362,7 @@ TEST_F(ClangdVFSTest, Reparse) {
   MockFSProvider FS;
   ErrorCheckingDiagConsumer DiagConsumer;
   MockCompilationDatabase CDB(/*AddFreestandingFlag=*/true);
-  ClangdServer Server(CDB, DiagConsumer, FS,
-                      /*RunSynchronously=*/false);
+  ClangdServer Server(CDB, DiagConsumer, FS, getDefaultAsyncThreadsCount());
 
   const auto SourceContents = R"cpp(
 #include "foo.h"
@@ -379,8 +406,7 @@ TEST_F(ClangdVFSTest, ReparseOnHeaderChange) {
   ErrorCheckingDiagConsumer DiagConsumer;
   MockCompilationDatabase CDB(/*AddFreestandingFlag=*/true);
 
-  ClangdServer Server(CDB, DiagConsumer, FS,
-                      /*RunSynchronously=*/false);
+  ClangdServer Server(CDB, DiagConsumer, FS, getDefaultAsyncThreadsCount());
 
   const auto SourceContents = R"cpp(
 #include "foo.h"
@@ -425,16 +451,17 @@ TEST_F(ClangdVFSTest, CheckVersions) {
   MockFSProvider FS;
   ErrorCheckingDiagConsumer DiagConsumer;
   MockCompilationDatabase CDB(/*AddFreestandingFlag=*/true);
+  // Run ClangdServer synchronously.
   ClangdServer Server(CDB, DiagConsumer, FS,
-                      /*RunSynchronously=*/true);
+                      /*AsyncThreadsCount=*/0);
 
   auto FooCpp = getVirtualTestFilePath("foo.cpp");
   const auto SourceContents = "int a;";
   FS.Files[FooCpp] = SourceContents;
   FS.ExpectedFile = FooCpp;
 
-  // No need to sync reparses, because RunSynchronously is set
-  // to true.
+  // No need to sync reparses, because requests are processed on the calling
+  // thread.
   FS.Tag = "123";
   Server.addDocument(FooCpp, SourceContents);
   EXPECT_EQ(Server.codeComplete(FooCpp, Position{0, 0}).Tag, FS.Tag);
@@ -457,8 +484,9 @@ TEST_F(ClangdVFSTest, SearchLibDir) {
                              {"-xc++", "-target", "x86_64-linux-unknown",
                               "-m64", "--gcc-toolchain=/randomusr",
                               "-stdlib=libstdc++"});
+  // Run ClangdServer synchronously.
   ClangdServer Server(CDB, DiagConsumer, FS,
-                      /*RunSynchronously=*/true);
+                      /*AsyncThreadsCount=*/0);
 
   // Just a random gcc version string
   SmallString<8> Version("4.9.3");
@@ -487,8 +515,8 @@ mock_string x;
 )cpp";
   FS.Files[FooCpp] = SourceContents;
 
-  // No need to sync reparses, because RunSynchronously is set
-  // to true.
+  // No need to sync reparses, because requests are processed on the calling
+  // thread.
   Server.addDocument(FooCpp, SourceContents);
   EXPECT_FALSE(DiagConsumer.hadErrorInLastDiags());
 
@@ -506,9 +534,9 @@ TEST_F(ClangdVFSTest, ForceReparseCompileCommand) {
   ErrorCheckingDiagConsumer DiagConsumer;
   MockCompilationDatabase CDB(/*AddFreestandingFlag=*/true);
   ClangdServer Server(CDB, DiagConsumer, FS,
-                      /*RunSynchronously=*/true);
-  // No need to sync reparses, because RunSynchronously is set
-  // to true.
+                      /*AsyncThreadsCount=*/0);
+  // No need to sync reparses, because reparses are performed on the calling
+  // thread to true.
 
   auto FooCpp = getVirtualTestFilePath("foo.cpp");
   const auto SourceContents1 = R"cpp(
@@ -564,8 +592,7 @@ TEST_F(ClangdCompletionTest, CheckContentsOverride) {
   ErrorCheckingDiagConsumer DiagConsumer;
   MockCompilationDatabase CDB(/*AddFreestandingFlag=*/true);
 
-  ClangdServer Server(CDB, DiagConsumer, FS,
-                      /*RunSynchronously=*/false);
+  ClangdServer Server(CDB, DiagConsumer, FS, getDefaultAsyncThreadsCount());
 
   auto FooCpp = getVirtualTestFilePath("foo.cpp");
   const auto SourceContents = R"cpp(
@@ -615,5 +642,248 @@ int b =   ;
   }
 }
 
+class ClangdThreadingTest : public ClangdVFSTest {};
+
+TEST_F(ClangdThreadingTest, StressTest) {
+  // Without 'static' clang gives an error for a usage inside TestDiagConsumer.
+  static const unsigned FilesCount = 5;
+  const unsigned RequestsCount = 500;
+  // Blocking requests wait for the parsing to complete, they slow down the test
+  // dramatically, so they are issued rarely. Each
+  // BlockingRequestInterval-request will be a blocking one.
+  const unsigned BlockingRequestInterval = 40;
+
+  const auto SourceContentsWithoutErrors = R"cpp(
+int a;
+int b;
+int c;
+int d;
+)cpp";
+
+  const auto SourceContentsWithErrors = R"cpp(
+int a = x;
+int b;
+int c;
+int d;
+)cpp";
+
+  // Giving invalid line and column number should not crash ClangdServer, but
+  // just to make sure we're sometimes hitting the bounds inside the file we
+  // limit the intervals of line and column number that are generated.
+  unsigned MaxLineForFileRequests = 7;
+  unsigned MaxColumnForFileRequests = 10;
+
+  std::vector<SmallString<32>> FilePaths;
+  FilePaths.reserve(FilesCount);
+  for (unsigned I = 0; I < FilesCount; ++I)
+    FilePaths.push_back(getVirtualTestFilePath(std::string("Foo") +
+                                               std::to_string(I) + ".cpp"));
+  // Mark all of those files as existing.
+  llvm::StringMap<std::string> FileContents;
+  for (auto &&FilePath : FilePaths)
+    FileContents[FilePath] = "";
+
+  ConstantFSProvider FS(buildTestFS(FileContents));
+
+  struct FileStat {
+    unsigned HitsWithoutErrors = 0;
+    unsigned HitsWithErrors = 0;
+    bool HadErrorsInLastDiags = false;
+  };
+
+  class TestDiagConsumer : public DiagnosticsConsumer {
+  public:
+    TestDiagConsumer() : Stats(FilesCount, FileStat()) {}
+
+    void onDiagnosticsReady(
+        PathRef File,
+        Tagged<std::vector<DiagWithFixIts>> Diagnostics) override {
+      StringRef FileIndexStr = llvm::sys::path::stem(File);
+      ASSERT_TRUE(FileIndexStr.consume_front("Foo"));
+
+      unsigned long FileIndex = std::stoul(FileIndexStr.str());
+
+      bool HadError = diagsContainErrors(Diagnostics.Value);
+
+      std::lock_guard<std::mutex> Lock(Mutex);
+      if (HadError)
+        Stats[FileIndex].HitsWithErrors++;
+      else
+        Stats[FileIndex].HitsWithoutErrors++;
+      Stats[FileIndex].HadErrorsInLastDiags = HadError;
+    }
+
+    std::vector<FileStat> takeFileStats() {
+      std::lock_guard<std::mutex> Lock(Mutex);
+      return std::move(Stats);
+    }
+
+  private:
+    std::mutex Mutex;
+    std::vector<FileStat> Stats;
+  };
+
+  struct RequestStats {
+    unsigned RequestsWithoutErrors = 0;
+    unsigned RequestsWithErrors = 0;
+    bool LastContentsHadErrors = false;
+    bool FileIsRemoved = true;
+    std::future<void> LastRequestFuture;
+  };
+
+  std::vector<RequestStats> ReqStats;
+  ReqStats.reserve(FilesCount);
+  for (unsigned FileIndex = 0; FileIndex < FilesCount; ++FileIndex)
+    ReqStats.emplace_back();
+
+  TestDiagConsumer DiagConsumer;
+  {
+    MockCompilationDatabase CDB(/*AddFreestandingFlag=*/true);
+    ClangdServer Server(CDB, DiagConsumer, FS, getDefaultAsyncThreadsCount());
+
+    // Prepare some random distributions for the test.
+    std::random_device RandGen;
+
+    std::uniform_int_distribution<unsigned> FileIndexDist(0, FilesCount - 1);
+    // Pass a text that contains compiler errors to addDocument in about 20% of
+    // all requests.
+    std::bernoulli_distribution ShouldHaveErrorsDist(0.2);
+    // Line and Column numbers for requests that need them.
+    std::uniform_int_distribution<int> LineDist(0, MaxLineForFileRequests);
+    std::uniform_int_distribution<int> ColumnDist(0, MaxColumnForFileRequests);
+
+    // Some helpers.
+    auto UpdateStatsOnAddDocument = [&](unsigned FileIndex, bool HadErrors,
+                                        std::future<void> Future) {
+      auto &Stats = ReqStats[FileIndex];
+
+      if (HadErrors)
+        ++Stats.RequestsWithErrors;
+      else
+        ++Stats.RequestsWithoutErrors;
+      Stats.LastContentsHadErrors = HadErrors;
+      Stats.FileIsRemoved = false;
+      Stats.LastRequestFuture = std::move(Future);
+    };
+
+    auto UpdateStatsOnRemoveDocument = [&](unsigned FileIndex,
+                                           std::future<void> Future) {
+      auto &Stats = ReqStats[FileIndex];
+
+      Stats.FileIsRemoved = true;
+      Stats.LastRequestFuture = std::move(Future);
+    };
+
+    auto UpdateStatsOnForceReparse = [&](unsigned FileIndex,
+                                         std::future<void> Future) {
+      auto &Stats = ReqStats[FileIndex];
+
+      Stats.LastRequestFuture = std::move(Future);
+      if (Stats.LastContentsHadErrors)
+        ++Stats.RequestsWithErrors;
+      else
+        ++Stats.RequestsWithoutErrors;
+    };
+
+    auto AddDocument = [&](unsigned FileIndex) {
+      bool ShouldHaveErrors = ShouldHaveErrorsDist(RandGen);
+      auto Future = Server.addDocument(
+          FilePaths[FileIndex], ShouldHaveErrors ? SourceContentsWithErrors
+                                                 : SourceContentsWithoutErrors);
+      UpdateStatsOnAddDocument(FileIndex, ShouldHaveErrors, std::move(Future));
+    };
+
+    // Various requests that we would randomly run.
+    auto AddDocumentRequest = [&]() {
+      unsigned FileIndex = FileIndexDist(RandGen);
+      AddDocument(FileIndex);
+    };
+
+    auto ForceReparseRequest = [&]() {
+      unsigned FileIndex = FileIndexDist(RandGen);
+      // Make sure we don't violate the ClangdServer's contract.
+      if (ReqStats[FileIndex].FileIsRemoved)
+        AddDocument(FileIndex);
+
+      auto Future = Server.forceReparse(FilePaths[FileIndex]);
+      UpdateStatsOnForceReparse(FileIndex, std::move(Future));
+    };
+
+    auto RemoveDocumentRequest = [&]() {
+      unsigned FileIndex = FileIndexDist(RandGen);
+      // Make sure we don't violate the ClangdServer's contract.
+      if (ReqStats[FileIndex].FileIsRemoved)
+        AddDocument(FileIndex);
+
+      auto Future = Server.removeDocument(FilePaths[FileIndex]);
+      UpdateStatsOnRemoveDocument(FileIndex, std::move(Future));
+    };
+
+    auto CodeCompletionRequest = [&]() {
+      unsigned FileIndex = FileIndexDist(RandGen);
+      // Make sure we don't violate the ClangdServer's contract.
+      if (ReqStats[FileIndex].FileIsRemoved)
+        AddDocument(FileIndex);
+
+      Position Pos{LineDist(RandGen), ColumnDist(RandGen)};
+      Server.codeComplete(FilePaths[FileIndex], Pos);
+    };
+
+    auto FindDefinitionsRequest = [&]() {
+      unsigned FileIndex = FileIndexDist(RandGen);
+      // Make sure we don't violate the ClangdServer's contract.
+      if (ReqStats[FileIndex].FileIsRemoved)
+        AddDocument(FileIndex);
+
+      Position Pos{LineDist(RandGen), ColumnDist(RandGen)};
+      Server.findDefinitions(FilePaths[FileIndex], Pos);
+    };
+
+    std::vector<std::function<void()>> AsyncRequests = {
+        AddDocumentRequest, ForceReparseRequest, RemoveDocumentRequest};
+    std::vector<std::function<void()>> BlockingRequests = {
+        CodeCompletionRequest, FindDefinitionsRequest};
+
+    // Bash requests to ClangdServer in a loop.
+    std::uniform_int_distribution<int> AsyncRequestIndexDist(
+        0, AsyncRequests.size() - 1);
+    std::uniform_int_distribution<int> BlockingRequestIndexDist(
+        0, BlockingRequests.size() - 1);
+    for (unsigned I = 1; I <= RequestsCount; ++I) {
+      if (I % BlockingRequestInterval != 0) {
+        // Issue an async request most of the time. It should be fast.
+        unsigned RequestIndex = AsyncRequestIndexDist(RandGen);
+        AsyncRequests[RequestIndex]();
+      } else {
+        // Issue a blocking request once in a while.
+        auto RequestIndex = BlockingRequestIndexDist(RandGen);
+        BlockingRequests[RequestIndex]();
+      }
+    }
+
+    // Wait for last requests to finish.
+    for (auto &ReqStat : ReqStats) {
+      if (!ReqStat.LastRequestFuture.valid())
+        continue; // We never ran any requests for this file.
+
+      // Future should be ready much earlier than in 5 seconds, the timeout is
+      // there to check we won't wait indefinitely.
+      ASSERT_EQ(ReqStat.LastRequestFuture.wait_for(std::chrono::seconds(5)),
+                std::future_status::ready);
+    }
+  } // Wait for ClangdServer to shutdown before proceeding.
+
+  // Check some invariants about the state of the program.
+  std::vector<FileStat> Stats = DiagConsumer.takeFileStats();
+  for (unsigned I = 0; I < FilesCount; ++I) {
+    if (!ReqStats[I].FileIsRemoved)
+      ASSERT_EQ(Stats[I].HadErrorsInLastDiags,
+                ReqStats[I].LastContentsHadErrors);
+
+    ASSERT_LE(Stats[I].HitsWithErrors, ReqStats[I].RequestsWithErrors);
+    ASSERT_LE(Stats[I].HitsWithoutErrors, ReqStats[I].RequestsWithoutErrors);
+  }
+}
+
 } // namespace clangd
 } // namespace clang