asix-omega-rtm-cli: Implement RTM for ASIX OMEGA via external CLI process
[libsigrok.git] / src / hardware / asix-omega-rtm-cli / protocol.c
1 /*
2  * This file is part of the libsigrok project.
3  *
4  * Copyright (C) 2021 Gerhard Sittig <gerhard.sittig@gmx.net>
5  *
6  * This program is free software: you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License as published by
8  * the Free Software Foundation, either version 3 of the License, or
9  * (at your option) any later version.
10  *
11  * This program is distributed in the hope that it will be useful,
12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14  * GNU General Public License for more details.
15  *
16  * You should have received a copy of the GNU General Public License
17  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
18  */
19
20 #include <config.h>
21
22 #include <string.h>
23 #include <unistd.h>
24
25 #include "protocol.h"
26
27 /*
28  * Start the external acquisition process (vendor's CLI application).
29  * Get the initial response to verify its operation.
30  */
31 SR_PRIV int omega_rtm_cli_open(const struct sr_dev_inst *sdi)
32 {
33         struct dev_context *devc;
34         gboolean ok;
35         gchar **argv;
36         GSpawnFlags flags;
37         GPid pid;
38         gint fd_in, fd_out;
39         GError *error;
40         GString *txt;
41         ssize_t rcvd;
42         uint8_t rsp[3 * sizeof(uint16_t)];
43         const uint8_t *rdptr;
44         uint16_t stamp, sample1, sample2;
45
46         if (!sdi)
47                 return SR_ERR_ARG;
48         devc = sdi->priv;
49         if (!devc)
50                 return SR_ERR_ARG;
51
52         if (devc->child.running) {
53                 sr_err("Vendor application already running.");
54                 return SR_ERR_BUG;
55         }
56
57         /* Prepare to feed sample data to the session. */
58         memset(&devc->rawdata, 0, sizeof(devc->rawdata));
59         memset(&devc->samples, 0, sizeof(devc->samples));
60         devc->samples.queue = feed_queue_logic_alloc(sdi,
61                 FEED_QUEUE_DEPTH, sizeof(devc->samples.last_sample));
62
63         /*
64          * Start the background process. May take considerable time
65          * before actual acquisition starts.
66          */
67         sr_dbg("Starting vendor application");
68         argv = devc->child.argv;
69         flags = devc->child.flags;
70         error = NULL;
71         ok = g_spawn_async_with_pipes(NULL, argv, NULL, flags, NULL, NULL,
72                 &pid, &fd_in, &fd_out, NULL, &error);
73         if (error) {
74                 sr_err("Cannot execute RTM CLI process: %s", error->message);
75                 g_error_free(error);
76                 ok = FALSE;
77         }
78         if (fd_in < 0 || fd_out < 0)
79                 ok = FALSE;
80         if (!ok) {
81                 sr_err("Vendor application start failed.");
82                 return SR_ERR_IO;
83         }
84         devc->child.pid = pid;
85         devc->child.fd_stdin_write = fd_in;
86         devc->child.fd_stdout_read = fd_out;
87         devc->child.running = TRUE;
88         sr_dbg("Started vendor application, in %d, out %d", fd_in, fd_out);
89         txt = sr_hexdump_new((const uint8_t *)&pid, sizeof(pid));
90         sr_dbg("Vendor application PID (OS dependent): %s", txt->str);
91         sr_hexdump_free(txt);
92
93         /*
94          * Get the initial response. Verifies its operation, and only
95          * returns with success when acquisition became operational.
96          */
97         rcvd = read(fd_out, rsp, sizeof(rsp));
98         sr_dbg("Read from vendor application, ret %zd", rcvd);
99         if (rcvd < 0)
100                 ok = FALSE;
101         if (ok && (size_t)rcvd != sizeof(rsp))
102                 ok = FALSE;
103         if (!ok) {
104                 omega_rtm_cli_close(sdi);
105                 return SR_ERR_IO;
106         }
107
108         /*
109          * Ignore the first timestamp. Grab the most recent sample data
110          * to feed the session from it upon later repetition.
111          */
112         rdptr = rsp;
113         stamp = read_u16le_inc(&rdptr);
114         sample1 = read_u16le_inc(&rdptr);
115         sample2 = read_u16le_inc(&rdptr);
116         sr_dbg("stamp %u, samples %x %x", stamp, sample1, sample2);
117         write_u16le(devc->samples.last_sample, sample2);
118
119         return SR_OK;
120 }
121
122 /*
123  * Terminate the external acquisition process (vendor's CLI application).
124  */
125 SR_PRIV int omega_rtm_cli_close(const struct sr_dev_inst *sdi)
126 {
127         struct dev_context *devc;
128
129         if (!sdi)
130                 return SR_ERR_ARG;
131         devc = sdi->priv;
132         if (!devc)
133                 return SR_ERR_ARG;
134
135         /* Close the external process' stdin. Discard its stdout. */
136         sr_dbg("Closing vendor application file descriptors.");
137         if (devc->child.fd_stdin_write >= 0) {
138                 sr_dbg("Closing vendor application stdin descriptor.");
139                 close(devc->child.fd_stdin_write);
140                 devc->child.fd_stdin_write = -1;
141         }
142         if (devc->child.fd_stdout_read >= 0) {
143                 sr_dbg("Closing vendor application stdout descriptor.");
144                 close(devc->child.fd_stdout_read);
145                 devc->child.fd_stdout_read = -1;
146         }
147
148         /* Terminate the external process. */
149         if (devc->child.running) {
150                 sr_dbg("Closing vendor application process.");
151                 (void)g_spawn_close_pid(devc->child.pid);
152                 memset(&devc->child.pid, 0, sizeof(devc->child.pid));
153                 devc->child.running = FALSE;
154         }
155
156         /* Release the session feed queue. */
157         if (devc->samples.queue) {
158                 feed_queue_logic_free(devc->samples.queue);
159                 devc->samples.queue = NULL;
160         }
161
162         sr_dbg("Done closing vendor application.");
163
164         return SR_OK;
165 }
166
167 /*
168  * Process received sample data, which comes in 6-byte chunks.
169  * Uncompress the RLE stream. Strictly enforce user specified sample
170  * count limits in the process, cap the submission when an uncompressed
171  * chunk would exceed the limit.
172  */
173 static int omega_rtm_cli_process_rawdata(const struct sr_dev_inst *sdi)
174 {
175         static const size_t chunk_size = 3 * sizeof(uint16_t);
176
177         struct dev_context *devc;
178         const uint8_t *rdptr;
179         size_t avail, taken, count;
180         uint16_t stamp, sample1, sample2;
181         int ret;
182
183         devc = sdi->priv;
184         rdptr = &devc->rawdata.buff[0];
185         avail = devc->rawdata.fill;
186         taken = 0;
187         ret = SR_OK;
188
189         /* Cope with previous errors, silently discard RX data. */
190         if (!devc->samples.queue)
191                 ret = SR_ERR_DATA;
192
193         /* Process those chunks whose reception has completed. */
194         while (ret == SR_OK && avail >= chunk_size) {
195                 stamp = read_u16le_inc(&rdptr);
196                 sample1 = read_u16le_inc(&rdptr);
197                 sample2 = read_u16le_inc(&rdptr);
198                 avail -= chunk_size;
199                 taken += chunk_size;
200
201                 /*
202                  * Uncompress the RLE stream by repeating the last
203                  * sample value when necessary. Notice that the stamp
204                  * has a resolution of 10ns and thus covers two times
205                  * the number of samples, these are taken each 5ns (at
206                  * 200MHz rate). A stamp value of 1 is immediately
207                  * adjacent to the last chunk.
208                  */
209                 if (stamp)
210                         stamp--;
211                 count = stamp * 2;
212                 if (devc->samples.check_count) {
213                         if (count > devc->samples.remain_count)
214                                 count = devc->samples.remain_count;
215                         devc->samples.remain_count -= count;
216                 }
217                 if (count) {
218                         ret = feed_queue_logic_submit(devc->samples.queue,
219                                 devc->samples.last_sample, count);
220                         if (ret != SR_OK)
221                                 break;
222                         sr_sw_limits_update_samples_read(&devc->limits, count);
223                 }
224                 if (devc->samples.check_count && !devc->samples.remain_count)
225                         break;
226
227                 /*
228                  * Also send the current samples. Keep the last value at
229                  * hand because future chunks might repeat it.
230                  */
231                 write_u16le(devc->samples.last_sample, sample1);
232                 ret = feed_queue_logic_submit(devc->samples.queue,
233                         devc->samples.last_sample, 1);
234                 if (ret != SR_OK)
235                         break;
236
237                 write_u16le(devc->samples.last_sample, sample2);
238                 ret = feed_queue_logic_submit(devc->samples.queue,
239                         devc->samples.last_sample, 1);
240                 if (ret != SR_OK)
241                         break;
242
243                 count = 2;
244                 sr_sw_limits_update_samples_read(&devc->limits, count);
245                 if (devc->samples.check_count) {
246                         if (count > devc->samples.remain_count)
247                                 count = devc->samples.remain_count;
248                         devc->samples.remain_count -= count;
249                         if (!devc->samples.remain_count)
250                                 break;
251                 }
252         }
253
254         /*
255          * Silently consume all chunks which were successfully received.
256          * These either completely got processed, or we are in an error
257          * path and discard unprocessed but complete sample data before
258          * propagating the error condition. This simplifies the logic
259          * above, and allows to keep draining the acquisition process'
260          * output, perhaps even resynchronize to it in a later attempt.
261          * The cost of this rare operation does not matter, robustness
262          * does.
263          */
264         while (avail >= chunk_size) {
265                 avail -= chunk_size;
266                 taken += chunk_size;
267         }
268
269         /*
270          * Shift remainders (incomplete chunks) down to the start of the
271          * receive buffer.
272          */
273         if (taken && avail) {
274                 memmove(&devc->rawdata.buff[0],
275                         &devc->rawdata.buff[taken], avail);
276         }
277         devc->rawdata.fill -= taken;
278
279         return ret;
280 }
281
282 SR_PRIV int omega_rtm_cli_receive_data(int fd, int revents, void *cb_data)
283 {
284         const struct sr_dev_inst *sdi;
285         struct dev_context *devc;
286         uint8_t *buff;
287         size_t space;
288         ssize_t rcvd;
289         int ret;
290
291         sdi = cb_data;
292         if (!sdi)
293                 return TRUE;
294         devc = sdi->priv;
295         if (!devc)
296                 return TRUE;
297
298         /* Process receive data when available. */
299         if (revents & G_IO_IN) do {
300                 buff = &devc->rawdata.buff[devc->rawdata.fill];
301                 space = sizeof(devc->rawdata.buff) - devc->rawdata.fill;
302                 rcvd = read(fd, buff, space);
303                 sr_spew("Read from vendor application, ret %zd", rcvd);
304                 if (rcvd <= 0)
305                         break;
306                 devc->rawdata.fill += (size_t)rcvd;
307                 ret = omega_rtm_cli_process_rawdata(sdi);
308                 if (ret != SR_OK) {
309                         sr_err("Could not process sample data.");
310                 }
311         } while (0);
312
313         /* Handle receive errors. */
314         if (revents & G_IO_ERR) {
315                 (void)feed_queue_logic_flush(devc->samples.queue);
316                 (void)sr_dev_acquisition_stop((struct sr_dev_inst *)sdi);
317         }
318
319         /* Handle optional acquisition limits. */
320         if (sr_sw_limits_check(&devc->limits)) {
321                 (void)feed_queue_logic_flush(devc->samples.queue);
322                 (void)sr_dev_acquisition_stop((struct sr_dev_inst *)sdi);
323         }
324
325         return TRUE;
326 }