Line data Source code
1 : #ifndef artdaq_DAQrate_RequestSender_hh
2 : #define artdaq_DAQrate_RequestSender_hh
3 :
4 : #include "artdaq/DAQrate/detail/RequestMessage.hh"
5 :
6 : #include "fhiclcpp/types/Atom.h"
7 : #include "fhiclcpp/types/Comment.h"
8 : #include "fhiclcpp/types/ConfigurationTable.h"
9 : #include "fhiclcpp/types/Name.h"
10 : #include "fhiclcpp/types/Table.h"
11 :
12 : #include <arpa/inet.h>
13 : #include <netinet/in.h>
14 : #include <sys/socket.h>
15 : #include <sys/types.h>
16 : #include <chrono>
17 : #include <cstdint>
18 : #include <map>
19 : #include <memory>
20 :
21 : namespace artdaq {
22 :
23 : /**
24 : * \brief The RequestSender contains methods used to send data requests and Routing tokens
25 : */
26 : class RequestSender
27 : {
28 : public:
29 : /// <summary>
30 : /// Configuration of the RequestSender. May be used for parameter validation
31 : /// </summary>
32 : struct Config
33 : {
34 : /// "send_requests" (Default: false): Whether to send DataRequests when new sequence IDs are seen
35 : fhicl::Atom<bool> send_requests{fhicl::Name{"send_requests"}, fhicl::Comment{"Enable sending Data Request messages"}, false};
36 : /// "request_port" (Default: 3001): Port to send DataRequests on
37 : fhicl::Atom<int> request_port{fhicl::Name{"request_port"}, fhicl::Comment{"Port to send DataRequests on"}, 3001};
38 : /// "request_interval_ms" (Default: 100): How often to send DataRequests
39 : fhicl::Atom<size_t> request_interval_ms{fhicl::Name{"request_interval_ms"}, fhicl::Comment{"How often to send DataRequests"}, 100};
40 : /// "request_delay_ms" (Default: 0): How long to wait before sending a new DataRequest
41 : fhicl::Atom<size_t> request_delay_ms{fhicl::Name{"request_delay_ms"}, fhicl::Comment{"How long to wait before sending a new DataRequest"}, 0};
42 : /// "request_shutdown_timeout_us" (Default: 100000 us): How long to wait for pending requests to be sent at shutdown
43 : fhicl::Atom<size_t> request_shutdown_timeout_us{fhicl::Name{"request_shutdown_timeout_us"}, fhicl::Comment{"How long to wait for pending requests to be sent at shutdown"}, 100000};
44 : /// "multicast_interface_ip" (Default: "0.0.0.0"): Use this hostname for multicast output (to assign to the proper NIC)
45 : fhicl::Atom<std::string> output_address{fhicl::Name{"multicast_interface_ip"}, fhicl::Comment{"Use this hostname for multicast output(to assign to the proper NIC)"}, "0.0.0.0"};
46 : /// "request_address" (Default: "227.128.12.26"): Multicast address to send DataRequests to
47 : fhicl::Atom<std::string> request_address{fhicl::Name{"request_address"}, fhicl::Comment{"Multicast address to send DataRequests to"}, "227.128.12.26"};
48 : };
49 : /// Used for ParameterSet validation (if desired)
50 : using Parameters = fhicl::WrappedTable<Config>;
51 :
52 : /**
53 : * \brief Default Constructor is deleted
54 : */
55 : RequestSender() = delete;
56 :
57 : /**
58 : * \brief Copy Constructor is deleted
59 : */
60 : RequestSender(RequestSender const&) = delete;
61 :
62 : /**
63 : * \brief Copy Assignment operator is deleted
64 : * \return RequestSender copy
65 : */
66 : RequestSender& operator=(RequestSender const&) = delete;
67 :
68 : RequestSender(RequestSender&&) = delete; ///< Move Constructor is deleted
69 : RequestSender& operator=(RequestSender&&) = delete; ///< Move-assignment operator is deleted
70 :
71 : /**
72 : * \brief RequestSender Constructor
73 : * \param pset ParameterSet used to configured RequestSender. See artdaq::RequestSender::Config
74 : */
75 : explicit RequestSender(const fhicl::ParameterSet& pset);
76 : /**
77 : * \brief RequestSender Destructor
78 : */
79 : virtual ~RequestSender();
80 :
81 : /**
82 : * \brief Set the mode for RequestMessages. Used to indicate when RequestSender should enter "EndOfRun" mode
83 : * \param mode Mode to set
84 : */
85 : void SetRequestMode(detail::RequestMessageMode const& mode);
86 :
87 : /**
88 : * \brief Get the mode for RequestMessages.
89 : * \return Current RequestMessageMode of the RequestSender
90 : */
91 1 : detail::RequestMessageMode GetRequestMode() const { return request_mode_; }
92 :
93 : /**
94 : * \brief Add a request to the request list
95 : * \param seqID Sequence ID for request
96 : * \param timestamp Timestamp to request
97 : */
98 : void AddRequest(Fragment::sequence_id_t const& seqID, Fragment::timestamp_t const& timestamp);
99 :
100 : /**
101 : * \brief Remove a request from the request list
102 : * \param seqID Sequence ID of request
103 : */
104 : void RemoveRequest(Fragment::sequence_id_t const& seqID);
105 :
106 : /**
107 : * \brief Set the run number to be used in request messages
108 : * \param run Run number
109 : */
110 17 : void SetRunNumber(uint32_t const& run) { run_number_ = run; }
111 :
112 : /**
113 : * @brief Get the number of requests sent by this RequestSender
114 : * @return The number of requests sent
115 : */
116 0 : size_t GetSentMessageCount() { return requests_sent_.load(); }
117 :
118 : private:
119 : private:
120 : // State variables
121 : std::atomic<bool> initialized_{false};
122 : std::atomic<bool> sender_thread_running_{false};
123 : std::atomic<bool> sender_thread_exited_{true};
124 : mutable std::mutex request_mutex_;
125 : std::condition_variable sender_cv_;
126 : std::unique_ptr<boost::thread> sender_thread_{nullptr};
127 : std::atomic<size_t> requests_sent_{0};
128 : uint32_t run_number_{0};
129 :
130 : // Main data storage
131 : std::map<Fragment::sequence_id_t, Fragment::timestamp_t> active_requests_{};
132 : detail::RequestMessageMode request_mode_{detail::RequestMessageMode::Normal};
133 :
134 : // Socket variables
135 : int request_socket_{-1};
136 : struct sockaddr_in request_addr_;
137 :
138 : // Configuration parameters
139 : bool send_requests_;
140 : std::string request_address_;
141 : int request_port_;
142 : std::chrono::microseconds request_shutdown_timeout_;
143 : std::string multicast_out_addr_;
144 : std::chrono::milliseconds request_interval_;
145 : std::chrono::milliseconds request_delay_;
146 :
147 : private:
148 : void setup_requests_();
149 :
150 : void start_send_request_thread_();
151 : void stop_send_request_thread_();
152 :
153 : void do_add_request_(Fragment::sequence_id_t const& seqID, Fragment::timestamp_t const& timestamp);
154 : void do_send_request_();
155 : };
156 : } // namespace artdaq
157 : #endif /* artdaq_DAQrate_RequestSender_hh */
|
