gio/gdbusprivate.c

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