Line data Source code
1 : /*
2 : * If not stated otherwise in this file or this component's LICENSE file the
3 : * following copyright and licenses apply:
4 : *
5 : * Copyright 2022 Sky UK
6 : *
7 : * Licensed under the Apache License, Version 2.0 (the "License");
8 : * you may not use this file except in compliance with the License.
9 : * You may obtain a copy of the License at
10 : *
11 : * http://www.apache.org/licenses/LICENSE-2.0
12 : *
13 : * Unless required by applicable law or agreed to in writing, software
14 : * distributed under the License is distributed on an "AS IS" BASIS,
15 : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 : * See the License for the specific language governing permissions and
17 : * limitations under the License.
18 : */
19 :
20 : #include "ControlIpc.h"
21 : #include "RialtoClientLogging.h"
22 : #include "SchemaVersion.h"
23 :
24 : namespace
25 : {
26 : firebolt::rialto::ApplicationState
27 4 : convertApplicationState(const firebolt::rialto::ApplicationStateChangeEvent_ApplicationState &state)
28 : {
29 4 : switch (state)
30 : {
31 1 : case firebolt::rialto::ApplicationStateChangeEvent_ApplicationState_RUNNING:
32 1 : return firebolt::rialto::ApplicationState::RUNNING;
33 1 : case firebolt::rialto::ApplicationStateChangeEvent_ApplicationState_INACTIVE:
34 1 : return firebolt::rialto::ApplicationState::INACTIVE;
35 2 : case firebolt::rialto::ApplicationStateChangeEvent_ApplicationState_UNKNOWN:
36 2 : break;
37 : }
38 2 : return firebolt::rialto::ApplicationState::UNKNOWN;
39 : }
40 : } // namespace
41 :
42 : namespace firebolt::rialto::client
43 : {
44 1 : std::shared_ptr<IControlIpcFactory> IControlIpcFactory::createFactory()
45 : {
46 1 : return ControlIpcFactory::createFactory();
47 : }
48 :
49 1 : std::shared_ptr<ControlIpcFactory> ControlIpcFactory::createFactory()
50 : {
51 1 : std::shared_ptr<ControlIpcFactory> factory;
52 :
53 : try
54 : {
55 1 : factory = std::make_shared<ControlIpcFactory>();
56 : }
57 0 : catch (const std::exception &e)
58 : {
59 0 : RIALTO_CLIENT_LOG_ERROR("Failed to create the rialto control ipc factory, reason: %s", e.what());
60 : }
61 :
62 1 : return factory;
63 : }
64 :
65 1 : std::shared_ptr<IControlIpc> ControlIpcFactory::createControlIpc(IControlClient *controlClient)
66 : {
67 1 : auto &ipcClient{IIpcClientAccessor::instance().getIpcClient()};
68 : auto controlIpc{std::make_shared<ControlIpc>(controlClient, ipcClient,
69 0 : firebolt::rialto::common::IEventThreadFactory::createFactory())};
70 0 : ipcClient.registerConnectionObserver(controlIpc);
71 0 : return controlIpc;
72 : }
73 :
74 25 : ControlIpc::ControlIpc(IControlClient *controlClient, IIpcClient &ipcClient,
75 25 : const std::shared_ptr<common::IEventThreadFactory> &eventThreadFactory)
76 25 : : IpcModule(ipcClient), m_controlClient(controlClient),
77 75 : m_eventThread(eventThreadFactory->createEventThread("rialto-control-events"))
78 : {
79 25 : if (!attachChannel())
80 : {
81 3 : throw std::runtime_error("Failed attach to the ipc channel");
82 : }
83 37 : }
84 :
85 22 : ControlIpc::~ControlIpc()
86 : {
87 : // detach the Ipc channel
88 22 : detachChannel();
89 :
90 : // destroy the thread processing async notifications
91 22 : m_eventThread.reset();
92 : }
93 :
94 4 : bool ControlIpc::getSharedMemory(int32_t &fd, uint32_t &size)
95 : {
96 4 : if (!reattachChannelIfRequired())
97 : {
98 1 : RIALTO_CLIENT_LOG_ERROR("Reattachment of the ipc channel failed, ipc disconnected");
99 1 : return false;
100 : }
101 :
102 3 : std::shared_ptr<::firebolt::rialto::ControlModule_Stub> controlStub = m_controlStub;
103 :
104 3 : firebolt::rialto::GetSharedMemoryRequest request;
105 :
106 3 : firebolt::rialto::GetSharedMemoryResponse response;
107 3 : auto ipcController = m_ipc.createRpcController();
108 3 : auto blockingClosure = m_ipc.createBlockingClosure();
109 3 : controlStub->getSharedMemory(ipcController.get(), &request, &response, blockingClosure.get());
110 :
111 : // wait for the call to complete
112 3 : blockingClosure->wait();
113 :
114 : // check the result
115 3 : if (ipcController->Failed())
116 : {
117 1 : RIALTO_CLIENT_LOG_ERROR("failed to get the shared memory due to '%s'", ipcController->ErrorText().c_str());
118 1 : return false;
119 : }
120 :
121 2 : fd = response.fd();
122 2 : size = response.size();
123 :
124 2 : return true;
125 3 : }
126 :
127 17 : bool ControlIpc::registerClient()
128 : {
129 17 : const auto kCurrentSchemaVersion{common::getCurrentSchemaVersion()};
130 :
131 17 : if (!reattachChannelIfRequired())
132 : {
133 1 : RIALTO_CLIENT_LOG_ERROR("Reattachment of the ipc channel failed, ipc disconnected");
134 1 : return false;
135 : }
136 16 : std::shared_ptr<::firebolt::rialto::ControlModule_Stub> controlStub = m_controlStub;
137 :
138 16 : firebolt::rialto::RegisterClientRequest request;
139 16 : firebolt::rialto::RegisterClientResponse response;
140 16 : firebolt::rialto::SchemaVersion clientSchemaVersion;
141 16 : request.mutable_client_schema_version()->set_major(kCurrentSchemaVersion.major());
142 16 : request.mutable_client_schema_version()->set_minor(kCurrentSchemaVersion.minor());
143 16 : request.mutable_client_schema_version()->set_patch(kCurrentSchemaVersion.patch());
144 16 : auto ipcController = m_ipc.createRpcController();
145 16 : auto blockingClosure = m_ipc.createBlockingClosure();
146 16 : controlStub->registerClient(ipcController.get(), &request, &response, blockingClosure.get());
147 :
148 : // wait for the call to complete
149 16 : blockingClosure->wait();
150 :
151 : // check the result
152 16 : if (ipcController->Failed())
153 : {
154 1 : RIALTO_CLIENT_LOG_ERROR("failed to register client due to '%s'", ipcController->ErrorText().c_str());
155 1 : return false;
156 : }
157 :
158 15 : m_controlHandle = response.control_handle();
159 :
160 15 : if (!response.has_server_schema_version())
161 : {
162 12 : RIALTO_CLIENT_LOG_WARN("Server proto schema version not known");
163 12 : return true;
164 : }
165 :
166 3 : const common::SchemaVersion kServerSchemaVersion{response.server_schema_version().major(),
167 6 : response.server_schema_version().minor(),
168 3 : response.server_schema_version().patch()};
169 3 : if (kCurrentSchemaVersion == kServerSchemaVersion)
170 : {
171 1 : RIALTO_CLIENT_LOG_DEBUG("Server and Client proto schema versions are equal");
172 : }
173 2 : else if (kCurrentSchemaVersion.isCompatible(kServerSchemaVersion))
174 : {
175 1 : RIALTO_CLIENT_LOG_INFO("Server and Client proto schema versions are compatible. Server schema version: %s, "
176 : "Client schema version: %s",
177 : kServerSchemaVersion.str().c_str(), kCurrentSchemaVersion.str().c_str());
178 : }
179 : else
180 : {
181 1 : RIALTO_CLIENT_LOG_ERROR("Server and Client proto schema versions are not compatible. Server schema version: "
182 : "%s, Client schema version: %s",
183 : kServerSchemaVersion.str().c_str(), kCurrentSchemaVersion.str().c_str());
184 1 : return false;
185 : }
186 :
187 2 : return true;
188 16 : }
189 :
190 26 : bool ControlIpc::createRpcStubs(const std::shared_ptr<ipc::IChannel> &ipcChannel)
191 : {
192 26 : m_controlStub = std::make_unique<::firebolt::rialto::ControlModule_Stub>(ipcChannel.get());
193 26 : if (!m_controlStub)
194 : {
195 0 : return false;
196 : }
197 26 : return true;
198 : }
199 :
200 26 : bool ControlIpc::subscribeToEvents(const std::shared_ptr<ipc::IChannel> &ipcChannel)
201 : {
202 26 : if (!ipcChannel)
203 : {
204 0 : return false;
205 : }
206 :
207 52 : int eventTag = ipcChannel->subscribe<firebolt::rialto::ApplicationStateChangeEvent>(
208 26 : [this](const std::shared_ptr<firebolt::rialto::ApplicationStateChangeEvent> &event)
209 30 : { m_eventThread->add(&ControlIpc::onApplicationStateUpdated, this, event); });
210 26 : if (eventTag < 0)
211 0 : return false;
212 26 : m_eventTags.push_back(eventTag);
213 :
214 52 : eventTag = ipcChannel->subscribe<firebolt::rialto::PingEvent>(
215 26 : [this](const std::shared_ptr<firebolt::rialto::PingEvent> &event)
216 5 : { m_eventThread->add(&ControlIpc::onPing, this, event); });
217 26 : if (eventTag < 0)
218 1 : return false;
219 25 : m_eventTags.push_back(eventTag);
220 :
221 25 : return true;
222 : }
223 :
224 1 : void ControlIpc::onConnectionBroken()
225 : {
226 2 : m_eventThread->add([this]() { m_controlClient->notifyApplicationState(ApplicationState::UNKNOWN); });
227 1 : }
228 :
229 4 : void ControlIpc::onApplicationStateUpdated(const std::shared_ptr<firebolt::rialto::ApplicationStateChangeEvent> &event)
230 : {
231 : // It's possible, that ApplicationStateChangeEvent comes before RegisterClientResponse, so do not check control_handle() here.
232 4 : m_controlClient->notifyApplicationState(convertApplicationState(event->application_state()));
233 : }
234 :
235 5 : void ControlIpc::onPing(const std::shared_ptr<firebolt::rialto::PingEvent> &event)
236 : {
237 5 : if (m_controlHandle != event->control_handle())
238 : {
239 1 : RIALTO_CLIENT_LOG_WARN("PingEvent received with wrong handle");
240 2 : return;
241 : }
242 :
243 4 : if (!reattachChannelIfRequired())
244 : {
245 1 : RIALTO_CLIENT_LOG_ERROR("Reattachment of the ipc channel failed, ipc disconnected");
246 1 : return;
247 : }
248 3 : firebolt::rialto::AckRequest request;
249 3 : request.set_control_handle(m_controlHandle);
250 3 : request.set_id(event->id());
251 3 : firebolt::rialto::AckResponse response;
252 3 : auto ipcController = m_ipc.createRpcController();
253 3 : auto blockingClosure = m_ipc.createBlockingClosure();
254 3 : m_controlStub->ack(ipcController.get(), &request, &response, blockingClosure.get());
255 :
256 : // wait for the call to complete
257 3 : blockingClosure->wait();
258 :
259 : // check the result
260 3 : if (ipcController->Failed())
261 : {
262 1 : RIALTO_CLIENT_LOG_ERROR("failed to ack due to '%s'", ipcController->ErrorText().c_str());
263 : }
264 3 : }
265 : }; // namespace firebolt::rialto::client
|
