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 "glibintl.h"
  40
  41
  42 /**
  43  * SECTION:gunixinputstream
  44  * @short_description: Streaming input operations for UNIX file descriptors
  45  * @include: gio/gunixinputstream.h
  46  * @see_also: #GInputStream
  47  *
  48  * #GUnixInputStream implements #GInputStream for reading from a
  49  * UNIX file descriptor, including asynchronous operations. The file
  50  * descriptor must be selectable, so it doesn't work with opened files.
  51  *
  52  * Note that <filename>&lt;gio/gunixinputstream.h&gt;</filename> belongs
  53  * to the UNIX-specific GIO interfaces, thus you have to use the
  54  * <filename>gio-unix-2.0.pc</filename> pkg-config file when using it.
  55  */
  56
  57 enum {
  58   PROP_0,
  59   PROP_FD,
  60   PROP_CLOSE_FD
  61 };
  62
  63 static void g_unix_input_stream_pollable_iface_init (GPollableInputStreamInterface *iface);
  64
  65 G_DEFINE_TYPE_WITH_CODE (GUnixInputStream, g_unix_input_stream, G_TYPE_INPUT_STREAM,
  66                          G_IMPLEMENT_INTERFACE (G_TYPE_POLLABLE_INPUT_STREAM,
  67                                                 g_unix_input_stream_pollable_iface_init)
  68                          );
  69
  70 struct _GUnixInputStreamPrivate {
  71   int fd;
  72   gboolean close_fd;
  73 };
  74
  75 static void     g_unix_input_stream_set_property (GObject              *object,
  76                                                   guint                 prop_id,
  77                                                   const GValue         *value,
  78                                                   GParamSpec           *pspec);
  79 static void     g_unix_input_stream_get_property (GObject              *object,
  80                                                   guint                 prop_id,
  81                                                   GValue               *value,
  82                                                   GParamSpec           *pspec);
  83 static gssize   g_unix_input_stream_read         (GInputStream         *stream,
  84                                                   void                 *buffer,
  85                                                   gsize                 count,
  86                                                   GCancellable         *cancellable,
  87                                                   GError              **error);
  88 static gboolean g_unix_input_stream_close        (GInputStream         *stream,
  89                                                   GCancellable         *cancellable,
  90                                                   GError              **error);
  91 static void     g_unix_input_stream_read_async   (GInputStream         *stream,
  92                                                   void                 *buffer,
  93                                                   gsize                 count,
  94                                                   int                   io_priority,
  95                                                   GCancellable         *cancellable,
  96                                                   GAsyncReadyCallback   callback,
  97                                                   gpointer              data);
  98 static gssize   g_unix_input_stream_read_finish  (GInputStream         *stream,
  99                                                   GAsyncResult         *result,
 100                                                   GError              **error);
 101 static void     g_unix_input_stream_skip_async   (GInputStream         *stream,
 102                                                   gsize                 count,
 103                                                   int                   io_priority,
 104                                                   GCancellable         *cancellable,
 105                                                   GAsyncReadyCallback   callback,
 106                                                   gpointer              data);
 107 static gssize   g_unix_input_stream_skip_finish  (GInputStream         *stream,
 108                                                   GAsyncResult         *result,
 109                                                   GError              **error);
 110 static void     g_unix_input_stream_close_async  (GInputStream         *stream,
 111                                                   int                   io_priority,
 112                                                   GCancellable         *cancellable,
 113                                                   GAsyncReadyCallback   callback,
 114                                                   gpointer              data);
 115 static gboolean g_unix_input_stream_close_finish (GInputStream         *stream,
 116                                                   GAsyncResult         *result,
 117                                                   GError              **error);
 118
 119 static gboolean g_unix_input_stream_pollable_is_readable   (GPollableInputStream *stream);
 120 static GSource *g_unix_input_stream_pollable_create_source (GPollableInputStream *stream,
 121                                                             GCancellable         *cancellable);
 122
 123 static void
 124 g_unix_input_stream_finalize (GObject *object)
 125 {
 126   G_OBJECT_CLASS (g_unix_input_stream_parent_class)->finalize (object);
 127 }
 128
 129 static void
 130 g_unix_input_stream_class_init (GUnixInputStreamClass *klass)
 131 {
 132   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 133   GInputStreamClass *stream_class = G_INPUT_STREAM_CLASS (klass);
 134
 135   g_type_class_add_private (klass, sizeof (GUnixInputStreamPrivate));
 136
 137   gobject_class->get_property = g_unix_input_stream_get_property;
 138   gobject_class->set_property = g_unix_input_stream_set_property;
 139   gobject_class->finalize = g_unix_input_stream_finalize;
 140
 141   stream_class->read_fn = g_unix_input_stream_read;
 142   stream_class->close_fn = g_unix_input_stream_close;
 143   stream_class->read_async = g_unix_input_stream_read_async;
 144   stream_class->read_finish = g_unix_input_stream_read_finish;
 145   if (0)
 146     {
 147       /* TODO: Implement instead of using fallbacks */
 148       stream_class->skip_async = g_unix_input_stream_skip_async;
 149       stream_class->skip_finish = g_unix_input_stream_skip_finish;
 150     }
 151   stream_class->close_async = g_unix_input_stream_close_async;
 152   stream_class->close_finish = g_unix_input_stream_close_finish;
 153
 154   /**
 155    * GUnixInputStream:fd:
 156    *
 157    * The file descriptor that the stream reads from.
 158    *
 159    * Since: 2.20
 160    */
 161   g_object_class_install_property (gobject_class,
 162                                    PROP_FD,
 163                                    g_param_spec_int ("fd",
 164                                                      P_("File descriptor"),
 165                                                      P_("The file descriptor to read from"),
 166                                                      G_MININT, G_MAXINT, -1,
 167                                                      G_PARAM_READABLE | G_PARAM_WRITABLE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_NAME | G_PARAM_STATIC_NICK | G_PARAM_STATIC_BLURB));
 168
 169   /**
 170    * GUnixInputStream:close-fd:
 171    *
 172    * Whether to close the file descriptor when the stream is closed.
 173    *
 174    * Since: 2.20
 175    */
 176   g_object_class_install_property (gobject_class,
 177                                    PROP_CLOSE_FD,
 178                                    g_param_spec_boolean ("close-fd",
 179                                                          P_("Close file descriptor"),
 180                                                          P_("Whether to close the file descriptor when the stream is closed"),
 181                                                          TRUE,
 182                                                          G_PARAM_READABLE | G_PARAM_WRITABLE | G_PARAM_STATIC_NAME | G_PARAM_STATIC_NICK | G_PARAM_STATIC_BLURB));
 183 }
 184
 185 static void
 186 g_unix_input_stream_pollable_iface_init (GPollableInputStreamInterface *iface)
 187 {
 188   iface->is_readable = g_unix_input_stream_pollable_is_readable;
 189   iface->create_source = g_unix_input_stream_pollable_create_source;
 190 }
 191
 192 static void
 193 g_unix_input_stream_set_property (GObject         *object,
 194                                   guint            prop_id,
 195                                   const GValue    *value,
 196                                   GParamSpec      *pspec)
 197 {
 198   GUnixInputStream *unix_stream;
 199
 200   unix_stream = G_UNIX_INPUT_STREAM (object);
 201
 202   switch (prop_id)
 203     {
 204     case PROP_FD:
 205       unix_stream->priv->fd = g_value_get_int (value);
 206       break;
 207     case PROP_CLOSE_FD:
 208       unix_stream->priv->close_fd = g_value_get_boolean (value);
 209       break;
 210     default:
 211       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 212       break;
 213     }
 214 }
 215
 216 static void
 217 g_unix_input_stream_get_property (GObject    *object,
 218                                   guint       prop_id,
 219                                   GValue     *value,
 220                                   GParamSpec *pspec)
 221 {
 222   GUnixInputStream *unix_stream;
 223
 224   unix_stream = G_UNIX_INPUT_STREAM (object);
 225
 226   switch (prop_id)
 227     {
 228     case PROP_FD:
 229       g_value_set_int (value, unix_stream->priv->fd);
 230       break;
 231     case PROP_CLOSE_FD:
 232       g_value_set_boolean (value, unix_stream->priv->close_fd);
 233       break;
 234     default:
 235       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 236     }
 237 }
 238
 239 static void
 240 g_unix_input_stream_init (GUnixInputStream *unix_stream)
 241 {
 242   unix_stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (unix_stream,
 243                                                    G_TYPE_UNIX_INPUT_STREAM,
 244                                                    GUnixInputStreamPrivate);
 245
 246   unix_stream->priv->fd = -1;
 247   unix_stream->priv->close_fd = TRUE;
 248 }
 249
 250 /**
 251  * g_unix_input_stream_new:
 252  * @fd: a UNIX file descriptor
 253  * @close_fd: %TRUE to close the file descriptor when done
 254  *
 255  * Creates a new #GUnixInputStream for the given @fd.
 256  *
 257  * If @close_fd is %TRUE, the file descriptor will be closed
 258  * when the stream is closed.
 259  *
 260  * Returns: a new #GUnixInputStream
 261  **/
 262 GInputStream *
 263 g_unix_input_stream_new (gint     fd,
 264                          gboolean close_fd)
 265 {
 266   GUnixInputStream *stream;
 267
 268   g_return_val_if_fail (fd != -1, NULL);
 269
 270   stream = g_object_new (G_TYPE_UNIX_INPUT_STREAM,
 271                          "fd", fd,
 272                          "close-fd", close_fd,
 273                          NULL);
 274
 275   return G_INPUT_STREAM (stream);
 276 }
 277
 278 /**
 279  * g_unix_input_stream_set_close_fd:
 280  * @stream: a #GUnixInputStream
 281  * @close_fd: %TRUE to close the file descriptor when done
 282  *
 283  * Sets whether the file descriptor of @stream shall be closed
 284  * when the stream is closed.
 285  *
 286  * Since: 2.20
 287  */
 288 void
 289 g_unix_input_stream_set_close_fd (GUnixInputStream *stream,
 290                                   gboolean          close_fd)
 291 {
 292   g_return_if_fail (G_IS_UNIX_INPUT_STREAM (stream));
 293
 294   close_fd = close_fd != FALSE;
 295   if (stream->priv->close_fd != close_fd)
 296     {
 297       stream->priv->close_fd = close_fd;
 298       g_object_notify (G_OBJECT (stream), "close-fd");
 299     }
 300 }
 301
 302 /**
 303  * g_unix_input_stream_get_close_fd:
 304  * @stream: a #GUnixInputStream
 305  *
 306  * Returns whether the file descriptor of @stream will be
 307  * closed when the stream is closed.
 308  *
 309  * Return value: %TRUE if the file descriptor is closed when done
 310  *
 311  * Since: 2.20
 312  */
 313 gboolean
 314 g_unix_input_stream_get_close_fd (GUnixInputStream *stream)
 315 {
 316   g_return_val_if_fail (G_IS_UNIX_INPUT_STREAM (stream), FALSE);
 317
 318   return stream->priv->close_fd;
 319 }
 320
 321 /**
 322  * g_unix_input_stream_get_fd:
 323  * @stream: a #GUnixInputStream
 324  *
 325  * Return the UNIX file descriptor that the stream reads from.
 326  *
 327  * Return value: The file descriptor of @stream
 328  *
 329  * Since: 2.20
 330  */
 331 gint
 332 g_unix_input_stream_get_fd (GUnixInputStream *stream)
 333 {
 334   g_return_val_if_fail (G_IS_UNIX_INPUT_STREAM (stream), -1);
 335
 336   return stream->priv->fd;
 337 }
 338
 339 static gssize
 340 g_unix_input_stream_read (GInputStream  *stream,
 341                           void          *buffer,
 342                           gsize          count,
 343                           GCancellable  *cancellable,
 344                           GError       **error)
 345 {
 346   GUnixInputStream *unix_stream;
 347   gssize res;
 348   GPollFD poll_fds[2];
 349   int poll_ret;
 350
 351   unix_stream = G_UNIX_INPUT_STREAM (stream);
 352
 353   if (g_cancellable_make_pollfd (cancellable, &poll_fds[1]))
 354     {
 355       poll_fds[0].fd = unix_stream->priv->fd;
 356       poll_fds[0].events = G_IO_IN;
 357       do
 358         poll_ret = g_poll (poll_fds, 2, -1);
 359       while (poll_ret == -1 && errno == EINTR);
 360       g_cancellable_release_fd (cancellable);
 361
 362       if (poll_ret == -1)
 363         {
 364           int errsv = errno;
 365
 366           g_set_error (error, G_IO_ERROR,
 367                        g_io_error_from_errno (errsv),
 368                        _("Error reading from unix: %s"),
 369                        g_strerror (errsv));
 370           return -1;
 371         }
 372     }
 373
 374   while (1)
 375     {
 376       if (g_cancellable_set_error_if_cancelled (cancellable, error))
 377         return -1;
 378       res = read (unix_stream->priv->fd, buffer, count);
 379       if (res == -1)
 380         {
 381           int errsv = errno;
 382
 383           if (errsv == EINTR)
 384             continue;
 385
 386           g_set_error (error, G_IO_ERROR,
 387                        g_io_error_from_errno (errsv),
 388                        _("Error reading from unix: %s"),
 389                        g_strerror (errsv));
 390         }
 391
 392       break;
 393     }
 394
 395   return res;
 396 }
 397
 398 static gboolean
 399 g_unix_input_stream_close (GInputStream  *stream,
 400                            GCancellable  *cancellable,
 401                            GError       **error)
 402 {
 403   GUnixInputStream *unix_stream;
 404   int res;
 405
 406   unix_stream = G_UNIX_INPUT_STREAM (stream);
 407
 408   if (!unix_stream->priv->close_fd)
 409     return TRUE;
 410
 411   while (1)
 412     {
 413       /* This might block during the close. Doesn't seem to be a way to avoid it though. */
 414       res = close (unix_stream->priv->fd);
 415       if (res == -1)
 416         {
 417           int errsv = errno;
 418
 419           g_set_error (error, G_IO_ERROR,
 420                        g_io_error_from_errno (errsv),
 421                        _("Error closing unix: %s"),
 422                        g_strerror (errsv));
 423         }
 424       break;
 425     }
 426
 427   return res != -1;
 428 }
 429
 430 typedef struct {
 431   gsize count;
 432   void *buffer;
 433   GAsyncReadyCallback callback;
 434   gpointer user_data;
 435   GCancellable *cancellable;
 436   GUnixInputStream *stream;
 437 } ReadAsyncData;
 438
 439 static gboolean
 440 read_async_cb (int            fd,
 441                GIOCondition   condition,
 442                ReadAsyncData *data)
 443 {
 444   GSimpleAsyncResult *simple;
 445   GError *error = NULL;
 446   gssize count_read;
 447
 448   /* We know that we can read from fd once without blocking */
 449   while (1)
 450     {
 451       if (g_cancellable_set_error_if_cancelled (data->cancellable, &error))
 452         {
 453           count_read = -1;
 454           break;
 455         }
 456       count_read = read (data->stream->priv->fd, data->buffer, data->count);
 457       if (count_read == -1)
 458         {
 459           int errsv = errno;
 460
 461           if (errsv == EINTR)
 462             continue;
 463
 464           g_set_error (&error, G_IO_ERROR,
 465                        g_io_error_from_errno (errsv),
 466                        _("Error reading from unix: %s"),
 467                        g_strerror (errsv));
 468         }
 469       break;
 470     }
 471
 472   simple = g_simple_async_result_new (G_OBJECT (data->stream),
 473                                       data->callback,
 474                                       data->user_data,
 475                                       g_unix_input_stream_read_async);
 476
 477   g_simple_async_result_set_op_res_gssize (simple, count_read);
 478
 479   if (count_read == -1)
 480     g_simple_async_result_take_error (simple, error);
 481
 482   /* Complete immediately, not in idle, since we're already in a mainloop callout */
 483   g_simple_async_result_complete (simple);
 484   g_object_unref (simple);
 485
 486   return FALSE;
 487 }
 488
 489 static void
 490 g_unix_input_stream_read_async (GInputStream        *stream,
 491                                 void                *buffer,
 492                                 gsize                count,
 493                                 int                  io_priority,
 494                                 GCancellable        *cancellable,
 495                                 GAsyncReadyCallback  callback,
 496                                 gpointer             user_data)
 497 {
 498   GSource *source;
 499   GUnixInputStream *unix_stream;
 500   ReadAsyncData *data;
 501
 502   unix_stream = G_UNIX_INPUT_STREAM (stream);
 503
 504   data = g_new0 (ReadAsyncData, 1);
 505   data->count = count;
 506   data->buffer = buffer;
 507   data->callback = callback;
 508   data->user_data = user_data;
 509   data->cancellable = cancellable;
 510   data->stream = unix_stream;
 511
 512   source = _g_fd_source_new (unix_stream->priv->fd,
 513                              G_IO_IN,
 514                              cancellable);
 515   g_source_set_name (source, "GUnixInputStream");
 516
 517   g_source_set_callback (source, (GSourceFunc)read_async_cb, data, g_free);
 518   g_source_attach (source, g_main_context_get_thread_default ());
 519
 520   g_source_unref (source);
 521 }
 522
 523 static gssize
 524 g_unix_input_stream_read_finish (GInputStream  *stream,
 525                                  GAsyncResult  *result,
 526                                  GError       **error)
 527 {
 528   GSimpleAsyncResult *simple;
 529   gssize nread;
 530
 531   simple = G_SIMPLE_ASYNC_RESULT (result);
 532   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_unix_input_stream_read_async);
 533
 534   nread = g_simple_async_result_get_op_res_gssize (simple);
 535   return nread;
 536 }
 537
 538 static void
 539 g_unix_input_stream_skip_async (GInputStream        *stream,
 540                                 gsize                count,
 541                                 int                  io_priority,
 542                                 GCancellable        *cancellable,
 543                                 GAsyncReadyCallback  callback,
 544                                 gpointer             data)
 545 {
 546   g_warn_if_reached ();
 547   /* TODO: Not implemented */
 548 }
 549
 550 static gssize
 551 g_unix_input_stream_skip_finish  (GInputStream  *stream,
 552                                   GAsyncResult  *result,
 553                                   GError       **error)
 554 {
 555   g_warn_if_reached ();
 556   return 0;
 557   /* TODO: Not implemented */
 558 }
 559
 560
 561 typedef struct {
 562   GInputStream *stream;
 563   GAsyncReadyCallback callback;
 564   gpointer user_data;
 565 } CloseAsyncData;
 566
 567 static void
 568 close_async_data_free (gpointer _data)
 569 {
 570   CloseAsyncData *data = _data;
 571
 572   g_free (data);
 573 }
 574
 575 static gboolean
 576 close_async_cb (CloseAsyncData *data)
 577 {
 578   GUnixInputStream *unix_stream;
 579   GSimpleAsyncResult *simple;
 580   GError *error = NULL;
 581   gboolean result;
 582   int res;
 583
 584   unix_stream = G_UNIX_INPUT_STREAM (data->stream);
 585
 586   if (!unix_stream->priv->close_fd)
 587     {
 588       result = TRUE;
 589       goto out;
 590     }
 591
 592   while (1)
 593     {
 594       res = close (unix_stream->priv->fd);
 595       if (res == -1)
 596         {
 597           int errsv = errno;
 598
 599           g_set_error (&error, G_IO_ERROR,
 600                        g_io_error_from_errno (errsv),
 601                        _("Error closing unix: %s"),
 602                        g_strerror (errsv));
 603         }
 604       break;
 605     }
 606
 607   result = res != -1;
 608
 609  out:
 610   simple = g_simple_async_result_new (G_OBJECT (data->stream),
 611                                       data->callback,
 612                                       data->user_data,
 613                                       g_unix_input_stream_close_async);
 614
 615   if (!result)
 616     g_simple_async_result_take_error (simple, error);
 617
 618   /* Complete immediately, not in idle, since we're already in a mainloop callout */
 619   g_simple_async_result_complete (simple);
 620   g_object_unref (simple);
 621
 622   return FALSE;
 623 }
 624
 625 static void
 626 g_unix_input_stream_close_async (GInputStream        *stream,
 627                                  int                  io_priority,
 628                                  GCancellable        *cancellable,
 629                                  GAsyncReadyCallback  callback,
 630                                  gpointer             user_data)
 631 {
 632   GSource *idle;
 633   CloseAsyncData *data;
 634
 635   data = g_new0 (CloseAsyncData, 1);
 636
 637   data->stream = stream;
 638   data->callback = callback;
 639   data->user_data = user_data;
 640
 641   idle = g_idle_source_new ();
 642   g_source_set_callback (idle, (GSourceFunc)close_async_cb, data, close_async_data_free);
 643   g_source_attach (idle, g_main_context_get_thread_default ());
 644   g_source_unref (idle);
 645 }
 646
 647 static gboolean
 648 g_unix_input_stream_close_finish (GInputStream  *stream,
 649                                   GAsyncResult  *result,
 650                                   GError       **error)
 651 {
 652   /* Failures handled in generic close_finish code */
 653   return TRUE;
 654 }
 655
 656 static gboolean
 657 g_unix_input_stream_pollable_is_readable (GPollableInputStream *stream)
 658 {
 659   GUnixInputStream *unix_stream = G_UNIX_INPUT_STREAM (stream);
 660   GPollFD poll_fd;
 661   gint result;
 662
 663   poll_fd.fd = unix_stream->priv->fd;
 664   poll_fd.events = G_IO_IN;
 665
 666   do
 667     result = g_poll (&poll_fd, 1, 0);
 668   while (result == -1 && errno == EINTR);
 669
 670   return poll_fd.revents != 0;
 671 }
 672
 673 static GSource *
 674 g_unix_input_stream_pollable_create_source (GPollableInputStream *stream,
 675                                             GCancellable         *cancellable)
 676 {
 677   GUnixInputStream *unix_stream = G_UNIX_INPUT_STREAM (stream);
 678   GSource *inner_source, *pollable_source;
 679
 680   pollable_source = g_pollable_source_new (G_OBJECT (stream));
 681
 682   inner_source = _g_fd_source_new (unix_stream->priv->fd, G_IO_IN, cancellable);
 683   g_source_set_dummy_callback (inner_source);
 684   g_source_add_child_source (pollable_source, inner_source);
 685   g_source_unref (inner_source);
 686
 687   return pollable_source;
 688 }