2 * Copyright (c) 2014 Samsung Electronics Co., Ltd All Rights Reserved
4 * Contact: Jan Olszak <j.olszak@samsung.com>
6 * Licensed under the Apache License, Version 2.0 (the "License");
7 * you may not use this file except in compliance with the License.
8 * You may obtain a copy of the License at
10 * http://www.apache.org/licenses/LICENSE-2.0
12 * Unless required by applicable law or agreed to in writing, software
13 * distributed under the License is distributed on an "AS IS" BASIS,
14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 * See the License for the specific language governing permissions and
16 * limitations under the License
21 * @author Jan Olszak (j.olszak@samsung.com)
22 * @brief Data and event processing thread
27 #include "ipc/exception.hpp"
28 #include "ipc/internals/processor.hpp"
29 #include "ipc/internals/utils.hpp"
35 #include <sys/socket.h>
38 namespace security_containers {
41 #define IGNORE_EXCEPTIONS(expr) \
46 catch (const std::exception& e){ \
47 LOGE("Callback threw an error: " << e.what()); \
50 const MethodID Processor::RETURN_METHOD_ID = std::numeric_limits<MethodID>::max();
51 const MethodID Processor::REGISTER_SIGNAL_METHOD_ID = std::numeric_limits<MethodID>::max() - 1;
53 Processor::Processor(const PeerCallback& newPeerCallback,
54 const PeerCallback& removedPeerCallback,
55 const unsigned int maxNumberOfPeers)
56 : mNewPeerCallback(newPeerCallback),
57 mRemovedPeerCallback(removedPeerCallback),
58 mMaxNumberOfPeers(maxNumberOfPeers)
60 LOGT("Creating Processor");
61 using namespace std::placeholders;
63 addMethodHandlerInternal<EmptyData, RegisterSignalsMessage>(REGISTER_SIGNAL_METHOD_ID,
64 std::bind(&Processor::onNewSignals, this, _1, _2));
68 Processor::~Processor()
70 LOGT("Destroying Processor");
73 } catch (IPCException& e) {
74 LOGE("Error in Processor's destructor: " << e.what());
76 LOGT("Destroyed Processor");
79 bool Processor::isStarted()
81 return mThread.joinable();
84 void Processor::start()
86 LOGT("Starting Processor");
88 mThread = std::thread(&Processor::run, this);
90 LOGT("Started Processor");
93 void Processor::stop()
95 LOGT("Stopping Processor");
97 mEventQueue.send(Event::FINISH);
100 LOGT("Stopped Processor");
103 void Processor::setNewPeerCallback(const PeerCallback& newPeerCallback)
105 Lock lock(mCallbacksMutex);
106 mNewPeerCallback = newPeerCallback;
109 void Processor::setRemovedPeerCallback(const PeerCallback& removedPeerCallback)
111 Lock lock(mCallbacksMutex);
112 mRemovedPeerCallback = removedPeerCallback;
115 void Processor::removeMethod(const MethodID methodID)
117 LOGT("Removing method " << methodID);
118 Lock lock(mCallsMutex);
119 mMethodsCallbacks.erase(methodID);
122 FileDescriptor Processor::addPeer(const std::shared_ptr<Socket>& socketPtr)
124 LOGT("Adding socket");
125 FileDescriptor peerFD;
127 Lock lock(mSocketsMutex);
128 peerFD = socketPtr->getFD();
129 SocketInfo socketInfo(peerFD, std::move(socketPtr));
130 mNewSockets.push(std::move(socketInfo));
132 LOGI("New peer added. Id: " << peerFD);
133 mEventQueue.send(Event::ADD_PEER);
138 void Processor::removePeer(const FileDescriptor peerFD)
140 std::shared_ptr<std::condition_variable> conditionPtr(new std::condition_variable());
143 Lock lock(mSocketsMutex);
144 RemovePeerRequest request(peerFD, conditionPtr);
145 mPeersToDelete.push(std::move(request));
148 mEventQueue.send(Event::REMOVE_PEER);
150 auto isPeerDeleted = [&peerFD, this]()->bool {
151 Lock lock(mSocketsMutex);
152 return mSockets.count(peerFD) == 0;
156 std::unique_lock<std::mutex> lock(mutex);
157 conditionPtr->wait(lock, isPeerDeleted);
160 void Processor::removePeerInternal(const FileDescriptor peerFD, Status status)
162 LOGW("Removing peer. ID: " << peerFD);
164 Lock lock(mSocketsMutex);
165 mSockets.erase(peerFD);
167 // Remove from signal addressees
168 for (auto it = mSignalsPeers.begin(); it != mSignalsPeers.end();) {
169 it->second.remove(peerFD);
170 if (it->second.empty()) {
171 it = mSignalsPeers.erase(it);
179 // Erase associated return value callbacks
180 Lock lock(mReturnCallbacksMutex);
182 std::shared_ptr<void> data;
183 for (auto it = mReturnCallbacks.begin(); it != mReturnCallbacks.end();) {
184 if (it->second.peerFD == peerFD) {
185 IGNORE_EXCEPTIONS(it->second.process(status, data));
186 it = mReturnCallbacks.erase(it);
195 Lock lock(mCallbacksMutex);
196 if (mRemovedPeerCallback) {
197 // Notify about the deletion
198 mRemovedPeerCallback(peerFD);
205 void Processor::resetPolling()
207 LOGI("Resetting polling");
208 // Setup polling on eventfd and sockets
209 Lock lock(mSocketsMutex);
210 mFDs.resize(mSockets.size() + 1);
212 mFDs[0].fd = mEventQueue.getFD();
213 mFDs[0].events = POLLIN;
215 auto socketIt = mSockets.begin();
216 for (unsigned int i = 1; i < mFDs.size(); ++i) {
217 mFDs[i].fd = socketIt->second->getFD();
218 mFDs[i].events = POLLIN | POLLHUP; // Listen for input events
220 // TODO: It's possible to block on writing to fd. Maybe listen for POLLOUT too?
224 void Processor::run()
230 LOGT("Waiting for communication...");
231 int ret = poll(mFDs.data(), mFDs.size(), -1 /*blocking call*/);
232 LOGT("... incoming communication!");
233 if (ret == -1 || ret == 0) {
234 if (errno == EINTR) {
237 LOGE("Error in poll: " << std::string(strerror(errno)));
238 throw IPCException("Error in poll: " + std::string(strerror(errno)));
241 // Check for lost connections:
242 if (handleLostConnections()) {
247 // Check for incoming data.
248 if (handleInputs()) {
253 // Check for incoming events
260 cleanCommunication();
264 bool Processor::handleLostConnections()
266 std::vector<FileDescriptor> peersToRemove;
269 Lock lock(mSocketsMutex);
270 for (unsigned int i = 1; i < mFDs.size(); ++i) {
271 if (mFDs[i].revents & POLLHUP) {
272 LOGI("Lost connection to peer: " << mFDs[i].fd);
273 mFDs[i].revents &= ~(POLLHUP);
274 peersToRemove.push_back(mFDs[i].fd);
279 for (const FileDescriptor peerFD : peersToRemove) {
280 removePeerInternal(peerFD, Status::PEER_DISCONNECTED);
283 return !peersToRemove.empty();
286 bool Processor::handleInputs()
288 std::vector<std::shared_ptr<Socket>> socketsWithInput;
290 Lock lock(mSocketsMutex);
291 for (unsigned int i = 1; i < mFDs.size(); ++i) {
292 if (mFDs[i].revents & POLLIN) {
293 mFDs[i].revents &= ~(POLLIN);
294 socketsWithInput.push_back(mSockets[mFDs[i].fd]);
299 bool pollChanged = false;
300 // Handle input outside the critical section
301 for (const auto& socketPtr : socketsWithInput) {
302 pollChanged = pollChanged || handleInput(*socketPtr);
307 bool Processor::handleInput(const Socket& socket)
309 LOGT("Handle incoming data");
313 Socket::Guard guard = socket.getGuard();
314 socket.read(&methodID, sizeof(methodID));
315 socket.read(&messageID, sizeof(messageID));
317 if (methodID == RETURN_METHOD_ID) {
318 return onReturnValue(socket, messageID);
321 Lock lock(mCallsMutex);
322 if (mMethodsCallbacks.count(methodID)) {
324 std::shared_ptr<MethodHandlers> methodCallbacks = mMethodsCallbacks.at(methodID);
326 return onRemoteCall(socket, methodID, messageID, methodCallbacks);
328 } else if (mSignalsCallbacks.count(methodID)) {
330 std::shared_ptr<SignalHandlers> signalCallbacks = mSignalsCallbacks.at(methodID);
332 return onRemoteSignal(socket, methodID, messageID, signalCallbacks);
337 LOGW("No method or signal callback for methodID: " << methodID);
338 removePeerInternal(socket.getFD(), Status::NAUGHTY_PEER);
347 std::shared_ptr<Processor::EmptyData> Processor::onNewSignals(const FileDescriptor peerFD,
348 std::shared_ptr<RegisterSignalsMessage>& data)
350 LOGD("New signals for peer: " << peerFD);
351 Lock lock(mSocketsMutex);
352 for (MethodID methodID : data->ids) {
353 mSignalsPeers[methodID].push_back(peerFD);
356 return std::make_shared<EmptyData>();
359 bool Processor::onReturnValue(const Socket& socket,
360 const MessageID messageID)
362 LOGI("Return value for messageID: " << messageID);
363 ReturnCallbacks returnCallbacks;
365 Lock lock(mReturnCallbacksMutex);
366 LOGT("Getting the return callback");
367 returnCallbacks = std::move(mReturnCallbacks.at(messageID));
368 mReturnCallbacks.erase(messageID);
369 } catch (const std::out_of_range&) {
370 LOGW("No return callback for messageID: " << messageID);
371 removePeerInternal(socket.getFD(), Status::NAUGHTY_PEER);
375 std::shared_ptr<void> data;
377 LOGT("Parsing incoming return data");
378 data = returnCallbacks.parse(socket.getFD());
379 } catch (const std::exception& e) {
380 LOGE("Exception during parsing: " << e.what());
381 IGNORE_EXCEPTIONS(returnCallbacks.process(Status::PARSING_ERROR, data));
382 removePeerInternal(socket.getFD(), Status::PARSING_ERROR);
386 LOGT("Process return value callback for messageID: " << messageID);
387 IGNORE_EXCEPTIONS(returnCallbacks.process(Status::OK, data));
392 bool Processor::onRemoteSignal(const Socket& socket,
393 const MethodID methodID,
394 const MessageID messageID,
395 std::shared_ptr<SignalHandlers> signalCallbacks)
397 LOGI("Remote signal; methodID: " << methodID << " messageID: " << messageID);
399 std::shared_ptr<void> data;
401 LOGT("Parsing incoming data");
402 data = signalCallbacks->parse(socket.getFD());
403 } catch (const std::exception& e) {
404 LOGE("Exception during parsing: " << e.what());
405 removePeerInternal(socket.getFD(), Status::PARSING_ERROR);
409 LOGT("Signal callback for methodID: " << methodID << "; messageID: " << messageID);
411 signalCallbacks->signal(socket.getFD(), data);
412 } catch (const std::exception& e) {
413 LOGE("Exception in method handler: " << e.what());
414 removePeerInternal(socket.getFD(), Status::NAUGHTY_PEER);
421 bool Processor::onRemoteCall(const Socket& socket,
422 const MethodID methodID,
423 const MessageID messageID,
424 std::shared_ptr<MethodHandlers> methodCallbacks)
426 LOGI("Remote call; methodID: " << methodID << " messageID: " << messageID);
428 std::shared_ptr<void> data;
430 LOGT("Parsing incoming data");
431 data = methodCallbacks->parse(socket.getFD());
432 } catch (const std::exception& e) {
433 LOGE("Exception during parsing: " << e.what());
434 removePeerInternal(socket.getFD(), Status::PARSING_ERROR);
438 LOGT("Process callback for methodID: " << methodID << "; messageID: " << messageID);
439 std::shared_ptr<void> returnData;
441 returnData = methodCallbacks->method(socket.getFD(), data);
442 } catch (const std::exception& e) {
443 LOGE("Exception in method handler: " << e.what());
444 removePeerInternal(socket.getFD(), Status::NAUGHTY_PEER);
448 LOGT("Sending return data; methodID: " << methodID << "; messageID: " << messageID);
450 // Send the call with the socket
451 Socket::Guard guard = socket.getGuard();
452 socket.write(&RETURN_METHOD_ID, sizeof(RETURN_METHOD_ID));
453 socket.write(&messageID, sizeof(messageID));
454 methodCallbacks->serialize(socket.getFD(), returnData);
455 } catch (const std::exception& e) {
456 LOGE("Exception during serialization: " << e.what());
457 removePeerInternal(socket.getFD(), Status::SERIALIZATION_ERROR);
464 bool Processor::handleEvent()
466 if (!(mFDs[0].revents & POLLIN)) {
471 mFDs[0].revents &= ~(POLLIN);
473 switch (mEventQueue.receive()) {
474 case Event::FINISH: {
475 LOGD("Event FINISH");
485 case Event::ADD_PEER: {
486 LOGD("Event ADD_PEER");
490 case Event::REMOVE_PEER: {
491 LOGD("Event REMOVE_PEER");
492 return onRemovePeer();
499 bool Processor::onNewPeer()
501 SocketInfo socketInfo;
503 Lock lock(mSocketsMutex);
505 socketInfo = std::move(mNewSockets.front());
508 if (mSockets.size() > mMaxNumberOfPeers) {
509 LOGE("There are too many peers. I don't accept the connection with " << socketInfo.peerFD);
512 if (mSockets.count(socketInfo.peerFD) != 0) {
513 LOGE("There already was a socket for peerFD: " << socketInfo.peerFD);
517 mSockets[socketInfo.peerFD] = std::move(socketInfo.socketPtr);
521 // Broadcast the new signal to peers
522 LOGW("Sending handled signals");
523 std::list<FileDescriptor> peersIDs;
525 Lock lock(mSocketsMutex);
526 for (const auto kv : mSockets) {
527 peersIDs.push_back(kv.first);
531 std::vector<MethodID> ids;
533 Lock lock(mSocketsMutex);
534 for (const auto kv : mSignalsCallbacks) {
535 ids.push_back(kv.first);
538 auto data = std::make_shared<RegisterSignalsMessage>(ids);
540 for (const FileDescriptor peerFD : peersIDs) {
541 callInternal<RegisterSignalsMessage, EmptyData>(REGISTER_SIGNAL_METHOD_ID,
544 discardResultHandler<EmptyData>);
546 LOGW("Sent handled signals");
552 Lock lock(mCallbacksMutex);
553 if (mNewPeerCallback) {
554 // Notify about the new user.
555 mNewPeerCallback(socketInfo.peerFD);
561 bool Processor::onRemovePeer()
563 RemovePeerRequest request;
565 Lock lock(mSocketsMutex);
566 request = std::move(mPeersToDelete.front());
567 mPeersToDelete.pop();
570 removePeerInternal(request.peerFD, Status::REMOVED_PEER);
571 request.conditionPtr->notify_all();
575 CallQueue::Call Processor::getCall()
577 Lock lock(mCallsMutex);
581 bool Processor::onCall()
583 LOGT("Handle call (from another thread) to send a message.");
584 CallQueue::Call call = getCall();
586 std::shared_ptr<Socket> socketPtr;
588 // Get the peer's socket
589 Lock lock(mSocketsMutex);
590 socketPtr = mSockets.at(call.peerFD);
591 } catch (const std::out_of_range&) {
592 LOGE("Peer disconnected. No socket with a peerFD: " << call.peerFD);
593 IGNORE_EXCEPTIONS(call.process(Status::PEER_DISCONNECTED, call.data));
597 if (call.parse && call.process) {
598 // Set what to do with the return message, but only if needed
599 Lock lock(mReturnCallbacksMutex);
600 if (mReturnCallbacks.count(call.messageID) != 0) {
601 LOGE("There already was a return callback for messageID: " << call.messageID);
603 mReturnCallbacks[call.messageID] = std::move(ReturnCallbacks(call.peerFD,
604 std::move(call.parse),
605 std::move(call.process)));
609 // Send the call with the socket
610 Socket::Guard guard = socketPtr->getGuard();
611 socketPtr->write(&call.methodID, sizeof(call.methodID));
612 socketPtr->write(&call.messageID, sizeof(call.messageID));
613 call.serialize(socketPtr->getFD(), call.data);
614 } catch (const std::exception& e) {
615 LOGE("Error during sending a message: " << e.what());
617 // Inform about the error
618 IGNORE_EXCEPTIONS(mReturnCallbacks[call.messageID].process(Status::SERIALIZATION_ERROR, call.data));
621 Lock lock(mReturnCallbacksMutex);
622 mReturnCallbacks.erase(call.messageID);
625 removePeerInternal(call.peerFD, Status::SERIALIZATION_ERROR);
632 void Processor::cleanCommunication()
634 while (!mEventQueue.isEmpty()) {
635 switch (mEventQueue.receive()) {
636 case Event::FINISH: {
637 LOGD("Event FINISH after FINISH");
641 LOGD("Event CALL after FINISH");
642 CallQueue::Call call = getCall();
644 IGNORE_EXCEPTIONS(call.process(Status::CLOSING, call.data));
649 case Event::ADD_PEER: {
650 LOGD("Event ADD_PEER after FINISH");
654 case Event::REMOVE_PEER: {
655 LOGD("Event REMOVE_PEER after FINISH");
656 RemovePeerRequest request;
658 Lock lock(mSocketsMutex);
659 request = std::move(mPeersToDelete.front());
660 mPeersToDelete.pop();
662 request.conditionPtr->notify_all();
670 } // namespace security_containers