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_sessiond_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  * Add stream on the relayd and assign stream handle to the stream_id argument.
 104  *
 105  * On success return 0 else return ret_code negative value.
 106  */
 107 int relayd_add_stream(struct lttcomm_sock *sock, const char *channel_name,
 108                 const char *pathname, uint64_t *stream_id)
 109 {
 110         int ret;
 111         struct lttcomm_relayd_add_stream msg;
 112         struct lttcomm_relayd_status_stream reply;
 113
 114         /* Code flow error. Safety net. */
 115         assert(sock);
 116         assert(channel_name);
 117         assert(pathname);
 118
 119         DBG("Relayd adding stream for channel name %s", channel_name);
 120
 121         strncpy(msg.channel_name, channel_name, sizeof(msg.channel_name));
 122         strncpy(msg.pathname, pathname, sizeof(msg.pathname));
 123
 124         /* Send command */
 125         ret = send_command(sock, RELAYD_ADD_STREAM, (void *) &msg, sizeof(msg), 0);
 126         if (ret < 0) {
 127                 goto error;
 128         }
 129
 130         /* Waiting for reply */
 131         ret = recv_reply(sock, (void *) &reply, sizeof(reply));
 132         if (ret < 0) {
 133                 goto error;
 134         }
 135
 136         /* Back to host bytes order. */
 137         reply.handle = be64toh(reply.handle);
 138         reply.ret_code = be32toh(reply.ret_code);
 139
 140         /* Return session id or negative ret code. */
 141         if (reply.ret_code != LTTNG_OK) {
 142                 ret = -reply.ret_code;
 143                 ERR("Relayd add stream replied error %d", ret);
 144         } else {
 145                 /* Success */
 146                 ret = 0;
 147                 *stream_id = reply.handle;
 148         }
 149
 150         DBG("Relayd stream added successfully with handle %" PRIu64,
 151                 reply.handle);
 152
 153 error:
 154         return ret;
 155 }
 156
 157 /*
 158  * Check version numbers on the relayd.
 159  *
 160  * Return 0 if compatible else negative value.
 161  */
 162 int relayd_version_check(struct lttcomm_sock *sock, uint32_t major,
 163                 uint32_t minor)
 164 {
 165         int ret;
 166         struct lttcomm_relayd_version reply;
 167
 168         /* Code flow error. Safety net. */
 169         assert(sock);
 170
 171         DBG("Relayd version check for major.minor %u.%u", major, minor);
 172
 173         /* Send command */
 174         ret = send_command(sock, RELAYD_VERSION, NULL, 0, 0);
 175         if (ret < 0) {
 176                 goto error;
 177         }
 178
 179         /* Recevie response */
 180         ret = recv_reply(sock, (void *) &reply, sizeof(reply));
 181         if (ret < 0) {
 182                 goto error;
 183         }
 184
 185         /* Set back to host bytes order */
 186         reply.major = be32toh(reply.major);
 187         reply.minor = be32toh(reply.minor);
 188
 189         /* Validate version */
 190         if (reply.major <= major) {
 191                 if (reply.minor <= minor) {
 192                         /* Compatible */
 193                         ret = 0;
 194                         DBG2("Relayd version is compatible");
 195                         goto error;
 196                 }
 197         }
 198
 199         /* Version number not compatible */
 200         DBG2("Relayd version is NOT compatible %u.%u > %u.%u", reply.major,
 201                         reply.minor, major, minor);
 202         ret = -1;
 203
 204 error:
 205         return ret;
 206 }
 207
 208 /*
 209  * Add stream on the relayd and assign stream handle to the stream_id argument.
 210  *
 211  * On success return 0 else return ret_code negative value.
 212  */
 213 int relayd_send_metadata(struct lttcomm_sock *sock, size_t len)
 214 {
 215         int ret;
 216
 217         /* Code flow error. Safety net. */
 218         assert(sock);
 219
 220         DBG("Relayd sending metadata of size %zu", len);
 221
 222         /* Send command */
 223         ret = send_command(sock, RELAYD_SEND_METADATA, NULL, len, 0);
 224         if (ret < 0) {
 225                 goto error;
 226         }
 227
 228         DBG2("Relayd metadata added successfully");
 229
 230         /*
 231          * After that call, the metadata data MUST be sent to the relayd so the
 232          * receive size on the other end matches the len of the metadata packet
 233          * header. This is why we don't wait for a reply here.
 234          */
 235
 236 error:
 237         return ret;
 238 }
 239
 240 /*
 241  * Connect to relay daemon with an allocated lttcomm_sock.
 242  */
 243 int relayd_connect(struct lttcomm_sock *sock)
 244 {
 245         /* Code flow error. Safety net. */
 246         assert(sock);
 247
 248         DBG3("Relayd connect ...");
 249
 250         return sock->ops->connect(sock);
 251 }
 252
 253 /*
 254  * Close relayd socket with an allocated lttcomm_sock.
 255  */
 256 int relayd_close(struct lttcomm_sock *sock)
 257 {
 258         /* Code flow error. Safety net. */
 259         assert(sock);
 260
 261         DBG3("Relayd closing socket %d", sock->fd);
 262
 263         return sock->ops->close(sock);
 264 }
 265
 266 /*
 267  * Send data header structure to the relayd.
 268  */
 269 int relayd_send_data_hdr(struct lttcomm_sock *sock,
 270                 struct lttcomm_relayd_data_hdr *hdr, size_t size)
 271 {
 272         int ret;
 273
 274         /* Code flow error. Safety net. */
 275         assert(sock);
 276         assert(hdr);
 277
 278         DBG3("Relayd sending data header of size %ld", size);
 279
 280         /* Again, safety net */
 281         if (size == 0) {
 282                 size = sizeof(struct lttcomm_relayd_data_hdr);
 283         }
 284
 285         /* Only send data header. */
 286         ret = sock->ops->sendmsg(sock, hdr, size, 0);
 287         if (ret < 0) {
 288                 ret = -errno;
 289                 goto error;
 290         }
 291
 292         /*
 293          * The data MUST be sent right after that command for the receive on the
 294          * other end to match the size in the header.
 295          */
 296
 297 error:
 298         return ret;
 299 }
 300
 301 /*
 302  * Send close stream command to the relayd.
 303  */
 304 int relayd_send_close_stream(struct lttcomm_sock *sock, uint64_t stream_id,
 305                 uint64_t last_net_seq_num)
 306 {
 307         int ret;
 308         struct lttcomm_relayd_close_stream msg;
 309         struct lttcomm_relayd_generic_reply reply;
 310
 311         /* Code flow error. Safety net. */
 312         assert(sock);
 313
 314         DBG("Relayd closing stream id %" PRIu64, stream_id);
 315
 316         msg.stream_id = htobe64(stream_id);
 317         msg.last_net_seq_num = htobe64(last_net_seq_num);
 318
 319         /* Send command */
 320         ret = send_command(sock, RELAYD_CLOSE_STREAM, (void *) &msg, sizeof(msg), 0);
 321         if (ret < 0) {
 322                 goto error;
 323         }
 324
 325         /* Recevie response */
 326         ret = recv_reply(sock, (void *) &reply, sizeof(reply));
 327         if (ret < 0) {
 328                 goto error;
 329         }
 330
 331         reply.ret_code = be32toh(reply.ret_code);
 332
 333         /* Return session id or negative ret code. */
 334         if (reply.ret_code != LTTNG_OK) {
 335                 ret = -reply.ret_code;
 336                 ERR("Relayd close stream replied error %d", ret);
 337         } else {
 338                 /* Success */
 339                 ret = 0;
 340         }
 341
 342         DBG("Relayd close stream id %" PRIu64 " successfully", stream_id);
 343
 344 error:
 345         return ret;
 346 }
 347
 348 /*
 349  * Check for data availability for a given stream id.
 350  *
 351  * Return 0 if NOT available, 1 if so and a negative value on error.
 352  */
 353 int relayd_data_available(struct lttcomm_sock *sock, uint64_t stream_id,
 354                 uint64_t last_net_seq_num)
 355 {
 356         int ret;
 357         struct lttcomm_relayd_data_available msg;
 358         struct lttcomm_relayd_generic_reply reply;
 359
 360         /* Code flow error. Safety net. */
 361         assert(sock);
 362
 363         DBG("Relayd data available for stream id %" PRIu64, stream_id);
 364
 365         msg.stream_id = htobe64(stream_id);
 366         msg.last_net_seq_num = htobe64(last_net_seq_num);
 367
 368         /* Send command */
 369         ret = send_command(sock, RELAYD_DATA_AVAILABLE, (void *) &msg,
 370                         sizeof(msg), 0);
 371         if (ret < 0) {
 372                 goto error;
 373         }
 374
 375         /* Recevie response */
 376         ret = recv_reply(sock, (void *) &reply, sizeof(reply));
 377         if (ret < 0) {
 378                 goto error;
 379         }
 380
 381         reply.ret_code = be32toh(reply.ret_code);
 382
 383         /* Return session id or negative ret code. */
 384         if (reply.ret_code >= LTTNG_OK) {
 385                 ret = -reply.ret_code;
 386                 ERR("Relayd data available replied error %d", ret);
 387         }
 388
 389         /* At this point, the ret code is either 1 or 0 */
 390         ret = reply.ret_code;
 391
 392         DBG("Relayd data is %s available for stream id %" PRIu64,
 393                         ret == 1 ? "" : "NOT", stream_id);
 394
 395 error:
 396         return ret;
 397 }
 398
 399 /*
 400  * Check on the relayd side for a quiescent state on the control socket.
 401  */
 402 int relayd_quiescent_control(struct lttcomm_sock *sock)
 403 {
 404         int ret;
 405         struct lttcomm_relayd_generic_reply reply;
 406
 407         /* Code flow error. Safety net. */
 408         assert(sock);
 409
 410         DBG("Relayd checking quiescent control state");
 411
 412         /* Send command */
 413         ret = send_command(sock, RELAYD_QUIESCENT_CONTROL, NULL, 0, 0);
 414         if (ret < 0) {
 415                 goto error;
 416         }
 417
 418         /* Recevie response */
 419         ret = recv_reply(sock, (void *) &reply, sizeof(reply));
 420         if (ret < 0) {
 421                 goto error;
 422         }
 423
 424         reply.ret_code = be32toh(reply.ret_code);
 425
 426         /* Return session id or negative ret code. */
 427         if (reply.ret_code != LTTNG_OK) {
 428                 ret = -reply.ret_code;
 429                 ERR("Relayd quiescent control replied error %d", ret);
 430                 goto error;
 431         }
 432
 433         /* Control socket is quiescent */
 434         return 1;
 435
 436 error:
 437         return ret;
 438 }