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