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