From f4091bb005b83b082cf15bcfe9bbdf07ffc6e39a Mon Sep 17 00:00:00 2001 From: Calin Culianu Date: Mon, 18 Nov 2019 14:32:48 +0200 Subject: [PATCH] Devised a scheme for asynch messaging to bitcoind And added some test code in Controller to use the scheme. It's very asynchronous. Needs some tuning long term to see how it deals with load and perhaps throttle the requests if we know bitcoind is under load. But for right now, for testing, it should work ok. Also in this commit small nits, plus bugfix to RPC::Message factory methods (I forgot to populate some instance properties when constructing the objects). Additionally, we were registering the qMetaType wrong for RPC::Message and RPC::Message::Id (this has been fixed). --- BitcoinD.cpp | 110 +++++++++++++++++++++++++++++++++++++++++++++---- BitcoinD.h | 24 ++++++++++- Controller.cpp | 7 ++++ RPC.cpp | 6 +++ RPC.h | 12 +++--- 5 files changed, 144 insertions(+), 15 deletions(-) diff --git a/BitcoinD.cpp b/BitcoinD.cpp index 8f098b5..8764a92 100644 --- a/BitcoinD.cpp +++ b/BitcoinD.cpp @@ -1,3 +1,5 @@ +#include + #include "BitcoinD.h" BitcoinDMgr::BitcoinDMgr(const QHostAddress &host, quint16 port, @@ -14,8 +16,6 @@ void BitcoinDMgr::startup() { Log() << objectName() << ": starting " << N_CLIENTS << " bitcoin rpc clients ..."; for (auto & client : clients) { - constexpr int miniTimeout = 333; - client = std::make_unique(host, port, user, pass); // connect client to us -- TODO: figure out workflow: how requests for work and results will get dispatched @@ -27,8 +27,8 @@ void BitcoinDMgr::startup() { Debug() << "got authenticated for id:" << b->id << " but isGood() is false!"; return; // false/stale signal } - const bool wasEmpty = goodBitcoinDs.empty(); - goodBitcoinDs.insert(b->id); + const bool wasEmpty = goodSet.empty(); + goodSet.insert(b->id); if (wasEmpty) emit gotFirstGoodConnection(b->id); }); @@ -38,11 +38,11 @@ void BitcoinDMgr::startup() { Debug() << "got lostConnection for id:" << c->id << " but isGood() is true!"; return; // false/stale signal } - goodBitcoinDs.erase(c->id); + goodSet.erase(c->id); auto constexpr chkTimer = "checkNoMoreBitcoinDs"; // we throttle the spamming of the allConnectionsLost signal via this mechanism callOnTimerSoonNoRepeat(miniTimeout, chkTimer, [this]{ - if (goodBitcoinDs.empty()) + if (goodSet.empty()) emit allConnectionsLost(); }, true); }); @@ -61,7 +61,7 @@ void BitcoinDMgr::cleanup() { for (auto & client : clients) { client.reset(); /// implicitly calls client->stop() } - goodBitcoinDs.clear(); + goodSet.clear(); Debug() << "BitcoinDMgr cleaned up"; } @@ -91,6 +91,100 @@ auto BitcoinDMgr::stats() const -> Stats return ret; } + +BitcoinD *BitcoinDMgr::getBitcoinD() +{ + BitcoinD *ret = nullptr; + if (goodSet.size() <= 1) + lastBitcoinDUsed = NO_ID; + if (!goodSet.empty()) { + // linear search for a bitcoind that is not lastBitcoinDUsed + for (auto & client : clients) { + if (goodSet.count(client->id) && lastBitcoinDUsed != client->id + // fixme here: will tinyTimeout expire easily under load? tune this. + && Util::CallOnObjectWithTimeoutNoThrow(tinyTimeout, client.get(), &BitcoinD::isGood).value_or(false)) { + ret = client.get(); + break; + } + } + } + if (ret) lastBitcoinDUsed = ret->id; + return ret; +} + +/// this is safe to call from any thread. internally it dispatches messages to this obejct's thread. +/// may throw (TODO list exceptions it may throw). Results/Error/Fail functions are called in the context of the sender's thread. +/// Returns the BitcoinD->id that was given the message. +void BitcoinDMgr::submitRequest(QObject *sender, const RPC::Message::Id &rid, const QString & method, const QVariantList & params, + const ResultsF & resf, const ErrorF & errf, const FailF & failf) +{ + QPointer context(new QObject(sender)); // this is a weak ref that gets killed when sender is killed. This way stuff just "goes away" if sender dies. + context->setObjectName(QString("context for '%1' request id: %2").arg(sender ? sender->objectName() : "").arg(rid.toString())); + auto killContext = [context, this] { + if (LIKELY(context)) { // need to check context because race conditions + Util::VoidFuncOnObjectNoThrow(this, [context, this] { if (LIKELY(context)) disconnect(this, &QObject::destroyed, context, nullptr); }, 0); + Util::VoidFuncOnObjectNoThrow(context, [context] { if (LIKELY(context)) context->deleteLater(); }, 0); + } + }; + connect(this, &QObject::destroyed, context, killContext); // make sure that if we die, to kill the context too to clean up resources. + + // schedule this ASAP + QTimer::singleShot(0, this, [this, context, resf, errf, failf, rid, method, params, killContext] { + auto replied = std::make_shared(false); // guards against spurious lostConnection arriving after reply msg + auto do_fail = [failf, context, killContext, rid, replied](const QString & reason){ + if (LIKELY(failf && context && !replied->exchange(true))) { + Util::VoidFuncOnObjectNoThrow(context, [context, failf, rid, reason]{ + if (LIKELY(context)) // need to check context again because race conditions + failf(rid, reason); + }); // fixme: this has an infinite timeout + } + killContext(); + }; + auto do_res = [resf, context, killContext, replied](const RPC::Message &resp){ + if (LIKELY(resf && context && !replied->exchange(true))) { + Util::VoidFuncOnObjectNoThrow(context, [context, resf, resp]{ + if (LIKELY(context)) // need to check context again because race conditions + resf(resp); + }); // fixme: this has an infinite timeout + } + killContext(); + }; + auto do_err = [errf, context, killContext, replied](const RPC::Message &resp){ + if (LIKELY(errf && context && !replied->exchange(true))) { + Util::VoidFuncOnObjectNoThrow(context.data(), [context, errf, resp]{ + if (LIKELY(context)) // need to check context again because race conditions + errf(resp); + }); // fixmne: this has an infinite timeout + } + killContext(); + }; + if (UNLIKELY(!context)) + // sender parent must have been deleted before we got a chance to run + return; + auto bd = getBitcoinD(); + if (UNLIKELY(!bd)) { + do_fail("Unable to find a good BitcoinD connection"); + return; + } + connect(bd, &BitcoinD::gotMessage, context, [do_res, rid](quint64, const RPC::Message &reply){ + if (reply.id == rid) // filter out messages not for us + do_res(reply); + }); + connect(bd, &BitcoinD::gotErrorMessage, context, [do_err, rid](quint64, const RPC::Message &errMsg){ + if (errMsg.id == rid) // filter out error messages not for us + do_err(errMsg); + }); + connect(bd, &BitcoinD::lostConnection, context, [do_fail, rid](AbstractConnection *){ + do_fail("connection lost"); + }); + + bd->sendRequest(rid, method, params); + }); + + // .. aand.. return right away +} + +/* --- BitcoinD --- */ auto BitcoinD::stats() const -> Stats { Stats m = RPC::HttpConnection::stats(); @@ -199,5 +293,5 @@ void BitcoinD::do_ping() Debug() << "Stale connection, reconnecting."; reconnect(); } else - emit sendRequest(newId(), "getblockcount"); + emit sendRequest(newId(), "ping"); } diff --git a/BitcoinD.h b/BitcoinD.h index aef1969..a76e221 100644 --- a/BitcoinD.h +++ b/BitcoinD.h @@ -25,6 +25,23 @@ public: static constexpr int N_CLIENTS = 2; + using ResultsF = std::function; + using ErrorF = ResultsF; // identical to ResultsF above except the message passed in is an error="" message. + using FailF = std::function; + + /// This is safe to call from any thread. + /// Internally it dispatches messages to `this` obejct's thread. Results/Error/Fail functions are called in the + /// context of the `sender` object's thread. Returns immediately regardless of which thread context it's called in, + /// with one of the 3 following being called later, in the thread context of `sender`, when results/errors/failure + /// is determined: + /// - ResultsF will be called exactly once on success returning the reults encapsulated in an RPC::Message + /// - If BitcoinD generated an error response, ErrorF will be called exactly once with the error wrapped in an + /// RPC::Message + /// - If some other error occurred (such as timeout, or BitcoinD not connected, or connection lost, etc), FailF + /// will be called exactly once with a string message. + void submitRequest(QObject *sender, const RPC::Message::Id &id, const QString & method, const QVariantList & params, + const ResultsF & = ResultsF(), const ErrorF & = ErrorF(), const FailF & = FailF()); + signals: void gotFirstGoodConnection(quint64 bitcoindId); // emitted whenever the first bitcoind after a "down" state (or after startup) gets its first good status (after successful authentication) void allConnectionsLost(); // emitted whenever all bitcoind rpc connections are down. @@ -43,9 +60,14 @@ private: const quint16 port; const QString user, pass; - std::set goodBitcoinDs; + static constexpr int miniTimeout = 333, tinyTimeout = 167; + + std::set goodSet; ///< set of bitcoind's (by id) that are `isGood` (connected, authed). This set is updated as we get signaled from BitcoinD objects. May be empty. Has at most N_CLIENTS elements. std::unique_ptr clients[N_CLIENTS]; + + quint64 lastBitcoinDUsed = NO_ID; + BitcoinD *getBitcoinD(); ///< may return nullptr if none are up. Otherwise does a round-robin of the ones present to grab one. to be called only in this thread. }; class BitcoinD : public RPC::HttpConnection, public ThreadObjectMixin /* NB: also inherits TimersByNameMixin via AbstractConnection base */ diff --git a/Controller.cpp b/Controller.cpp index 766f470..5368bc9 100644 --- a/Controller.cpp +++ b/Controller.cpp @@ -71,4 +71,11 @@ void Controller::process() Debug() << "Process called..."; if (!sm) sm = std::make_unique(); // create statemachine if doest not exist + bitcoindmgr->submitRequest(this, IdMixin::newId(), "getmempoolinfo", {}, [](auto resp){ // testing + Trace() << resp.id.toString() << ": result reply: " << resp.toJsonString(); + }, [](auto resp){ + Trace() << resp.id.toString() << ": error response: " << resp.toJsonString(); + }, [](auto id, auto msg) { + Warning() << id.toString() << ": FAIL: " << msg; + }); } diff --git a/RPC.cpp b/RPC.cpp index be51270..6956ce1 100644 --- a/RPC.cpp +++ b/RPC.cpp @@ -169,6 +169,7 @@ namespace RPC { if (!v1) map["jsonrpc"] = RPC::jsonRpcVersion; ret.v1 = v1; + ret.id = id; map["id"] = id; // may be "null" QVariantMap errMap; errMap["code"] = code; @@ -190,6 +191,7 @@ namespace RPC { map["error"] = QVariant(); // v1: always set the "error" key to null map["id"] = reqId; map["result"] = result; + ret.id = reqId; return ret; } @@ -199,6 +201,7 @@ namespace RPC { Message ret = makeNotification(methodName, params, v1); auto & map = ret.data; map["id"] = id; + ret.id = id; return ret; } @@ -208,6 +211,7 @@ namespace RPC { Message ret = makeNotification(methodName, params, v1); auto & map = ret.data; map["id"] = id; + ret.id = id; return ret; } @@ -223,6 +227,7 @@ namespace RPC { map["id"] = QVariant(); // v1: always has the "id" key as null for a notif map["method"] = methodName; map["params"] = params; + ret.method = methodName; return ret; } @@ -238,6 +243,7 @@ namespace RPC { map["id"] = QVariant(); // v1: always has the "id" key as null for a notif map["method"] = methodName; map["params"] = params; + ret.method = methodName; return ret; } diff --git a/RPC.h b/RPC.h index dc64c5d..8742127 100644 --- a/RPC.h +++ b/RPC.h @@ -248,13 +248,13 @@ namespace RPC { signals: /// call (emit) this to send a request to the peer - void sendRequest(const Message::Id & reqid, const QString &method, const QVariantList & params = QVariantList()); + void sendRequest(const RPC::Message::Id & reqid, const QString &method, const QVariantList & params = QVariantList()); /// call (emit) this to send a notification to the peer void sendNotification(const QString &method, const QVariantList & params = QVariantList()); /// call (emit) this to send a request to the peer - void sendError(bool disconnectAfterSend, int errorCode, const QString &message, const Message::Id & reqid = Message::Id()); + void sendError(bool disconnectAfterSend, int errorCode, const QString &message, const RPC::Message::Id & reqid = Message::Id()); /// call (emit) this to send a result reply to the peer (result= message) - void sendResult(const Message::Id & reqid, const QString &method, const QVariant & result = QVariant()); + void sendResult(const RPC::Message::Id & reqid, const QString &method, const QVariant & result = QVariant()); /// this is emitted when a new message arrives that was successfully parsed and matches /// a known method described in the 'methods' MethodMap. Unknown messages will eventually result @@ -269,13 +269,13 @@ namespace RPC { protected slots: /// Actual implentation that prepares the request. Is connected to sendRequest() above. Runs in this object's /// thread context. Eventually calls send() -> do_write() (from superclass). - virtual void _sendRequest(const Message::Id & reqid, const QString &method, const QVariantList & params = QVariantList()); + virtual void _sendRequest(const RPC::Message::Id & reqid, const QString &method, const QVariantList & params = QVariantList()); // ditto for notifications virtual void _sendNotification(const QString &method, const QVariantList & params = QVariantList()); /// Actual implementation of sendError, runs in our thread context. - virtual void _sendError(bool disconnect, int errorCode, const QString &message, const Message::Id &reqid = Message::Id()); + virtual void _sendError(bool disconnect, int errorCode, const QString &message, const RPC::Message::Id &reqid = Message::Id()); /// Actual implementation of sendResult, runs in our thread context. - virtual void _sendResult(const Message::Id & reqid, const QString &method, const QVariant & result = QVariant()); + virtual void _sendResult(const RPC::Message::Id & reqid, const QString &method, const QVariant & result = QVariant()); protected: /// chains to base, connects sendRequest signal to _sendRequest slot