Merge remote branch 'gvdb/master'
[platform/upstream/glib.git] / gio / gthreadedsocketservice.c
1 /* GIO - GLib Input, Output and Streaming Library
2  *
3  * Copyright © 2009 Codethink Limited
4  * Copyright © 2009 Red Hat, Inc
5  *
6  * This program is free software: you can redistribute it and/or modify
7  * it under the terms of the GNU Lesser General Public License as published
8  * by the Free Software Foundation; either version 2 of the licence or (at
9  * your option) any later version.
10  *
11  * This library is distributed in the hope that it will be useful,
12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14  * Lesser General Public License for more details.
15  *
16  * You should have received a copy of the GNU Lesser General
17  * Public License along with this library; if not, write to the
18  * Free Software Foundation, Inc., 59 Temple Place, Suite 330,
19  * Boston, MA 02111-1307, USA.
20  *
21  * Authors: Ryan Lortie <desrt@desrt.ca>
22  *          Alexander Larsson <alexl@redhat.com>
23  */
24
25 /**
26  * SECTION: gthreadedsocketservice
27  * @title: GThreadedSocketService
28  * @short_description: A threaded GSocketService
29  * @see_also: #GSocketService.
30  *
31  * A #GThreadedSocketService is a simple subclass of #GSocketService
32  * that handles incoming connections by creating a worker thread and
33  * dispatching the connection to it by emitting the ::run signal in
34  * the new thread.
35  *
36  * The signal handler may perform blocking IO and need not return
37  * until the connection is closed.
38  *
39  * The service is implemented using a thread pool, so there is a
40  * limited amount of threads availible to serve incomming requests.
41  * The service automatically stops the #GSocketService from accepting
42  * new connections when all threads are busy.
43  *
44  * As with #GSocketService, you may connect to #GThreadedSocketService:run,
45  * or subclass and override the default handler.
46  */
47
48 #include "config.h"
49 #include "gsocketconnection.h"
50 #include "gthreadedsocketservice.h"
51 #include "glibintl.h"
52
53 #include "gio-marshal.h"
54
55 #include "gioalias.h"
56
57 static guint g_threaded_socket_service_run_signal;
58
59 G_DEFINE_TYPE (GThreadedSocketService,
60                g_threaded_socket_service,
61                G_TYPE_SOCKET_SERVICE);
62
63 enum
64 {
65   PROP_0,
66   PROP_MAX_THREADS
67 };
68
69
70 G_LOCK_DEFINE_STATIC(job_count);
71
72 struct _GThreadedSocketServicePrivate
73 {
74   GThreadPool *thread_pool;
75   int max_threads;
76   gint job_count;
77 };
78
79 typedef struct
80 {
81   GThreadedSocketService *service;
82   GSocketConnection *connection;
83   GObject *source_object;
84 } GThreadedSocketServiceData;
85
86 static void
87 g_threaded_socket_service_func (gpointer _data,
88                                 gpointer user_data)
89 {
90   GThreadedSocketService *threaded = user_data;
91   GThreadedSocketServiceData *data = _data;
92   gboolean result;
93
94   g_signal_emit (data->service, g_threaded_socket_service_run_signal,
95                  0, data->connection, data->source_object, &result);
96
97   g_object_unref (data->service);
98   g_object_unref (data->connection);
99   if (data->source_object)
100     g_object_unref (data->source_object);
101   g_slice_free (GThreadedSocketServiceData, data);
102
103   G_LOCK (job_count);
104   if (threaded->priv->job_count-- == threaded->priv->max_threads)
105     g_socket_service_start (G_SOCKET_SERVICE (threaded));
106   G_UNLOCK (job_count);
107 }
108
109 static gboolean
110 g_threaded_socket_service_incoming (GSocketService    *service,
111                                     GSocketConnection *connection,
112                                     GObject           *source_object)
113 {
114   GThreadedSocketService *threaded;
115   GThreadedSocketServiceData *data;
116
117   threaded = G_THREADED_SOCKET_SERVICE (service);
118
119   data = g_slice_new (GThreadedSocketServiceData);
120   data->service = g_object_ref (service);
121   data->connection = g_object_ref (connection);
122   if (source_object)
123     data->source_object = g_object_ref (source_object);
124   else
125     data->source_object = NULL;
126
127   G_LOCK (job_count);
128   if (++threaded->priv->job_count == threaded->priv->max_threads)
129     g_socket_service_stop (service);
130   G_UNLOCK (job_count);
131
132   g_thread_pool_push (threaded->priv->thread_pool, data, NULL);
133
134
135
136   return FALSE;
137 }
138
139 static void
140 g_threaded_socket_service_init (GThreadedSocketService *service)
141 {
142   service->priv = G_TYPE_INSTANCE_GET_PRIVATE (service,
143                                                G_TYPE_THREADED_SOCKET_SERVICE,
144                                                GThreadedSocketServicePrivate);
145   service->priv->max_threads = 10;
146 }
147
148 static void
149 g_threaded_socket_service_constructed (GObject *object)
150 {
151   GThreadedSocketService *service = G_THREADED_SOCKET_SERVICE (object);
152
153   service->priv->thread_pool =
154     g_thread_pool_new  (g_threaded_socket_service_func,
155                         service,
156                         service->priv->max_threads,
157                         FALSE,
158                         NULL);
159 }
160
161
162 static void
163 g_threaded_socket_service_finalize (GObject *object)
164 {
165   GThreadedSocketService *service = G_THREADED_SOCKET_SERVICE (object);
166
167   g_thread_pool_free (service->priv->thread_pool, FALSE, TRUE);
168
169   G_OBJECT_CLASS (g_threaded_socket_service_parent_class)
170     ->finalize (object);
171 }
172
173 static void
174 g_threaded_socket_service_get_property (GObject    *object,
175                                         guint       prop_id,
176                                         GValue     *value,
177                                         GParamSpec *pspec)
178 {
179   GThreadedSocketService *service = G_THREADED_SOCKET_SERVICE (object);
180
181   switch (prop_id)
182     {
183       case PROP_MAX_THREADS:
184         g_value_set_int (value, service->priv->max_threads);
185         break;
186
187       default:
188         G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
189     }
190 }
191
192 static void
193 g_threaded_socket_service_set_property (GObject      *object,
194                                         guint         prop_id,
195                                         const GValue *value,
196                                         GParamSpec   *pspec)
197 {
198   GThreadedSocketService *service = G_THREADED_SOCKET_SERVICE (object);
199
200   switch (prop_id)
201     {
202       case PROP_MAX_THREADS:
203         service->priv->max_threads = g_value_get_int (value);
204         break;
205
206       default:
207         G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
208     }
209 }
210
211
212 static void
213 g_threaded_socket_service_class_init (GThreadedSocketServiceClass *class)
214 {
215   GObjectClass *gobject_class = G_OBJECT_CLASS (class);
216   GSocketServiceClass *ss_class = &class->parent_class;
217
218   g_type_class_add_private (class, sizeof (GThreadedSocketServicePrivate));
219
220   gobject_class->constructed = g_threaded_socket_service_constructed;
221   gobject_class->finalize = g_threaded_socket_service_finalize;
222   gobject_class->set_property = g_threaded_socket_service_set_property;
223   gobject_class->get_property = g_threaded_socket_service_get_property;
224
225   ss_class->incoming = g_threaded_socket_service_incoming;
226
227   /**
228    * GThreadedSocketService::run:
229    * @service: the #GThreadedSocketService.
230    * @connection: a new #GSocketConnection object.
231    * @source_object: the source_object passed to g_socket_listener_add_address().
232    *
233    * The ::run signal is emitted in a worker thread in response to an
234    * incoming connection. This thread is dedicated to handling
235    * @connection and may perform blocking IO. The signal handler need
236    * not return until the connection is closed.
237    *
238    * Returns: %TRUE to stope further signal handlers from being called
239    */
240   g_threaded_socket_service_run_signal =
241     g_signal_new ("run", G_TYPE_FROM_CLASS (class), G_SIGNAL_RUN_LAST,
242                   G_STRUCT_OFFSET (GThreadedSocketServiceClass, run),
243                   g_signal_accumulator_true_handled, NULL,
244                   _gio_marshal_BOOLEAN__OBJECT_OBJECT, G_TYPE_BOOLEAN,
245                   2, G_TYPE_SOCKET_CONNECTION, G_TYPE_OBJECT);
246
247   g_object_class_install_property (gobject_class, PROP_MAX_THREADS,
248                                    g_param_spec_int ("max-threads",
249                                                      P_("Max threads"),
250                                                      P_("The max number of threads handling clients for this service"),
251                                                      -1,
252                                                      G_MAXINT,
253                                                      10,
254                                                      G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
255 }
256
257 /**
258  * g_threaded_socket_service_new:
259  * @max_threads: the maximal number of threads to execute concurrently
260  *   handling incoming clients, -1 means no limit
261  *
262  * Creates a new #GThreadedSocketService with no listeners. Listeners
263  * must be added with g_socket_service_add_listeners().
264  *
265  * Returns: a new #GSocketService.
266  *
267  * Since: 2.22
268  */
269 GSocketService *
270 g_threaded_socket_service_new (int max_threads)
271 {
272   return g_object_new (G_TYPE_THREADED_SOCKET_SERVICE,
273                        "max-threads", max_threads,
274                        NULL);
275 }
276
277 #define __G_THREADED_SOCKET_SERVICE_C__
278 #include "gioaliasdef.c"