[clangd] Use multiple working threads in clangd.

[arphaman]

clangd/ClangdServer.h

107

Typo: immediately

[klimek]

clangd/ClangdServer.h

105

I think calling it "Options" is more idiomatic.

110

Typo: Indicates *that*.

121–122

Why not: 1 -> run synchronously, > 1, run in parallel?

[ilya-biryukov]

Addressed review comments.

• Fixed typos.
• Renamed SchedulingParams to SchedulingOptions.

clangd/ClangdServer.h

121–122

Currently 1 means: start 1 worker thread to run async operations (that thread is separate from the main thread).
This makes sense for clangd, as if you do that, you still get code completion that doesn't wait for diagnostics to finish.
On the other hand, it's useful to have -run-synchronously for some tests.

[krasimir]

Great!

clangd/ClangdServer.h

131

typo: "on separate threads"

176

typo: "on separate threads"

clangd/tool/ClangdMain.cpp

24

Extract this somewhere when it can be reused both by this and in clangd/ClangdServer.cpp:90.

69

Consider inverting this condition.

[ilya-biryukov]

• Removed SchedulingOptions altogether, replaced with AsyncThreadsCount.

[ilya-biryukov]

• Fixed more typos/inconsistences in comments, pointed out by @krasimir.

clangd/tool/ClangdMain.cpp

69

Thanks, good point. I also like non-negated conditions more.
That code was removed in the updated version, though.

[ilya-biryukov]

clangd/ClangdServer.h

121–122

As discussed, replaced with unsigned AsyncThreadsCount. Zero means run synchronously.

[klimek]

High-level question: Why can't we use llvm::ThreadPool?

[ilya-biryukov]

In D36261#834902, @klimek wrote:

High-level question: Why can't we use llvm::ThreadPool?

It is not an in-place replacement as it does not allow to prioritize new tasks over old ones (new tasks are usually more important for clangd as the old ones are often outdated when new ones are added).
I looked into using llvm::ThreadPool before, but decided to stay with our own implementation.
It gives us more flexibility over threading and it is not hard to maintain (I think it's under 100 lines of code, and the code is rather simple).

[ilya-biryukov]

• Added a stress test for multithreading.

[ilya-biryukov]

Added a test that exercises some multi-threading behaviour.
It really finds bugs, at least it found a problem that was fixed by https://reviews.llvm.org/D36397. (Will make sure to submit that other change before this one).

As discussed with @klimek, will move to llvm::ThreadPool after submitting a fix for it to allow functions with moveable-only arguments to be used there (that's what clangd uses).

[klimek]

lg

clangd/ClangdLSPServer.h

 namespace clangd {
 
 class JSONOutput;
 
 /// This class provides implementation of an LSP server, glueing the JSON
 /// dispatch and ClangdServer together.
 class ClangdLSPServer {
 public:
-  ClangdLSPServer(JSONOutput &Out, bool RunSynchronously,
+  ClangdLSPServer(JSONOutput &Out, SchedulingOptions SchedOpts,
                   llvm::Optional<StringRef> ResourceDir);
 
   /// Run LSP server loop, receiving input for it from \p In. \p In must be
   /// opened in binary mode. Output will be written using Out variable passed to
   /// class constructor. This method must not be executed more than once for
   /// each instance of ClangdLSPServer.
   void run(std::istream &In);
 

clangd/ClangdLSPServer.cpp

void ClangdLSPServer::LSPProtocolCallbacks::onGoToDefinition(
   }
   if (!Locations.empty())
     Locations.pop_back();
   Out.writeMessage(
       R"({"jsonrpc":"2.0","id":)" + ID.str() +
       R"(,"result":[)" + Locations + R"(]})");
 }
 
-ClangdLSPServer::ClangdLSPServer(JSONOutput &Out, bool RunSynchronously,
+ClangdLSPServer::ClangdLSPServer(JSONOutput &Out, SchedulingOptions SchedOpts,
                                  llvm::Optional<StringRef> ResourceDir)
     : Out(Out), DiagConsumer(*this),
-      Server(CDB, DiagConsumer, FSProvider, RunSynchronously, ResourceDir) {}
+      Server(CDB, DiagConsumer, FSProvider, SchedOpts, ResourceDir) {}
 
 void ClangdLSPServer::run(std::istream &In) {
   assert(!IsDone && "Run was called before");
 
   // Set up JSONRPCDispatcher.
   LSPProtocolCallbacks Callbacks(*this);
   JSONRPCDispatcher Dispatcher(llvm::make_unique<Handler>(Out));
   regiterCallbackHandlers(Dispatcher, Out, Callbacks);

clangd/ClangdServer.h

 public:
   /// \return getRealFileSystem() tagged with default tag, i.e. VFSTag()
   Tagged<IntrusiveRefCntPtr<vfs::FileSystem>>
   getTaggedFileSystem(PathRef File) override;
 };
 
 class ClangdServer;
 
+/// A helper class to pass concurrency parameters to ClangdScheduler.
+class SchedulingOptions {
+public:
+  /// Indicates that requests must be executed immediately on the calling
+  /// thread.
+  static SchedulingOptions RunOnCallingThread();
+  /// Indicates that requests should be executed on separate worker threads. In
+  /// total \p ThreadsCount working threads will be created.
+  /// Asserts that \p ThreadsCount is > 0.
+  static SchedulingOptions RunOnWorkerThreads(unsigned ThreadsCount);
+  /// Uses this::thread::hardware_concurrency() for the count of threads.
+  static SchedulingOptions RunOnDefaultNumberOfThreads();
+
+private:
+  SchedulingOptions(unsigned ThreadsCount, bool RunSynchronously);
+
+public:
+  /// Returns the number of threads to use when shouldRunsynchronously() is
+  /// false. Must not be called if shouldRunsynchronously() is true.
+  unsigned getThreadsCount();
+  bool shouldRunSynchronously();
+
+private:
+  unsigned ThreadsCount;
+  bool RunSynchronously;
+};
+
 /// Handles running WorkerRequests of ClangdServer on a separate threads.
 /// Currently runs only one worker thread.
 class ClangdScheduler {
 public:
-  ClangdScheduler(bool RunSynchronously);
+  ClangdScheduler(SchedulingOptions Opts);
   ~ClangdScheduler();
 
   /// Add a new request to run function \p F with args \p As to the start of the
   /// queue. The request will be run on a separate thread.
   template <class Func, class... Args>
   void addToFront(Func &&F, Args &&... As) {
     if (RunSynchronously) {
       std::forward<Func>(F)(std::forward<Args>(As)...);
if (RunSynchronously) {
                                         std::forward<Args>(As)...));
     }
     RequestCV.notify_one();
   }
 
 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;
+  std::vector<std::thread> Workers;
   /// Setting Done to true will make the worker thread 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.
   std::condition_variable RequestCV;
 };
 
 /// Provides API to manage ASTs for a collection of C++ files and request
 /// various language features(currently, only codeCompletion and async
 /// 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,
+  /// created using \p SchedOpts, will be created. 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
   /// (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, SchedulingOptions SchedOpts,
                llvm::Optional<StringRef> ResourceDir = llvm::None);
 
   /// 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.

clangd/ClangdServer.cpp

Position clangd::offsetToPosition(StringRef Code, size_t Offset) {
   return {Lines, Cols};
 }
 
 Tagged<IntrusiveRefCntPtr<vfs::FileSystem>>
 RealFileSystemProvider::getTaggedFileSystem(PathRef File) {
   return make_tagged(vfs::getRealFileSystem(), VFSTag());
 }
 
-ClangdScheduler::ClangdScheduler(bool RunSynchronously)
-    : RunSynchronously(RunSynchronously) {
+SchedulingOptions SchedulingOptions::RunOnCallingThread() {
+  return SchedulingOptions(1, true);
+}
+
+SchedulingOptions SchedulingOptions::RunOnWorkerThreads(unsigned ThreadsCount) {
+  assert(ThreadsCount > 0);
+  return SchedulingOptions(ThreadsCount, false);
+}
+
+SchedulingOptions SchedulingOptions::RunOnDefaultNumberOfThreads() {
+  unsigned Threads = std::thread::hardware_concurrency();
+  if (Threads == 0)
+    Threads = 1; // C++ standard says hardware_concurrency() can return 0 on
+                 // some platforms.
+  return RunOnWorkerThreads(Threads);
+}
+
+SchedulingOptions::SchedulingOptions(unsigned ThreadsCount,
+                                     bool RunSynchronously)
+    : ThreadsCount(ThreadsCount), RunSynchronously(RunSynchronously) {
+  if (ThreadsCount == 0) {
+    assert(false && "ThreadsCount should never be 0 to tolerate misuses of an "
+                    "API with disabled asserts.");
+    ThreadsCount = 1;
+  }
+}
+
+unsigned SchedulingOptions::getThreadsCount() {
+  assert(!shouldRunSynchronously());
+  return ThreadsCount;
+}
+
+bool SchedulingOptions::shouldRunSynchronously() { return RunSynchronously; }
+
+ClangdScheduler::ClangdScheduler(SchedulingOptions Opts)
+    : RunSynchronously(Opts.shouldRunSynchronously()) {
   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]() {
+  unsigned ThreadsCount = Opts.getThreadsCount();
+
+  Workers.reserve(ThreadsCount);
+  for (unsigned I = 0; I < ThreadsCount; ++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; });
+          RequestCV.wait(Lock,
+                         [this] { return !RequestQueue.empty() || Done; });
           if (Done)
             return;
 
           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
 
         Request.get();
       }
-  });
+    }));
+  }
 }
 
 ClangdScheduler::~ClangdScheduler() {
   if (RunSynchronously)
     return; // no worker thread is running in that case
 
   {
     std::lock_guard<std::mutex> Lock(Mutex);
     // Wake up the worker thread
     Done = true;
   } // unlock Mutex
-  RequestCV.notify_one();
+  RequestCV.notify_all();
+
+  for (auto &Worker : Workers)
     Worker.join();
 }
 
 ClangdServer::ClangdServer(GlobalCompilationDatabase &CDB,
                            DiagnosticsConsumer &DiagConsumer,
                            FileSystemProvider &FSProvider,
-                           bool RunSynchronously,
+                           SchedulingOptions SchedOpts,
                            llvm::Optional<StringRef> ResourceDir)
     : CDB(CDB), DiagConsumer(DiagConsumer), FSProvider(FSProvider),
       ResourceDir(ResourceDir ? ResourceDir->str() : getStandardResourceDir()),
       PCHs(std::make_shared<PCHContainerOperations>()),
-      WorkScheduler(RunSynchronously) {}
+      WorkScheduler(SchedOpts) {}
 
 std::future<void> ClangdServer::addDocument(PathRef File, StringRef Contents) {
   DocVersion Version = DraftMgr.updateDraft(File, Contents);
 
   auto TaggedFS = FSProvider.getTaggedFileSystem(File);
   std::shared_ptr<CppFile> Resources =
       Units.getOrCreateFile(File, ResourceDir, CDB, PCHs, TaggedFS.Value);
 

clangd/tool/ClangdMain.cpp

 #include "JSONRPCDispatcher.h"
 #include "llvm/Support/CommandLine.h"
 #include "llvm/Support/FileSystem.h"
 #include "llvm/Support/Program.h"
 
 #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 unsigned getDefaultThreadsCount() {
+  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;
+}
+
+static llvm::cl::opt<unsigned>
+    ThreadsCount("j",
+                 llvm::cl::desc("Number of parallel workers used by clangd."),
+                 llvm::cl::init(getDefaultThreadsCount()));
+
+static llvm::cl::opt<bool> RunSynchronously(
+    "run-synchronously",
+    llvm::cl::desc("Parse on main thread. If set, -j is ignored."),
 
 static llvm::cl::opt<std::string>
     ResourceDir("resource-dir",
                 llvm::cl::desc("Directory for system clang headers"),
                 llvm::cl::init(""), llvm::cl::Hidden);
 
 int main(int argc, char *argv[]) {
   llvm::cl::ParseCommandLineOptions(argc, argv, "clangd");
 
+  if (!RunSynchronously && ThreadsCount == 0) {
+    llvm::errs() << "Number of threads must be positive.";
+    return 1;
+  }
+
   llvm::raw_ostream &Outs = llvm::outs();
   llvm::raw_ostream &Logs = llvm::errs();
   JSONOutput Out(Outs, Logs);
 
   // Change stdin to binary to not lose \r\n on windows.
   llvm::sys::ChangeStdinToBinary();
 
   llvm::Optional<StringRef> ResourceDirRef = None;
   if (!ResourceDir.empty())
     ResourceDirRef = ResourceDir;
-  ClangdLSPServer LSPServer(Out, RunSynchronously, ResourceDirRef);
+
+  SchedulingOptions SchedOpts =
+      !RunSynchronously ? SchedulingOptions::RunOnWorkerThreads(ThreadsCount)
+                        : SchedulingOptions::RunOnCallingThread();
+
+  ClangdLSPServer LSPServer(Out, SchedOpts, ResourceDirRef);
   LSPServer.run(std::cin);
 }

unittests/clangd/ClangdTests.cpp

#endif
   std::string parseSourceAndDumpAST(
       PathRef SourceFileRelPath, StringRef SourceContents,
       std::vector<std::pair<PathRef, StringRef>> ExtraFiles = {},
       bool ExpectErrors = false) {
     MockFSProvider FS;
     ErrorCheckingDiagConsumer DiagConsumer;
     MockCompilationDatabase CDB(/*AddFreestandingFlag=*/true);
     ClangdServer Server(CDB, DiagConsumer, FS,
-                        /*RunSynchronously=*/false);
+                        SchedulingOptions::RunOnDefaultNumberOfThreads());
     for (const auto &FileWithContents : ExtraFiles)
       FS.Files[getVirtualTestFilePath(FileWithContents.first)] =
           FileWithContents.second;
 
     auto SourceFilename = getVirtualTestFilePath(SourceFileRelPath);
 
     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);
 
     // Wait for reparse to finish before checking for errors.
     EXPECT_EQ(AddDocFuture.wait_for(DefaultFutureTimeout),
               std::future_status::ready);
     EXPECT_EQ(ExpectErrors, DiagConsumer.hadErrorInLastDiags());
parseSourceAndDumpAST("foo.cpp", SourceContents, {{"foo.h", "int a;"}},
                         /*ExpectErrors=*/false);
 }
 
 TEST_F(ClangdVFSTest, Reparse) {
   MockFSProvider FS;
   ErrorCheckingDiagConsumer DiagConsumer;
   MockCompilationDatabase CDB(/*AddFreestandingFlag=*/true);
   ClangdServer Server(CDB, DiagConsumer, FS,
-                      /*RunSynchronously=*/false);
+                      SchedulingOptions::RunOnDefaultNumberOfThreads());
 
   const auto SourceContents = R"cpp(
 #include "foo.h"
 int b = a;
 )cpp";
 
   auto FooCpp = getVirtualTestFilePath("foo.cpp");
   auto FooH = getVirtualTestFilePath("foo.h");

 }
 
 TEST_F(ClangdVFSTest, ReparseOnHeaderChange) {
   MockFSProvider FS;
   ErrorCheckingDiagConsumer DiagConsumer;
   MockCompilationDatabase CDB(/*AddFreestandingFlag=*/true);
 
   ClangdServer Server(CDB, DiagConsumer, FS,
-                      /*RunSynchronously=*/false);
+                      SchedulingOptions::RunOnDefaultNumberOfThreads());
 
   const auto SourceContents = R"cpp(
 #include "foo.h"
 int b = a;
 )cpp";
 
   auto FooCpp = getVirtualTestFilePath("foo.cpp");
   auto FooH = getVirtualTestFilePath("foo.h");
)cpp";
   EXPECT_NE(DumpParse1, DumpParseDifferent);
 }
 
 TEST_F(ClangdVFSTest, CheckVersions) {
   MockFSProvider FS;
   ErrorCheckingDiagConsumer DiagConsumer;
   MockCompilationDatabase CDB(/*AddFreestandingFlag=*/true);
   ClangdServer Server(CDB, DiagConsumer, FS,
-                      /*RunSynchronously=*/true);
+                      SchedulingOptions::RunOnCallingThread());
 
   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);
   EXPECT_EQ(DiagConsumer.lastVFSTag(), FS.Tag);
 
   FS.Tag = "321";
   Server.addDocument(FooCpp, SourceContents);
   EXPECT_EQ(DiagConsumer.lastVFSTag(), FS.Tag);
   EXPECT_EQ(Server.codeComplete(FooCpp, Position{0, 0}).Tag, FS.Tag);
 }
 
 // Only enable this test on Unix
 #ifdef LLVM_ON_UNIX
 TEST_F(ClangdVFSTest, SearchLibDir) {
   // Checks that searches for GCC installation is done through vfs.
   MockFSProvider FS;
   ErrorCheckingDiagConsumer DiagConsumer;
   MockCompilationDatabase CDB(/*AddFreestandingFlag=*/true);
   CDB.ExtraClangFlags.insert(CDB.ExtraClangFlags.end(),
                              {"-xc++", "-target", "x86_64-linux-unknown",
                               "-m64", "--gcc-toolchain=/randomusr"});
   ClangdServer Server(CDB, DiagConsumer, FS,
-                      /*RunSynchronously=*/true);
+                      SchedulingOptions::RunOnCallingThread());
 
   // Just a random gcc version string
   SmallString<8> Version("4.9.3");
 
   // A lib dir for gcc installation
   SmallString<64> LibDir("/randomusr/lib/gcc/x86_64-linux-gnu");
   llvm::sys::path::append(LibDir, Version);
 
TEST_F(ClangdVFSTest, SearchLibDir) {
 
   auto FooCpp = getVirtualTestFilePath("foo.cpp");
   const auto SourceContents = R"cpp(
 #include <string>
 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());
 
   const auto SourceContentsWithError = R"cpp(
 #include <string>
 std::string x;
 )cpp";
   Server.addDocument(FooCpp, SourceContentsWithError);

 };
 
 TEST_F(ClangdCompletionTest, CheckContentsOverride) {
   MockFSProvider FS;
   ErrorCheckingDiagConsumer DiagConsumer;
   MockCompilationDatabase CDB(/*AddFreestandingFlag=*/true);
 
   ClangdServer Server(CDB, DiagConsumer, FS,
-                      /*RunSynchronously=*/false);
+                      SchedulingOptions::RunOnDefaultNumberOfThreads());
 
   auto FooCpp = getVirtualTestFilePath("foo.cpp");
   const auto SourceContents = R"cpp(
 int aba;
 int b =   ;
 )cpp";
 
   const auto OverridenSourceContents = R"cpp(