1//===--- Server.cpp - gRPC-based Remote Index Server ---------------------===//
2//
3// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
4// See https://llvm.org/LICENSE.txt for license information.
5// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
6//
7//===----------------------------------------------------------------------===//
8
9#include "Feature.h"
10#include "Index.pb.h"
11#include "MonitoringService.grpc.pb.h"
12#include "MonitoringService.pb.h"
13#include "Service.grpc.pb.h"
14#include "Service.pb.h"
15#include "index/Index.h"
16#include "index/Serialization.h"
17#include "index/Symbol.h"
18#include "index/remote/marshalling/Marshalling.h"
19#include "support/Context.h"
20#include "support/Logger.h"
21#include "support/Shutdown.h"
22#include "support/ThreadsafeFS.h"
23#include "support/Trace.h"
24#include "llvm/ADT/IntrusiveRefCntPtr.h"
25#include "llvm/ADT/StringRef.h"
26#include "llvm/Support/Chrono.h"
27#include "llvm/Support/CommandLine.h"
28#include "llvm/Support/Error.h"
29#include "llvm/Support/FileSystem.h"
30#include "llvm/Support/FormatVariadic.h"
31#include "llvm/Support/Path.h"
32#include "llvm/Support/Signals.h"
33#include "llvm/Support/VirtualFileSystem.h"
34
35#include <chrono>
36#include <grpc++/grpc++.h>
37#include <grpc++/health_check_service_interface.h>
38#include <memory>
39#include <optional>
40#include <string>
41#include <thread>
42#include <utility>
43
44#if ENABLE_GRPC_REFLECTION
45#include <grpc++/ext/proto_server_reflection_plugin.h>
46#endif
47
48#ifdef __GLIBC__
49#include <malloc.h>
50#endif
51
52namespace clang {
53namespace clangd {
54namespace remote {
55namespace {
56
57static constexpr char Overview[] = R"(
58This is an experimental remote index implementation. The server opens Dex and
59awaits gRPC lookup requests from the client.
60)";
61
62llvm::cl::opt<std::string> IndexPath(llvm::cl::desc("<INDEX FILE>"),
63 llvm::cl::Positional, llvm::cl::Required);
64
65llvm::cl::opt<std::string> IndexRoot(llvm::cl::desc("<PROJECT ROOT>"),
66 llvm::cl::Positional, llvm::cl::Required);
67
68llvm::cl::opt<Logger::Level> LogLevel{
69 "log",
70 llvm::cl::desc("Verbosity of log messages written to stderr"),
71 values(clEnumValN(Logger::Error, "error", "Error messages only"),
72 clEnumValN(Logger::Info, "info", "High level execution tracing"),
73 clEnumValN(Logger::Debug, "verbose", "Low level details")),
74 llvm::cl::init(Val: Logger::Info),
75};
76
77llvm::cl::opt<bool> LogPublic{
78 "log-public",
79 llvm::cl::desc("Avoid logging potentially-sensitive request details"),
80 llvm::cl::init(Val: false),
81};
82
83llvm::cl::opt<std::string> LogPrefix{
84 "log-prefix",
85 llvm::cl::desc("A string that'll be prepended to all log statements. "
86 "Useful when running multiple instances on same host."),
87};
88
89llvm::cl::opt<std::string> TraceFile(
90 "trace-file",
91 llvm::cl::desc("Path to the file where tracer logs will be stored"));
92
93llvm::cl::opt<bool> PrettyPrint{
94 "pretty",
95 llvm::cl::desc("Pretty-print JSON output in the trace"),
96 llvm::cl::init(Val: false),
97};
98
99llvm::cl::opt<std::string> ServerAddress(
100 "server-address", llvm::cl::init(Val: "0.0.0.0:50051"),
101 llvm::cl::desc("Address of the invoked server. Defaults to 0.0.0.0:50051"));
102
103llvm::cl::opt<size_t> IdleTimeoutSeconds(
104 "idle-timeout", llvm::cl::init(Val: 8 * 60),
105 llvm::cl::desc("Maximum time a channel may stay idle until server closes "
106 "the connection, in seconds. Defaults to 480."));
107
108llvm::cl::opt<size_t> LimitResults(
109 "limit-results", llvm::cl::init(Val: 10000),
110 llvm::cl::desc("Maximum number of results to stream as a response to "
111 "single request. Limit is to keep the server from being "
112 "DOS'd. Defaults to 10000."));
113
114static Key<grpc::ServerContext *> CurrentRequest;
115
116class RemoteIndexServer final : public v1::SymbolIndex::Service {
117public:
118 RemoteIndexServer(clangd::SymbolIndex &Index, llvm::StringRef IndexRoot)
119 : Index(Index) {
120 llvm::SmallString<256> NativePath = IndexRoot;
121 llvm::sys::path::native(path&: NativePath);
122 ProtobufMarshaller = std::unique_ptr<Marshaller>(new Marshaller(
123 /*RemoteIndexRoot=*/llvm::StringRef(NativePath),
124 /*LocalIndexRoot=*/""));
125 }
126
127private:
128 using stopwatch = std::chrono::steady_clock;
129
130 grpc::Status Lookup(grpc::ServerContext *Context,
131 const LookupRequest *Request,
132 grpc::ServerWriter<LookupReply> *Reply) override {
133 auto StartTime = stopwatch::now();
134 WithContextValue WithRequestContext(CurrentRequest, Context);
135 logRequest(*Request);
136 trace::Span Tracer("LookupRequest");
137 auto Req = ProtobufMarshaller->fromProtobuf(Message: Request);
138 if (!Req) {
139 elog(Fmt: "Can not parse LookupRequest from protobuf: {0}", Vals: Req.takeError());
140 return grpc::Status::CANCELLED;
141 }
142 unsigned Sent = 0;
143 unsigned FailedToSend = 0;
144 bool HasMore = false;
145 Index.lookup(Req: *Req, Callback: [&](const clangd::Symbol &Item) {
146 if (Sent >= LimitResults) {
147 HasMore = true;
148 return;
149 }
150 auto SerializedItem = ProtobufMarshaller->toProtobuf(From: Item);
151 if (!SerializedItem) {
152 elog(Fmt: "Unable to convert Symbol to protobuf: {0}",
153 Vals: SerializedItem.takeError());
154 ++FailedToSend;
155 return;
156 }
157 LookupReply NextMessage;
158 *NextMessage.mutable_stream_result() = *SerializedItem;
159 logResponse(NextMessage);
160 Reply->Write(NextMessage);
161 ++Sent;
162 });
163 if (HasMore)
164 log(Fmt: "[public] Limiting result size for Lookup request.");
165 LookupReply LastMessage;
166 LastMessage.mutable_final_result()->set_has_more(HasMore);
167 logResponse(LastMessage);
168 Reply->Write(LastMessage);
169 SPAN_ATTACH(Tracer, "Sent", Sent);
170 SPAN_ATTACH(Tracer, "Failed to send", FailedToSend);
171 logRequestSummary(RequestName: "v1/Lookup", Sent, StartTime);
172 return grpc::Status::OK;
173 }
174
175 grpc::Status FuzzyFind(grpc::ServerContext *Context,
176 const FuzzyFindRequest *Request,
177 grpc::ServerWriter<FuzzyFindReply> *Reply) override {
178 auto StartTime = stopwatch::now();
179 WithContextValue WithRequestContext(CurrentRequest, Context);
180 logRequest(*Request);
181 trace::Span Tracer("FuzzyFindRequest");
182 auto Req = ProtobufMarshaller->fromProtobuf(Message: Request);
183 if (!Req) {
184 elog(Fmt: "Can not parse FuzzyFindRequest from protobuf: {0}",
185 Vals: Req.takeError());
186 return grpc::Status::CANCELLED;
187 }
188 if (!Req->Limit || *Req->Limit > LimitResults) {
189 log(Fmt: "[public] Limiting result size for FuzzyFind request from {0} to {1}",
190 Vals&: Req->Limit, Vals&: LimitResults);
191 Req->Limit = LimitResults;
192 }
193 unsigned Sent = 0;
194 unsigned FailedToSend = 0;
195 bool HasMore = Index.fuzzyFind(Req: *Req, Callback: [&](const clangd::Symbol &Item) {
196 auto SerializedItem = ProtobufMarshaller->toProtobuf(From: Item);
197 if (!SerializedItem) {
198 elog(Fmt: "Unable to convert Symbol to protobuf: {0}",
199 Vals: SerializedItem.takeError());
200 ++FailedToSend;
201 return;
202 }
203 FuzzyFindReply NextMessage;
204 *NextMessage.mutable_stream_result() = *SerializedItem;
205 logResponse(NextMessage);
206 Reply->Write(NextMessage);
207 ++Sent;
208 });
209 FuzzyFindReply LastMessage;
210 LastMessage.mutable_final_result()->set_has_more(HasMore);
211 logResponse(LastMessage);
212 Reply->Write(LastMessage);
213 SPAN_ATTACH(Tracer, "Sent", Sent);
214 SPAN_ATTACH(Tracer, "Failed to send", FailedToSend);
215 logRequestSummary(RequestName: "v1/FuzzyFind", Sent, StartTime);
216 return grpc::Status::OK;
217 }
218
219 grpc::Status Refs(grpc::ServerContext *Context, const RefsRequest *Request,
220 grpc::ServerWriter<RefsReply> *Reply) override {
221 auto StartTime = stopwatch::now();
222 WithContextValue WithRequestContext(CurrentRequest, Context);
223 logRequest(*Request);
224 trace::Span Tracer("RefsRequest");
225 auto Req = ProtobufMarshaller->fromProtobuf(Message: Request);
226 if (!Req) {
227 elog(Fmt: "Can not parse RefsRequest from protobuf: {0}", Vals: Req.takeError());
228 return grpc::Status::CANCELLED;
229 }
230 if (!Req->Limit || *Req->Limit > LimitResults) {
231 log(Fmt: "[public] Limiting result size for Refs request from {0} to {1}.",
232 Vals&: Req->Limit, Vals&: LimitResults);
233 Req->Limit = LimitResults;
234 }
235 unsigned Sent = 0;
236 unsigned FailedToSend = 0;
237 bool HasMore = Index.refs(Req: *Req, Callback: [&](const clangd::Ref &Item) {
238 auto SerializedItem = ProtobufMarshaller->toProtobuf(From: Item);
239 if (!SerializedItem) {
240 elog(Fmt: "Unable to convert Ref to protobuf: {0}",
241 Vals: SerializedItem.takeError());
242 ++FailedToSend;
243 return;
244 }
245 RefsReply NextMessage;
246 *NextMessage.mutable_stream_result() = *SerializedItem;
247 logResponse(NextMessage);
248 Reply->Write(NextMessage);
249 ++Sent;
250 });
251 RefsReply LastMessage;
252 LastMessage.mutable_final_result()->set_has_more(HasMore);
253 logResponse(LastMessage);
254 Reply->Write(LastMessage);
255 SPAN_ATTACH(Tracer, "Sent", Sent);
256 SPAN_ATTACH(Tracer, "Failed to send", FailedToSend);
257 logRequestSummary(RequestName: "v1/Refs", Sent, StartTime);
258 return grpc::Status::OK;
259 }
260
261 grpc::Status Relations(grpc::ServerContext *Context,
262 const RelationsRequest *Request,
263 grpc::ServerWriter<RelationsReply> *Reply) override {
264 auto StartTime = stopwatch::now();
265 WithContextValue WithRequestContext(CurrentRequest, Context);
266 logRequest(*Request);
267 trace::Span Tracer("RelationsRequest");
268 auto Req = ProtobufMarshaller->fromProtobuf(Message: Request);
269 if (!Req) {
270 elog(Fmt: "Can not parse RelationsRequest from protobuf: {0}",
271 Vals: Req.takeError());
272 return grpc::Status::CANCELLED;
273 }
274 if (!Req->Limit || *Req->Limit > LimitResults) {
275 log(Fmt: "[public] Limiting result size for Relations request from {0} to "
276 "{1}.",
277 Vals&: Req->Limit, Vals&: LimitResults);
278 Req->Limit = LimitResults;
279 }
280 unsigned Sent = 0;
281 unsigned FailedToSend = 0;
282 Index.relations(
283 Req: *Req, Callback: [&](const SymbolID &Subject, const clangd::Symbol &Object) {
284 auto SerializedItem = ProtobufMarshaller->toProtobuf(Subject, Object);
285 if (!SerializedItem) {
286 elog(Fmt: "Unable to convert Relation to protobuf: {0}",
287 Vals: SerializedItem.takeError());
288 ++FailedToSend;
289 return;
290 }
291 RelationsReply NextMessage;
292 *NextMessage.mutable_stream_result() = *SerializedItem;
293 logResponse(NextMessage);
294 Reply->Write(NextMessage);
295 ++Sent;
296 });
297 RelationsReply LastMessage;
298 LastMessage.mutable_final_result()->set_has_more(true);
299 logResponse(LastMessage);
300 Reply->Write(LastMessage);
301 SPAN_ATTACH(Tracer, "Sent", Sent);
302 SPAN_ATTACH(Tracer, "Failed to send", FailedToSend);
303 logRequestSummary(RequestName: "v1/Relations", Sent, StartTime);
304 return grpc::Status::OK;
305 }
306
307 // Proxy object to allow proto messages to be lazily serialized as text.
308 struct TextProto {
309 const google::protobuf::Message &M;
310 friend llvm::raw_ostream &operator<<(llvm::raw_ostream &OS,
311 const TextProto &P) {
312 return OS << P.M.DebugString();
313 }
314 };
315
316 void logRequest(const google::protobuf::Message &M) {
317 vlog("<<< {0}\n{1}", M.GetDescriptor()->name(), TextProto{M});
318 }
319 void logResponse(const google::protobuf::Message &M) {
320 vlog(">>> {0}\n{1}", M.GetDescriptor()->name(), TextProto{M});
321 }
322 void logRequestSummary(llvm::StringLiteral RequestName, unsigned Sent,
323 stopwatch::time_point StartTime) {
324 auto Duration = stopwatch::now() - StartTime;
325 auto Millis =
326 std::chrono::duration_cast<std::chrono::milliseconds>(d: Duration).count();
327 log(Fmt: "[public] request {0} => OK: {1} results in {2}ms", Vals&: RequestName, Vals&: Sent,
328 Vals&: Millis);
329 }
330
331 std::unique_ptr<Marshaller> ProtobufMarshaller;
332 clangd::SymbolIndex &Index;
333};
334
335class Monitor final : public v1::Monitor::Service {
336public:
337 Monitor(llvm::sys::TimePoint<> IndexAge)
338 : StartTime(std::chrono::system_clock::now()), IndexBuildTime(IndexAge) {}
339
340 void updateIndex(llvm::sys::TimePoint<> UpdateTime) {
341 IndexBuildTime.exchange(i: UpdateTime);
342 }
343
344private:
345 // FIXME(kirillbobyrev): Most fields should be populated when the index
346 // reloads (probably in adjacent metadata.txt file next to loaded .idx) but
347 // they aren't right now.
348 grpc::Status MonitoringInfo(grpc::ServerContext *Context,
349 const v1::MonitoringInfoRequest *Request,
350 v1::MonitoringInfoReply *Reply) override {
351 Reply->set_uptime_seconds(std::chrono::duration_cast<std::chrono::seconds>(
352 d: std::chrono::system_clock::now() - StartTime)
353 .count());
354 // FIXME(kirillbobyrev): We are currently making use of the last
355 // modification time of the index artifact to deduce its age. This is wrong
356 // as it doesn't account for the indexing delay. Propagate some metadata
357 // with the index artifacts to indicate time of the commit we indexed.
358 Reply->set_index_age_seconds(
359 std::chrono::duration_cast<std::chrono::seconds>(
360 d: std::chrono::system_clock::now() - IndexBuildTime.load())
361 .count());
362 return grpc::Status::OK;
363 }
364
365 const llvm::sys::TimePoint<> StartTime;
366 std::atomic<llvm::sys::TimePoint<>> IndexBuildTime;
367};
368
369void maybeTrimMemory() {
370#if defined(__GLIBC__) && CLANGD_MALLOC_TRIM
371 malloc_trim(pad: 0);
372#endif
373}
374
375// Detect changes in \p IndexPath file and load new versions of the index
376// whenever they become available.
377void hotReload(clangd::SwapIndex &Index, llvm::StringRef IndexPath,
378 llvm::vfs::Status &LastStatus,
379 llvm::IntrusiveRefCntPtr<llvm::vfs::FileSystem> &FS,
380 Monitor &Monitor) {
381 // glibc malloc doesn't shrink an arena if there are items living at the end,
382 // which might happen since we destroy the old index after building new one.
383 // Trim more aggresively to keep memory usage of the server low.
384 // Note that we do it deliberately here rather than after Index.reset(),
385 // because old index might still be kept alive after the reset call if we are
386 // serving requests.
387 maybeTrimMemory();
388 auto Status = FS->status(Path: IndexPath);
389 // Requested file is same as loaded index: no reload is needed.
390 if (!Status || (Status->getLastModificationTime() ==
391 LastStatus.getLastModificationTime() &&
392 Status->getSize() == LastStatus.getSize()))
393 return;
394 vlog(Fmt: "Found different index version: existing index was modified at "
395 "{0}, new index was modified at {1}. Attempting to reload.",
396 Vals: LastStatus.getLastModificationTime(), Vals: Status->getLastModificationTime());
397 LastStatus = *Status;
398 std::unique_ptr<clang::clangd::SymbolIndex> NewIndex =
399 loadIndex(Filename: IndexPath, Origin: SymbolOrigin::Static);
400 if (!NewIndex) {
401 elog(Fmt: "Failed to load new index. Old index will be served.");
402 return;
403 }
404 Index.reset(std::move(NewIndex));
405 Monitor.updateIndex(UpdateTime: Status->getLastModificationTime());
406 log(Fmt: "New index version loaded. Last modification time: {0}, size: {1} bytes.",
407 Vals: Status->getLastModificationTime(), Vals: Status->getSize());
408}
409
410void runServerAndWait(clangd::SymbolIndex &Index, llvm::StringRef ServerAddress,
411 llvm::StringRef IndexPath, Monitor &Monitor) {
412 RemoteIndexServer Service(Index, IndexRoot);
413
414 grpc::EnableDefaultHealthCheckService(enable: true);
415#if ENABLE_GRPC_REFLECTION
416 grpc::reflection::InitProtoReflectionServerBuilderPlugin();
417#endif
418 grpc::ServerBuilder Builder;
419 Builder.AddListeningPort(addr_uri: ServerAddress.str(),
420 creds: grpc::InsecureServerCredentials());
421 Builder.AddChannelArgument(GRPC_ARG_MAX_CONNECTION_IDLE_MS,
422 value: IdleTimeoutSeconds * 1000);
423 Builder.RegisterService(&Service);
424 Builder.RegisterService(&Monitor);
425 std::unique_ptr<grpc::Server> Server(Builder.BuildAndStart());
426 log(Fmt: "Server listening on {0}", Vals&: ServerAddress);
427
428 std::thread ServerShutdownWatcher([&]() {
429 static constexpr auto WatcherFrequency = std::chrono::seconds(5);
430 while (!clang::clangd::shutdownRequested())
431 std::this_thread::sleep_for(rtime: WatcherFrequency);
432 Server->Shutdown();
433 });
434
435 Server->Wait();
436 ServerShutdownWatcher.join();
437}
438
439std::unique_ptr<Logger> makeLogger(llvm::StringRef LogPrefix,
440 llvm::raw_ostream &OS) {
441 std::unique_ptr<Logger> Base;
442 if (LogPublic) {
443 // Redacted mode:
444 // - messages outside the scope of a request: log fully
445 // - messages tagged [public]: log fully
446 // - errors: log the format string
447 // - others: drop
448 class RedactedLogger : public StreamLogger {
449 public:
450 using StreamLogger::StreamLogger;
451 void log(Level L, const char *Fmt,
452 const llvm::formatv_object_base &Message) override {
453 if (Context::current().get(Key: CurrentRequest) == nullptr ||
454 llvm::StringRef(Fmt).starts_with(Prefix: "[public]"))
455 return StreamLogger::log(L, Fmt, Message);
456 if (L >= Error)
457 return StreamLogger::log(L, Fmt,
458 Message: llvm::formatv(Fmt: "[redacted] {0}", Vals&: Fmt));
459 }
460 };
461 Base = std::make_unique<RedactedLogger>(args&: OS, args&: LogLevel);
462 } else {
463 Base = std::make_unique<StreamLogger>(args&: OS, args&: LogLevel);
464 }
465
466 if (LogPrefix.empty())
467 return Base;
468 class PrefixedLogger : public Logger {
469 std::string LogPrefix;
470 std::unique_ptr<Logger> Base;
471
472 public:
473 PrefixedLogger(llvm::StringRef LogPrefix, std::unique_ptr<Logger> Base)
474 : LogPrefix(LogPrefix.str()), Base(std::move(Base)) {}
475 void log(Level L, const char *Fmt,
476 const llvm::formatv_object_base &Message) override {
477 Base->log(L, Fmt, Message: llvm::formatv(Fmt: "[{0}] {1}", Vals&: LogPrefix, Vals: Message));
478 }
479 };
480 return std::make_unique<PrefixedLogger>(args&: LogPrefix, args: std::move(Base));
481}
482
483} // namespace
484} // namespace remote
485} // namespace clangd
486} // namespace clang
487
488using clang::clangd::elog;
489
490int main(int argc, char *argv[]) {
491 using namespace clang::clangd::remote;
492 llvm::cl::ParseCommandLineOptions(argc, argv, Overview);
493 llvm::sys::PrintStackTraceOnErrorSignal(Argv0: argv[0]);
494 llvm::sys::SetInterruptFunction(&clang::clangd::requestShutdown);
495
496 if (!llvm::sys::path::is_absolute(path: IndexRoot)) {
497 llvm::errs() << "Index root should be an absolute path.\n";
498 return -1;
499 }
500
501 llvm::errs().SetBuffered();
502 // Don't flush stdout when logging for thread safety.
503 llvm::errs().tie(TieTo: nullptr);
504 auto Logger = makeLogger(LogPrefix: LogPrefix.getValue(), OS&: llvm::errs());
505 clang::clangd::LoggingSession LoggingSession(*Logger);
506
507 std::optional<llvm::raw_fd_ostream> TracerStream;
508 std::unique_ptr<clang::clangd::trace::EventTracer> Tracer;
509 if (!TraceFile.empty()) {
510 std::error_code EC;
511 TracerStream.emplace(args&: TraceFile, args&: EC,
512 args: llvm::sys::fs::FA_Read | llvm::sys::fs::FA_Write);
513 if (EC) {
514 TracerStream.reset();
515 elog(Fmt: "Error while opening trace file {0}: {1}", Vals&: TraceFile, Vals: EC.message());
516 } else {
517 // FIXME(kirillbobyrev): Also create metrics tracer to track latency and
518 // accumulate other request statistics.
519 Tracer = clang::clangd::trace::createJSONTracer(OS&: *TracerStream,
520 /*PrettyPrint=*/Pretty: false);
521 clang::clangd::vlog(Fmt: "Successfully created a tracer.");
522 }
523 }
524
525 std::optional<clang::clangd::trace::Session> TracingSession;
526 if (Tracer)
527 TracingSession.emplace(args&: *Tracer);
528
529 clang::clangd::RealThreadsafeFS TFS;
530 auto FS = TFS.view(CWD: std::nullopt);
531 auto Status = FS->status(Path: IndexPath);
532 if (!Status) {
533 elog(Fmt: "{0} does not exist.", Vals&: IndexPath);
534 return Status.getError().value();
535 }
536
537 auto SymIndex =
538 clang::clangd::loadIndex(Filename: IndexPath, Origin: clang::clangd::SymbolOrigin::Static);
539 if (!SymIndex) {
540 llvm::errs() << "Failed to open the index.\n";
541 return -1;
542 }
543 clang::clangd::SwapIndex Index(std::move(SymIndex));
544
545 Monitor Monitor(Status->getLastModificationTime());
546
547 std::thread HotReloadThread([&Index, &Status, &FS, &Monitor]() {
548 llvm::vfs::Status LastStatus = *Status;
549 static constexpr auto RefreshFrequency = std::chrono::seconds(30);
550 while (!clang::clangd::shutdownRequested()) {
551 hotReload(Index, IndexPath: llvm::StringRef(IndexPath), LastStatus, FS, Monitor);
552 std::this_thread::sleep_for(rtime: RefreshFrequency);
553 }
554 });
555
556 runServerAndWait(Index, ServerAddress, IndexPath, Monitor);
557
558 HotReloadThread.join();
559}
560

source code of clang-tools-extra/clangd/index/remote/server/Server.cpp