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/compat/endian.h>
  29 #include <common/sessiond-comm/relayd.h>
  30 #include <common/index/ctf-index.h>
  31
  32 #include "relayd.h"
  33
  34 /*
  35  * Send command. Fill up the header and append the data.
  36  */
  37 static int send_command(struct lttcomm_relayd_sock *rsock,
  38                 enum lttcomm_relayd_command cmd, void *data, size_t size,
  39                 int flags)
  40 {
  41         int ret;
  42         struct lttcomm_relayd_hdr header;
  43         char *buf;
  44         uint64_t buf_size = sizeof(header);
  45
  46         if (rsock->sock.fd < 0) {
  47                 return -ECONNRESET;
  48         }
  49
  50         if (data) {
  51                 buf_size += size;
  52         }
  53
  54         buf = zmalloc(buf_size);
  55         if (buf == NULL) {
  56                 PERROR("zmalloc relayd send command buf");
  57                 ret = -1;
  58                 goto alloc_error;
  59         }
  60
  61         memset(&header, 0, sizeof(header));
  62         header.cmd = htobe32(cmd);
  63         header.data_size = htobe64(size);
  64
  65         /* Zeroed for now since not used. */
  66         header.cmd_version = 0;
  67         header.circuit_id = 0;
  68
  69         /* Prepare buffer to send. */
  70         memcpy(buf, &header, sizeof(header));
  71         if (data) {
  72                 memcpy(buf + sizeof(header), data, size);
  73         }
  74
  75         ret = rsock->sock.ops->sendmsg(&rsock->sock, buf, buf_size, flags);
  76         if (ret < 0) {
  77                 ret = -errno;
  78                 goto error;
  79         }
  80
  81         DBG3("Relayd sending command %d of size %" PRIu64, cmd, buf_size);
  82
  83 error:
  84         free(buf);
  85 alloc_error:
  86         return ret;
  87 }
  88
  89 /*
  90  * Receive reply data on socket. This MUST be call after send_command or else
  91  * could result in unexpected behavior(s).
  92  */
  93 static int recv_reply(struct lttcomm_relayd_sock *rsock, void *data, size_t size)
  94 {
  95         int ret;
  96
  97         if (rsock->sock.fd < 0) {
  98                 return -ECONNRESET;
  99         }
 100
 101         DBG3("Relayd waiting for reply of size %zu", size);
 102
 103         ret = rsock->sock.ops->recvmsg(&rsock->sock, data, size, 0);
 104         if (ret <= 0 || ret != size) {
 105                 if (ret == 0) {
 106                         /* Orderly shutdown. */
 107                         DBG("Socket %d has performed an orderly shutdown", rsock->sock.fd);
 108                 } else {
 109                         DBG("Receiving reply failed on sock %d for size %zu with ret %d",
 110                                         rsock->sock.fd, size, ret);
 111                 }
 112                 /* Always return -1 here and the caller can use errno. */
 113                 ret = -1;
 114                 goto error;
 115         }
 116
 117 error:
 118         return ret;
 119 }
 120
 121 /*
 122  * Starting at 2.4, RELAYD_CREATE_SESSION takes additional parameters to
 123  * support the live reading capability.
 124  */
 125 static int relayd_create_session_2_4(struct lttcomm_relayd_sock *rsock,
 126                 uint64_t *session_id, char *session_name, char *hostname,
 127                 int session_live_timer, unsigned int snapshot)
 128 {
 129         int ret;
 130         struct lttcomm_relayd_create_session_2_4 msg;
 131
 132         strncpy(msg.session_name, session_name, sizeof(msg.session_name));
 133         strncpy(msg.hostname, hostname, sizeof(msg.hostname));
 134         msg.live_timer = htobe32(session_live_timer);
 135         msg.snapshot = htobe32(snapshot);
 136
 137         /* Send command */
 138         ret = send_command(rsock, RELAYD_CREATE_SESSION, &msg, sizeof(msg), 0);
 139         if (ret < 0) {
 140                 goto error;
 141         }
 142
 143 error:
 144         return ret;
 145 }
 146
 147 /*
 148  * RELAYD_CREATE_SESSION from 2.1 to 2.3.
 149  */
 150 static int relayd_create_session_2_1(struct lttcomm_relayd_sock *rsock,
 151                 uint64_t *session_id)
 152 {
 153         int ret;
 154
 155         /* Send command */
 156         ret = send_command(rsock, RELAYD_CREATE_SESSION, NULL, 0, 0);
 157         if (ret < 0) {
 158                 goto error;
 159         }
 160
 161 error:
 162         return ret;
 163 }
 164
 165 /*
 166  * Send a RELAYD_CREATE_SESSION command to the relayd with the given socket and
 167  * set session_id of the relayd if we have a successful reply from the relayd.
 168  *
 169  * On success, return 0 else a negative value which is either an errno error or
 170  * a lttng error code from the relayd.
 171  */
 172 int relayd_create_session(struct lttcomm_relayd_sock *rsock, uint64_t *session_id,
 173                 char *session_name, char *hostname, int session_live_timer,
 174                 unsigned int snapshot)
 175 {
 176         int ret;
 177         struct lttcomm_relayd_status_session reply;
 178
 179         assert(rsock);
 180         assert(session_id);
 181
 182         DBG("Relayd create session");
 183
 184         switch(rsock->minor) {
 185                 case 1:
 186                 case 2:
 187                 case 3:
 188                         ret = relayd_create_session_2_1(rsock, session_id);
 189                         break;
 190                 case 4:
 191                 default:
 192                         ret = relayd_create_session_2_4(rsock, session_id, session_name,
 193                                         hostname, session_live_timer, snapshot);
 194                         break;
 195         }
 196
 197         if (ret < 0) {
 198                 goto error;
 199         }
 200
 201         /* Receive response */
 202         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 203         if (ret < 0) {
 204                 goto error;
 205         }
 206
 207         reply.session_id = be64toh(reply.session_id);
 208         reply.ret_code = be32toh(reply.ret_code);
 209
 210         /* Return session id or negative ret code. */
 211         if (reply.ret_code != LTTNG_OK) {
 212                 ret = -1;
 213                 ERR("Relayd create session replied error %d", reply.ret_code);
 214                 goto error;
 215         } else {
 216                 ret = 0;
 217                 *session_id = reply.session_id;
 218         }
 219
 220         DBG("Relayd session created with id %" PRIu64, reply.session_id);
 221
 222 error:
 223         return ret;
 224 }
 225
 226 /*
 227  * Add stream on the relayd and assign stream handle to the stream_id argument.
 228  *
 229  * On success return 0 else return ret_code negative value.
 230  */
 231 int relayd_add_stream(struct lttcomm_relayd_sock *rsock, const char *channel_name,
 232                 const char *pathname, uint64_t *stream_id,
 233                 uint64_t tracefile_size, uint64_t tracefile_count)
 234 {
 235         int ret;
 236         struct lttcomm_relayd_add_stream msg;
 237         struct lttcomm_relayd_add_stream_2_2 msg_2_2;
 238         struct lttcomm_relayd_status_stream reply;
 239
 240         /* Code flow error. Safety net. */
 241         assert(rsock);
 242         assert(channel_name);
 243         assert(pathname);
 244
 245         DBG("Relayd adding stream for channel name %s", channel_name);
 246
 247         /* Compat with relayd 2.1 */
 248         if (rsock->minor == 1) {
 249                 memset(&msg, 0, sizeof(msg));
 250                 strncpy(msg.channel_name, channel_name, sizeof(msg.channel_name));
 251                 strncpy(msg.pathname, pathname, sizeof(msg.pathname));
 252
 253                 /* Send command */
 254                 ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg, sizeof(msg), 0);
 255                 if (ret < 0) {
 256                         goto error;
 257                 }
 258         } else {
 259                 memset(&msg_2_2, 0, sizeof(msg_2_2));
 260                 /* Compat with relayd 2.2+ */
 261                 strncpy(msg_2_2.channel_name, channel_name, sizeof(msg_2_2.channel_name));
 262                 strncpy(msg_2_2.pathname, pathname, sizeof(msg_2_2.pathname));
 263                 msg_2_2.tracefile_size = htobe64(tracefile_size);
 264                 msg_2_2.tracefile_count = htobe64(tracefile_count);
 265
 266                 /* Send command */
 267                 ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg_2_2, sizeof(msg_2_2), 0);
 268                 if (ret < 0) {
 269                         goto error;
 270                 }
 271         }
 272
 273         /* Waiting for reply */
 274         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 275         if (ret < 0) {
 276                 goto error;
 277         }
 278
 279         /* Back to host bytes order. */
 280         reply.handle = be64toh(reply.handle);
 281         reply.ret_code = be32toh(reply.ret_code);
 282
 283         /* Return session id or negative ret code. */
 284         if (reply.ret_code != LTTNG_OK) {
 285                 ret = -1;
 286                 ERR("Relayd add stream replied error %d", reply.ret_code);
 287         } else {
 288                 /* Success */
 289                 ret = 0;
 290                 *stream_id = reply.handle;
 291         }
 292
 293         DBG("Relayd stream added successfully with handle %" PRIu64,
 294                 reply.handle);
 295
 296 error:
 297         return ret;
 298 }
 299
 300 /*
 301  * Inform the relay that all the streams for the current channel has been sent.
 302  *
 303  * On success return 0 else return ret_code negative value.
 304  */
 305 int relayd_streams_sent(struct lttcomm_relayd_sock *rsock)
 306 {
 307         int ret;
 308         struct lttcomm_relayd_generic_reply reply;
 309
 310         /* Code flow error. Safety net. */
 311         assert(rsock);
 312
 313         DBG("Relayd sending streams sent.");
 314
 315         /* This feature was introduced in 2.4, ignore it for earlier versions. */
 316         if (rsock->minor < 4) {
 317                 ret = 0;
 318                 goto end;
 319         }
 320
 321         /* Send command */
 322         ret = send_command(rsock, RELAYD_STREAMS_SENT, NULL, 0, 0);
 323         if (ret < 0) {
 324                 goto error;
 325         }
 326
 327         /* Waiting for reply */
 328         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 329         if (ret < 0) {
 330                 goto error;
 331         }
 332
 333         /* Back to host bytes order. */
 334         reply.ret_code = be32toh(reply.ret_code);
 335
 336         /* Return session id or negative ret code. */
 337         if (reply.ret_code != LTTNG_OK) {
 338                 ret = -1;
 339                 ERR("Relayd streams sent replied error %d", reply.ret_code);
 340                 goto error;
 341         } else {
 342                 /* Success */
 343                 ret = 0;
 344         }
 345
 346         DBG("Relayd streams sent success");
 347
 348 error:
 349 end:
 350         return ret;
 351 }
 352
 353 /*
 354  * Check version numbers on the relayd.
 355  * If major versions are compatible, we assign minor_to_use to the
 356  * minor version of the procotol we are going to use for this session.
 357  *
 358  * Return 0 if compatible else negative value.
 359  */
 360 int relayd_version_check(struct lttcomm_relayd_sock *rsock)
 361 {
 362         int ret;
 363         struct lttcomm_relayd_version msg;
 364
 365         /* Code flow error. Safety net. */
 366         assert(rsock);
 367
 368         DBG("Relayd version check for major.minor %u.%u", rsock->major,
 369                         rsock->minor);
 370
 371         memset(&msg, 0, sizeof(msg));
 372         /* Prepare network byte order before transmission. */
 373         msg.major = htobe32(rsock->major);
 374         msg.minor = htobe32(rsock->minor);
 375
 376         /* Send command */
 377         ret = send_command(rsock, RELAYD_VERSION, (void *) &msg, sizeof(msg), 0);
 378         if (ret < 0) {
 379                 goto error;
 380         }
 381
 382         /* Receive response */
 383         ret = recv_reply(rsock, (void *) &msg, sizeof(msg));
 384         if (ret < 0) {
 385                 goto error;
 386         }
 387
 388         /* Set back to host bytes order */
 389         msg.major = be32toh(msg.major);
 390         msg.minor = be32toh(msg.minor);
 391
 392         /*
 393          * Only validate the major version. If the other side is higher,
 394          * communication is not possible. Only major version equal can talk to each
 395          * other. If the minor version differs, the lowest version is used by both
 396          * sides.
 397          */
 398         if (msg.major != rsock->major) {
 399                 /* Not compatible */
 400                 ret = -1;
 401                 DBG2("Relayd version is NOT compatible. Relayd version %u != %u (us)",
 402                                 msg.major, rsock->major);
 403                 goto error;
 404         }
 405
 406         /*
 407          * If the relayd's minor version is higher, it will adapt to our version so
 408          * we can continue to use the latest relayd communication data structure.
 409          * If the received minor version is higher, the relayd should adapt to us.
 410          */
 411         if (rsock->minor > msg.minor) {
 412                 rsock->minor = msg.minor;
 413         }
 414
 415         /* Version number compatible */
 416         DBG2("Relayd version is compatible, using protocol version %u.%u",
 417                         rsock->major, rsock->minor);
 418         ret = 0;
 419
 420 error:
 421         return ret;
 422 }
 423
 424 /*
 425  * Add stream on the relayd and assign stream handle to the stream_id argument.
 426  *
 427  * On success return 0 else return ret_code negative value.
 428  */
 429 int relayd_send_metadata(struct lttcomm_relayd_sock *rsock, size_t len)
 430 {
 431         int ret;
 432
 433         /* Code flow error. Safety net. */
 434         assert(rsock);
 435
 436         DBG("Relayd sending metadata of size %zu", len);
 437
 438         /* Send command */
 439         ret = send_command(rsock, RELAYD_SEND_METADATA, NULL, len, 0);
 440         if (ret < 0) {
 441                 goto error;
 442         }
 443
 444         DBG2("Relayd metadata added successfully");
 445
 446         /*
 447          * After that call, the metadata data MUST be sent to the relayd so the
 448          * receive size on the other end matches the len of the metadata packet
 449          * header. This is why we don't wait for a reply here.
 450          */
 451
 452 error:
 453         return ret;
 454 }
 455
 456 /*
 457  * Connect to relay daemon with an allocated lttcomm_relayd_sock.
 458  */
 459 int relayd_connect(struct lttcomm_relayd_sock *rsock)
 460 {
 461         /* Code flow error. Safety net. */
 462         assert(rsock);
 463
 464         if (!rsock->sock.ops) {
 465                 /*
 466                  * Attempting a connect on a non-initialized socket.
 467                  */
 468                 return -ECONNRESET;
 469         }
 470
 471         DBG3("Relayd connect ...");
 472
 473         return rsock->sock.ops->connect(&rsock->sock);
 474 }
 475
 476 /*
 477  * Close relayd socket with an allocated lttcomm_relayd_sock.
 478  *
 479  * If no socket operations are found, simply return 0 meaning that everything
 480  * is fine. Without operations, the socket can not possibly be opened or used.
 481  * This is possible if the socket was allocated but not created. However, the
 482  * caller could simply use it to store a valid file descriptor for instance
 483  * passed over a Unix socket and call this to cleanup but still without a valid
 484  * ops pointer.
 485  *
 486  * Return the close returned value. On error, a negative value is usually
 487  * returned back from close(2).
 488  */
 489 int relayd_close(struct lttcomm_relayd_sock *rsock)
 490 {
 491         int ret;
 492
 493         /* Code flow error. Safety net. */
 494         assert(rsock);
 495
 496         /* An invalid fd is fine, return success. */
 497         if (rsock->sock.fd < 0) {
 498                 ret = 0;
 499                 goto end;
 500         }
 501
 502         DBG3("Relayd closing socket %d", rsock->sock.fd);
 503
 504         if (rsock->sock.ops) {
 505                 ret = rsock->sock.ops->close(&rsock->sock);
 506         } else {
 507                 /* Default call if no specific ops found. */
 508                 ret = close(rsock->sock.fd);
 509                 if (ret < 0) {
 510                         PERROR("relayd_close default close");
 511                 }
 512         }
 513         rsock->sock.fd = -1;
 514
 515 end:
 516         return ret;
 517 }
 518
 519 /*
 520  * Send data header structure to the relayd.
 521  */
 522 int relayd_send_data_hdr(struct lttcomm_relayd_sock *rsock,
 523                 struct lttcomm_relayd_data_hdr *hdr, size_t size)
 524 {
 525         int ret;
 526
 527         /* Code flow error. Safety net. */
 528         assert(rsock);
 529         assert(hdr);
 530
 531         if (rsock->sock.fd < 0) {
 532                 return -ECONNRESET;
 533         }
 534
 535         DBG3("Relayd sending data header of size %zu", size);
 536
 537         /* Again, safety net */
 538         if (size == 0) {
 539                 size = sizeof(struct lttcomm_relayd_data_hdr);
 540         }
 541
 542         /* Only send data header. */
 543         ret = rsock->sock.ops->sendmsg(&rsock->sock, hdr, size, 0);
 544         if (ret < 0) {
 545                 ret = -errno;
 546                 goto error;
 547         }
 548
 549         /*
 550          * The data MUST be sent right after that command for the receive on the
 551          * other end to match the size in the header.
 552          */
 553
 554 error:
 555         return ret;
 556 }
 557
 558 /*
 559  * Send close stream command to the relayd.
 560  */
 561 int relayd_send_close_stream(struct lttcomm_relayd_sock *rsock, uint64_t stream_id,
 562                 uint64_t last_net_seq_num)
 563 {
 564         int ret;
 565         struct lttcomm_relayd_close_stream msg;
 566         struct lttcomm_relayd_generic_reply reply;
 567
 568         /* Code flow error. Safety net. */
 569         assert(rsock);
 570
 571         DBG("Relayd closing stream id %" PRIu64, stream_id);
 572
 573         memset(&msg, 0, sizeof(msg));
 574         msg.stream_id = htobe64(stream_id);
 575         msg.last_net_seq_num = htobe64(last_net_seq_num);
 576
 577         /* Send command */
 578         ret = send_command(rsock, RELAYD_CLOSE_STREAM, (void *) &msg, sizeof(msg), 0);
 579         if (ret < 0) {
 580                 goto error;
 581         }
 582
 583         /* Receive response */
 584         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 585         if (ret < 0) {
 586                 goto error;
 587         }
 588
 589         reply.ret_code = be32toh(reply.ret_code);
 590
 591         /* Return session id or negative ret code. */
 592         if (reply.ret_code != LTTNG_OK) {
 593                 ret = -1;
 594                 ERR("Relayd close stream replied error %d", reply.ret_code);
 595         } else {
 596                 /* Success */
 597                 ret = 0;
 598         }
 599
 600         DBG("Relayd close stream id %" PRIu64 " successfully", stream_id);
 601
 602 error:
 603         return ret;
 604 }
 605
 606 /*
 607  * Check for data availability for a given stream id.
 608  *
 609  * Return 0 if NOT pending, 1 if so and a negative value on error.
 610  */
 611 int relayd_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t stream_id,
 612                 uint64_t last_net_seq_num)
 613 {
 614         int ret;
 615         struct lttcomm_relayd_data_pending msg;
 616         struct lttcomm_relayd_generic_reply reply;
 617
 618         /* Code flow error. Safety net. */
 619         assert(rsock);
 620
 621         DBG("Relayd data pending for stream id %" PRIu64, stream_id);
 622
 623         memset(&msg, 0, sizeof(msg));
 624         msg.stream_id = htobe64(stream_id);
 625         msg.last_net_seq_num = htobe64(last_net_seq_num);
 626
 627         /* Send command */
 628         ret = send_command(rsock, RELAYD_DATA_PENDING, (void *) &msg,
 629                         sizeof(msg), 0);
 630         if (ret < 0) {
 631                 goto error;
 632         }
 633
 634         /* Receive response */
 635         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 636         if (ret < 0) {
 637                 goto error;
 638         }
 639
 640         reply.ret_code = be32toh(reply.ret_code);
 641
 642         /* Return session id or negative ret code. */
 643         if (reply.ret_code >= LTTNG_OK) {
 644                 ERR("Relayd data pending replied error %d", reply.ret_code);
 645         }
 646
 647         /* At this point, the ret code is either 1 or 0 */
 648         ret = reply.ret_code;
 649
 650         DBG("Relayd data is %s pending for stream id %" PRIu64,
 651                         ret == 1 ? "" : "NOT", stream_id);
 652
 653 error:
 654         return ret;
 655 }
 656
 657 /*
 658  * Check on the relayd side for a quiescent state on the control socket.
 659  */
 660 int relayd_quiescent_control(struct lttcomm_relayd_sock *rsock,
 661                 uint64_t metadata_stream_id)
 662 {
 663         int ret;
 664         struct lttcomm_relayd_quiescent_control msg;
 665         struct lttcomm_relayd_generic_reply reply;
 666
 667         /* Code flow error. Safety net. */
 668         assert(rsock);
 669
 670         DBG("Relayd checking quiescent control state");
 671
 672         memset(&msg, 0, sizeof(msg));
 673         msg.stream_id = htobe64(metadata_stream_id);
 674
 675         /* Send command */
 676         ret = send_command(rsock, RELAYD_QUIESCENT_CONTROL, &msg, sizeof(msg), 0);
 677         if (ret < 0) {
 678                 goto error;
 679         }
 680
 681         /* Receive response */
 682         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 683         if (ret < 0) {
 684                 goto error;
 685         }
 686
 687         reply.ret_code = be32toh(reply.ret_code);
 688
 689         /* Return session id or negative ret code. */
 690         if (reply.ret_code != LTTNG_OK) {
 691                 ret = -1;
 692                 ERR("Relayd quiescent control replied error %d", reply.ret_code);
 693                 goto error;
 694         }
 695
 696         /* Control socket is quiescent */
 697         return 0;
 698
 699 error:
 700         return ret;
 701 }
 702
 703 /*
 704  * Begin a data pending command for a specific session id.
 705  */
 706 int relayd_begin_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id)
 707 {
 708         int ret;
 709         struct lttcomm_relayd_begin_data_pending msg;
 710         struct lttcomm_relayd_generic_reply reply;
 711
 712         /* Code flow error. Safety net. */
 713         assert(rsock);
 714
 715         DBG("Relayd begin data pending");
 716
 717         memset(&msg, 0, sizeof(msg));
 718         msg.session_id = htobe64(id);
 719
 720         /* Send command */
 721         ret = send_command(rsock, RELAYD_BEGIN_DATA_PENDING, &msg, sizeof(msg), 0);
 722         if (ret < 0) {
 723                 goto error;
 724         }
 725
 726         /* Receive response */
 727         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 728         if (ret < 0) {
 729                 goto error;
 730         }
 731
 732         reply.ret_code = be32toh(reply.ret_code);
 733
 734         /* Return session id or negative ret code. */
 735         if (reply.ret_code != LTTNG_OK) {
 736                 ret = -1;
 737                 ERR("Relayd begin data pending replied error %d", reply.ret_code);
 738                 goto error;
 739         }
 740
 741         return 0;
 742
 743 error:
 744         return ret;
 745 }
 746
 747 /*
 748  * End a data pending command for a specific session id.
 749  *
 750  * Return 0 on success and set is_data_inflight to 0 if no data is being
 751  * streamed or 1 if it is the case.
 752  */
 753 int relayd_end_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id,
 754                 unsigned int *is_data_inflight)
 755 {
 756         int ret, recv_ret;
 757         struct lttcomm_relayd_end_data_pending msg;
 758         struct lttcomm_relayd_generic_reply reply;
 759
 760         /* Code flow error. Safety net. */
 761         assert(rsock);
 762
 763         DBG("Relayd end data pending");
 764
 765         memset(&msg, 0, sizeof(msg));
 766         msg.session_id = htobe64(id);
 767
 768         /* Send command */
 769         ret = send_command(rsock, RELAYD_END_DATA_PENDING, &msg, sizeof(msg), 0);
 770         if (ret < 0) {
 771                 goto error;
 772         }
 773
 774         /* Receive response */
 775         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 776         if (ret < 0) {
 777                 goto error;
 778         }
 779
 780         recv_ret = be32toh(reply.ret_code);
 781         if (recv_ret < 0) {
 782                 ret = recv_ret;
 783                 goto error;
 784         }
 785
 786         *is_data_inflight = recv_ret;
 787
 788         DBG("Relayd end data pending is data inflight: %d", recv_ret);
 789
 790         return 0;
 791
 792 error:
 793         return ret;
 794 }
 795
 796 /*
 797  * Send index to the relayd.
 798  */
 799 int relayd_send_index(struct lttcomm_relayd_sock *rsock,
 800                 struct ctf_packet_index *index, uint64_t relay_stream_id,
 801                 uint64_t net_seq_num)
 802 {
 803         int ret;
 804         struct lttcomm_relayd_index msg;
 805         struct lttcomm_relayd_generic_reply reply;
 806
 807         /* Code flow error. Safety net. */
 808         assert(rsock);
 809
 810         if (rsock->minor < 4) {
 811                 DBG("Not sending indexes before protocol 2.4");
 812                 ret = 0;
 813                 goto error;
 814         }
 815
 816         DBG("Relayd sending index for stream ID %" PRIu64, relay_stream_id);
 817
 818         memset(&msg, 0, sizeof(msg));
 819         msg.relay_stream_id = htobe64(relay_stream_id);
 820         msg.net_seq_num = htobe64(net_seq_num);
 821
 822         /* The index is already in big endian. */
 823         msg.packet_size = index->packet_size;
 824         msg.content_size = index->content_size;
 825         msg.timestamp_begin = index->timestamp_begin;
 826         msg.timestamp_end = index->timestamp_end;
 827         msg.events_discarded = index->events_discarded;
 828         msg.stream_id = index->stream_id;
 829
 830         /* Send command */
 831         ret = send_command(rsock, RELAYD_SEND_INDEX, &msg, sizeof(msg), 0);
 832         if (ret < 0) {
 833                 goto error;
 834         }
 835
 836         /* Receive response */
 837         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 838         if (ret < 0) {
 839                 goto error;
 840         }
 841
 842         reply.ret_code = be32toh(reply.ret_code);
 843
 844         /* Return session id or negative ret code. */
 845         if (reply.ret_code != LTTNG_OK) {
 846                 ret = -1;
 847                 ERR("Relayd send index replied error %d", reply.ret_code);
 848         } else {
 849                 /* Success */
 850                 ret = 0;
 851         }
 852
 853 error:
 854         return ret;
 855 }