[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

103–104

typo: "on separate threads"

157–158

typo: "on separate threads"

clangd/tool/ClangdMain.cpp

24–32

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

65

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

65

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, unsigned AsyncThreadsCount,
                   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, 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");
 
   // Set up JSONRPCDispatcher.
   LSPProtocolCallbacks Callbacks(*this);
   JSONRPCDispatcher Dispatcher(llvm::make_unique<Handler>(Out));
   regiterCallbackHandlers(Dispatcher, Out, Callbacks);

clangd/ClangdServer.h

 class RealFileSystemProvider : public FileSystemProvider {
 public:
   /// \return getRealFileSystem() tagged with default tag, i.e. VFSTag()
   Tagged<IntrusiveRefCntPtr<vfs::FileSystem>>
   getTaggedFileSystem(PathRef File) override;
 };
 
 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
   /// 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)...);
       return;
     }
 
     {
       std::lock_guard<std::mutex> Lock(Mutex);
if (RunSynchronously) {
                                         std::forward<Func>(F),
                                         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;
-  /// 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;
 };
 
 /// 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,
+  /// 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
   /// \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) {
+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]() {
+  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; });
+          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,
+                           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);
 
   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 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"),
 
 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 && 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);
 
   // 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);
+
+  ClangdLSPServer LSPServer(Out, WorkerThreadsCount, 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);
+    ClangdServer Server(CDB, DiagConsumer, FS, getDefaultAsyncThreadsCount());
     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());
)cpp";
   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);
+  ClangdServer Server(CDB, DiagConsumer, FS, getDefaultAsyncThreadsCount());
 
   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, DumpParseEmpty);
 }
 
 TEST_F(ClangdVFSTest, ReparseOnHeaderChange) {
   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"
 int b = a;
 )cpp";
 
   auto FooCpp = getVirtualTestFilePath("foo.cpp");
   auto FooH = getVirtualTestFilePath("foo.h");
)cpp";
   EXPECT_EQ(DumpParse1, DumpParse2);
   EXPECT_NE(DumpParse1, DumpParseDifferent);
 }
 
 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);
   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"});
+  // Run ClangdServer synchronously.
   ClangdServer Server(CDB, DiagConsumer, FS,
-                      /*RunSynchronously=*/true);
+                      /*AsyncThreadsCount=*/0);
 
   // 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);
protected:
   }
 };
 
 TEST_F(ClangdCompletionTest, CheckContentsOverride) {
   MockFSProvider FS;
   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(
 int aba;
 int b =   ;
 )cpp";
 
   const auto OverridenSourceContents = R"cpp(