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 <urcu/rculist.h>
  23
  24 #include "lttng-relayd.h"
  25 #include "ctf-trace.h"
  26 #include "session.h"
  27 #include "stream.h"
  28 #include "sessiond-trace-chunks.h"
  29
  30 /* Global session id used in the session creation. */
  31 static uint64_t last_relay_session_id;
  32 static pthread_mutex_t last_relay_session_id_lock = PTHREAD_MUTEX_INITIALIZER;
  33
  34 /*
  35  * Create a new session by assigning a new session ID.
  36  *
  37  * Return allocated session or else NULL.
  38  */
  39 struct relay_session *session_create(const char *session_name,
  40                 const char *hostname, uint32_t live_timer,
  41                 bool snapshot, const lttng_uuid sessiond_uuid,
  42                 uint32_t major, uint32_t minor)
  43 {
  44         int ret;
  45         struct relay_session *session;
  46
  47         session = zmalloc(sizeof(*session));
  48         if (!session) {
  49                 PERROR("relay session zmalloc");
  50                 goto error;
  51         }
  52         if (lttng_strncpy(session->session_name, session_name,
  53                         sizeof(session->session_name))) {
  54                 goto error;
  55         }
  56         if (lttng_strncpy(session->hostname, hostname,
  57                         sizeof(session->hostname))) {
  58                 goto error;
  59         }
  60         session->ctf_traces_ht = lttng_ht_new(0, LTTNG_HT_TYPE_STRING);
  61         if (!session->ctf_traces_ht) {
  62                 goto error;
  63         }
  64
  65         pthread_mutex_lock(&last_relay_session_id_lock);
  66         session->id = ++last_relay_session_id;
  67         pthread_mutex_unlock(&last_relay_session_id_lock);
  68
  69         session->major = major;
  70         session->minor = minor;
  71         lttng_ht_node_init_u64(&session->session_n, session->id);
  72         urcu_ref_init(&session->ref);
  73         CDS_INIT_LIST_HEAD(&session->recv_list);
  74         pthread_mutex_init(&session->lock, NULL);
  75         pthread_mutex_init(&session->recv_list_lock, NULL);
  76
  77         session->live_timer = live_timer;
  78         session->snapshot = snapshot;
  79         lttng_uuid_copy(session->sessiond_uuid, sessiond_uuid);
  80
  81         ret = sessiond_trace_chunk_registry_session_created(
  82                         sessiond_trace_chunk_registry, sessiond_uuid);
  83         if (ret) {
  84                 goto error;
  85         }
  86
  87         lttng_ht_add_unique_u64(sessions_ht, &session->session_n);
  88         return session;
  89
  90 error:
  91         free(session);
  92         return NULL;
  93 }
  94
  95 /* Should be called with RCU read-side lock held. */
  96 bool session_get(struct relay_session *session)
  97 {
  98         return urcu_ref_get_unless_zero(&session->ref);
  99 }
 100
 101 /*
 102  * Lookup a session within the session hash table using the session id
 103  * as key. A session reference is taken when a session is returned.
 104  * session_put() must be called on that session.
 105  *
 106  * Return session or NULL if not found.
 107  */
 108 struct relay_session *session_get_by_id(uint64_t id)
 109 {
 110         struct relay_session *session = NULL;
 111         struct lttng_ht_node_u64 *node;
 112         struct lttng_ht_iter iter;
 113
 114         rcu_read_lock();
 115         lttng_ht_lookup(sessions_ht, &id, &iter);
 116         node = lttng_ht_iter_get_node_u64(&iter);
 117         if (!node) {
 118                 DBG("Session find by ID %" PRIu64 " id NOT found", id);
 119                 goto end;
 120         }
 121         session = caa_container_of(node, struct relay_session, session_n);
 122         DBG("Session find by ID %" PRIu64 " id found", id);
 123         if (!session_get(session)) {
 124                 session = NULL;
 125         }
 126 end:
 127         rcu_read_unlock();
 128         return session;
 129 }
 130
 131 static void rcu_destroy_session(struct rcu_head *rcu_head)
 132 {
 133         struct relay_session *session =
 134                         caa_container_of(rcu_head, struct relay_session,
 135                                 rcu_node);
 136         /*
 137          * Since each trace has a reference on the session, it means
 138          * that if we are at the point where we teardown the session, no
 139          * trace belonging to that session exist at this point.
 140          * Calling lttng_ht_destroy in call_rcu worker thread so we
 141          * don't hold the RCU read-side lock while calling it.
 142          */
 143         lttng_ht_destroy(session->ctf_traces_ht);
 144         free(session);
 145 }
 146
 147 /*
 148  * Delete session from the given hash table.
 149  *
 150  * Return lttng ht del error code being 0 on success and 1 on failure.
 151  */
 152 static int session_delete(struct relay_session *session)
 153 {
 154         struct lttng_ht_iter iter;
 155
 156         iter.iter.node = &session->session_n.node;
 157         return lttng_ht_del(sessions_ht, &iter);
 158 }
 159
 160
 161 static void destroy_session(struct relay_session *session)
 162 {
 163         int ret;
 164
 165         ret = session_delete(session);
 166         assert(!ret);
 167         lttng_trace_chunk_put(session->current_trace_chunk);
 168         ret = sessiond_trace_chunk_registry_session_destroyed(
 169                         sessiond_trace_chunk_registry, session->sessiond_uuid);
 170         assert(!ret);
 171         lttng_trace_chunk_put(session->current_trace_chunk);
 172         session->current_trace_chunk = NULL;
 173         call_rcu(&session->rcu_node, rcu_destroy_session);
 174 }
 175
 176 void session_release(struct urcu_ref *ref)
 177 {
 178         struct relay_session *session =
 179                         caa_container_of(ref, struct relay_session, ref);
 180
 181         destroy_session(session);
 182 }
 183
 184 void session_put(struct relay_session *session)
 185 {
 186         rcu_read_lock();
 187         urcu_ref_put(&session->ref, session_release);
 188         rcu_read_unlock();
 189 }
 190
 191 int session_close(struct relay_session *session)
 192 {
 193         int ret = 0;
 194         struct ctf_trace *trace;
 195         struct lttng_ht_iter iter;
 196         struct relay_stream *stream;
 197
 198         pthread_mutex_lock(&session->lock);
 199         DBG("closing session %" PRIu64 ": is conn already closed %d",
 200                         session->id, session->connection_closed);
 201         session->connection_closed = true;
 202         pthread_mutex_unlock(&session->lock);
 203
 204         rcu_read_lock();
 205         cds_lfht_for_each_entry(session->ctf_traces_ht->ht,
 206                         &iter.iter, trace, node.node) {
 207                 ret = ctf_trace_close(trace);
 208                 if (ret) {
 209                         goto rcu_unlock;
 210                 }
 211         }
 212         cds_list_for_each_entry_rcu(stream, &session->recv_list,
 213                         recv_node) {
 214                 /* Close streams which have not been published yet. */
 215                 try_stream_close(stream);
 216         }
 217 rcu_unlock:
 218         rcu_read_unlock();
 219         if (ret) {
 220                 return ret;
 221         }
 222         /* Put self-reference from create. */
 223         session_put(session);
 224         return ret;
 225 }
 226
 227 int session_abort(struct relay_session *session)
 228 {
 229         int ret = 0;
 230
 231         if (!session) {
 232                 return 0;
 233         }
 234
 235         pthread_mutex_lock(&session->lock);
 236         DBG("aborting session %" PRIu64, session->id);
 237         session->aborted = true;
 238         pthread_mutex_unlock(&session->lock);
 239         return ret;
 240 }
 241
 242 void print_sessions(void)
 243 {
 244         struct lttng_ht_iter iter;
 245         struct relay_session *session;
 246
 247         if (!sessions_ht) {
 248                 return;
 249         }
 250
 251         rcu_read_lock();
 252         cds_lfht_for_each_entry(sessions_ht->ht, &iter.iter, session,
 253                         session_n.node) {
 254                 if (!session_get(session)) {
 255                         continue;
 256                 }
 257                 DBG("session %p refcount %ld session %" PRIu64,
 258                         session,
 259                         session->ref.refcount,
 260                         session->id);
 261                 session_put(session);
 262         }
 263         rcu_read_unlock();
 264 }