Initial code for getXferTelemetry API.

Author: mkhazraee

src/api/cpp/nixl.h

@@ -289,6 +289,18 @@ class nixlAgent {
         nixl_status_t
         getXferStatus (nixlXferReqH* req_hndl) const;
 
+
+        /**
+         * @brief  Get the telemetry data associated with `req_hndl`.
+         *
+         * @param  req_hndl        Transfer request handle obtained from makeXferReq/createXferReq
+         * @param  telemetry [out] Output telemetry information
+         * @return nixl_status_t   Error code if call was not successful
+         */
+        nixl_status_t
+        getXferTelemetry (const nixlXferReqH* req_hndl,
+                          nixl_xfer_telem_t &telemetry) const;
+
         /**
          * @brief  Query the backend associated with `req_hndl`. E.g., if for genNotif
          *         the same backend as a transfer is desired.

src/api/cpp/nixl_types.h

@@ -20,6 +20,7 @@
 #include <string>
 #include <unordered_map>
 #include <optional>
+#include <chrono>
 
 
 /*** Forward declarations ***/
@@ -62,7 +63,8 @@ enum nixl_status_t {
     NIXL_ERR_UNKNOWN = -8,
     NIXL_ERR_NOT_SUPPORTED = -9,
     NIXL_ERR_REMOTE_DISCONNECT = -10,
-    NIXL_ERR_CANCELED = -11
+    NIXL_ERR_CANCELED = -11,
+    NIXL_ERR_NO_TELEMETRY = -12
 };
 
 /**
@@ -223,6 +225,65 @@ struct nixlAgentOptionalArgs {
  */
 using nixl_opt_args_t = nixlAgentOptionalArgs;
 
+/**
+ * @brief A typedefs for a point in time
+ */
+using chrono_point_t = std::chrono::steady_clock::time_point;
+
+/**
+ * @brief A constant indicating min chrono_point_t value
+ */
+constexpr auto min_chrono_time = std::chrono::steady_clock::time_point::min();
+
+/**
+ * @brief A typedefs for a period of time
+ */
+using chrono_period_t = std::chrono::microseconds;
+
+/**
+ * @struct nixlXferTelemetry
+ * @brief A structure for telemetry output from agent API
+ */
+struct nixlXferTelemetry {
+    /**
+     * @var backendType Type of the backend performing the transfer
+     */
+    nixl_backend_t backendType;
+
+    /**
+     * @var startTime Time that the transfer was posted
+     */
+    chrono_point_t startTime;
+
+    /**
+     * @var postDuration Time it took to do the post operation
+     */
+    chrono_period_t postDuration;
+
+    /**
+     * @var xferDuration Time it took to complete the transfer
+     *      if checkXferReq is called late, that might impact this result
+     */
+    chrono_period_t xferDuration;
+
+    /**
+     * @var totalBytes Amount of bytes transfered in the request
+     */
+    size_t totalBytes;
+
+    /**
+     * @var descCount Number of descriptors in the transfer request.
+     *      If any merging of descriptors were performed, it will be reflected here.
+     */
+    size_t descCount;
+};
+
+/**
+ * @brief A typedef for a nixlXferTelemetry
+ *        for telemetry output.
+ */
+using nixl_xfer_telem_t = nixlXferTelemetry;
+
 /**
  * @brief A define for an empty string, that indicates the descriptor list is being
  *        prepared for the local agent as an initiator in prepXferDlist method.

src/core/nixl_agent.cpp

@@ -89,24 +89,24 @@ nixlXferReqH::updateRequestStats(std::unique_ptr<nixlTelemetry> &telemetry_pub,
     static const std::array<std::string, 3> nixl_post_status_str = {
         " Posted", " Posted and Completed", " Completed"};
     auto duration = std::chrono::duration_cast<std::chrono::microseconds>(
-        std::chrono::steady_clock::now() - telemetry.startTime);
+        std::chrono::steady_clock::now() - telemetry.startTime_);
 
     if (stat_status == NIXL_TELEMETRY_POST) {
         telemetry.postDuration_ = duration;
     } else if (stat_status == NIXL_TELEMETRY_POST_AND_FINISH) {
         telemetry.postDuration_ = duration;
         telemetry.xferDuration_ = duration;
         telemetry_pub->addPostTime(duration);
-        telemetry_pub->addXferTime(duration, backendOp == NIXL_WRITE, telemetry.totalBytes);
+        telemetry_pub->addXferTime(duration, backendOp == NIXL_WRITE, telemetry.totalBytes_);
     } else { // stat_status == NIXL_TELEMETRY_FINISH
         telemetry.xferDuration_ = duration;
         telemetry_pub->addPostTime(telemetry.postDuration_);
-        telemetry_pub->addXferTime(duration, backendOp == NIXL_WRITE, telemetry.totalBytes);
+        telemetry_pub->addXferTime(duration, backendOp == NIXL_WRITE, telemetry.totalBytes_);
     }
 
     NIXL_TRACE << "[NIXL TELEMETRY]: From backend " << engine->getType()
                << nixl_post_status_str[stat_status] << " Xfer with " << initiatorDescs->descCount()
-               << " descriptors of total size " << telemetry.totalBytes << "B in "
+               << " descriptors of total size " << telemetry.totalBytes_ << "B in "
                << duration.count() << "us.";
 }
 
@@ -751,7 +751,7 @@ nixlAgent::makeXferReq (const nixl_xfer_op_t &operation,
     handle->hasNotif = opt_args.hasNotif;
     handle->backendOp = operation;
     handle->status = NIXL_ERR_NOT_POSTED;
-    handle->telemetry.totalBytes = total_bytes;
+    handle->telemetry.totalBytes_ = total_bytes;
 
     ret = handle->engine->prepXfer (handle->backendOp,
                                     *handle->initiatorDescs,
@@ -877,7 +877,7 @@ nixlAgent::createXferReq(const nixl_xfer_op_t &operation,
     handle->status = NIXL_ERR_NOT_POSTED;
     handle->notifMsg = opt_args.notifMsg;
     handle->hasNotif = opt_args.hasNotif;
-    handle->telemetry.totalBytes = total_bytes;
+    handle->telemetry.totalBytes_ = total_bytes;
 
     ret1 = handle->engine->prepXfer (handle->backendOp,
                                      *handle->initiatorDescs,
@@ -941,7 +941,7 @@ nixlAgent::postXferReq(nixlXferReqH *req_hndl,
         return NIXL_ERR_INVALID_PARAM;
     }
 
-    if (data->telemetry_) req_hndl->telemetry.startTime = std::chrono::steady_clock::now();
+    if (data->telemetry_) req_hndl->telemetry.startTime_ = std::chrono::steady_clock::now();
 
     NIXL_SHARED_LOCK_GUARD(data->lock);
     // Check if the remote was invalidated before post/repost
@@ -1049,6 +1049,25 @@ nixlAgent::getXferStatus (nixlXferReqH *req_hndl) const {
     return req_hndl->status;
 }
 
+nixl_status_t
+nixlAgent::getXferTelemetry (const nixlXferReqH* req_hndl,
+                             nixl_xfer_telem_t &telemetry) const{
+    if (!data->telemetry_)
+        return NIXL_ERR_NO_TELEMETRY;
+
+    if (req_hndl->status <0)
+        return req_hndl->status;
+
+    // NIXL_SUCCESS or NIXL_IN_PROG, values are initialized
+    telemetry.backendType = req_hndl->engine->getType();
+    telemetry.startTime = req_hndl->telemetry.startTime_;
+    telemetry.postDuration = req_hndl->telemetry.postDuration_;
+    telemetry.xferDuration = req_hndl->telemetry.xferDuration_;
+    telemetry.totalBytes = req_hndl->telemetry.totalBytes_;
+    telemetry.descCount = req_hndl->initiatorDescs->descCount();
+
+    return req_hndl->status;
+}
 
 nixl_status_t
 nixlAgent::queryXferBackend(const nixlXferReqH* req_hndl,

src/core/transfer_request.h

@@ -17,7 +17,6 @@
 #ifndef __TRANSFER_REQUEST_H_
 #define __TRANSFER_REQUEST_H_
 
-#include <chrono>
 #include <string>
 #include <unordered_map>
 #include <memory>
@@ -26,9 +25,6 @@
 #include "backend_engine.h"
 #include "telemetry.h"
 
-using chrono_point_t = std::chrono::steady_clock::time_point;
-using std::chrono::microseconds;
-
 enum nixl_telemetry_stat_status_t {
     NIXL_TELEMETRY_POST = 0,
     NIXL_TELEMETRY_POST_AND_FINISH = 1,
@@ -52,11 +48,12 @@ class nixlXferReqH {
         nixl_xfer_op_t     backendOp;
         nixl_status_t      status;
 
+        // Initializing time values, as getXferTelem might be called before postXfer.
         struct {
-            chrono_point_t startTime;
-            microseconds postDuration_ = microseconds(0);
-            microseconds xferDuration_ = microseconds(0);
-            size_t totalBytes;
+            chrono_point_t startTime_ = min_chrono_time;
+            chrono_period_t postDuration_ = chrono_period_t(0);
+            chrono_period_t xferDuration_ = chrono_period_t(0);
+            size_t totalBytes_;
         } telemetry;
 
     public: