gst/rtsp-server/rtsp-thread-pool.c

   1 /* GStreamer
   2  * Copyright (C) 2013 Wim Taymans <wim.taymans at gmail.com>
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Library General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2 of the License, or (at your option) any later version.
   8  *
   9  * This library is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Library General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Library General Public
  15  * License along with this library; if not, write to the
  16  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  17  * Boston, MA 02110-1301, USA.
  18  */
  19 /**
  20  * SECTION:rtsp-thread-pool
  21  * @short_description: A pool of threads
  22  * @see_also: #GstRTSPMedia, #GstRTSPClient
  23  *
  24  * A #GstRTSPThreadPool manages reusable threads for various server tasks.
  25  * Currently the defined thread types can be found in #GstRTSPThreadType.
  26  *
  27  * Threads of type #GST_RTSP_THREAD_TYPE_CLIENT are used to handle requests from
  28  * a connected client. With gst_rtsp_thread_pool_get_max_threads() a maximum
  29  * number of threads can be set after which the pool will start to reuse the
  30  * same thread for multiple clients.
  31  *
  32  * Threads of type #GST_RTSP_THREAD_TYPE_MEDIA will be used to perform the state
  33  * changes of the media pipelines and handle its bus messages.
  34  *
  35  * gst_rtsp_thread_pool_get_thread() can be used to create a #GstRTSPThread
  36  * object of the right type. The thread object contains a mainloop and context
  37  * that run in a seperate thread and can be used to attached sources to.
  38  *
  39  * gst_rtsp_thread_reuse() can be used to reuse a thread for multiple purposes.
  40  * If all gst_rtsp_thread_reuse() calls are matched with a
  41  * gst_rtsp_thread_stop() call, the mainloop will be quit and the thread will
  42  * stop.
  43  *
  44  * To configure the threads, a subclass of this object should be made and the
  45  * virtual methods should be overriden to implement the desired functionality.
  46  *
  47  * Last reviewed on 2013-07-11 (1.0.0)
  48  */
  49
  50 #include <string.h>
  51
  52 #include "rtsp-thread-pool.h"
  53
  54 typedef struct _GstRTSPThreadImpl
  55 {
  56   GstRTSPThread thread;
  57
  58   gint reused;
  59   GSource *source;
  60 } GstRTSPThreadImpl;
  61
  62 GST_DEFINE_MINI_OBJECT_TYPE (GstRTSPThread, gst_rtsp_thread);
  63
  64 static void gst_rtsp_thread_init (GstRTSPThreadImpl * impl);
  65
  66 static void
  67 _gst_rtsp_thread_free (GstRTSPThreadImpl * impl)
  68 {
  69   GST_DEBUG ("free thread %p", impl);
  70
  71   g_source_unref (impl->source);
  72   g_main_loop_unref (impl->thread.loop);
  73   g_main_context_unref (impl->thread.context);
  74   g_slice_free1 (sizeof (GstRTSPThreadImpl), impl);
  75 }
  76
  77 static GstRTSPThread *
  78 _gst_rtsp_thread_copy (GstRTSPThreadImpl * impl)
  79 {
  80   GstRTSPThreadImpl *copy;
  81
  82   GST_DEBUG ("copy thread %p", impl);
  83
  84   copy = g_slice_new0 (GstRTSPThreadImpl);
  85   gst_rtsp_thread_init (copy);
  86   copy->thread.context = g_main_context_ref (impl->thread.context);
  87   copy->thread.loop = g_main_loop_ref (impl->thread.loop);
  88
  89   return GST_RTSP_THREAD (copy);
  90 }
  91
  92 static void
  93 gst_rtsp_thread_init (GstRTSPThreadImpl * impl)
  94 {
  95   gst_mini_object_init (GST_MINI_OBJECT_CAST (impl), 0,
  96       GST_TYPE_RTSP_THREAD,
  97       (GstMiniObjectCopyFunction) _gst_rtsp_thread_copy, NULL,
  98       (GstMiniObjectFreeFunction) _gst_rtsp_thread_free);
  99
 100   g_atomic_int_set (&impl->reused, 1);
 101 }
 102
 103 /**
 104  * gst_rtsp_thread_new:
 105  * @type: the thread type
 106  *
 107  * Create a new thread object that can run a mainloop.
 108  *
 109  * Returns: (transfer full): a #GstRTSPThread.
 110  */
 111 GstRTSPThread *
 112 gst_rtsp_thread_new (GstRTSPThreadType type)
 113 {
 114   GstRTSPThreadImpl *impl;
 115
 116   impl = g_slice_new0 (GstRTSPThreadImpl);
 117
 118   gst_rtsp_thread_init (impl);
 119   impl->thread.type = type;
 120   impl->thread.context = g_main_context_new ();
 121   impl->thread.loop = g_main_loop_new (impl->thread.context, TRUE);
 122
 123   return GST_RTSP_THREAD (impl);
 124 }
 125
 126 /**
 127  * gst_rtsp_thread_reuse:
 128  * @thread: (transfer none): a #GstRTSPThread
 129  *
 130  * Reuse the mainloop of @thread
 131  *
 132  * Returns: %TRUE if the mainloop could be reused
 133  */
 134 gboolean
 135 gst_rtsp_thread_reuse (GstRTSPThread * thread)
 136 {
 137   GstRTSPThreadImpl *impl = (GstRTSPThreadImpl *) thread;
 138   gboolean res;
 139
 140   g_return_val_if_fail (GST_IS_RTSP_THREAD (thread), FALSE);
 141
 142   GST_DEBUG ("reuse thread %p", thread);
 143
 144   res = g_atomic_int_add (&impl->reused, 1) > 0;
 145   if (res)
 146     gst_rtsp_thread_ref (thread);
 147
 148   return res;
 149 }
 150
 151 static gboolean
 152 do_quit (GstRTSPThread * thread)
 153 {
 154   GST_DEBUG ("stop mainloop of thread %p", thread);
 155   g_main_loop_quit (thread->loop);
 156   return FALSE;
 157 }
 158
 159 /**
 160  * gst_rtsp_thread_stop:
 161  * @thread: (transfer full): a #GstRTSPThread
 162  *
 163  * Stop and unref @thread. When no threads are using the mainloop, the thread
 164  * will be stopped and the final ref to @thread will be released.
 165  */
 166 void
 167 gst_rtsp_thread_stop (GstRTSPThread * thread)
 168 {
 169   GstRTSPThreadImpl *impl = (GstRTSPThreadImpl *) thread;
 170
 171   g_return_if_fail (GST_IS_RTSP_THREAD (thread));
 172
 173   GST_DEBUG ("stop thread %p", thread);
 174
 175   if (g_atomic_int_dec_and_test (&impl->reused)) {
 176     GST_DEBUG ("add idle source to quit mainloop of thread %p", thread);
 177     impl->source = g_idle_source_new ();
 178     g_source_set_callback (impl->source, (GSourceFunc) do_quit,
 179         thread, (GDestroyNotify) gst_rtsp_thread_unref);
 180     g_source_attach (impl->source, thread->context);
 181   } else
 182     gst_rtsp_thread_unref (thread);
 183 }
 184
 185 #define GST_RTSP_THREAD_POOL_GET_PRIVATE(obj)  \
 186    (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_RTSP_THREAD_POOL, GstRTSPThreadPoolPrivate))
 187
 188 struct _GstRTSPThreadPoolPrivate
 189 {
 190   GMutex lock;
 191
 192   gint max_threads;
 193   /* currently used mainloops */
 194   GQueue threads;
 195 };
 196
 197 #define DEFAULT_MAX_THREADS 1
 198
 199 enum
 200 {
 201   PROP_0,
 202   PROP_MAX_THREADS,
 203   PROP_LAST
 204 };
 205
 206 GST_DEBUG_CATEGORY_STATIC (rtsp_thread_pool_debug);
 207 #define GST_CAT_DEFAULT rtsp_thread_pool_debug
 208
 209 static GQuark thread_pool;
 210
 211 static void gst_rtsp_thread_pool_get_property (GObject * object, guint propid,
 212     GValue * value, GParamSpec * pspec);
 213 static void gst_rtsp_thread_pool_set_property (GObject * object, guint propid,
 214     const GValue * value, GParamSpec * pspec);
 215 static void gst_rtsp_thread_pool_finalize (GObject * obj);
 216
 217 static gpointer do_loop (GstRTSPThread * thread);
 218 static GstRTSPThread *default_get_thread (GstRTSPThreadPool * pool,
 219     GstRTSPThreadType type, GstRTSPContext * ctx);
 220
 221 G_DEFINE_TYPE (GstRTSPThreadPool, gst_rtsp_thread_pool, G_TYPE_OBJECT);
 222
 223 static void
 224 gst_rtsp_thread_pool_class_init (GstRTSPThreadPoolClass * klass)
 225 {
 226   GObjectClass *gobject_class;
 227
 228   g_type_class_add_private (klass, sizeof (GstRTSPThreadPoolPrivate));
 229
 230   gobject_class = G_OBJECT_CLASS (klass);
 231
 232   gobject_class->get_property = gst_rtsp_thread_pool_get_property;
 233   gobject_class->set_property = gst_rtsp_thread_pool_set_property;
 234   gobject_class->finalize = gst_rtsp_thread_pool_finalize;
 235
 236   /**
 237    * GstRTSPThreadPool::max-threads:
 238    *
 239    * The maximum amount of threads to use for client connections. A value of
 240    * 0 means to use only the mainloop, -1 means an unlimited amount of
 241    * threads.
 242    */
 243   g_object_class_install_property (gobject_class, PROP_MAX_THREADS,
 244       g_param_spec_int ("max-threads", "Max Threads",
 245           "The maximum amount of threads to use for client connections "
 246           "(0 = only mainloop, -1 = unlimited)", -1, G_MAXINT,
 247           DEFAULT_MAX_THREADS, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 248
 249   klass->get_thread = default_get_thread;
 250
 251   GST_DEBUG_CATEGORY_INIT (rtsp_thread_pool_debug, "rtspthreadpool", 0,
 252       "GstRTSPThreadPool");
 253
 254   thread_pool = g_quark_from_string ("gst.rtsp.thread.pool");
 255 }
 256
 257 static void
 258 gst_rtsp_thread_pool_init (GstRTSPThreadPool * pool)
 259 {
 260   GstRTSPThreadPoolPrivate *priv;
 261
 262   pool->priv = priv = GST_RTSP_THREAD_POOL_GET_PRIVATE (pool);
 263
 264   g_mutex_init (&priv->lock);
 265   priv->max_threads = DEFAULT_MAX_THREADS;
 266   g_queue_init (&priv->threads);
 267 }
 268
 269 static void
 270 gst_rtsp_thread_pool_finalize (GObject * obj)
 271 {
 272   GstRTSPThreadPool *pool = GST_RTSP_THREAD_POOL (obj);
 273   GstRTSPThreadPoolPrivate *priv = pool->priv;
 274
 275   GST_INFO ("finalize pool %p", pool);
 276
 277   g_queue_clear (&priv->threads);
 278   g_mutex_clear (&priv->lock);
 279
 280   G_OBJECT_CLASS (gst_rtsp_thread_pool_parent_class)->finalize (obj);
 281 }
 282
 283 static void
 284 gst_rtsp_thread_pool_get_property (GObject * object, guint propid,
 285     GValue * value, GParamSpec * pspec)
 286 {
 287   GstRTSPThreadPool *pool = GST_RTSP_THREAD_POOL (object);
 288
 289   switch (propid) {
 290     case PROP_MAX_THREADS:
 291       g_value_set_int (value, gst_rtsp_thread_pool_get_max_threads (pool));
 292       break;
 293     default:
 294       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, propid, pspec);
 295   }
 296 }
 297
 298 static void
 299 gst_rtsp_thread_pool_set_property (GObject * object, guint propid,
 300     const GValue * value, GParamSpec * pspec)
 301 {
 302   GstRTSPThreadPool *pool = GST_RTSP_THREAD_POOL (object);
 303
 304   switch (propid) {
 305     case PROP_MAX_THREADS:
 306       gst_rtsp_thread_pool_set_max_threads (pool, g_value_get_int (value));
 307       break;
 308     default:
 309       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, propid, pspec);
 310   }
 311 }
 312
 313 static gpointer
 314 do_loop (GstRTSPThread * thread)
 315 {
 316   GstRTSPThreadPoolPrivate *priv;
 317   GstRTSPThreadPoolClass *klass;
 318   GstRTSPThreadPool *pool;
 319
 320   pool = gst_mini_object_get_qdata (GST_MINI_OBJECT (thread), thread_pool);
 321   priv = pool->priv;
 322
 323   klass = GST_RTSP_THREAD_POOL_GET_CLASS (pool);
 324
 325   if (klass->thread_enter)
 326     klass->thread_enter (pool, thread);
 327
 328   GST_INFO ("enter mainloop of thread %p", thread);
 329   g_main_loop_run (thread->loop);
 330   GST_INFO ("exit mainloop of thread %p", thread);
 331
 332   if (klass->thread_leave)
 333     klass->thread_leave (pool, thread);
 334
 335   g_mutex_lock (&priv->lock);
 336   g_queue_remove (&priv->threads, thread);
 337   g_mutex_unlock (&priv->lock);
 338
 339   gst_rtsp_thread_unref (thread);
 340
 341   return NULL;
 342 }
 343
 344 /**
 345  * gst_rtsp_thread_pool_new:
 346  *
 347  * Create a new #GstRTSPThreadPool instance.
 348  *
 349  * Returns: (transfer full): a new #GstRTSPThreadPool
 350  */
 351 GstRTSPThreadPool *
 352 gst_rtsp_thread_pool_new (void)
 353 {
 354   GstRTSPThreadPool *result;
 355
 356   result = g_object_new (GST_TYPE_RTSP_THREAD_POOL, NULL);
 357
 358   return result;
 359 }
 360
 361 /**
 362  * gst_rtsp_thread_pool_set_max_threads:
 363  * @pool: a #GstRTSPThreadPool
 364  * @max_threads: maximum threads
 365  *
 366  * Set the maximum threads used by the pool to handle client requests.
 367  * A value of 0 will use the pool mainloop, a value of -1 will use an
 368  * unlimited number of threads.
 369  */
 370 void
 371 gst_rtsp_thread_pool_set_max_threads (GstRTSPThreadPool * pool,
 372     gint max_threads)
 373 {
 374   GstRTSPThreadPoolPrivate *priv;
 375
 376   g_return_if_fail (GST_IS_RTSP_THREAD_POOL (pool));
 377
 378   priv = pool->priv;
 379
 380   g_mutex_lock (&priv->lock);
 381   priv->max_threads = max_threads;
 382   g_mutex_unlock (&priv->lock);
 383 }
 384
 385 /**
 386  * gst_rtsp_thread_pool_get_max_threads:
 387  * @pool: a #GstRTSPThreadPool
 388  *
 389  * Get the maximum number of threads used for client connections.
 390  * See gst_rtsp_thread_pool_set_max_threads().
 391  *
 392  * Returns: the maximum number of threads.
 393  */
 394 gint
 395 gst_rtsp_thread_pool_get_max_threads (GstRTSPThreadPool * pool)
 396 {
 397   GstRTSPThreadPoolPrivate *priv;
 398   gint res;
 399
 400   g_return_val_if_fail (GST_IS_RTSP_THREAD_POOL (pool), -1);
 401
 402   priv = pool->priv;
 403
 404   g_mutex_lock (&priv->lock);
 405   res = priv->max_threads;
 406   g_mutex_unlock (&priv->lock);
 407
 408   return res;
 409 }
 410
 411 static GstRTSPThread *
 412 make_thread (GstRTSPThreadPool * pool, GstRTSPThreadType type,
 413     GstRTSPContext * ctx)
 414 {
 415   GstRTSPThreadPoolClass *klass;
 416   GstRTSPThread *thread;
 417
 418   klass = GST_RTSP_THREAD_POOL_GET_CLASS (pool);
 419
 420   thread = gst_rtsp_thread_new (type);
 421   gst_mini_object_set_qdata (GST_MINI_OBJECT (thread), thread_pool,
 422       g_object_ref (pool), g_object_unref);
 423
 424   GST_DEBUG_OBJECT (pool, "new thread %p", thread);
 425
 426   if (klass->configure_thread)
 427     klass->configure_thread (pool, thread, ctx);
 428
 429   return thread;
 430 }
 431
 432 static GstRTSPThread *
 433 default_get_thread (GstRTSPThreadPool * pool,
 434     GstRTSPThreadType type, GstRTSPContext * ctx)
 435 {
 436   GstRTSPThreadPoolPrivate *priv = pool->priv;
 437   GstRTSPThreadPoolClass *klass;
 438   GstRTSPThread *thread;
 439   GError *error = NULL;
 440
 441   klass = GST_RTSP_THREAD_POOL_GET_CLASS (pool);
 442
 443   switch (type) {
 444     case GST_RTSP_THREAD_TYPE_CLIENT:
 445       if (priv->max_threads == 0) {
 446         /* no threads allowed */
 447         GST_DEBUG_OBJECT (pool, "no client threads allowed");
 448         thread = NULL;
 449       } else {
 450         g_mutex_lock (&priv->lock);
 451       retry:
 452         if (priv->max_threads > 0 &&
 453             g_queue_get_length (&priv->threads) >= priv->max_threads) {
 454           /* max threads reached, recycle from queue */
 455           thread = g_queue_pop_head (&priv->threads);
 456           GST_DEBUG_OBJECT (pool, "recycle client thread %p", thread);
 457           if (!gst_rtsp_thread_reuse (thread)) {
 458             GST_DEBUG_OBJECT (pool, "thread %p stopping, retry", thread);
 459             /* this can happen if we just decremented the reuse counter of the
 460              * thread and signaled the mainloop that it should stop. We leave
 461              * the thread out of the queue now, there is no point to add it
 462              * again, it will be removed from the mainloop otherwise after it
 463              * stops. */
 464             goto retry;
 465           }
 466         } else {
 467           /* make more threads */
 468           GST_DEBUG_OBJECT (pool, "make new client thread");
 469           thread = make_thread (pool, type, ctx);
 470
 471           if (!g_thread_pool_push (klass->pool, gst_rtsp_thread_ref (thread),
 472                   &error))
 473             goto thread_error;
 474         }
 475         g_queue_push_tail (&priv->threads, thread);
 476         g_mutex_unlock (&priv->lock);
 477       }
 478       break;
 479     case GST_RTSP_THREAD_TYPE_MEDIA:
 480       GST_DEBUG_OBJECT (pool, "make new media thread");
 481       thread = make_thread (pool, type, ctx);
 482
 483       if (!g_thread_pool_push (klass->pool, gst_rtsp_thread_ref (thread),
 484               &error))
 485         goto thread_error;
 486       break;
 487     default:
 488       thread = NULL;
 489       break;
 490   }
 491   return thread;
 492
 493   /* ERRORS */
 494 thread_error:
 495   {
 496     GST_ERROR_OBJECT (pool, "failed to push thread %s", error->message);
 497     gst_rtsp_thread_unref (thread);
 498     /* drop also the ref dedicated for the pool */
 499     gst_rtsp_thread_unref (thread);
 500     g_clear_error (&error);
 501     return NULL;
 502   }
 503 }
 504
 505 /**
 506  * gst_rtsp_thread_pool_get_thread:
 507  * @pool: a #GstRTSPThreadPool
 508  * @type: the #GstRTSPThreadType
 509  * @ctx: (transfer none): a #GstRTSPContext
 510  *
 511  * Get a new #GstRTSPThread for @type and @ctx.
 512  *
 513  * Returns: (transfer full): a new #GstRTSPThread, gst_rtsp_thread_stop() after usage
 514  */
 515 GstRTSPThread *
 516 gst_rtsp_thread_pool_get_thread (GstRTSPThreadPool * pool,
 517     GstRTSPThreadType type, GstRTSPContext * ctx)
 518 {
 519   GstRTSPThreadPoolClass *klass;
 520   GstRTSPThread *result = NULL;
 521
 522   g_return_val_if_fail (GST_IS_RTSP_THREAD_POOL (pool), NULL);
 523
 524   klass = GST_RTSP_THREAD_POOL_GET_CLASS (pool);
 525
 526   /* We want to be thread safe as there might be 2 threads wanting to get new
 527    * #GstRTSPThread at the same time
 528    */
 529   if (G_UNLIKELY (!g_atomic_pointer_get (&klass->pool))) {
 530     GThreadPool *t_pool;
 531     t_pool = g_thread_pool_new ((GFunc) do_loop, klass, -1, FALSE, NULL);
 532     if (!g_atomic_pointer_compare_and_exchange (&klass->pool, NULL, t_pool))
 533       g_thread_pool_free (t_pool, FALSE, TRUE);
 534   }
 535
 536   if (klass->get_thread)
 537     result = klass->get_thread (pool, type, ctx);
 538
 539   return result;
 540 }
 541
 542 /**
 543  * gst_rtsp_thread_pool_cleanup:
 544  *
 545  * Wait for all tasks to be stopped and free all allocated resources. This is
 546  * mainly used in test suites to ensure proper cleanup of internal data
 547  * structures.
 548  */
 549 void
 550 gst_rtsp_thread_pool_cleanup (void)
 551 {
 552   GstRTSPThreadPoolClass *klass;
 553
 554   klass =
 555       GST_RTSP_THREAD_POOL_CLASS (g_type_class_peek
 556       (gst_rtsp_thread_pool_get_type ()));
 557   if (klass->pool != NULL) {
 558     g_thread_pool_free (klass->pool, FALSE, TRUE);
 559     klass->pool = NULL;
 560   }
 561 }