diff --git a/applications/analyzer/eavesdropperthread.cpp b/applications/analyzer/eavesdropperthread.cpp index 982f2b5..c6155f7 100644 --- a/applications/analyzer/eavesdropperthread.cpp +++ b/applications/analyzer/eavesdropperthread.cpp @@ -1,96 +1,96 @@ /* Copyright (C) 2013 Andreas Hartmetz This library is free software; you can redistribute it and/or modify it under the terms of the GNU Library General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more details. You should have received a copy of the GNU Library General Public License along with this library; see the file COPYING.LGPL. If not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. Alternatively, this file is available under the Mozilla Public License Version 1.1. You may obtain a copy of the License at http://www.mozilla.org/MPL/ */ #include "eavesdropperthread.h" #include "arguments.h" #include "connectaddress.h" #include "eavesdroppermodel.h" #include "error.h" #include "eventdispatcher.h" #include "localsocket.h" #include "message.h" #include "connection.h" EavesdropperThread::EavesdropperThread(EavesdropperModel *model) { // do not parent this to the model; it doesn't work across threads moveToThread(&m_thread); connect(this, SIGNAL(messageReceived(Message *, qint64)), model, SLOT(addMessage(Message *, qint64)), Qt::QueuedConnection); connect(&m_thread, SIGNAL(started()), SLOT(run())); m_thread.start(); } EavesdropperThread::~EavesdropperThread() { m_dispatcher->interrupt(); m_thread.wait(); delete m_connection; delete m_dispatcher; } static Message createEavesdropMessage(const char *messageType) { Message ret = Message::createCall("/org/freedesktop/DBus", "org.freedesktop.DBus", "AddMatch"); ret.setDestination("org.freedesktop.DBus"); Arguments::Writer writer; std::string str = "eavesdrop=true,type="; str += messageType; writer.writeString(cstring(str.c_str())); ret.setArguments(writer.finish()); return ret; } void EavesdropperThread::run() { m_timer.start(); m_dispatcher = new EventDispatcher; m_connection = new Connection(m_dispatcher, ConnectAddress::StandardBus::Session); m_connection->setSpontaneousMessageReceiver(this); { static const int messageTypeCount = 4; const char *messageType[messageTypeCount] = { "signal", "method_call", "method_return", "error" }; for (int i = 0; i < messageTypeCount; i++) { m_connection->sendNoReply(createEavesdropMessage(messageType[i])); } } Q_ASSERT(m_connection->isConnected()); while (m_dispatcher->poll()) { } m_thread.quit(); } -void EavesdropperThread::handleSpontaneousMessageReceived(Message message) +void EavesdropperThread::handleSpontaneousMessageReceived(Message message, Connection *) { emit messageReceived(new Message(std::move(message)), m_timer.nsecsElapsed()); } diff --git a/applications/analyzer/eavesdropperthread.h b/applications/analyzer/eavesdropperthread.h index edf158d..fa54995 100644 --- a/applications/analyzer/eavesdropperthread.h +++ b/applications/analyzer/eavesdropperthread.h @@ -1,64 +1,64 @@ /* Copyright (C) 2013 Andreas Hartmetz This library is free software; you can redistribute it and/or modify it under the terms of the GNU Library General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more details. You should have received a copy of the GNU Library General Public License along with this library; see the file COPYING.LGPL. If not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. Alternatively, this file is available under the Mozilla Public License Version 1.1. You may obtain a copy of the License at http://www.mozilla.org/MPL/ */ #ifndef EAVESDROPPERTHREAD_H #define EAVESDROPPERTHREAD_H #include "imessagereceiver.h" #include #include class EavesdropperModel; class EventDispatcher; class Message; class Connection; // This is a separate thread mainly for accurate timestamps. If this was running in the main // thread, GUI and other processing would delay the calls to messageReceived() and therefore // QDateTime::currentDateTime(). class EavesdropperThread : public QObject, public IMessageReceiver { Q_OBJECT public: EavesdropperThread(EavesdropperModel *model); ~EavesdropperThread() override; // reimplemented IMessageReceiver method - void handleSpontaneousMessageReceived(Message message) override; + void handleSpontaneousMessageReceived(Message message, Connection *) override; signals: void messageReceived(Message *message, qint64 timestamp); private slots: void run(); private: QThread m_thread; QElapsedTimer m_timer; EventDispatcher *m_dispatcher; Connection *m_connection; }; #endif // EAVESDROPPERTHREAD_H diff --git a/applications/dfer.cpp b/applications/dfer.cpp index 13e6950..f7e65d3 100644 --- a/applications/dfer.cpp +++ b/applications/dfer.cpp @@ -1,108 +1,108 @@ /* Copyright (C) 2013 Andreas Hartmetz This library is free software; you can redistribute it and/or modify it under the terms of the GNU Library General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more details. You should have received a copy of the GNU Library General Public License along with this library; see the file COPYING.LGPL. If not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. Alternatively, this file is available under the Mozilla Public License Version 1.1. You may obtain a copy of the License at http://www.mozilla.org/MPL/ */ #include "arguments.h" #include "connectaddress.h" #include "error.h" #include "eventdispatcher.h" #include "imessagereceiver.h" #include "message.h" #include "connection.h" #include #include static Message createEavesdropMessage(const char *messageType) { Message ret = Message::createCall("/org/freedesktop/DBus", "org.freedesktop.DBus", "AddMatch"); ret.setDestination("org.freedesktop.DBus"); Arguments::Writer writer; std::string str = "eavesdrop=true,type="; str += messageType; writer.writeString(cstring(str.c_str())); ret.setArguments(writer.finish()); return ret; } class ReplyPrinter : public IMessageReceiver { // reimplemented from IMessageReceiver - void handleSpontaneousMessageReceived(Message m) override; + void handleSpontaneousMessageReceived(Message m, Connection *) override; }; -void ReplyPrinter::handleSpontaneousMessageReceived(Message m) +void ReplyPrinter::handleSpontaneousMessageReceived(Message m, Connection *) { std::cout << '\n' << m.prettyPrint(); } static void printHelp() { std::cout << "dfer options:\n" " --session-bus Monitor the session bus [the default]\n" " --system-bus Monitor the system bus\n" " --help Show this help and exit\n"; } int main(int argc, char *argv[]) { EventDispatcher dispatcher; ConnectAddress::StandardBus bus = ConnectAddress::StandardBus::Session; for (int i = 1; i < argc; i++) { std::string s = argv[i]; if (s == "--help") { printHelp(); exit(0); } else if (s == "--system-bus") { bus = ConnectAddress::StandardBus::System; } else if (s == "--session-bus") { bus = ConnectAddress::StandardBus::Session; } else { std::cerr << "Unknown option \"" << s << "\".\n"; printHelp(); exit(1); } } Connection connection(&dispatcher, bus); ReplyPrinter receiver; connection.setSpontaneousMessageReceiver(&receiver); { static const int messageTypeCount = 4; const char *messageType[messageTypeCount] = { "signal", "method_call", "method_return", "error" }; for (int i = 0; i < messageTypeCount; i++) { connection.sendNoReply(createEavesdropMessage(messageType[i])); } } while (true) { dispatcher.poll(); } return 0; } diff --git a/connection/connection.cpp b/connection/connection.cpp index 7d3e3d7..d49c48e 100644 --- a/connection/connection.cpp +++ b/connection/connection.cpp @@ -1,781 +1,782 @@ /* Copyright (C) 2013 Andreas Hartmetz This library is free software; you can redistribute it and/or modify it under the terms of the GNU Library General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more details. You should have received a copy of the GNU Library General Public License along with this library; see the file COPYING.LGPL. If not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. Alternatively, this file is available under the Mozilla Public License Version 1.1. You may obtain a copy of the License at http://www.mozilla.org/MPL/ */ #include "connection.h" #include "connection_p.h" #include "arguments.h" #include "authclient.h" #include "event.h" #include "eventdispatcher_p.h" #include "icompletionlistener.h" #include "imessagereceiver.h" #include "iserver.h" #include "localsocket.h" #include "message.h" #include "message_p.h" #include "pendingreply.h" #include "pendingreply_p.h" #include "stringtools.h" #include #include #include class HelloReceiver : public IMessageReceiver { public: - void handlePendingReplyFinished(PendingReply *pr) override + void handlePendingReplyFinished(PendingReply *pr, Connection *) override { assert(pr == &m_helloReply); (void) pr; m_parent->handleHelloReply(); } PendingReply m_helloReply; // keep it here so it conveniently goes away when it's done ConnectionPrivate *m_parent; }; class ClientConnectedHandler : public ICompletionListener { public: ~ClientConnectedHandler() override { delete m_server; } void handleCompletion(void *) override { m_parent->handleClientConnected(); } IServer *m_server; ConnectionPrivate *m_parent; }; ConnectionPrivate::ConnectionPrivate(Connection *connection, EventDispatcher *dispatcher) : m_state(Unconnected), m_connection(connection), m_client(nullptr), m_receivingMessage(nullptr), m_transport(nullptr), m_helloReceiver(nullptr), m_clientConnectedHandler(nullptr), m_eventDispatcher(dispatcher), m_authClient(nullptr), m_defaultTimeout(25000), m_sendSerial(1), m_mainThreadConnection(nullptr) { } Connection::Connection(EventDispatcher *dispatcher, const ConnectAddress &ca) : d(new ConnectionPrivate(this, dispatcher)) { d->m_connectAddress = ca; assert(d->m_eventDispatcher); EventDispatcherPrivate::get(d->m_eventDispatcher)->m_connectionToNotify = d; if (ca.type() == ConnectAddress::Type::None || ca.role() == ConnectAddress::Role::None) { std::cerr << "\nConnection: connection constructor Exit A\n\n"; return; } if (ca.role() == ConnectAddress::Role::PeerServer) { // this sets up a server that will be destroyed after accepting exactly one connection d->m_clientConnectedHandler = new ClientConnectedHandler; ConnectAddress dummyClientAddress; d->m_clientConnectedHandler->m_server = IServer::create(ca, &dummyClientAddress); d->m_clientConnectedHandler->m_server->setEventDispatcher(dispatcher); d->m_clientConnectedHandler->m_server->setNewConnectionListener(d->m_clientConnectedHandler); d->m_clientConnectedHandler->m_parent = d; d->m_state = ConnectionPrivate::ServerWaitingForClient; } else { d->m_transport = ITransport::create(ca); d->m_transport->setEventDispatcher(dispatcher); if (ca.role() == ConnectAddress::Role::BusClient) { d->startAuthentication(); d->m_state = ConnectionPrivate::Authenticating; } else { assert(ca.role() == ConnectAddress::Role::PeerClient); // get ready to receive messages right away d->receiveNextMessage(); d->m_state = ConnectionPrivate::Connected; } } } Connection::Connection(EventDispatcher *dispatcher, CommRef mainConnectionRef) : d(new ConnectionPrivate(this, dispatcher)) { EventDispatcherPrivate::get(d->m_eventDispatcher)->m_connectionToNotify = d; d->m_mainThreadLink = std::move(mainConnectionRef.commutex); CommutexLocker locker(&d->m_mainThreadLink); assert(locker.hasLock()); Commutex *const id = d->m_mainThreadLink.id(); if (!id) { assert(false); std::cerr << "\nConnection: slave constructor Exit A\n\n"; return; // stay in Unconnected state } // TODO how do we handle m_state? d->m_mainThreadConnection = mainConnectionRef.connection; ConnectionPrivate *mainD = d->m_mainThreadConnection; // get the current values - if we got them from e.g. the CommRef they could be outdated // and we don't want to wait for more event ping-pong SpinLocker mainLocker(&mainD->m_lock); d->m_connectAddress = mainD->m_connectAddress; // register with the main Connection SecondaryConnectionConnectEvent *evt = new SecondaryConnectionConnectEvent(); evt->connection = d; evt->id = id; EventDispatcherPrivate::get(mainD->m_eventDispatcher) ->queueEvent(std::unique_ptr(evt)); } Connection::Connection(ITransport *transport, const ConnectAddress &address) : d(new ConnectionPrivate(this, transport->eventDispatcher())) { // TODO FULLY validate address, also in the other constructors and in ITransport::create() // and in IServer::create()! assert(address.role() == ConnectAddress::Role::PeerServer); assert(d->m_eventDispatcher); d->m_transport = transport; d->m_connectAddress = address; EventDispatcherPrivate::get(d->m_eventDispatcher)->m_connectionToNotify = d; #if 0 // TODO make the client authenticate itself, roughly along these lines // this sets up a server that will be destroyed after accepting exactly one connection d->m_clientConnectedHandler = new ClientConnectedHandler; d->m_clientConnectedHandler->m_server = IServer::create(ca); d->m_clientConnectedHandler->m_server->setEventDispatcher(dispatcher); d->m_clientConnectedHandler->m_server->setNewConnectionListener(d->m_clientConnectedHandler); d->m_clientConnectedHandler->m_parent = d; #endif d->receiveNextMessage(); d->m_state = ConnectionPrivate::Connected; } Connection::Connection(Connection &&other) { d = other.d; other.d = nullptr; if (d) { d->m_connection = this; } } Connection &Connection::operator=(Connection &&other) { this->~Connection(); d = other.d; other.d = nullptr; if (d) { d->m_connection = this; } return *this; } Connection::~Connection() { if (!d) { return; } d->close(); delete d->m_transport; delete d->m_authClient; delete d->m_helloReceiver; delete d->m_receivingMessage; delete d; d = nullptr; } void Connection::close() { d->close(); } void ConnectionPrivate::close() { // Can't be main and secondary at the main time - it could be made to work, but what for? assert(m_secondaryThreadLinks.empty() || !m_mainThreadConnection); if (m_mainThreadConnection) { CommutexUnlinker unlinker(&m_mainThreadLink); if (unlinker.hasLock()) { SecondaryConnectionDisconnectEvent *evt = new SecondaryConnectionDisconnectEvent(); evt->connection = this; EventDispatcherPrivate::get(m_mainThreadConnection->m_eventDispatcher) ->queueEvent(std::unique_ptr(evt)); } } // Destroy whatever is suitable and available at a given time, in order to avoid things like // one secondary thread blocking another indefinitely and smaller dependency-related slowdowns. while (!m_secondaryThreadLinks.empty()) { for (auto it = m_secondaryThreadLinks.begin(); it != m_secondaryThreadLinks.end(); ) { CommutexUnlinker unlinker(&it->second, false); if (unlinker.willSucceed()) { if (unlinker.hasLock()) { MainConnectionDisconnectEvent *evt = new MainConnectionDisconnectEvent(); EventDispatcherPrivate::get(it->first->m_eventDispatcher) ->queueEvent(std::unique_ptr(evt)); } unlinker.unlinkNow(); // don't access the element after erasing it, finish it now it = m_secondaryThreadLinks.erase(it); } else { ++it; // don't block, try again next iteration } } } cancelAllPendingReplies(); EventDispatcherPrivate::get(m_eventDispatcher)->m_connectionToNotify = nullptr; } void ConnectionPrivate::startAuthentication() { m_authClient = new AuthClient(m_transport); m_authClient->setCompletionListener(this); } void ConnectionPrivate::handleHelloReply() { if (!m_helloReceiver->m_helloReply.hasNonErrorReply()) { delete m_helloReceiver; m_helloReceiver = nullptr; m_state = Unconnected; // TODO set an error, provide access to it, also set it on messages when trying to send / receive them return; } Arguments argList = m_helloReceiver->m_helloReply.reply()->arguments(); delete m_helloReceiver; m_helloReceiver = nullptr; Arguments::Reader reader(argList); assert(reader.state() == Arguments::String); cstring busName = reader.readString(); assert(reader.state() == Arguments::Finished); m_uniqueName = toStdString(busName); // tell current secondaries UniqueNameReceivedEvent evt; evt.uniqueName = m_uniqueName; for (auto &it : m_secondaryThreadLinks) { CommutexLocker otherLocker(&it.second); if (otherLocker.hasLock()) { EventDispatcherPrivate::get(it.first->m_eventDispatcher) ->queueEvent(std::unique_ptr(new UniqueNameReceivedEvent(evt))); } } m_state = Connected; } void ConnectionPrivate::handleClientConnected() { m_transport = m_clientConnectedHandler->m_server->takeNextClient(); delete m_clientConnectedHandler; m_clientConnectedHandler = nullptr; assert(m_transport); m_transport->setEventDispatcher(m_eventDispatcher); receiveNextMessage(); m_state = Connected; } void Connection::setDefaultReplyTimeout(int msecs) { d->m_defaultTimeout = msecs; } int Connection::defaultReplyTimeout() const { return d->m_defaultTimeout; } uint32 ConnectionPrivate::takeNextSerial() { uint32 ret; do { ret = m_sendSerial.fetch_add(1, std::memory_order_relaxed); } while (unlikely(ret == 0)); return ret; } Error ConnectionPrivate::prepareSend(Message *msg) { if (msg->serial() == 0) { if (!m_mainThreadConnection) { msg->setSerial(takeNextSerial()); } else { // we take a serial from the other Connection and then serialize locally in order to keep the CPU // expense of serialization local, even though it's more complicated than doing everything in the // other thread / Connection. CommutexLocker locker(&m_mainThreadLink); if (locker.hasLock()) { msg->setSerial(m_mainThreadConnection->takeNextSerial()); } else { return Error::LocalDisconnect; } } } MessagePrivate *const mpriv = MessagePrivate::get(msg); // this is unchanged by move()ing the owning Message. if (!mpriv->serialize()) { return mpriv->m_error; } return Error::NoError; } void ConnectionPrivate::sendPreparedMessage(Message msg) { MessagePrivate *const mpriv = MessagePrivate::get(&msg); mpriv->setCompletionListener(this); m_sendQueue.push_back(std::move(msg)); if (m_state == ConnectionPrivate::Connected && m_sendQueue.size() == 1) { // first in queue, don't wait for some other event to trigger sending mpriv->send(m_transport); } } PendingReply Connection::send(Message m, int timeoutMsecs) { if (timeoutMsecs == DefaultTimeout) { timeoutMsecs = d->m_defaultTimeout; } Error error = d->prepareSend(&m); PendingReplyPrivate *pendingPriv = new PendingReplyPrivate(d->m_eventDispatcher, timeoutMsecs); pendingPriv->m_connectionOrReply.connection = d; pendingPriv->m_receiver = nullptr; pendingPriv->m_serial = m.serial(); // even if we're handing off I/O to a main Connection, keep a record because that simplifies // aborting all pending replies when we disconnect from the main Connection, no matter which // side initiated the disconnection. d->m_pendingReplies.emplace(m.serial(), pendingPriv); if (error.isError()) { // Signal the error asynchronously, in order to get the same delayed completion callback as in // the non-error case. This should make the behavior more predictable and client code harder to // accidentally get wrong. To detect errors immediately, PendingReply::error() can be used. pendingPriv->m_error = error; pendingPriv->m_replyTimeout.start(0); } else { if (!d->m_mainThreadConnection) { d->sendPreparedMessage(std::move(m)); } else { CommutexLocker locker(&d->m_mainThreadLink); if (locker.hasLock()) { std::unique_ptr evt(new SendMessageWithPendingReplyEvent); evt->message = std::move(m); evt->connection = d; EventDispatcherPrivate::get(d->m_mainThreadConnection->m_eventDispatcher) ->queueEvent(std::move(evt)); } else { pendingPriv->m_error = Error::LocalDisconnect; } } } return PendingReply(pendingPriv); } Error Connection::sendNoReply(Message m) { // ### (when not called from send()) warn if sending a message without the noreply flag set? // doing that is wasteful, but might be common. needs investigation. Error error = d->prepareSend(&m); if (error.isError()) { return error; } // pass ownership to the send queue now because if the IO system decided to send the message without // going through an event loop iteration, handleCompletion would be called and expects the message to // be in the queue if (!d->m_mainThreadConnection) { d->sendPreparedMessage(std::move(m)); } else { CommutexLocker locker(&d->m_mainThreadLink); if (locker.hasLock()) { std::unique_ptr evt(new SendMessageEvent); evt->message = std::move(m); EventDispatcherPrivate::get(d->m_mainThreadConnection->m_eventDispatcher) ->queueEvent(std::move(evt)); } else { return Error::LocalDisconnect; } } return Error::NoError; } void Connection::waitForConnectionEstablished() { if (d->m_state != ConnectionPrivate::Authenticating) { return; } while (d->m_state == ConnectionPrivate::Authenticating) { d->m_authClient->handleTransportCanRead(); } if (d->m_state != ConnectionPrivate::AwaitingUniqueName) { return; } // Send the hello message assert(!d->m_sendQueue.empty()); // the hello message should be in the queue MessagePrivate *helloPriv = MessagePrivate::get(&d->m_sendQueue.front()); helloPriv->handleTransportCanWrite(); // Receive the hello reply while (d->m_state == ConnectionPrivate::AwaitingUniqueName) { MessagePrivate::get(d->m_receivingMessage)->handleTransportCanRead(); } } ConnectAddress Connection::connectAddress() const { return d->m_connectAddress; } std::string Connection::uniqueName() const { return d->m_uniqueName; } bool Connection::isConnected() const { return d->m_transport && d->m_transport->isOpen(); } EventDispatcher *Connection::eventDispatcher() const { return d->m_eventDispatcher; } IMessageReceiver *Connection::spontaneousMessageReceiver() const { return d->m_client; } void Connection::setSpontaneousMessageReceiver(IMessageReceiver *receiver) { d->m_client = receiver; } void ConnectionPrivate::handleCompletion(void *task) { switch (m_state) { case Authenticating: { assert(task == m_authClient); if (!m_authClient->isAuthenticated()) { m_state = Unconnected; } delete m_authClient; m_authClient = nullptr; if (m_state == Unconnected) { break; } m_state = AwaitingUniqueName; // Announce our presence to the bus and have it send some introductory information of its own Message hello = Message::createCall("/org/freedesktop/DBus", "org.freedesktop.DBus", "Hello"); hello.setExpectsReply(false); hello.setDestination(std::string("org.freedesktop.DBus")); MessagePrivate *const helloPriv = MessagePrivate::get(&hello); m_helloReceiver = new HelloReceiver; m_helloReceiver->m_helloReply = m_connection->send(std::move(hello)); // Small hack: Connection::send() refuses to really start sending if the connection isn't in // Connected state. So force the sending here to actually get to Connected state. helloPriv->send(m_transport); // Also ensure that the hello message is sent before any other messages that may have been // already enqueued by an API client hello = std::move(m_sendQueue.back()); m_sendQueue.pop_back(); m_sendQueue.push_front(std::move(hello)); m_helloReceiver->m_helloReply.setReceiver(m_helloReceiver); m_helloReceiver->m_parent = this; // get ready to receive the first message, the hello reply receiveNextMessage(); break; } case AwaitingUniqueName: // the code paths for these two states only diverge in the PendingReply handler case Connected: { assert(!m_authClient); if (!m_sendQueue.empty() && task == &m_sendQueue.front()) { m_sendQueue.pop_front(); if (!m_sendQueue.empty()) { MessagePrivate::get(&m_sendQueue.front())->send(m_transport); } } else { assert(task == m_receivingMessage); Message *const receivedMessage = m_receivingMessage; receiveNextMessage(); if (!maybeDispatchToPendingReply(receivedMessage)) { if (m_client) { - m_client->handleSpontaneousMessageReceived(Message(std::move(*receivedMessage))); + m_client->handleSpontaneousMessageReceived(Message(std::move(*receivedMessage)), + m_connection); } // dispatch to other threads listening to spontaneous messages, if any for (auto it = m_secondaryThreadLinks.begin(); it != m_secondaryThreadLinks.end(); ) { SpontaneousMessageReceivedEvent *evt = new SpontaneousMessageReceivedEvent(); evt->message = *receivedMessage; CommutexLocker otherLocker(&it->second); if (otherLocker.hasLock()) { EventDispatcherPrivate::get(it->first->m_eventDispatcher) ->queueEvent(std::unique_ptr(evt)); ++it; } else { ConnectionPrivate *connection = it->first; it = m_secondaryThreadLinks.erase(it); discardPendingRepliesForSecondaryThread(connection); delete evt; } } delete receivedMessage; } } break; } default: // ### decide what to do here break; }; } bool ConnectionPrivate::maybeDispatchToPendingReply(Message *receivedMessage) { if (receivedMessage->type() != Message::MethodReturnMessage && receivedMessage->type() != Message::ErrorMessage) { return false; } auto it = m_pendingReplies.find(receivedMessage->replySerial()); if (it == m_pendingReplies.end()) { return false; } if (PendingReplyPrivate *pr = it->second.asPendingReply()) { m_pendingReplies.erase(it); assert(!pr->m_isFinished); pr->handleReceived(receivedMessage); } else { // forward to other thread's Connection ConnectionPrivate *connection = it->second.asConnection(); m_pendingReplies.erase(it); assert(connection); PendingReplySuccessEvent *evt = new PendingReplySuccessEvent; evt->reply = std::move(*receivedMessage); delete receivedMessage; EventDispatcherPrivate::get(connection->m_eventDispatcher)->queueEvent(std::unique_ptr(evt)); } return true; } void ConnectionPrivate::receiveNextMessage() { m_receivingMessage = new Message; MessagePrivate *const mpriv = MessagePrivate::get(m_receivingMessage); mpriv->setCompletionListener(this); mpriv->receive(m_transport); } void ConnectionPrivate::unregisterPendingReply(PendingReplyPrivate *p) { if (m_mainThreadConnection) { CommutexLocker otherLocker(&m_mainThreadLink); if (otherLocker.hasLock()) { PendingReplyCancelEvent *evt = new PendingReplyCancelEvent; evt->serial = p->m_serial; EventDispatcherPrivate::get(m_mainThreadConnection->m_eventDispatcher) ->queueEvent(std::unique_ptr(evt)); } } #ifndef NDEBUG auto it = m_pendingReplies.find(p->m_serial); assert(it != m_pendingReplies.end()); if (!m_mainThreadConnection) { assert(it->second.asPendingReply()); assert(it->second.asPendingReply() == p); } #endif m_pendingReplies.erase(p->m_serial); } void ConnectionPrivate::cancelAllPendingReplies() { // No locking because we should have no connections to other threads anymore at this point. // No const iteration followed by container clear because that has different semantics - many // things can happen in a callback... // In case we have pending replies for secondary threads, and we cancel all pending replies, // that is because we're shutting down, which we told the secondary thread, and it will deal // with bulk cancellation of replies. We just throw away our records about them. for (auto it = m_pendingReplies.begin() ; it != m_pendingReplies.end(); ) { PendingReplyPrivate *pendingPriv = it->second.asPendingReply(); it = m_pendingReplies.erase(it); if (pendingPriv) { // if from this thread pendingPriv->handleError(Error::LocalDisconnect); } } } void ConnectionPrivate::discardPendingRepliesForSecondaryThread(ConnectionPrivate *connection) { for (auto it = m_pendingReplies.begin() ; it != m_pendingReplies.end(); ) { if (it->second.asConnection() == connection) { it = m_pendingReplies.erase(it); // notification and deletion are handled on the event's source thread } else { ++it; } } } void ConnectionPrivate::processEvent(Event *evt) { // std::cerr << "ConnectionPrivate::processEvent() with event type " << evt->type << std::endl; switch (evt->type) { case Event::SendMessage: sendPreparedMessage(std::move(static_cast(evt)->message)); break; case Event::SendMessageWithPendingReply: { SendMessageWithPendingReplyEvent *pre = static_cast(evt); m_pendingReplies.emplace(pre->message.serial(), pre->connection); sendPreparedMessage(std::move(pre->message)); break; } case Event::SpontaneousMessageReceived: if (m_client) { SpontaneousMessageReceivedEvent *smre = static_cast(evt); - m_client->handleSpontaneousMessageReceived(Message(std::move(smre->message))); + m_client->handleSpontaneousMessageReceived(Message(std::move(smre->message)), m_connection); } break; case Event::PendingReplySuccess: maybeDispatchToPendingReply(&static_cast(evt)->reply); break; case Event::PendingReplyFailure: { PendingReplyFailureEvent *prfe = static_cast(evt); const auto it = m_pendingReplies.find(prfe->m_serial); if (it == m_pendingReplies.end()) { // not a disaster, but when it happens in debug mode I want to check it out assert(false); break; } PendingReplyPrivate *pendingPriv = it->second.asPendingReply(); m_pendingReplies.erase(it); pendingPriv->handleError(prfe->m_error); break; } case Event::PendingReplyCancel: // This comes from a secondary thread, which handles PendingReply notification itself. m_pendingReplies.erase(static_cast(evt)->serial); break; case Event::SecondaryConnectionConnect: { SecondaryConnectionConnectEvent *sce = static_cast(evt); const auto it = find_if(m_unredeemedCommRefs.begin(), m_unredeemedCommRefs.end(), [sce](const CommutexPeer &item) { return item.id() == sce->id; } ); assert(it != m_unredeemedCommRefs.end()); const auto emplaced = m_secondaryThreadLinks.emplace(sce->connection, std::move(*it)).first; m_unredeemedCommRefs.erase(it); // "welcome package" - it's done (only) as an event to avoid locking order issues CommutexLocker locker(&emplaced->second); if (locker.hasLock()) { UniqueNameReceivedEvent *evt = new UniqueNameReceivedEvent; evt->uniqueName = m_uniqueName; EventDispatcherPrivate::get(sce->connection->m_eventDispatcher) ->queueEvent(std::unique_ptr(evt)); } break; } case Event::SecondaryConnectionDisconnect: { SecondaryConnectionDisconnectEvent *sde = static_cast(evt); // delete our records to make sure we don't call into it in the future! const auto found = m_secondaryThreadLinks.find(sde->connection); if (found == m_secondaryThreadLinks.end()) { // looks like we've noticed the disappearance of the other thread earlier return; } m_secondaryThreadLinks.erase(found); discardPendingRepliesForSecondaryThread(sde->connection); break; } case Event::MainConnectionDisconnect: // since the main thread *sent* us the event, it already knows to drop all our PendingReplies m_mainThreadConnection = nullptr; cancelAllPendingReplies(); break; case Event::UniqueNameReceived: // We get this when the unique name became available after we were linked up with the main thread m_uniqueName = static_cast(evt)->uniqueName; break; } } Connection::CommRef Connection::createCommRef() { // TODO this is a good time to clean up "dead" CommRefs, where the counterpart was destroyed. CommRef ret; ret.connection = d; std::pair link = CommutexPeer::createLink(); { SpinLocker mainLocker(&d->m_lock); d->m_unredeemedCommRefs.emplace_back(std::move(link.first)); } ret.commutex = std::move(link.second); return ret; } bool Connection::supportsPassingFileDescriptors() const { return d->m_transport && d->m_transport->supportsPassingFileDescriptors(); } diff --git a/connection/imessagereceiver.cpp b/connection/imessagereceiver.cpp index 1869f01..2babcde 100644 --- a/connection/imessagereceiver.cpp +++ b/connection/imessagereceiver.cpp @@ -1,41 +1,41 @@ /* Copyright (C) 2014 Andreas Hartmetz This library is free software; you can redistribute it and/or modify it under the terms of the GNU Library General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more details. You should have received a copy of the GNU Library General Public License along with this library; see the file COPYING.LGPL. If not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. Alternatively, this file is available under the Mozilla Public License Version 1.1. You may obtain a copy of the License at http://www.mozilla.org/MPL/ */ #include "imessagereceiver.h" #include "message.h" IMessageReceiver::~IMessageReceiver() { } -void IMessageReceiver::handleSpontaneousMessageReceived(Message /* message */) +void IMessageReceiver::handleSpontaneousMessageReceived(Message /* message */, Connection * /* connection */) { // *poof* goes the message when this method returns! } -void IMessageReceiver::handlePendingReplyFinished(PendingReply * /* pendingReply */) +void IMessageReceiver::handlePendingReplyFinished(PendingReply * /* pendingReply */, Connection *) { // if we get here that might be bad! but it also might not be under special circumstances, so // don't complain. } diff --git a/connection/imessagereceiver.h b/connection/imessagereceiver.h index 0a84e09..8de7ead 100644 --- a/connection/imessagereceiver.h +++ b/connection/imessagereceiver.h @@ -1,46 +1,47 @@ /* Copyright (C) 2013 Andreas Hartmetz This library is free software; you can redistribute it and/or modify it under the terms of the GNU Library General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more details. You should have received a copy of the GNU Library General Public License along with this library; see the file COPYING.LGPL. If not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. Alternatively, this file is available under the Mozilla Public License Version 1.1. You may obtain a copy of the License at http://www.mozilla.org/MPL/ */ #ifndef IMESSAGERECEIVER_H #define IMESSAGERECEIVER_H #include "export.h" +class Connection; class Message; class PendingReply; class DFERRY_EXPORT IMessageReceiver { public: virtual ~IMessageReceiver(); // This hands over ownership of the Message. The default implementation is empty, so the Message // is destroyed upon going out of scope there. - virtual void handleSpontaneousMessageReceived(Message message); + virtual void handleSpontaneousMessageReceived(Message message, Connection *connection); // This assumes that client code already owns the PendingReply; if the PendingReply was destroyed, the // reply would be considered a spontaneous message. The received message is owned by the PendingReply. // The default implementation does nothing since somebody must still have the PendingReply, so the // Message is still reachable. That's a somewhat strange but valid situation. - virtual void handlePendingReplyFinished(PendingReply *pendingReply); + virtual void handlePendingReplyFinished(PendingReply *pendingReply, Connection *connection); }; #endif // IMESSAGERECEIVER_H diff --git a/connection/pendingreply.cpp b/connection/pendingreply.cpp index b45c5d6..7a53eed 100644 --- a/connection/pendingreply.cpp +++ b/connection/pendingreply.cpp @@ -1,203 +1,205 @@ /* Copyright (C) 2014 Andreas Hartmetz This library is free software; you can redistribute it and/or modify it under the terms of the GNU Library General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more details. You should have received a copy of the GNU Library General Public License along with this library; see the file COPYING.LGPL. If not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. Alternatively, this file is available under the Mozilla Public License Version 1.1. You may obtain a copy of the License at http://www.mozilla.org/MPL/ */ #include "pendingreply.h" #include "pendingreply_p.h" #include "imessagereceiver.h" #include "connection.h" #include "connection_p.h" #include #include PendingReply::PendingReply() : d(nullptr) { } PendingReply::~PendingReply() { if (!d) { return; } if (!d->m_isFinished) { if (d->m_connectionOrReply.connection) { d->m_connectionOrReply.connection->unregisterPendingReply(d); } } else { if (d->m_connectionOrReply.reply) { delete d->m_connectionOrReply.reply; } } delete d; d = nullptr; } PendingReply::PendingReply(PendingReplyPrivate *priv) : d(priv) { d->m_owner = this; } PendingReply::PendingReply(PendingReply &&other) : d(other.d) { other.d = nullptr; if (d) { d->m_owner = this; } } PendingReply &PendingReply::operator=(PendingReply &&other) { if (this == &other) { return *this; } delete d; d = other.d; other.d = nullptr; // note that in this class, !d is a valid state; otherwise this check wouldn't be necessary because // moved-from objects (that also have !d) are not safe for any operation but destruction if (d) { d->m_owner = this; } return *this; } void PendingReplyPrivate::handleReceived(Message *reply) { m_isFinished = true; // Connection has already unregistered us because it knows this reply is done + Connection *const connection = m_connectionOrReply.connection->m_connection; m_connectionOrReply.reply = reply; m_replyTimeout.stop(); if (m_receiver) { - m_receiver->handlePendingReplyFinished(m_owner); + m_receiver->handlePendingReplyFinished(m_owner, connection); } } void PendingReply::dumpState() { std::cerr << "PendingReply::dumpState() " << d << '\n'; if (d) { std::cerr << d->m_owner << " " << d->m_connectionOrReply.reply << " " << d->m_serial << " " << int(d->m_error.code()) << " " /* << d->m_reply->type() */ << '\n'; } } bool PendingReply::isNull() const { return !d; } bool PendingReply::isFinished() const { return !d || d->m_isFinished; } bool PendingReply::hasNonErrorReply() const { return d && d->m_isFinished && !d->m_error.isError(); } Error PendingReply::error() const { if (!d) { return Error::DetachedPendingReply; } return d->m_error; } bool PendingReply::isError() const { return d && d->m_error.isError(); } void PendingReply::setCookie(void *cookie) { d->m_cookie = cookie; } void *PendingReply::cookie() const { return d->m_cookie; } void PendingReply::setReceiver(IMessageReceiver *receiver) { if (d) { d->m_receiver = receiver; } else { // if !d, this is a detached (invalid) instance, and that can't be changed. std::cerr << "PendingReply::setReceiver() on a detached instance does nothing.\n"; } } IMessageReceiver *PendingReply::receiver() const { return d ? d->m_receiver : nullptr; } const Message *PendingReply::reply() const { return d->m_isFinished ? d->m_connectionOrReply.reply : nullptr; } Message PendingReply::takeReply() { Message reply; if (d->m_isFinished) { reply = std::move(*d->m_connectionOrReply.reply); delete d->m_connectionOrReply.reply; d->m_connectionOrReply.reply = nullptr; } return reply; } void PendingReplyPrivate::handleCompletion(void *task) { assert(task == &m_replyTimeout); (void) task; assert(!m_isFinished); // if a reply comes after the timout, it's too late and the reply is probably served as a spontaneous // message by Connection if (m_connectionOrReply.connection) { m_connectionOrReply.connection->unregisterPendingReply(this); } handleError(Error::Timeout); } void PendingReplyPrivate::handleError(Error error) { // When there is an error before or during sending, we already have an error, and the timeout it set to // zero seconds instead of calling the callback right away, in order to provide more consistent behavior // to API clients. In that case, the timeout itself is not the error. if (!m_error.isError()) { m_error = error; } m_isFinished = true; + Connection *const connection = m_connectionOrReply.connection->m_connection; m_connectionOrReply.reply = nullptr; if (m_receiver) { - m_receiver->handlePendingReplyFinished(m_owner); + m_receiver->handlePendingReplyFinished(m_owner, connection); } } diff --git a/tests/connection/tst_pendingreply.cpp b/tests/connection/tst_pendingreply.cpp index d568757..7dcefb6 100644 --- a/tests/connection/tst_pendingreply.cpp +++ b/tests/connection/tst_pendingreply.cpp @@ -1,137 +1,137 @@ /* Copyright (C) 2013 Andreas Hartmetz This library is free software; you can redistribute it and/or modify it under the terms of the GNU Library General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more details. You should have received a copy of the GNU Library General Public License along with this library; see the file COPYING.LGPL. If not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. Alternatively, this file is available under the Mozilla Public License Version 1.1. You may obtain a copy of the License at http://www.mozilla.org/MPL/ */ #include "arguments.h" #include "connectaddress.h" #include "eventdispatcher.h" #include "imessagereceiver.h" #include "message.h" #include "pendingreply.h" #include "connection.h" #include "../testutil.h" #include #include static void addressMessageToBus(Message *msg) { msg->setType(Message::MethodCallMessage); msg->setDestination("org.freedesktop.DBus"); msg->setInterface("org.freedesktop.DBus"); msg->setPath("/org/freedesktop/DBus"); } class ReplyCheck : public IMessageReceiver { public: EventDispatcher *m_eventDispatcher; - void handlePendingReplyFinished(PendingReply *pr) override + void handlePendingReplyFinished(PendingReply *pr, Connection *) override { pr->dumpState(); std::cout << "got it!\n" << pr->reply()->arguments().prettyPrint(); TEST(pr->isFinished()); TEST(!pr->isError()); // This is really a different test, it used to reproduce a memory leak under Valgrind Message reply = pr->takeReply(); m_eventDispatcher->interrupt(); } }; static void testBusAddress(bool waitForConnected) { EventDispatcher eventDispatcher; Connection conn(&eventDispatcher, ConnectAddress::StandardBus::Session); Message msg; addressMessageToBus(&msg); msg.setMethod("RequestName"); Arguments::Writer writer; writer.writeString("Bana.nana"); // requested name writer.writeUint32(4); // TODO proper enum or so: 4 == DBUS_NAME_FLAG_DO_NOT_QUEUE msg.setArguments(writer.finish()); if (waitForConnected) { // finish creating the connection while (conn.uniqueName().empty()) { eventDispatcher.poll(); } } PendingReply busNameReply = conn.send(std::move(msg)); ReplyCheck replyCheck; replyCheck.m_eventDispatcher = &eventDispatcher; busNameReply.setReceiver(&replyCheck); while (eventDispatcher.poll()) { } } class TimeoutCheck : public IMessageReceiver { public: EventDispatcher *m_eventDispatcher; - void handlePendingReplyFinished(PendingReply *reply) override + void handlePendingReplyFinished(PendingReply *reply, Connection *) override { TEST(reply->isFinished()); TEST(!reply->hasNonErrorReply()); TEST(reply->error().code() == Error::Timeout); std::cout << "We HAVE timed out.\n"; m_eventDispatcher->interrupt(); } }; static void testTimeout() { EventDispatcher eventDispatcher; Connection conn(&eventDispatcher, ConnectAddress::StandardBus::Session); // finish creating the connection; we need to know our own name so we can send the message to // ourself so we can make sure that there will be no reply :) while (conn.uniqueName().empty()) { eventDispatcher.poll(); } Message msg = Message::createCall("/some/dummy/path", "org.no_interface", "non_existent_method"); msg.setDestination(conn.uniqueName()); PendingReply neverGonnaGetReply = conn.send(std::move(msg), 200); TimeoutCheck timeoutCheck; timeoutCheck.m_eventDispatcher = &eventDispatcher; neverGonnaGetReply.setReceiver(&timeoutCheck); while (eventDispatcher.poll()) { } } int main(int, char *[]) { testBusAddress(false); testBusAddress(true); testTimeout(); // TODO testBadCall std::cout << "Passed!\n"; } diff --git a/tests/connection/tst_threads.cpp b/tests/connection/tst_threads.cpp index 2f7c669..edb7c6e 100644 --- a/tests/connection/tst_threads.cpp +++ b/tests/connection/tst_threads.cpp @@ -1,242 +1,242 @@ /* Copyright (C) 2014 Andreas Hartmetz This library is free software; you can redistribute it and/or modify it under the terms of the GNU Library General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more details. You should have received a copy of the GNU Library General Public License along with this library; see the file COPYING.LGPL. If not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. Alternatively, this file is available under the Mozilla Public License Version 1.1. You may obtain a copy of the License at http://www.mozilla.org/MPL/ */ #include "arguments.h" #include "connectaddress.h" #include "eventdispatcher.h" #include "imessagereceiver.h" #include "message.h" #include "pendingreply.h" #include "stringtools.h" #include "connection.h" #include "../testutil.h" #include #include #include static const char *echoPath = "/echo"; // make the name "fairly unique" because the interface name is our only protection against replying // to the wrong message static const char *echoInterface = "org.example_fb39a8dbd0aa66d2.echo"; static const char *echoMethod = "echo"; //////////////// Multi-thread ping-pong test //////////////// static const char *pingPayload = "-> J. Random PING"; static const char *pongPayload = "<- J. Random Pong"; class PongSender : public IMessageReceiver { public: Connection *m_connection; - void handleSpontaneousMessageReceived(Message ping) override + void handleSpontaneousMessageReceived(Message ping, Connection *) override { if (ping.interface() != echoInterface) { // This is not the ping... it is probably still something from connection setup. // We can possibly receive many things here that we were not expecting. return; } { Arguments args = ping.arguments(); Arguments::Reader reader(args); cstring payload = reader.readString(); TEST(!reader.error().isError()); TEST(reader.isFinished()); std::cout << "we have ping with payload: " << payload.ptr << std::endl; } { Message pong = Message::createReplyTo(ping); Arguments::Writer writer; writer.writeString(pongPayload); pong.setArguments(writer.finish()); std::cout << "\n\nSending pong!\n\n"; Error replyError = m_connection->sendNoReply(std::move(pong)); TEST(!replyError.isError()); m_connection->eventDispatcher()->interrupt(); } } }; static void pongThreadRun(Connection::CommRef mainConnectionRef, std::atomic *pongThreadReady) { std::cout << " Pong thread starting!\n"; EventDispatcher eventDispatcher; Connection conn(&eventDispatcher, std::move(mainConnectionRef)); PongSender pongSender; pongSender.m_connection = &conn; conn.setSpontaneousMessageReceiver(&pongSender); while (eventDispatcher.poll()) { std::cout << " Pong thread waking up!\n"; if (conn.uniqueName().length()) { pongThreadReady->store(true); // HACK: we do this only to wake up the main thread's event loop std::cout << "\n\nSending WAKEUP package!!\n\n"; Message wakey = Message::createCall(echoPath, "org.notexample.foo", echoMethod); wakey.setDestination(conn.uniqueName()); conn.sendNoReply(std::move(wakey)); } else { std::cout << " Pong thread: NO NAME YET!\n"; } // receive ping message // send pong message } std::cout << " Pong thread almost finished!\n"; } class PongReceiver : public IMessageReceiver { public: - void handlePendingReplyFinished(PendingReply *pongReply) override + void handlePendingReplyFinished(PendingReply *pongReply, Connection *) override { TEST(!pongReply->error().isError()); Message pong = pongReply->takeReply(); Arguments args = pong.arguments(); Arguments::Reader reader(args); std::string strPayload = toStdString(reader.readString()); TEST(!reader.error().isError()); TEST(reader.isFinished()); TEST(strPayload == pongPayload); } }; static void testPingPong() { EventDispatcher eventDispatcher; Connection conn(&eventDispatcher, ConnectAddress::StandardBus::Session); std::atomic pongThreadReady(false); std::thread pongThread(pongThreadRun, conn.createCommRef(), &pongThreadReady); // finish creating the connection while (conn.uniqueName().empty()) { std::cout << "."; eventDispatcher.poll(); } std::cout << "we have connection! " << conn.uniqueName() << "\n"; // send ping message to other thread Message ping = Message::createCall(echoPath, echoInterface, echoMethod); Arguments::Writer writer; writer.writeString(pingPayload); ping.setArguments(writer.finish()); ping.setDestination(conn.uniqueName()); PongReceiver pongReceiver; PendingReply pongReply; bool sentPing = false; while (!sentPing || !pongReply.isFinished()) { eventDispatcher.poll(); if (pongThreadReady.load() && !sentPing) { std::cout << "\n\nSending ping!!\n\n"; pongReply = conn.send(std::move(ping)); pongReply.setReceiver(&pongReceiver); sentPing = true; } } TEST(pongReply.hasNonErrorReply()); std::cout << "we have pong!\n"; pongThread.join(); } //////////////// Multi-threaded timeout test //////////////// class TimeoutReceiver : public IMessageReceiver { public: - void handlePendingReplyFinished(PendingReply *reply) override + void handlePendingReplyFinished(PendingReply *reply, Connection *) override { TEST(reply->isFinished()); TEST(!reply->hasNonErrorReply()); TEST(reply->error().code() == Error::Timeout); std::cout << "We HAVE timed out.\n"; } }; static void timeoutThreadRun(Connection::CommRef mainConnectionRef, std::atomic *done) { // TODO v turn this into proper documentation in Connection // Open a Connection "slaved" to the other Connection - it runs its own event loop in this thread // and has message I/O handled by the Connection in the "master" thread through message passing. // The main purpose of that is to use just one DBus connection per application( module), which is often // more convenient for client programmers and brings some limited ordering guarantees. std::cout << " Other thread starting!\n"; EventDispatcher eventDispatcher; Connection conn(&eventDispatcher, std::move(mainConnectionRef)); while (!conn.uniqueName().length()) { eventDispatcher.poll(); } Message notRepliedTo = Message::createCall(echoPath, echoInterface, echoMethod); notRepliedTo.setDestination(conn.uniqueName()); PendingReply deadReply = conn.send(std::move(notRepliedTo), 50); TimeoutReceiver timeoutReceiver; deadReply.setReceiver(&timeoutReceiver); while (!deadReply.isFinished()) { eventDispatcher.poll(); } *done = true; } static void testThreadedTimeout() { EventDispatcher eventDispatcher; Connection conn(&eventDispatcher, ConnectAddress::StandardBus::Session); std::atomic done(false); std::thread timeoutThread(timeoutThreadRun, conn.createCommRef(), &done); while (!done) { eventDispatcher.poll(); } timeoutThread.join(); } // more things to test: // - (do we want to do this, and if so here??) blocking on a reply through other thread's connection // - ping-pong with several messages queued - every message should arrive exactly once and messages // should arrive in sending order (can use serials for that as simplificitaion) int main(int, char *[]) { testPingPong(); testThreadedTimeout(); std::cout << "Passed!\n"; } diff --git a/tests/serialization/tst_message.cpp b/tests/serialization/tst_message.cpp index 423c51b..981bd3c 100644 --- a/tests/serialization/tst_message.cpp +++ b/tests/serialization/tst_message.cpp @@ -1,301 +1,301 @@ /* Copyright (C) 2013 Andreas Hartmetz This library is free software; you can redistribute it and/or modify it under the terms of the GNU Library General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more details. You should have received a copy of the GNU Library General Public License along with this library; see the file COPYING.LGPL. If not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. Alternatively, this file is available under the Mozilla Public License Version 1.1. You may obtain a copy of the License at http://www.mozilla.org/MPL/ */ #include "arguments.h" #include "connectaddress.h" #include "error.h" #include "eventdispatcher.h" #include "imessagereceiver.h" #include "message.h" #include "pendingreply.h" #include "testutil.h" #include "connection.h" #include #include #include #include static void test_signatureHeader() { Message msg; Arguments::Writer writer; writer.writeByte(123); writer.writeUint64(1); msg.setArguments(writer.finish()); TEST(msg.signature() == "yt"); } class PrintAndTerminateClient : public IMessageReceiver { public: - void handleSpontaneousMessageReceived(Message msg) override + void handleSpontaneousMessageReceived(Message msg, Connection *) override { std::cout << msg.prettyPrint(); m_eventDispatcher->interrupt(); } EventDispatcher *m_eventDispatcher; }; class PrintAndReplyClient : public IMessageReceiver { public: - void handleSpontaneousMessageReceived(Message msg) override + void handleSpontaneousMessageReceived(Message msg, Connection *) override { std::cout << msg.prettyPrint(); m_connection->sendNoReply(Message::createErrorReplyTo(msg, "Unable to get out of hammock!")); //m_connection->eventDispatcher()->interrupt(); } Connection *m_connection; }; // used during implementation, is supposed to not crash and be valgrind-clean afterwards void testBasic(const ConnectAddress &clientAddress) { EventDispatcher dispatcher; ConnectAddress serverAddress = clientAddress; serverAddress.setRole(ConnectAddress::Role::PeerServer); Connection serverConnection(&dispatcher, serverAddress); std::cout << "Created server connection. " << &serverConnection << std::endl; Connection clientConnection(&dispatcher, clientAddress); std::cout << "Created client connection. " << &clientConnection << std::endl; PrintAndReplyClient printAndReplyClient; printAndReplyClient.m_connection = &serverConnection; serverConnection.setSpontaneousMessageReceiver(&printAndReplyClient); PrintAndTerminateClient printAndTerminateClient; printAndTerminateClient.m_eventDispatcher = &dispatcher; clientConnection.setSpontaneousMessageReceiver(&printAndTerminateClient); Message msg = Message::createCall("/foo", "org.foo.interface", "laze"); Arguments::Writer writer; writer.writeString("couch"); msg.setArguments(writer.finish()); clientConnection.sendNoReply(std::move(msg)); while (dispatcher.poll()) { } } void testMessageLength() { static const uint32 bufferSize = Arguments::MaxArrayLength + 1024; byte *buffer = static_cast(malloc(bufferSize)); memset(buffer, 0, bufferSize); for (int i = 0; i < 2; i++) { const bool makeTooLong = i == 1; Arguments::Writer writer; writer.writePrimitiveArray(Arguments::Byte, chunk(buffer, Arguments::MaxArrayLength)); // Our minimal Message is going to have the following variable headers (in that order): // Array: 4 byte length prefix // PathHeader: 4 byte length prefix // MethodHeader: 4 byte length prefix // SignatureHeader: 1 byte length prefix // This is VERY tedious to calculate, so let's just take it as an experimentally determined value uint32 left = Arguments::MaxMessageLength - Arguments::MaxArrayLength - 72; if (makeTooLong) { left += 1; } writer.writePrimitiveArray(Arguments::Byte, chunk(buffer, left)); Message msg = Message::createCall("/a", "x"); msg.setSerial(1); msg.setArguments(writer.finish()); std::vector saved = msg.save(); TEST(msg.error().isError() == makeTooLong); } } enum { // a small integer could be confused with an index into the fd array (in the implementation), // so make it large DummyFdOffset = 1000000 }; static Arguments createArgumentsWithDummyFileDescriptors(uint fdCount) { Arguments::Writer writer; for (uint i = 0; i < fdCount; i++) { writer.writeUnixFd(DummyFdOffset - i); } return writer.finish(); } void testFileDescriptorsInArguments() { // Note: This replaces round-trip tests with file descriptors in tst_arguments. // A full roundtrip test must go through Message due to the out-of-band way that file // descriptors are stored (which is so because they are also transmitted out-of-band). Message msg = Message::createCall("/foo", "org.foo.interface", "doNothing"); for (uint i = 0; i < 4; i++) { msg.setArguments(createArgumentsWithDummyFileDescriptors(i)); { // const ref to arguments const Arguments &args = msg.arguments(); Arguments::Reader reader(args); for (uint j = 0; j < i; j++) { TEST(reader.readUnixFd() == int(DummyFdOffset - j)); TEST(reader.isValid()); } TEST(reader.isFinished()); } { // copy of arguments Arguments args = msg.arguments(); Arguments::Reader reader(args); for (uint j = 0; j < i; j++) { TEST(reader.readUnixFd() == int(DummyFdOffset - j)); TEST(reader.isValid()); } TEST(reader.isFinished()); } } } void testTooManyFileDescriptors() { // TODO re-think what is the best place to catch too many file descriptors... Arguments::Writer writer; } void testFileDescriptorsHeader() { Message msg = Message::createCall("/foo", "org.foo.interface", "doNothing"); for (uint i = 0; i < 4; i++) { msg.setArguments(createArgumentsWithDummyFileDescriptors(i)); TEST(msg.unixFdCount() == i); } } enum { // for pipe2() file descriptor array ReadSide = 0, WriteSide = 1, // how many file descriptors to send in test FdCountToSend = 10 }; class FileDescriptorTestReceiver : public IMessageReceiver { public: - void handleSpontaneousMessageReceived(Message msg) override + void handleSpontaneousMessageReceived(Message msg, Connection *) override { // we're on the session bus, so we'll receive all kinds of notifications we don't care about here if (msg.type() != Message::MethodCallMessage || msg.method() != "testFileDescriptorsForDataTransfer") { return; } Arguments::Reader reader(msg.arguments()); for (uint i = 0; i < FdCountToSend; i++) { int fd = reader.readUnixFd(); uint readBuf = 12345; ::read(fd, &readBuf, sizeof(uint)); ::close(fd); TEST(readBuf == i); } Message reply = Message::createReplyTo(msg); m_connection->sendNoReply(std::move(reply)); } Connection *m_connection = nullptr; }; void testFileDescriptorsForDataTransfer() { EventDispatcher eventDispatcher; Connection conn(&eventDispatcher, ConnectAddress::StandardBus::Session); conn.waitForConnectionEstablished(); TEST(conn.isConnected()); int pipeFds[2 * FdCountToSend]; Message msg = Message::createCall("/foo", "org.foo.interface", "testFileDescriptorsForDataTransfer"); msg.setDestination(conn.uniqueName()); Arguments::Writer writer; for (uint i = 0; i < FdCountToSend; i++) { TEST(pipe2(pipeFds + 2 * i, O_NONBLOCK) == 0); // write into write side of the pipe... will be read when the message is received back from bus ::write(pipeFds[2 * i + WriteSide], &i, sizeof(uint)); writer.writeUnixFd(pipeFds[2 * i + ReadSide]); } msg.setArguments(writer.finish()); PendingReply reply = conn.send(std::move(msg), 500 /* fail quickly */); FileDescriptorTestReceiver fdTestReceiver; conn.setSpontaneousMessageReceiver(&fdTestReceiver); fdTestReceiver.m_connection = &conn; while (!reply.isFinished()) { eventDispatcher.poll(); } TEST(reply.hasNonErrorReply()); // otherwise timeout, the message exchange failed somehow for (uint i = 0; i < FdCountToSend; i++) { ::close(pipeFds[2 * i + WriteSide]); } } int main(int, char *[]) { test_signatureHeader(); #ifdef __linux__ { ConnectAddress clientAddress; clientAddress.setType(ConnectAddress::Type::AbstractUnixPath); clientAddress.setRole(ConnectAddress::Role::PeerClient); clientAddress.setPath("dferry.Test.Message"); testBasic(clientAddress); } #endif // TODO: SocketType::Unix works on any Unix-compatible OS, but we'll need to construct a path { ConnectAddress clientAddress; clientAddress.setType(ConnectAddress::Type::Tcp); clientAddress.setPort(6800); clientAddress.setRole(ConnectAddress::Role::PeerClient); testBasic(clientAddress); } testMessageLength(); testFileDescriptorsInArguments(); testTooManyFileDescriptors(); testFileDescriptorsHeader(); testFileDescriptorsForDataTransfer(); // TODO testSaveLoad(); // TODO testDeepCopy(); std::cout << "\nNote that the hammock error is part of the test.\nPassed!\n"; }