gio/gdbusprivate.c

   1 /* GDBus - GLib D-Bus Library
   2  *
   3  * Copyright (C) 2008-2010 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.1 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, see <http://www.gnu.org/licenses/>.
  17  *
  18  * Author: David Zeuthen <davidz@redhat.com>
  19  */
  20
  21 #include "config.h"
  22
  23 #include <stdlib.h>
  24 #include <string.h>
  25
  26 #include "giotypes.h"
  27 #include "gioenumtypes.h"
  28 #include "gsocket.h"
  29 #include "gdbusauthobserver.h"
  30 #include "gdbusprivate.h"
  31 #include "gdbusmessage.h"
  32 #include "gdbusconnection.h"
  33 #include "gdbusproxy.h"
  34 #include "gdbuserror.h"
  35 #include "gdbusintrospection.h"
  36 #include "gdbusdaemon.h"
  37 #include "giomodule-priv.h"
  38 #include "gtask.h"
  39 #include "ginputstream.h"
  40 #include "gmemoryinputstream.h"
  41 #include "giostream.h"
  42 #include "glib/gstdio.h"
  43 #include "gsocketaddress.h"
  44 #include "gsocketcontrolmessage.h"
  45 #include "gsocketconnection.h"
  46 #include "gsocketoutputstream.h"
  47
  48 #ifdef G_OS_UNIX
  49 #include "gunixfdmessage.h"
  50 #include "gunixconnection.h"
  51 #include "gunixcredentialsmessage.h"
  52 #endif
  53
  54 #ifdef G_OS_WIN32
  55 #include <windows.h>
  56 #include <io.h>
  57 #include <conio.h>
  58 #endif
  59
  60 #include "glibintl.h"
  61
  62 static gboolean _g_dbus_worker_do_initial_read (gpointer data);
  63 static void schedule_pending_close (GDBusWorker *worker);
  64
  65 /* ---------------------------------------------------------------------------------------------------- */
  66
  67 gchar *
  68 _g_dbus_hexdump (const gchar *data, gsize len, guint indent)
  69 {
  70  guint n, m;
  71  GString *ret;
  72
  73  ret = g_string_new (NULL);
  74
  75  for (n = 0; n < len; n += 16)
  76    {
  77      g_string_append_printf (ret, "%*s%04x: ", indent, "", n);
  78
  79      for (m = n; m < n + 16; m++)
  80        {
  81          if (m > n && (m%4) == 0)
  82            g_string_append_c (ret, ' ');
  83          if (m < len)
  84            g_string_append_printf (ret, "%02x ", (guchar) data[m]);
  85          else
  86            g_string_append (ret, "   ");
  87        }
  88
  89      g_string_append (ret, "   ");
  90
  91      for (m = n; m < len && m < n + 16; m++)
  92        g_string_append_c (ret, g_ascii_isprint (data[m]) ? data[m] : '.');
  93
  94      g_string_append_c (ret, '\n');
  95    }
  96
  97  return g_string_free (ret, FALSE);
  98 }
  99
 100 /* ---------------------------------------------------------------------------------------------------- */
 101
 102 /* Unfortunately ancillary messages are discarded when reading from a
 103  * socket using the GSocketInputStream abstraction. So we provide a
 104  * very GInputStream-ish API that uses GSocket in this case (very
 105  * similar to GSocketInputStream).
 106  */
 107
 108 typedef struct
 109 {
 110   void *buffer;
 111   gsize count;
 112
 113   GSocketControlMessage ***messages;
 114   gint *num_messages;
 115 } ReadWithControlData;
 116
 117 static void
 118 read_with_control_data_free (ReadWithControlData *data)
 119 {
 120   g_slice_free (ReadWithControlData, data);
 121 }
 122
 123 static gboolean
 124 _g_socket_read_with_control_messages_ready (GSocket      *socket,
 125                                             GIOCondition  condition,
 126                                             gpointer      user_data)
 127 {
 128   GTask *task = user_data;
 129   ReadWithControlData *data = g_task_get_task_data (task);
 130   GError *error;
 131   gssize result;
 132   GInputVector vector;
 133
 134   error = NULL;
 135   vector.buffer = data->buffer;
 136   vector.size = data->count;
 137   result = g_socket_receive_message (socket,
 138                                      NULL, /* address */
 139                                      &vector,
 140                                      1,
 141                                      data->messages,
 142                                      data->num_messages,
 143                                      NULL,
 144                                      g_task_get_cancellable (task),
 145                                      &error);
 146
 147   if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
 148     {
 149       g_error_free (error);
 150       return TRUE;
 151     }
 152
 153   g_assert (result >= 0 || error != NULL);
 154   if (result >= 0)
 155     g_task_return_int (task, result);
 156   else
 157     g_task_return_error (task, error);
 158   g_object_unref (task);
 159
 160   return FALSE;
 161 }
 162
 163 static void
 164 _g_socket_read_with_control_messages (GSocket                 *socket,
 165                                       void                    *buffer,
 166                                       gsize                    count,
 167                                       GSocketControlMessage ***messages,
 168                                       gint                    *num_messages,
 169                                       gint                     io_priority,
 170                                       GCancellable            *cancellable,
 171                                       GAsyncReadyCallback      callback,
 172                                       gpointer                 user_data)
 173 {
 174   GTask *task;
 175   ReadWithControlData *data;
 176   GSource *source;
 177
 178   data = g_slice_new0 (ReadWithControlData);
 179   data->buffer = buffer;
 180   data->count = count;
 181   data->messages = messages;
 182   data->num_messages = num_messages;
 183
 184   task = g_task_new (socket, cancellable, callback, user_data);
 185   g_task_set_source_tag (task, _g_socket_read_with_control_messages);
 186   g_task_set_name (task, "[gio] D-Bus read");
 187   g_task_set_task_data (task, data, (GDestroyNotify) read_with_control_data_free);
 188
 189   if (g_socket_condition_check (socket, G_IO_IN))
 190     {
 191       if (!_g_socket_read_with_control_messages_ready (socket, G_IO_IN, task))
 192         return;
 193     }
 194
 195   source = g_socket_create_source (socket,
 196                                    G_IO_IN | G_IO_HUP | G_IO_ERR,
 197                                    cancellable);
 198   g_task_attach_source (task, source, (GSourceFunc) _g_socket_read_with_control_messages_ready);
 199   g_source_unref (source);
 200 }
 201
 202 static gssize
 203 _g_socket_read_with_control_messages_finish (GSocket       *socket,
 204                                              GAsyncResult  *result,
 205                                              GError       **error)
 206 {
 207   g_return_val_if_fail (G_IS_SOCKET (socket), -1);
 208   g_return_val_if_fail (g_task_is_valid (result, socket), -1);
 209
 210   return g_task_propagate_int (G_TASK (result), error);
 211 }
 212
 213 /* ---------------------------------------------------------------------------------------------------- */
 214
 215 /* Work-around for https://bugzilla.gnome.org/show_bug.cgi?id=674885
 216    and see also the original https://bugzilla.gnome.org/show_bug.cgi?id=627724  */
 217
 218 static GPtrArray *ensured_classes = NULL;
 219
 220 static void
 221 ensure_type (GType gtype)
 222 {
 223   g_ptr_array_add (ensured_classes, g_type_class_ref (gtype));
 224 }
 225
 226 static void
 227 release_required_types (void)
 228 {
 229   g_ptr_array_foreach (ensured_classes, (GFunc) g_type_class_unref, NULL);
 230   g_ptr_array_unref (ensured_classes);
 231   ensured_classes = NULL;
 232 }
 233
 234 static void
 235 ensure_required_types (void)
 236 {
 237   g_assert (ensured_classes == NULL);
 238   ensured_classes = g_ptr_array_new ();
 239   /* Generally in this list, you should initialize types which are used as
 240    * properties first, then the class which has them. For example, GDBusProxy
 241    * has a type of GDBusConnection, so we initialize GDBusConnection first.
 242    * And because GDBusConnection has a property of type GDBusConnectionFlags,
 243    * we initialize that first.
 244    *
 245    * Similarly, GSocket has a type of GSocketAddress.
 246    *
 247    * We don't fill out the whole dependency tree right now because in practice
 248    * it tends to be just types that GDBus use that cause pain, and there
 249    * is work on a more general approach in https://bugzilla.gnome.org/show_bug.cgi?id=674885
 250    */
 251   ensure_type (G_TYPE_TASK);
 252   ensure_type (G_TYPE_MEMORY_INPUT_STREAM);
 253   ensure_type (G_TYPE_DBUS_CONNECTION_FLAGS);
 254   ensure_type (G_TYPE_DBUS_CAPABILITY_FLAGS);
 255   ensure_type (G_TYPE_DBUS_AUTH_OBSERVER);
 256   ensure_type (G_TYPE_DBUS_CONNECTION);
 257   ensure_type (G_TYPE_DBUS_PROXY);
 258   ensure_type (G_TYPE_SOCKET_FAMILY);
 259   ensure_type (G_TYPE_SOCKET_TYPE);
 260   ensure_type (G_TYPE_SOCKET_PROTOCOL);
 261   ensure_type (G_TYPE_SOCKET_ADDRESS);
 262   ensure_type (G_TYPE_SOCKET);
 263 }
 264 /* ---------------------------------------------------------------------------------------------------- */
 265
 266 typedef struct
 267 {
 268   volatile gint refcount;
 269   GThread *thread;
 270   GMainContext *context;
 271   GMainLoop *loop;
 272 } SharedThreadData;
 273
 274 static gpointer
 275 gdbus_shared_thread_func (gpointer user_data)
 276 {
 277   SharedThreadData *data = user_data;
 278
 279   g_main_context_push_thread_default (data->context);
 280   g_main_loop_run (data->loop);
 281   g_main_context_pop_thread_default (data->context);
 282
 283   release_required_types ();
 284
 285   return NULL;
 286 }
 287
 288 /* ---------------------------------------------------------------------------------------------------- */
 289
 290 static SharedThreadData *
 291 _g_dbus_shared_thread_ref (void)
 292 {
 293   static gsize shared_thread_data = 0;
 294   SharedThreadData *ret;
 295
 296   if (g_once_init_enter (&shared_thread_data))
 297     {
 298       SharedThreadData *data;
 299
 300       data = g_new0 (SharedThreadData, 1);
 301       data->refcount = 0;
 302
 303       data->context = g_main_context_new ();
 304       data->loop = g_main_loop_new (data->context, FALSE);
 305       data->thread = g_thread_new ("gdbus",
 306                                    gdbus_shared_thread_func,
 307                                    data);
 308       /* We can cast between gsize and gpointer safely */
 309       g_once_init_leave (&shared_thread_data, (gsize) data);
 310     }
 311
 312   ret = (SharedThreadData*) shared_thread_data;
 313   g_atomic_int_inc (&ret->refcount);
 314   return ret;
 315 }
 316
 317 static void
 318 _g_dbus_shared_thread_unref (SharedThreadData *data)
 319 {
 320   /* TODO: actually destroy the shared thread here */
 321 #if 0
 322   g_assert (data != NULL);
 323   if (g_atomic_int_dec_and_test (&data->refcount))
 324     {
 325       g_main_loop_quit (data->loop);
 326       //g_thread_join (data->thread);
 327       g_main_loop_unref (data->loop);
 328       g_main_context_unref (data->context);
 329     }
 330 #endif
 331 }
 332
 333 /* ---------------------------------------------------------------------------------------------------- */
 334
 335 typedef enum {
 336     PENDING_NONE = 0,
 337     PENDING_WRITE,
 338     PENDING_FLUSH,
 339     PENDING_CLOSE
 340 } OutputPending;
 341
 342 struct GDBusWorker
 343 {
 344   volatile gint                       ref_count;
 345
 346   SharedThreadData                   *shared_thread_data;
 347
 348   /* really a boolean, but GLib 2.28 lacks atomic boolean ops */
 349   volatile gint                       stopped;
 350
 351   /* TODO: frozen (e.g. G_DBUS_CONNECTION_FLAGS_DELAY_MESSAGE_PROCESSING) currently
 352    * only affects messages received from the other peer (since GDBusServer is the
 353    * only user) - we might want it to affect messages sent to the other peer too?
 354    */
 355   gboolean                            frozen;
 356   GDBusCapabilityFlags                capabilities;
 357   GQueue                             *received_messages_while_frozen;
 358
 359   GIOStream                          *stream;
 360   GCancellable                       *cancellable;
 361   GDBusWorkerMessageReceivedCallback  message_received_callback;
 362   GDBusWorkerMessageAboutToBeSentCallback message_about_to_be_sent_callback;
 363   GDBusWorkerDisconnectedCallback     disconnected_callback;
 364   gpointer                            user_data;
 365
 366   /* if not NULL, stream is GSocketConnection */
 367   GSocket *socket;
 368
 369   /* used for reading */
 370   GMutex                              read_lock;
 371   gchar                              *read_buffer;
 372   gsize                               read_buffer_allocated_size;
 373   gsize                               read_buffer_cur_size;
 374   gsize                               read_buffer_bytes_wanted;
 375   GUnixFDList                        *read_fd_list;
 376   GSocketControlMessage             **read_ancillary_messages;
 377   gint                                read_num_ancillary_messages;
 378
 379   /* Whether an async write, flush or close, or none of those, is pending.
 380    * Only the worker thread may change its value, and only with the write_lock.
 381    * Other threads may read its value when holding the write_lock.
 382    * The worker thread may read its value at any time.
 383    */
 384   OutputPending                       output_pending;
 385   /* used for writing */
 386   GMutex                              write_lock;
 387   /* queue of MessageToWriteData, protected by write_lock */
 388   GQueue                             *write_queue;
 389   /* protected by write_lock */
 390   guint64                             write_num_messages_written;
 391   /* number of messages we'd written out last time we flushed;
 392    * protected by write_lock
 393    */
 394   guint64                             write_num_messages_flushed;
 395   /* list of FlushData, protected by write_lock */
 396   GList                              *write_pending_flushes;
 397   /* list of CloseData, protected by write_lock */
 398   GList                              *pending_close_attempts;
 399   /* no lock - only used from the worker thread */
 400   gboolean                            close_expected;
 401 };
 402
 403 static void _g_dbus_worker_unref (GDBusWorker *worker);
 404
 405 /* ---------------------------------------------------------------------------------------------------- */
 406
 407 typedef struct
 408 {
 409   GMutex  mutex;
 410   GCond   cond;
 411   guint64 number_to_wait_for;
 412   gboolean finished;
 413   GError *error;
 414 } FlushData;
 415
 416 struct _MessageToWriteData ;
 417 typedef struct _MessageToWriteData MessageToWriteData;
 418
 419 static void message_to_write_data_free (MessageToWriteData *data);
 420
 421 static void read_message_print_transport_debug (gssize bytes_read,
 422                                                 GDBusWorker *worker);
 423
 424 static void write_message_print_transport_debug (gssize bytes_written,
 425                                                  MessageToWriteData *data);
 426
 427 typedef struct {
 428     GDBusWorker *worker;
 429     GTask *task;
 430 } CloseData;
 431
 432 static void close_data_free (CloseData *close_data)
 433 {
 434   g_clear_object (&close_data->task);
 435
 436   _g_dbus_worker_unref (close_data->worker);
 437   g_slice_free (CloseData, close_data);
 438 }
 439
 440 /* ---------------------------------------------------------------------------------------------------- */
 441
 442 static GDBusWorker *
 443 _g_dbus_worker_ref (GDBusWorker *worker)
 444 {
 445   g_atomic_int_inc (&worker->ref_count);
 446   return worker;
 447 }
 448
 449 static void
 450 _g_dbus_worker_unref (GDBusWorker *worker)
 451 {
 452   if (g_atomic_int_dec_and_test (&worker->ref_count))
 453     {
 454       g_assert (worker->write_pending_flushes == NULL);
 455
 456       _g_dbus_shared_thread_unref (worker->shared_thread_data);
 457
 458       g_object_unref (worker->stream);
 459
 460       g_mutex_clear (&worker->read_lock);
 461       g_object_unref (worker->cancellable);
 462       if (worker->read_fd_list != NULL)
 463         g_object_unref (worker->read_fd_list);
 464
 465       g_queue_free_full (worker->received_messages_while_frozen, (GDestroyNotify) g_object_unref);
 466       g_mutex_clear (&worker->write_lock);
 467       g_queue_free_full (worker->write_queue, (GDestroyNotify) message_to_write_data_free);
 468       g_free (worker->read_buffer);
 469
 470       g_free (worker);
 471     }
 472 }
 473
 474 static void
 475 _g_dbus_worker_emit_disconnected (GDBusWorker  *worker,
 476                                   gboolean      remote_peer_vanished,
 477                                   GError       *error)
 478 {
 479   if (!g_atomic_int_get (&worker->stopped))
 480     worker->disconnected_callback (worker, remote_peer_vanished, error, worker->user_data);
 481 }
 482
 483 static void
 484 _g_dbus_worker_emit_message_received (GDBusWorker  *worker,
 485                                       GDBusMessage *message)
 486 {
 487   if (!g_atomic_int_get (&worker->stopped))
 488     worker->message_received_callback (worker, message, worker->user_data);
 489 }
 490
 491 static GDBusMessage *
 492 _g_dbus_worker_emit_message_about_to_be_sent (GDBusWorker  *worker,
 493                                               GDBusMessage *message)
 494 {
 495   GDBusMessage *ret;
 496   if (!g_atomic_int_get (&worker->stopped))
 497     ret = worker->message_about_to_be_sent_callback (worker, g_steal_pointer (&message), worker->user_data);
 498   else
 499     ret = g_steal_pointer (&message);
 500   return ret;
 501 }
 502
 503 /* can only be called from private thread with read-lock held - takes ownership of @message */
 504 static void
 505 _g_dbus_worker_queue_or_deliver_received_message (GDBusWorker  *worker,
 506                                                   GDBusMessage *message)
 507 {
 508   if (worker->frozen || g_queue_get_length (worker->received_messages_while_frozen) > 0)
 509     {
 510       /* queue up */
 511       g_queue_push_tail (worker->received_messages_while_frozen, g_steal_pointer (&message));
 512     }
 513   else
 514     {
 515       /* not frozen, nor anything in queue */
 516       _g_dbus_worker_emit_message_received (worker, message);
 517       g_clear_object (&message);
 518     }
 519 }
 520
 521 /* called in private thread shared by all GDBusConnection instances (without read-lock held) */
 522 static gboolean
 523 unfreeze_in_idle_cb (gpointer user_data)
 524 {
 525   GDBusWorker *worker = user_data;
 526   GDBusMessage *message;
 527
 528   g_mutex_lock (&worker->read_lock);
 529   if (worker->frozen)
 530     {
 531       while ((message = g_queue_pop_head (worker->received_messages_while_frozen)) != NULL)
 532         {
 533           _g_dbus_worker_emit_message_received (worker, message);
 534           g_clear_object (&message);
 535         }
 536       worker->frozen = FALSE;
 537     }
 538   else
 539     {
 540       g_assert (g_queue_get_length (worker->received_messages_while_frozen) == 0);
 541     }
 542   g_mutex_unlock (&worker->read_lock);
 543   return FALSE;
 544 }
 545
 546 /* can be called from any thread */
 547 void
 548 _g_dbus_worker_unfreeze (GDBusWorker *worker)
 549 {
 550   GSource *idle_source;
 551   idle_source = g_idle_source_new ();
 552   g_source_set_priority (idle_source, G_PRIORITY_DEFAULT);
 553   g_source_set_callback (idle_source,
 554                          unfreeze_in_idle_cb,
 555                          _g_dbus_worker_ref (worker),
 556                          (GDestroyNotify) _g_dbus_worker_unref);
 557   g_source_set_name (idle_source, "[gio] unfreeze_in_idle_cb");
 558   g_source_attach (idle_source, worker->shared_thread_data->context);
 559   g_source_unref (idle_source);
 560 }
 561
 562 /* ---------------------------------------------------------------------------------------------------- */
 563
 564 static void _g_dbus_worker_do_read_unlocked (GDBusWorker *worker);
 565
 566 /* called in private thread shared by all GDBusConnection instances (without read-lock held) */
 567 static void
 568 _g_dbus_worker_do_read_cb (GInputStream  *input_stream,
 569                            GAsyncResult  *res,
 570                            gpointer       user_data)
 571 {
 572   GDBusWorker *worker = user_data;
 573   GError *error;
 574   gssize bytes_read;
 575
 576   g_mutex_lock (&worker->read_lock);
 577
 578   /* If already stopped, don't even process the reply */
 579   if (g_atomic_int_get (&worker->stopped))
 580     goto out;
 581
 582   error = NULL;
 583   if (worker->socket == NULL)
 584     bytes_read = g_input_stream_read_finish (g_io_stream_get_input_stream (worker->stream),
 585                                              res,
 586                                              &error);
 587   else
 588     bytes_read = _g_socket_read_with_control_messages_finish (worker->socket,
 589                                                               res,
 590                                                               &error);
 591   if (worker->read_num_ancillary_messages > 0)
 592     {
 593       gint n;
 594       for (n = 0; n < worker->read_num_ancillary_messages; n++)
 595         {
 596           GSocketControlMessage *control_message = G_SOCKET_CONTROL_MESSAGE (worker->read_ancillary_messages[n]);
 597
 598           if (FALSE)
 599             {
 600             }
 601 #ifdef G_OS_UNIX
 602           else if (G_IS_UNIX_FD_MESSAGE (control_message))
 603             {
 604               GUnixFDMessage *fd_message;
 605               gint *fds;
 606               gint num_fds;
 607
 608               fd_message = G_UNIX_FD_MESSAGE (control_message);
 609               fds = g_unix_fd_message_steal_fds (fd_message, &num_fds);
 610               if (worker->read_fd_list == NULL)
 611                 {
 612                   worker->read_fd_list = g_unix_fd_list_new_from_array (fds, num_fds);
 613                 }
 614               else
 615                 {
 616                   gint n;
 617                   for (n = 0; n < num_fds; n++)
 618                     {
 619                       /* TODO: really want a append_steal() */
 620                       g_unix_fd_list_append (worker->read_fd_list, fds[n], NULL);
 621                       (void) g_close (fds[n], NULL);
 622                     }
 623                 }
 624               g_free (fds);
 625             }
 626           else if (G_IS_UNIX_CREDENTIALS_MESSAGE (control_message))
 627             {
 628               /* do nothing */
 629             }
 630 #endif
 631           else
 632             {
 633               if (error == NULL)
 634                 {
 635                   g_set_error (&error,
 636                                G_IO_ERROR,
 637                                G_IO_ERROR_FAILED,
 638                                "Unexpected ancillary message of type %s received from peer",
 639                                g_type_name (G_TYPE_FROM_INSTANCE (control_message)));
 640                   _g_dbus_worker_emit_disconnected (worker, TRUE, error);
 641                   g_error_free (error);
 642                   g_object_unref (control_message);
 643                   n++;
 644                   while (n < worker->read_num_ancillary_messages)
 645                     g_object_unref (worker->read_ancillary_messages[n++]);
 646                   g_free (worker->read_ancillary_messages);
 647                   goto out;
 648                 }
 649             }
 650           g_object_unref (control_message);
 651         }
 652       g_free (worker->read_ancillary_messages);
 653     }
 654
 655   if (bytes_read == -1)
 656     {
 657       if (G_UNLIKELY (_g_dbus_debug_transport ()))
 658         {
 659           _g_dbus_debug_print_lock ();
 660           g_print ("========================================================================\n"
 661                    "GDBus-debug:Transport:\n"
 662                    "  ---- READ ERROR on stream of type %s:\n"
 663                    "  ---- %s %d: %s\n",
 664                    g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_input_stream (worker->stream))),
 665                    g_quark_to_string (error->domain), error->code,
 666                    error->message);
 667           _g_dbus_debug_print_unlock ();
 668         }
 669
 670       /* Every async read that uses this callback uses worker->cancellable
 671        * as its GCancellable. worker->cancellable gets cancelled if and only
 672        * if the GDBusConnection tells us to close (either via
 673        * _g_dbus_worker_stop, which is called on last-unref, or directly),
 674        * so a cancelled read must mean our connection was closed locally.
 675        *
 676        * If we're closing, other errors are possible - notably,
 677        * G_IO_ERROR_CLOSED can be seen if we close the stream with an async
 678        * read in-flight. It seems sensible to treat all read errors during
 679        * closing as an expected thing that doesn't trip exit-on-close.
 680        *
 681        * Because close_expected can't be set until we get into the worker
 682        * thread, but the cancellable is signalled sooner (from another
 683        * thread), we do still need to check the error.
 684        */
 685       if (worker->close_expected ||
 686           g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED))
 687         _g_dbus_worker_emit_disconnected (worker, FALSE, NULL);
 688       else
 689         _g_dbus_worker_emit_disconnected (worker, TRUE, error);
 690
 691       g_error_free (error);
 692       goto out;
 693     }
 694
 695 #if 0
 696   g_debug ("read %d bytes (is_closed=%d blocking=%d condition=0x%02x) stream %p, %p",
 697            (gint) bytes_read,
 698            g_socket_is_closed (g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker->stream))),
 699            g_socket_get_blocking (g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker->stream))),
 700            g_socket_condition_check (g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker->stream)),
 701                                      G_IO_IN | G_IO_OUT | G_IO_HUP),
 702            worker->stream,
 703            worker);
 704 #endif
 705
 706   /* The read failed, which could mean the dbus-daemon was sent SIGTERM. */
 707   if (bytes_read == 0)
 708     {
 709       g_set_error (&error,
 710                    G_IO_ERROR,
 711                    G_IO_ERROR_FAILED,
 712                    "Underlying GIOStream returned 0 bytes on an async read");
 713       _g_dbus_worker_emit_disconnected (worker, TRUE, error);
 714       g_error_free (error);
 715       goto out;
 716     }
 717
 718   read_message_print_transport_debug (bytes_read, worker);
 719
 720   worker->read_buffer_cur_size += bytes_read;
 721   if (worker->read_buffer_bytes_wanted == worker->read_buffer_cur_size)
 722     {
 723       /* OK, got what we asked for! */
 724       if (worker->read_buffer_bytes_wanted == 16)
 725         {
 726           gssize message_len;
 727           /* OK, got the header - determine how many more bytes are needed */
 728           error = NULL;
 729           message_len = g_dbus_message_bytes_needed ((guchar *) worker->read_buffer,
 730                                                      16,
 731                                                      &error);
 732           if (message_len == -1)
 733             {
 734               g_warning ("_g_dbus_worker_do_read_cb: error determining bytes needed: %s", error->message);
 735               _g_dbus_worker_emit_disconnected (worker, FALSE, error);
 736               g_error_free (error);
 737               goto out;
 738             }
 739
 740           worker->read_buffer_bytes_wanted = message_len;
 741           _g_dbus_worker_do_read_unlocked (worker);
 742         }
 743       else
 744         {
 745           GDBusMessage *message;
 746           error = NULL;
 747
 748           /* TODO: use connection->priv->auth to decode the message */
 749
 750           message = g_dbus_message_new_from_blob ((guchar *) worker->read_buffer,
 751                                                   worker->read_buffer_cur_size,
 752                                                   worker->capabilities,
 753                                                   &error);
 754           if (message == NULL)
 755             {
 756               gchar *s;
 757               s = _g_dbus_hexdump (worker->read_buffer, worker->read_buffer_cur_size, 2);
 758               g_warning ("Error decoding D-Bus message of %" G_GSIZE_FORMAT " bytes\n"
 759                          "The error is: %s\n"
 760                          "The payload is as follows:\n"
 761                          "%s",
 762                          worker->read_buffer_cur_size,
 763                          error->message,
 764                          s);
 765               g_free (s);
 766               _g_dbus_worker_emit_disconnected (worker, FALSE, error);
 767               g_error_free (error);
 768               goto out;
 769             }
 770
 771 #ifdef G_OS_UNIX
 772           if (worker->read_fd_list != NULL)
 773             {
 774               g_dbus_message_set_unix_fd_list (message, worker->read_fd_list);
 775               g_object_unref (worker->read_fd_list);
 776               worker->read_fd_list = NULL;
 777             }
 778 #endif
 779
 780           if (G_UNLIKELY (_g_dbus_debug_message ()))
 781             {
 782               gchar *s;
 783               _g_dbus_debug_print_lock ();
 784               g_print ("========================================================================\n"
 785                        "GDBus-debug:Message:\n"
 786                        "  <<<< RECEIVED D-Bus message (%" G_GSIZE_FORMAT " bytes)\n",
 787                        worker->read_buffer_cur_size);
 788               s = g_dbus_message_print (message, 2);
 789               g_print ("%s", s);
 790               g_free (s);
 791               if (G_UNLIKELY (_g_dbus_debug_payload ()))
 792                 {
 793                   s = _g_dbus_hexdump (worker->read_buffer, worker->read_buffer_cur_size, 2);
 794                   g_print ("%s\n", s);
 795                   g_free (s);
 796                 }
 797               _g_dbus_debug_print_unlock ();
 798             }
 799
 800           /* yay, got a message, go deliver it */
 801           _g_dbus_worker_queue_or_deliver_received_message (worker, g_steal_pointer (&message));
 802
 803           /* start reading another message! */
 804           worker->read_buffer_bytes_wanted = 0;
 805           worker->read_buffer_cur_size = 0;
 806           _g_dbus_worker_do_read_unlocked (worker);
 807         }
 808     }
 809   else
 810     {
 811       /* didn't get all the bytes we requested - so repeat the request... */
 812       _g_dbus_worker_do_read_unlocked (worker);
 813     }
 814
 815  out:
 816   g_mutex_unlock (&worker->read_lock);
 817
 818   /* check if there is any pending close */
 819   schedule_pending_close (worker);
 820
 821   /* gives up the reference acquired when calling g_input_stream_read_async() */
 822   _g_dbus_worker_unref (worker);
 823 }
 824
 825 /* called in private thread shared by all GDBusConnection instances (with read-lock held) */
 826 static void
 827 _g_dbus_worker_do_read_unlocked (GDBusWorker *worker)
 828 {
 829   /* Note that we do need to keep trying to read even if close_expected is
 830    * true, because only failing a read causes us to signal 'closed'.
 831    */
 832
 833   /* if bytes_wanted is zero, it means start reading a message */
 834   if (worker->read_buffer_bytes_wanted == 0)
 835     {
 836       worker->read_buffer_cur_size = 0;
 837       worker->read_buffer_bytes_wanted = 16;
 838     }
 839
 840   /* ensure we have a (big enough) buffer */
 841   if (worker->read_buffer == NULL || worker->read_buffer_bytes_wanted > worker->read_buffer_allocated_size)
 842     {
 843       /* TODO: 4096 is randomly chosen; might want a better chosen default minimum */
 844       worker->read_buffer_allocated_size = MAX (worker->read_buffer_bytes_wanted, 4096);
 845       worker->read_buffer = g_realloc (worker->read_buffer, worker->read_buffer_allocated_size);
 846     }
 847
 848   if (worker->socket == NULL)
 849     g_input_stream_read_async (g_io_stream_get_input_stream (worker->stream),
 850                                worker->read_buffer + worker->read_buffer_cur_size,
 851                                worker->read_buffer_bytes_wanted - worker->read_buffer_cur_size,
 852                                G_PRIORITY_DEFAULT,
 853                                worker->cancellable,
 854                                (GAsyncReadyCallback) _g_dbus_worker_do_read_cb,
 855                                _g_dbus_worker_ref (worker));
 856   else
 857     {
 858       worker->read_ancillary_messages = NULL;
 859       worker->read_num_ancillary_messages = 0;
 860       _g_socket_read_with_control_messages (worker->socket,
 861                                             worker->read_buffer + worker->read_buffer_cur_size,
 862                                             worker->read_buffer_bytes_wanted - worker->read_buffer_cur_size,
 863                                             &worker->read_ancillary_messages,
 864                                             &worker->read_num_ancillary_messages,
 865                                             G_PRIORITY_DEFAULT,
 866                                             worker->cancellable,
 867                                             (GAsyncReadyCallback) _g_dbus_worker_do_read_cb,
 868                                             _g_dbus_worker_ref (worker));
 869     }
 870 }
 871
 872 /* called in private thread shared by all GDBusConnection instances (without read-lock held) */
 873 static gboolean
 874 _g_dbus_worker_do_initial_read (gpointer data)
 875 {
 876   GDBusWorker *worker = data;
 877   g_mutex_lock (&worker->read_lock);
 878   _g_dbus_worker_do_read_unlocked (worker);
 879   g_mutex_unlock (&worker->read_lock);
 880   return FALSE;
 881 }
 882
 883 /* ---------------------------------------------------------------------------------------------------- */
 884
 885 struct _MessageToWriteData
 886 {
 887   GDBusWorker  *worker;
 888   GDBusMessage *message;
 889   gchar        *blob;
 890   gsize         blob_size;
 891
 892   gsize         total_written;
 893   GTask        *task;
 894 };
 895
 896 static void
 897 message_to_write_data_free (MessageToWriteData *data)
 898 {
 899   _g_dbus_worker_unref (data->worker);
 900   if (data->message)
 901     g_object_unref (data->message);
 902   g_free (data->blob);
 903   g_slice_free (MessageToWriteData, data);
 904 }
 905
 906 /* ---------------------------------------------------------------------------------------------------- */
 907
 908 static void write_message_continue_writing (MessageToWriteData *data);
 909
 910 /* called in private thread shared by all GDBusConnection instances
 911  *
 912  * write-lock is not held on entry
 913  * output_pending is PENDING_WRITE on entry
 914  */
 915 static void
 916 write_message_async_cb (GObject      *source_object,
 917                         GAsyncResult *res,
 918                         gpointer      user_data)
 919 {
 920   MessageToWriteData *data = user_data;
 921   GTask *task;
 922   gssize bytes_written;
 923   GError *error;
 924
 925   /* Note: we can't access data->task after calling g_task_return_* () because the
 926    * callback can free @data and we're not completing in idle. So use a copy of the pointer.
 927    */
 928   task = data->task;
 929
 930   error = NULL;
 931   bytes_written = g_output_stream_write_finish (G_OUTPUT_STREAM (source_object),
 932                                                 res,
 933                                                 &error);
 934   if (bytes_written == -1)
 935     {
 936       g_task_return_error (task, error);
 937       g_object_unref (task);
 938       goto out;
 939     }
 940   g_assert (bytes_written > 0); /* zero is never returned */
 941
 942   write_message_print_transport_debug (bytes_written, data);
 943
 944   data->total_written += bytes_written;
 945   g_assert (data->total_written <= data->blob_size);
 946   if (data->total_written == data->blob_size)
 947     {
 948       g_task_return_boolean (task, TRUE);
 949       g_object_unref (task);
 950       goto out;
 951     }
 952
 953   write_message_continue_writing (data);
 954
 955  out:
 956   ;
 957 }
 958
 959 /* called in private thread shared by all GDBusConnection instances
 960  *
 961  * write-lock is not held on entry
 962  * output_pending is PENDING_WRITE on entry
 963  */
 964 #ifdef G_OS_UNIX
 965 static gboolean
 966 on_socket_ready (GSocket      *socket,
 967                  GIOCondition  condition,
 968                  gpointer      user_data)
 969 {
 970   MessageToWriteData *data = user_data;
 971   write_message_continue_writing (data);
 972   return FALSE; /* remove source */
 973 }
 974 #endif
 975
 976 /* called in private thread shared by all GDBusConnection instances
 977  *
 978  * write-lock is not held on entry
 979  * output_pending is PENDING_WRITE on entry
 980  */
 981 static void
 982 write_message_continue_writing (MessageToWriteData *data)
 983 {
 984   GOutputStream *ostream;
 985 #ifdef G_OS_UNIX
 986   GTask *task;
 987   GUnixFDList *fd_list;
 988 #endif
 989
 990 #ifdef G_OS_UNIX
 991   /* Note: we can't access data->task after calling g_task_return_* () because the
 992    * callback can free @data and we're not completing in idle. So use a copy of the pointer.
 993    */
 994   task = data->task;
 995 #endif
 996
 997   ostream = g_io_stream_get_output_stream (data->worker->stream);
 998 #ifdef G_OS_UNIX
 999   fd_list = g_dbus_message_get_unix_fd_list (data->message);
1000 #endif
1001
1002   g_assert (!g_output_stream_has_pending (ostream));
1003   g_assert_cmpint (data->total_written, <, data->blob_size);
1004
1005   if (FALSE)
1006     {
1007     }
1008 #ifdef G_OS_UNIX
1009   else if (G_IS_SOCKET_OUTPUT_STREAM (ostream) && data->total_written == 0)
1010     {
1011       GOutputVector vector;
1012       GSocketControlMessage *control_message;
1013       gssize bytes_written;
1014       GError *error;
1015
1016       vector.buffer = data->blob;
1017       vector.size = data->blob_size;
1018
1019       control_message = NULL;
1020       if (fd_list != NULL && g_unix_fd_list_get_length (fd_list) > 0)
1021         {
1022           if (!(data->worker->capabilities & G_DBUS_CAPABILITY_FLAGS_UNIX_FD_PASSING))
1023             {
1024               g_task_return_new_error (task,
1025                                        G_IO_ERROR,
1026                                        G_IO_ERROR_FAILED,
1027                                        "Tried sending a file descriptor but remote peer does not support this capability");
1028               g_object_unref (task);
1029               goto out;
1030             }
1031           control_message = g_unix_fd_message_new_with_fd_list (fd_list);
1032         }
1033
1034       error = NULL;
1035       bytes_written = g_socket_send_message (data->worker->socket,
1036                                              NULL, /* address */
1037                                              &vector,
1038                                              1,
1039                                              control_message != NULL ? &control_message : NULL,
1040                                              control_message != NULL ? 1 : 0,
1041                                              G_SOCKET_MSG_NONE,
1042                                              data->worker->cancellable,
1043                                              &error);
1044       if (control_message != NULL)
1045         g_object_unref (control_message);
1046
1047       if (bytes_written == -1)
1048         {
1049           /* Handle WOULD_BLOCK by waiting until there's room in the buffer */
1050           if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
1051             {
1052               GSource *source;
1053               source = g_socket_create_source (data->worker->socket,
1054                                                G_IO_OUT | G_IO_HUP | G_IO_ERR,
1055                                                data->worker->cancellable);
1056               g_source_set_callback (source,
1057                                      (GSourceFunc) on_socket_ready,
1058                                      data,
1059                                      NULL); /* GDestroyNotify */
1060               g_source_attach (source, g_main_context_get_thread_default ());
1061               g_source_unref (source);
1062               g_error_free (error);
1063               goto out;
1064             }
1065           g_task_return_error (task, error);
1066           g_object_unref (task);
1067           goto out;
1068         }
1069       g_assert (bytes_written > 0); /* zero is never returned */
1070
1071       write_message_print_transport_debug (bytes_written, data);
1072
1073       data->total_written += bytes_written;
1074       g_assert (data->total_written <= data->blob_size);
1075       if (data->total_written == data->blob_size)
1076         {
1077           g_task_return_boolean (task, TRUE);
1078           g_object_unref (task);
1079           goto out;
1080         }
1081
1082       write_message_continue_writing (data);
1083     }
1084 #endif
1085   else
1086     {
1087 #ifdef G_OS_UNIX
1088       if (data->total_written == 0 && fd_list != NULL)
1089         {
1090           /* We were trying to write byte 0 of the message, which needs
1091            * the fd list to be attached to it, but this connection doesn't
1092            * support doing that. */
1093           g_task_return_new_error (task,
1094                                    G_IO_ERROR,
1095                                    G_IO_ERROR_FAILED,
1096                                    "Tried sending a file descriptor on unsupported stream of type %s",
1097                                    g_type_name (G_TYPE_FROM_INSTANCE (ostream)));
1098           g_object_unref (task);
1099           goto out;
1100         }
1101 #endif
1102
1103       g_output_stream_write_async (ostream,
1104                                    (const gchar *) data->blob + data->total_written,
1105                                    data->blob_size - data->total_written,
1106                                    G_PRIORITY_DEFAULT,
1107                                    data->worker->cancellable,
1108                                    write_message_async_cb,
1109                                    data);
1110     }
1111 #ifdef G_OS_UNIX
1112  out:
1113 #endif
1114   ;
1115 }
1116
1117 /* called in private thread shared by all GDBusConnection instances
1118  *
1119  * write-lock is not held on entry
1120  * output_pending is PENDING_WRITE on entry
1121  */
1122 static void
1123 write_message_async (GDBusWorker         *worker,
1124                      MessageToWriteData  *data,
1125                      GAsyncReadyCallback  callback,
1126                      gpointer             user_data)
1127 {
1128   data->task = g_task_new (NULL, NULL, callback, user_data);
1129   g_task_set_source_tag (data->task, write_message_async);
1130   g_task_set_name (data->task, "[gio] D-Bus write message");
1131   data->total_written = 0;
1132   write_message_continue_writing (data);
1133 }
1134
1135 /* called in private thread shared by all GDBusConnection instances (with write-lock held) */
1136 static gboolean
1137 write_message_finish (GAsyncResult   *res,
1138                       GError        **error)
1139 {
1140   g_return_val_if_fail (g_task_is_valid (res, NULL), FALSE);
1141
1142   return g_task_propagate_boolean (G_TASK (res), error);
1143 }
1144 /* ---------------------------------------------------------------------------------------------------- */
1145
1146 static void continue_writing (GDBusWorker *worker);
1147
1148 typedef struct
1149 {
1150   GDBusWorker *worker;
1151   GList *flushers;
1152 } FlushAsyncData;
1153
1154 static void
1155 flush_data_list_complete (const GList  *flushers,
1156                           const GError *error)
1157 {
1158   const GList *l;
1159
1160   for (l = flushers; l != NULL; l = l->next)
1161     {
1162       FlushData *f = l->data;
1163
1164       f->error = error != NULL ? g_error_copy (error) : NULL;
1165
1166       g_mutex_lock (&f->mutex);
1167       f->finished = TRUE;
1168       g_cond_signal (&f->cond);
1169       g_mutex_unlock (&f->mutex);
1170     }
1171 }
1172
1173 /* called in private thread shared by all GDBusConnection instances
1174  *
1175  * write-lock is not held on entry
1176  * output_pending is PENDING_FLUSH on entry
1177  */
1178 static void
1179 ostream_flush_cb (GObject      *source_object,
1180                   GAsyncResult *res,
1181                   gpointer      user_data)
1182 {
1183   FlushAsyncData *data = user_data;
1184   GError *error;
1185
1186   error = NULL;
1187   g_output_stream_flush_finish (G_OUTPUT_STREAM (source_object),
1188                                 res,
1189                                 &error);
1190
1191   if (error == NULL)
1192     {
1193       if (G_UNLIKELY (_g_dbus_debug_transport ()))
1194         {
1195           _g_dbus_debug_print_lock ();
1196           g_print ("========================================================================\n"
1197                    "GDBus-debug:Transport:\n"
1198                    "  ---- FLUSHED stream of type %s\n",
1199                    g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_output_stream (data->worker->stream))));
1200           _g_dbus_debug_print_unlock ();
1201         }
1202     }
1203
1204   /* Make sure we tell folks that we don't have additional
1205      flushes pending */
1206   g_mutex_lock (&data->worker->write_lock);
1207   data->worker->write_num_messages_flushed = data->worker->write_num_messages_written;
1208   g_assert (data->worker->output_pending == PENDING_FLUSH);
1209   data->worker->output_pending = PENDING_NONE;
1210   g_mutex_unlock (&data->worker->write_lock);
1211
1212   g_assert (data->flushers != NULL);
1213   flush_data_list_complete (data->flushers, error);
1214   g_list_free (data->flushers);
1215   if (error != NULL)
1216     g_error_free (error);
1217
1218   /* OK, cool, finally kick off the next write */
1219   continue_writing (data->worker);
1220
1221   _g_dbus_worker_unref (data->worker);
1222   g_free (data);
1223 }
1224
1225 /* called in private thread shared by all GDBusConnection instances
1226  *
1227  * write-lock is not held on entry
1228  * output_pending is PENDING_FLUSH on entry
1229  */
1230 static void
1231 start_flush (FlushAsyncData *data)
1232 {
1233   g_output_stream_flush_async (g_io_stream_get_output_stream (data->worker->stream),
1234                                G_PRIORITY_DEFAULT,
1235                                data->worker->cancellable,
1236                                ostream_flush_cb,
1237                                data);
1238 }
1239
1240 /* called in private thread shared by all GDBusConnection instances
1241  *
1242  * write-lock is held on entry
1243  * output_pending is PENDING_NONE on entry
1244  */
1245 static void
1246 message_written_unlocked (GDBusWorker *worker,
1247                           MessageToWriteData *message_data)
1248 {
1249   if (G_UNLIKELY (_g_dbus_debug_message ()))
1250     {
1251       gchar *s;
1252       _g_dbus_debug_print_lock ();
1253       g_print ("========================================================================\n"
1254                "GDBus-debug:Message:\n"
1255                "  >>>> SENT D-Bus message (%" G_GSIZE_FORMAT " bytes)\n",
1256                message_data->blob_size);
1257       s = g_dbus_message_print (message_data->message, 2);
1258       g_print ("%s", s);
1259       g_free (s);
1260       if (G_UNLIKELY (_g_dbus_debug_payload ()))
1261         {
1262           s = _g_dbus_hexdump (message_data->blob, message_data->blob_size, 2);
1263           g_print ("%s\n", s);
1264           g_free (s);
1265         }
1266       _g_dbus_debug_print_unlock ();
1267     }
1268
1269   worker->write_num_messages_written += 1;
1270 }
1271
1272 /* called in private thread shared by all GDBusConnection instances
1273  *
1274  * write-lock is held on entry
1275  * output_pending is PENDING_NONE on entry
1276  *
1277  * Returns: non-%NULL, setting @output_pending, if we need to flush now
1278  */
1279 static FlushAsyncData *
1280 prepare_flush_unlocked (GDBusWorker *worker)
1281 {
1282   GList *l;
1283   GList *ll;
1284   GList *flushers;
1285
1286   flushers = NULL;
1287   for (l = worker->write_pending_flushes; l != NULL; l = ll)
1288     {
1289       FlushData *f = l->data;
1290       ll = l->next;
1291
1292       if (f->number_to_wait_for == worker->write_num_messages_written)
1293         {
1294           flushers = g_list_append (flushers, f);
1295           worker->write_pending_flushes = g_list_delete_link (worker->write_pending_flushes, l);
1296         }
1297     }
1298   if (flushers != NULL)
1299     {
1300       g_assert (worker->output_pending == PENDING_NONE);
1301       worker->output_pending = PENDING_FLUSH;
1302     }
1303
1304   if (flushers != NULL)
1305     {
1306       FlushAsyncData *data;
1307
1308       data = g_new0 (FlushAsyncData, 1);
1309       data->worker = _g_dbus_worker_ref (worker);
1310       data->flushers = flushers;
1311       return data;
1312     }
1313
1314   return NULL;
1315 }
1316
1317 /* called in private thread shared by all GDBusConnection instances
1318  *
1319  * write-lock is not held on entry
1320  * output_pending is PENDING_WRITE on entry
1321  */
1322 static void
1323 write_message_cb (GObject       *source_object,
1324                   GAsyncResult  *res,
1325                   gpointer       user_data)
1326 {
1327   MessageToWriteData *data = user_data;
1328   GError *error;
1329
1330   g_mutex_lock (&data->worker->write_lock);
1331   g_assert (data->worker->output_pending == PENDING_WRITE);
1332   data->worker->output_pending = PENDING_NONE;
1333
1334   error = NULL;
1335   if (!write_message_finish (res, &error))
1336     {
1337       g_mutex_unlock (&data->worker->write_lock);
1338
1339       /* TODO: handle */
1340       _g_dbus_worker_emit_disconnected (data->worker, TRUE, error);
1341       g_error_free (error);
1342
1343       g_mutex_lock (&data->worker->write_lock);
1344     }
1345
1346   message_written_unlocked (data->worker, data);
1347
1348   g_mutex_unlock (&data->worker->write_lock);
1349
1350   continue_writing (data->worker);
1351
1352   message_to_write_data_free (data);
1353 }
1354
1355 /* called in private thread shared by all GDBusConnection instances
1356  *
1357  * write-lock is not held on entry
1358  * output_pending is PENDING_CLOSE on entry
1359  */
1360 static void
1361 iostream_close_cb (GObject      *source_object,
1362                    GAsyncResult *res,
1363                    gpointer      user_data)
1364 {
1365   GDBusWorker *worker = user_data;
1366   GError *error = NULL;
1367   GList *pending_close_attempts, *pending_flush_attempts;
1368   GQueue *send_queue;
1369
1370   g_io_stream_close_finish (worker->stream, res, &error);
1371
1372   g_mutex_lock (&worker->write_lock);
1373
1374   pending_close_attempts = worker->pending_close_attempts;
1375   worker->pending_close_attempts = NULL;
1376
1377   pending_flush_attempts = worker->write_pending_flushes;
1378   worker->write_pending_flushes = NULL;
1379
1380   send_queue = worker->write_queue;
1381   worker->write_queue = g_queue_new ();
1382
1383   g_assert (worker->output_pending == PENDING_CLOSE);
1384   worker->output_pending = PENDING_NONE;
1385
1386   /* Ensure threads waiting for pending flushes to finish will be unblocked. */
1387   worker->write_num_messages_flushed =
1388     worker->write_num_messages_written + g_list_length(pending_flush_attempts);
1389
1390   g_mutex_unlock (&worker->write_lock);
1391
1392   while (pending_close_attempts != NULL)
1393     {
1394       CloseData *close_data = pending_close_attempts->data;
1395
1396       pending_close_attempts = g_list_delete_link (pending_close_attempts,
1397                                                    pending_close_attempts);
1398
1399       if (close_data->task != NULL)
1400         {
1401           if (error != NULL)
1402             g_task_return_error (close_data->task, g_error_copy (error));
1403           else
1404             g_task_return_boolean (close_data->task, TRUE);
1405         }
1406
1407       close_data_free (close_data);
1408     }
1409
1410   g_clear_error (&error);
1411
1412   /* all messages queued for sending are discarded */
1413   g_queue_free_full (send_queue, (GDestroyNotify) message_to_write_data_free);
1414   /* all queued flushes fail */
1415   error = g_error_new (G_IO_ERROR, G_IO_ERROR_CANCELLED,
1416                        _("Operation was cancelled"));
1417   flush_data_list_complete (pending_flush_attempts, error);
1418   g_list_free (pending_flush_attempts);
1419   g_clear_error (&error);
1420
1421   _g_dbus_worker_unref (worker);
1422 }
1423
1424 /* called in private thread shared by all GDBusConnection instances
1425  *
1426  * write-lock is not held on entry
1427  * output_pending must be PENDING_NONE on entry
1428  */
1429 static void
1430 continue_writing (GDBusWorker *worker)
1431 {
1432   MessageToWriteData *data;
1433   FlushAsyncData *flush_async_data;
1434
1435  write_next:
1436   /* we mustn't try to write two things at once */
1437   g_assert (worker->output_pending == PENDING_NONE);
1438
1439   g_mutex_lock (&worker->write_lock);
1440
1441   data = NULL;
1442   flush_async_data = NULL;
1443
1444   /* if we want to close the connection, that takes precedence */
1445   if (worker->pending_close_attempts != NULL)
1446     {
1447       GInputStream *input = g_io_stream_get_input_stream (worker->stream);
1448
1449       if (!g_input_stream_has_pending (input))
1450         {
1451           worker->close_expected = TRUE;
1452           worker->output_pending = PENDING_CLOSE;
1453
1454           g_io_stream_close_async (worker->stream, G_PRIORITY_DEFAULT,
1455                                    NULL, iostream_close_cb,
1456                                    _g_dbus_worker_ref (worker));
1457         }
1458     }
1459   else
1460     {
1461       flush_async_data = prepare_flush_unlocked (worker);
1462
1463       if (flush_async_data == NULL)
1464         {
1465           data = g_queue_pop_head (worker->write_queue);
1466
1467           if (data != NULL)
1468             worker->output_pending = PENDING_WRITE;
1469         }
1470     }
1471
1472   g_mutex_unlock (&worker->write_lock);
1473
1474   /* Note that write_lock is only used for protecting the @write_queue
1475    * and @output_pending fields of the GDBusWorker struct ... which we
1476    * need to modify from arbitrary threads in _g_dbus_worker_send_message().
1477    *
1478    * Therefore, it's fine to drop it here when calling back into user
1479    * code and then writing the message out onto the GIOStream since this
1480    * function only runs on the worker thread.
1481    */
1482
1483   if (flush_async_data != NULL)
1484     {
1485       start_flush (flush_async_data);
1486       g_assert (data == NULL);
1487     }
1488   else if (data != NULL)
1489     {
1490       GDBusMessage *old_message;
1491       guchar *new_blob;
1492       gsize new_blob_size;
1493       GError *error;
1494
1495       old_message = data->message;
1496       data->message = _g_dbus_worker_emit_message_about_to_be_sent (worker, data->message);
1497       if (data->message == old_message)
1498         {
1499           /* filters had no effect - do nothing */
1500         }
1501       else if (data->message == NULL)
1502         {
1503           /* filters dropped message */
1504           g_mutex_lock (&worker->write_lock);
1505           worker->output_pending = PENDING_NONE;
1506           g_mutex_unlock (&worker->write_lock);
1507           message_to_write_data_free (data);
1508           goto write_next;
1509         }
1510       else
1511         {
1512           /* filters altered the message -> re-encode */
1513           error = NULL;
1514           new_blob = g_dbus_message_to_blob (data->message,
1515                                              &new_blob_size,
1516                                              worker->capabilities,
1517                                              &error);
1518           if (new_blob == NULL)
1519             {
1520               /* if filter make the GDBusMessage unencodeable, just complain on stderr and send
1521                * the old message instead
1522                */
1523               g_warning ("Error encoding GDBusMessage with serial %d altered by filter function: %s",
1524                          g_dbus_message_get_serial (data->message),
1525                          error->message);
1526               g_error_free (error);
1527             }
1528           else
1529             {
1530               g_free (data->blob);
1531               data->blob = (gchar *) new_blob;
1532               data->blob_size = new_blob_size;
1533             }
1534         }
1535
1536       write_message_async (worker,
1537                            data,
1538                            write_message_cb,
1539                            data);
1540     }
1541 }
1542
1543 /* called in private thread shared by all GDBusConnection instances
1544  *
1545  * write-lock is not held on entry
1546  * output_pending may be anything
1547  */
1548 static gboolean
1549 continue_writing_in_idle_cb (gpointer user_data)
1550 {
1551   GDBusWorker *worker = user_data;
1552
1553   /* Because this is the worker thread, we can read this struct member
1554    * without holding the lock: no other thread ever modifies it.
1555    */
1556   if (worker->output_pending == PENDING_NONE)
1557     continue_writing (worker);
1558
1559   return FALSE;
1560 }
1561
1562 /*
1563  * @write_data: (transfer full) (nullable):
1564  * @flush_data: (transfer full) (nullable):
1565  * @close_data: (transfer full) (nullable):
1566  *
1567  * Can be called from any thread
1568  *
1569  * write_lock is held on entry
1570  * output_pending may be anything
1571  */
1572 static void
1573 schedule_writing_unlocked (GDBusWorker        *worker,
1574                            MessageToWriteData *write_data,
1575                            FlushData          *flush_data,
1576                            CloseData          *close_data)
1577 {
1578   if (write_data != NULL)
1579     g_queue_push_tail (worker->write_queue, write_data);
1580
1581   if (flush_data != NULL)
1582     worker->write_pending_flushes = g_list_prepend (worker->write_pending_flushes, flush_data);
1583
1584   if (close_data != NULL)
1585     worker->pending_close_attempts = g_list_prepend (worker->pending_close_attempts,
1586                                                      close_data);
1587
1588   /* If we had output pending, the next bit of output will happen
1589    * automatically when it finishes, so we only need to do this
1590    * if nothing was pending.
1591    *
1592    * The idle callback will re-check that output_pending is still
1593    * PENDING_NONE, to guard against output starting before the idle.
1594    */
1595   if (worker->output_pending == PENDING_NONE)
1596     {
1597       GSource *idle_source;
1598       idle_source = g_idle_source_new ();
1599       g_source_set_priority (idle_source, G_PRIORITY_DEFAULT);
1600       g_source_set_callback (idle_source,
1601                              continue_writing_in_idle_cb,
1602                              _g_dbus_worker_ref (worker),
1603                              (GDestroyNotify) _g_dbus_worker_unref);
1604       g_source_set_name (idle_source, "[gio] continue_writing_in_idle_cb");
1605       g_source_attach (idle_source, worker->shared_thread_data->context);
1606       g_source_unref (idle_source);
1607     }
1608 }
1609
1610 static void
1611 schedule_pending_close (GDBusWorker *worker)
1612 {
1613   g_mutex_lock (&worker->write_lock);
1614   if (worker->pending_close_attempts)
1615     schedule_writing_unlocked (worker, NULL, NULL, NULL);
1616   g_mutex_unlock (&worker->write_lock);
1617 }
1618
1619 /* ---------------------------------------------------------------------------------------------------- */
1620
1621 /* can be called from any thread - steals blob
1622  *
1623  * write_lock is not held on entry
1624  * output_pending may be anything
1625  */
1626 void
1627 _g_dbus_worker_send_message (GDBusWorker    *worker,
1628                              GDBusMessage   *message,
1629                              gchar          *blob,
1630                              gsize           blob_len)
1631 {
1632   MessageToWriteData *data;
1633
1634   g_return_if_fail (G_IS_DBUS_MESSAGE (message));
1635   g_return_if_fail (blob != NULL);
1636   g_return_if_fail (blob_len > 16);
1637
1638   data = g_slice_new0 (MessageToWriteData);
1639   data->worker = _g_dbus_worker_ref (worker);
1640   data->message = g_object_ref (message);
1641   data->blob = blob; /* steal! */
1642   data->blob_size = blob_len;
1643
1644   g_mutex_lock (&worker->write_lock);
1645   schedule_writing_unlocked (worker, data, NULL, NULL);
1646   g_mutex_unlock (&worker->write_lock);
1647 }
1648
1649 /* ---------------------------------------------------------------------------------------------------- */
1650
1651 GDBusWorker *
1652 _g_dbus_worker_new (GIOStream                              *stream,
1653                     GDBusCapabilityFlags                    capabilities,
1654                     gboolean                                initially_frozen,
1655                     GDBusWorkerMessageReceivedCallback      message_received_callback,
1656                     GDBusWorkerMessageAboutToBeSentCallback message_about_to_be_sent_callback,
1657                     GDBusWorkerDisconnectedCallback         disconnected_callback,
1658                     gpointer                                user_data)
1659 {
1660   GDBusWorker *worker;
1661   GSource *idle_source;
1662
1663   g_return_val_if_fail (G_IS_IO_STREAM (stream), NULL);
1664   g_return_val_if_fail (message_received_callback != NULL, NULL);
1665   g_return_val_if_fail (message_about_to_be_sent_callback != NULL, NULL);
1666   g_return_val_if_fail (disconnected_callback != NULL, NULL);
1667
1668   worker = g_new0 (GDBusWorker, 1);
1669   worker->ref_count = 1;
1670
1671   g_mutex_init (&worker->read_lock);
1672   worker->message_received_callback = message_received_callback;
1673   worker->message_about_to_be_sent_callback = message_about_to_be_sent_callback;
1674   worker->disconnected_callback = disconnected_callback;
1675   worker->user_data = user_data;
1676   worker->stream = g_object_ref (stream);
1677   worker->capabilities = capabilities;
1678   worker->cancellable = g_cancellable_new ();
1679   worker->output_pending = PENDING_NONE;
1680
1681   worker->frozen = initially_frozen;
1682   worker->received_messages_while_frozen = g_queue_new ();
1683
1684   g_mutex_init (&worker->write_lock);
1685   worker->write_queue = g_queue_new ();
1686
1687   if (G_IS_SOCKET_CONNECTION (worker->stream))
1688     worker->socket = g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker->stream));
1689
1690   worker->shared_thread_data = _g_dbus_shared_thread_ref ();
1691
1692   /* begin reading */
1693   idle_source = g_idle_source_new ();
1694   g_source_set_priority (idle_source, G_PRIORITY_DEFAULT);
1695   g_source_set_callback (idle_source,
1696                          _g_dbus_worker_do_initial_read,
1697                          _g_dbus_worker_ref (worker),
1698                          (GDestroyNotify) _g_dbus_worker_unref);
1699   g_source_set_name (idle_source, "[gio] _g_dbus_worker_do_initial_read");
1700   g_source_attach (idle_source, worker->shared_thread_data->context);
1701   g_source_unref (idle_source);
1702
1703   return worker;
1704 }
1705
1706 /* ---------------------------------------------------------------------------------------------------- */
1707
1708 /* can be called from any thread
1709  *
1710  * write_lock is not held on entry
1711  * output_pending may be anything
1712  */
1713 void
1714 _g_dbus_worker_close (GDBusWorker         *worker,
1715                       GTask               *task)
1716 {
1717   CloseData *close_data;
1718
1719   close_data = g_slice_new0 (CloseData);
1720   close_data->worker = _g_dbus_worker_ref (worker);
1721   close_data->task = (task == NULL ? NULL : g_object_ref (task));
1722
1723   /* Don't set worker->close_expected here - we're in the wrong thread.
1724    * It'll be set before the actual close happens.
1725    */
1726   g_cancellable_cancel (worker->cancellable);
1727   g_mutex_lock (&worker->write_lock);
1728   schedule_writing_unlocked (worker, NULL, NULL, close_data);
1729   g_mutex_unlock (&worker->write_lock);
1730 }
1731
1732 /* This can be called from any thread - frees worker. Note that
1733  * callbacks might still happen if called from another thread than the
1734  * worker - use your own synchronization primitive in the callbacks.
1735  *
1736  * write_lock is not held on entry
1737  * output_pending may be anything
1738  */
1739 void
1740 _g_dbus_worker_stop (GDBusWorker *worker)
1741 {
1742   g_atomic_int_set (&worker->stopped, TRUE);
1743
1744   /* Cancel any pending operations and schedule a close of the underlying I/O
1745    * stream in the worker thread
1746    */
1747   _g_dbus_worker_close (worker, NULL);
1748
1749   /* _g_dbus_worker_close holds a ref until after an idle in the worker
1750    * thread has run, so we no longer need to unref in an idle like in
1751    * commit 322e25b535
1752    */
1753   _g_dbus_worker_unref (worker);
1754 }
1755
1756 /* ---------------------------------------------------------------------------------------------------- */
1757
1758 /* can be called from any thread (except the worker thread) - blocks
1759  * calling thread until all queued outgoing messages are written and
1760  * the transport has been flushed
1761  *
1762  * write_lock is not held on entry
1763  * output_pending may be anything
1764  */
1765 gboolean
1766 _g_dbus_worker_flush_sync (GDBusWorker    *worker,
1767                            GCancellable   *cancellable,
1768                            GError        **error)
1769 {
1770   gboolean ret;
1771   FlushData *data;
1772   guint64 pending_writes;
1773
1774   data = NULL;
1775   ret = TRUE;
1776
1777   g_mutex_lock (&worker->write_lock);
1778
1779   /* if the queue is empty, no write is in-flight and we haven't written
1780    * anything since the last flush, then there's nothing to wait for
1781    */
1782   pending_writes = g_queue_get_length (worker->write_queue);
1783
1784   /* if a write is in-flight, we shouldn't be satisfied until the first
1785    * flush operation that follows it
1786    */
1787   if (worker->output_pending == PENDING_WRITE)
1788     pending_writes += 1;
1789
1790   if (pending_writes > 0 ||
1791       worker->write_num_messages_written != worker->write_num_messages_flushed)
1792     {
1793       data = g_new0 (FlushData, 1);
1794       g_mutex_init (&data->mutex);
1795       g_cond_init (&data->cond);
1796       data->number_to_wait_for = worker->write_num_messages_written + pending_writes;
1797       data->finished = FALSE;
1798       g_mutex_lock (&data->mutex);
1799
1800       schedule_writing_unlocked (worker, NULL, data, NULL);
1801     }
1802   g_mutex_unlock (&worker->write_lock);
1803
1804   if (data != NULL)
1805     {
1806       /* Wait for flush operations to finish. */
1807       while (!data->finished)
1808         {
1809           g_cond_wait (&data->cond, &data->mutex);
1810         }
1811
1812       g_mutex_unlock (&data->mutex);
1813       g_cond_clear (&data->cond);
1814       g_mutex_clear (&data->mutex);
1815       if (data->error != NULL)
1816         {
1817           ret = FALSE;
1818           g_propagate_error (error, data->error);
1819         }
1820       g_free (data);
1821     }
1822
1823   return ret;
1824 }
1825
1826 /* ---------------------------------------------------------------------------------------------------- */
1827
1828 #define G_DBUS_DEBUG_AUTHENTICATION (1<<0)
1829 #define G_DBUS_DEBUG_TRANSPORT      (1<<1)
1830 #define G_DBUS_DEBUG_MESSAGE        (1<<2)
1831 #define G_DBUS_DEBUG_PAYLOAD        (1<<3)
1832 #define G_DBUS_DEBUG_CALL           (1<<4)
1833 #define G_DBUS_DEBUG_SIGNAL         (1<<5)
1834 #define G_DBUS_DEBUG_INCOMING       (1<<6)
1835 #define G_DBUS_DEBUG_RETURN         (1<<7)
1836 #define G_DBUS_DEBUG_EMISSION       (1<<8)
1837 #define G_DBUS_DEBUG_ADDRESS        (1<<9)
1838 #define G_DBUS_DEBUG_PROXY          (1<<10)
1839
1840 static gint _gdbus_debug_flags = 0;
1841
1842 gboolean
1843 _g_dbus_debug_authentication (void)
1844 {
1845   _g_dbus_initialize ();
1846   return (_gdbus_debug_flags & G_DBUS_DEBUG_AUTHENTICATION) != 0;
1847 }
1848
1849 gboolean
1850 _g_dbus_debug_transport (void)
1851 {
1852   _g_dbus_initialize ();
1853   return (_gdbus_debug_flags & G_DBUS_DEBUG_TRANSPORT) != 0;
1854 }
1855
1856 gboolean
1857 _g_dbus_debug_message (void)
1858 {
1859   _g_dbus_initialize ();
1860   return (_gdbus_debug_flags & G_DBUS_DEBUG_MESSAGE) != 0;
1861 }
1862
1863 gboolean
1864 _g_dbus_debug_payload (void)
1865 {
1866   _g_dbus_initialize ();
1867   return (_gdbus_debug_flags & G_DBUS_DEBUG_PAYLOAD) != 0;
1868 }
1869
1870 gboolean
1871 _g_dbus_debug_call (void)
1872 {
1873   _g_dbus_initialize ();
1874   return (_gdbus_debug_flags & G_DBUS_DEBUG_CALL) != 0;
1875 }
1876
1877 gboolean
1878 _g_dbus_debug_signal (void)
1879 {
1880   _g_dbus_initialize ();
1881   return (_gdbus_debug_flags & G_DBUS_DEBUG_SIGNAL) != 0;
1882 }
1883
1884 gboolean
1885 _g_dbus_debug_incoming (void)
1886 {
1887   _g_dbus_initialize ();
1888   return (_gdbus_debug_flags & G_DBUS_DEBUG_INCOMING) != 0;
1889 }
1890
1891 gboolean
1892 _g_dbus_debug_return (void)
1893 {
1894   _g_dbus_initialize ();
1895   return (_gdbus_debug_flags & G_DBUS_DEBUG_RETURN) != 0;
1896 }
1897
1898 gboolean
1899 _g_dbus_debug_emission (void)
1900 {
1901   _g_dbus_initialize ();
1902   return (_gdbus_debug_flags & G_DBUS_DEBUG_EMISSION) != 0;
1903 }
1904
1905 gboolean
1906 _g_dbus_debug_address (void)
1907 {
1908   _g_dbus_initialize ();
1909   return (_gdbus_debug_flags & G_DBUS_DEBUG_ADDRESS) != 0;
1910 }
1911
1912 gboolean
1913 _g_dbus_debug_proxy (void)
1914 {
1915   _g_dbus_initialize ();
1916   return (_gdbus_debug_flags & G_DBUS_DEBUG_PROXY) != 0;
1917 }
1918
1919 G_LOCK_DEFINE_STATIC (print_lock);
1920
1921 void
1922 _g_dbus_debug_print_lock (void)
1923 {
1924   G_LOCK (print_lock);
1925 }
1926
1927 void
1928 _g_dbus_debug_print_unlock (void)
1929 {
1930   G_UNLOCK (print_lock);
1931 }
1932
1933 /**
1934  * _g_dbus_initialize:
1935  *
1936  * Does various one-time init things such as
1937  *
1938  *  - registering the G_DBUS_ERROR error domain
1939  *  - parses the G_DBUS_DEBUG environment variable
1940  */
1941 void
1942 _g_dbus_initialize (void)
1943 {
1944   static volatile gsize initialized = 0;
1945
1946   if (g_once_init_enter (&initialized))
1947     {
1948       volatile GQuark g_dbus_error_domain;
1949       const gchar *debug;
1950
1951       g_dbus_error_domain = G_DBUS_ERROR;
1952       (g_dbus_error_domain); /* To avoid -Wunused-but-set-variable */
1953
1954       debug = g_getenv ("G_DBUS_DEBUG");
1955       if (debug != NULL)
1956         {
1957           const GDebugKey keys[] = {
1958             { "authentication", G_DBUS_DEBUG_AUTHENTICATION },
1959             { "transport",      G_DBUS_DEBUG_TRANSPORT      },
1960             { "message",        G_DBUS_DEBUG_MESSAGE        },
1961             { "payload",        G_DBUS_DEBUG_PAYLOAD        },
1962             { "call",           G_DBUS_DEBUG_CALL           },
1963             { "signal",         G_DBUS_DEBUG_SIGNAL         },
1964             { "incoming",       G_DBUS_DEBUG_INCOMING       },
1965             { "return",         G_DBUS_DEBUG_RETURN         },
1966             { "emission",       G_DBUS_DEBUG_EMISSION       },
1967             { "address",        G_DBUS_DEBUG_ADDRESS        },
1968             { "proxy",          G_DBUS_DEBUG_PROXY          }
1969           };
1970
1971           _gdbus_debug_flags = g_parse_debug_string (debug, keys, G_N_ELEMENTS (keys));
1972           if (_gdbus_debug_flags & G_DBUS_DEBUG_PAYLOAD)
1973             _gdbus_debug_flags |= G_DBUS_DEBUG_MESSAGE;
1974         }
1975
1976       /* Work-around for https://bugzilla.gnome.org/show_bug.cgi?id=627724 */
1977       ensure_required_types ();
1978
1979       g_once_init_leave (&initialized, 1);
1980     }
1981 }
1982
1983 /* ---------------------------------------------------------------------------------------------------- */
1984
1985 GVariantType *
1986 _g_dbus_compute_complete_signature (GDBusArgInfo **args)
1987 {
1988   const GVariantType *arg_types[256];
1989   guint n;
1990
1991   if (args)
1992     for (n = 0; args[n] != NULL; n++)
1993       {
1994         /* DBus places a hard limit of 255 on signature length.
1995          * therefore number of args must be less than 256.
1996          */
1997         g_assert (n < 256);
1998
1999         arg_types[n] = G_VARIANT_TYPE (args[n]->signature);
2000
2001         if G_UNLIKELY (arg_types[n] == NULL)
2002           return NULL;
2003       }
2004   else
2005     n = 0;
2006
2007   return g_variant_type_new_tuple (arg_types, n);
2008 }
2009
2010 /* ---------------------------------------------------------------------------------------------------- */
2011
2012 #ifdef G_OS_WIN32
2013
2014 extern BOOL WINAPI ConvertSidToStringSidA (PSID Sid, LPSTR *StringSid);
2015
2016 gchar *
2017 _g_dbus_win32_get_user_sid (void)
2018 {
2019   HANDLE h;
2020   TOKEN_USER *user;
2021   DWORD token_information_len;
2022   PSID psid;
2023   gchar *sid;
2024   gchar *ret;
2025
2026   ret = NULL;
2027   user = NULL;
2028   h = INVALID_HANDLE_VALUE;
2029
2030   if (!OpenProcessToken (GetCurrentProcess (), TOKEN_QUERY, &h))
2031     {
2032       g_warning ("OpenProcessToken failed with error code %d", (gint) GetLastError ());
2033       goto out;
2034     }
2035
2036   /* Get length of buffer */
2037   token_information_len = 0;
2038   if (!GetTokenInformation (h, TokenUser, NULL, 0, &token_information_len))
2039     {
2040       if (GetLastError () != ERROR_INSUFFICIENT_BUFFER)
2041         {
2042           g_warning ("GetTokenInformation() failed with error code %d", (gint) GetLastError ());
2043           goto out;
2044         }
2045     }
2046   user = g_malloc (token_information_len);
2047   if (!GetTokenInformation (h, TokenUser, user, token_information_len, &token_information_len))
2048     {
2049       g_warning ("GetTokenInformation() failed with error code %d", (gint) GetLastError ());
2050       goto out;
2051     }
2052
2053   psid = user->User.Sid;
2054   if (!IsValidSid (psid))
2055     {
2056       g_warning ("Invalid SID");
2057       goto out;
2058     }
2059
2060   if (!ConvertSidToStringSidA (psid, &sid))
2061     {
2062       g_warning ("Invalid SID");
2063       goto out;
2064     }
2065
2066   ret = g_strdup (sid);
2067   LocalFree (sid);
2068
2069 out:
2070   g_free (user);
2071   if (h != INVALID_HANDLE_VALUE)
2072     CloseHandle (h);
2073   return ret;
2074 }
2075
2076
2077 #define DBUS_DAEMON_ADDRESS_INFO "DBusDaemonAddressInfo"
2078 #define DBUS_DAEMON_MUTEX "DBusDaemonMutex"
2079 #define UNIQUE_DBUS_INIT_MUTEX "UniqueDBusInitMutex"
2080 #define DBUS_AUTOLAUNCH_MUTEX "DBusAutolaunchMutex"
2081
2082 static void
2083 release_mutex (HANDLE mutex)
2084 {
2085   ReleaseMutex (mutex);
2086   CloseHandle (mutex);
2087 }
2088
2089 static HANDLE
2090 acquire_mutex (const char *mutexname)
2091 {
2092   HANDLE mutex;
2093   DWORD res;
2094
2095   mutex = CreateMutexA (NULL, FALSE, mutexname);
2096   if (!mutex)
2097     return 0;
2098
2099   res = WaitForSingleObject (mutex, INFINITE);
2100   switch (res)
2101     {
2102     case WAIT_ABANDONED:
2103       release_mutex (mutex);
2104       return 0;
2105     case WAIT_FAILED:
2106     case WAIT_TIMEOUT:
2107       return 0;
2108     }
2109
2110   return mutex;
2111 }
2112
2113 static gboolean
2114 is_mutex_owned (const char *mutexname)
2115 {
2116   HANDLE mutex;
2117   gboolean res = FALSE;
2118
2119   mutex = CreateMutexA (NULL, FALSE, mutexname);
2120   if (WaitForSingleObject (mutex, 10) == WAIT_TIMEOUT)
2121     res = TRUE;
2122   else
2123     ReleaseMutex (mutex);
2124   CloseHandle (mutex);
2125
2126   return res;
2127 }
2128
2129 static char *
2130 read_shm (const char *shm_name)
2131 {
2132   HANDLE shared_mem;
2133   char *shared_data;
2134   char *res;
2135   int i;
2136
2137   res = NULL;
2138
2139   for (i = 0; i < 20; i++)
2140     {
2141       shared_mem = OpenFileMappingA (FILE_MAP_READ, FALSE, shm_name);
2142       if (shared_mem != 0)
2143         break;
2144       Sleep (100);
2145     }
2146
2147   if (shared_mem != 0)
2148     {
2149       shared_data = MapViewOfFile (shared_mem, FILE_MAP_READ, 0, 0, 0);
2150       /* It looks that a race is possible here:
2151        * if the dbus process already created mapping but didn't fill it
2152        * the code below may read incorrect address.
2153        * Also this is a bit complicated by the fact that
2154        * any change in the "synchronization contract" between processes
2155        * should be accompanied with renaming all of used win32 named objects:
2156        * otherwise libgio-2.0-0.dll of different versions shipped with
2157        * different apps may break each other due to protocol difference.
2158        */
2159       if (shared_data != NULL)
2160         {
2161           res = g_strdup (shared_data);
2162           UnmapViewOfFile (shared_data);
2163         }
2164       CloseHandle (shared_mem);
2165     }
2166
2167   return res;
2168 }
2169
2170 static HANDLE
2171 set_shm (const char *shm_name, const char *value)
2172 {
2173   HANDLE shared_mem;
2174   char *shared_data;
2175
2176   shared_mem = CreateFileMappingA (INVALID_HANDLE_VALUE, NULL, PAGE_READWRITE,
2177                                    0, strlen (value) + 1, shm_name);
2178   if (shared_mem == 0)
2179     return 0;
2180
2181   shared_data = MapViewOfFile (shared_mem, FILE_MAP_WRITE, 0, 0, 0 );
2182   if (shared_data == NULL)
2183     return 0;
2184
2185   strcpy (shared_data, value);
2186
2187   UnmapViewOfFile (shared_data);
2188
2189   return shared_mem;
2190 }
2191
2192 /* These keep state between publish_session_bus and unpublish_session_bus */
2193 static HANDLE published_daemon_mutex;
2194 static HANDLE published_shared_mem;
2195
2196 static gboolean
2197 publish_session_bus (const char *address)
2198 {
2199   HANDLE init_mutex;
2200
2201   init_mutex = acquire_mutex (UNIQUE_DBUS_INIT_MUTEX);
2202
2203   published_daemon_mutex = CreateMutexA (NULL, FALSE, DBUS_DAEMON_MUTEX);
2204   if (WaitForSingleObject (published_daemon_mutex, 10 ) != WAIT_OBJECT_0)
2205     {
2206       release_mutex (init_mutex);
2207       CloseHandle (published_daemon_mutex);
2208       published_daemon_mutex = NULL;
2209       return FALSE;
2210     }
2211
2212   published_shared_mem = set_shm (DBUS_DAEMON_ADDRESS_INFO, address);
2213   if (!published_shared_mem)
2214     {
2215       release_mutex (init_mutex);
2216       CloseHandle (published_daemon_mutex);
2217       published_daemon_mutex = NULL;
2218       return FALSE;
2219     }
2220
2221   release_mutex (init_mutex);
2222   return TRUE;
2223 }
2224
2225 static void
2226 unpublish_session_bus (void)
2227 {
2228   HANDLE init_mutex;
2229
2230   init_mutex = acquire_mutex (UNIQUE_DBUS_INIT_MUTEX);
2231
2232   CloseHandle (published_shared_mem);
2233   published_shared_mem = NULL;
2234
2235   release_mutex (published_daemon_mutex);
2236   published_daemon_mutex = NULL;
2237
2238   release_mutex (init_mutex);
2239 }
2240
2241 static void
2242 wait_console_window (void)
2243 {
2244   FILE *console = fopen ("CONOUT$", "w");
2245
2246   SetConsoleTitleW (L"gdbus-daemon output. Type any character to close this window.");
2247   fprintf (console, _("(Type any character to close this window)\n"));
2248   fflush (console);
2249   _getch ();
2250 }
2251
2252 static void
2253 open_console_window (void)
2254 {
2255   if (((HANDLE) _get_osfhandle (fileno (stdout)) == INVALID_HANDLE_VALUE ||
2256        (HANDLE) _get_osfhandle (fileno (stderr)) == INVALID_HANDLE_VALUE) && AllocConsole ())
2257     {
2258       if ((HANDLE) _get_osfhandle (fileno (stdout)) == INVALID_HANDLE_VALUE)
2259         freopen ("CONOUT$", "w", stdout);
2260
2261       if ((HANDLE) _get_osfhandle (fileno (stderr)) == INVALID_HANDLE_VALUE)
2262         freopen ("CONOUT$", "w", stderr);
2263
2264       SetConsoleTitleW (L"gdbus-daemon debug output.");
2265
2266       atexit (wait_console_window);
2267     }
2268 }
2269
2270 static void
2271 idle_timeout_cb (GDBusDaemon *daemon, gpointer user_data)
2272 {
2273   GMainLoop *loop = user_data;
2274   g_main_loop_quit (loop);
2275 }
2276
2277 /* Satisfies STARTF_FORCEONFEEDBACK */
2278 static void
2279 turn_off_the_starting_cursor (void)
2280 {
2281   MSG msg;
2282   BOOL bRet;
2283
2284   PostQuitMessage (0);
2285
2286   while ((bRet = GetMessage (&msg, 0, 0, 0)) != 0)
2287     {
2288       if (bRet == -1)
2289         continue;
2290
2291       TranslateMessage (&msg);
2292       DispatchMessage (&msg);
2293     }
2294 }
2295
2296 __declspec(dllexport) void __stdcall
2297 g_win32_run_session_bus (void* hwnd, void* hinst, const char* cmdline, int cmdshow)
2298 {
2299   GDBusDaemon *daemon;
2300   GMainLoop *loop;
2301   const char *address;
2302   GError *error = NULL;
2303
2304   turn_off_the_starting_cursor ();
2305
2306   if (g_getenv ("GDBUS_DAEMON_DEBUG") != NULL)
2307     open_console_window ();
2308
2309   address = "nonce-tcp:";
2310   daemon = _g_dbus_daemon_new (address, NULL, &error);
2311   if (daemon == NULL)
2312     {
2313       g_printerr ("Can't init bus: %s\n", error->message);
2314       g_error_free (error);
2315       return;
2316     }
2317
2318   loop = g_main_loop_new (NULL, FALSE);
2319
2320   /* There is a subtle detail with "idle-timeout" signal of dbus daemon:
2321    * It is fired on idle after last client disconnection,
2322    * but (at least with glib 2.59.1) it is NEVER fired
2323    * if no clients connect to daemon at all.
2324    * This may lead to infinite run of this daemon process.
2325    */
2326   g_signal_connect (daemon, "idle-timeout", G_CALLBACK (idle_timeout_cb), loop);
2327
2328   if (publish_session_bus (_g_dbus_daemon_get_address (daemon)))
2329     {
2330       g_main_loop_run (loop);
2331
2332       unpublish_session_bus ();
2333     }
2334
2335   g_main_loop_unref (loop);
2336   g_object_unref (daemon);
2337 }
2338
2339 static gboolean autolaunch_binary_absent = FALSE;
2340
2341 gchar *
2342 _g_dbus_win32_get_session_address_dbus_launch (GError **error)
2343 {
2344   HANDLE autolaunch_mutex, init_mutex;
2345   char *address = NULL;
2346
2347   autolaunch_mutex = acquire_mutex (DBUS_AUTOLAUNCH_MUTEX);
2348
2349   init_mutex = acquire_mutex (UNIQUE_DBUS_INIT_MUTEX);
2350
2351   if (is_mutex_owned (DBUS_DAEMON_MUTEX))
2352     address = read_shm (DBUS_DAEMON_ADDRESS_INFO);
2353
2354   release_mutex (init_mutex);
2355
2356   if (address == NULL && !autolaunch_binary_absent)
2357     {
2358       wchar_t gio_path[MAX_PATH + 2] = { 0 };
2359       int gio_path_len = GetModuleFileNameW (_g_io_win32_get_module (), gio_path, MAX_PATH + 1);
2360
2361       /* The <= MAX_PATH check prevents truncated path usage */
2362       if (gio_path_len > 0 && gio_path_len <= MAX_PATH)
2363         {
2364           PROCESS_INFORMATION pi = { 0 };
2365           STARTUPINFOW si = { 0 };
2366           BOOL res = FALSE;
2367           wchar_t exe_path[MAX_PATH + 100] = { 0 };
2368           /* calculate index of first char of dll file name inside full path */
2369           int gio_name_index = gio_path_len;
2370           for (; gio_name_index > 0; --gio_name_index)
2371           {
2372             wchar_t prev_char = gio_path[gio_name_index - 1];
2373             if (prev_char == L'\\' || prev_char == L'/')
2374               break;
2375           }
2376           gio_path[gio_name_index] = L'\0';
2377           wcscpy (exe_path, gio_path);
2378           wcscat (exe_path, L"\\gdbus.exe");
2379
2380           if (GetFileAttributesW (exe_path) == INVALID_FILE_ATTRIBUTES)
2381             {
2382               /* warning won't be raised another time
2383                * since autolaunch_binary_absent would be already set.
2384                */
2385               autolaunch_binary_absent = TRUE;
2386               g_warning ("win32 session dbus binary not found: %S", exe_path );
2387             }
2388           else
2389             {
2390               wchar_t args[MAX_PATH*2 + 100] = { 0 };
2391               wcscpy (args, L"\"");
2392               wcscat (args, exe_path);
2393               wcscat (args, L"\" ");
2394 #define _L_PREFIX_FOR_EXPANDED(arg) L##arg
2395 #define _L_PREFIX(arg) _L_PREFIX_FOR_EXPANDED (arg)
2396               wcscat (args, _L_PREFIX (_GDBUS_ARG_WIN32_RUN_SESSION_BUS));
2397 #undef _L_PREFIX
2398 #undef _L_PREFIX_FOR_EXPANDED
2399
2400               res = CreateProcessW (exe_path, args,
2401                                     0, 0, FALSE,
2402                                     NORMAL_PRIORITY_CLASS | CREATE_NO_WINDOW | DETACHED_PROCESS,
2403                                     0, gio_path,
2404                                     &si, &pi);
2405             }
2406           if (res)
2407             {
2408               address = read_shm (DBUS_DAEMON_ADDRESS_INFO);
2409               if (address == NULL)
2410                 g_warning ("%S dbus binary failed to launch bus, maybe incompatible version", exe_path );
2411             }
2412         }
2413     }
2414
2415   release_mutex (autolaunch_mutex);
2416
2417   if (address == NULL)
2418     g_set_error (error,
2419                  G_IO_ERROR,
2420                  G_IO_ERROR_FAILED,
2421                  _("Session dbus not running, and autolaunch failed"));
2422
2423   return address;
2424 }
2425
2426 #endif
2427
2428 /* ---------------------------------------------------------------------------------------------------- */
2429
2430 gchar *
2431 _g_dbus_get_machine_id (GError **error)
2432 {
2433 #ifdef G_OS_WIN32
2434   HW_PROFILE_INFOA info;
2435   char *src, *dest, *res;
2436   int i;
2437
2438   if (!GetCurrentHwProfileA (&info))
2439     {
2440       char *message = g_win32_error_message (GetLastError ());
2441       g_set_error (error,
2442                    G_IO_ERROR,
2443                    G_IO_ERROR_FAILED,
2444                    _("Unable to get Hardware profile: %s"), message);
2445       g_free (message);
2446       return NULL;
2447     }
2448
2449   /* Form: {12340001-4980-1920-6788-123456789012} */
2450   src = &info.szHwProfileGuid[0];
2451
2452   res = g_malloc (32+1);
2453   dest = res;
2454
2455   src++; /* Skip { */
2456   for (i = 0; i < 8; i++)
2457     *dest++ = *src++;
2458   src++; /* Skip - */
2459   for (i = 0; i < 4; i++)
2460     *dest++ = *src++;
2461   src++; /* Skip - */
2462   for (i = 0; i < 4; i++)
2463     *dest++ = *src++;
2464   src++; /* Skip - */
2465   for (i = 0; i < 4; i++)
2466     *dest++ = *src++;
2467   src++; /* Skip - */
2468   for (i = 0; i < 12; i++)
2469     *dest++ = *src++;
2470   *dest = 0;
2471
2472   return res;
2473 #else
2474   gchar *ret;
2475   GError *first_error;
2476   /* TODO: use PACKAGE_LOCALSTATEDIR ? */
2477   ret = NULL;
2478   first_error = NULL;
2479   if (!g_file_get_contents ("/var/lib/dbus/machine-id",
2480                             &ret,
2481                             NULL,
2482                             &first_error) &&
2483       !g_file_get_contents ("/etc/machine-id",
2484                             &ret,
2485                             NULL,
2486                             NULL))
2487     {
2488       g_propagate_prefixed_error (error, first_error,
2489                                   _("Unable to load /var/lib/dbus/machine-id or /etc/machine-id: "));
2490     }
2491   else
2492     {
2493       /* ignore the error from the first try, if any */
2494       g_clear_error (&first_error);
2495       /* TODO: validate value */
2496       g_strstrip (ret);
2497     }
2498   return ret;
2499 #endif
2500 }
2501
2502 /* ---------------------------------------------------------------------------------------------------- */
2503
2504 gchar *
2505 _g_dbus_enum_to_string (GType enum_type, gint value)
2506 {
2507   gchar *ret;
2508   GEnumClass *klass;
2509   GEnumValue *enum_value;
2510
2511   klass = g_type_class_ref (enum_type);
2512   enum_value = g_enum_get_value (klass, value);
2513   if (enum_value != NULL)
2514     ret = g_strdup (enum_value->value_nick);
2515   else
2516     ret = g_strdup_printf ("unknown (value %d)", value);
2517   g_type_class_unref (klass);
2518   return ret;
2519 }
2520
2521 /* ---------------------------------------------------------------------------------------------------- */
2522
2523 static void
2524 write_message_print_transport_debug (gssize bytes_written,
2525                                      MessageToWriteData *data)
2526 {
2527   if (G_LIKELY (!_g_dbus_debug_transport ()))
2528     goto out;
2529
2530   _g_dbus_debug_print_lock ();
2531   g_print ("========================================================================\n"
2532            "GDBus-debug:Transport:\n"
2533            "  >>>> WROTE %" G_GSSIZE_FORMAT " bytes of message with serial %d and\n"
2534            "       size %" G_GSIZE_FORMAT " from offset %" G_GSIZE_FORMAT " on a %s\n",
2535            bytes_written,
2536            g_dbus_message_get_serial (data->message),
2537            data->blob_size,
2538            data->total_written,
2539            g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_output_stream (data->worker->stream))));
2540   _g_dbus_debug_print_unlock ();
2541  out:
2542   ;
2543 }
2544
2545 /* ---------------------------------------------------------------------------------------------------- */
2546
2547 static void
2548 read_message_print_transport_debug (gssize bytes_read,
2549                                     GDBusWorker *worker)
2550 {
2551   gsize size;
2552   gint32 serial;
2553   gint32 message_length;
2554
2555   if (G_LIKELY (!_g_dbus_debug_transport ()))
2556     goto out;
2557
2558   size = bytes_read + worker->read_buffer_cur_size;
2559   serial = 0;
2560   message_length = 0;
2561   if (size >= 16)
2562     message_length = g_dbus_message_bytes_needed ((guchar *) worker->read_buffer, size, NULL);
2563   if (size >= 1)
2564     {
2565       switch (worker->read_buffer[0])
2566         {
2567         case 'l':
2568           if (size >= 12)
2569             serial = GUINT32_FROM_LE (((guint32 *) worker->read_buffer)[2]);
2570           break;
2571         case 'B':
2572           if (size >= 12)
2573             serial = GUINT32_FROM_BE (((guint32 *) worker->read_buffer)[2]);
2574           break;
2575         default:
2576           /* an error will be set elsewhere if this happens */
2577           goto out;
2578         }
2579     }
2580
2581     _g_dbus_debug_print_lock ();
2582   g_print ("========================================================================\n"
2583            "GDBus-debug:Transport:\n"
2584            "  <<<< READ %" G_GSSIZE_FORMAT " bytes of message with serial %d and\n"
2585            "       size %d to offset %" G_GSIZE_FORMAT " from a %s\n",
2586            bytes_read,
2587            serial,
2588            message_length,
2589            worker->read_buffer_cur_size,
2590            g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_input_stream (worker->stream))));
2591   _g_dbus_debug_print_unlock ();
2592  out:
2593   ;
2594 }
2595
2596 /* ---------------------------------------------------------------------------------------------------- */
2597
2598 gboolean
2599 _g_signal_accumulator_false_handled (GSignalInvocationHint *ihint,
2600                                      GValue                *return_accu,
2601                                      const GValue          *handler_return,
2602                                      gpointer               dummy)
2603 {
2604   gboolean continue_emission;
2605   gboolean signal_return;
2606
2607   signal_return = g_value_get_boolean (handler_return);
2608   g_value_set_boolean (return_accu, signal_return);
2609   continue_emission = signal_return;
2610
2611   return continue_emission;
2612 }
2613
2614 /* ---------------------------------------------------------------------------------------------------- */
2615
2616 static void
2617 append_nibble (GString *s, gint val)
2618 {
2619   g_string_append_c (s, val >= 10 ? ('a' + val - 10) : ('0' + val));
2620 }
2621
2622 /* ---------------------------------------------------------------------------------------------------- */
2623
2624 gchar *
2625 _g_dbus_hexencode (const gchar *str,
2626                    gsize        str_len)
2627 {
2628   gsize n;
2629   GString *s;
2630
2631   s = g_string_new (NULL);
2632   for (n = 0; n < str_len; n++)
2633     {
2634       gint val;
2635       gint upper_nibble;
2636       gint lower_nibble;
2637
2638       val = ((const guchar *) str)[n];
2639       upper_nibble = val >> 4;
2640       lower_nibble = val & 0x0f;
2641
2642       append_nibble (s, upper_nibble);
2643       append_nibble (s, lower_nibble);
2644     }
2645
2646   return g_string_free (s, FALSE);
2647 }