1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
|
//===--- Client.cpp ----------------------------------------------*- C++-*-===//
//
// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
// See https://llvm.org/LICENSE.txt for license information.
// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
//
//===----------------------------------------------------------------------===//
#include <grpc++/grpc++.h>
#include "Client.h"
#include "Feature.h"
#include "Service.grpc.pb.h"
#include "index/Index.h"
#include "marshalling/Marshalling.h"
#include "support/Logger.h"
#include "support/Trace.h"
#include "llvm/ADT/SmallString.h"
#include "llvm/ADT/StringRef.h"
#include "llvm/Support/Error.h"
#include <atomic>
#include <chrono>
#include <memory>
namespace clang {
namespace clangd {
namespace remote {
namespace {
llvm::StringRef toString(const grpc_connectivity_state &State) {
switch (State) {
case GRPC_CHANNEL_IDLE:
return "idle";
case GRPC_CHANNEL_CONNECTING:
return "connecting";
case GRPC_CHANNEL_READY:
return "ready";
case GRPC_CHANNEL_TRANSIENT_FAILURE:
return "transient failure";
case GRPC_CHANNEL_SHUTDOWN:
return "shutdown";
}
llvm_unreachable("Not a valid grpc_connectivity_state.");
}
class IndexClient : public clangd::SymbolIndex {
void updateConnectionStatus() const {
auto NewStatus = Channel->GetState(/*try_to_connect=*/false);
auto OldStatus = ConnectionStatus.exchange(NewStatus);
if (OldStatus != NewStatus)
vlog("Remote index connection [{0}]: {1} => {2}", ServerAddress,
toString(OldStatus), toString(NewStatus));
}
template <typename RequestT, typename ReplyT>
using StreamingCall = std::unique_ptr<grpc::ClientReader<ReplyT>> (
remote::v1::SymbolIndex::Stub::*)(grpc::ClientContext *,
const RequestT &);
template <typename RequestT, typename ReplyT, typename ClangdRequestT,
typename CallbackT>
bool streamRPC(ClangdRequestT Request,
StreamingCall<RequestT, ReplyT> RPCCall,
CallbackT Callback) const {
updateConnectionStatus();
// We initialize to true because stream might be broken before we see the
// final message. In such a case there are actually more results on the
// stream, but we couldn't get to them.
bool HasMore = true;
trace::Span Tracer(RequestT::descriptor()->name());
const auto RPCRequest = ProtobufMarshaller->toProtobuf(Request);
SPAN_ATTACH(Tracer, "Request", RPCRequest.DebugString());
grpc::ClientContext Context;
Context.AddMetadata("version", versionString());
Context.AddMetadata("features", featureString());
Context.AddMetadata("platform", platformString());
std::chrono::system_clock::time_point StartTime =
std::chrono::system_clock::now();
auto Deadline = StartTime + DeadlineWaitingTime;
Context.set_deadline(Deadline);
auto Reader = (Stub.get()->*RPCCall)(&Context, RPCRequest);
dlog("Sending {0}: {1}", RequestT::descriptor()->name(),
RPCRequest.DebugString());
ReplyT Reply;
unsigned Successful = 0;
unsigned FailedToParse = 0;
while (Reader->Read(&Reply)) {
if (!Reply.has_stream_result()) {
HasMore = Reply.final_result().has_more();
continue;
}
auto Response = ProtobufMarshaller->fromProtobuf(Reply.stream_result());
if (!Response) {
elog("Received invalid {0}: {1}. Reason: {2}",
ReplyT::descriptor()->name(), Reply.stream_result().DebugString(),
Response.takeError());
++FailedToParse;
continue;
}
Callback(*Response);
++Successful;
}
auto Millis = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::system_clock::now() - StartTime)
.count();
vlog("Remote index [{0}]: {1} => {2} results in {3}ms.", ServerAddress,
RequestT::descriptor()->name(), Successful, Millis);
SPAN_ATTACH(Tracer, "Status", Reader->Finish().ok());
SPAN_ATTACH(Tracer, "Successful", Successful);
SPAN_ATTACH(Tracer, "Failed to parse", FailedToParse);
updateConnectionStatus();
return HasMore;
}
public:
IndexClient(
std::shared_ptr<grpc::Channel> Channel, llvm::StringRef Address,
llvm::StringRef ProjectRoot,
std::chrono::milliseconds DeadlineTime = std::chrono::milliseconds(1000))
: Stub(remote::v1::SymbolIndex::NewStub(Channel)), Channel(Channel),
ServerAddress(Address),
ConnectionStatus(Channel->GetState(/*try_to_connect=*/true)),
ProtobufMarshaller(new Marshaller(/*RemoteIndexRoot=*/"",
/*LocalIndexRoot=*/ProjectRoot)),
DeadlineWaitingTime(DeadlineTime) {
assert(!ProjectRoot.empty());
}
void lookup(const clangd::LookupRequest &Request,
llvm::function_ref<void(const clangd::Symbol &)> Callback)
const override {
streamRPC(Request, &remote::v1::SymbolIndex::Stub::Lookup, Callback);
}
bool fuzzyFind(const clangd::FuzzyFindRequest &Request,
llvm::function_ref<void(const clangd::Symbol &)> Callback)
const override {
return streamRPC(Request, &remote::v1::SymbolIndex::Stub::FuzzyFind,
Callback);
}
bool
refs(const clangd::RefsRequest &Request,
llvm::function_ref<void(const clangd::Ref &)> Callback) const override {
return streamRPC(Request, &remote::v1::SymbolIndex::Stub::Refs, Callback);
}
void
relations(const clangd::RelationsRequest &Request,
llvm::function_ref<void(const SymbolID &, const clangd::Symbol &)>
Callback) const override {
streamRPC(Request, &remote::v1::SymbolIndex::Stub::Relations,
// Unpack protobuf Relation.
[&](std::pair<SymbolID, clangd::Symbol> SubjectAndObject) {
Callback(SubjectAndObject.first, SubjectAndObject.second);
});
}
llvm::unique_function<IndexContents(llvm::StringRef) const>
indexedFiles() const override {
// FIXME: For now we always return IndexContents::None regardless of whether
// the file was indexed or not. A possible implementation could be
// based on the idea that we do not want to send a request at every
// call of a function returned by IndexClient::indexedFiles().
return [](llvm::StringRef) { return IndexContents::None; };
}
// IndexClient does not take any space since the data is stored on the
// server.
size_t estimateMemoryUsage() const override { return 0; }
private:
std::unique_ptr<remote::v1::SymbolIndex::Stub> Stub;
std::shared_ptr<grpc::Channel> Channel;
llvm::SmallString<256> ServerAddress;
mutable std::atomic<grpc_connectivity_state> ConnectionStatus;
std::unique_ptr<Marshaller> ProtobufMarshaller;
// Each request will be terminated if it takes too long.
std::chrono::milliseconds DeadlineWaitingTime;
};
} // namespace
std::unique_ptr<clangd::SymbolIndex> getClient(llvm::StringRef Address,
llvm::StringRef ProjectRoot) {
const auto Channel =
grpc::CreateChannel(Address.str(), grpc::InsecureChannelCredentials());
return std::unique_ptr<clangd::SymbolIndex>(
new IndexClient(Channel, Address, ProjectRoot));
}
} // namespace remote
} // namespace clangd
} // namespace clang
|
