1 /*
2 * Copyright (C) 2017 The Android Open Source Project
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17 #include "src/ipc/client_impl.h"
18
19 #include <fcntl.h>
20 #include <inttypes.h>
21 #include <unistd.h>
22
23 #include <utility>
24
25 #include "perfetto/base/task_runner.h"
26 #include "perfetto/base/utils.h"
27 #include "perfetto/ipc/service_descriptor.h"
28 #include "perfetto/ipc/service_proxy.h"
29
30 // TODO(primiano): Add ThreadChecker everywhere.
31
32 // TODO(primiano): Add timeouts.
33
34 namespace perfetto {
35 namespace ipc {
36
37 // static
CreateInstance(const char * socket_name,base::TaskRunner * task_runner)38 std::unique_ptr<Client> Client::CreateInstance(const char* socket_name,
39 base::TaskRunner* task_runner) {
40 std::unique_ptr<Client> client(new ClientImpl(socket_name, task_runner));
41 return client;
42 }
43
ClientImpl(const char * socket_name,base::TaskRunner * task_runner)44 ClientImpl::ClientImpl(const char* socket_name, base::TaskRunner* task_runner)
45 : task_runner_(task_runner), weak_ptr_factory_(this) {
46 GOOGLE_PROTOBUF_VERIFY_VERSION;
47 sock_ = UnixSocket::Connect(socket_name, this, task_runner);
48 }
49
~ClientImpl()50 ClientImpl::~ClientImpl() {
51 // Ensure we are not destroyed in the middle of invoking a reply.
52 PERFETTO_DCHECK(!invoking_method_reply_);
53 OnDisconnect(nullptr); // The UnixSocket* ptr is not used in OnDisconnect().
54 }
55
BindService(base::WeakPtr<ServiceProxy> service_proxy)56 void ClientImpl::BindService(base::WeakPtr<ServiceProxy> service_proxy) {
57 if (!service_proxy)
58 return;
59 if (!sock_->is_connected())
60 return queued_bindings_.emplace_back(service_proxy);
61 RequestID request_id = ++last_request_id_;
62 Frame frame;
63 frame.set_request_id(request_id);
64 Frame::BindService* req = frame.mutable_msg_bind_service();
65 const char* const service_name = service_proxy->GetDescriptor().service_name;
66 req->set_service_name(service_name);
67 if (!SendFrame(frame)) {
68 PERFETTO_DLOG("BindService(%s) failed", service_name);
69 return service_proxy->OnConnect(false /* success */);
70 }
71 QueuedRequest qr;
72 qr.type = Frame::kMsgBindService;
73 qr.request_id = request_id;
74 qr.service_proxy = service_proxy;
75 queued_requests_.emplace(request_id, std::move(qr));
76 }
77
UnbindService(ServiceID service_id)78 void ClientImpl::UnbindService(ServiceID service_id) {
79 service_bindings_.erase(service_id);
80 }
81
BeginInvoke(ServiceID service_id,const std::string & method_name,MethodID remote_method_id,const ProtoMessage & method_args,bool drop_reply,base::WeakPtr<ServiceProxy> service_proxy,int fd)82 RequestID ClientImpl::BeginInvoke(ServiceID service_id,
83 const std::string& method_name,
84 MethodID remote_method_id,
85 const ProtoMessage& method_args,
86 bool drop_reply,
87 base::WeakPtr<ServiceProxy> service_proxy,
88 int fd) {
89 std::string args_proto;
90 RequestID request_id = ++last_request_id_;
91 Frame frame;
92 frame.set_request_id(request_id);
93 Frame::InvokeMethod* req = frame.mutable_msg_invoke_method();
94 req->set_service_id(service_id);
95 req->set_method_id(remote_method_id);
96 req->set_drop_reply(drop_reply);
97 bool did_serialize = method_args.SerializeToString(&args_proto);
98 req->set_args_proto(args_proto);
99 if (!did_serialize || !SendFrame(frame, fd)) {
100 PERFETTO_DLOG("BeginInvoke() failed while sending the frame");
101 return 0;
102 }
103 if (drop_reply)
104 return 0;
105 QueuedRequest qr;
106 qr.type = Frame::kMsgInvokeMethod;
107 qr.request_id = request_id;
108 qr.method_name = method_name;
109 qr.service_proxy = std::move(service_proxy);
110 queued_requests_.emplace(request_id, std::move(qr));
111 return request_id;
112 }
113
SendFrame(const Frame & frame,int fd)114 bool ClientImpl::SendFrame(const Frame& frame, int fd) {
115 // Serialize the frame into protobuf, add the size header, and send it.
116 std::string buf = BufferedFrameDeserializer::Serialize(frame);
117
118 // TODO(primiano): this should do non-blocking I/O. But then what if the
119 // socket buffer is full? We might want to either drop the request or throttle
120 // the send and PostTask the reply later? Right now we are making Send()
121 // blocking as a workaround. Propagate bakpressure to the caller instead.
122 bool res = sock_->Send(buf.data(), buf.size(), fd,
123 UnixSocket::BlockingMode::kBlocking);
124 PERFETTO_CHECK(res || !sock_->is_connected());
125 return res;
126 }
127
OnConnect(UnixSocket *,bool connected)128 void ClientImpl::OnConnect(UnixSocket*, bool connected) {
129 // Drain the BindService() calls that were queued before establishig the
130 // connection with the host.
131 for (base::WeakPtr<ServiceProxy>& service_proxy : queued_bindings_) {
132 if (connected) {
133 BindService(service_proxy);
134 } else if (service_proxy) {
135 service_proxy->OnConnect(false /* success */);
136 }
137 }
138 queued_bindings_.clear();
139 }
140
OnDisconnect(UnixSocket *)141 void ClientImpl::OnDisconnect(UnixSocket*) {
142 for (auto it : service_bindings_) {
143 base::WeakPtr<ServiceProxy>& service_proxy = it.second;
144 task_runner_->PostTask([service_proxy] {
145 if (service_proxy)
146 service_proxy->OnDisconnect();
147 });
148 }
149 service_bindings_.clear();
150 queued_bindings_.clear();
151 }
152
OnDataAvailable(UnixSocket *)153 void ClientImpl::OnDataAvailable(UnixSocket*) {
154 size_t rsize;
155 do {
156 auto buf = frame_deserializer_.BeginReceive();
157 base::ScopedFile fd;
158 rsize = sock_->Receive(buf.data, buf.size, &fd);
159 if (fd) {
160 PERFETTO_DCHECK(!received_fd_);
161 int res = fcntl(*fd, F_SETFD, FD_CLOEXEC);
162 PERFETTO_DCHECK(res == 0);
163 received_fd_ = std::move(fd);
164 }
165 if (!frame_deserializer_.EndReceive(rsize)) {
166 // The endpoint tried to send a frame that is way too large.
167 return sock_->Shutdown(true); // In turn will trigger an OnDisconnect().
168 // TODO(fmayer): check this.
169 }
170 } while (rsize > 0);
171
172 while (std::unique_ptr<Frame> frame = frame_deserializer_.PopNextFrame())
173 OnFrameReceived(*frame);
174 }
175
OnFrameReceived(const Frame & frame)176 void ClientImpl::OnFrameReceived(const Frame& frame) {
177 auto queued_requests_it = queued_requests_.find(frame.request_id());
178 if (queued_requests_it == queued_requests_.end()) {
179 PERFETTO_DLOG("OnFrameReceived(): got invalid request_id=%" PRIu64,
180 static_cast<uint64_t>(frame.request_id()));
181 return;
182 }
183 QueuedRequest req = std::move(queued_requests_it->second);
184 queued_requests_.erase(queued_requests_it);
185
186 if (req.type == Frame::kMsgBindService &&
187 frame.msg_case() == Frame::kMsgBindServiceReply) {
188 return OnBindServiceReply(std::move(req), frame.msg_bind_service_reply());
189 }
190 if (req.type == Frame::kMsgInvokeMethod &&
191 frame.msg_case() == Frame::kMsgInvokeMethodReply) {
192 return OnInvokeMethodReply(std::move(req), frame.msg_invoke_method_reply());
193 }
194 if (frame.msg_case() == Frame::kMsgRequestError) {
195 PERFETTO_DLOG("Host error: %s", frame.msg_request_error().error().c_str());
196 return;
197 }
198
199 PERFETTO_DLOG(
200 "OnFrameReceived() request msg_type=%d, received msg_type=%d in reply to "
201 "request_id=%" PRIu64,
202 req.type, frame.msg_case(), static_cast<uint64_t>(frame.request_id()));
203 }
204
OnBindServiceReply(QueuedRequest req,const Frame::BindServiceReply & reply)205 void ClientImpl::OnBindServiceReply(QueuedRequest req,
206 const Frame::BindServiceReply& reply) {
207 base::WeakPtr<ServiceProxy>& service_proxy = req.service_proxy;
208 if (!service_proxy)
209 return;
210 const char* svc_name = service_proxy->GetDescriptor().service_name;
211 if (!reply.success()) {
212 PERFETTO_DLOG("BindService(): unknown service_name=\"%s\"", svc_name);
213 return service_proxy->OnConnect(false /* success */);
214 }
215
216 auto prev_service = service_bindings_.find(reply.service_id());
217 if (prev_service != service_bindings_.end() && prev_service->second.get()) {
218 PERFETTO_DLOG(
219 "BindService(): Trying to bind service \"%s\" but another service "
220 "named \"%s\" is already bound with the same ID.",
221 svc_name, prev_service->second->GetDescriptor().service_name);
222 return service_proxy->OnConnect(false /* success */);
223 }
224
225 // Build the method [name] -> [remote_id] map.
226 std::map<std::string, MethodID> methods;
227 for (const auto& method : reply.methods()) {
228 if (method.name().empty() || method.id() <= 0) {
229 PERFETTO_DLOG("OnBindServiceReply(): invalid method \"%s\" -> %" PRIu64,
230 method.name().c_str(), static_cast<uint64_t>(method.id()));
231 continue;
232 }
233 methods[method.name()] = method.id();
234 }
235 service_proxy->InitializeBinding(weak_ptr_factory_.GetWeakPtr(),
236 reply.service_id(), std::move(methods));
237 service_bindings_[reply.service_id()] = service_proxy;
238 service_proxy->OnConnect(true /* success */);
239 }
240
OnInvokeMethodReply(QueuedRequest req,const Frame::InvokeMethodReply & reply)241 void ClientImpl::OnInvokeMethodReply(QueuedRequest req,
242 const Frame::InvokeMethodReply& reply) {
243 base::WeakPtr<ServiceProxy> service_proxy = req.service_proxy;
244 if (!service_proxy)
245 return;
246 std::unique_ptr<ProtoMessage> decoded_reply;
247 if (reply.success()) {
248 // TODO(fmayer): this could be optimized, stop doing method name string
249 // lookups.
250 for (const auto& method : service_proxy->GetDescriptor().methods) {
251 if (req.method_name == method.name) {
252 decoded_reply = method.reply_proto_decoder(reply.reply_proto());
253 break;
254 }
255 }
256 }
257 const RequestID request_id = req.request_id;
258 invoking_method_reply_ = true;
259 service_proxy->EndInvoke(request_id, std::move(decoded_reply),
260 reply.has_more());
261 invoking_method_reply_ = false;
262
263 // If this is a streaming method and future replies will be resolved, put back
264 // the |req| with the callback into the set of active requests.
265 if (reply.has_more())
266 queued_requests_.emplace(request_id, std::move(req));
267 }
268
269 ClientImpl::QueuedRequest::QueuedRequest() = default;
270
TakeReceivedFD()271 base::ScopedFile ClientImpl::TakeReceivedFD() {
272 return std::move(received_fd_);
273 }
274
275 } // namespace ipc
276 } // namespace perfetto
277