#include "gasyncresult.h"
#include "gsimpleasyncresult.h"
#include "ginputstream.h"
+#include "gmemoryinputstream.h"
#include "giostream.h"
#include "gsocketcontrolmessage.h"
#include "gsocketconnection.h"
+#include "gsocketoutputstream.h"
#ifdef G_OS_UNIX
#include "gunixfdmessage.h"
#endif
#include "glibintl.h"
-#include "gioalias.h"
/* ---------------------------------------------------------------------------------------------------- */
else
{
g_assert (error != NULL);
- g_simple_async_result_set_from_error (data->simple, error);
- g_error_free (error);
+ g_simple_async_result_take_error (data->simple, error);
}
if (data->from_mainloop)
/* ---------------------------------------------------------------------------------------------------- */
+/* Work-around for https://bugzilla.gnome.org/show_bug.cgi?id=627724 */
+
+static GPtrArray *ensured_classes = NULL;
+
+static void
+ensure_type (GType gtype)
+{
+ g_ptr_array_add (ensured_classes, g_type_class_ref (gtype));
+}
+
+static void
+released_required_types (void)
+{
+ g_ptr_array_foreach (ensured_classes, (GFunc) g_type_class_unref, NULL);
+ g_ptr_array_unref (ensured_classes);
+ ensured_classes = NULL;
+}
+
+static void
+ensure_required_types (void)
+{
+ g_assert (ensured_classes == NULL);
+ ensured_classes = g_ptr_array_new ();
+ ensure_type (G_TYPE_SIMPLE_ASYNC_RESULT);
+ ensure_type (G_TYPE_MEMORY_INPUT_STREAM);
+}
+/* ---------------------------------------------------------------------------------------------------- */
+
G_LOCK_DEFINE_STATIC (shared_thread_lock);
typedef struct
static SharedThreadData *shared_thread_data = NULL;
static gpointer
-shared_thread_func (gpointer data)
+gdbus_shared_thread_func (gpointer data)
{
g_main_context_push_thread_default (shared_thread_data->context);
g_main_loop_run (shared_thread_data->loop);
return FALSE;
}
+/* ---------------------------------------------------------------------------------------------------- */
+
static void
_g_dbus_shared_thread_ref (GDBusSharedThreadFunc func,
gpointer user_data)
GError *error;
GSource *idle_source;
CallerData *data;
+ gboolean release_types;
G_LOCK (shared_thread_lock);
+ release_types = FALSE;
+
if (shared_thread_data != NULL)
{
shared_thread_data->num_users += 1;
shared_thread_data = g_new0 (SharedThreadData, 1);
shared_thread_data->num_users = 1;
+ /* Work-around for https://bugzilla.gnome.org/show_bug.cgi?id=627724 */
+ ensure_required_types ();
+ release_types = TRUE;
+
error = NULL;
shared_thread_data->context = g_main_context_new ();
shared_thread_data->loop = g_main_loop_new (shared_thread_data->context, FALSE);
- shared_thread_data->thread = g_thread_create (shared_thread_func,
+ shared_thread_data->thread = g_thread_create (gdbus_shared_thread_func,
NULL,
TRUE,
&error);
while (!data->done)
g_thread_yield ();
+ if (release_types)
+ released_required_types ();
+
g_free (data);
G_UNLOCK (shared_thread_lock);
/* used for writing */
GMutex *write_lock;
GQueue *write_queue;
- gboolean write_is_pending;
+ gint num_writes_pending;
+ guint64 write_num_messages_written;
+ GList *write_pending_flushes;
+ gboolean flush_pending;
};
+/* ---------------------------------------------------------------------------------------------------- */
+
+typedef struct
+{
+ GMutex *mutex;
+ GCond *cond;
+ guint64 number_to_wait_for;
+ GError *error;
+} FlushData;
+
struct _MessageToWriteData ;
typedef struct _MessageToWriteData MessageToWriteData;
static void message_to_write_data_free (MessageToWriteData *data);
+static void read_message_print_transport_debug (gssize bytes_read,
+ GDBusWorker *worker);
+
+static void write_message_print_transport_debug (gssize bytes_written,
+ MessageToWriteData *data);
+
+/* ---------------------------------------------------------------------------------------------------- */
+
static GDBusWorker *
_g_dbus_worker_ref (GDBusWorker *worker)
{
{
if (g_atomic_int_dec_and_test (&worker->ref_count))
{
+ g_assert (worker->write_pending_flushes == NULL);
+
_g_dbus_shared_thread_unref ();
g_object_unref (worker->stream);
g_queue_foreach (worker->write_queue, (GFunc) message_to_write_data_free, NULL);
g_queue_free (worker->write_queue);
+ g_free (worker->read_buffer);
+
g_free (worker);
}
}
worker->message_received_callback (worker, message, worker->user_data);
}
-static gboolean
+static GDBusMessage *
_g_dbus_worker_emit_message_about_to_be_sent (GDBusWorker *worker,
GDBusMessage *message)
{
- gboolean ret;
- ret = FALSE;
+ GDBusMessage *ret;
if (!worker->stopped)
ret = worker->message_about_to_be_sent_callback (worker, message, worker->user_data);
+ else
+ ret = message;
return ret;
}
goto out;
}
+ read_message_print_transport_debug (bytes_read, worker);
+
worker->read_buffer_cur_size += bytes_read;
if (worker->read_buffer_bytes_wanted == worker->read_buffer_cur_size)
{
if (worker->read_fd_list != NULL)
{
g_dbus_message_set_unix_fd_list (message, worker->read_fd_list);
+ g_object_unref (worker->read_fd_list);
worker->read_fd_list = NULL;
}
#endif
struct _MessageToWriteData
{
+ GDBusWorker *worker;
GDBusMessage *message;
gchar *blob;
gsize blob_size;
+
+ gsize total_written;
+ GSimpleAsyncResult *simple;
+
};
static void
message_to_write_data_free (MessageToWriteData *data)
{
- g_object_unref (data->message);
+ _g_dbus_worker_unref (data->worker);
+ if (data->message)
+ g_object_unref (data->message);
g_free (data->blob);
g_free (data);
}
/* ---------------------------------------------------------------------------------------------------- */
+static void write_message_continue_writing (MessageToWriteData *data);
+
/* called in private thread shared by all GDBusConnection instances (without write-lock held) */
-static gboolean
-write_message (GDBusWorker *worker,
- MessageToWriteData *data,
- GError **error)
+static void
+write_message_async_cb (GObject *source_object,
+ GAsyncResult *res,
+ gpointer user_data)
{
- gboolean ret;
+ MessageToWriteData *data = user_data;
+ GSimpleAsyncResult *simple;
+ gssize bytes_written;
+ GError *error;
- g_return_val_if_fail (data->blob_size > 16, FALSE);
+ /* Note: we can't access data->simple after calling g_async_result_complete () because the
+ * callback can free @data and we're not completing in idle. So use a copy of the pointer.
+ */
+ simple = data->simple;
- ret = FALSE;
+ error = NULL;
+ bytes_written = g_output_stream_write_finish (G_OUTPUT_STREAM (source_object),
+ res,
+ &error);
+ if (bytes_written == -1)
+ {
+ g_simple_async_result_take_error (simple, error);
+ g_simple_async_result_complete (simple);
+ g_object_unref (simple);
+ goto out;
+ }
+ g_assert (bytes_written > 0); /* zero is never returned */
+
+ write_message_print_transport_debug (bytes_written, data);
+
+ data->total_written += bytes_written;
+ g_assert (data->total_written <= data->blob_size);
+ if (data->total_written == data->blob_size)
+ {
+ g_simple_async_result_complete (simple);
+ g_object_unref (simple);
+ goto out;
+ }
- /* First, the initial 16 bytes - special case UNIX sockets here
- * since it may involve writing an ancillary message with file
- * descriptors
+ write_message_continue_writing (data);
+
+ out:
+ ;
+}
+
+/* called in private thread shared by all GDBusConnection instances (without write-lock held) */
+static gboolean
+on_socket_ready (GSocket *socket,
+ GIOCondition condition,
+ gpointer user_data)
+{
+ MessageToWriteData *data = user_data;
+ write_message_continue_writing (data);
+ return FALSE; /* remove source */
+}
+
+/* called in private thread shared by all GDBusConnection instances (without write-lock held) */
+static void
+write_message_continue_writing (MessageToWriteData *data)
+{
+ GOutputStream *ostream;
+ GSimpleAsyncResult *simple;
+#ifdef G_OS_UNIX
+ GUnixFDList *fd_list;
+#endif
+
+ /* Note: we can't access data->simple after calling g_async_result_complete () because the
+ * callback can free @data and we're not completing in idle. So use a copy of the pointer.
*/
+ simple = data->simple;
+
+ ostream = g_io_stream_get_output_stream (data->worker->stream);
#ifdef G_OS_UNIX
- {
- GOutputVector vector;
- GSocketControlMessage *message;
- GUnixFDList *fd_list;
- gssize bytes_written;
+ fd_list = g_dbus_message_get_unix_fd_list (data->message);
+#endif
- fd_list = g_dbus_message_get_unix_fd_list (data->message);
+ g_assert (!g_output_stream_has_pending (ostream));
+ g_assert_cmpint (data->total_written, <, data->blob_size);
- message = NULL;
- if (fd_list != NULL)
- {
- if (!G_IS_UNIX_CONNECTION (worker->stream))
- {
- g_set_error (error,
- G_IO_ERROR,
- G_IO_ERROR_INVALID_ARGUMENT,
- "Tried sending a file descriptor on unsupported stream of type %s",
- g_type_name (G_TYPE_FROM_INSTANCE (worker->stream)));
- goto out;
- }
- else if (!(worker->capabilities & G_DBUS_CAPABILITY_FLAGS_UNIX_FD_PASSING))
- {
- g_set_error_literal (error,
- G_IO_ERROR,
- G_IO_ERROR_INVALID_ARGUMENT,
- "Tried sending a file descriptor but remote peer does not support this capability");
- goto out;
- }
- message = g_unix_fd_message_new_with_fd_list (fd_list);
- }
+ if (FALSE)
+ {
+ }
+#ifdef G_OS_UNIX
+ else if (G_IS_SOCKET_OUTPUT_STREAM (ostream) && data->total_written == 0)
+ {
+ GOutputVector vector;
+ GSocketControlMessage *control_message;
+ gssize bytes_written;
+ GError *error;
- vector.buffer = data->blob;
- vector.size = 16;
-
- bytes_written = g_socket_send_message (worker->socket,
- NULL, /* address */
- &vector,
- 1,
- message != NULL ? &message : NULL,
- message != NULL ? 1 : 0,
- G_SOCKET_MSG_NONE,
- worker->cancellable,
- error);
- if (bytes_written == -1)
- {
- g_prefix_error (error, _("Error writing first 16 bytes of message to socket: "));
- if (message != NULL)
- g_object_unref (message);
- goto out;
- }
- if (message != NULL)
- g_object_unref (message);
+ vector.buffer = data->blob;
+ vector.size = data->blob_size;
- if (bytes_written < 16)
- {
- /* TODO: I think this needs to be handled ... are we guaranteed that the ancillary
- * messages are sent?
- */
- g_assert_not_reached ();
- }
- }
-#else
- /* write the first 16 bytes (guaranteed to return an error if everything can't be written) */
- if (!g_output_stream_write_all (g_io_stream_get_output_stream (worker->stream),
- (const gchar *) data->blob,
- 16,
- NULL, /* bytes_written */
- worker->cancellable, /* cancellable */
- error))
- goto out;
+ control_message = NULL;
+ if (fd_list != NULL && g_unix_fd_list_get_length (fd_list) > 0)
+ {
+ if (!(data->worker->capabilities & G_DBUS_CAPABILITY_FLAGS_UNIX_FD_PASSING))
+ {
+ g_simple_async_result_set_error (simple,
+ G_IO_ERROR,
+ G_IO_ERROR_FAILED,
+ "Tried sending a file descriptor but remote peer does not support this capability");
+ g_simple_async_result_complete (simple);
+ g_object_unref (simple);
+ goto out;
+ }
+ control_message = g_unix_fd_message_new_with_fd_list (fd_list);
+ }
+
+ error = NULL;
+ bytes_written = g_socket_send_message (data->worker->socket,
+ NULL, /* address */
+ &vector,
+ 1,
+ control_message != NULL ? &control_message : NULL,
+ control_message != NULL ? 1 : 0,
+ G_SOCKET_MSG_NONE,
+ data->worker->cancellable,
+ &error);
+ if (control_message != NULL)
+ g_object_unref (control_message);
+
+ if (bytes_written == -1)
+ {
+ /* Handle WOULD_BLOCK by waiting until there's room in the buffer */
+ if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
+ {
+ GSource *source;
+ source = g_socket_create_source (data->worker->socket,
+ G_IO_OUT | G_IO_HUP | G_IO_ERR,
+ data->worker->cancellable);
+ g_source_set_callback (source,
+ (GSourceFunc) on_socket_ready,
+ data,
+ NULL); /* GDestroyNotify */
+ g_source_attach (source, g_main_context_get_thread_default ());
+ g_source_unref (source);
+ g_error_free (error);
+ goto out;
+ }
+ g_simple_async_result_take_error (simple, error);
+ g_simple_async_result_complete (simple);
+ g_object_unref (simple);
+ goto out;
+ }
+ g_assert (bytes_written > 0); /* zero is never returned */
+
+ write_message_print_transport_debug (bytes_written, data);
+
+ data->total_written += bytes_written;
+ g_assert (data->total_written <= data->blob_size);
+ if (data->total_written == data->blob_size)
+ {
+ g_simple_async_result_complete (simple);
+ g_object_unref (simple);
+ goto out;
+ }
+
+ write_message_continue_writing (data);
+ }
+#endif
+ else
+ {
+#ifdef G_OS_UNIX
+ if (fd_list != NULL)
+ {
+ g_simple_async_result_set_error (simple,
+ G_IO_ERROR,
+ G_IO_ERROR_FAILED,
+ "Tried sending a file descriptor on unsupported stream of type %s",
+ g_type_name (G_TYPE_FROM_INSTANCE (ostream)));
+ g_simple_async_result_complete (simple);
+ g_object_unref (simple);
+ goto out;
+ }
#endif
- /* Then write the rest of the message (guaranteed to return an error if everything can't be written) */
- if (!g_output_stream_write_all (g_io_stream_get_output_stream (worker->stream),
- (const gchar *) data->blob + 16,
- data->blob_size - 16,
- NULL, /* bytes_written */
- worker->cancellable, /* cancellable */
- error))
- goto out;
+ g_output_stream_write_async (ostream,
+ (const gchar *) data->blob + data->total_written,
+ data->blob_size - data->total_written,
+ G_PRIORITY_DEFAULT,
+ data->worker->cancellable,
+ write_message_async_cb,
+ data);
+ }
+ out:
+ ;
+}
- ret = TRUE;
+/* called in private thread shared by all GDBusConnection instances (without write-lock held) */
+static void
+write_message_async (GDBusWorker *worker,
+ MessageToWriteData *data,
+ GAsyncReadyCallback callback,
+ gpointer user_data)
+{
+ data->simple = g_simple_async_result_new (NULL,
+ callback,
+ user_data,
+ write_message_async);
+ data->total_written = 0;
+ write_message_continue_writing (data);
+}
+
+/* called in private thread shared by all GDBusConnection instances (without write-lock held) */
+static gboolean
+write_message_finish (GAsyncResult *res,
+ GError **error)
+{
+ g_warn_if_fail (g_simple_async_result_get_source_tag (G_SIMPLE_ASYNC_RESULT (res)) == write_message_async);
+ if (g_simple_async_result_propagate_error (G_SIMPLE_ASYNC_RESULT (res), error))
+ return FALSE;
+ else
+ return TRUE;
+}
+/* ---------------------------------------------------------------------------------------------------- */
+
+static void maybe_write_next_message (GDBusWorker *worker);
+
+typedef struct
+{
+ GDBusWorker *worker;
+ GList *flushers;
+} FlushAsyncData;
+
+/* called in private thread shared by all GDBusConnection instances (without write-lock held) */
+static void
+ostream_flush_cb (GObject *source_object,
+ GAsyncResult *res,
+ gpointer user_data)
+{
+ FlushAsyncData *data = user_data;
+ GError *error;
+ GList *l;
+
+ error = NULL;
+ g_output_stream_flush_finish (G_OUTPUT_STREAM (source_object),
+ res,
+ &error);
+ if (error == NULL)
+ {
+ if (G_UNLIKELY (_g_dbus_debug_transport ()))
+ {
+ _g_dbus_debug_print_lock ();
+ g_print ("========================================================================\n"
+ "GDBus-debug:Transport:\n"
+ " ---- FLUSHED stream of type %s\n",
+ g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_output_stream (data->worker->stream))));
+ _g_dbus_debug_print_unlock ();
+ }
+ }
+
+ g_assert (data->flushers != NULL);
+ for (l = data->flushers; l != NULL; l = l->next)
+ {
+ FlushData *f = l->data;
+
+ f->error = error != NULL ? g_error_copy (error) : NULL;
+
+ g_mutex_lock (f->mutex);
+ g_cond_signal (f->cond);
+ g_mutex_unlock (f->mutex);
+ }
+ g_list_free (data->flushers);
+
+ if (error != NULL)
+ g_error_free (error);
+
+ /* Make sure we tell folks that we don't have additional
+ flushes pending */
+ g_mutex_lock (data->worker->write_lock);
+ data->worker->flush_pending = FALSE;
+ g_mutex_unlock (data->worker->write_lock);
+
+ /* OK, cool, finally kick off the next write */
+ maybe_write_next_message (data->worker);
+
+ _g_dbus_worker_unref (data->worker);
+ g_free (data);
+}
+
+/* called in private thread shared by all GDBusConnection instances (without write-lock held) */
+static void
+message_written (GDBusWorker *worker,
+ MessageToWriteData *message_data)
+{
+ GList *l;
+ GList *ll;
+ GList *flushers;
+
+ /* first log the fact that we wrote a message */
if (G_UNLIKELY (_g_dbus_debug_message ()))
{
gchar *s;
g_print ("========================================================================\n"
"GDBus-debug:Message:\n"
" >>>> SENT D-Bus message (%" G_GSIZE_FORMAT " bytes)\n",
- data->blob_size);
- s = g_dbus_message_print (data->message, 2);
+ message_data->blob_size);
+ s = g_dbus_message_print (message_data->message, 2);
g_print ("%s", s);
g_free (s);
if (G_UNLIKELY (_g_dbus_debug_payload ()))
{
- s = _g_dbus_hexdump (data->blob, data->blob_size, 2);
+ s = _g_dbus_hexdump (message_data->blob, message_data->blob_size, 2);
g_print ("%s\n", s);
g_free (s);
}
_g_dbus_debug_print_unlock ();
}
- out:
- return ret;
+ /* then first wake up pending flushes and, if needed, flush the stream */
+ flushers = NULL;
+ g_mutex_lock (worker->write_lock);
+ worker->write_num_messages_written += 1;
+ for (l = worker->write_pending_flushes; l != NULL; l = ll)
+ {
+ FlushData *f = l->data;
+ ll = l->next;
+
+ if (f->number_to_wait_for == worker->write_num_messages_written)
+ {
+ flushers = g_list_append (flushers, f);
+ worker->write_pending_flushes = g_list_delete_link (worker->write_pending_flushes, l);
+ }
+ }
+ if (flushers != NULL)
+ {
+ worker->flush_pending = TRUE;
+ }
+ g_mutex_unlock (worker->write_lock);
+
+ if (flushers != NULL)
+ {
+ FlushAsyncData *data;
+ data = g_new0 (FlushAsyncData, 1);
+ data->worker = _g_dbus_worker_ref (worker);
+ data->flushers = flushers;
+ /* flush the stream before writing the next message */
+ g_output_stream_flush_async (g_io_stream_get_output_stream (worker->stream),
+ G_PRIORITY_DEFAULT,
+ worker->cancellable,
+ ostream_flush_cb,
+ data);
+ }
+ else
+ {
+ /* kick off the next write! */
+ maybe_write_next_message (worker);
+ }
}
-/* ---------------------------------------------------------------------------------------------------- */
+/* called in private thread shared by all GDBusConnection instances (without write-lock held) */
+static void
+write_message_cb (GObject *source_object,
+ GAsyncResult *res,
+ gpointer user_data)
+{
+ MessageToWriteData *data = user_data;
+ GError *error;
+
+ g_mutex_lock (data->worker->write_lock);
+ data->worker->num_writes_pending -= 1;
+ g_mutex_unlock (data->worker->write_lock);
+
+ error = NULL;
+ if (!write_message_finish (res, &error))
+ {
+ /* TODO: handle */
+ _g_dbus_worker_emit_disconnected (data->worker, TRUE, error);
+ g_error_free (error);
+ }
+
+ /* this function will also kick of the next write (it might need to
+ * flush so writing the next message might happen much later
+ * e.g. async)
+ */
+ message_written (data->worker, data);
+
+ message_to_write_data_free (data);
+}
/* called in private thread shared by all GDBusConnection instances (without write-lock held) */
-static gboolean
-write_message_in_idle_cb (gpointer user_data)
+static void
+maybe_write_next_message (GDBusWorker *worker)
{
- GDBusWorker *worker = user_data;
- gboolean more_writes_are_pending;
MessageToWriteData *data;
- gboolean message_was_dropped;
- GError *error;
+
+ write_next:
g_mutex_lock (worker->write_lock);
data = g_queue_pop_head (worker->write_queue);
- g_assert (data != NULL);
- more_writes_are_pending = (g_queue_get_length (worker->write_queue) > 0);
- worker->write_is_pending = more_writes_are_pending;
+ if (data != NULL)
+ worker->num_writes_pending += 1;
g_mutex_unlock (worker->write_lock);
/* Note that write_lock is only used for protecting the @write_queue
- * and @write_is_pending fields of the GDBusWorker struct ... which we
+ * and @num_writes_pending fields of the GDBusWorker struct ... which we
* need to modify from arbitrary threads in _g_dbus_worker_send_message().
*
* Therefore, it's fine to drop it here when calling back into user
* code and then writing the message out onto the GIOStream since this
* function only runs on the worker thread.
*/
- message_was_dropped = _g_dbus_worker_emit_message_about_to_be_sent (worker, data->message);
- if (G_LIKELY (!message_was_dropped))
+ if (data != NULL)
{
- error = NULL;
- if (!write_message (worker,
- data,
- &error))
+ GDBusMessage *old_message;
+ guchar *new_blob;
+ gsize new_blob_size;
+ GError *error;
+
+ old_message = data->message;
+ data->message = _g_dbus_worker_emit_message_about_to_be_sent (worker, data->message);
+ if (data->message == old_message)
+ {
+ /* filters had no effect - do nothing */
+ }
+ else if (data->message == NULL)
+ {
+ /* filters dropped message */
+ g_mutex_lock (worker->write_lock);
+ worker->num_writes_pending -= 1;
+ g_mutex_unlock (worker->write_lock);
+ message_to_write_data_free (data);
+ goto write_next;
+ }
+ else
{
- /* TODO: handle */
- _g_dbus_worker_emit_disconnected (worker, TRUE, error);
- g_error_free (error);
+ /* filters altered the message -> reencode */
+ error = NULL;
+ new_blob = g_dbus_message_to_blob (data->message,
+ &new_blob_size,
+ worker->capabilities,
+ &error);
+ if (new_blob == NULL)
+ {
+ /* if filter make the GDBusMessage unencodeable, just complain on stderr and send
+ * the old message instead
+ */
+ g_warning ("Error encoding GDBusMessage with serial %d altered by filter function: %s",
+ g_dbus_message_get_serial (data->message),
+ error->message);
+ g_error_free (error);
+ }
+ else
+ {
+ g_free (data->blob);
+ data->blob = (gchar *) new_blob;
+ data->blob_size = new_blob_size;
+ }
}
+
+ write_message_async (worker,
+ data,
+ write_message_cb,
+ data);
}
- message_to_write_data_free (data);
+}
- return more_writes_are_pending;
+/* called in private thread shared by all GDBusConnection instances (without write-lock held) */
+static gboolean
+write_message_in_idle_cb (gpointer user_data)
+{
+ GDBusWorker *worker = user_data;
+ if (worker->num_writes_pending == 0 && !worker->flush_pending)
+ maybe_write_next_message (worker);
+ return FALSE;
}
/* ---------------------------------------------------------------------------------------------------- */
g_return_if_fail (blob_len > 16);
data = g_new0 (MessageToWriteData, 1);
+ data->worker = _g_dbus_worker_ref (worker);
data->message = g_object_ref (message);
data->blob = blob; /* steal! */
data->blob_size = blob_len;
g_mutex_lock (worker->write_lock);
g_queue_push_tail (worker->write_queue, data);
- if (!worker->write_is_pending)
+ if (worker->num_writes_pending == 0)
{
GSource *idle_source;
-
- worker->write_is_pending = TRUE;
-
idle_source = g_idle_source_new ();
g_source_set_priority (idle_source, G_PRIORITY_DEFAULT);
g_source_set_callback (idle_source,
worker->stream = g_object_ref (stream);
worker->capabilities = capabilities;
worker->cancellable = g_cancellable_new ();
+ worker->flush_pending = FALSE;
worker->frozen = initially_frozen;
worker->received_messages_while_frozen = g_queue_new ();
return worker;
}
-/* This can be called from any thread - frees worker - guarantees no callbacks
- * will ever be issued again
+/* ---------------------------------------------------------------------------------------------------- */
+
+/* This can be called from any thread - frees worker. Note that
+ * callbacks might still happen if called from another thread than the
+ * worker - use your own synchronization primitive in the callbacks.
*/
void
_g_dbus_worker_stop (GDBusWorker *worker)
{
- /* If we're called in the worker thread it means we are called from
- * a worker callback. This is fine, we just can't lock in that case since
- * we're already holding the lock...
- */
- if (g_thread_self () != worker->thread)
- g_mutex_lock (worker->read_lock);
worker->stopped = TRUE;
- if (g_thread_self () != worker->thread)
- g_mutex_unlock (worker->read_lock);
-
g_cancellable_cancel (worker->cancellable);
_g_dbus_worker_unref (worker);
}
+/* ---------------------------------------------------------------------------------------------------- */
+
+/* can be called from any thread (except the worker thread) - blocks
+ * calling thread until all queued outgoing messages are written and
+ * the transport has been flushed
+ */
+gboolean
+_g_dbus_worker_flush_sync (GDBusWorker *worker,
+ GCancellable *cancellable,
+ GError **error)
+{
+ gboolean ret;
+ FlushData *data;
+
+ data = NULL;
+ ret = TRUE;
+
+ /* if the queue is empty, there's nothing to wait for */
+ g_mutex_lock (worker->write_lock);
+ if (g_queue_get_length (worker->write_queue) > 0)
+ {
+ data = g_new0 (FlushData, 1);
+ data->mutex = g_mutex_new ();
+ data->cond = g_cond_new ();
+ data->number_to_wait_for = worker->write_num_messages_written + g_queue_get_length (worker->write_queue);
+ g_mutex_lock (data->mutex);
+ worker->write_pending_flushes = g_list_prepend (worker->write_pending_flushes, data);
+ }
+ g_mutex_unlock (worker->write_lock);
+
+ if (data != NULL)
+ {
+ g_cond_wait (data->cond, data->mutex);
+ g_mutex_unlock (data->mutex);
+
+ /* note:the element is removed from worker->write_pending_flushes in flush_cb() above */
+ g_cond_free (data->cond);
+ g_mutex_free (data->mutex);
+ if (data->error != NULL)
+ {
+ ret = FALSE;
+ g_propagate_error (error, data->error);
+ }
+ g_free (data);
+ }
+
+ return ret;
+}
+
+/* ---------------------------------------------------------------------------------------------------- */
+
#define G_DBUS_DEBUG_AUTHENTICATION (1<<0)
-#define G_DBUS_DEBUG_MESSAGE (1<<1)
-#define G_DBUS_DEBUG_PAYLOAD (1<<2)
-#define G_DBUS_DEBUG_CALL (1<<3)
-#define G_DBUS_DEBUG_SIGNAL (1<<4)
-#define G_DBUS_DEBUG_INCOMING (1<<5)
-#define G_DBUS_DEBUG_EMISSION (1<<6)
-#define G_DBUS_DEBUG_ADDRESS (1<<7)
-#define G_DBUS_DEBUG_ALL 0xffffffff
+#define G_DBUS_DEBUG_TRANSPORT (1<<1)
+#define G_DBUS_DEBUG_MESSAGE (1<<2)
+#define G_DBUS_DEBUG_PAYLOAD (1<<3)
+#define G_DBUS_DEBUG_CALL (1<<4)
+#define G_DBUS_DEBUG_SIGNAL (1<<5)
+#define G_DBUS_DEBUG_INCOMING (1<<6)
+#define G_DBUS_DEBUG_RETURN (1<<7)
+#define G_DBUS_DEBUG_EMISSION (1<<8)
+#define G_DBUS_DEBUG_ADDRESS (1<<9)
+
static gint _gdbus_debug_flags = 0;
gboolean
}
gboolean
+_g_dbus_debug_transport (void)
+{
+ _g_dbus_initialize ();
+ return (_gdbus_debug_flags & G_DBUS_DEBUG_TRANSPORT) != 0;
+}
+
+gboolean
_g_dbus_debug_message (void)
{
_g_dbus_initialize ();
}
gboolean
+_g_dbus_debug_return (void)
+{
+ _g_dbus_initialize ();
+ return (_gdbus_debug_flags & G_DBUS_DEBUG_RETURN) != 0;
+}
+
+gboolean
_g_dbus_debug_emission (void)
{
_g_dbus_initialize ();
const gchar *debug;
g_dbus_error_domain = G_DBUS_ERROR;
+ (g_dbus_error_domain); /* To avoid -Wunused-but-set-variable */
debug = g_getenv ("G_DBUS_DEBUG");
if (debug != NULL)
{
- gchar **tokens;
- guint n;
- tokens = g_strsplit (debug, ",", 0);
- for (n = 0; tokens[n] != NULL; n++)
- {
- if (g_strcmp0 (tokens[n], "authentication") == 0)
- _gdbus_debug_flags |= G_DBUS_DEBUG_AUTHENTICATION;
- else if (g_strcmp0 (tokens[n], "message") == 0)
- _gdbus_debug_flags |= G_DBUS_DEBUG_MESSAGE;
- else if (g_strcmp0 (tokens[n], "payload") == 0) /* implies `message' */
- _gdbus_debug_flags |= (G_DBUS_DEBUG_MESSAGE | G_DBUS_DEBUG_PAYLOAD);
- else if (g_strcmp0 (tokens[n], "call") == 0)
- _gdbus_debug_flags |= G_DBUS_DEBUG_CALL;
- else if (g_strcmp0 (tokens[n], "signal") == 0)
- _gdbus_debug_flags |= G_DBUS_DEBUG_SIGNAL;
- else if (g_strcmp0 (tokens[n], "incoming") == 0)
- _gdbus_debug_flags |= G_DBUS_DEBUG_INCOMING;
- else if (g_strcmp0 (tokens[n], "emission") == 0)
- _gdbus_debug_flags |= G_DBUS_DEBUG_EMISSION;
- else if (g_strcmp0 (tokens[n], "address") == 0)
- _gdbus_debug_flags |= G_DBUS_DEBUG_ADDRESS;
- else if (g_strcmp0 (tokens[n], "all") == 0)
- _gdbus_debug_flags |= G_DBUS_DEBUG_ALL;
- }
- g_strfreev (tokens);
+ const GDebugKey keys[] = {
+ { "authentication", G_DBUS_DEBUG_AUTHENTICATION },
+ { "transport", G_DBUS_DEBUG_TRANSPORT },
+ { "message", G_DBUS_DEBUG_MESSAGE },
+ { "payload", G_DBUS_DEBUG_PAYLOAD },
+ { "call", G_DBUS_DEBUG_CALL },
+ { "signal", G_DBUS_DEBUG_SIGNAL },
+ { "incoming", G_DBUS_DEBUG_INCOMING },
+ { "return", G_DBUS_DEBUG_RETURN },
+ { "emission", G_DBUS_DEBUG_EMISSION },
+ { "address", G_DBUS_DEBUG_ADDRESS }
+ };
+
+ _gdbus_debug_flags = g_parse_debug_string (debug, keys, G_N_ELEMENTS (keys));
+ if (_gdbus_debug_flags & G_DBUS_DEBUG_PAYLOAD)
+ _gdbus_debug_flags |= G_DBUS_DEBUG_MESSAGE;
}
g_once_init_leave (&initialized, 1);
/* ---------------------------------------------------------------------------------------------------- */
-#define __G_DBUS_PRIVATE_C__
-#include "gioaliasdef.c"
+static void
+write_message_print_transport_debug (gssize bytes_written,
+ MessageToWriteData *data)
+{
+ if (G_LIKELY (!_g_dbus_debug_transport ()))
+ goto out;
+
+ _g_dbus_debug_print_lock ();
+ g_print ("========================================================================\n"
+ "GDBus-debug:Transport:\n"
+ " >>>> WROTE %" G_GSIZE_FORMAT " bytes of message with serial %d and\n"
+ " size %" G_GSIZE_FORMAT " from offset %" G_GSIZE_FORMAT " on a %s\n",
+ bytes_written,
+ g_dbus_message_get_serial (data->message),
+ data->blob_size,
+ data->total_written,
+ g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_output_stream (data->worker->stream))));
+ _g_dbus_debug_print_unlock ();
+ out:
+ ;
+}
+
+/* ---------------------------------------------------------------------------------------------------- */
+
+static void
+read_message_print_transport_debug (gssize bytes_read,
+ GDBusWorker *worker)
+{
+ gsize size;
+ gint32 serial;
+ gint32 message_length;
+
+ if (G_LIKELY (!_g_dbus_debug_transport ()))
+ goto out;
+
+ size = bytes_read + worker->read_buffer_cur_size;
+ serial = 0;
+ message_length = 0;
+ if (size >= 16)
+ message_length = g_dbus_message_bytes_needed ((guchar *) worker->read_buffer, size, NULL);
+ if (size >= 1)
+ {
+ switch (worker->read_buffer[0])
+ {
+ case 'l':
+ if (size >= 12)
+ serial = GUINT32_FROM_LE (((guint32 *) worker->read_buffer)[2]);
+ break;
+ case 'B':
+ if (size >= 12)
+ serial = GUINT32_FROM_BE (((guint32 *) worker->read_buffer)[2]);
+ break;
+ default:
+ /* an error will be set elsewhere if this happens */
+ goto out;
+ }
+ }
+
+ _g_dbus_debug_print_lock ();
+ g_print ("========================================================================\n"
+ "GDBus-debug:Transport:\n"
+ " <<<< READ %" G_GSIZE_FORMAT " bytes of message with serial %d and\n"
+ " size %d to offset %" G_GSIZE_FORMAT " from a %s\n",
+ bytes_read,
+ serial,
+ message_length,
+ worker->read_buffer_cur_size,
+ g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_input_stream (worker->stream))));
+ _g_dbus_debug_print_unlock ();
+ out:
+ ;
+}
projects / platform / upstream / glib.git / blobdiff