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