src/bin/lttng-relayd/session.c

   1 /*
   2  * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
   3  *                      David Goulet <dgoulet@efficios.com>
   4  *               2015 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   5  *
   6  * This program is free software; you can redistribute it and/or modify it
   7  * under the terms of the GNU General Public License, version 2 only, as
   8  * published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but WITHOUT
  11  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  12  * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
  13  * more details.
  14  *
  15  * You should have received a copy of the GNU General Public License along with
  16  * this program; if not, write to the Free Software Foundation, Inc., 51
  17  * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  18  */
  19
  20 #define _LGPL_SOURCE
  21 #include <common/common.h>
  22 #include <common/compat/uuid.h>
  23 #include <urcu/rculist.h>
  24
  25 #include "lttng-relayd.h"
  26 #include "ctf-trace.h"
  27 #include "session.h"
  28 #include "stream.h"
  29 #include "sessiond-trace-chunks.h"
  30
  31 /* Global session id used in the session creation. */
  32 static uint64_t last_relay_session_id;
  33 static pthread_mutex_t last_relay_session_id_lock = PTHREAD_MUTEX_INITIALIZER;
  34
  35 static int session_set_anonymous_chunk(struct relay_session *session)
  36 {
  37         int ret = 0;
  38         struct lttng_trace_chunk *chunk = NULL;
  39         enum lttng_trace_chunk_status status;
  40         struct lttng_directory_handle output_directory;
  41
  42         ret = lttng_directory_handle_init(&output_directory, opt_output_path);
  43         if (ret) {
  44                 goto end;
  45         }
  46
  47         chunk = lttng_trace_chunk_create_anonymous();
  48         if (!chunk) {
  49                 goto end;
  50         }
  51
  52         status = lttng_trace_chunk_set_credentials_current_user(chunk);
  53         if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
  54                 ret = -1;
  55                 goto end;
  56         }
  57
  58         status = lttng_trace_chunk_set_as_owner(chunk, &output_directory);
  59         if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
  60                 ret = -1;
  61                 goto end;
  62         }
  63         session->current_trace_chunk = chunk;
  64         chunk = NULL;
  65 end:
  66         lttng_trace_chunk_put(chunk);
  67         lttng_directory_handle_fini(&output_directory);
  68         return ret;
  69 }
  70
  71 /*
  72  * Create a new session by assigning a new session ID.
  73  *
  74  * Return allocated session or else NULL.
  75  */
  76 struct relay_session *session_create(const char *session_name,
  77                 const char *hostname,
  78                 uint32_t live_timer,
  79                 bool snapshot,
  80                 const lttng_uuid sessiond_uuid,
  81                 const uint64_t *id_sessiond,
  82                 const uint64_t *current_chunk_id,
  83                 const time_t *creation_time,
  84                 uint32_t major,
  85                 uint32_t minor)
  86 {
  87         int ret;
  88         struct relay_session *session;
  89
  90         session = zmalloc(sizeof(*session));
  91         if (!session) {
  92                 PERROR("Failed to allocate session");
  93                 goto error;
  94         }
  95         if (lttng_strncpy(session->session_name, session_name,
  96                         sizeof(session->session_name))) {
  97                 WARN("Session name exceeds maximal allowed length");
  98                 goto error;
  99         }
 100         if (lttng_strncpy(session->hostname, hostname,
 101                         sizeof(session->hostname))) {
 102                 WARN("Hostname exceeds maximal allowed length");
 103                 goto error;
 104         }
 105         session->ctf_traces_ht = lttng_ht_new(0, LTTNG_HT_TYPE_STRING);
 106         if (!session->ctf_traces_ht) {
 107                 goto error;
 108         }
 109
 110         pthread_mutex_lock(&last_relay_session_id_lock);
 111         session->id = ++last_relay_session_id;
 112         pthread_mutex_unlock(&last_relay_session_id_lock);
 113
 114         session->major = major;
 115         session->minor = minor;
 116         lttng_ht_node_init_u64(&session->session_n, session->id);
 117         urcu_ref_init(&session->ref);
 118         CDS_INIT_LIST_HEAD(&session->recv_list);
 119         pthread_mutex_init(&session->lock, NULL);
 120         pthread_mutex_init(&session->recv_list_lock, NULL);
 121
 122         session->live_timer = live_timer;
 123         session->snapshot = snapshot;
 124         lttng_uuid_copy(session->sessiond_uuid, sessiond_uuid);
 125
 126         if (id_sessiond) {
 127                 LTTNG_OPTIONAL_SET(&session->id_sessiond, *id_sessiond);
 128         }
 129
 130         ret = sessiond_trace_chunk_registry_session_created(
 131                         sessiond_trace_chunk_registry, sessiond_uuid);
 132         if (ret) {
 133                 goto error;
 134         }
 135
 136         if (id_sessiond && current_chunk_id) {
 137                 session->current_trace_chunk =
 138                                 sessiond_trace_chunk_registry_get_chunk(
 139                                         sessiond_trace_chunk_registry,
 140                                         session->sessiond_uuid,
 141                                         session->id_sessiond.value,
 142                                         *current_chunk_id);
 143                 if (!session->current_trace_chunk) {
 144                         char uuid_str[UUID_STR_LEN];
 145
 146                         lttng_uuid_to_str(sessiond_uuid, uuid_str);
 147                         ERR("Could not find trace chunk: sessiond = {%s}, sessiond session id = %" PRIu64 ", trace chunk id = %" PRIu64,
 148                                         uuid_str, *id_sessiond,
 149                                         *current_chunk_id);
 150                 }
 151         } else if (!id_sessiond) {
 152                 /*
 153                  * Pre-2.11 peers will not announce trace chunks. An
 154                  * anonymous trace chunk which will remain set for the
 155                  * duration of the session is created.
 156                  */
 157                 ret = session_set_anonymous_chunk(session);
 158                 if (ret) {
 159                         goto error;
 160                 }
 161         }
 162
 163         lttng_ht_add_unique_u64(sessions_ht, &session->session_n);
 164         if (creation_time) {
 165                 LTTNG_OPTIONAL_SET(&session->creation_time, *creation_time);
 166         }
 167         return session;
 168
 169 error:
 170         session_put(session);
 171         return NULL;
 172 }
 173
 174 /* Should be called with RCU read-side lock held. */
 175 bool session_get(struct relay_session *session)
 176 {
 177         return urcu_ref_get_unless_zero(&session->ref);
 178 }
 179
 180 /*
 181  * Lookup a session within the session hash table using the session id
 182  * as key. A session reference is taken when a session is returned.
 183  * session_put() must be called on that session.
 184  *
 185  * Return session or NULL if not found.
 186  */
 187 struct relay_session *session_get_by_id(uint64_t id)
 188 {
 189         struct relay_session *session = NULL;
 190         struct lttng_ht_node_u64 *node;
 191         struct lttng_ht_iter iter;
 192
 193         rcu_read_lock();
 194         lttng_ht_lookup(sessions_ht, &id, &iter);
 195         node = lttng_ht_iter_get_node_u64(&iter);
 196         if (!node) {
 197                 DBG("Session find by ID %" PRIu64 " id NOT found", id);
 198                 goto end;
 199         }
 200         session = caa_container_of(node, struct relay_session, session_n);
 201         DBG("Session find by ID %" PRIu64 " id found", id);
 202         if (!session_get(session)) {
 203                 session = NULL;
 204         }
 205 end:
 206         rcu_read_unlock();
 207         return session;
 208 }
 209
 210 static void rcu_destroy_session(struct rcu_head *rcu_head)
 211 {
 212         struct relay_session *session =
 213                         caa_container_of(rcu_head, struct relay_session,
 214                                 rcu_node);
 215         /*
 216          * Since each trace has a reference on the session, it means
 217          * that if we are at the point where we teardown the session, no
 218          * trace belonging to that session exist at this point.
 219          * Calling lttng_ht_destroy in call_rcu worker thread so we
 220          * don't hold the RCU read-side lock while calling it.
 221          */
 222         lttng_ht_destroy(session->ctf_traces_ht);
 223         free(session);
 224 }
 225
 226 /*
 227  * Delete session from the given hash table.
 228  *
 229  * Return lttng ht del error code being 0 on success and 1 on failure.
 230  */
 231 static int session_delete(struct relay_session *session)
 232 {
 233         struct lttng_ht_iter iter;
 234
 235         iter.iter.node = &session->session_n.node;
 236         return lttng_ht_del(sessions_ht, &iter);
 237 }
 238
 239
 240 static void destroy_session(struct relay_session *session)
 241 {
 242         int ret;
 243
 244         ret = session_delete(session);
 245         assert(!ret);
 246         lttng_trace_chunk_put(session->current_trace_chunk);
 247         session->current_trace_chunk = NULL;
 248         ret = sessiond_trace_chunk_registry_session_destroyed(
 249                         sessiond_trace_chunk_registry, session->sessiond_uuid);
 250         assert(!ret);
 251         call_rcu(&session->rcu_node, rcu_destroy_session);
 252 }
 253
 254 void session_release(struct urcu_ref *ref)
 255 {
 256         struct relay_session *session =
 257                         caa_container_of(ref, struct relay_session, ref);
 258
 259         destroy_session(session);
 260 }
 261
 262 void session_put(struct relay_session *session)
 263 {
 264         rcu_read_lock();
 265         urcu_ref_put(&session->ref, session_release);
 266         rcu_read_unlock();
 267 }
 268
 269 int session_close(struct relay_session *session)
 270 {
 271         int ret = 0;
 272         struct ctf_trace *trace;
 273         struct lttng_ht_iter iter;
 274         struct relay_stream *stream;
 275
 276         pthread_mutex_lock(&session->lock);
 277         DBG("closing session %" PRIu64 ": is conn already closed %d",
 278                         session->id, session->connection_closed);
 279         session->connection_closed = true;
 280         pthread_mutex_unlock(&session->lock);
 281
 282         rcu_read_lock();
 283         cds_lfht_for_each_entry(session->ctf_traces_ht->ht,
 284                         &iter.iter, trace, node.node) {
 285                 ret = ctf_trace_close(trace);
 286                 if (ret) {
 287                         goto rcu_unlock;
 288                 }
 289         }
 290         cds_list_for_each_entry_rcu(stream, &session->recv_list,
 291                         recv_node) {
 292                 /* Close streams which have not been published yet. */
 293                 try_stream_close(stream);
 294         }
 295 rcu_unlock:
 296         rcu_read_unlock();
 297         if (ret) {
 298                 return ret;
 299         }
 300         /* Put self-reference from create. */
 301         session_put(session);
 302         return ret;
 303 }
 304
 305 int session_abort(struct relay_session *session)
 306 {
 307         int ret = 0;
 308
 309         if (!session) {
 310                 return 0;
 311         }
 312
 313         pthread_mutex_lock(&session->lock);
 314         DBG("aborting session %" PRIu64, session->id);
 315         session->aborted = true;
 316         pthread_mutex_unlock(&session->lock);
 317         return ret;
 318 }
 319
 320 void print_sessions(void)
 321 {
 322         struct lttng_ht_iter iter;
 323         struct relay_session *session;
 324
 325         if (!sessions_ht) {
 326                 return;
 327         }
 328
 329         rcu_read_lock();
 330         cds_lfht_for_each_entry(sessions_ht->ht, &iter.iter, session,
 331                         session_n.node) {
 332                 if (!session_get(session)) {
 333                         continue;
 334                 }
 335                 DBG("session %p refcount %ld session %" PRIu64,
 336                         session,
 337                         session->ref.refcount,
 338                         session->id);
 339                 session_put(session);
 340         }
 341         rcu_read_unlock();
 342 }