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