| /* GIO - GLib Input, Output and Streaming Library |
| * |
| * Copyright © 2009 Codethink Limited |
| * Copyright © 2009 Red Hat, Inc |
| * |
| * This library is free software; you can redistribute it and/or |
| * modify it under the terms of the GNU Lesser General Public |
| * License as published by the Free Software Foundation; either |
| * version 2.1 of the License, or (at your option) any later version. |
| * |
| * This library is distributed in the hope that it will be useful, |
| * but WITHOUT ANY WARRANTY; without even the implied warranty of |
| * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU |
| * Lesser General Public License for more details. |
| * |
| * You should have received a copy of the GNU Lesser General |
| * Public License along with this library; if not, see <http://www.gnu.org/licenses/>. |
| * |
| * Authors: Ryan Lortie <desrt@desrt.ca> |
| * Alexander Larsson <alexl@redhat.com> |
| */ |
| |
| /** |
| * SECTION:gthreadedsocketservice |
| * @title: GThreadedSocketService |
| * @short_description: A threaded GSocketService |
| * @include: gio/gio.h |
| * @see_also: #GSocketService. |
| * |
| * A #GThreadedSocketService is a simple subclass of #GSocketService |
| * that handles incoming connections by creating a worker thread and |
| * dispatching the connection to it by emitting the |
| * #GThreadedSocketService::run signal in the new thread. |
| * |
| * The signal handler may perform blocking IO and need not return |
| * until the connection is closed. |
| * |
| * The service is implemented using a thread pool, so there is a |
| * limited amount of threads available to serve incoming requests. |
| * The service automatically stops the #GSocketService from accepting |
| * new connections when all threads are busy. |
| * |
| * As with #GSocketService, you may connect to #GThreadedSocketService::run, |
| * or subclass and override the default handler. |
| */ |
| |
| #include "config.h" |
| #include "gsocketconnection.h" |
| #include "gthreadedsocketservice.h" |
| #include "glibintl.h" |
| |
| struct _GThreadedSocketServicePrivate |
| { |
| GThreadPool *thread_pool; |
| int max_threads; |
| gint job_count; |
| }; |
| |
| static guint g_threaded_socket_service_run_signal; |
| |
| G_DEFINE_TYPE_WITH_PRIVATE (GThreadedSocketService, |
| g_threaded_socket_service, |
| G_TYPE_SOCKET_SERVICE) |
| |
| enum |
| { |
| PROP_0, |
| PROP_MAX_THREADS |
| }; |
| |
| G_LOCK_DEFINE_STATIC(job_count); |
| |
| typedef struct |
| { |
| GSocketConnection *connection; |
| GObject *source_object; |
| } GThreadedSocketServiceData; |
| |
| static void |
| g_threaded_socket_service_func (gpointer _data, |
| gpointer user_data) |
| { |
| GThreadedSocketService *threaded = user_data; |
| GThreadedSocketServiceData *data = _data; |
| gboolean result; |
| |
| g_signal_emit (threaded, g_threaded_socket_service_run_signal, |
| 0, data->connection, data->source_object, &result); |
| |
| g_object_unref (data->connection); |
| if (data->source_object) |
| g_object_unref (data->source_object); |
| g_slice_free (GThreadedSocketServiceData, data); |
| |
| G_LOCK (job_count); |
| if (threaded->priv->job_count-- == threaded->priv->max_threads) |
| g_socket_service_start (G_SOCKET_SERVICE (threaded)); |
| G_UNLOCK (job_count); |
| |
| g_object_unref (threaded); |
| } |
| |
| static gboolean |
| g_threaded_socket_service_incoming (GSocketService *service, |
| GSocketConnection *connection, |
| GObject *source_object) |
| { |
| GThreadedSocketService *threaded; |
| GThreadedSocketServiceData *data; |
| |
| threaded = G_THREADED_SOCKET_SERVICE (service); |
| |
| data = g_slice_new (GThreadedSocketServiceData); |
| |
| /* Ref the socket service for the thread */ |
| g_object_ref (service); |
| |
| data->connection = g_object_ref (connection); |
| if (source_object) |
| data->source_object = g_object_ref (source_object); |
| else |
| data->source_object = NULL; |
| |
| G_LOCK (job_count); |
| if (++threaded->priv->job_count == threaded->priv->max_threads) |
| g_socket_service_stop (service); |
| G_UNLOCK (job_count); |
| |
| g_thread_pool_push (threaded->priv->thread_pool, data, NULL); |
| |
| |
| |
| return FALSE; |
| } |
| |
| static void |
| g_threaded_socket_service_init (GThreadedSocketService *service) |
| { |
| service->priv = g_threaded_socket_service_get_instance_private (service); |
| service->priv->max_threads = 10; |
| } |
| |
| static void |
| g_threaded_socket_service_constructed (GObject *object) |
| { |
| GThreadedSocketService *service = G_THREADED_SOCKET_SERVICE (object); |
| |
| service->priv->thread_pool = |
| g_thread_pool_new (g_threaded_socket_service_func, |
| service, |
| service->priv->max_threads, |
| FALSE, |
| NULL); |
| } |
| |
| |
| static void |
| g_threaded_socket_service_finalize (GObject *object) |
| { |
| GThreadedSocketService *service = G_THREADED_SOCKET_SERVICE (object); |
| |
| g_thread_pool_free (service->priv->thread_pool, FALSE, FALSE); |
| |
| G_OBJECT_CLASS (g_threaded_socket_service_parent_class) |
| ->finalize (object); |
| } |
| |
| static void |
| g_threaded_socket_service_get_property (GObject *object, |
| guint prop_id, |
| GValue *value, |
| GParamSpec *pspec) |
| { |
| GThreadedSocketService *service = G_THREADED_SOCKET_SERVICE (object); |
| |
| switch (prop_id) |
| { |
| case PROP_MAX_THREADS: |
| g_value_set_int (value, service->priv->max_threads); |
| break; |
| |
| default: |
| G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); |
| } |
| } |
| |
| static void |
| g_threaded_socket_service_set_property (GObject *object, |
| guint prop_id, |
| const GValue *value, |
| GParamSpec *pspec) |
| { |
| GThreadedSocketService *service = G_THREADED_SOCKET_SERVICE (object); |
| |
| switch (prop_id) |
| { |
| case PROP_MAX_THREADS: |
| service->priv->max_threads = g_value_get_int (value); |
| break; |
| |
| default: |
| G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); |
| } |
| } |
| |
| |
| static void |
| g_threaded_socket_service_class_init (GThreadedSocketServiceClass *class) |
| { |
| GObjectClass *gobject_class = G_OBJECT_CLASS (class); |
| GSocketServiceClass *ss_class = &class->parent_class; |
| |
| gobject_class->constructed = g_threaded_socket_service_constructed; |
| gobject_class->finalize = g_threaded_socket_service_finalize; |
| gobject_class->set_property = g_threaded_socket_service_set_property; |
| gobject_class->get_property = g_threaded_socket_service_get_property; |
| |
| ss_class->incoming = g_threaded_socket_service_incoming; |
| |
| /** |
| * GThreadedSocketService::run: |
| * @service: the #GThreadedSocketService. |
| * @connection: a new #GSocketConnection object. |
| * @source_object: the source_object passed to g_socket_listener_add_address(). |
| * |
| * The ::run signal is emitted in a worker thread in response to an |
| * incoming connection. This thread is dedicated to handling |
| * @connection and may perform blocking IO. The signal handler need |
| * not return until the connection is closed. |
| * |
| * Returns: %TRUE to stop further signal handlers from being called |
| */ |
| g_threaded_socket_service_run_signal = |
| g_signal_new (I_("run"), G_TYPE_FROM_CLASS (class), G_SIGNAL_RUN_LAST, |
| G_STRUCT_OFFSET (GThreadedSocketServiceClass, run), |
| g_signal_accumulator_true_handled, NULL, |
| NULL, G_TYPE_BOOLEAN, |
| 2, G_TYPE_SOCKET_CONNECTION, G_TYPE_OBJECT); |
| |
| g_object_class_install_property (gobject_class, PROP_MAX_THREADS, |
| g_param_spec_int ("max-threads", |
| P_("Max threads"), |
| P_("The max number of threads handling clients for this service"), |
| -1, |
| G_MAXINT, |
| 10, |
| G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); |
| } |
| |
| /** |
| * g_threaded_socket_service_new: |
| * @max_threads: the maximal number of threads to execute concurrently |
| * handling incoming clients, -1 means no limit |
| * |
| * Creates a new #GThreadedSocketService with no listeners. Listeners |
| * must be added with one of the #GSocketListener "add" methods. |
| * |
| * Returns: a new #GSocketService. |
| * |
| * Since: 2.22 |
| */ |
| GSocketService * |
| g_threaded_socket_service_new (int max_threads) |
| { |
| return g_object_new (G_TYPE_THREADED_SOCKET_SERVICE, |
| "max-threads", max_threads, |
| NULL); |
| } |