Merge pull request #1105 from slaclab/writer-bandwidth

Add bandwidth tracking and display to StreamWriter and PyDM DataWriter

Author: ruck314

include/rogue/utilities/fileio/StreamWriter.h

@@ -36,12 +36,15 @@
 
 #include <stdint.h>
 
+#include <chrono>
 #include <condition_variable>
+#include <deque>
 #include <map>
 #include <memory>
 #include <mutex>
 #include <string>
 #include <thread>
+#include <utility>
 
 #include "rogue/EnableSharedFromThis.h"
 #include "rogue/Logging.h"
@@ -100,6 +103,12 @@ class StreamWriter : public rogue::EnableSharedFromThis<rogue::utilities::fileio
     //! Total number of frames in file
     uint32_t frameCount_;
 
+    //! Bytes recorded in the last one second window
+    uint64_t bandwidthBytes_;
+
+    //! Time tagged byte counts for bandwidth calculation
+    std::deque<std::pair<std::chrono::steady_clock::time_point, uint32_t>> bandwidthHistory_;
+
     //! Internal method for file writing
     void intWrite(void* data, uint32_t size);
 
@@ -109,6 +118,12 @@ class StreamWriter : public rogue::EnableSharedFromThis<rogue::utilities::fileio
     //! Flush file
     void flush();
 
+    //! Update bandwidth accounting
+    void recordBandwidth(uint32_t size);
+
+    //! Remove stale bandwidth samples
+    void pruneBandwidth(std::chrono::steady_clock::time_point now);
+
     //! Write raw data
     bool raw_;
 
@@ -166,6 +181,9 @@ class StreamWriter : public rogue::EnableSharedFromThis<rogue::utilities::fileio
     //! Get current file size
     uint64_t getCurrentSize();
 
+    //! Get instantaneous bandwidth in bytes per second
+    double getBandwidth();
+
     //! Get current frame count
     uint32_t getFrameCount();
 

python/pyrogue/_DataWriter.py

@@ -98,6 +98,17 @@ def __init__(self, *, hidden=True, bufferSize=0, maxFileSize=0, **kwargs):
             localGet=self._getTotalSize,
             description='Total bytes written.'))
 
+        self.add(pr.LocalVariable(
+            name='Bandwidth',
+            mode='RO',
+            value=0.0,
+            typeStr='Float64',
+            pollInterval=1,
+            localGet=self._getBandwidth,
+            units='Bytes/sec',
+            disp='{:,.3f}',
+            description='Instantaneous write bandwidth in bytes per second.'))
+
         self.add(pr.LocalVariable(
             name='FrameCount',
             mode='RO',
@@ -162,6 +173,10 @@ def _getTotalSize(self):
         """get total file size. Override in sub-class"""
         return 0
 
+    def _getBandwidth(self, dev=None, var=None):
+        """get instantaneous bandwidth. Override in sub-class"""
+        return 0.0
+
     def _getFrameCount(self):
         """get current file frame count. Override in sub-class"""
         return 0

python/pyrogue/pydm/widgets/data_writer.py

@@ -18,7 +18,7 @@
 from pyrogue.pydm.widgets import PyRogueLineEdit
 from pyrogue.pydm.data_plugins.rogue_plugin import nodeFromAddress
 from qtpy.QtCore import Qt, Slot
-from qtpy.QtWidgets import QVBoxLayout, QHBoxLayout, QPushButton
+from qtpy.QtWidgets import QVBoxLayout, QHBoxLayout, QPushButton, QWidget, QLabel
 from qtpy.QtWidgets import QFormLayout, QGroupBox, QFileDialog
 import datetime
 
@@ -114,6 +114,24 @@ def connection_changed(self, connected):
         w.alarmSensitiveBorder  = True
         fl.addRow('Frame Count:',w)
 
+        bwWidget = QWidget()
+        bwLayout = QHBoxLayout()
+        bwLayout.setContentsMargins(0,0,0,0)
+        bwLayout.setSpacing(4)
+        bwWidget.setLayout(bwLayout)
+
+        w = PyDMLabel(parent=None, init_channel=self._path + '.Bandwidth/disp')
+        w.alarmSensitiveContent = False
+        w.alarmSensitiveBorder  = True
+        bwLayout.addWidget(w)
+
+        units = QLabel('Bytes/sec')
+        units.setAlignment(Qt.AlignLeft | Qt.AlignVCenter)
+        bwLayout.addWidget(units)
+        bwLayout.setAlignment(Qt.AlignLeft | Qt.AlignVCenter)
+
+        fl.addRow('Bandwidth:',bwWidget)
+
         w = PyDMLabel(parent=None, init_channel=self._path + '.TotalSize/disp')
         w.alarmSensitiveContent = False
         w.alarmSensitiveBorder  = True

python/pyrogue/utilities/fileio/_StreamWriter.py

@@ -72,6 +72,9 @@ def _getTotalSize(self,dev,var):
 
         return self._writer.getTotalSize()
 
+    def _getBandwidth(self,dev,var):
+        return self._writer.getBandwidth()
+
     def _getFrameCount(self,dev,var):
         return self._writer.getFrameCount()
 

src/rogue/utilities/fileio/StreamWriter.cpp

@@ -32,6 +32,7 @@
 
 #include "rogue/utilities/fileio/StreamWriter.h"
 
+#include <chrono>
 #include <fcntl.h>
 #include <inttypes.h>
 #include <stdint.h>
@@ -79,6 +80,7 @@ void ruf::StreamWriter::setup_python() {
         .def("getChannel", &ruf::StreamWriter::getChannel)
         .def("getTotalSize", &ruf::StreamWriter::getTotalSize)
         .def("getCurrentSize", &ruf::StreamWriter::getCurrentSize)
+        .def("getBandwidth", &ruf::StreamWriter::getBandwidth)
         .def("getFrameCount", &ruf::StreamWriter::getFrameCount)
         .def("waitFrameCount", &ruf::StreamWriter::waitFrameCount);
 #endif
@@ -94,6 +96,7 @@ ruf::StreamWriter::StreamWriter() {
     totSize_    = 0;
     buffer_     = NULL;
     frameCount_ = 0;
+    bandwidthBytes_ = 0;
     currBuffer_ = 0;
     dropErrors_ = false;
     isOpen_     = false;
@@ -134,6 +137,8 @@ void ruf::StreamWriter::open(std::string file) {
     totSize_    = 0;
     currSize_   = 0;
     frameCount_ = 0;
+    bandwidthBytes_ = 0;
+    bandwidthHistory_.clear();
     currBuffer_ = 0;
 
     // Iterate over all channels and reset their frame counts
@@ -150,6 +155,8 @@ void ruf::StreamWriter::close() {
     std::lock_guard<std::mutex> lock(mtx_);
     isOpen_ = false;
     flush();
+    bandwidthBytes_ = 0;
+    bandwidthHistory_.clear();
     if (fd_ >= 0) ::close(fd_);
     fd_ = -1;
 }
@@ -231,6 +238,22 @@ uint64_t ruf::StreamWriter::getCurrentSize() {
     return (currSize_ + currBuffer_);
 }
 
+//! Get instantaneous bandwidth in bytes per second over the last second
+double ruf::StreamWriter::getBandwidth() {
+    rogue::GilRelease noGil;
+    std::lock_guard<std::mutex> lock(mtx_);
+
+    auto now = std::chrono::steady_clock::now();
+    pruneBandwidth(now);
+
+    if (bandwidthHistory_.empty()) return (0.0);
+
+    auto windowStart = bandwidthHistory_.front().first;
+    double seconds   = std::chrono::duration<double>(now - windowStart).count();
+    if (seconds <= 0.0) return (static_cast<double>(bandwidthBytes_));
+    return (static_cast<double>(bandwidthBytes_) / seconds);
+}
+
 //! Get current frame count
 uint32_t ruf::StreamWriter::getFrameCount() {
     return (frameCount_);
@@ -266,6 +289,22 @@ bool ruf::StreamWriter::waitFrameCount(uint32_t count, uint64_t timeout) {
     return (frameCount_ >= count);
 }
 
+void ruf::StreamWriter::pruneBandwidth(std::chrono::steady_clock::time_point now) {
+    auto cutoff = now - std::chrono::seconds(1);
+    while (!bandwidthHistory_.empty() && bandwidthHistory_.front().first < cutoff) {
+        bandwidthBytes_ -= bandwidthHistory_.front().second;
+        bandwidthHistory_.pop_front();
+    }
+}
+
+void ruf::StreamWriter::recordBandwidth(uint32_t size) {
+    if (size == 0) return;
+    auto now = std::chrono::steady_clock::now();
+    pruneBandwidth(now);
+    bandwidthHistory_.emplace_back(now, size);
+    bandwidthBytes_ += size;
+}
+
 //! Write data to file. Called from StreamWriterChannel
 void ruf::StreamWriter::writeFile(uint8_t channel, std::shared_ptr<rogue::interfaces::stream::Frame> frame) {
     ris::Frame::BufferIterator it;
@@ -333,6 +372,8 @@ void ruf::StreamWriter::intWrite(void* data, uint32_t size) {
         std::memcpy(buffer_ + currBuffer_, data, size);
         currBuffer_ += size;
     }
+
+    recordBandwidth(size);
 }
 
 //! Check file size for next write