src/bin/lttng-relayd/connection.c

   1 /*
   2  * Copyright (C) 2013 Julien Desfossez <jdesfossez@efficios.com>
   3  * Copyright (C) 2013 David Goulet <dgoulet@efficios.com>
   4  * Copyright (C) 2015 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   5  *
   6  * SPDX-License-Identifier: GPL-2.0-only
   7  *
   8  */
   9
  10 #define _LGPL_SOURCE
  11 #include <common/common.h>
  12 #include <urcu/rculist.h>
  13
  14 #include "connection.h"
  15 #include "stream.h"
  16 #include "viewer-session.h"
  17
  18 bool connection_get(struct relay_connection *conn)
  19 {
  20         return urcu_ref_get_unless_zero(&conn->ref);
  21 }
  22
  23 struct relay_connection *connection_get_by_sock(struct lttng_ht *relay_connections_ht,
  24                 int sock)
  25 {
  26         struct lttng_ht_node_ulong *node;
  27         struct lttng_ht_iter iter;
  28         struct relay_connection *conn = NULL;
  29
  30         assert(sock >= 0);
  31
  32         rcu_read_lock();
  33         lttng_ht_lookup(relay_connections_ht, (void *)((unsigned long) sock),
  34                         &iter);
  35         node = lttng_ht_iter_get_node_ulong(&iter);
  36         if (!node) {
  37                 DBG2("Relay connection by sock %d not found", sock);
  38                 goto end;
  39         }
  40         conn = caa_container_of(node, struct relay_connection, sock_n);
  41         if (!connection_get(conn)) {
  42                 conn = NULL;
  43         }
  44 end:
  45         rcu_read_unlock();
  46         return conn;
  47 }
  48
  49 int connection_reset_protocol_state(struct relay_connection *connection)
  50 {
  51         int ret = 0;
  52
  53         switch (connection->type) {
  54         case RELAY_DATA:
  55                 connection->protocol.data.state_id =
  56                                 DATA_CONNECTION_STATE_RECEIVE_HEADER;
  57                 memset(&connection->protocol.data.state.receive_header,
  58                                 0,
  59                                 sizeof(connection->protocol.data.state.receive_header));
  60                 connection->protocol.data.state.receive_header.left_to_receive =
  61                                 sizeof(struct lttcomm_relayd_data_hdr);
  62                 break;
  63         case RELAY_CONTROL:
  64                 connection->protocol.ctrl.state_id =
  65                                 CTRL_CONNECTION_STATE_RECEIVE_HEADER;
  66                 memset(&connection->protocol.ctrl.state.receive_header,
  67                                 0,
  68                                 sizeof(connection->protocol.ctrl.state.receive_header));
  69                 connection->protocol.data.state.receive_header.left_to_receive =
  70                                 sizeof(struct lttcomm_relayd_hdr);
  71                 ret = lttng_dynamic_buffer_set_size(
  72                                 &connection->protocol.ctrl.reception_buffer,
  73                                 sizeof(struct lttcomm_relayd_hdr));
  74                 if (ret) {
  75                         ERR("Failed to reinitialize control connection reception buffer size to %zu bytes.", sizeof(struct lttcomm_relayd_hdr));
  76                         goto end;
  77                 }
  78                 break;
  79         default:
  80                 goto end;
  81         }
  82         DBG("Reset communication state of relay connection (fd = %i)",
  83                         connection->sock->fd);
  84 end:
  85         return ret;
  86 }
  87
  88 struct relay_connection *connection_create(struct lttcomm_sock *sock,
  89                 enum connection_type type)
  90 {
  91         struct relay_connection *conn;
  92
  93         conn = zmalloc(sizeof(*conn));
  94         if (!conn) {
  95                 PERROR("zmalloc relay connection");
  96                 goto end;
  97         }
  98         urcu_ref_init(&conn->ref);
  99         conn->type = type;
 100         conn->sock = sock;
 101         lttng_ht_node_init_ulong(&conn->sock_n, (unsigned long) conn->sock->fd);
 102         if (conn->type == RELAY_CONTROL) {
 103                 lttng_dynamic_buffer_init(&conn->protocol.ctrl.reception_buffer);
 104         }
 105         connection_reset_protocol_state(conn);
 106 end:
 107         return conn;
 108 }
 109
 110 static void rcu_free_connection(struct rcu_head *head)
 111 {
 112         struct relay_connection *conn =
 113                 caa_container_of(head, struct relay_connection, rcu_node);
 114
 115         lttcomm_destroy_sock(conn->sock);
 116         if (conn->viewer_session) {
 117                 viewer_session_destroy(conn->viewer_session);
 118                 conn->viewer_session = NULL;
 119         }
 120         if (conn->type == RELAY_CONTROL) {
 121                 lttng_dynamic_buffer_reset(
 122                                 &conn->protocol.ctrl.reception_buffer);
 123         }
 124         free(conn);
 125 }
 126
 127 static void destroy_connection(struct relay_connection *conn)
 128 {
 129         call_rcu(&conn->rcu_node, rcu_free_connection);
 130 }
 131
 132 static void connection_release(struct urcu_ref *ref)
 133 {
 134         struct relay_connection *conn =
 135                 caa_container_of(ref, struct relay_connection, ref);
 136
 137         if (conn->in_socket_ht) {
 138                 struct lttng_ht_iter iter;
 139                 int ret;
 140
 141                 iter.iter.node = &conn->sock_n.node;
 142                 ret = lttng_ht_del(conn->socket_ht, &iter);
 143                 assert(!ret);
 144         }
 145
 146         if (conn->session) {
 147                 if (session_close(conn->session)) {
 148                         ERR("session_close");
 149                 }
 150                 conn->session = NULL;
 151         }
 152         if (conn->viewer_session) {
 153                 viewer_session_close(conn->viewer_session);
 154         }
 155         destroy_connection(conn);
 156 }
 157
 158 void connection_put(struct relay_connection *conn)
 159 {
 160         rcu_read_lock();
 161         urcu_ref_put(&conn->ref, connection_release);
 162         rcu_read_unlock();
 163 }
 164
 165 void connection_ht_add(struct lttng_ht *relay_connections_ht,
 166                 struct relay_connection *conn)
 167 {
 168         assert(!conn->in_socket_ht);
 169         lttng_ht_add_unique_ulong(relay_connections_ht, &conn->sock_n);
 170         conn->in_socket_ht = 1;
 171         conn->socket_ht = relay_connections_ht;
 172 }
 173
 174 int connection_set_session(struct relay_connection *conn,
 175                 struct relay_session *session)
 176 {
 177         int ret = 0;
 178
 179         assert(conn);
 180         assert(session);
 181         assert(!conn->session);
 182
 183         if (connection_get(conn)) {
 184                 if (session_get(session)) {
 185                         conn->session = session;
 186                 } else {
 187                         ERR("Failed to get session reference in connection_set_session()");
 188                         ret = -1;
 189                 }
 190                 connection_put(conn);
 191         } else {
 192                 ERR("Failed to get connection reference in connection_set_session()");
 193                 ret = -1;
 194         }
 195         return ret;
 196 }