gio: use GPollable* to implement fallback read_async/write_async
[platform/upstream/glib.git] / gio / gunixinputstream.c
1 /* GIO - GLib Input, Output and Streaming Library
2  * 
3  * Copyright (C) 2006-2007 Red Hat, Inc.
4  *
5  * This library is free software; you can redistribute it and/or
6  * modify it under the terms of the GNU Lesser General Public
7  * License as published by the Free Software Foundation; either
8  * version 2 of the License, or (at your option) any later version.
9  *
10  * This library is distributed in the hope that it will be useful,
11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * Lesser General Public License for more details.
14  *
15  * You should have received a copy of the GNU Lesser General
16  * Public License along with this library; if not, write to the
17  * Free Software Foundation, Inc., 59 Temple Place, Suite 330,
18  * Boston, MA 02111-1307, USA.
19  *
20  * Author: Alexander Larsson <alexl@redhat.com>
21  */
22
23 #include "config.h"
24
25 #include <sys/types.h>
26 #include <sys/stat.h>
27 #include <unistd.h>
28 #include <errno.h>
29 #include <stdio.h>
30 #include <fcntl.h>
31
32 #include <glib.h>
33 #include <glib/gstdio.h>
34 #include "gioerror.h"
35 #include "gsimpleasyncresult.h"
36 #include "gunixinputstream.h"
37 #include "gcancellable.h"
38 #include "gasynchelper.h"
39 #include "gfiledescriptorbased.h"
40 #include "glibintl.h"
41
42
43 /**
44  * SECTION:gunixinputstream
45  * @short_description: Streaming input operations for UNIX file descriptors
46  * @include: gio/gunixinputstream.h
47  * @see_also: #GInputStream
48  *
49  * #GUnixInputStream implements #GInputStream for reading from a UNIX
50  * file descriptor, including asynchronous operations. (If the file
51  * descriptor refers to a socket or pipe, this will use poll() to do
52  * asynchronous I/O. If it refers to a regular file, it will fall back
53  * to doing asynchronous I/O in another thread.)
54  *
55  * Note that <filename>&lt;gio/gunixinputstream.h&gt;</filename> belongs
56  * to the UNIX-specific GIO interfaces, thus you have to use the
57  * <filename>gio-unix-2.0.pc</filename> pkg-config file when using it.
58  */
59
60 enum {
61   PROP_0,
62   PROP_FD,
63   PROP_CLOSE_FD
64 };
65
66 static void g_unix_input_stream_pollable_iface_init (GPollableInputStreamInterface *iface);
67 static void g_unix_input_stream_file_descriptor_based_iface_init (GFileDescriptorBasedIface *iface);
68
69 G_DEFINE_TYPE_WITH_CODE (GUnixInputStream, g_unix_input_stream, G_TYPE_INPUT_STREAM,
70                          G_IMPLEMENT_INTERFACE (G_TYPE_POLLABLE_INPUT_STREAM,
71                                                 g_unix_input_stream_pollable_iface_init)
72                          G_IMPLEMENT_INTERFACE (G_TYPE_FILE_DESCRIPTOR_BASED,
73                                                 g_unix_input_stream_file_descriptor_based_iface_init)
74                          )
75
76 struct _GUnixInputStreamPrivate {
77   int fd;
78   guint close_fd : 1;
79   guint is_pipe_or_socket : 1;
80 };
81
82 static void     g_unix_input_stream_set_property (GObject              *object,
83                                                   guint                 prop_id,
84                                                   const GValue         *value,
85                                                   GParamSpec           *pspec);
86 static void     g_unix_input_stream_get_property (GObject              *object,
87                                                   guint                 prop_id,
88                                                   GValue               *value,
89                                                   GParamSpec           *pspec);
90 static gssize   g_unix_input_stream_read         (GInputStream         *stream,
91                                                   void                 *buffer,
92                                                   gsize                 count,
93                                                   GCancellable         *cancellable,
94                                                   GError              **error);
95 static gboolean g_unix_input_stream_close        (GInputStream         *stream,
96                                                   GCancellable         *cancellable,
97                                                   GError              **error);
98 static void     g_unix_input_stream_skip_async   (GInputStream         *stream,
99                                                   gsize                 count,
100                                                   int                   io_priority,
101                                                   GCancellable         *cancellable,
102                                                   GAsyncReadyCallback   callback,
103                                                   gpointer              data);
104 static gssize   g_unix_input_stream_skip_finish  (GInputStream         *stream,
105                                                   GAsyncResult         *result,
106                                                   GError              **error);
107 static void     g_unix_input_stream_close_async  (GInputStream         *stream,
108                                                   int                   io_priority,
109                                                   GCancellable         *cancellable,
110                                                   GAsyncReadyCallback   callback,
111                                                   gpointer              data);
112 static gboolean g_unix_input_stream_close_finish (GInputStream         *stream,
113                                                   GAsyncResult         *result,
114                                                   GError              **error);
115
116 static gboolean g_unix_input_stream_pollable_can_poll      (GPollableInputStream *stream);
117 static gboolean g_unix_input_stream_pollable_is_readable   (GPollableInputStream *stream);
118 static GSource *g_unix_input_stream_pollable_create_source (GPollableInputStream *stream,
119                                                             GCancellable         *cancellable);
120
121 static void
122 g_unix_input_stream_finalize (GObject *object)
123 {
124   G_OBJECT_CLASS (g_unix_input_stream_parent_class)->finalize (object);
125 }
126
127 static void
128 g_unix_input_stream_class_init (GUnixInputStreamClass *klass)
129 {
130   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
131   GInputStreamClass *stream_class = G_INPUT_STREAM_CLASS (klass);
132   
133   g_type_class_add_private (klass, sizeof (GUnixInputStreamPrivate));
134
135   gobject_class->get_property = g_unix_input_stream_get_property;
136   gobject_class->set_property = g_unix_input_stream_set_property;
137   gobject_class->finalize = g_unix_input_stream_finalize;
138
139   stream_class->read_fn = g_unix_input_stream_read;
140   stream_class->close_fn = g_unix_input_stream_close;
141   if (0)
142     {
143       /* TODO: Implement instead of using fallbacks */
144       stream_class->skip_async = g_unix_input_stream_skip_async;
145       stream_class->skip_finish = g_unix_input_stream_skip_finish;
146     }
147   stream_class->close_async = g_unix_input_stream_close_async;
148   stream_class->close_finish = g_unix_input_stream_close_finish;
149
150   /**
151    * GUnixInputStream:fd:
152    *
153    * The file descriptor that the stream reads from.
154    *
155    * Since: 2.20
156    */
157   g_object_class_install_property (gobject_class,
158                                    PROP_FD,
159                                    g_param_spec_int ("fd",
160                                                      P_("File descriptor"),
161                                                      P_("The file descriptor to read from"),
162                                                      G_MININT, G_MAXINT, -1,
163                                                      G_PARAM_READABLE | G_PARAM_WRITABLE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_NAME | G_PARAM_STATIC_NICK | G_PARAM_STATIC_BLURB));
164
165   /**
166    * GUnixInputStream:close-fd:
167    *
168    * Whether to close the file descriptor when the stream is closed.
169    *
170    * Since: 2.20
171    */
172   g_object_class_install_property (gobject_class,
173                                    PROP_CLOSE_FD,
174                                    g_param_spec_boolean ("close-fd",
175                                                          P_("Close file descriptor"),
176                                                          P_("Whether to close the file descriptor when the stream is closed"),
177                                                          TRUE,
178                                                          G_PARAM_READABLE | G_PARAM_WRITABLE | G_PARAM_STATIC_NAME | G_PARAM_STATIC_NICK | G_PARAM_STATIC_BLURB));
179 }
180
181 static void
182 g_unix_input_stream_pollable_iface_init (GPollableInputStreamInterface *iface)
183 {
184   iface->can_poll = g_unix_input_stream_pollable_can_poll;
185   iface->is_readable = g_unix_input_stream_pollable_is_readable;
186   iface->create_source = g_unix_input_stream_pollable_create_source;
187 }
188
189 static void
190 g_unix_input_stream_file_descriptor_based_iface_init (GFileDescriptorBasedIface *iface)
191 {
192   iface->get_fd = (int (*) (GFileDescriptorBased *))g_unix_input_stream_get_fd;
193 }
194
195 static void
196 g_unix_input_stream_set_property (GObject         *object,
197                                   guint            prop_id,
198                                   const GValue    *value,
199                                   GParamSpec      *pspec)
200 {
201   GUnixInputStream *unix_stream;
202   
203   unix_stream = G_UNIX_INPUT_STREAM (object);
204
205   switch (prop_id)
206     {
207     case PROP_FD:
208       unix_stream->priv->fd = g_value_get_int (value);
209       if (lseek (unix_stream->priv->fd, 0, SEEK_CUR) == -1 && errno == ESPIPE)
210         unix_stream->priv->is_pipe_or_socket = TRUE;
211       else
212         unix_stream->priv->is_pipe_or_socket = FALSE;
213       break;
214     case PROP_CLOSE_FD:
215       unix_stream->priv->close_fd = g_value_get_boolean (value);
216       break;
217     default:
218       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
219       break;
220     }
221 }
222
223 static void
224 g_unix_input_stream_get_property (GObject    *object,
225                                   guint       prop_id,
226                                   GValue     *value,
227                                   GParamSpec *pspec)
228 {
229   GUnixInputStream *unix_stream;
230
231   unix_stream = G_UNIX_INPUT_STREAM (object);
232
233   switch (prop_id)
234     {
235     case PROP_FD:
236       g_value_set_int (value, unix_stream->priv->fd);
237       break;
238     case PROP_CLOSE_FD:
239       g_value_set_boolean (value, unix_stream->priv->close_fd);
240       break;
241     default:
242       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
243     }
244 }
245
246 static void
247 g_unix_input_stream_init (GUnixInputStream *unix_stream)
248 {
249   unix_stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (unix_stream,
250                                                    G_TYPE_UNIX_INPUT_STREAM,
251                                                    GUnixInputStreamPrivate);
252
253   unix_stream->priv->fd = -1;
254   unix_stream->priv->close_fd = TRUE;
255 }
256
257 /**
258  * g_unix_input_stream_new:
259  * @fd: a UNIX file descriptor
260  * @close_fd: %TRUE to close the file descriptor when done
261  * 
262  * Creates a new #GUnixInputStream for the given @fd. 
263  *
264  * If @close_fd is %TRUE, the file descriptor will be closed 
265  * when the stream is closed.
266  * 
267  * Returns: a new #GUnixInputStream
268  **/
269 GInputStream *
270 g_unix_input_stream_new (gint     fd,
271                          gboolean close_fd)
272 {
273   GUnixInputStream *stream;
274
275   g_return_val_if_fail (fd != -1, NULL);
276
277   stream = g_object_new (G_TYPE_UNIX_INPUT_STREAM,
278                          "fd", fd,
279                          "close-fd", close_fd,
280                          NULL);
281
282   return G_INPUT_STREAM (stream);
283 }
284
285 /**
286  * g_unix_input_stream_set_close_fd:
287  * @stream: a #GUnixInputStream
288  * @close_fd: %TRUE to close the file descriptor when done
289  *
290  * Sets whether the file descriptor of @stream shall be closed
291  * when the stream is closed.
292  *
293  * Since: 2.20
294  */
295 void
296 g_unix_input_stream_set_close_fd (GUnixInputStream *stream,
297                                   gboolean          close_fd)
298 {
299   g_return_if_fail (G_IS_UNIX_INPUT_STREAM (stream));
300
301   close_fd = close_fd != FALSE;
302   if (stream->priv->close_fd != close_fd)
303     {
304       stream->priv->close_fd = close_fd;
305       g_object_notify (G_OBJECT (stream), "close-fd");
306     }
307 }
308
309 /**
310  * g_unix_input_stream_get_close_fd:
311  * @stream: a #GUnixInputStream
312  *
313  * Returns whether the file descriptor of @stream will be
314  * closed when the stream is closed.
315  *
316  * Return value: %TRUE if the file descriptor is closed when done
317  *
318  * Since: 2.20
319  */
320 gboolean
321 g_unix_input_stream_get_close_fd (GUnixInputStream *stream)
322 {
323   g_return_val_if_fail (G_IS_UNIX_INPUT_STREAM (stream), FALSE);
324
325   return stream->priv->close_fd;
326 }
327
328 /**
329  * g_unix_input_stream_get_fd:
330  * @stream: a #GUnixInputStream
331  *
332  * Return the UNIX file descriptor that the stream reads from.
333  *
334  * Return value: The file descriptor of @stream
335  *
336  * Since: 2.20
337  */
338 gint
339 g_unix_input_stream_get_fd (GUnixInputStream *stream)
340 {
341   g_return_val_if_fail (G_IS_UNIX_INPUT_STREAM (stream), -1);
342   
343   return stream->priv->fd;
344 }
345
346 static gssize
347 g_unix_input_stream_read (GInputStream  *stream,
348                           void          *buffer,
349                           gsize          count,
350                           GCancellable  *cancellable,
351                           GError       **error)
352 {
353   GUnixInputStream *unix_stream;
354   gssize res = -1;
355   GPollFD poll_fds[2];
356   int nfds;
357   int poll_ret;
358
359   unix_stream = G_UNIX_INPUT_STREAM (stream);
360
361   poll_fds[0].fd = unix_stream->priv->fd;
362   poll_fds[0].events = G_IO_IN;
363   if (unix_stream->priv->is_pipe_or_socket &&
364       g_cancellable_make_pollfd (cancellable, &poll_fds[1]))
365     nfds = 2;
366   else
367     nfds = 1;
368
369   while (1)
370     {
371       poll_fds[0].revents = poll_fds[1].revents = 0;
372       do
373         poll_ret = g_poll (poll_fds, nfds, -1);
374       while (poll_ret == -1 && errno == EINTR);
375
376       if (poll_ret == -1)
377         {
378           int errsv = errno;
379
380           g_set_error (error, G_IO_ERROR,
381                        g_io_error_from_errno (errsv),
382                        _("Error reading from file descriptor: %s"),
383                        g_strerror (errsv));
384           break;
385         }
386
387       if (g_cancellable_set_error_if_cancelled (cancellable, error))
388         break;
389
390       if (!poll_fds[0].revents)
391         continue;
392
393       res = read (unix_stream->priv->fd, buffer, count);
394       if (res == -1)
395         {
396           int errsv = errno;
397
398           if (errsv == EINTR || errsv == EAGAIN)
399             continue;
400
401           g_set_error (error, G_IO_ERROR,
402                        g_io_error_from_errno (errsv),
403                        _("Error reading from file descriptor: %s"),
404                        g_strerror (errsv));
405         }
406
407       break;
408     }
409
410   if (nfds == 2)
411     g_cancellable_release_fd (cancellable);
412   return res;
413 }
414
415 static gboolean
416 g_unix_input_stream_close (GInputStream  *stream,
417                            GCancellable  *cancellable,
418                            GError       **error)
419 {
420   GUnixInputStream *unix_stream;
421   int res;
422
423   unix_stream = G_UNIX_INPUT_STREAM (stream);
424
425   if (!unix_stream->priv->close_fd)
426     return TRUE;
427   
428   while (1)
429     {
430       /* This might block during the close. Doesn't seem to be a way to avoid it though. */
431       res = close (unix_stream->priv->fd);
432       if (res == -1)
433         {
434           int errsv = errno;
435
436           g_set_error (error, G_IO_ERROR,
437                        g_io_error_from_errno (errsv),
438                        _("Error closing file descriptor: %s"),
439                        g_strerror (errsv));
440         }
441       break;
442     }
443   
444   return res != -1;
445 }
446
447 static void
448 g_unix_input_stream_skip_async (GInputStream        *stream,
449                                 gsize                count,
450                                 int                  io_priority,
451                                 GCancellable        *cancellable,
452                                 GAsyncReadyCallback  callback,
453                                 gpointer             data)
454 {
455   g_warn_if_reached ();
456   /* TODO: Not implemented */
457 }
458
459 static gssize
460 g_unix_input_stream_skip_finish  (GInputStream  *stream,
461                                   GAsyncResult  *result,
462                                   GError       **error)
463 {
464   g_warn_if_reached ();
465   return 0;
466   /* TODO: Not implemented */
467 }
468
469
470 typedef struct {
471   GInputStream *stream;
472   GAsyncReadyCallback callback;
473   gpointer user_data;
474 } CloseAsyncData;
475
476 static void
477 close_async_data_free (gpointer _data)
478 {
479   CloseAsyncData *data = _data;
480
481   g_free (data);
482 }
483
484 static gboolean
485 close_async_cb (CloseAsyncData *data)
486 {
487   GUnixInputStream *unix_stream;
488   GSimpleAsyncResult *simple;
489   GError *error = NULL;
490   gboolean result;
491   int res;
492
493   unix_stream = G_UNIX_INPUT_STREAM (data->stream);
494
495   if (!unix_stream->priv->close_fd)
496     {
497       result = TRUE;
498       goto out;
499     }
500   
501   while (1)
502     {
503       res = close (unix_stream->priv->fd);
504       if (res == -1)
505         {
506           int errsv = errno;
507
508           g_set_error (&error, G_IO_ERROR,
509                        g_io_error_from_errno (errsv),
510                        _("Error closing file descriptor: %s"),
511                        g_strerror (errsv));
512         }
513       break;
514     }
515   
516   result = res != -1;
517
518  out:
519   simple = g_simple_async_result_new (G_OBJECT (data->stream),
520                                       data->callback,
521                                       data->user_data,
522                                       g_unix_input_stream_close_async);
523
524   if (!result)
525     g_simple_async_result_take_error (simple, error);
526
527   /* Complete immediately, not in idle, since we're already in a mainloop callout */
528   g_simple_async_result_complete (simple);
529   g_object_unref (simple);
530   
531   return FALSE;
532 }
533
534 static void
535 g_unix_input_stream_close_async (GInputStream        *stream,
536                                  int                  io_priority,
537                                  GCancellable        *cancellable,
538                                  GAsyncReadyCallback  callback,
539                                  gpointer             user_data)
540 {
541   GSource *idle;
542   CloseAsyncData *data;
543
544   data = g_new0 (CloseAsyncData, 1);
545
546   data->stream = stream;
547   data->callback = callback;
548   data->user_data = user_data;
549   
550   idle = g_idle_source_new ();
551   g_source_set_callback (idle, (GSourceFunc)close_async_cb, data, close_async_data_free);
552   g_source_attach (idle, g_main_context_get_thread_default ());
553   g_source_unref (idle);
554 }
555
556 static gboolean
557 g_unix_input_stream_close_finish (GInputStream  *stream,
558                                   GAsyncResult  *result,
559                                   GError       **error)
560 {
561   /* Failures handled in generic close_finish code */
562   return TRUE;
563 }
564
565 static gboolean
566 g_unix_input_stream_pollable_can_poll (GPollableInputStream *stream)
567 {
568   return G_UNIX_INPUT_STREAM (stream)->priv->is_pipe_or_socket;
569 }
570
571 static gboolean
572 g_unix_input_stream_pollable_is_readable (GPollableInputStream *stream)
573 {
574   GUnixInputStream *unix_stream = G_UNIX_INPUT_STREAM (stream);
575   GPollFD poll_fd;
576   gint result;
577
578   poll_fd.fd = unix_stream->priv->fd;
579   poll_fd.events = G_IO_IN;
580
581   do
582     result = g_poll (&poll_fd, 1, 0);
583   while (result == -1 && errno == EINTR);
584
585   return poll_fd.revents != 0;
586 }
587
588 static GSource *
589 g_unix_input_stream_pollable_create_source (GPollableInputStream *stream,
590                                             GCancellable         *cancellable)
591 {
592   GUnixInputStream *unix_stream = G_UNIX_INPUT_STREAM (stream);
593   GSource *inner_source, *pollable_source;
594
595   pollable_source = g_pollable_source_new (G_OBJECT (stream));
596
597   inner_source = _g_fd_source_new (unix_stream->priv->fd, G_IO_IN, cancellable);
598   g_source_set_dummy_callback (inner_source);
599   g_source_add_child_source (pollable_source, inner_source);
600   g_source_unref (inner_source);
601
602   return pollable_source;
603 }