Line data    Source code

       1              : #ifndef _artdaq_DAQrate_RequestBuffer_hh_
       2              : #define _artdaq_DAQrate_RequestBuffer_hh_
       3              : 
       4              : #include "TRACE/tracemf.h"  // Pre-empt TRACE/trace.h from Fragment.hh.
       5              : #include "artdaq-core/Data/Fragment.hh"
       6              : 
       7              : #include <condition_variable>
       8              : #include <map>
       9              : #include <set>
      10              : 
      11              : namespace artdaq {
      12              : /**
      13              :  * @brief Holds requests from RequestReceiver while they are being processed
      14              :  */
      15              : class RequestBuffer
      16              : {
      17              : public:
      18              :         /**
      19              :          * @brief RequestBuffer Constructor
      20              :          */
      21              :         RequestBuffer();
      22              : 
      23              :         /**
      24              :          * @brief RequestBuffer Destructor
      25              :          */
      26              :         virtual ~RequestBuffer();
      27              : 
      28              :         /**
      29              :          * @brief Add a Request to the buffer
      30              :          * @param seq Sequence ID of the request
      31              :          * @param ts Timestamp for the request
      32              :          */
      33              :         void push(artdaq::Fragment::sequence_id_t seq, artdaq::Fragment::timestamp_t ts);
      34              : 
      35              :         /**
      36              :          * @brief Reset RequestBuffer, discarding all requests and tracking information
      37              :          */
      38              :         void reset();
      39              : 
      40              :         /// <summary>
      41              :         /// Get the current requests
      42              :         /// </summary>
      43              :         /// <returns>Map relating sequence IDs to timestamps</returns>
      44              :         std::map<artdaq::Fragment::sequence_id_t, artdaq::Fragment::timestamp_t> GetRequests() const;
      45              : 
      46              :         /// <summary>
      47              :         /// Get the current requests that are older than a specified time point
      48              :         /// </summary>
      49              :         /// <param name="older_than">Return requests that were received before the given time point</param>
      50              :         /// <returns>Map relating sequence IDs to timestamps</returns>
      51              :         std::map<artdaq::Fragment::sequence_id_t, artdaq::Fragment::timestamp_t> GetRequests(std::chrono::steady_clock::time_point older_than) const;
      52              : 
      53              :         /// <summary>
      54              :         /// Get the "next" request, i.e. the first unsatisfied request that has not already been returned by GetNextRequest
      55              :         /// </summary>
      56              :         /// <returns>Request data for "next" request. Will return (0,0) if there is no "next" request</returns>
      57              :         ///
      58              :         /// This function uses last_next_request_ to ensure that it does not return the same request more than once
      59              :         std::pair<artdaq::Fragment::sequence_id_t, artdaq::Fragment::timestamp_t> GetNextRequest();
      60              : 
      61              :         /// <summary>
      62              :         /// Remove the request with the given sequence ID from the request map
      63              :         /// </summary>
      64              :         /// <param name="reqID">Request ID to remove</param>
      65              :         void RemoveRequest(artdaq::Fragment::sequence_id_t reqID);
      66              : 
      67              :         /// <summary>
      68              :         /// Clear all requests from the map
      69              :         /// </summary>
      70              :         void ClearRequests();
      71              :         /// <summary>
      72              :         /// Get the current requests, then clear the map
      73              :         /// </summary>
      74              :         /// <returns>Map relating sequence IDs to timestamps</returns>
      75              :         std::map<artdaq::Fragment::sequence_id_t, artdaq::Fragment::timestamp_t> GetAndClearRequests();
      76              : 
      77              :         /// <summary>
      78              :         /// Get the number of requests currently stored in the RequestReceiver
      79              :         /// </summary>
      80              :         /// <returns>The number of requests stored in the RequestReceiver</returns>
      81              :         size_t size();
      82              :         /// <summary>
      83              :         /// Wait for a new request message, up to the timeout given
      84              :         /// </summary>
      85              :         /// <param name="timeout_ms">Milliseconds to wait for a new request to arrive</param>
      86              :         /// <returns>True if any requests are present in the request map</returns>
      87              :         bool WaitForRequests(int timeout_ms);
      88              : 
      89              :         /// <summary>
      90              :         /// Get the time a given request was received
      91              :         /// </summary>
      92              :         /// <param name="reqID">Request ID of the request</param>
      93              :         /// <returns>steady_clock::time_point corresponding to when the request was received</returns>
      94              :         std::chrono::steady_clock::time_point GetRequestTime(artdaq::Fragment::sequence_id_t reqID);
      95              : 
      96              :         /**
      97              :          * @brief Determine whether the RequestBuffer is active
      98              :          * @return
      99              :          */
     100            3 :         bool isRunning() const { return receiver_running_; }
     101              :         /**
     102              :          * @brief Set whether the RequestBuffer is active
     103              :          * @param running Whether the RequestBuffer is active
     104              :          */
     105           28 :         void setRunning(bool running) { receiver_running_ = running; }
     106              : 
     107              : private:
     108              :         std::map<artdaq::Fragment::sequence_id_t, artdaq::Fragment::timestamp_t> requests_;
     109              :         std::map<artdaq::Fragment::sequence_id_t, std::chrono::steady_clock::time_point> request_timing_;
     110              :         std::set<artdaq::Fragment::sequence_id_t> next_requests_;
     111              :         std::set<artdaq::Fragment::sequence_id_t> recently_completed_requests_;
     112              :         size_t recently_completed_request_history_size_ = 1000;
     113              :         mutable std::mutex request_mutex_;
     114              :         std::condition_variable request_cv_;
     115              : 
     116              :         std::atomic<bool> receiver_running_;
     117              : };
     118              : }  // namespace artdaq
     119              : 
     120              : #endif  // _artdaq_DAQrate_RequestBuffer_hh_