Line data Source code
1 : /**
2 : *
3 : * Implementations of DaphneV2Interface's functions
4 : *
5 : * This is part of the DUNE DAQ Software Suite, copyright 2020.
6 : * Licensing/copyright details are in the COPYING file that you should have
7 : */
8 :
9 : #include "DaphneV2Interface.hpp"
10 : #include "logging/Logging.hpp"
11 : #include <sys/time.h>
12 :
13 : using namespace dunedaq::daphnemodules;
14 :
15 1 : DaphneV2Interface::DaphneV2Interface(const char* ipaddr, int port, std::chrono::milliseconds timeout)
16 1 : : m_timeout(timeout)
17 : {
18 :
19 1 : m_connection_id = socket(AF_INET, SOCK_DGRAM, 0);
20 :
21 1 : if (m_connection_id < 0)
22 0 : throw SocketCreationError(ERS_HERE);
23 :
24 1 : m_target.sin_family = AF_INET;
25 1 : m_target.sin_port = htons(port);
26 1 : auto ret = inet_pton(AF_INET, ipaddr, &(m_target.sin_addr));
27 1 : if (ret <= 0)
28 1 : throw InvalidIPAddress(ERS_HERE, ipaddr);
29 :
30 0 : m_ip = ipaddr;
31 :
32 0 : if (!validate_connection())
33 0 : throw FailedPing(ERS_HERE, ipaddr, port);
34 1 : }
35 :
36 : void
37 0 : DaphneV2Interface::close()
38 : {
39 :
40 0 : ::close(m_connection_id);
41 0 : }
42 :
43 : bool
44 0 : DaphneV2Interface::validate_connection() const
45 : {
46 :
47 0 : auto ret = read_register(0xaa55, 1);
48 :
49 0 : static const uint64_t good_value = 0xdeadbeef;
50 0 : return ret[0] == good_value;
51 0 : }
52 :
53 : command_result
54 0 : DaphneV2Interface::send_command_retry(std::string cmd, size_t retry) const
55 : {
56 :
57 0 : do {
58 0 : try {
59 0 : auto ret = send_command(cmd);
60 0 : return ret;
61 0 : } catch (const CommandTimeout& e) {
62 0 : ers::warning(e);
63 0 : --retry;
64 0 : } catch (const ers::Issue& e) {
65 0 : throw FailedSocketInteraction(ERS_HERE, cmd, e);
66 0 : }
67 :
68 0 : } while (retry > 0);
69 :
70 0 : throw FailedSocketInteraction(ERS_HERE, cmd);
71 : }
72 :
73 : command_result
74 0 : DaphneV2Interface::send_command_interruptible(std::string cmd, std::function<bool()> can_retry) const
75 : {
76 :
77 0 : do {
78 0 : try {
79 0 : auto ret = send_command(cmd);
80 0 : return ret;
81 0 : } catch (const CommandTimeout& e) {
82 0 : ers::warning(e);
83 0 : } catch (const ers::Issue& e) {
84 0 : throw FailedSocketInteraction(ERS_HERE, cmd, e);
85 0 : }
86 :
87 0 : } while (can_retry());
88 : }
89 :
90 : command_result
91 0 : DaphneV2Interface::send_command(std::string cmd) const
92 : {
93 :
94 0 : TLOG() << "Board: " << m_ip << ", sending command " << cmd;
95 0 : std::vector<uint64_t> bytes;
96 0 : for (char ch : cmd) {
97 0 : bytes.push_back(static_cast<uint64_t>(ch));
98 : }
99 0 : bytes.push_back(0x0d); // dedicated command flag
100 :
101 0 : const std::lock_guard<std::mutex> lock(m_command_mutex);
102 :
103 : // we send the bytes in chunks of 50 words
104 0 : for (size_t i = 0; i < (bytes.size() + 49) / 50; ++i) {
105 0 : std::vector<uint64_t> part(bytes.begin() + i * 50, bytes.begin() + std::min((i + 1) * 50, bytes.size()));
106 0 : write_buffer(0x90000000, std::move(part));
107 0 : }
108 :
109 0 : TLOG() << "Board: " << m_ip << ", Command sent, waiting for result";
110 :
111 0 : command_result res;
112 0 : std::string* writing_pointer = nullptr;
113 :
114 0 : auto start_time = std::chrono::high_resolution_clock::now();
115 :
116 0 : int more = 40;
117 0 : while (more > 0) {
118 0 : auto data_block = read_buffer(0x90000000, 50);
119 0 : for (size_t i = 0; i < data_block.size(); ++i) {
120 0 : if (data_block[i] == 255) {
121 : break;
122 0 : } else if (data_block[i] == 1) {
123 : // the following data are returning the command that was issued
124 0 : writing_pointer = &res.command;
125 0 : } else if (data_block[i] == 2) {
126 : // the following data are the immediate command response
127 0 : writing_pointer = &res.result;
128 0 : } else if (data_block[i] == 3) {
129 : // this is the message end
130 : writing_pointer = nullptr;
131 0 : } else if (isprint(static_cast<int>(data_block[i]))) {
132 0 : more = 40;
133 0 : char c = static_cast<char>(data_block[i]);
134 0 : if (writing_pointer) {
135 0 : *writing_pointer += static_cast<char>(data_block[i]);
136 : } else {
137 0 : TLOG() << "Failed adding charachter " << c;
138 : }
139 : }
140 : }
141 0 : auto now = std::chrono::high_resolution_clock::now();
142 :
143 0 : auto delay = now - start_time;
144 :
145 0 : if (delay > m_timeout) {
146 0 : TLOG() << "Details of timeout";
147 0 : for (size_t i = 0; i < data_block.size(); ++i) {
148 0 : TLOG() << i << "\t" << std::hex << data_block[i] << std::dec;
149 : }
150 0 : TLOG() << "Received so far: " << res.result;
151 0 : auto delay_us = std::chrono::duration_cast<std::chrono::microseconds>(delay);
152 0 : throw CommandTimeout(ERS_HERE, cmd, delay_us.count());
153 : }
154 :
155 0 : std::this_thread::sleep_for(std::chrono::milliseconds(1));
156 0 : --more;
157 0 : }
158 :
159 0 : return res;
160 0 : }
161 :
162 : std::vector<uint64_t>
163 0 : DaphneV2Interface::read(uint8_t command_id, uint64_t addr, uint8_t size) const
164 : {
165 :
166 0 : const std::lock_guard<std::mutex> lock(m_access_mutex);
167 :
168 0 : uint8_t cmd[10];
169 0 : cmd[0] = command_id;
170 0 : cmd[1] = size;
171 0 : memcpy(cmd + 2, &addr, sizeof(uint64_t));
172 0 : auto result = sendto(m_connection_id, cmd, sizeof(cmd), 0, (struct sockaddr*)&m_target, sizeof(m_target));
173 :
174 0 : if (result < 0)
175 0 : throw FailedSocketInteraction(ERS_HERE, "sendto");
176 :
177 0 : struct timeval timeout;
178 0 : timeout.tv_sec = m_timeout.count() / 1000;
179 0 : timeout.tv_usec = (m_timeout.count() % 1000) * 1000;
180 0 : fd_set readfds, masterfds;
181 :
182 0 : FD_ZERO(&masterfds);
183 0 : FD_SET(m_connection_id, &masterfds);
184 :
185 0 : memcpy(&readfds, &masterfds, sizeof(fd_set));
186 :
187 0 : auto start_time = std::chrono::high_resolution_clock::now();
188 :
189 0 : if (select(m_connection_id + 1, &readfds, NULL, NULL, &timeout) < 0) {
190 0 : throw FailedSocketInteraction(ERS_HERE, "select");
191 : }
192 :
193 0 : std::vector<uint64_t> ret_value;
194 :
195 0 : if (FD_ISSET(m_connection_id, &readfds)) {
196 0 : uint8_t buffer[2 + (8 * size)];
197 0 : socklen_t addrlen = sizeof(m_target);
198 0 : result = recvfrom(m_connection_id, buffer, sizeof(buffer), 0, (struct sockaddr*)&m_target, &addrlen);
199 :
200 0 : if (result <= 0)
201 0 : throw FailedSocketInteraction(ERS_HERE, "recvfrom");
202 :
203 0 : uint8_t fmt[4 + size];
204 0 : fmt[0] = '<';
205 0 : fmt[1] = 'B';
206 0 : fmt[2] = 'B';
207 0 : fmt[3] = size;
208 0 : for (int i = 0; i < size; i++) {
209 0 : fmt[4 + i] = 'Q';
210 : }
211 :
212 0 : for (int i = 0; i < size; i++) {
213 0 : uint64_t value;
214 0 : memcpy(&value, buffer + 2 + (8 * i), sizeof(uint64_t));
215 0 : ret_value.push_back(value);
216 : }
217 :
218 0 : } else {
219 : // the socket timedout
220 0 : auto end_time = std::chrono::high_resolution_clock::now();
221 0 : auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time);
222 :
223 0 : throw SocketTimeout(ERS_HERE, duration.count());
224 : }
225 :
226 0 : return ret_value;
227 0 : }
228 :
229 : void
230 0 : DaphneV2Interface::write(uint8_t command_id, uint64_t addr, std::vector<uint64_t>&& data) const
231 : {
232 :
233 0 : const std::lock_guard<std::mutex> lock(m_access_mutex);
234 :
235 0 : uint8_t cmd[10 + (8 * data.size())];
236 0 : cmd[0] = command_id;
237 0 : cmd[1] = data.size();
238 0 : memcpy(cmd + 2, &addr, sizeof(uint64_t));
239 0 : for (size_t i = 0; i < data.size(); i++) {
240 0 : memcpy(cmd + 10 + (8 * i), &(data[i]), sizeof(uint64_t));
241 : }
242 :
243 0 : auto result = sendto(m_connection_id, cmd, sizeof(cmd), 0, (struct sockaddr*)&m_target, sizeof(m_target));
244 0 : if (result < 0)
245 0 : throw FailedSocketInteraction(ERS_HERE, "sendto");
246 0 : }
|
