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 _LGPL_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/compat/string.h>
  30 #include <common/sessiond-comm/relayd.h>
  31 #include <common/index/ctf-index.h>
  32 #include <common/trace-chunk.h>
  33 #include <common/string-utils/format.h>
  34
  35 #include "relayd.h"
  36
  37 /*
  38  * Send command. Fill up the header and append the data.
  39  */
  40 static int send_command(struct lttcomm_relayd_sock *rsock,
  41                 enum lttcomm_relayd_command cmd, const void *data, size_t size,
  42                 int flags)
  43 {
  44         int ret;
  45         struct lttcomm_relayd_hdr header;
  46         char *buf;
  47         uint64_t buf_size = sizeof(header);
  48
  49         if (rsock->sock.fd < 0) {
  50                 return -ECONNRESET;
  51         }
  52
  53         if (data) {
  54                 buf_size += size;
  55         }
  56
  57         buf = zmalloc(buf_size);
  58         if (buf == NULL) {
  59                 PERROR("zmalloc relayd send command buf");
  60                 ret = -1;
  61                 goto alloc_error;
  62         }
  63
  64         memset(&header, 0, sizeof(header));
  65         header.cmd = htobe32(cmd);
  66         header.data_size = htobe64(size);
  67
  68         /* Zeroed for now since not used. */
  69         header.cmd_version = 0;
  70         header.circuit_id = 0;
  71
  72         /* Prepare buffer to send. */
  73         memcpy(buf, &header, sizeof(header));
  74         if (data) {
  75                 memcpy(buf + sizeof(header), data, size);
  76         }
  77
  78         DBG3("Relayd sending command %d of size %" PRIu64, (int) cmd, buf_size);
  79         ret = rsock->sock.ops->sendmsg(&rsock->sock, buf, buf_size, flags);
  80         if (ret < 0) {
  81                 PERROR("Failed to send command %d of size %" PRIu64,
  82                                 (int) cmd, buf_size);
  83                 ret = -errno;
  84                 goto error;
  85         }
  86 error:
  87         free(buf);
  88 alloc_error:
  89         return ret;
  90 }
  91
  92 /*
  93  * Receive reply data on socket. This MUST be call after send_command or else
  94  * could result in unexpected behavior(s).
  95  */
  96 static int recv_reply(struct lttcomm_relayd_sock *rsock, void *data, size_t size)
  97 {
  98         int ret;
  99
 100         if (rsock->sock.fd < 0) {
 101                 return -ECONNRESET;
 102         }
 103
 104         DBG3("Relayd waiting for reply of size %zu", size);
 105
 106         ret = rsock->sock.ops->recvmsg(&rsock->sock, data, size, 0);
 107         if (ret <= 0 || ret != size) {
 108                 if (ret == 0) {
 109                         /* Orderly shutdown. */
 110                         DBG("Socket %d has performed an orderly shutdown", rsock->sock.fd);
 111                 } else {
 112                         DBG("Receiving reply failed on sock %d for size %zu with ret %d",
 113                                         rsock->sock.fd, size, ret);
 114                 }
 115                 /* Always return -1 here and the caller can use errno. */
 116                 ret = -1;
 117                 goto error;
 118         }
 119
 120 error:
 121         return ret;
 122 }
 123
 124 /*
 125  * Starting from 2.11, RELAYD_CREATE_SESSION payload (session_name & hostname)
 126  * have no length restriction on the sender side.
 127  * Length for both payloads is stored in the msg struct. A new dynamic size
 128  * payload size is introduced.
 129  */
 130 static int relayd_create_session_2_11(struct lttcomm_relayd_sock *rsock,
 131                 const char *session_name, const char *hostname,
 132                 int session_live_timer, unsigned int snapshot,
 133                 uint64_t sessiond_session_id, const lttng_uuid sessiond_uuid,
 134                 const uint64_t *current_chunk_id,
 135                 time_t creation_time)
 136 {
 137         int ret;
 138         struct lttcomm_relayd_create_session_2_11 *msg = NULL;
 139         size_t session_name_len;
 140         size_t hostname_len;
 141         size_t msg_length;
 142
 143         /* The two names are sent with a '\0' delimiter between them. */
 144         session_name_len = strlen(session_name) + 1;
 145         hostname_len = strlen(hostname) + 1;
 146
 147         msg_length = sizeof(*msg) + session_name_len + hostname_len;
 148         msg = zmalloc(msg_length);
 149         if (!msg) {
 150                 PERROR("zmalloc create_session_2_11 command message");
 151                 ret = -1;
 152                 goto error;
 153         }
 154
 155         assert(session_name_len <= UINT32_MAX);
 156         msg->session_name_len = htobe32(session_name_len);
 157
 158         assert(hostname_len <= UINT32_MAX);
 159         msg->hostname_len = htobe32(hostname_len);
 160
 161         if (lttng_strncpy(msg->names, session_name, session_name_len)) {
 162                 ret = -1;
 163                 goto error;
 164         }
 165         if (lttng_strncpy(msg->names + session_name_len, hostname, hostname_len)) {
 166                 ret = -1;
 167                 goto error;
 168         }
 169
 170         msg->live_timer = htobe32(session_live_timer);
 171         msg->snapshot = !!snapshot;
 172
 173         lttng_uuid_copy(msg->sessiond_uuid, sessiond_uuid);
 174         msg->session_id = htobe64(sessiond_session_id);
 175
 176         if (current_chunk_id) {
 177                 LTTNG_OPTIONAL_SET(&msg->current_chunk_id,
 178                                 htobe64(*current_chunk_id));
 179         }
 180
 181         msg->creation_time = htobe64((uint64_t) creation_time);
 182
 183         /* Send command */
 184         ret = send_command(rsock, RELAYD_CREATE_SESSION, msg, msg_length, 0);
 185         if (ret < 0) {
 186                 goto error;
 187         }
 188 error:
 189         free(msg);
 190         return ret;
 191 }
 192 /*
 193  * From 2.4 to 2.10, RELAYD_CREATE_SESSION takes additional parameters to
 194  * support the live reading capability.
 195  */
 196 static int relayd_create_session_2_4(struct lttcomm_relayd_sock *rsock,
 197                 const char *session_name, const char *hostname,
 198                 int session_live_timer, unsigned int snapshot)
 199 {
 200         int ret;
 201         struct lttcomm_relayd_create_session_2_4 msg;
 202
 203         if (lttng_strncpy(msg.session_name, session_name,
 204                         sizeof(msg.session_name))) {
 205                 ret = -1;
 206                 goto error;
 207         }
 208         if (lttng_strncpy(msg.hostname, hostname, sizeof(msg.hostname))) {
 209                 ret = -1;
 210                 goto error;
 211         }
 212         msg.live_timer = htobe32(session_live_timer);
 213         msg.snapshot = htobe32(snapshot);
 214
 215         /* Send command */
 216         ret = send_command(rsock, RELAYD_CREATE_SESSION, &msg, sizeof(msg), 0);
 217         if (ret < 0) {
 218                 goto error;
 219         }
 220
 221 error:
 222         return ret;
 223 }
 224
 225 /*
 226  * RELAYD_CREATE_SESSION from 2.1 to 2.3.
 227  */
 228 static int relayd_create_session_2_1(struct lttcomm_relayd_sock *rsock)
 229 {
 230         int ret;
 231
 232         /* Send command */
 233         ret = send_command(rsock, RELAYD_CREATE_SESSION, NULL, 0, 0);
 234         if (ret < 0) {
 235                 goto error;
 236         }
 237
 238 error:
 239         return ret;
 240 }
 241
 242 /*
 243  * Send a RELAYD_CREATE_SESSION command to the relayd with the given socket and
 244  * set session_id of the relayd if we have a successful reply from the relayd.
 245  *
 246  * On success, return 0 else a negative value which is either an errno error or
 247  * a lttng error code from the relayd.
 248  */
 249 int relayd_create_session(struct lttcomm_relayd_sock *rsock,
 250                 uint64_t *relayd_session_id,
 251                 const char *session_name, const char *hostname,
 252                 int session_live_timer,
 253                 unsigned int snapshot, uint64_t sessiond_session_id,
 254                 const lttng_uuid sessiond_uuid,
 255                 const uint64_t *current_chunk_id,
 256                 time_t creation_time)
 257 {
 258         int ret;
 259         struct lttcomm_relayd_status_session reply;
 260
 261         assert(rsock);
 262         assert(relayd_session_id);
 263
 264         DBG("Relayd create session");
 265
 266         if (rsock->minor < 4) {
 267                 /* From 2.1 to 2.3 */
 268                 ret = relayd_create_session_2_1(rsock);
 269         } else if (rsock->minor >= 4 && rsock->minor < 11) {
 270                 /* From 2.4 to 2.10 */
 271                 ret = relayd_create_session_2_4(rsock, session_name,
 272                                 hostname, session_live_timer, snapshot);
 273         } else {
 274                 /* From 2.11 to ... */
 275                 ret = relayd_create_session_2_11(rsock, session_name,
 276                                 hostname, session_live_timer, snapshot,
 277                                 sessiond_session_id, sessiond_uuid,
 278                                 current_chunk_id, creation_time);
 279         }
 280
 281         if (ret < 0) {
 282                 goto error;
 283         }
 284
 285         /* Receive response */
 286         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 287         if (ret < 0) {
 288                 goto error;
 289         }
 290
 291         reply.session_id = be64toh(reply.session_id);
 292         reply.ret_code = be32toh(reply.ret_code);
 293
 294         /* Return session id or negative ret code. */
 295         if (reply.ret_code != LTTNG_OK) {
 296                 ret = -1;
 297                 ERR("Relayd create session replied error %d", reply.ret_code);
 298                 goto error;
 299         } else {
 300                 ret = 0;
 301                 *relayd_session_id = reply.session_id;
 302         }
 303
 304         DBG("Relayd session created with id %" PRIu64, reply.session_id);
 305
 306 error:
 307         return ret;
 308 }
 309
 310 static int relayd_add_stream_2_1(struct lttcomm_relayd_sock *rsock,
 311                 const char *channel_name, const char *pathname)
 312 {
 313         int ret;
 314         struct lttcomm_relayd_add_stream msg;
 315
 316         memset(&msg, 0, sizeof(msg));
 317         if (lttng_strncpy(msg.channel_name, channel_name,
 318                                 sizeof(msg.channel_name))) {
 319                 ret = -1;
 320                 goto error;
 321         }
 322
 323         if (lttng_strncpy(msg.pathname, pathname,
 324                                 sizeof(msg.pathname))) {
 325                 ret = -1;
 326                 goto error;
 327         }
 328
 329         /* Send command */
 330         ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg, sizeof(msg), 0);
 331         if (ret < 0) {
 332                 ret = -1;
 333                 goto error;
 334         }
 335         ret = 0;
 336 error:
 337         return ret;
 338 }
 339
 340 static int relayd_add_stream_2_2(struct lttcomm_relayd_sock *rsock,
 341                 const char *channel_name, const char *pathname,
 342                 uint64_t tracefile_size, uint64_t tracefile_count)
 343 {
 344         int ret;
 345         struct lttcomm_relayd_add_stream_2_2 msg;
 346
 347         memset(&msg, 0, sizeof(msg));
 348         /* Compat with relayd 2.2 to 2.10 */
 349         if (lttng_strncpy(msg.channel_name, channel_name,
 350                                 sizeof(msg.channel_name))) {
 351                 ret = -1;
 352                 goto error;
 353         }
 354         if (lttng_strncpy(msg.pathname, pathname,
 355                                 sizeof(msg.pathname))) {
 356                 ret = -1;
 357                 goto error;
 358         }
 359         msg.tracefile_size = htobe64(tracefile_size);
 360         msg.tracefile_count = htobe64(tracefile_count);
 361
 362         /* Send command */
 363         ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg, sizeof(msg), 0);
 364         if (ret < 0) {
 365                 goto error;
 366         }
 367         ret = 0;
 368 error:
 369         return ret;
 370 }
 371
 372 static int relayd_add_stream_2_11(struct lttcomm_relayd_sock *rsock,
 373                 const char *channel_name, const char *pathname,
 374                 uint64_t tracefile_size, uint64_t tracefile_count,
 375                 uint64_t trace_archive_id)
 376 {
 377         int ret;
 378         struct lttcomm_relayd_add_stream_2_11 *msg = NULL;
 379         size_t channel_name_len;
 380         size_t pathname_len;
 381         size_t msg_length;
 382
 383         /* The two names are sent with a '\0' delimiter between them. */
 384         channel_name_len = strlen(channel_name) + 1;
 385         pathname_len = strlen(pathname) + 1;
 386
 387         msg_length = sizeof(*msg) + channel_name_len + pathname_len;
 388         msg = zmalloc(msg_length);
 389         if (!msg) {
 390                 PERROR("zmalloc add_stream_2_11 command message");
 391                 ret = -1;
 392                 goto error;
 393         }
 394
 395         assert(channel_name_len <= UINT32_MAX);
 396         msg->channel_name_len = htobe32(channel_name_len);
 397
 398         assert(pathname_len <= UINT32_MAX);
 399         msg->pathname_len = htobe32(pathname_len);
 400
 401         if (lttng_strncpy(msg->names, channel_name, channel_name_len)) {
 402                 ret = -1;
 403                 goto error;
 404         }
 405         if (lttng_strncpy(msg->names + channel_name_len, pathname, pathname_len)) {
 406                 ret = -1;
 407                 goto error;
 408         }
 409
 410         msg->tracefile_size = htobe64(tracefile_size);
 411         msg->tracefile_count = htobe64(tracefile_count);
 412         msg->trace_chunk_id = htobe64(trace_archive_id);
 413
 414         /* Send command */
 415         ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) msg, msg_length, 0);
 416         if (ret < 0) {
 417                 goto error;
 418         }
 419         ret = 0;
 420 error:
 421         free(msg);
 422         return ret;
 423 }
 424
 425 /*
 426  * Add stream on the relayd and assign stream handle to the stream_id argument.
 427  *
 428  * On success return 0 else return ret_code negative value.
 429  */
 430 int relayd_add_stream(struct lttcomm_relayd_sock *rsock, const char *channel_name,
 431                 const char *pathname, uint64_t *stream_id,
 432                 uint64_t tracefile_size, uint64_t tracefile_count,
 433                 struct lttng_trace_chunk *trace_chunk)
 434 {
 435         int ret;
 436         struct lttcomm_relayd_status_stream reply;
 437
 438         /* Code flow error. Safety net. */
 439         assert(rsock);
 440         assert(channel_name);
 441         assert(pathname);
 442
 443         DBG("Relayd adding stream for channel name %s", channel_name);
 444
 445         /* Compat with relayd 2.1 */
 446         if (rsock->minor == 1) {
 447                 /* For 2.1 */
 448                 assert(!trace_chunk);
 449                 ret = relayd_add_stream_2_1(rsock, channel_name, pathname);
 450
 451         } else if (rsock->minor > 1 && rsock->minor < 11) {
 452                 /* From 2.2 to 2.10 */
 453                 assert(!trace_chunk);
 454                 ret = relayd_add_stream_2_2(rsock, channel_name, pathname,
 455                                 tracefile_size, tracefile_count);
 456         } else {
 457                 enum lttng_trace_chunk_status chunk_status;
 458                 uint64_t chunk_id;
 459
 460                 assert(trace_chunk);
 461                 chunk_status = lttng_trace_chunk_get_id(trace_chunk,
 462                                 &chunk_id);
 463                 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
 464
 465                 /* From 2.11 to ...*/
 466                 ret = relayd_add_stream_2_11(rsock, channel_name, pathname,
 467                                 tracefile_size, tracefile_count,
 468                                 chunk_id);
 469         }
 470
 471         if (ret) {
 472                 ret = -1;
 473                 goto error;
 474         }
 475
 476         /* Waiting for reply */
 477         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 478         if (ret < 0) {
 479                 goto error;
 480         }
 481
 482         /* Back to host bytes order. */
 483         reply.handle = be64toh(reply.handle);
 484         reply.ret_code = be32toh(reply.ret_code);
 485
 486         /* Return session id or negative ret code. */
 487         if (reply.ret_code != LTTNG_OK) {
 488                 ret = -1;
 489                 ERR("Relayd add stream replied error %d", reply.ret_code);
 490         } else {
 491                 /* Success */
 492                 ret = 0;
 493                 *stream_id = reply.handle;
 494         }
 495
 496         DBG("Relayd stream added successfully with handle %" PRIu64,
 497                         reply.handle);
 498
 499 error:
 500         return ret;
 501 }
 502
 503 /*
 504  * Inform the relay that all the streams for the current channel has been sent.
 505  *
 506  * On success return 0 else return ret_code negative value.
 507  */
 508 int relayd_streams_sent(struct lttcomm_relayd_sock *rsock)
 509 {
 510         int ret;
 511         struct lttcomm_relayd_generic_reply reply;
 512
 513         /* Code flow error. Safety net. */
 514         assert(rsock);
 515
 516         DBG("Relayd sending streams sent.");
 517
 518         /* This feature was introduced in 2.4, ignore it for earlier versions. */
 519         if (rsock->minor < 4) {
 520                 ret = 0;
 521                 goto end;
 522         }
 523
 524         /* Send command */
 525         ret = send_command(rsock, RELAYD_STREAMS_SENT, NULL, 0, 0);
 526         if (ret < 0) {
 527                 goto error;
 528         }
 529
 530         /* Waiting for reply */
 531         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 532         if (ret < 0) {
 533                 goto error;
 534         }
 535
 536         /* Back to host bytes order. */
 537         reply.ret_code = be32toh(reply.ret_code);
 538
 539         /* Return session id or negative ret code. */
 540         if (reply.ret_code != LTTNG_OK) {
 541                 ret = -1;
 542                 ERR("Relayd streams sent replied error %d", reply.ret_code);
 543                 goto error;
 544         } else {
 545                 /* Success */
 546                 ret = 0;
 547         }
 548
 549         DBG("Relayd streams sent success");
 550
 551 error:
 552 end:
 553         return ret;
 554 }
 555
 556 /*
 557  * Check version numbers on the relayd.
 558  * If major versions are compatible, we assign minor_to_use to the
 559  * minor version of the procotol we are going to use for this session.
 560  *
 561  * Return 0 if the two daemons are compatible, LTTNG_ERR_RELAYD_VERSION_FAIL
 562  * otherwise, or a negative value on network errors.
 563  */
 564 int relayd_version_check(struct lttcomm_relayd_sock *rsock)
 565 {
 566         int ret;
 567         struct lttcomm_relayd_version msg;
 568
 569         /* Code flow error. Safety net. */
 570         assert(rsock);
 571
 572         DBG("Relayd version check for major.minor %u.%u", rsock->major,
 573                         rsock->minor);
 574
 575         memset(&msg, 0, sizeof(msg));
 576         /* Prepare network byte order before transmission. */
 577         msg.major = htobe32(rsock->major);
 578         msg.minor = htobe32(rsock->minor);
 579
 580         /* Send command */
 581         ret = send_command(rsock, RELAYD_VERSION, (void *) &msg, sizeof(msg), 0);
 582         if (ret < 0) {
 583                 goto error;
 584         }
 585
 586         /* Receive response */
 587         ret = recv_reply(rsock, (void *) &msg, sizeof(msg));
 588         if (ret < 0) {
 589                 goto error;
 590         }
 591
 592         /* Set back to host bytes order */
 593         msg.major = be32toh(msg.major);
 594         msg.minor = be32toh(msg.minor);
 595
 596         /*
 597          * Only validate the major version. If the other side is higher,
 598          * communication is not possible. Only major version equal can talk to each
 599          * other. If the minor version differs, the lowest version is used by both
 600          * sides.
 601          */
 602         if (msg.major != rsock->major) {
 603                 /* Not compatible */
 604                 ret = LTTNG_ERR_RELAYD_VERSION_FAIL;
 605                 DBG2("Relayd version is NOT compatible. Relayd version %u != %u (us)",
 606                                 msg.major, rsock->major);
 607                 goto error;
 608         }
 609
 610         /*
 611          * If the relayd's minor version is higher, it will adapt to our version so
 612          * we can continue to use the latest relayd communication data structure.
 613          * If the received minor version is higher, the relayd should adapt to us.
 614          */
 615         if (rsock->minor > msg.minor) {
 616                 rsock->minor = msg.minor;
 617         }
 618
 619         /* Version number compatible */
 620         DBG2("Relayd version is compatible, using protocol version %u.%u",
 621                         rsock->major, rsock->minor);
 622         ret = 0;
 623
 624 error:
 625         return ret;
 626 }
 627
 628 /*
 629  * Add stream on the relayd and assign stream handle to the stream_id argument.
 630  *
 631  * On success return 0 else return ret_code negative value.
 632  */
 633 int relayd_send_metadata(struct lttcomm_relayd_sock *rsock, size_t len)
 634 {
 635         int ret;
 636
 637         /* Code flow error. Safety net. */
 638         assert(rsock);
 639
 640         DBG("Relayd sending metadata of size %zu", len);
 641
 642         /* Send command */
 643         ret = send_command(rsock, RELAYD_SEND_METADATA, NULL, len, 0);
 644         if (ret < 0) {
 645                 goto error;
 646         }
 647
 648         DBG2("Relayd metadata added successfully");
 649
 650         /*
 651          * After that call, the metadata data MUST be sent to the relayd so the
 652          * receive size on the other end matches the len of the metadata packet
 653          * header. This is why we don't wait for a reply here.
 654          */
 655
 656 error:
 657         return ret;
 658 }
 659
 660 /*
 661  * Connect to relay daemon with an allocated lttcomm_relayd_sock.
 662  */
 663 int relayd_connect(struct lttcomm_relayd_sock *rsock)
 664 {
 665         /* Code flow error. Safety net. */
 666         assert(rsock);
 667
 668         if (!rsock->sock.ops) {
 669                 /*
 670                  * Attempting a connect on a non-initialized socket.
 671                  */
 672                 return -ECONNRESET;
 673         }
 674
 675         DBG3("Relayd connect ...");
 676
 677         return rsock->sock.ops->connect(&rsock->sock);
 678 }
 679
 680 /*
 681  * Close relayd socket with an allocated lttcomm_relayd_sock.
 682  *
 683  * If no socket operations are found, simply return 0 meaning that everything
 684  * is fine. Without operations, the socket can not possibly be opened or used.
 685  * This is possible if the socket was allocated but not created. However, the
 686  * caller could simply use it to store a valid file descriptor for instance
 687  * passed over a Unix socket and call this to cleanup but still without a valid
 688  * ops pointer.
 689  *
 690  * Return the close returned value. On error, a negative value is usually
 691  * returned back from close(2).
 692  */
 693 int relayd_close(struct lttcomm_relayd_sock *rsock)
 694 {
 695         int ret;
 696
 697         /* Code flow error. Safety net. */
 698         assert(rsock);
 699
 700         /* An invalid fd is fine, return success. */
 701         if (rsock->sock.fd < 0) {
 702                 ret = 0;
 703                 goto end;
 704         }
 705
 706         DBG3("Relayd closing socket %d", rsock->sock.fd);
 707
 708         if (rsock->sock.ops) {
 709                 ret = rsock->sock.ops->close(&rsock->sock);
 710         } else {
 711                 /* Default call if no specific ops found. */
 712                 ret = close(rsock->sock.fd);
 713                 if (ret < 0) {
 714                         PERROR("relayd_close default close");
 715                 }
 716         }
 717         rsock->sock.fd = -1;
 718
 719 end:
 720         return ret;
 721 }
 722
 723 /*
 724  * Send data header structure to the relayd.
 725  */
 726 int relayd_send_data_hdr(struct lttcomm_relayd_sock *rsock,
 727                 struct lttcomm_relayd_data_hdr *hdr, size_t size)
 728 {
 729         int ret;
 730
 731         /* Code flow error. Safety net. */
 732         assert(rsock);
 733         assert(hdr);
 734
 735         if (rsock->sock.fd < 0) {
 736                 return -ECONNRESET;
 737         }
 738
 739         DBG3("Relayd sending data header of size %zu", size);
 740
 741         /* Again, safety net */
 742         if (size == 0) {
 743                 size = sizeof(struct lttcomm_relayd_data_hdr);
 744         }
 745
 746         /* Only send data header. */
 747         ret = rsock->sock.ops->sendmsg(&rsock->sock, hdr, size, 0);
 748         if (ret < 0) {
 749                 ret = -errno;
 750                 goto error;
 751         }
 752
 753         /*
 754          * The data MUST be sent right after that command for the receive on the
 755          * other end to match the size in the header.
 756          */
 757
 758 error:
 759         return ret;
 760 }
 761
 762 /*
 763  * Send close stream command to the relayd.
 764  */
 765 int relayd_send_close_stream(struct lttcomm_relayd_sock *rsock, uint64_t stream_id,
 766                 uint64_t last_net_seq_num)
 767 {
 768         int ret;
 769         struct lttcomm_relayd_close_stream msg;
 770         struct lttcomm_relayd_generic_reply reply;
 771
 772         /* Code flow error. Safety net. */
 773         assert(rsock);
 774
 775         DBG("Relayd closing stream id %" PRIu64, stream_id);
 776
 777         memset(&msg, 0, sizeof(msg));
 778         msg.stream_id = htobe64(stream_id);
 779         msg.last_net_seq_num = htobe64(last_net_seq_num);
 780
 781         /* Send command */
 782         ret = send_command(rsock, RELAYD_CLOSE_STREAM, (void *) &msg, sizeof(msg), 0);
 783         if (ret < 0) {
 784                 goto error;
 785         }
 786
 787         /* Receive response */
 788         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 789         if (ret < 0) {
 790                 goto error;
 791         }
 792
 793         reply.ret_code = be32toh(reply.ret_code);
 794
 795         /* Return session id or negative ret code. */
 796         if (reply.ret_code != LTTNG_OK) {
 797                 ret = -1;
 798                 ERR("Relayd close stream replied error %d", reply.ret_code);
 799         } else {
 800                 /* Success */
 801                 ret = 0;
 802         }
 803
 804         DBG("Relayd close stream id %" PRIu64 " successfully", stream_id);
 805
 806 error:
 807         return ret;
 808 }
 809
 810 /*
 811  * Check for data availability for a given stream id.
 812  *
 813  * Return 0 if NOT pending, 1 if so and a negative value on error.
 814  */
 815 int relayd_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t stream_id,
 816                 uint64_t last_net_seq_num)
 817 {
 818         int ret;
 819         struct lttcomm_relayd_data_pending msg;
 820         struct lttcomm_relayd_generic_reply reply;
 821
 822         /* Code flow error. Safety net. */
 823         assert(rsock);
 824
 825         DBG("Relayd data pending for stream id %" PRIu64, stream_id);
 826
 827         memset(&msg, 0, sizeof(msg));
 828         msg.stream_id = htobe64(stream_id);
 829         msg.last_net_seq_num = htobe64(last_net_seq_num);
 830
 831         /* Send command */
 832         ret = send_command(rsock, RELAYD_DATA_PENDING, (void *) &msg,
 833                         sizeof(msg), 0);
 834         if (ret < 0) {
 835                 goto error;
 836         }
 837
 838         /* Receive response */
 839         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 840         if (ret < 0) {
 841                 goto error;
 842         }
 843
 844         reply.ret_code = be32toh(reply.ret_code);
 845
 846         /* Return session id or negative ret code. */
 847         if (reply.ret_code >= LTTNG_OK) {
 848                 ERR("Relayd data pending replied error %d", reply.ret_code);
 849         }
 850
 851         /* At this point, the ret code is either 1 or 0 */
 852         ret = reply.ret_code;
 853
 854         DBG("Relayd data is %s pending for stream id %" PRIu64,
 855                         ret == 1 ? "" : "NOT", stream_id);
 856
 857 error:
 858         return ret;
 859 }
 860
 861 /*
 862  * Check on the relayd side for a quiescent state on the control socket.
 863  */
 864 int relayd_quiescent_control(struct lttcomm_relayd_sock *rsock,
 865                 uint64_t metadata_stream_id)
 866 {
 867         int ret;
 868         struct lttcomm_relayd_quiescent_control msg;
 869         struct lttcomm_relayd_generic_reply reply;
 870
 871         /* Code flow error. Safety net. */
 872         assert(rsock);
 873
 874         DBG("Relayd checking quiescent control state");
 875
 876         memset(&msg, 0, sizeof(msg));
 877         msg.stream_id = htobe64(metadata_stream_id);
 878
 879         /* Send command */
 880         ret = send_command(rsock, RELAYD_QUIESCENT_CONTROL, &msg, sizeof(msg), 0);
 881         if (ret < 0) {
 882                 goto error;
 883         }
 884
 885         /* Receive response */
 886         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 887         if (ret < 0) {
 888                 goto error;
 889         }
 890
 891         reply.ret_code = be32toh(reply.ret_code);
 892
 893         /* Return session id or negative ret code. */
 894         if (reply.ret_code != LTTNG_OK) {
 895                 ret = -1;
 896                 ERR("Relayd quiescent control replied error %d", reply.ret_code);
 897                 goto error;
 898         }
 899
 900         /* Control socket is quiescent */
 901         return 0;
 902
 903 error:
 904         return ret;
 905 }
 906
 907 /*
 908  * Begin a data pending command for a specific session id.
 909  */
 910 int relayd_begin_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id)
 911 {
 912         int ret;
 913         struct lttcomm_relayd_begin_data_pending msg;
 914         struct lttcomm_relayd_generic_reply reply;
 915
 916         /* Code flow error. Safety net. */
 917         assert(rsock);
 918
 919         DBG("Relayd begin data pending");
 920
 921         memset(&msg, 0, sizeof(msg));
 922         msg.session_id = htobe64(id);
 923
 924         /* Send command */
 925         ret = send_command(rsock, RELAYD_BEGIN_DATA_PENDING, &msg, sizeof(msg), 0);
 926         if (ret < 0) {
 927                 goto error;
 928         }
 929
 930         /* Receive response */
 931         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 932         if (ret < 0) {
 933                 goto error;
 934         }
 935
 936         reply.ret_code = be32toh(reply.ret_code);
 937
 938         /* Return session id or negative ret code. */
 939         if (reply.ret_code != LTTNG_OK) {
 940                 ret = -1;
 941                 ERR("Relayd begin data pending replied error %d", reply.ret_code);
 942                 goto error;
 943         }
 944
 945         return 0;
 946
 947 error:
 948         return ret;
 949 }
 950
 951 /*
 952  * End a data pending command for a specific session id.
 953  *
 954  * Return 0 on success and set is_data_inflight to 0 if no data is being
 955  * streamed or 1 if it is the case.
 956  */
 957 int relayd_end_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id,
 958                 unsigned int *is_data_inflight)
 959 {
 960         int ret, recv_ret;
 961         struct lttcomm_relayd_end_data_pending msg;
 962         struct lttcomm_relayd_generic_reply reply;
 963
 964         /* Code flow error. Safety net. */
 965         assert(rsock);
 966
 967         DBG("Relayd end data pending");
 968
 969         memset(&msg, 0, sizeof(msg));
 970         msg.session_id = htobe64(id);
 971
 972         /* Send command */
 973         ret = send_command(rsock, RELAYD_END_DATA_PENDING, &msg, sizeof(msg), 0);
 974         if (ret < 0) {
 975                 goto error;
 976         }
 977
 978         /* Receive response */
 979         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
 980         if (ret < 0) {
 981                 goto error;
 982         }
 983
 984         recv_ret = be32toh(reply.ret_code);
 985         if (recv_ret < 0) {
 986                 ret = recv_ret;
 987                 goto error;
 988         }
 989
 990         *is_data_inflight = recv_ret;
 991
 992         DBG("Relayd end data pending is data inflight: %d", recv_ret);
 993
 994         return 0;
 995
 996 error:
 997         return ret;
 998 }
 999
1000 /*
1001  * Send index to the relayd.
1002  */
1003 int relayd_send_index(struct lttcomm_relayd_sock *rsock,
1004                 struct ctf_packet_index *index, uint64_t relay_stream_id,
1005                 uint64_t net_seq_num)
1006 {
1007         int ret;
1008         struct lttcomm_relayd_index msg;
1009         struct lttcomm_relayd_generic_reply reply;
1010
1011         /* Code flow error. Safety net. */
1012         assert(rsock);
1013
1014         if (rsock->minor < 4) {
1015                 DBG("Not sending indexes before protocol 2.4");
1016                 ret = 0;
1017                 goto error;
1018         }
1019
1020         DBG("Relayd sending index for stream ID %" PRIu64, relay_stream_id);
1021
1022         memset(&msg, 0, sizeof(msg));
1023         msg.relay_stream_id = htobe64(relay_stream_id);
1024         msg.net_seq_num = htobe64(net_seq_num);
1025
1026         /* The index is already in big endian. */
1027         msg.packet_size = index->packet_size;
1028         msg.content_size = index->content_size;
1029         msg.timestamp_begin = index->timestamp_begin;
1030         msg.timestamp_end = index->timestamp_end;
1031         msg.events_discarded = index->events_discarded;
1032         msg.stream_id = index->stream_id;
1033
1034         if (rsock->minor >= 8) {
1035                 msg.stream_instance_id = index->stream_instance_id;
1036                 msg.packet_seq_num = index->packet_seq_num;
1037         }
1038
1039         /* Send command */
1040         ret = send_command(rsock, RELAYD_SEND_INDEX, &msg,
1041                 lttcomm_relayd_index_len(lttng_to_index_major(rsock->major,
1042                                                                 rsock->minor),
1043                                 lttng_to_index_minor(rsock->major, rsock->minor)),
1044                                 0);
1045         if (ret < 0) {
1046                 goto error;
1047         }
1048
1049         /* Receive response */
1050         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
1051         if (ret < 0) {
1052                 goto error;
1053         }
1054
1055         reply.ret_code = be32toh(reply.ret_code);
1056
1057         /* Return session id or negative ret code. */
1058         if (reply.ret_code != LTTNG_OK) {
1059                 ret = -1;
1060                 ERR("Relayd send index replied error %d", reply.ret_code);
1061         } else {
1062                 /* Success */
1063                 ret = 0;
1064         }
1065
1066 error:
1067         return ret;
1068 }
1069
1070 /*
1071  * Ask the relay to reset the metadata trace file (regeneration).
1072  */
1073 int relayd_reset_metadata(struct lttcomm_relayd_sock *rsock,
1074                 uint64_t stream_id, uint64_t version)
1075 {
1076         int ret;
1077         struct lttcomm_relayd_reset_metadata msg;
1078         struct lttcomm_relayd_generic_reply reply;
1079
1080         /* Code flow error. Safety net. */
1081         assert(rsock);
1082
1083         /* Should have been prevented by the sessiond. */
1084         if (rsock->minor < 8) {
1085                 ERR("Metadata regeneration unsupported before 2.8");
1086                 ret = -1;
1087                 goto error;
1088         }
1089
1090         DBG("Relayd reset metadata stream id %" PRIu64, stream_id);
1091
1092         memset(&msg, 0, sizeof(msg));
1093         msg.stream_id = htobe64(stream_id);
1094         msg.version = htobe64(version);
1095
1096         /* Send command */
1097         ret = send_command(rsock, RELAYD_RESET_METADATA, (void *) &msg, sizeof(msg), 0);
1098         if (ret < 0) {
1099                 goto error;
1100         }
1101
1102         /* Receive response */
1103         ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
1104         if (ret < 0) {
1105                 goto error;
1106         }
1107
1108         reply.ret_code = be32toh(reply.ret_code);
1109
1110         /* Return session id or negative ret code. */
1111         if (reply.ret_code != LTTNG_OK) {
1112                 ret = -1;
1113                 ERR("Relayd reset metadata replied error %d", reply.ret_code);
1114         } else {
1115                 /* Success */
1116                 ret = 0;
1117         }
1118
1119         DBG("Relayd reset metadata stream id %" PRIu64 " successfully", stream_id);
1120
1121 error:
1122         return ret;
1123 }
1124
1125 int relayd_rotate_streams(struct lttcomm_relayd_sock *sock,
1126                 unsigned int stream_count, uint64_t *new_chunk_id,
1127                 const struct relayd_stream_rotation_position *positions)
1128 {
1129         int ret;
1130         unsigned int i;
1131         struct lttng_dynamic_buffer payload;
1132         struct lttcomm_relayd_generic_reply reply = {};
1133         const struct lttcomm_relayd_rotate_streams msg = {
1134                 .stream_count = htobe32((uint32_t) stream_count),
1135                 .new_chunk_id = (typeof(msg.new_chunk_id)) {
1136                         .is_set = !!new_chunk_id,
1137                         .value = htobe64(new_chunk_id ? *new_chunk_id : 0),
1138                 },
1139         };
1140         char new_chunk_id_buf[MAX_INT_DEC_LEN(*new_chunk_id)] = {};
1141         const char *new_chunk_id_str;
1142
1143         lttng_dynamic_buffer_init(&payload);
1144
1145         /* Code flow error. Safety net. */
1146         assert(sock);
1147
1148         if (new_chunk_id) {
1149                 ret = snprintf(new_chunk_id_buf, sizeof(new_chunk_id_buf),
1150                                 "%" PRIu64, *new_chunk_id);
1151                 if (ret == -1 || ret >= sizeof(new_chunk_id_buf)) {
1152                         new_chunk_id_str = "formatting error";
1153                 } else {
1154                         new_chunk_id_str = new_chunk_id_buf;
1155                 }
1156         } else {
1157                 new_chunk_id_str = "none";
1158         }
1159
1160         DBG("Preparing \"rotate streams\" command payload: new_chunk_id = %s, stream_count = %u",
1161                         new_chunk_id_str, stream_count);
1162
1163         ret = lttng_dynamic_buffer_append(&payload, &msg, sizeof(msg));
1164         if (ret) {
1165                 ERR("Failed to allocate \"rotate streams\" command payload");
1166                 goto error;
1167         }
1168
1169         for (i = 0; i < stream_count; i++) {
1170                 const struct relayd_stream_rotation_position *position =
1171                                 &positions[i];
1172                 const struct lttcomm_relayd_stream_rotation_position comm_position = {
1173                         .stream_id = htobe64(position->stream_id),
1174                         .rotate_at_seq_num = htobe64(
1175                                         position->rotate_at_seq_num),
1176                 };
1177
1178                 DBG("Rotate stream %" PRIu64 "at sequence number %" PRIu64,
1179                                 position->stream_id,
1180                                 position->rotate_at_seq_num);
1181                 ret = lttng_dynamic_buffer_append(&payload, &comm_position,
1182                                 sizeof(comm_position));
1183                 if (ret) {
1184                         ERR("Failed to allocate \"rotate streams\" command payload");
1185                         goto error;
1186                 }
1187         }
1188
1189         /* Send command. */
1190         ret = send_command(sock, RELAYD_ROTATE_STREAMS, payload.data,
1191                         payload.size, 0);
1192         if (ret < 0) {
1193                 ERR("Failed to send \"rotate stream\" command");
1194                 goto error;
1195         }
1196
1197         /* Receive response. */
1198         ret = recv_reply(sock, &reply, sizeof(reply));
1199         if (ret < 0) {
1200                 ERR("Failed to receive \"rotate streams\" command reply");
1201                 goto error;
1202         }
1203
1204         reply.ret_code = be32toh(reply.ret_code);
1205         if (reply.ret_code != LTTNG_OK) {
1206                 ret = -1;
1207                 ERR("Relayd rotate streams replied error %d", reply.ret_code);
1208         } else {
1209                 /* Success. */
1210                 ret = 0;
1211                 DBG("Relayd rotated streams successfully");
1212         }
1213
1214 error:
1215         lttng_dynamic_buffer_reset(&payload);
1216         return ret;
1217 }
1218
1219 int relayd_create_trace_chunk(struct lttcomm_relayd_sock *sock,
1220                 struct lttng_trace_chunk *chunk)
1221 {
1222         int ret = 0;
1223         enum lttng_trace_chunk_status status;
1224         struct lttcomm_relayd_create_trace_chunk msg = {};
1225         struct lttcomm_relayd_generic_reply reply = {};
1226         struct lttng_dynamic_buffer payload;
1227         uint64_t chunk_id;
1228         time_t creation_timestamp;
1229         const char *chunk_name;
1230         size_t chunk_name_length;
1231         bool overriden_name;
1232
1233         lttng_dynamic_buffer_init(&payload);
1234
1235         status = lttng_trace_chunk_get_id(chunk, &chunk_id);
1236         if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1237                 ret = -1;
1238                 goto end;
1239         }
1240
1241         status = lttng_trace_chunk_get_creation_timestamp(
1242                         chunk, &creation_timestamp);
1243         if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1244                 ret = -1;
1245                 goto end;
1246         }
1247
1248         status = lttng_trace_chunk_get_name(
1249                         chunk, &chunk_name, &overriden_name);
1250         if (status != LTTNG_TRACE_CHUNK_STATUS_OK &&
1251                         status != LTTNG_TRACE_CHUNK_STATUS_NONE) {
1252                 ret = -1;
1253                 goto end;
1254         }
1255
1256         chunk_name_length = overriden_name ? (strlen(chunk_name) + 1) : 0;
1257         msg = (typeof(msg)){
1258                 .chunk_id = htobe64(chunk_id),
1259                 .creation_timestamp = htobe64((uint64_t) creation_timestamp),
1260                 .override_name_length = htobe32((uint32_t) chunk_name_length),
1261         };
1262
1263         ret = lttng_dynamic_buffer_append(&payload, &msg, sizeof(msg));
1264         if (ret) {
1265                 goto end;
1266         }
1267         if (chunk_name_length) {
1268                 ret = lttng_dynamic_buffer_append(
1269                                 &payload, chunk_name, chunk_name_length);
1270                 if (ret) {
1271                         goto end;
1272                 }
1273         }
1274
1275         ret = send_command(sock, RELAYD_CREATE_TRACE_CHUNK, payload.data,
1276                         payload.size, 0);
1277         if (ret < 0) {
1278                 ERR("Failed to send trace chunk creation command to relay daemon");
1279                 goto end;
1280         }
1281
1282         ret = recv_reply(sock, &reply, sizeof(reply));
1283         if (ret < 0) {
1284                 ERR("Failed to receive relay daemon trace chunk creation command reply");
1285                 goto end;
1286         }
1287
1288         reply.ret_code = be32toh(reply.ret_code);
1289         if (reply.ret_code != LTTNG_OK) {
1290                 ret = -1;
1291                 ERR("Relayd trace chunk create replied error %d",
1292                                 reply.ret_code);
1293         } else {
1294                 ret = 0;
1295                 DBG("Relayd successfully created trace chunk: chunk_id = %" PRIu64,
1296                                 chunk_id);
1297         }
1298
1299 end:
1300         lttng_dynamic_buffer_reset(&payload);
1301         return ret;
1302 }
1303
1304 int relayd_close_trace_chunk(struct lttcomm_relayd_sock *sock,
1305                 struct lttng_trace_chunk *chunk)
1306 {
1307         int ret = 0;
1308         enum lttng_trace_chunk_status status;
1309         struct lttcomm_relayd_close_trace_chunk msg = {};
1310         struct lttcomm_relayd_generic_reply reply = {};
1311         uint64_t chunk_id;
1312         time_t close_timestamp;
1313         LTTNG_OPTIONAL(enum lttng_trace_chunk_command_type) close_command = {};
1314
1315         status = lttng_trace_chunk_get_id(chunk, &chunk_id);
1316         if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1317                 ERR("Failed to get trace chunk id");
1318                 ret = -1;
1319                 goto end;
1320         }
1321
1322         status = lttng_trace_chunk_get_close_timestamp(chunk, &close_timestamp);
1323         if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1324                 ERR("Failed to get trace chunk close timestamp");
1325                 ret = -1;
1326                 goto end;
1327         }
1328
1329         status = lttng_trace_chunk_get_close_command(chunk,
1330                         &close_command.value);
1331         switch (status) {
1332         case LTTNG_TRACE_CHUNK_STATUS_OK:
1333                 close_command.is_set = 1;
1334                 break;
1335         case LTTNG_TRACE_CHUNK_STATUS_NONE:
1336                 break;
1337         default:
1338                 ERR("Failed to get trace chunk close command");
1339                 ret = -1;
1340                 goto end;
1341         }
1342
1343         msg = (typeof(msg)){
1344                 .chunk_id = htobe64(chunk_id),
1345                 .close_timestamp = htobe64((uint64_t) close_timestamp),
1346                 .close_command = {
1347                         .value = htobe32((uint32_t) close_command.value),
1348                         .is_set = close_command.is_set,
1349                 },
1350         };
1351
1352         ret = send_command(sock, RELAYD_CLOSE_TRACE_CHUNK, &msg, sizeof(msg),
1353                         0);
1354         if (ret < 0) {
1355                 ERR("Failed to send trace chunk close command to relay daemon");
1356                 goto end;
1357         }
1358
1359         ret = recv_reply(sock, &reply, sizeof(reply));
1360         if (ret < 0) {
1361                 ERR("Failed to receive relay daemon trace chunk close command reply");
1362                 goto end;
1363         }
1364
1365         reply.ret_code = be32toh(reply.ret_code);
1366         if (reply.ret_code != LTTNG_OK) {
1367                 ret = -1;
1368                 ERR("Relayd trace chunk close replied error %d",
1369                                 reply.ret_code);
1370         } else {
1371                 ret = 0;
1372                 DBG("Relayd successfully closed trace chunk: chunk_id = %" PRIu64,
1373                                 chunk_id);
1374         }
1375 end:
1376         return ret;
1377 }
1378
1379 int relayd_trace_chunk_exists(struct lttcomm_relayd_sock *sock,
1380                 uint64_t chunk_id, bool *chunk_exists)
1381 {
1382         int ret = 0;
1383         struct lttcomm_relayd_trace_chunk_exists msg = {};
1384         struct lttcomm_relayd_trace_chunk_exists_reply reply = {};
1385
1386         msg = (typeof(msg)){
1387                         .chunk_id = htobe64(chunk_id),
1388         };
1389
1390         ret = send_command(sock, RELAYD_TRACE_CHUNK_EXISTS, &msg, sizeof(msg),
1391                         0);
1392         if (ret < 0) {
1393                 ERR("Failed to send trace chunk exists command to relay daemon");
1394                 goto end;
1395         }
1396
1397         ret = recv_reply(sock, &reply, sizeof(reply));
1398         if (ret < 0) {
1399                 ERR("Failed to receive relay daemon trace chunk close command reply");
1400                 goto end;
1401         }
1402
1403         reply.generic.ret_code = be32toh(reply.generic.ret_code);
1404         if (reply.generic.ret_code != LTTNG_OK) {
1405                 ret = -1;
1406                 ERR("Relayd trace chunk close replied error %d",
1407                                 reply.generic.ret_code);
1408         } else {
1409                 ret = 0;
1410                 DBG("Relayd successfully checked trace chunk existence: chunk_id = %" PRIu64
1411                                 ", exists = %s", chunk_id,
1412                                 reply.trace_chunk_exists ? "true" : "false");
1413                 *chunk_exists = !!reply.trace_chunk_exists;
1414         }
1415 end:
1416         return ret;
1417 }