Index: CMakeLists.txt =================================================================== diff -u -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 -re9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65 --- CMakeLists.txt (.../CMakeLists.txt) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) +++ CMakeLists.txt (.../CMakeLists.txt) (revision e9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65) @@ -14,6 +14,7 @@ set(CMAKE_DISABLE_IN_SOURCE_BUILD ON) add_compile_options(-Wall -Werror -Wextra) +add_compile_options($<$:-U_FORTIFY_SOURCE>) add_subdirectory(lib) add_subdirectory(tools) Index: CloudConnect/CloudConnect.pro =================================================================== diff -u -raaebfee335c74b0250864a6dce0555f866adadea -re9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65 --- CloudConnect/CloudConnect.pro (.../CloudConnect.pro) (revision aaebfee335c74b0250864a6dce0555f866adadea) +++ CloudConnect/CloudConnect.pro (.../CloudConnect.pro) (revision e9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65) @@ -6,6 +6,8 @@ QMAKE_CXXFLAGS += -Wall -Werror -Wextra +CONFIG(debug, debug|release): QMAKE_CXXFLAGS += -U_FORTIFY_SOURCE + DESTDIR = $$PWD/../bin include($$PWD/../lib/MsgUtils/codegen_tools.pri) Index: CloudConnect/CloudConnectController.cpp =================================================================== diff -u -raaebfee335c74b0250864a6dce0555f866adadea -re9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65 --- CloudConnect/CloudConnectController.cpp (.../CloudConnectController.cpp) (revision aaebfee335c74b0250864a6dce0555f866adadea) +++ CloudConnect/CloudConnectController.cpp (.../CloudConnectController.cpp) (revision e9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65) @@ -16,12 +16,18 @@ #include #include +#include #include #include #include #include #include +#include +#include +#include +#include + #include "CloudConnectController.h" #include "LeahiMsgDefs.pb.h" #include "LeahiMsgProtoUtils.h" @@ -30,18 +36,16 @@ Q_LOGGING_CATEGORY(logProto, "proto") Q_LOGGING_CATEGORY(logCanRouting, "can.routing") Q_LOGGING_CATEGORY(logMqtt, "mqtt") -Q_LOGGING_CATEGORY(logStats, "mqtt.stats") +Q_LOGGING_CATEGORY(logMqttStats, "mqtt.stats") +Q_LOGGING_CATEGORY(logCanStats, "can.stats") /*! * \brief CloudConnectController::CloudConnectController * \details Constructor. Wires the frame→dispatcher→completed-message pipeline. * The CAN device itself is started by startCan() once this object is * on its worker thread. - * \param configPath - path to the settings file - * \param canRoutingPath - path to the CAN message routing INI - * \param parent - optional QObject parent + * \param[in] parent - optional QObject parent * \note Must be constructed on the main thread, but later moved to a worker thread. - * _canInterface is parented to this object so it migrates with it. */ CloudConnectController::CloudConnectController(QObject *parent) : QObject(parent), @@ -61,6 +65,12 @@ */ CloudConnectController::~CloudConnectController() = default; +/*! + * \brief CloudConnectController::loadConfig + * \details Loads the CloudConnect configuration INI file and validates the content. + * \param[in] configPath - file path of CloudConnect configuration INI + * \return True if config is loaded successfully, otherwise false. + */ bool CloudConnectController::loadConfig(const QString &configPath) { static auto resolvePath = [](const QString &path) -> QString { @@ -110,7 +120,7 @@ /*! * \brief CloudConnectController::initThread * \details Moves this object and its children onto a worker thread and starts it. - * \param thread Thread to move to, owned by the caller. + * \param[in] thread - thread to move to, owned by the caller. * \note Must be called from the main thread, before the event loop runs. */ void CloudConnectController::initThread(QThread &thread) @@ -127,11 +137,7 @@ * \brief CloudConnectController::startCan * \details Creates and connects the CAN device. * \return true if the device was created and connected. - * \note Must run on the controller thread, after initThread(). connectDevice() - * installs the CAN socket notifier on the calling thread, so starting - * here keeps the notifier, the frame queue and the drain loop on one - * thread. QCanBusDevice::framesAvailable() is unguarded in Qt 5.15, so - * a cross-thread split of those is a data race, not just a style issue. + * \note Must run on the controller thread, after initThread(). */ bool CloudConnectController::startCan() { @@ -145,14 +151,15 @@ * \brief CloudConnectController::listenForApp * \details Creates the app socket server, then starts listening. * \return true if the server bound successfully, false otherwise. - * \note Must run on the controller thread, after initThread(). QLocalServer ties - * its socket engine to the thread that calls listen(), and accepted client - * sockets are parented to that engine. + * \note Must run on the controller thread, after initThread(). + * QLocalServer ties its socket engine to the thread that calls + * listen(), and accepted client sockets are parented to that engine. */ bool CloudConnectController::listenForApp() { + // use Q_ASSERT only during object creation or thread move Q_ASSERT_X(QThread::currentThread() == thread(), __func__, - "listenForApp() must run on the controller thread"); // use Q_ASSERT only during object creation or thread move + "listenForApp() must run on the controller thread"); if (_appServer == nullptr) { _appServer = QSharedPointer::create(this); @@ -163,8 +170,8 @@ /*! * \brief CloudConnectController::connectToCloud - * \details Configures the MQTT interface from the [Mqtt] settings and starts connecting. - * \return true if the attempt was started, or if MQTT is disabled. + * \details Starts connecting to the cloud server with the configuration loaded in loadConfig. + * \return true if the attempt was started, or false if MQTT is disabled. * \note Must run on the controller thread, after initThread(). The interface's * QMqttClient socket belongs to the thread that constructs it. */ @@ -179,7 +186,7 @@ /*! * \brief CloudConnectController::loadCanRouting * \details Parses CAN message routing INI and populates _canRouting. - * \param canRoutingPath - path to the CAN message routing INI file + * \param[in] canRoutingPath - path to the CAN message routing INI file * \return true if CanRouting config successfully loads, otherwise false */ bool CloudConnectController::loadCanRouting(const QString &canRoutingPath) @@ -214,7 +221,7 @@ const Can::MsgId msgId = static_cast(group.toUInt(&ok, 16)); if (!ok) { qCWarning(logCanRouting).noquote() << QString("could not convert group \"%1\" to MsgId in CAN mesg routing config file %2") - .arg(group, canRoutingPath); + .arg(group, canRoutingPath); continue; } @@ -227,17 +234,18 @@ canRoutingConfig.endGroup(); CanRouting canRouting; + // TODO: uncomment the following line after testing // canRouting.action = actionMap.value(actionStr, CanAction::Drop); - canRouting.action = actionMap.value(actionStr, CanAction::SendAlways); + canRouting.action = actionMap.value(actionStr, CanAction::SendAlways); // TODO: remove after testing canRouting.topic = topicMap.value(topicStr, CloudConnectFrame::Topic::NormalPriority); if (actionStr.length() > 0 && !actionMap.contains(actionStr)) { - qCWarning(logCanRouting).noquote() << QString("unknown message action \"%1\" for msgId=0x%2 — defaulting to Drop") + qCWarning(logCanRouting).noquote() << QString("unknown message action \"%1\" for msgId=0x%2, defaulting to Drop") .arg(actionStr, QString("%1").arg(quint16(msgId), 4, 16, QChar('0').toUpper())); } if (topicStr.length() > 0 && !topicMap.contains(topicStr)) { - qCWarning(logCanRouting).noquote() << QString("unknown message topic \"%1\" for msgId=0x%2 — defaulting to NormalPriority") + qCWarning(logCanRouting).noquote() << QString("unknown message topic \"%1\" for msgId=0x%2, defaulting to NormalPriority") .arg(topicStr, QString("%1").arg(quint16(msgId), 4, 16, QChar('0').toUpper())); } @@ -249,6 +257,28 @@ } /*! + * \brief CloudConnectController::mqttTopic + * \details Builds the MQTT topic for a message class. + * \param topic - message class from the CAN message routing INI + * \return Topic of the form {prefix}/{deviceId}/{suffix}. + */ +QString CloudConnectController::mqttTopic(CloudConnectFrame::Topic topic) const +{ + static const QMap suffixes = { + { CloudConnectFrame::Topic::HighPriority, QStringLiteral("high") }, + { CloudConnectFrame::Topic::NormalPriority, QStringLiteral("normal") }, + { CloudConnectFrame::Topic::DeviceLogFile, QStringLiteral("log") }, + { CloudConnectFrame::Topic::TreatmentLogFile, QStringLiteral("tx_log") }, + { CloudConnectFrame::Topic::CloudSyncLogFile, QStringLiteral("cs_log") }, + }; + + // TODO: uncomment the following line after testing + // return suffixes.value(topic, QStringLiteral("normal")); + Q_UNUSED(topic) + return QString("%1/%2/%3").arg(_topicPrefix, _deviceId, "clinical"); // TODO: remove after test +} + +/*! * \brief CloudConnectController::onCanFrameReceive * \details Unpacks a CAN frame and sends it to the dispatcher to reassemble the entire CAN message. * \param frame - the received CAN frame @@ -265,6 +295,21 @@ */ void CloudConnectController::onCanMessageReceive(const Can::Message &msg) { + // BEGIN: CAN Mesg Stats + if (!_canTimer.isValid()) { + _canTimer.start(); + } + const auto now = _canTimer.nsecsElapsed() / 1000; + const auto last = _canLastRecv; + _canLastRecv = now; + _canCount++; + if (_canCount > 1) { + _canTotalTime += (now - last); + _canAvgTime = _canTotalTime / qint64(_canCount - 1); + } + qCInfo(logCanStats).noquote() << canStats(msg.msgId, (_canCount > 1) ? (now - last) : 0); + // END: CAN Mesg Stats + const auto it = _canRouting.constFind(msg.msgId); if (it == _canRouting.constEnd()) { qCInfo(logCanRouting).noquote() << QString("no action defined for %1 (0x%2), dropping") @@ -288,60 +333,44 @@ QByteArray payload; if (leahi::canMessageToProtobufByteArray(QDateTime::currentDateTime(), msg, payload)) { - // Read the token out before it is advanced: the value handed to publish() is - // what the delivery ledger reports, so logging the post-increment value made - // every trace-back off by one. - qint32 msgId; - if (_mqttClient.publish(mqttTopic(it->topic), payload, msgId)) { - // Logs the token so an unacknowledged publish reported by the MQTT delivery - // ledger can be traced back to the message it carried. - qCWarning(logMqtt).noquote() << QString("%1 (0x%2) published (msgId=%3)") - .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0')).arg(msgId); + qint32 pubId; + if (_mqttClient.publish(mqttTopic(it->topic), payload, pubId)) { + qCWarning(logMqtt).noquote() << QString("message[ msgId=0x%1 (%2), seq=%3 ], published (pubId=%4)") + .arg(QString("%1").arg(msg.msgId, 4, 16, QChar('0')).toUpper()) + .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))) + .arg(msg.sequence) + .arg(pubId); + + // BEGIN: CAN Mesg Stats + if (_inflight.contains(pubId)) { + // The client reuses packet ids once they retire, so a collision means the + // previous publish never reached a terminal state. Report the loss rather + // than let the old entry disappear silently. + const Inflight &stale = _inflight.value(pubId); + qCWarning(logMqttStats).noquote() << QString("publish id %1 still in flight for 0x%2 seq=%3, replacing") + .arg(pubId).arg(stale.msgId, 4, 16, QChar('0')).arg(stale.sequence); + } + _msgSentCount++; + _inflight.insert(pubId, Inflight { Inflight::Sent, msg.msgId, msg.sequence, msgElapsedUs() }); + // END: CAN Mesg Stats } else { // TODO: spool the message if publish fails qCWarning(logMqtt).noquote() << QString("could not publish %1 (0x%2), message lost") .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0')); } - - // NOTE: this records that the client ACCEPTED the message, not that AWS - // acknowledged it. Only the PUBACK counted in the MqttClient ledger - // means delivered. - // received = published; } else { qCWarning(logProto).noquote() << QString("could not serialize %1 (0x%2) message to protobuf, message dropped") .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0')); } - // Cache the message only after it has been successfully sent. - // Note: saving before successfully send may prevent a message of with that msgId from being - // sent until a delta message is received if SendDelta is specified for the msgId. + // Cache the message only after it has been successfully sent + // TODO: this may need to be moved so the message is cached on ACK cachedMsg = msg; } /*! - * \brief CloudConnectController::mqttTopic - * \details Builds the MQTT topic for a message class. - * \param topic - message class from the CAN message routing INI - * \return Topic of the form {prefix}/{deviceId}/{suffix}. - */ -QString CloudConnectController::mqttTopic(CloudConnectFrame::Topic topic) const -{ - static const QMap suffixes = { - { CloudConnectFrame::Topic::HighPriority, QStringLiteral("high") }, - { CloudConnectFrame::Topic::NormalPriority, QStringLiteral("normal") }, - { CloudConnectFrame::Topic::DeviceLogFile, QStringLiteral("log") }, - { CloudConnectFrame::Topic::TreatmentLogFile, QStringLiteral("tx_log") }, - { CloudConnectFrame::Topic::CloudSyncLogFile, QStringLiteral("cs_log") }, - }; - - // return suffixes.value(topic, QStringLiteral("normal")); - Q_UNUSED(topic) - return QString("%1/%2/%3").arg(_topicPrefix, _deviceId, "clinical"); -} - -/*! * \brief CloudConnectController::onCloudStateChanged * \details Re-establishes the echo subscriptions on every new session, and * retires anything still awaiting an echo when one ends. @@ -355,11 +384,28 @@ switch (state) { case QMqttClient::Disconnected: // clear any pending messages - // expireInFlight(true); + _inflight.clear(); // TODO: Mesg Stats break; case QMqttClient::Connecting: break; case QMqttClient::Connected: + // BEGIN: Mesg Stats + _canCount = 0; + _canAvgTime = 0; + _canTotalTime = 0; + _canLastRecv = 0; + _canTimer.invalidate(); + _inflight.clear(); + _msgTimer.invalidate(); + _msgSentCount = 0; + _msgAckCount = 0; + _msgRecvCount = 0; + _msgRecvMatchCount = 0; + _msgAckTotalTime = 0; + _msgRecvTotalTime = 0; + _msgAckAvgTime = 0; + _msgRecvAvgTime = 0; + // END: Mesg Stats (void)_mqttClient.subscribe(mqttTopic(CloudConnectFrame::Topic::NormalPriority)); break; default: @@ -371,76 +417,93 @@ * \brief CloudConnectController::onCloudMessageStatusChanged * \details Correlates one message delivered by the broker back to the publish * that produced it and credits that msgId. - * \param msgId - message identifier - * \param status - new message status - * \param properties - additional properties specified by the server/broker + * \param[in] msgId - message identifier + * \param[in] status - new message status + * \param[in] properties - additional properties specified by the server/broker */ void CloudConnectController::onCloudMessageStatusChanged(qint32 id, QMqtt::MessageStatus status, const QMqttMessageStatusProperties &properties) { - Q_UNUSED(id) // SQ Q_UNUSED(properties) // SQ switch (status) { - case QMqtt::MessageStatus::Unknown: - qCInfo(logMqtt).noquote() << "message status changed to Unknown"; // SQ + case QMqtt::MessageStatus::Acknowledged: { // PUBACK (QoS 1&2) + // BEGIN: Mesg Stats + const auto it = _inflight.find(id); + if (it == _inflight.end()) { + qCWarning(logMqttStats).noquote() << QString("PUBACK for unknown publish id %1, ignoring").arg(id); + break; + } + it->status = Inflight::ACK; + + const qint64 ackUs = msgElapsedUs() - it->sentUs; + _msgAckCount++; + _msgAckTotalTime += ackUs; + _msgAckAvgTime = _msgAckTotalTime / qint64(_msgAckCount); + + qCInfo(logMqttStats).noquote() << msgStats(QStringLiteral("PUBACK"), it->msgId, it->sequence, + QStringLiteral("publish to ack"), ackUs, _msgAckAvgTime); + // END: Mesg Stats break; - case QMqtt::MessageStatus::Published: - qCInfo(logMqtt).noquote() << "message status changed to Published"; // SQ - break; - case QMqtt::MessageStatus::Acknowledged: - qCInfo(logMqtt).noquote() << "message status changed to Acknowledged"; // SQ - break; - case QMqtt::MessageStatus::Received: - qCInfo(logMqtt).noquote() << "message status changed to Received"; // SQ - break; - case QMqtt::MessageStatus::Released: - qCInfo(logMqtt).noquote() << "message status changed to Released"; // SQ - break; - case QMqtt::MessageStatus::Completed: - qCInfo(logMqtt).noquote() << "message status changed to Completed"; // SQ - break; + } + case QMqtt::MessageStatus::Unknown: + case QMqtt::MessageStatus::Published: // PUBLISH (QoS 1) + case QMqtt::MessageStatus::Received: // PUBREC (QoS 2) + case QMqtt::MessageStatus::Released: // PUBREL (QoS 2) + case QMqtt::MessageStatus::Completed: // PUBCOMP (QoS 2) default: break; } } /*! * \brief CloudConnectController::onCloudMessageReceived - * \details Correlates one message delivered by the broker back to the publish - * that produced it and credits that msgId. - * \param topic - topic the message was delivered on - * \param payload - serialised protobuf message - * \note The msgId is read from the message itself: every Leahi message carries - * Header at field 1, and Envelope exists to parse exactly that much - * without knowing the concrete message type. + * \details Handle received MQTT messages + * \param[in] message - MQTT message data + * \param[in] topic - MQTT topic */ -void CloudConnectController::onCloudMessageReceived(const QString &topic, const QByteArray &payload) +void CloudConnectController::onCloudMessageReceived(const QByteArray &message, const QMqttTopicName &topic) { - qCInfo(logMqtt).noquote() << QString("received message with topic=%1").arg(topic); // SQ + qCInfo(logMqtt).noquote() << QString("received message with topic=%1").arg(topic.name()); // SQ // TODO: filter by topic leahi::messages::Envelope envelope; - if (!envelope.ParseFromArray(payload.constData(), payload.size()) || !envelope.has_header()) { - // Unknown fields parse cleanly in proto3, so a foreign payload usually - // lands here as a header-less Envelope rather than as a parse failure. - qCWarning(logMqtt).noquote() << QString("received mesg (size=%1) for topic %2 do not contain a valid protobuf payload, ignoring") - .arg(payload.size()).arg(topic); + if (!envelope.ParseFromArray(message.constData(), message.size()) || !envelope.has_header()) { + qCWarning(logMqtt).noquote() << QString("received mesg (size=%1) for topic %2 do not contain a valid protobuf message, ignoring") + .arg(message.size()).arg(topic.name()); return; } const leahi::messages::Header &header = envelope.header(); const Can::MsgId msgId = Can::MsgId(header.msgid()); + const Can::Sequence sequence = static_cast(header.sequence()); - // No serial here: Header carries only timestamp/sequence/msgId, and the device - // identity travels in the topic (logged by handlePublishPacket) rather than the - // payload. Printing the timestamp instead also proves updateHeader populated it. const qint64 headerMsecs = header.timestamp().seconds() * 1000 + header.timestamp().nanos() / 1000000; qCInfo(logProto).noquote() << QString("header: msgId=0x%1 (%2), seq=%3, timestamp=%4") .arg(QString("%1").arg(msgId, 4, 16, QChar('0')).toUpper(), leahi::msgIdString(static_cast(msgId))) .arg(header.sequence()) .arg(QDateTime::fromMSecsSinceEpoch(headerMsecs).toString(Qt::ISODateWithMs)); + // BEGIN: Mesg Stats + const qint64 now = msgElapsedUs(); + _msgRecvCount++; + qint64 recvUs = -1; + for (auto it = _inflight.begin(); it != _inflight.end(); ++it) { + if (it->msgId == msgId && it->sequence == sequence) { + recvUs = now - it->sentUs; + _inflight.erase(it); + break; + } + } + if (recvUs >= 0) { + _msgRecvMatchCount++; + _msgRecvTotalTime += recvUs; + _msgRecvAvgTime = _msgRecvTotalTime / qint64(_msgRecvMatchCount); + } + qCInfo(logMqttStats).noquote() << msgStats(QStringLiteral("SUB"), msgId, sequence, + QStringLiteral("publish to receive"), recvUs, _msgRecvAvgTime); + // END: Mesg Stats + const std::string &typeName = leahi::msgIdToProtoName(msgId); if (typeName.empty()) { qCWarning(logProto).noquote() << QString("no protobuf mapping for message with msgId=0x%1, ignoring").arg(msgId, 4, 16, QChar('0')); @@ -456,7 +519,7 @@ google::protobuf::DynamicMessageFactory factory; std::unique_ptr body(factory.GetPrototype(descriptor)->New()); - if (!body->ParseFromArray(payload.constData(), static_cast(payload.size()))) { + if (!body->ParseFromArray(message.constData(), static_cast(message.size()))) { qCWarning(logProto).noquote() << QString("could not parse protobuf message %1, ignoring").arg(QString::fromStdString(typeName)); return; } @@ -468,3 +531,36 @@ google::protobuf::util::MessageToJsonString(*body, &json, options); qCDebug(logProto).noquote() << QString::fromStdString(typeName) << ":" << Qt::endl << QString::fromStdString(json); } + +// BEGIN: Mesg Stats +QString CloudConnectController::canStats(Can::MsgId msgId, qint64 sinceLastUs) const +{ + return QString("0x%1: count=%2, time since last=%3ms, avg time=%4ms") + .arg(QString("%1").arg(msgId, 4, 16, QChar('0')).toUpper()) + .arg(_canCount) + .arg(sinceLastUs / 1000.0, 6, 'f', 3, QChar(' ')) + .arg(_canAvgTime / 1000.0, 6, 'f', 3, QChar(' ')); +} + +qint64 CloudConnectController::msgElapsedUs() +{ + if (!_msgTimer.isValid()) { + _msgTimer.start(); + } + return _msgTimer.nsecsElapsed() / 1000; +} + +QString CloudConnectController::msgStats(const QString &label, Can::MsgId msgId, qint16 sequence, + const QString &latencyLabel, qint64 latencyUs, qint64 avgUs) const +{ + return QString("%1: msgId=0x%2 (%3), seq=%4, %5=%6, avg %7=%8ms") + .arg(label) + .arg(QString("%1").arg(msgId, 4, 16, QChar('0')).toUpper(), leahi::msgIdString(leahi::MsgId(msgId))) + .arg(sequence) + .arg(latencyLabel, (latencyUs >= 0) + ? QString("%1ms").arg(latencyUs / 1000.0, 6, 'f', 3, QChar(' ')) + : QStringLiteral("unknown")) + .arg(latencyLabel) + .arg(avgUs / 1000.0, 6, 'f', 3, QChar(' ')); +} +// END: Mesg Stats Index: CloudConnect/CloudConnectController.h =================================================================== diff -u -raaebfee335c74b0250864a6dce0555f866adadea -re9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65 --- CloudConnect/CloudConnectController.h (.../CloudConnectController.h) (revision aaebfee335c74b0250864a6dce0555f866adadea) +++ CloudConnect/CloudConnectController.h (.../CloudConnectController.h) (revision e9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65) @@ -15,11 +15,16 @@ #include #include #include +#include #include #include #include #include +#include +#include +#include + #include "CanInterface.h" #include "CanMessage.h" #include "CloudConnectFrame.h" @@ -41,10 +46,10 @@ ~CloudConnectController(); bool loadConfig(const QString &configPath); + void initThread(QThread &thread); bool startCan(); bool listenForApp(); bool connectToCloud(); - void initThread(QThread &thread); private: /*! @@ -74,18 +79,55 @@ QString _appServerSocketPath; QSharedPointer _appServer; MqttClient _mqttClient; + quint64 _publishSeqNum = 0; QString _topicPrefix; // TODO: define in INI? QString _deviceId; // TODO: this needs to be sent from Leahi app or retrieved from somewhere - QString _serialNumber; - QString _deviceCertPath; - QString _deviceKeyPath; + // BEGIN: Mesg Stats + struct MsgStats { + quint64 published = 0; ///< handed to the broker + quint64 echoed = 0; ///< came back on our own subscription and matched a publish + quint64 stale = 0; ///< came back but matched no in-flight publish + quint64 expired = 0; ///< published, never echoed within the echo window + }; + struct Inflight { + enum Status { + Unknown, + Sent, + ACK, + Published, + }; + Status status = Unknown; + Can::MsgId msgId = 0; + Can::Sequence sequence = 0; + qint64 sentUs = 0; ///< _msgTimer reading when the publish was handed to the broker + }; + QString canStats(Can::MsgId msgId, qint64 sinceLastUs) const; + qint64 msgElapsedUs(); + QString msgStats(const QString &label, Can::MsgId msgId, qint16 sequence, + const QString &latencyLabel, qint64 latencyUs, qint64 avgUs) const; + quint64 _canCount = 0; + qint64 _canAvgTime = 0; + qint64 _canTotalTime = 0; + qint64 _canLastRecv = 0; + QElapsedTimer _canTimer; + QElapsedTimer _msgTimer; + QHash _inflight; + quint64 _msgSentCount = 0; + quint64 _msgAckCount = 0; + quint64 _msgRecvCount = 0; + quint64 _msgRecvMatchCount = 0; + qint64 _msgAckTotalTime = 0; + qint64 _msgRecvTotalTime = 0; + qint64 _msgAckAvgTime = 0; + qint64 _msgRecvAvgTime = 0; + // END: Mesg Stats private Q_SLOTS: void onCanFrameReceive(const QCanBusFrame &frame); void onCanMessageReceive(const Can::Message &msg); void onCloudStateChanged(QMqttClient::ClientState state); void onCloudMessageStatusChanged(qint32 id, QMqtt::MessageStatus status, const QMqttMessageStatusProperties &properties); - void onCloudMessageReceived(const QString &topic, const QByteArray &payload); + void onCloudMessageReceived(const QByteArray &message, const QMqttTopicName &topic); }; Index: lib/Comms/Comms.pro =================================================================== diff -u -raaebfee335c74b0250864a6dce0555f866adadea -re9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65 --- lib/Comms/Comms.pro (.../Comms.pro) (revision aaebfee335c74b0250864a6dce0555f866adadea) +++ lib/Comms/Comms.pro (.../Comms.pro) (revision e9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65) @@ -7,6 +7,8 @@ QMAKE_CXXFLAGS += -Wall -Werror -Wextra +CONFIG(debug, debug|release): QMAKE_CXXFLAGS += -U_FORTIFY_SOURCE + DESTDIR = $$PWD/lib HEADERS = \ Index: lib/Comms/include/MqttClient.h =================================================================== diff -u -raaebfee335c74b0250864a6dce0555f866adadea -re9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65 --- lib/Comms/include/MqttClient.h (.../MqttClient.h) (revision aaebfee335c74b0250864a6dce0555f866adadea) +++ lib/Comms/include/MqttClient.h (.../MqttClient.h) (revision e9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65) @@ -75,7 +75,7 @@ * \brief Emitted for every message the broker delivers to this client. * \details Covers all active subscriptions; the receiver demultiplexes by topic. */ - void didMessageReceived(const QString &topic, const QByteArray &payload); + void didMessageReceived(const QByteArray &message, const QMqttTopicName &topic); private Q_SLOTS: void onStateChanged(); Index: lib/Comms/src/MqttClient.cpp =================================================================== diff -u -raaebfee335c74b0250864a6dce0555f866adadea -re9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65 --- lib/Comms/src/MqttClient.cpp (.../MqttClient.cpp) (revision aaebfee335c74b0250864a6dce0555f866adadea) +++ lib/Comms/src/MqttClient.cpp (.../MqttClient.cpp) (revision e9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65) @@ -146,11 +146,7 @@ connect(_client, &QMqttClient::stateChanged, this, &MqttClient::onStateChanged); connect(_client, &QMqttClient::messageStatusChanged, this, &MqttClient::didMessageStatusChanged); // Fires for every delivery on this connection, whichever subscription matched. - connect(_client, &QMqttClient::messageReceived, this, - [this](const QByteArray &message, const QMqttTopicName &topic) { - Q_EMIT didMessageReceived(topic.name(), message); - } - ); + connect(_client, &QMqttClient::messageReceived, this, &MqttClient::didMessageReceived); connect(_client, &QMqttClient::errorChanged, this, [](QMqttClient::ClientError error) { if (error != QMqttClient::NoError) { @@ -307,7 +303,7 @@ qCInfo(logMqtt).noquote() << "client connecting"; Q_EMIT didStateChanged(_client->state()); break; - case QMqttClient::Connected: + case QMqttClient::Connected: // CONNACK qCInfo(logMqtt).noquote() << "client connected"; // Subscriptions did not survive the previous client, so this is where // a receiver re-establishes them. Index: lib/MsgUtils/MsgUtils.pro =================================================================== diff -u -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 -re9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65 --- lib/MsgUtils/MsgUtils.pro (.../MsgUtils.pro) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) +++ lib/MsgUtils/MsgUtils.pro (.../MsgUtils.pro) (revision e9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65) @@ -7,6 +7,8 @@ QMAKE_CXXFLAGS += -Wall -Werror -Wextra +CONFIG(debug, debug|release): QMAKE_CXXFLAGS += -U_FORTIFY_SOURCE + DESTDIR = $$PWD/lib include(codegen_tools.pri) Index: tools/CANDumpPlayer/CANDumpPlayer.pro =================================================================== diff -u -r51e99f2578e0901d9da91a4cb60d1b8858cfe971 -re9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65 --- tools/CANDumpPlayer/CANDumpPlayer.pro (.../CANDumpPlayer.pro) (revision 51e99f2578e0901d9da91a4cb60d1b8858cfe971) +++ tools/CANDumpPlayer/CANDumpPlayer.pro (.../CANDumpPlayer.pro) (revision e9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65) @@ -6,6 +6,8 @@ QMAKE_CXXFLAGS += -Wall -Werror -Wextra +CONFIG(debug, debug|release): QMAKE_CXXFLAGS += -U_FORTIFY_SOURCE + DESTDIR = $$PWD/../bin SOURCES = \ Index: tools/CANDumpPlayer/main.cpp =================================================================== diff -u -r46d7af4d93047a9c416f93447bd91f94513a000a -re9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65 --- tools/CANDumpPlayer/main.cpp (.../main.cpp) (revision 46d7af4d93047a9c416f93447bd91f94513a000a) +++ tools/CANDumpPlayer/main.cpp (.../main.cpp) (revision e9aa2c82ccb8cb5662f05ed878d2f7b5ef9f4a65) @@ -13,6 +13,9 @@ #include #include +constexpr quint8 syncByte = 0xA5; +constexpr int headerMsgIdOffset = 3; + int main(int argc, char *argv[]) { QCoreApplication app(argc, argv); @@ -90,7 +93,7 @@ // Candump files may use either Unix epoch ("1234567890.123456") or // wall-clock ("YYYY-MM-DD HH:MM:SS.ffffff") timestamp format. - // We parse both and normalise to microseconds since epoch as qint64. + // Parse both and normalise to microseconds since epoch as qint64. const QString datetimeFormat = QStringLiteral("yyyy-MM-dd HH:mm:ss.zzz"); auto parseTimestampUs = [&](const QString &raw) -> qint64 { @@ -115,16 +118,31 @@ }; // Record the monotonic clock origin at the start of replay. - // Each frame's absolute wake time is: origin + (logOffset / speed). - // This absorbs per-frame overshoot automatically — no drift accumulation. + // Use origin + (logOffset / speed) for frame replay time to prevent accumulating drift. struct timespec replayOrigin; clock_gettime(CLOCK_MONOTONIC, &replayOrigin); qint64 firstTimestampUs = -1; + qint64 lastTimestampUs = -1; qint64 prevTimestampUs = -1; QElapsedTimer wallTimer; unsigned int lineCount = 1; + quint64 playedCount = 0; + qint64 totalDeltaNs = 0; + qint64 avgDeltaNs = 0; + QElapsedTimer replayTimer; + replayTimer.start(); + + quint64 messageCount = 0; + QElapsedTimer msgTimer; + qint64 msgTotalDeltaNs = 0; + qint64 msgAvgDeltaNs = 0; + qint64 msgDeltaNs = 0; + qint64 msgCalcDeltaUs = 0; + qint64 msgPrevTimestampUs = -1; + unsigned int msgId = 0; + while (stream.readLineInto(&line)) { auto match = regexEntry.match(line); if (match.hasMatch()) { @@ -135,13 +153,22 @@ payload.append(static_cast(payloadMatch.captured(1).toUInt(nullptr, 16))); } + // Grossly count sent messages based off frames that start with the sync byte, + // this may not be completely accurate, but close enough. + const bool isMessageStart = !payload.isEmpty() && static_cast(payload.at(0)) == syncByte; + const qint64 timestampUs = parseTimestampUs(match.captured(QStringLiteral("timestamp"))); const qint64 calcDeltaUs = (prevTimestampUs >= 0 && timestampUs >= 0) ? timestampUs - prevTimestampUs : 0; - if (speed > 0.0 && timestampUs >= 0) { + // Save the first and last time stamps for reporting later + if (timestampUs >= 0) { if (firstTimestampUs < 0) { firstTimestampUs = timestampUs; } + lastTimestampUs = timestampUs; + } + + if (speed > 0.0 && timestampUs >= 0) { const qint64 offsetUs = static_cast((timestampUs - firstTimestampUs) / speed); struct timespec wakeTime; wakeTime.tv_sec = replayOrigin.tv_sec + offsetUs / 1'000'000; @@ -151,20 +178,61 @@ clock_nanosleep(CLOCK_MONOTONIC, TIMER_ABSTIME, &wakeTime, nullptr); } - if (testMode) { - const qint64 actualDeltaUs = (prevTimestampUs >= 0) ? wallTimer.nsecsElapsed() / 1000 : 0; - wallTimer.restart(); - qInfo().noquote() << QString("delta(calc=%1us, actual=%2us) %3") - .arg(calcDeltaUs, 6) - .arg(actualDeltaUs, 6) - .arg(line.trimmed()); + // Wall-clock interval since the previous frame. + // start() on an already-running timer restarts it. + const qint64 actualDeltaNs = wallTimer.isValid() ? wallTimer.nsecsElapsed() : 0; + wallTimer.start(); + playedCount++; + if (playedCount > 1) { + totalDeltaNs += actualDeltaNs; + avgDeltaNs = totalDeltaNs / qint64(playedCount - 1); } - else { - const unsigned int canId = match.captured(QStringLiteral("can_id")).toUInt(nullptr, 16); + + // Same interval measurement, but only across message starts, so it spans a + // whole message rather than one frame of it. + if (isMessageStart) { + msgDeltaNs = msgTimer.isValid() ? msgTimer.nsecsElapsed() : 0; + msgTimer.start(); + messageCount++; + if (messageCount > 1) { + msgTotalDeltaNs += msgDeltaNs; + msgAvgDeltaNs = msgTotalDeltaNs / qint64(messageCount - 1); + } + // Logged interval between sync byte frames + msgCalcDeltaUs = (msgPrevTimestampUs >= 0 && timestampUs >= 0) ? timestampUs - msgPrevTimestampUs : 0; + msgPrevTimestampUs = timestampUs; + + msgId = (payload.size() >= headerMsgIdOffset + 2) + ? (static_cast(payload.at(headerMsgIdOffset)) << 8) + | static_cast(payload.at(headerMsgIdOffset + 1)) + : 0; + } + + const unsigned int canId = match.captured(QStringLiteral("can_id")).toUInt(nullptr, 16); + if (!testMode) { QCanBusFrame frame(canId, payload); canDevice->writeFrame(frame); } + qInfo().noquote() << QString("Frame %1: canId=0x%2, time since last frame [actual=%3ms, calc=%4ms], " + "avg time between frames=%5ms, avg time between messages=%6ms") + .arg(playedCount) + .arg(QString("%1").arg(canId, 4, 16, QChar('0')).toUpper()) + .arg((playedCount > 1) ? actualDeltaNs / 1'000'000.0 : 0.0, 6, 'f', 3, QChar(' ')) + .arg((playedCount > 1) ? calcDeltaUs / 1000.0 : 0.0, 6, 'f', 3, QChar(' ')) + .arg(avgDeltaNs / 1'000'000.0, 6, 'f', 3, QChar(' ')) + .arg(msgAvgDeltaNs / 1'000'000.0, 6, 'f', 3, QChar(' ')); + + if (isMessageStart) { + qInfo().noquote() << QString("Message %1: msgId=0x%2, time since last [actual=%3ms, calc=%4ms], " + "avg time between=%5ms.") + .arg(messageCount) + .arg(QString("%1").arg(QString("%1").arg(msgId, 4, 16, QChar('0')).toUpper()).toUpper()) + .arg((messageCount > 1) ? msgDeltaNs / 1'000'000.0 : 0.0, 6, 'f', 3, QChar(' ')) + .arg((messageCount > 1) ? msgCalcDeltaUs / 1000.0 : 0.0, 6, 'f', 3, QChar(' ')) + .arg(msgAvgDeltaNs / 1'000'000.0, 6, 'f', 3, QChar(' ')); + } + prevTimestampUs = timestampUs; } else { @@ -174,6 +242,19 @@ lineCount++; } + // Span the dump covers, i.e. the last frame timestamp minus the first. Compare + // against the replay time to see how faithfully the replay tracked the log. + const qint64 logDurationUs = (firstTimestampUs >= 0) ? (lastTimestampUs - firstTimestampUs) : 0; + + qInfo().noquote() << QString("replay complete: messages(approx)=%1, frames=%2, avg time between frames=%3ms, " + "avg time between messages=%4ms, total replay time=%5ms, total log time=%6m") + .arg(messageCount) + .arg(playedCount) + .arg(avgDeltaNs / 1'000'000.0, 0, 'f', 3, QChar(' ')) + .arg(msgAvgDeltaNs / 1'000'000.0, 0, 'f', 3, QChar(' ')) + .arg(replayTimer.nsecsElapsed() / 1'000'000.0, 0, 'f', 3, QChar(' ')) + .arg(logDurationUs / 60'000'000.0, 0, 'f', 2, QChar(' ')); + file.close(); if (canDevice) { canDevice->disconnectDevice();