1 /* GDBus - GLib D-Bus Library
3 * Copyright (C) 2008-2010 Red Hat, Inc.
5 * This library is free software; you can redistribute it and/or
6 * modify it under the terms of the GNU Lesser General Public
7 * License as published by the Free Software Foundation; either
8 * version 2 of the License, or (at your option) any later version.
10 * This library is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 * Lesser General Public License for more details.
15 * You should have received a copy of the GNU Lesser General
16 * Public License along with this library; if not, see <http://www.gnu.org/licenses/>.
18 * Author: David Zeuthen <davidz@redhat.com>
28 #include "gdbusprivate.h"
29 #include "gdbusmessage.h"
30 #include "gdbusconnection.h"
31 #include "gdbusproxy.h"
32 #include "gdbuserror.h"
33 #include "gdbusintrospection.h"
35 #include "ginputstream.h"
36 #include "gmemoryinputstream.h"
37 #include "giostream.h"
38 #include "glib/gstdio.h"
39 #include "gsocketcontrolmessage.h"
40 #include "gsocketconnection.h"
41 #include "gsocketoutputstream.h"
44 #include "gunixfdmessage.h"
45 #include "gunixconnection.h"
46 #include "gunixcredentialsmessage.h"
55 static gboolean
_g_dbus_worker_do_initial_read (gpointer data
);
56 static void schedule_pending_close (GDBusWorker
*worker
);
58 /* ---------------------------------------------------------------------------------------------------- */
61 _g_dbus_hexdump (const gchar
*data
, gsize len
, guint indent
)
66 ret
= g_string_new (NULL
);
68 for (n
= 0; n
< len
; n
+= 16)
70 g_string_append_printf (ret
, "%*s%04x: ", indent
, "", n
);
72 for (m
= n
; m
< n
+ 16; m
++)
74 if (m
> n
&& (m
%4) == 0)
75 g_string_append_c (ret
, ' ');
77 g_string_append_printf (ret
, "%02x ", (guchar
) data
[m
]);
79 g_string_append (ret
, " ");
82 g_string_append (ret
, " ");
84 for (m
= n
; m
< len
&& m
< n
+ 16; m
++)
85 g_string_append_c (ret
, g_ascii_isprint (data
[m
]) ? data
[m
] : '.');
87 g_string_append_c (ret
, '\n');
90 return g_string_free (ret
, FALSE
);
93 /* ---------------------------------------------------------------------------------------------------- */
95 /* Unfortunately ancillary messages are discarded when reading from a
96 * socket using the GSocketInputStream abstraction. So we provide a
97 * very GInputStream-ish API that uses GSocket in this case (very
98 * similar to GSocketInputStream).
106 GSocketControlMessage
***messages
;
108 } ReadWithControlData
;
111 read_with_control_data_free (ReadWithControlData
*data
)
113 g_slice_free (ReadWithControlData
, data
);
117 _g_socket_read_with_control_messages_ready (GSocket
*socket
,
118 GIOCondition condition
,
121 GTask
*task
= user_data
;
122 ReadWithControlData
*data
= g_task_get_task_data (task
);
128 vector
.buffer
= data
->buffer
;
129 vector
.size
= data
->count
;
130 result
= g_socket_receive_message (socket
,
137 g_task_get_cancellable (task
),
140 if (g_error_matches (error
, G_IO_ERROR
, G_IO_ERROR_WOULD_BLOCK
))
142 g_error_free (error
);
146 g_assert (result
>= 0 || error
!= NULL
);
148 g_task_return_int (task
, result
);
150 g_task_return_error (task
, error
);
151 g_object_unref (task
);
157 _g_socket_read_with_control_messages (GSocket
*socket
,
160 GSocketControlMessage
***messages
,
163 GCancellable
*cancellable
,
164 GAsyncReadyCallback callback
,
168 ReadWithControlData
*data
;
171 data
= g_slice_new0 (ReadWithControlData
);
172 data
->buffer
= buffer
;
174 data
->messages
= messages
;
175 data
->num_messages
= num_messages
;
177 task
= g_task_new (socket
, cancellable
, callback
, user_data
);
178 g_task_set_source_tag (task
, _g_socket_read_with_control_messages
);
179 g_task_set_task_data (task
, data
, (GDestroyNotify
) read_with_control_data_free
);
181 if (g_socket_condition_check (socket
, G_IO_IN
))
183 if (!_g_socket_read_with_control_messages_ready (socket
, G_IO_IN
, task
))
187 source
= g_socket_create_source (socket
,
188 G_IO_IN
| G_IO_HUP
| G_IO_ERR
,
190 g_task_attach_source (task
, source
, (GSourceFunc
) _g_socket_read_with_control_messages_ready
);
191 g_source_unref (source
);
195 _g_socket_read_with_control_messages_finish (GSocket
*socket
,
196 GAsyncResult
*result
,
199 g_return_val_if_fail (G_IS_SOCKET (socket
), -1);
200 g_return_val_if_fail (g_task_is_valid (result
, socket
), -1);
202 return g_task_propagate_int (G_TASK (result
), error
);
205 /* ---------------------------------------------------------------------------------------------------- */
207 /* Work-around for https://bugzilla.gnome.org/show_bug.cgi?id=674885
208 and see also the original https://bugzilla.gnome.org/show_bug.cgi?id=627724 */
210 static GPtrArray
*ensured_classes
= NULL
;
213 ensure_type (GType gtype
)
215 g_ptr_array_add (ensured_classes
, g_type_class_ref (gtype
));
219 release_required_types (void)
221 g_ptr_array_foreach (ensured_classes
, (GFunc
) g_type_class_unref
, NULL
);
222 g_ptr_array_unref (ensured_classes
);
223 ensured_classes
= NULL
;
227 ensure_required_types (void)
229 g_assert (ensured_classes
== NULL
);
230 ensured_classes
= g_ptr_array_new ();
231 ensure_type (G_TYPE_TASK
);
232 ensure_type (G_TYPE_MEMORY_INPUT_STREAM
);
233 ensure_type (G_TYPE_DBUS_CONNECTION
);
234 ensure_type (G_TYPE_DBUS_PROXY
);
236 /* ---------------------------------------------------------------------------------------------------- */
240 volatile gint refcount
;
242 GMainContext
*context
;
247 gdbus_shared_thread_func (gpointer user_data
)
249 SharedThreadData
*data
= user_data
;
251 g_main_context_push_thread_default (data
->context
);
252 g_main_loop_run (data
->loop
);
253 g_main_context_pop_thread_default (data
->context
);
255 release_required_types ();
260 /* ---------------------------------------------------------------------------------------------------- */
262 static SharedThreadData
*
263 _g_dbus_shared_thread_ref (void)
265 static gsize shared_thread_data
= 0;
266 SharedThreadData
*ret
;
268 if (g_once_init_enter (&shared_thread_data
))
270 SharedThreadData
*data
;
272 data
= g_new0 (SharedThreadData
, 1);
275 data
->context
= g_main_context_new ();
276 data
->loop
= g_main_loop_new (data
->context
, FALSE
);
277 data
->thread
= g_thread_new ("gdbus",
278 gdbus_shared_thread_func
,
280 /* We can cast between gsize and gpointer safely */
281 g_once_init_leave (&shared_thread_data
, (gsize
) data
);
284 ret
= (SharedThreadData
*) shared_thread_data
;
285 g_atomic_int_inc (&ret
->refcount
);
290 _g_dbus_shared_thread_unref (SharedThreadData
*data
)
292 /* TODO: actually destroy the shared thread here */
294 g_assert (data
!= NULL
);
295 if (g_atomic_int_dec_and_test (&data
->refcount
))
297 g_main_loop_quit (data
->loop
);
298 //g_thread_join (data->thread);
299 g_main_loop_unref (data
->loop
);
300 g_main_context_unref (data
->context
);
305 /* ---------------------------------------------------------------------------------------------------- */
316 volatile gint ref_count
;
318 SharedThreadData
*shared_thread_data
;
320 /* really a boolean, but GLib 2.28 lacks atomic boolean ops */
321 volatile gint stopped
;
323 /* TODO: frozen (e.g. G_DBUS_CONNECTION_FLAGS_DELAY_MESSAGE_PROCESSING) currently
324 * only affects messages received from the other peer (since GDBusServer is the
325 * only user) - we might want it to affect messages sent to the other peer too?
328 GDBusCapabilityFlags capabilities
;
329 GQueue
*received_messages_while_frozen
;
332 GCancellable
*cancellable
;
333 GDBusWorkerMessageReceivedCallback message_received_callback
;
334 GDBusWorkerMessageAboutToBeSentCallback message_about_to_be_sent_callback
;
335 GDBusWorkerDisconnectedCallback disconnected_callback
;
338 /* if not NULL, stream is GSocketConnection */
341 /* used for reading */
344 gsize read_buffer_allocated_size
;
345 gsize read_buffer_cur_size
;
346 gsize read_buffer_bytes_wanted
;
347 GUnixFDList
*read_fd_list
;
348 GSocketControlMessage
**read_ancillary_messages
;
349 gint read_num_ancillary_messages
;
351 /* Whether an async write, flush or close, or none of those, is pending.
352 * Only the worker thread may change its value, and only with the write_lock.
353 * Other threads may read its value when holding the write_lock.
354 * The worker thread may read its value at any time.
356 OutputPending output_pending
;
357 /* used for writing */
359 /* queue of MessageToWriteData, protected by write_lock */
361 /* protected by write_lock */
362 guint64 write_num_messages_written
;
363 /* number of messages we'd written out last time we flushed;
364 * protected by write_lock
366 guint64 write_num_messages_flushed
;
367 /* list of FlushData, protected by write_lock */
368 GList
*write_pending_flushes
;
369 /* list of CloseData, protected by write_lock */
370 GList
*pending_close_attempts
;
371 /* no lock - only used from the worker thread */
372 gboolean close_expected
;
375 static void _g_dbus_worker_unref (GDBusWorker
*worker
);
377 /* ---------------------------------------------------------------------------------------------------- */
383 guint64 number_to_wait_for
;
387 struct _MessageToWriteData
;
388 typedef struct _MessageToWriteData MessageToWriteData
;
390 static void message_to_write_data_free (MessageToWriteData
*data
);
392 static void read_message_print_transport_debug (gssize bytes_read
,
393 GDBusWorker
*worker
);
395 static void write_message_print_transport_debug (gssize bytes_written
,
396 MessageToWriteData
*data
);
403 static void close_data_free (CloseData
*close_data
)
405 g_clear_object (&close_data
->task
);
407 _g_dbus_worker_unref (close_data
->worker
);
408 g_slice_free (CloseData
, close_data
);
411 /* ---------------------------------------------------------------------------------------------------- */
414 _g_dbus_worker_ref (GDBusWorker
*worker
)
416 g_atomic_int_inc (&worker
->ref_count
);
421 _g_dbus_worker_unref (GDBusWorker
*worker
)
423 if (g_atomic_int_dec_and_test (&worker
->ref_count
))
425 g_assert (worker
->write_pending_flushes
== NULL
);
427 _g_dbus_shared_thread_unref (worker
->shared_thread_data
);
429 g_object_unref (worker
->stream
);
431 g_mutex_clear (&worker
->read_lock
);
432 g_object_unref (worker
->cancellable
);
433 if (worker
->read_fd_list
!= NULL
)
434 g_object_unref (worker
->read_fd_list
);
436 g_queue_free_full (worker
->received_messages_while_frozen
, (GDestroyNotify
) g_object_unref
);
437 g_mutex_clear (&worker
->write_lock
);
438 g_queue_free_full (worker
->write_queue
, (GDestroyNotify
) message_to_write_data_free
);
439 g_free (worker
->read_buffer
);
446 _g_dbus_worker_emit_disconnected (GDBusWorker
*worker
,
447 gboolean remote_peer_vanished
,
450 if (!g_atomic_int_get (&worker
->stopped
))
451 worker
->disconnected_callback (worker
, remote_peer_vanished
, error
, worker
->user_data
);
455 _g_dbus_worker_emit_message_received (GDBusWorker
*worker
,
456 GDBusMessage
*message
)
458 if (!g_atomic_int_get (&worker
->stopped
))
459 worker
->message_received_callback (worker
, message
, worker
->user_data
);
462 static GDBusMessage
*
463 _g_dbus_worker_emit_message_about_to_be_sent (GDBusWorker
*worker
,
464 GDBusMessage
*message
)
467 if (!g_atomic_int_get (&worker
->stopped
))
468 ret
= worker
->message_about_to_be_sent_callback (worker
, message
, worker
->user_data
);
474 /* can only be called from private thread with read-lock held - takes ownership of @message */
476 _g_dbus_worker_queue_or_deliver_received_message (GDBusWorker
*worker
,
477 GDBusMessage
*message
)
479 if (worker
->frozen
|| g_queue_get_length (worker
->received_messages_while_frozen
) > 0)
482 g_queue_push_tail (worker
->received_messages_while_frozen
, message
);
486 /* not frozen, nor anything in queue */
487 _g_dbus_worker_emit_message_received (worker
, message
);
488 g_object_unref (message
);
492 /* called in private thread shared by all GDBusConnection instances (without read-lock held) */
494 unfreeze_in_idle_cb (gpointer user_data
)
496 GDBusWorker
*worker
= user_data
;
497 GDBusMessage
*message
;
499 g_mutex_lock (&worker
->read_lock
);
502 while ((message
= g_queue_pop_head (worker
->received_messages_while_frozen
)) != NULL
)
504 _g_dbus_worker_emit_message_received (worker
, message
);
505 g_object_unref (message
);
507 worker
->frozen
= FALSE
;
511 g_assert (g_queue_get_length (worker
->received_messages_while_frozen
) == 0);
513 g_mutex_unlock (&worker
->read_lock
);
517 /* can be called from any thread */
519 _g_dbus_worker_unfreeze (GDBusWorker
*worker
)
521 GSource
*idle_source
;
522 idle_source
= g_idle_source_new ();
523 g_source_set_priority (idle_source
, G_PRIORITY_DEFAULT
);
524 g_source_set_callback (idle_source
,
526 _g_dbus_worker_ref (worker
),
527 (GDestroyNotify
) _g_dbus_worker_unref
);
528 g_source_set_name (idle_source
, "[gio] unfreeze_in_idle_cb");
529 g_source_attach (idle_source
, worker
->shared_thread_data
->context
);
530 g_source_unref (idle_source
);
533 /* ---------------------------------------------------------------------------------------------------- */
535 static void _g_dbus_worker_do_read_unlocked (GDBusWorker
*worker
);
537 /* called in private thread shared by all GDBusConnection instances (without read-lock held) */
539 _g_dbus_worker_do_read_cb (GInputStream
*input_stream
,
543 GDBusWorker
*worker
= user_data
;
547 g_mutex_lock (&worker
->read_lock
);
549 /* If already stopped, don't even process the reply */
550 if (g_atomic_int_get (&worker
->stopped
))
554 if (worker
->socket
== NULL
)
555 bytes_read
= g_input_stream_read_finish (g_io_stream_get_input_stream (worker
->stream
),
559 bytes_read
= _g_socket_read_with_control_messages_finish (worker
->socket
,
562 if (worker
->read_num_ancillary_messages
> 0)
565 for (n
= 0; n
< worker
->read_num_ancillary_messages
; n
++)
567 GSocketControlMessage
*control_message
= G_SOCKET_CONTROL_MESSAGE (worker
->read_ancillary_messages
[n
]);
573 else if (G_IS_UNIX_FD_MESSAGE (control_message
))
575 GUnixFDMessage
*fd_message
;
579 fd_message
= G_UNIX_FD_MESSAGE (control_message
);
580 fds
= g_unix_fd_message_steal_fds (fd_message
, &num_fds
);
581 if (worker
->read_fd_list
== NULL
)
583 worker
->read_fd_list
= g_unix_fd_list_new_from_array (fds
, num_fds
);
588 for (n
= 0; n
< num_fds
; n
++)
590 /* TODO: really want a append_steal() */
591 g_unix_fd_list_append (worker
->read_fd_list
, fds
[n
], NULL
);
592 (void) g_close (fds
[n
], NULL
);
597 else if (G_IS_UNIX_CREDENTIALS_MESSAGE (control_message
))
609 "Unexpected ancillary message of type %s received from peer",
610 g_type_name (G_TYPE_FROM_INSTANCE (control_message
)));
611 _g_dbus_worker_emit_disconnected (worker
, TRUE
, error
);
612 g_error_free (error
);
613 g_object_unref (control_message
);
615 while (n
< worker
->read_num_ancillary_messages
)
616 g_object_unref (worker
->read_ancillary_messages
[n
++]);
617 g_free (worker
->read_ancillary_messages
);
621 g_object_unref (control_message
);
623 g_free (worker
->read_ancillary_messages
);
626 if (bytes_read
== -1)
628 if (G_UNLIKELY (_g_dbus_debug_transport ()))
630 _g_dbus_debug_print_lock ();
631 g_print ("========================================================================\n"
632 "GDBus-debug:Transport:\n"
633 " ---- READ ERROR on stream of type %s:\n"
635 g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_input_stream (worker
->stream
))),
636 g_quark_to_string (error
->domain
), error
->code
,
638 _g_dbus_debug_print_unlock ();
641 /* Every async read that uses this callback uses worker->cancellable
642 * as its GCancellable. worker->cancellable gets cancelled if and only
643 * if the GDBusConnection tells us to close (either via
644 * _g_dbus_worker_stop, which is called on last-unref, or directly),
645 * so a cancelled read must mean our connection was closed locally.
647 * If we're closing, other errors are possible - notably,
648 * G_IO_ERROR_CLOSED can be seen if we close the stream with an async
649 * read in-flight. It seems sensible to treat all read errors during
650 * closing as an expected thing that doesn't trip exit-on-close.
652 * Because close_expected can't be set until we get into the worker
653 * thread, but the cancellable is signalled sooner (from another
654 * thread), we do still need to check the error.
656 if (worker
->close_expected
||
657 g_error_matches (error
, G_IO_ERROR
, G_IO_ERROR_CANCELLED
))
658 _g_dbus_worker_emit_disconnected (worker
, FALSE
, NULL
);
660 _g_dbus_worker_emit_disconnected (worker
, TRUE
, error
);
662 g_error_free (error
);
667 g_debug ("read %d bytes (is_closed=%d blocking=%d condition=0x%02x) stream %p, %p",
669 g_socket_is_closed (g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker
->stream
))),
670 g_socket_get_blocking (g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker
->stream
))),
671 g_socket_condition_check (g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker
->stream
)),
672 G_IO_IN
| G_IO_OUT
| G_IO_HUP
),
677 /* TODO: hmm, hmm... */
683 "Underlying GIOStream returned 0 bytes on an async read");
684 _g_dbus_worker_emit_disconnected (worker
, TRUE
, error
);
685 g_error_free (error
);
689 read_message_print_transport_debug (bytes_read
, worker
);
691 worker
->read_buffer_cur_size
+= bytes_read
;
692 if (worker
->read_buffer_bytes_wanted
== worker
->read_buffer_cur_size
)
694 /* OK, got what we asked for! */
695 if (worker
->read_buffer_bytes_wanted
== 16)
698 /* OK, got the header - determine how many more bytes are needed */
700 message_len
= g_dbus_message_bytes_needed ((guchar
*) worker
->read_buffer
,
703 if (message_len
== -1)
705 g_warning ("_g_dbus_worker_do_read_cb: error determining bytes needed: %s", error
->message
);
706 _g_dbus_worker_emit_disconnected (worker
, FALSE
, error
);
707 g_error_free (error
);
711 worker
->read_buffer_bytes_wanted
= message_len
;
712 _g_dbus_worker_do_read_unlocked (worker
);
716 GDBusMessage
*message
;
719 /* TODO: use connection->priv->auth to decode the message */
721 message
= g_dbus_message_new_from_blob ((guchar
*) worker
->read_buffer
,
722 worker
->read_buffer_cur_size
,
723 worker
->capabilities
,
728 s
= _g_dbus_hexdump (worker
->read_buffer
, worker
->read_buffer_cur_size
, 2);
729 g_warning ("Error decoding D-Bus message of %" G_GSIZE_FORMAT
" bytes\n"
731 "The payload is as follows:\n"
733 worker
->read_buffer_cur_size
,
737 _g_dbus_worker_emit_disconnected (worker
, FALSE
, error
);
738 g_error_free (error
);
743 if (worker
->read_fd_list
!= NULL
)
745 g_dbus_message_set_unix_fd_list (message
, worker
->read_fd_list
);
746 g_object_unref (worker
->read_fd_list
);
747 worker
->read_fd_list
= NULL
;
751 if (G_UNLIKELY (_g_dbus_debug_message ()))
754 _g_dbus_debug_print_lock ();
755 g_print ("========================================================================\n"
756 "GDBus-debug:Message:\n"
757 " <<<< RECEIVED D-Bus message (%" G_GSIZE_FORMAT
" bytes)\n",
758 worker
->read_buffer_cur_size
);
759 s
= g_dbus_message_print (message
, 2);
762 if (G_UNLIKELY (_g_dbus_debug_payload ()))
764 s
= _g_dbus_hexdump (worker
->read_buffer
, worker
->read_buffer_cur_size
, 2);
768 _g_dbus_debug_print_unlock ();
771 /* yay, got a message, go deliver it */
772 _g_dbus_worker_queue_or_deliver_received_message (worker
, message
);
774 /* start reading another message! */
775 worker
->read_buffer_bytes_wanted
= 0;
776 worker
->read_buffer_cur_size
= 0;
777 _g_dbus_worker_do_read_unlocked (worker
);
782 /* didn't get all the bytes we requested - so repeat the request... */
783 _g_dbus_worker_do_read_unlocked (worker
);
787 g_mutex_unlock (&worker
->read_lock
);
789 /* gives up the reference acquired when calling g_input_stream_read_async() */
790 _g_dbus_worker_unref (worker
);
792 /* check if there is any pending close */
793 schedule_pending_close (worker
);
796 /* called in private thread shared by all GDBusConnection instances (with read-lock held) */
798 _g_dbus_worker_do_read_unlocked (GDBusWorker
*worker
)
800 /* Note that we do need to keep trying to read even if close_expected is
801 * true, because only failing a read causes us to signal 'closed'.
804 /* if bytes_wanted is zero, it means start reading a message */
805 if (worker
->read_buffer_bytes_wanted
== 0)
807 worker
->read_buffer_cur_size
= 0;
808 worker
->read_buffer_bytes_wanted
= 16;
811 /* ensure we have a (big enough) buffer */
812 if (worker
->read_buffer
== NULL
|| worker
->read_buffer_bytes_wanted
> worker
->read_buffer_allocated_size
)
814 /* TODO: 4096 is randomly chosen; might want a better chosen default minimum */
815 worker
->read_buffer_allocated_size
= MAX (worker
->read_buffer_bytes_wanted
, 4096);
816 worker
->read_buffer
= g_realloc (worker
->read_buffer
, worker
->read_buffer_allocated_size
);
819 if (worker
->socket
== NULL
)
820 g_input_stream_read_async (g_io_stream_get_input_stream (worker
->stream
),
821 worker
->read_buffer
+ worker
->read_buffer_cur_size
,
822 worker
->read_buffer_bytes_wanted
- worker
->read_buffer_cur_size
,
825 (GAsyncReadyCallback
) _g_dbus_worker_do_read_cb
,
826 _g_dbus_worker_ref (worker
));
829 worker
->read_ancillary_messages
= NULL
;
830 worker
->read_num_ancillary_messages
= 0;
831 _g_socket_read_with_control_messages (worker
->socket
,
832 worker
->read_buffer
+ worker
->read_buffer_cur_size
,
833 worker
->read_buffer_bytes_wanted
- worker
->read_buffer_cur_size
,
834 &worker
->read_ancillary_messages
,
835 &worker
->read_num_ancillary_messages
,
838 (GAsyncReadyCallback
) _g_dbus_worker_do_read_cb
,
839 _g_dbus_worker_ref (worker
));
843 /* called in private thread shared by all GDBusConnection instances (without read-lock held) */
845 _g_dbus_worker_do_initial_read (gpointer data
)
847 GDBusWorker
*worker
= data
;
848 g_mutex_lock (&worker
->read_lock
);
849 _g_dbus_worker_do_read_unlocked (worker
);
850 g_mutex_unlock (&worker
->read_lock
);
854 /* ---------------------------------------------------------------------------------------------------- */
856 struct _MessageToWriteData
859 GDBusMessage
*message
;
868 message_to_write_data_free (MessageToWriteData
*data
)
870 _g_dbus_worker_unref (data
->worker
);
872 g_object_unref (data
->message
);
874 g_slice_free (MessageToWriteData
, data
);
877 /* ---------------------------------------------------------------------------------------------------- */
879 static void write_message_continue_writing (MessageToWriteData
*data
);
881 /* called in private thread shared by all GDBusConnection instances
883 * write-lock is not held on entry
884 * output_pending is PENDING_WRITE on entry
887 write_message_async_cb (GObject
*source_object
,
891 MessageToWriteData
*data
= user_data
;
893 gssize bytes_written
;
896 /* Note: we can't access data->task after calling g_task_return_* () because the
897 * callback can free @data and we're not completing in idle. So use a copy of the pointer.
902 bytes_written
= g_output_stream_write_finish (G_OUTPUT_STREAM (source_object
),
905 if (bytes_written
== -1)
907 g_task_return_error (task
, error
);
908 g_object_unref (task
);
911 g_assert (bytes_written
> 0); /* zero is never returned */
913 write_message_print_transport_debug (bytes_written
, data
);
915 data
->total_written
+= bytes_written
;
916 g_assert (data
->total_written
<= data
->blob_size
);
917 if (data
->total_written
== data
->blob_size
)
919 g_task_return_boolean (task
, TRUE
);
920 g_object_unref (task
);
924 write_message_continue_writing (data
);
930 /* called in private thread shared by all GDBusConnection instances
932 * write-lock is not held on entry
933 * output_pending is PENDING_WRITE on entry
937 on_socket_ready (GSocket
*socket
,
938 GIOCondition condition
,
941 MessageToWriteData
*data
= user_data
;
942 write_message_continue_writing (data
);
943 return FALSE
; /* remove source */
947 /* called in private thread shared by all GDBusConnection instances
949 * write-lock is not held on entry
950 * output_pending is PENDING_WRITE on entry
953 write_message_continue_writing (MessageToWriteData
*data
)
955 GOutputStream
*ostream
;
958 GUnixFDList
*fd_list
;
962 /* Note: we can't access data->task after calling g_task_return_* () because the
963 * callback can free @data and we're not completing in idle. So use a copy of the pointer.
968 ostream
= g_io_stream_get_output_stream (data
->worker
->stream
);
970 fd_list
= g_dbus_message_get_unix_fd_list (data
->message
);
973 g_assert (!g_output_stream_has_pending (ostream
));
974 g_assert_cmpint (data
->total_written
, <, data
->blob_size
);
980 else if (G_IS_SOCKET_OUTPUT_STREAM (ostream
) && data
->total_written
== 0)
982 GOutputVector vector
;
983 GSocketControlMessage
*control_message
;
984 gssize bytes_written
;
987 vector
.buffer
= data
->blob
;
988 vector
.size
= data
->blob_size
;
990 control_message
= NULL
;
991 if (fd_list
!= NULL
&& g_unix_fd_list_get_length (fd_list
) > 0)
993 if (!(data
->worker
->capabilities
& G_DBUS_CAPABILITY_FLAGS_UNIX_FD_PASSING
))
995 g_task_return_new_error (task
,
998 "Tried sending a file descriptor but remote peer does not support this capability");
999 g_object_unref (task
);
1002 control_message
= g_unix_fd_message_new_with_fd_list (fd_list
);
1006 bytes_written
= g_socket_send_message (data
->worker
->socket
,
1010 control_message
!= NULL
? &control_message
: NULL
,
1011 control_message
!= NULL
? 1 : 0,
1013 data
->worker
->cancellable
,
1015 if (control_message
!= NULL
)
1016 g_object_unref (control_message
);
1018 if (bytes_written
== -1)
1020 /* Handle WOULD_BLOCK by waiting until there's room in the buffer */
1021 if (g_error_matches (error
, G_IO_ERROR
, G_IO_ERROR_WOULD_BLOCK
))
1024 source
= g_socket_create_source (data
->worker
->socket
,
1025 G_IO_OUT
| G_IO_HUP
| G_IO_ERR
,
1026 data
->worker
->cancellable
);
1027 g_source_set_callback (source
,
1028 (GSourceFunc
) on_socket_ready
,
1030 NULL
); /* GDestroyNotify */
1031 g_source_attach (source
, g_main_context_get_thread_default ());
1032 g_source_unref (source
);
1033 g_error_free (error
);
1036 g_task_return_error (task
, error
);
1037 g_object_unref (task
);
1040 g_assert (bytes_written
> 0); /* zero is never returned */
1042 write_message_print_transport_debug (bytes_written
, data
);
1044 data
->total_written
+= bytes_written
;
1045 g_assert (data
->total_written
<= data
->blob_size
);
1046 if (data
->total_written
== data
->blob_size
)
1048 g_task_return_boolean (task
, TRUE
);
1049 g_object_unref (task
);
1053 write_message_continue_writing (data
);
1059 if (fd_list
!= NULL
)
1061 g_task_return_new_error (task
,
1064 "Tried sending a file descriptor on unsupported stream of type %s",
1065 g_type_name (G_TYPE_FROM_INSTANCE (ostream
)));
1066 g_object_unref (task
);
1071 g_output_stream_write_async (ostream
,
1072 (const gchar
*) data
->blob
+ data
->total_written
,
1073 data
->blob_size
- data
->total_written
,
1075 data
->worker
->cancellable
,
1076 write_message_async_cb
,
1085 /* called in private thread shared by all GDBusConnection instances
1087 * write-lock is not held on entry
1088 * output_pending is PENDING_WRITE on entry
1091 write_message_async (GDBusWorker
*worker
,
1092 MessageToWriteData
*data
,
1093 GAsyncReadyCallback callback
,
1096 data
->task
= g_task_new (NULL
, NULL
, callback
, user_data
);
1097 g_task_set_source_tag (data
->task
, write_message_async
);
1098 data
->total_written
= 0;
1099 write_message_continue_writing (data
);
1102 /* called in private thread shared by all GDBusConnection instances (with write-lock held) */
1104 write_message_finish (GAsyncResult
*res
,
1107 g_return_val_if_fail (g_task_is_valid (res
, NULL
), FALSE
);
1109 return g_task_propagate_boolean (G_TASK (res
), error
);
1111 /* ---------------------------------------------------------------------------------------------------- */
1113 static void continue_writing (GDBusWorker
*worker
);
1117 GDBusWorker
*worker
;
1122 flush_data_list_complete (const GList
*flushers
,
1123 const GError
*error
)
1127 for (l
= flushers
; l
!= NULL
; l
= l
->next
)
1129 FlushData
*f
= l
->data
;
1131 f
->error
= error
!= NULL
? g_error_copy (error
) : NULL
;
1133 g_mutex_lock (&f
->mutex
);
1134 g_cond_signal (&f
->cond
);
1135 g_mutex_unlock (&f
->mutex
);
1139 /* called in private thread shared by all GDBusConnection instances
1141 * write-lock is not held on entry
1142 * output_pending is PENDING_FLUSH on entry
1145 ostream_flush_cb (GObject
*source_object
,
1149 FlushAsyncData
*data
= user_data
;
1153 g_output_stream_flush_finish (G_OUTPUT_STREAM (source_object
),
1159 if (G_UNLIKELY (_g_dbus_debug_transport ()))
1161 _g_dbus_debug_print_lock ();
1162 g_print ("========================================================================\n"
1163 "GDBus-debug:Transport:\n"
1164 " ---- FLUSHED stream of type %s\n",
1165 g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_output_stream (data
->worker
->stream
))));
1166 _g_dbus_debug_print_unlock ();
1170 g_assert (data
->flushers
!= NULL
);
1171 flush_data_list_complete (data
->flushers
, error
);
1172 g_list_free (data
->flushers
);
1175 g_error_free (error
);
1177 /* Make sure we tell folks that we don't have additional
1179 g_mutex_lock (&data
->worker
->write_lock
);
1180 data
->worker
->write_num_messages_flushed
= data
->worker
->write_num_messages_written
;
1181 g_assert (data
->worker
->output_pending
== PENDING_FLUSH
);
1182 data
->worker
->output_pending
= PENDING_NONE
;
1183 g_mutex_unlock (&data
->worker
->write_lock
);
1185 /* OK, cool, finally kick off the next write */
1186 continue_writing (data
->worker
);
1188 _g_dbus_worker_unref (data
->worker
);
1192 /* called in private thread shared by all GDBusConnection instances
1194 * write-lock is not held on entry
1195 * output_pending is PENDING_FLUSH on entry
1198 start_flush (FlushAsyncData
*data
)
1200 g_output_stream_flush_async (g_io_stream_get_output_stream (data
->worker
->stream
),
1202 data
->worker
->cancellable
,
1207 /* called in private thread shared by all GDBusConnection instances
1209 * write-lock is held on entry
1210 * output_pending is PENDING_NONE on entry
1213 message_written_unlocked (GDBusWorker
*worker
,
1214 MessageToWriteData
*message_data
)
1216 if (G_UNLIKELY (_g_dbus_debug_message ()))
1219 _g_dbus_debug_print_lock ();
1220 g_print ("========================================================================\n"
1221 "GDBus-debug:Message:\n"
1222 " >>>> SENT D-Bus message (%" G_GSIZE_FORMAT
" bytes)\n",
1223 message_data
->blob_size
);
1224 s
= g_dbus_message_print (message_data
->message
, 2);
1227 if (G_UNLIKELY (_g_dbus_debug_payload ()))
1229 s
= _g_dbus_hexdump (message_data
->blob
, message_data
->blob_size
, 2);
1230 g_print ("%s\n", s
);
1233 _g_dbus_debug_print_unlock ();
1236 worker
->write_num_messages_written
+= 1;
1239 /* called in private thread shared by all GDBusConnection instances
1241 * write-lock is held on entry
1242 * output_pending is PENDING_NONE on entry
1244 * Returns: non-%NULL, setting @output_pending, if we need to flush now
1246 static FlushAsyncData
*
1247 prepare_flush_unlocked (GDBusWorker
*worker
)
1254 for (l
= worker
->write_pending_flushes
; l
!= NULL
; l
= ll
)
1256 FlushData
*f
= l
->data
;
1259 if (f
->number_to_wait_for
== worker
->write_num_messages_written
)
1261 flushers
= g_list_append (flushers
, f
);
1262 worker
->write_pending_flushes
= g_list_delete_link (worker
->write_pending_flushes
, l
);
1265 if (flushers
!= NULL
)
1267 g_assert (worker
->output_pending
== PENDING_NONE
);
1268 worker
->output_pending
= PENDING_FLUSH
;
1271 if (flushers
!= NULL
)
1273 FlushAsyncData
*data
;
1275 data
= g_new0 (FlushAsyncData
, 1);
1276 data
->worker
= _g_dbus_worker_ref (worker
);
1277 data
->flushers
= flushers
;
1284 /* called in private thread shared by all GDBusConnection instances
1286 * write-lock is not held on entry
1287 * output_pending is PENDING_WRITE on entry
1290 write_message_cb (GObject
*source_object
,
1294 MessageToWriteData
*data
= user_data
;
1297 g_mutex_lock (&data
->worker
->write_lock
);
1298 g_assert (data
->worker
->output_pending
== PENDING_WRITE
);
1299 data
->worker
->output_pending
= PENDING_NONE
;
1302 if (!write_message_finish (res
, &error
))
1304 g_mutex_unlock (&data
->worker
->write_lock
);
1307 _g_dbus_worker_emit_disconnected (data
->worker
, TRUE
, error
);
1308 g_error_free (error
);
1310 g_mutex_lock (&data
->worker
->write_lock
);
1313 message_written_unlocked (data
->worker
, data
);
1315 g_mutex_unlock (&data
->worker
->write_lock
);
1317 continue_writing (data
->worker
);
1319 message_to_write_data_free (data
);
1322 /* called in private thread shared by all GDBusConnection instances
1324 * write-lock is not held on entry
1325 * output_pending is PENDING_CLOSE on entry
1328 iostream_close_cb (GObject
*source_object
,
1332 GDBusWorker
*worker
= user_data
;
1333 GError
*error
= NULL
;
1334 GList
*pending_close_attempts
, *pending_flush_attempts
;
1337 g_io_stream_close_finish (worker
->stream
, res
, &error
);
1339 g_mutex_lock (&worker
->write_lock
);
1341 pending_close_attempts
= worker
->pending_close_attempts
;
1342 worker
->pending_close_attempts
= NULL
;
1344 pending_flush_attempts
= worker
->write_pending_flushes
;
1345 worker
->write_pending_flushes
= NULL
;
1347 send_queue
= worker
->write_queue
;
1348 worker
->write_queue
= g_queue_new ();
1350 g_assert (worker
->output_pending
== PENDING_CLOSE
);
1351 worker
->output_pending
= PENDING_NONE
;
1353 g_mutex_unlock (&worker
->write_lock
);
1355 while (pending_close_attempts
!= NULL
)
1357 CloseData
*close_data
= pending_close_attempts
->data
;
1359 pending_close_attempts
= g_list_delete_link (pending_close_attempts
,
1360 pending_close_attempts
);
1362 if (close_data
->task
!= NULL
)
1365 g_task_return_error (close_data
->task
, g_error_copy (error
));
1367 g_task_return_boolean (close_data
->task
, TRUE
);
1370 close_data_free (close_data
);
1373 g_clear_error (&error
);
1375 /* all messages queued for sending are discarded */
1376 g_queue_free_full (send_queue
, (GDestroyNotify
) message_to_write_data_free
);
1377 /* all queued flushes fail */
1378 error
= g_error_new (G_IO_ERROR
, G_IO_ERROR_CANCELLED
,
1379 _("Operation was cancelled"));
1380 flush_data_list_complete (pending_flush_attempts
, error
);
1381 g_list_free (pending_flush_attempts
);
1382 g_clear_error (&error
);
1384 _g_dbus_worker_unref (worker
);
1387 /* called in private thread shared by all GDBusConnection instances
1389 * write-lock is not held on entry
1390 * output_pending must be PENDING_NONE on entry
1393 continue_writing (GDBusWorker
*worker
)
1395 MessageToWriteData
*data
;
1396 FlushAsyncData
*flush_async_data
;
1399 /* we mustn't try to write two things at once */
1400 g_assert (worker
->output_pending
== PENDING_NONE
);
1402 g_mutex_lock (&worker
->write_lock
);
1405 flush_async_data
= NULL
;
1407 /* if we want to close the connection, that takes precedence */
1408 if (worker
->pending_close_attempts
!= NULL
)
1410 GInputStream
*input
= g_io_stream_get_input_stream (worker
->stream
);
1412 if (!g_input_stream_has_pending (input
))
1414 worker
->close_expected
= TRUE
;
1415 worker
->output_pending
= PENDING_CLOSE
;
1417 g_io_stream_close_async (worker
->stream
, G_PRIORITY_DEFAULT
,
1418 NULL
, iostream_close_cb
,
1419 _g_dbus_worker_ref (worker
));
1424 flush_async_data
= prepare_flush_unlocked (worker
);
1426 if (flush_async_data
== NULL
)
1428 data
= g_queue_pop_head (worker
->write_queue
);
1431 worker
->output_pending
= PENDING_WRITE
;
1435 g_mutex_unlock (&worker
->write_lock
);
1437 /* Note that write_lock is only used for protecting the @write_queue
1438 * and @output_pending fields of the GDBusWorker struct ... which we
1439 * need to modify from arbitrary threads in _g_dbus_worker_send_message().
1441 * Therefore, it's fine to drop it here when calling back into user
1442 * code and then writing the message out onto the GIOStream since this
1443 * function only runs on the worker thread.
1446 if (flush_async_data
!= NULL
)
1448 start_flush (flush_async_data
);
1449 g_assert (data
== NULL
);
1451 else if (data
!= NULL
)
1453 GDBusMessage
*old_message
;
1455 gsize new_blob_size
;
1458 old_message
= data
->message
;
1459 data
->message
= _g_dbus_worker_emit_message_about_to_be_sent (worker
, data
->message
);
1460 if (data
->message
== old_message
)
1462 /* filters had no effect - do nothing */
1464 else if (data
->message
== NULL
)
1466 /* filters dropped message */
1467 g_mutex_lock (&worker
->write_lock
);
1468 worker
->output_pending
= PENDING_NONE
;
1469 g_mutex_unlock (&worker
->write_lock
);
1470 message_to_write_data_free (data
);
1475 /* filters altered the message -> reencode */
1477 new_blob
= g_dbus_message_to_blob (data
->message
,
1479 worker
->capabilities
,
1481 if (new_blob
== NULL
)
1483 /* if filter make the GDBusMessage unencodeable, just complain on stderr and send
1484 * the old message instead
1486 g_warning ("Error encoding GDBusMessage with serial %d altered by filter function: %s",
1487 g_dbus_message_get_serial (data
->message
),
1489 g_error_free (error
);
1493 g_free (data
->blob
);
1494 data
->blob
= (gchar
*) new_blob
;
1495 data
->blob_size
= new_blob_size
;
1499 write_message_async (worker
,
1506 /* called in private thread shared by all GDBusConnection instances
1508 * write-lock is not held on entry
1509 * output_pending may be anything
1512 continue_writing_in_idle_cb (gpointer user_data
)
1514 GDBusWorker
*worker
= user_data
;
1516 /* Because this is the worker thread, we can read this struct member
1517 * without holding the lock: no other thread ever modifies it.
1519 if (worker
->output_pending
== PENDING_NONE
)
1520 continue_writing (worker
);
1526 * @write_data: (transfer full) (nullable):
1527 * @flush_data: (transfer full) (nullable):
1528 * @close_data: (transfer full) (nullable):
1530 * Can be called from any thread
1532 * write_lock is held on entry
1533 * output_pending may be anything
1536 schedule_writing_unlocked (GDBusWorker
*worker
,
1537 MessageToWriteData
*write_data
,
1538 FlushData
*flush_data
,
1539 CloseData
*close_data
)
1541 if (write_data
!= NULL
)
1542 g_queue_push_tail (worker
->write_queue
, write_data
);
1544 if (flush_data
!= NULL
)
1545 worker
->write_pending_flushes
= g_list_prepend (worker
->write_pending_flushes
, flush_data
);
1547 if (close_data
!= NULL
)
1548 worker
->pending_close_attempts
= g_list_prepend (worker
->pending_close_attempts
,
1551 /* If we had output pending, the next bit of output will happen
1552 * automatically when it finishes, so we only need to do this
1553 * if nothing was pending.
1555 * The idle callback will re-check that output_pending is still
1556 * PENDING_NONE, to guard against output starting before the idle.
1558 if (worker
->output_pending
== PENDING_NONE
)
1560 GSource
*idle_source
;
1561 idle_source
= g_idle_source_new ();
1562 g_source_set_priority (idle_source
, G_PRIORITY_DEFAULT
);
1563 g_source_set_callback (idle_source
,
1564 continue_writing_in_idle_cb
,
1565 _g_dbus_worker_ref (worker
),
1566 (GDestroyNotify
) _g_dbus_worker_unref
);
1567 g_source_set_name (idle_source
, "[gio] continue_writing_in_idle_cb");
1568 g_source_attach (idle_source
, worker
->shared_thread_data
->context
);
1569 g_source_unref (idle_source
);
1574 schedule_pending_close (GDBusWorker
*worker
)
1576 g_mutex_lock (&worker
->write_lock
);
1577 if (worker
->pending_close_attempts
)
1578 schedule_writing_unlocked (worker
, NULL
, NULL
, NULL
);
1579 g_mutex_unlock (&worker
->write_lock
);
1582 /* ---------------------------------------------------------------------------------------------------- */
1584 /* can be called from any thread - steals blob
1586 * write_lock is not held on entry
1587 * output_pending may be anything
1590 _g_dbus_worker_send_message (GDBusWorker
*worker
,
1591 GDBusMessage
*message
,
1595 MessageToWriteData
*data
;
1597 g_return_if_fail (G_IS_DBUS_MESSAGE (message
));
1598 g_return_if_fail (blob
!= NULL
);
1599 g_return_if_fail (blob_len
> 16);
1601 data
= g_slice_new0 (MessageToWriteData
);
1602 data
->worker
= _g_dbus_worker_ref (worker
);
1603 data
->message
= g_object_ref (message
);
1604 data
->blob
= blob
; /* steal! */
1605 data
->blob_size
= blob_len
;
1607 g_mutex_lock (&worker
->write_lock
);
1608 schedule_writing_unlocked (worker
, data
, NULL
, NULL
);
1609 g_mutex_unlock (&worker
->write_lock
);
1612 /* ---------------------------------------------------------------------------------------------------- */
1615 _g_dbus_worker_new (GIOStream
*stream
,
1616 GDBusCapabilityFlags capabilities
,
1617 gboolean initially_frozen
,
1618 GDBusWorkerMessageReceivedCallback message_received_callback
,
1619 GDBusWorkerMessageAboutToBeSentCallback message_about_to_be_sent_callback
,
1620 GDBusWorkerDisconnectedCallback disconnected_callback
,
1623 GDBusWorker
*worker
;
1624 GSource
*idle_source
;
1626 g_return_val_if_fail (G_IS_IO_STREAM (stream
), NULL
);
1627 g_return_val_if_fail (message_received_callback
!= NULL
, NULL
);
1628 g_return_val_if_fail (message_about_to_be_sent_callback
!= NULL
, NULL
);
1629 g_return_val_if_fail (disconnected_callback
!= NULL
, NULL
);
1631 worker
= g_new0 (GDBusWorker
, 1);
1632 worker
->ref_count
= 1;
1634 g_mutex_init (&worker
->read_lock
);
1635 worker
->message_received_callback
= message_received_callback
;
1636 worker
->message_about_to_be_sent_callback
= message_about_to_be_sent_callback
;
1637 worker
->disconnected_callback
= disconnected_callback
;
1638 worker
->user_data
= user_data
;
1639 worker
->stream
= g_object_ref (stream
);
1640 worker
->capabilities
= capabilities
;
1641 worker
->cancellable
= g_cancellable_new ();
1642 worker
->output_pending
= PENDING_NONE
;
1644 worker
->frozen
= initially_frozen
;
1645 worker
->received_messages_while_frozen
= g_queue_new ();
1647 g_mutex_init (&worker
->write_lock
);
1648 worker
->write_queue
= g_queue_new ();
1650 if (G_IS_SOCKET_CONNECTION (worker
->stream
))
1651 worker
->socket
= g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker
->stream
));
1653 worker
->shared_thread_data
= _g_dbus_shared_thread_ref ();
1656 idle_source
= g_idle_source_new ();
1657 g_source_set_priority (idle_source
, G_PRIORITY_DEFAULT
);
1658 g_source_set_callback (idle_source
,
1659 _g_dbus_worker_do_initial_read
,
1660 _g_dbus_worker_ref (worker
),
1661 (GDestroyNotify
) _g_dbus_worker_unref
);
1662 g_source_set_name (idle_source
, "[gio] _g_dbus_worker_do_initial_read");
1663 g_source_attach (idle_source
, worker
->shared_thread_data
->context
);
1664 g_source_unref (idle_source
);
1669 /* ---------------------------------------------------------------------------------------------------- */
1671 /* can be called from any thread
1673 * write_lock is not held on entry
1674 * output_pending may be anything
1677 _g_dbus_worker_close (GDBusWorker
*worker
,
1680 CloseData
*close_data
;
1682 close_data
= g_slice_new0 (CloseData
);
1683 close_data
->worker
= _g_dbus_worker_ref (worker
);
1684 close_data
->task
= (task
== NULL
? NULL
: g_object_ref (task
));
1686 /* Don't set worker->close_expected here - we're in the wrong thread.
1687 * It'll be set before the actual close happens.
1689 g_cancellable_cancel (worker
->cancellable
);
1690 g_mutex_lock (&worker
->write_lock
);
1691 schedule_writing_unlocked (worker
, NULL
, NULL
, close_data
);
1692 g_mutex_unlock (&worker
->write_lock
);
1695 /* This can be called from any thread - frees worker. Note that
1696 * callbacks might still happen if called from another thread than the
1697 * worker - use your own synchronization primitive in the callbacks.
1699 * write_lock is not held on entry
1700 * output_pending may be anything
1703 _g_dbus_worker_stop (GDBusWorker
*worker
)
1705 g_atomic_int_set (&worker
->stopped
, TRUE
);
1707 /* Cancel any pending operations and schedule a close of the underlying I/O
1708 * stream in the worker thread
1710 _g_dbus_worker_close (worker
, NULL
);
1712 /* _g_dbus_worker_close holds a ref until after an idle in the worker
1713 * thread has run, so we no longer need to unref in an idle like in
1716 _g_dbus_worker_unref (worker
);
1719 /* ---------------------------------------------------------------------------------------------------- */
1721 /* can be called from any thread (except the worker thread) - blocks
1722 * calling thread until all queued outgoing messages are written and
1723 * the transport has been flushed
1725 * write_lock is not held on entry
1726 * output_pending may be anything
1729 _g_dbus_worker_flush_sync (GDBusWorker
*worker
,
1730 GCancellable
*cancellable
,
1735 guint64 pending_writes
;
1740 g_mutex_lock (&worker
->write_lock
);
1742 /* if the queue is empty, no write is in-flight and we haven't written
1743 * anything since the last flush, then there's nothing to wait for
1745 pending_writes
= g_queue_get_length (worker
->write_queue
);
1747 /* if a write is in-flight, we shouldn't be satisfied until the first
1748 * flush operation that follows it
1750 if (worker
->output_pending
== PENDING_WRITE
)
1751 pending_writes
+= 1;
1753 if (pending_writes
> 0 ||
1754 worker
->write_num_messages_written
!= worker
->write_num_messages_flushed
)
1756 data
= g_new0 (FlushData
, 1);
1757 g_mutex_init (&data
->mutex
);
1758 g_cond_init (&data
->cond
);
1759 data
->number_to_wait_for
= worker
->write_num_messages_written
+ pending_writes
;
1760 g_mutex_lock (&data
->mutex
);
1762 schedule_writing_unlocked (worker
, NULL
, data
, NULL
);
1764 g_mutex_unlock (&worker
->write_lock
);
1768 g_cond_wait (&data
->cond
, &data
->mutex
);
1769 g_mutex_unlock (&data
->mutex
);
1771 /* note:the element is removed from worker->write_pending_flushes in flush_cb() above */
1772 g_cond_clear (&data
->cond
);
1773 g_mutex_clear (&data
->mutex
);
1774 if (data
->error
!= NULL
)
1777 g_propagate_error (error
, data
->error
);
1785 /* ---------------------------------------------------------------------------------------------------- */
1787 #define G_DBUS_DEBUG_AUTHENTICATION (1<<0)
1788 #define G_DBUS_DEBUG_TRANSPORT (1<<1)
1789 #define G_DBUS_DEBUG_MESSAGE (1<<2)
1790 #define G_DBUS_DEBUG_PAYLOAD (1<<3)
1791 #define G_DBUS_DEBUG_CALL (1<<4)
1792 #define G_DBUS_DEBUG_SIGNAL (1<<5)
1793 #define G_DBUS_DEBUG_INCOMING (1<<6)
1794 #define G_DBUS_DEBUG_RETURN (1<<7)
1795 #define G_DBUS_DEBUG_EMISSION (1<<8)
1796 #define G_DBUS_DEBUG_ADDRESS (1<<9)
1798 static gint _gdbus_debug_flags
= 0;
1801 _g_dbus_debug_authentication (void)
1803 _g_dbus_initialize ();
1804 return (_gdbus_debug_flags
& G_DBUS_DEBUG_AUTHENTICATION
) != 0;
1808 _g_dbus_debug_transport (void)
1810 _g_dbus_initialize ();
1811 return (_gdbus_debug_flags
& G_DBUS_DEBUG_TRANSPORT
) != 0;
1815 _g_dbus_debug_message (void)
1817 _g_dbus_initialize ();
1818 return (_gdbus_debug_flags
& G_DBUS_DEBUG_MESSAGE
) != 0;
1822 _g_dbus_debug_payload (void)
1824 _g_dbus_initialize ();
1825 return (_gdbus_debug_flags
& G_DBUS_DEBUG_PAYLOAD
) != 0;
1829 _g_dbus_debug_call (void)
1831 _g_dbus_initialize ();
1832 return (_gdbus_debug_flags
& G_DBUS_DEBUG_CALL
) != 0;
1836 _g_dbus_debug_signal (void)
1838 _g_dbus_initialize ();
1839 return (_gdbus_debug_flags
& G_DBUS_DEBUG_SIGNAL
) != 0;
1843 _g_dbus_debug_incoming (void)
1845 _g_dbus_initialize ();
1846 return (_gdbus_debug_flags
& G_DBUS_DEBUG_INCOMING
) != 0;
1850 _g_dbus_debug_return (void)
1852 _g_dbus_initialize ();
1853 return (_gdbus_debug_flags
& G_DBUS_DEBUG_RETURN
) != 0;
1857 _g_dbus_debug_emission (void)
1859 _g_dbus_initialize ();
1860 return (_gdbus_debug_flags
& G_DBUS_DEBUG_EMISSION
) != 0;
1864 _g_dbus_debug_address (void)
1866 _g_dbus_initialize ();
1867 return (_gdbus_debug_flags
& G_DBUS_DEBUG_ADDRESS
) != 0;
1870 G_LOCK_DEFINE_STATIC (print_lock
);
1873 _g_dbus_debug_print_lock (void)
1875 G_LOCK (print_lock
);
1879 _g_dbus_debug_print_unlock (void)
1881 G_UNLOCK (print_lock
);
1885 * _g_dbus_initialize:
1887 * Does various one-time init things such as
1889 * - registering the G_DBUS_ERROR error domain
1890 * - parses the G_DBUS_DEBUG environment variable
1893 _g_dbus_initialize (void)
1895 static volatile gsize initialized
= 0;
1897 if (g_once_init_enter (&initialized
))
1899 volatile GQuark g_dbus_error_domain
;
1902 g_dbus_error_domain
= G_DBUS_ERROR
;
1903 (g_dbus_error_domain
); /* To avoid -Wunused-but-set-variable */
1905 debug
= g_getenv ("G_DBUS_DEBUG");
1908 const GDebugKey keys
[] = {
1909 { "authentication", G_DBUS_DEBUG_AUTHENTICATION
},
1910 { "transport", G_DBUS_DEBUG_TRANSPORT
},
1911 { "message", G_DBUS_DEBUG_MESSAGE
},
1912 { "payload", G_DBUS_DEBUG_PAYLOAD
},
1913 { "call", G_DBUS_DEBUG_CALL
},
1914 { "signal", G_DBUS_DEBUG_SIGNAL
},
1915 { "incoming", G_DBUS_DEBUG_INCOMING
},
1916 { "return", G_DBUS_DEBUG_RETURN
},
1917 { "emission", G_DBUS_DEBUG_EMISSION
},
1918 { "address", G_DBUS_DEBUG_ADDRESS
}
1921 _gdbus_debug_flags
= g_parse_debug_string (debug
, keys
, G_N_ELEMENTS (keys
));
1922 if (_gdbus_debug_flags
& G_DBUS_DEBUG_PAYLOAD
)
1923 _gdbus_debug_flags
|= G_DBUS_DEBUG_MESSAGE
;
1926 /* Work-around for https://bugzilla.gnome.org/show_bug.cgi?id=627724 */
1927 ensure_required_types ();
1929 g_once_init_leave (&initialized
, 1);
1933 /* ---------------------------------------------------------------------------------------------------- */
1936 _g_dbus_compute_complete_signature (GDBusArgInfo
**args
)
1938 const GVariantType
*arg_types
[256];
1942 for (n
= 0; args
[n
] != NULL
; n
++)
1944 /* DBus places a hard limit of 255 on signature length.
1945 * therefore number of args must be less than 256.
1949 arg_types
[n
] = G_VARIANT_TYPE (args
[n
]->signature
);
1951 if G_UNLIKELY (arg_types
[n
] == NULL
)
1957 return g_variant_type_new_tuple (arg_types
, n
);
1960 /* ---------------------------------------------------------------------------------------------------- */
1964 extern BOOL WINAPI
ConvertSidToStringSidA (PSID Sid
, LPSTR
*StringSid
);
1967 _g_dbus_win32_get_user_sid (void)
1971 DWORD token_information_len
;
1978 h
= INVALID_HANDLE_VALUE
;
1980 if (!OpenProcessToken (GetCurrentProcess (), TOKEN_QUERY
, &h
))
1982 g_warning ("OpenProcessToken failed with error code %d", (gint
) GetLastError ());
1986 /* Get length of buffer */
1987 token_information_len
= 0;
1988 if (!GetTokenInformation (h
, TokenUser
, NULL
, 0, &token_information_len
))
1990 if (GetLastError () != ERROR_INSUFFICIENT_BUFFER
)
1992 g_warning ("GetTokenInformation() failed with error code %d", (gint
) GetLastError ());
1996 user
= g_malloc (token_information_len
);
1997 if (!GetTokenInformation (h
, TokenUser
, user
, token_information_len
, &token_information_len
))
1999 g_warning ("GetTokenInformation() failed with error code %d", (gint
) GetLastError ());
2003 psid
= user
->User
.Sid
;
2004 if (!IsValidSid (psid
))
2006 g_warning ("Invalid SID");
2010 if (!ConvertSidToStringSidA (psid
, &sid
))
2012 g_warning ("Invalid SID");
2016 ret
= g_strdup (sid
);
2021 if (h
!= INVALID_HANDLE_VALUE
)
2027 /* ---------------------------------------------------------------------------------------------------- */
2030 _g_dbus_get_machine_id (GError
**error
)
2033 HW_PROFILE_INFOA info
;
2034 char *src
, *dest
, *res
;
2037 if (!GetCurrentHwProfileA (&info
))
2039 char *message
= g_win32_error_message (GetLastError ());
2043 _("Unable to get Hardware profile: %s"), message
);
2048 /* Form: {12340001-4980-1920-6788-123456789012} */
2049 src
= &info
.szHwProfileGuid
[0];
2051 res
= g_malloc (32+1);
2055 for (i
= 0; i
< 8; i
++)
2058 for (i
= 0; i
< 4; i
++)
2061 for (i
= 0; i
< 4; i
++)
2064 for (i
= 0; i
< 4; i
++)
2067 for (i
= 0; i
< 12; i
++)
2074 GError
*first_error
;
2075 /* TODO: use PACKAGE_LOCALSTATEDIR ? */
2078 if (!g_file_get_contents ("/var/lib/dbus/machine-id",
2082 !g_file_get_contents ("/etc/machine-id",
2087 g_propagate_prefixed_error (error
, first_error
,
2088 _("Unable to load /var/lib/dbus/machine-id or /etc/machine-id: "));
2092 /* ignore the error from the first try, if any */
2093 g_clear_error (&first_error
);
2094 /* TODO: validate value */
2101 /* ---------------------------------------------------------------------------------------------------- */
2104 _g_dbus_enum_to_string (GType enum_type
, gint value
)
2108 GEnumValue
*enum_value
;
2110 klass
= g_type_class_ref (enum_type
);
2111 enum_value
= g_enum_get_value (klass
, value
);
2112 if (enum_value
!= NULL
)
2113 ret
= g_strdup (enum_value
->value_nick
);
2115 ret
= g_strdup_printf ("unknown (value %d)", value
);
2116 g_type_class_unref (klass
);
2120 /* ---------------------------------------------------------------------------------------------------- */
2123 write_message_print_transport_debug (gssize bytes_written
,
2124 MessageToWriteData
*data
)
2126 if (G_LIKELY (!_g_dbus_debug_transport ()))
2129 _g_dbus_debug_print_lock ();
2130 g_print ("========================================================================\n"
2131 "GDBus-debug:Transport:\n"
2132 " >>>> WROTE %" G_GSSIZE_FORMAT
" bytes of message with serial %d and\n"
2133 " size %" G_GSIZE_FORMAT
" from offset %" G_GSIZE_FORMAT
" on a %s\n",
2135 g_dbus_message_get_serial (data
->message
),
2137 data
->total_written
,
2138 g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_output_stream (data
->worker
->stream
))));
2139 _g_dbus_debug_print_unlock ();
2144 /* ---------------------------------------------------------------------------------------------------- */
2147 read_message_print_transport_debug (gssize bytes_read
,
2148 GDBusWorker
*worker
)
2152 gint32 message_length
;
2154 if (G_LIKELY (!_g_dbus_debug_transport ()))
2157 size
= bytes_read
+ worker
->read_buffer_cur_size
;
2161 message_length
= g_dbus_message_bytes_needed ((guchar
*) worker
->read_buffer
, size
, NULL
);
2164 switch (worker
->read_buffer
[0])
2168 serial
= GUINT32_FROM_LE (((guint32
*) worker
->read_buffer
)[2]);
2172 serial
= GUINT32_FROM_BE (((guint32
*) worker
->read_buffer
)[2]);
2175 /* an error will be set elsewhere if this happens */
2180 _g_dbus_debug_print_lock ();
2181 g_print ("========================================================================\n"
2182 "GDBus-debug:Transport:\n"
2183 " <<<< READ %" G_GSSIZE_FORMAT
" bytes of message with serial %d and\n"
2184 " size %d to offset %" G_GSIZE_FORMAT
" from a %s\n",
2188 worker
->read_buffer_cur_size
,
2189 g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_input_stream (worker
->stream
))));
2190 _g_dbus_debug_print_unlock ();
2195 /* ---------------------------------------------------------------------------------------------------- */
2198 _g_signal_accumulator_false_handled (GSignalInvocationHint
*ihint
,
2199 GValue
*return_accu
,
2200 const GValue
*handler_return
,
2203 gboolean continue_emission
;
2204 gboolean signal_return
;
2206 signal_return
= g_value_get_boolean (handler_return
);
2207 g_value_set_boolean (return_accu
, signal_return
);
2208 continue_emission
= signal_return
;
2210 return continue_emission
;