2ec6ef2db39aebac9451f3b560884bfb0bde5f66
[lttng-tools.git] / src / common / relayd / relayd.c
1 /*
2 * Copyright (C) 2012 - David Goulet <dgoulet@efficios.com>
3 *
4 * This program is free software; you can redistribute it and/or modify it
5 * under the terms of the GNU General Public License, version 2 only, as
6 * published by the Free Software Foundation.
7 *
8 * This program is distributed in the hope that it will be useful, but WITHOUT
9 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
11 * more details.
12 *
13 * You should have received a copy of the GNU General Public License along with
14 * this program; if not, write to the Free Software Foundation, Inc., 51
15 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
16 */
17
18 #define _GNU_SOURCE
19 #include <assert.h>
20 #include <stdio.h>
21 #include <stdlib.h>
22 #include <string.h>
23 #include <sys/stat.h>
24 #include <inttypes.h>
25
26 #include <common/common.h>
27 #include <common/defaults.h>
28 #include <common/sessiond-comm/relayd.h>
29
30 #include "relayd.h"
31
32 /*
33 * Send command. Fill up the header and append the data.
34 */
35 static int send_command(struct lttcomm_sock *sock,
36 enum lttcomm_relayd_command cmd, void *data, size_t size,
37 int flags)
38 {
39 int ret;
40 struct lttcomm_relayd_hdr header;
41 char *buf;
42 uint64_t buf_size = sizeof(header);
43
44 if (data) {
45 buf_size += size;
46 }
47
48 buf = zmalloc(buf_size);
49 if (buf == NULL) {
50 PERROR("zmalloc relayd send command buf");
51 ret = -1;
52 goto alloc_error;
53 }
54
55 header.cmd = htobe32(cmd);
56 header.data_size = htobe64(size);
57
58 /* Zeroed for now since not used. */
59 header.cmd_version = 0;
60 header.circuit_id = 0;
61
62 /* Prepare buffer to send. */
63 memcpy(buf, &header, sizeof(header));
64 if (data) {
65 memcpy(buf + sizeof(header), data, size);
66 }
67
68 ret = sock->ops->sendmsg(sock, buf, buf_size, flags);
69 if (ret < 0) {
70 ret = -errno;
71 goto error;
72 }
73
74 DBG3("Relayd sending command %d of size %" PRIu64, cmd, buf_size);
75
76 error:
77 free(buf);
78 alloc_error:
79 return ret;
80 }
81
82 /*
83 * Receive reply data on socket. This MUST be call after send_command or else
84 * could result in unexpected behavior(s).
85 */
86 static int recv_reply(struct lttcomm_sock *sock, void *data, size_t size)
87 {
88 int ret;
89
90 DBG3("Relayd waiting for reply of size %ld", size);
91
92 ret = sock->ops->recvmsg(sock, data, size, 0);
93 if (ret < 0) {
94 ret = -errno;
95 goto error;
96 }
97
98 error:
99 return ret;
100 }
101
102 /*
103 * Send a RELAYD_CREATE_SESSION command to the relayd with the given socket and
104 * set session_id of the relayd if we have a successful reply from the relayd.
105 *
106 * On success, return 0 else a negative value being a lttng_error_code returned
107 * from the relayd.
108 */
109 int relayd_create_session(struct lttcomm_sock *sock, uint64_t *session_id)
110 {
111 int ret;
112 struct lttcomm_relayd_status_session reply;
113
114 assert(sock);
115 assert(session_id);
116
117 DBG("Relayd create session");
118
119 /* Send command */
120 ret = send_command(sock, RELAYD_CREATE_SESSION, NULL, 0, 0);
121 if (ret < 0) {
122 goto error;
123 }
124
125 /* Recevie response */
126 ret = recv_reply(sock, (void *) &reply, sizeof(reply));
127 if (ret < 0) {
128 goto error;
129 }
130
131 reply.session_id = be64toh(reply.session_id);
132 reply.ret_code = be32toh(reply.ret_code);
133
134 /* Return session id or negative ret code. */
135 if (reply.ret_code != LTTNG_OK) {
136 ret = -reply.ret_code;
137 ERR("Relayd create session replied error %d", ret);
138 goto error;
139 } else {
140 ret = 0;
141 *session_id = reply.session_id;
142 }
143
144 DBG("Relayd session created with id %" PRIu64, reply.session_id);
145
146 error:
147 return ret;
148 }
149
150 /*
151 * Add stream on the relayd and assign stream handle to the stream_id argument.
152 *
153 * On success return 0 else return ret_code negative value.
154 */
155 int relayd_add_stream(struct lttcomm_sock *sock, const char *channel_name,
156 const char *pathname, uint64_t *stream_id)
157 {
158 int ret;
159 struct lttcomm_relayd_add_stream msg;
160 struct lttcomm_relayd_status_stream reply;
161
162 /* Code flow error. Safety net. */
163 assert(sock);
164 assert(channel_name);
165 assert(pathname);
166
167 DBG("Relayd adding stream for channel name %s", channel_name);
168
169 strncpy(msg.channel_name, channel_name, sizeof(msg.channel_name));
170 strncpy(msg.pathname, pathname, sizeof(msg.pathname));
171
172 /* Send command */
173 ret = send_command(sock, RELAYD_ADD_STREAM, (void *) &msg, sizeof(msg), 0);
174 if (ret < 0) {
175 goto error;
176 }
177
178 /* Waiting for reply */
179 ret = recv_reply(sock, (void *) &reply, sizeof(reply));
180 if (ret < 0) {
181 goto error;
182 }
183
184 /* Back to host bytes order. */
185 reply.handle = be64toh(reply.handle);
186 reply.ret_code = be32toh(reply.ret_code);
187
188 /* Return session id or negative ret code. */
189 if (reply.ret_code != LTTNG_OK) {
190 ret = -reply.ret_code;
191 ERR("Relayd add stream replied error %d", ret);
192 } else {
193 /* Success */
194 ret = 0;
195 *stream_id = reply.handle;
196 }
197
198 DBG("Relayd stream added successfully with handle %" PRIu64,
199 reply.handle);
200
201 error:
202 return ret;
203 }
204
205 /*
206 * Check version numbers on the relayd.
207 *
208 * Return 0 if compatible else negative value.
209 */
210 int relayd_version_check(struct lttcomm_sock *sock, uint32_t major,
211 uint32_t minor)
212 {
213 int ret;
214 struct lttcomm_relayd_version msg;
215
216 /* Code flow error. Safety net. */
217 assert(sock);
218
219 DBG("Relayd version check for major.minor %u.%u", major, minor);
220
221 /* Prepare network byte order before transmission. */
222 msg.major = htobe32(major);
223 msg.minor = htobe32(minor);
224
225 /* Send command */
226 ret = send_command(sock, RELAYD_VERSION, (void *) &msg, sizeof(msg), 0);
227 if (ret < 0) {
228 goto error;
229 }
230
231 /* Recevie response */
232 ret = recv_reply(sock, (void *) &msg, sizeof(msg));
233 if (ret < 0) {
234 goto error;
235 }
236
237 /* Set back to host bytes order */
238 msg.major = be32toh(msg.major);
239 msg.minor = be32toh(msg.minor);
240
241 /*
242 * Only validate the major version. If the other side is higher,
243 * communication is not possible. Only major version equal can talk to each
244 * other. If the minor version differs, the lowest version is used by both
245 * sides.
246 *
247 * For now, before 2.1.0 stable release, we don't have to check the minor
248 * because this new mechanism with the relayd will only be available with
249 * 2.1 and NOT 2.0.x.
250 */
251 if (msg.major == major) {
252 /* Compatible */
253 ret = 0;
254 DBG2("Relayd version is compatible");
255 goto error;
256 }
257
258 /*
259 * After 2.1.0 release, for the 2.2 release, at this point will have to
260 * check the minor version in order for the session daemon to know which
261 * structure to use to communicate with the relayd. If the relayd's minor
262 * version is higher, it will adapt to our version so we can continue to
263 * use the latest relayd communication data structure.
264 */
265
266 /* Version number not compatible */
267 DBG2("Relayd version is NOT compatible. Relayd version %u != %u (us)",
268 msg.major, major);
269 ret = -1;
270
271 error:
272 return ret;
273 }
274
275 /*
276 * Add stream on the relayd and assign stream handle to the stream_id argument.
277 *
278 * On success return 0 else return ret_code negative value.
279 */
280 int relayd_send_metadata(struct lttcomm_sock *sock, size_t len)
281 {
282 int ret;
283
284 /* Code flow error. Safety net. */
285 assert(sock);
286
287 DBG("Relayd sending metadata of size %zu", len);
288
289 /* Send command */
290 ret = send_command(sock, RELAYD_SEND_METADATA, NULL, len, 0);
291 if (ret < 0) {
292 goto error;
293 }
294
295 DBG2("Relayd metadata added successfully");
296
297 /*
298 * After that call, the metadata data MUST be sent to the relayd so the
299 * receive size on the other end matches the len of the metadata packet
300 * header. This is why we don't wait for a reply here.
301 */
302
303 error:
304 return ret;
305 }
306
307 /*
308 * Connect to relay daemon with an allocated lttcomm_sock.
309 */
310 int relayd_connect(struct lttcomm_sock *sock)
311 {
312 /* Code flow error. Safety net. */
313 assert(sock);
314
315 DBG3("Relayd connect ...");
316
317 return sock->ops->connect(sock);
318 }
319
320 /*
321 * Close relayd socket with an allocated lttcomm_sock.
322 */
323 int relayd_close(struct lttcomm_sock *sock)
324 {
325 /* Code flow error. Safety net. */
326 assert(sock);
327
328 DBG3("Relayd closing socket %d", sock->fd);
329
330 return sock->ops->close(sock);
331 }
332
333 /*
334 * Send data header structure to the relayd.
335 */
336 int relayd_send_data_hdr(struct lttcomm_sock *sock,
337 struct lttcomm_relayd_data_hdr *hdr, size_t size)
338 {
339 int ret;
340
341 /* Code flow error. Safety net. */
342 assert(sock);
343 assert(hdr);
344
345 DBG3("Relayd sending data header of size %ld", size);
346
347 /* Again, safety net */
348 if (size == 0) {
349 size = sizeof(struct lttcomm_relayd_data_hdr);
350 }
351
352 /* Only send data header. */
353 ret = sock->ops->sendmsg(sock, hdr, size, 0);
354 if (ret < 0) {
355 ret = -errno;
356 goto error;
357 }
358
359 /*
360 * The data MUST be sent right after that command for the receive on the
361 * other end to match the size in the header.
362 */
363
364 error:
365 return ret;
366 }
367
368 /*
369 * Send close stream command to the relayd.
370 */
371 int relayd_send_close_stream(struct lttcomm_sock *sock, uint64_t stream_id,
372 uint64_t last_net_seq_num)
373 {
374 int ret;
375 struct lttcomm_relayd_close_stream msg;
376 struct lttcomm_relayd_generic_reply reply;
377
378 /* Code flow error. Safety net. */
379 assert(sock);
380
381 DBG("Relayd closing stream id %" PRIu64, stream_id);
382
383 msg.stream_id = htobe64(stream_id);
384 msg.last_net_seq_num = htobe64(last_net_seq_num);
385
386 /* Send command */
387 ret = send_command(sock, RELAYD_CLOSE_STREAM, (void *) &msg, sizeof(msg), 0);
388 if (ret < 0) {
389 goto error;
390 }
391
392 /* Recevie response */
393 ret = recv_reply(sock, (void *) &reply, sizeof(reply));
394 if (ret < 0) {
395 goto error;
396 }
397
398 reply.ret_code = be32toh(reply.ret_code);
399
400 /* Return session id or negative ret code. */
401 if (reply.ret_code != LTTNG_OK) {
402 ret = -reply.ret_code;
403 ERR("Relayd close stream replied error %d", ret);
404 } else {
405 /* Success */
406 ret = 0;
407 }
408
409 DBG("Relayd close stream id %" PRIu64 " successfully", stream_id);
410
411 error:
412 return ret;
413 }
414
415 /*
416 * Check for data availability for a given stream id.
417 *
418 * Return 0 if NOT pending, 1 if so and a negative value on error.
419 */
420 int relayd_data_pending(struct lttcomm_sock *sock, uint64_t stream_id,
421 uint64_t last_net_seq_num)
422 {
423 int ret;
424 struct lttcomm_relayd_data_pending msg;
425 struct lttcomm_relayd_generic_reply reply;
426
427 /* Code flow error. Safety net. */
428 assert(sock);
429
430 DBG("Relayd data pending for stream id %" PRIu64, stream_id);
431
432 msg.stream_id = htobe64(stream_id);
433 msg.last_net_seq_num = htobe64(last_net_seq_num);
434
435 /* Send command */
436 ret = send_command(sock, RELAYD_DATA_PENDING, (void *) &msg,
437 sizeof(msg), 0);
438 if (ret < 0) {
439 goto error;
440 }
441
442 /* Recevie response */
443 ret = recv_reply(sock, (void *) &reply, sizeof(reply));
444 if (ret < 0) {
445 goto error;
446 }
447
448 reply.ret_code = be32toh(reply.ret_code);
449
450 /* Return session id or negative ret code. */
451 if (reply.ret_code >= LTTNG_OK) {
452 ret = -reply.ret_code;
453 ERR("Relayd data pending replied error %d", ret);
454 }
455
456 /* At this point, the ret code is either 1 or 0 */
457 ret = reply.ret_code;
458
459 DBG("Relayd data is %s pending for stream id %" PRIu64,
460 ret == 1 ? "NOT" : "", stream_id);
461
462 error:
463 return ret;
464 }
465
466 /*
467 * Check on the relayd side for a quiescent state on the control socket.
468 */
469 int relayd_quiescent_control(struct lttcomm_sock *sock)
470 {
471 int ret;
472 struct lttcomm_relayd_generic_reply reply;
473
474 /* Code flow error. Safety net. */
475 assert(sock);
476
477 DBG("Relayd checking quiescent control state");
478
479 /* Send command */
480 ret = send_command(sock, RELAYD_QUIESCENT_CONTROL, NULL, 0, 0);
481 if (ret < 0) {
482 goto error;
483 }
484
485 /* Recevie response */
486 ret = recv_reply(sock, (void *) &reply, sizeof(reply));
487 if (ret < 0) {
488 goto error;
489 }
490
491 reply.ret_code = be32toh(reply.ret_code);
492
493 /* Return session id or negative ret code. */
494 if (reply.ret_code != LTTNG_OK) {
495 ret = -reply.ret_code;
496 ERR("Relayd quiescent control replied error %d", ret);
497 goto error;
498 }
499
500 /* Control socket is quiescent */
501 return 0;
502
503 error:
504 return ret;
505 }
506
507 /*
508 * Begin a data pending command for a specific session id.
509 */
510 int relayd_begin_data_pending(struct lttcomm_sock *sock, uint64_t id)
511 {
512 int ret;
513 struct lttcomm_relayd_begin_data_pending msg;
514 struct lttcomm_relayd_generic_reply reply;
515
516 /* Code flow error. Safety net. */
517 assert(sock);
518
519 DBG("Relayd begin data pending");
520
521 msg.session_id = htobe64(id);
522
523 /* Send command */
524 ret = send_command(sock, RELAYD_BEGIN_DATA_PENDING, &msg, sizeof(msg), 0);
525 if (ret < 0) {
526 goto error;
527 }
528
529 /* Recevie response */
530 ret = recv_reply(sock, (void *) &reply, sizeof(reply));
531 if (ret < 0) {
532 goto error;
533 }
534
535 reply.ret_code = be32toh(reply.ret_code);
536
537 /* Return session id or negative ret code. */
538 if (reply.ret_code != LTTNG_OK) {
539 ret = -reply.ret_code;
540 ERR("Relayd begin data pending replied error %d", ret);
541 goto error;
542 }
543
544 return 0;
545
546 error:
547 return ret;
548 }
549
550 /*
551 * End a data pending command for a specific session id.
552 *
553 * Return 0 on success and set is_data_inflight to 0 if no data is being
554 * streamed or 1 if it is the case.
555 */
556 int relayd_end_data_pending(struct lttcomm_sock *sock, uint64_t id,
557 unsigned int *is_data_inflight)
558 {
559 int ret;
560 struct lttcomm_relayd_end_data_pending msg;
561 struct lttcomm_relayd_generic_reply reply;
562
563 /* Code flow error. Safety net. */
564 assert(sock);
565
566 DBG("Relayd end data pending");
567
568 msg.session_id = htobe64(id);
569
570 /* Send command */
571 ret = send_command(sock, RELAYD_END_DATA_PENDING, &msg, sizeof(msg), 0);
572 if (ret < 0) {
573 goto error;
574 }
575
576 /* Recevie response */
577 ret = recv_reply(sock, (void *) &reply, sizeof(reply));
578 if (ret < 0) {
579 goto error;
580 }
581
582 reply.ret_code = be32toh(reply.ret_code);
583 if (reply.ret_code < 0) {
584 ret = reply.ret_code;
585 goto error;
586 }
587
588 *is_data_inflight = reply.ret_code;
589
590 DBG("Relayd end data pending is data inflight: %d", reply.ret_code);
591
592 return 0;
593
594 error:
595 return ret;
596 }
This page took 0.041278 seconds and 3 git commands to generate.