mirror of
https://github.com/cculianu/Fulcrum.git
synced 2026-08-18 13:09:12 +02:00
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).
This commit is contained in:
Calin Culianu
parent
0b6d6f0b70
commit
f4091bb005
5 changed files with 144 additions and 15 deletions
110
BitcoinD.cpp
110
BitcoinD.cpp
|
|
@ -1,3 +1,5 @@
|
|||
#include <QPointer>
|
||||
|
||||
#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<BitcoinD>(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<bool>(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<QObject> 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<std::atomic_bool>(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");
|
||||
}
|
||||
|
|
|
|||
24
BitcoinD.h
24
BitcoinD.h
|
|
@ -25,6 +25,23 @@ public:
|
|||
|
||||
static constexpr int N_CLIENTS = 2;
|
||||
|
||||
using ResultsF = std::function<void(const RPC::Message &response)>;
|
||||
using ErrorF = ResultsF; // identical to ResultsF above except the message passed in is an error="" message.
|
||||
using FailF = std::function<void(const RPC::Message::Id &origId, const QString & failureReason)>;
|
||||
|
||||
/// 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<quint64> goodBitcoinDs;
|
||||
static constexpr int miniTimeout = 333, tinyTimeout = 167;
|
||||
|
||||
std::set<quint64> 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<BitcoinD> 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 */
|
||||
|
|
|
|||
|
|
@ -71,4 +71,11 @@ void Controller::process()
|
|||
Debug() << "Process called...";
|
||||
if (!sm) sm = std::make_unique<StateMachine>(); // 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;
|
||||
});
|
||||
}
|
||||
|
|
|
|||
6
RPC.cpp
6
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;
|
||||
}
|
||||
|
||||
|
|
|
|||
12
RPC.h
12
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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue