+2003-04-16 Havoc Pennington <hp@redhat.com>
+
+ * dbus/dbus-message.c (_dbus_message_add_size_counter): keep a
+ list of size counters
+ (_dbus_message_loader_putback_message_link): put back a popped link
+
+ * dbus/dbus-connection.c
+ (dbus_connection_set_max_live_messages_size): rename
+ max_received_size
+ (dbus_connection_get_outgoing_size): get size of outgoing
+ queue
+ (_dbus_connection_set_connection_counter): remove this cruft
+
2003-04-14 Havoc Pennington <hp@redhat.com>
* dbus/dbus-userdb.c: user database abstraction, mostly to get
void _dbus_connection_notify_disconnected (DBusConnection *connection);
void _dbus_connection_handler_destroyed_locked (DBusConnection *connection,
DBusMessageHandler *handler);
-void _dbus_connection_set_connection_counter (DBusConnection *connection,
- DBusCounter *counter);
dbus_bool_t _dbus_message_handler_add_connection (DBusMessageHandler *handler,
DBusConnection *connection);
void _dbus_message_handler_remove_connection (DBusMessageHandler *handler,
int n_outgoing; /**< Length of outgoing queue. */
int n_incoming; /**< Length of incoming queue. */
+
+ DBusCounter *outgoing_counter; /**< Counts size of outgoing messages. */
DBusTransport *transport; /**< Object that sends/receives messages over network. */
DBusWatchList *watches; /**< Stores active watches. */
DBusDataSlotList slot_list; /**< Data stored by allocated integer ID */
DBusHashTable *pending_replies; /**< Hash of message serials and their message handlers. */
- DBusCounter *connection_counter; /**< Counter that we decrement when finalized */
int client_serial; /**< Client serial. Increments each time a message is sent */
DBusList *disconnect_message_link; /**< Preallocated list node for queueing the disconnection message */
message, dbus_message_get_name (message),
connection, connection->n_outgoing);
+ _dbus_message_remove_size_counter (message, connection->outgoing_counter);
+
dbus_message_unref (message);
if (connection->n_outgoing == 0)
DBusCondVar *io_path_cond;
DBusList *disconnect_link;
DBusMessage *disconnect_message;
-
+ DBusCounter *outgoing_counter;
+
watch_list = NULL;
connection = NULL;
handler_table = NULL;
io_path_cond = NULL;
disconnect_link = NULL;
disconnect_message = NULL;
+ outgoing_counter = NULL;
watch_list = _dbus_watch_list_new ();
if (watch_list == NULL)
if (disconnect_link == NULL)
goto error;
+ outgoing_counter = _dbus_counter_new ();
+ if (outgoing_counter == NULL)
+ goto error;
+
if (_dbus_modify_sigpipe)
_dbus_disable_sigpipe ();
connection->timeouts = timeout_list;
connection->handler_table = handler_table;
connection->pending_replies = pending_replies;
+ connection->outgoing_counter = outgoing_counter;
connection->filter_list = NULL;
connection->last_dispatch_status = DBUS_DISPATCH_COMPLETE; /* so we're notified first time there's data */
if (timeout_list)
_dbus_timeout_list_free (timeout_list);
+
+ if (outgoing_counter)
+ _dbus_counter_unref (outgoing_counter);
return NULL;
}
dbus_mutex_unlock (connection->mutex);
}
-/**
- * Adds the counter used to count the number of open connections.
- * Increments the counter by one, and saves it to be decremented
- * again when this connection is finalized.
- *
- * @param connection a #DBusConnection
- * @param counter counter that tracks number of connections
- */
-void
-_dbus_connection_set_connection_counter (DBusConnection *connection,
- DBusCounter *counter)
-{
- _dbus_assert (connection->connection_counter == NULL);
-
- connection->connection_counter = counter;
- _dbus_counter_ref (connection->connection_counter);
- _dbus_counter_adjust (connection->connection_counter, 1);
-}
-
/** @} */
/**
connection->refcount += 1;
}
+static void
+free_outgoing_message (void *element,
+ void *data)
+{
+ DBusMessage *message = element;
+ DBusConnection *connection = data;
+
+ _dbus_message_remove_size_counter (message,
+ connection->outgoing_counter);
+ dbus_message_unref (message);
+}
/* This is run without the mutex held, but after the last reference
* to the connection has been dropped we should have no thread-related
* you won't get the disconnected message.
*/
_dbus_assert (!_dbus_transport_get_is_connected (connection->transport));
-
- if (connection->connection_counter != NULL)
- {
- /* subtract ourselves from the counter */
- _dbus_counter_adjust (connection->connection_counter, - 1);
- _dbus_counter_unref (connection->connection_counter);
- connection->connection_counter = NULL;
- }
/* ---- We're going to call various application callbacks here, hope it doesn't break anything... */
dbus_connection_set_dispatch_status_function (connection, NULL, NULL, NULL);
_dbus_list_clear (&connection->filter_list);
_dbus_list_foreach (&connection->outgoing_messages,
- (DBusForeachFunction) dbus_message_unref,
- NULL);
+ free_outgoing_message,
+ connection);
_dbus_list_clear (&connection->outgoing_messages);
_dbus_list_foreach (&connection->incoming_messages,
(DBusForeachFunction) dbus_message_unref,
NULL);
_dbus_list_clear (&connection->incoming_messages);
+
+ _dbus_counter_unref (connection->outgoing_counter);
_dbus_transport_unref (connection->transport);
return res;
}
+struct DBusPreallocatedSend
+{
+ DBusConnection *connection;
+ DBusList *queue_link;
+ DBusList *counter_link;
+};
+
+
/**
* Preallocates resources needed to send a message, allowing the message
* to be sent without the possibility of memory allocation failure.
DBusPreallocatedSend*
dbus_connection_preallocate_send (DBusConnection *connection)
{
- /* we store "connection" in the link just to enforce via
- * assertion that preallocated links are only used
- * with the connection they were created for.
- */
- return (DBusPreallocatedSend*) _dbus_list_alloc_link (connection);
+ DBusPreallocatedSend *preallocated;
+
+ preallocated = dbus_new (DBusPreallocatedSend, 1);
+ if (preallocated == NULL)
+ return NULL;
+
+ preallocated->queue_link = _dbus_list_alloc_link (NULL);
+ if (preallocated->queue_link == NULL)
+ goto failed_0;
+
+ preallocated->counter_link = _dbus_list_alloc_link (connection->outgoing_counter);
+ if (preallocated->counter_link == NULL)
+ goto failed_1;
+
+ _dbus_counter_ref (preallocated->counter_link->data);
+
+ preallocated->connection = connection;
+
+ return preallocated;
+
+ failed_1:
+ _dbus_list_free_link (preallocated->queue_link);
+ failed_0:
+ dbus_free (preallocated);
+
+ return NULL;
}
/**
dbus_connection_free_preallocated_send (DBusConnection *connection,
DBusPreallocatedSend *preallocated)
{
- DBusList *link = (DBusList*) preallocated;
- _dbus_assert (link->data == connection);
- _dbus_list_free_link (link);
+ _dbus_assert (connection == preallocated->connection);
+
+ _dbus_list_free_link (preallocated->queue_link);
+ _dbus_counter_unref (preallocated->counter_link->data);
+ _dbus_list_free_link (preallocated->counter_link);
+ dbus_free (preallocated);
}
/**
DBusMessage *message,
dbus_int32_t *client_serial)
{
- DBusList *link = (DBusList*) preallocated;
dbus_int32_t serial;
- _dbus_assert (link->data == connection);
+ _dbus_assert (preallocated->connection == connection);
_dbus_assert (dbus_message_get_name (message) != NULL);
dbus_mutex_lock (connection->mutex);
- link->data = message;
+ preallocated->queue_link->data = message;
_dbus_list_prepend_link (&connection->outgoing_messages,
- link);
+ preallocated->queue_link);
+ _dbus_message_add_size_counter_link (message,
+ preallocated->counter_link);
+
+ dbus_free (preallocated);
+ preallocated = NULL;
+
dbus_message_ref (message);
+
connection->n_outgoing += 1;
_dbus_verbose ("Message %p (%s) added to outgoing queue %p, %d pending to send\n",
* @param size the maximum size in bytes of all outstanding messages
*/
void
-dbus_connection_set_max_live_messages_size (DBusConnection *connection,
- long size)
+dbus_connection_set_max_received_size (DBusConnection *connection,
+ long size)
{
dbus_mutex_lock (connection->mutex);
- _dbus_transport_set_max_live_messages_size (connection->transport,
- size);
+ _dbus_transport_set_max_received_size (connection->transport,
+ size);
dbus_mutex_unlock (connection->mutex);
}
/**
- * Gets the value set by dbus_connection_set_max_live_messages_size().
+ * Gets the value set by dbus_connection_set_max_received_size().
*
* @param connection the connection
* @returns the max size of all live messages
*/
long
-dbus_connection_get_max_live_messages_size (DBusConnection *connection)
+dbus_connection_get_max_received_size (DBusConnection *connection)
+{
+ long res;
+ dbus_mutex_lock (connection->mutex);
+ res = _dbus_transport_get_max_received_size (connection->transport);
+ dbus_mutex_unlock (connection->mutex);
+ return res;
+}
+
+/**
+ * Gets the approximate size in bytes of all messages in the outgoing
+ * message queue. The size is approximate in that you shouldn't use
+ * it to decide how many bytes to read off the network or anything
+ * of that nature, as optimizations may choose to tell small white lies
+ * to avoid performance overhead.
+ *
+ * @param connection the connection
+ * @returns the number of bytes that have been queued up but not sent
+ */
+long
+dbus_connection_get_outgoing_size (DBusConnection *connection)
{
long res;
dbus_mutex_lock (connection->mutex);
- res = _dbus_transport_get_max_live_messages_size (connection->transport);
+ res = _dbus_counter_get_value (connection->outgoing_counter);
dbus_mutex_unlock (connection->mutex);
return res;
}
void dbus_connection_set_max_message_size (DBusConnection *connection,
long size);
long dbus_connection_get_max_message_size (DBusConnection *connection);
-void dbus_connection_set_max_live_messages_size (DBusConnection *connection,
+void dbus_connection_set_max_received_size (DBusConnection *connection,
long size);
-long dbus_connection_get_max_live_messages_size (DBusConnection *connection);
+long dbus_connection_get_max_received_size (DBusConnection *connection);
+long dbus_connection_get_outgoing_size (DBusConnection *connection);
DBusPreallocatedSend* dbus_connection_preallocate_send (DBusConnection *connection);
void dbus_connection_free_preallocated_send (DBusConnection *connection,
const DBusString **header,
const DBusString **body);
-void _dbus_message_lock (DBusMessage *message);
-void _dbus_message_unlock (DBusMessage *message);
-void _dbus_message_set_serial (DBusMessage *message,
- dbus_int32_t serial);
-void _dbus_message_add_size_counter (DBusMessage *message,
- DBusCounter *counter);
+void _dbus_message_lock (DBusMessage *message);
+void _dbus_message_unlock (DBusMessage *message);
+void _dbus_message_set_serial (DBusMessage *message,
+ dbus_int32_t serial);
+dbus_bool_t _dbus_message_add_size_counter (DBusMessage *message,
+ DBusCounter *counter);
+void _dbus_message_add_size_counter_link (DBusMessage *message,
+ DBusList *link);
+void _dbus_message_remove_size_counter (DBusMessage *message,
+ DBusCounter *counter);
DBusMessageLoader* _dbus_message_loader_new (void);
void _dbus_message_loader_ref (DBusMessageLoader *loader);
DBusMessage* _dbus_message_loader_peek_message (DBusMessageLoader *loader);
DBusMessage* _dbus_message_loader_pop_message (DBusMessageLoader *loader);
DBusList* _dbus_message_loader_pop_message_link (DBusMessageLoader *loader);
+void _dbus_message_loader_putback_message_link (DBusMessageLoader *loader,
+ DBusList *link);
dbus_bool_t _dbus_message_loader_get_is_corrupted (DBusMessageLoader *loader);
char byte_order; /**< Message byte order. */
- DBusCounter *size_counter; /**< Counter for the size of the message, or #NULL */
- long size_counter_delta; /**< Size we incremented the size counter by. */
+ DBusList *size_counters; /**< 0-N DBusCounter used to track message size. */
+ long size_counter_delta; /**< Size we incremented the size counters by. */
dbus_uint32_t changed_stamp;
* Adds a counter to be incremented immediately with the
* size of this message, and decremented by the size
* of this message when this message if finalized.
+ * The link contains a counter with its refcount already
+ * incremented, but the counter itself not incremented.
+ * Ownership of link and counter refcount is passed to
+ * the message.
*
* @param message the message
- * @param counter the counter
+ * @param link link with counter as data
*/
void
+_dbus_message_add_size_counter_link (DBusMessage *message,
+ DBusList *link)
+{
+ /* right now we don't recompute the delta when message
+ * size changes, and that's OK for current purposes
+ * I think, but could be important to change later.
+ * Do recompute it whenever there are no outstanding counters,
+ * since it's basically free.
+ */
+ if (message->size_counters == NULL)
+ {
+ message->size_counter_delta =
+ _dbus_string_get_length (&message->header) +
+ _dbus_string_get_length (&message->body);
+
+#if 0
+ _dbus_verbose ("message has size %ld\n",
+ message->size_counter_delta);
+#endif
+ }
+
+ _dbus_list_append_link (&message->size_counters, link);
+
+ _dbus_counter_adjust (link->data, message->size_counter_delta);
+}
+
+/**
+ * Adds a counter to be incremented immediately with the
+ * size of this message, and decremented by the size
+ * of this message when this message if finalized.
+ *
+ * @param message the message
+ * @param counter the counter
+ * @returns #FALSE if no memory
+ */
+dbus_bool_t
_dbus_message_add_size_counter (DBusMessage *message,
DBusCounter *counter)
{
- _dbus_assert (message->size_counter == NULL); /* If this fails we may need to keep a list of
- * counters instead of just one
- */
+ DBusList *link;
- message->size_counter = counter;
- _dbus_counter_ref (message->size_counter);
+ link = _dbus_list_alloc_link (counter);
+ if (link == NULL)
+ return FALSE;
- /* When we can change message size, we may want to
- * update this each time we do so, or we may want to
- * just KISS like this.
- */
- message->size_counter_delta =
- _dbus_string_get_length (&message->header) +
- _dbus_string_get_length (&message->body);
+ _dbus_counter_ref (counter);
+ _dbus_message_add_size_counter_link (message, link);
-#if 0
- _dbus_verbose ("message has size %ld\n",
- message->size_counter_delta);
-#endif
-
- _dbus_counter_adjust (message->size_counter, message->size_counter_delta);
+ return TRUE;
+}
+
+/**
+ * Removes a counter tracking the size of this message, and decrements
+ * the counter by the size of this message.
+ *
+ * @param message the message
+ * @param counter the counter
+ */
+void
+_dbus_message_remove_size_counter (DBusMessage *message,
+ DBusCounter *counter)
+{
+ if (!_dbus_list_remove_last (&message->size_counters,
+ counter))
+ _dbus_assert_not_reached ("Removed a message size counter that was not added");
+
+ _dbus_counter_adjust (counter, message->size_counter_delta);
+
+ _dbus_counter_unref (counter);
}
static dbus_bool_t
_dbus_assert (refcount > 1);
}
+static void
+free_size_counter (void *element,
+ void *data)
+{
+ DBusCounter *counter = element;
+ DBusMessage *message = data;
+
+ _dbus_counter_adjust (counter, - message->size_counter_delta);
+
+ _dbus_counter_unref (counter);
+}
+
/**
* Decrements the reference count of a DBusMessage.
*
if (refcount == 0)
{
- if (message->size_counter != NULL)
- {
- _dbus_counter_adjust (message->size_counter,
- - message->size_counter_delta);
- _dbus_counter_unref (message->size_counter);
- }
+ _dbus_list_foreach (&message->size_counters,
+ free_size_counter, message);
+ _dbus_list_clear (&message->size_counters);
_dbus_string_free (&message->header);
_dbus_string_free (&message->body);
}
/**
+ * Returns a popped message link, used to undo a pop.
+ *
+ * @param loader the loader
+ * @param link the link with a message in it
+ */
+void
+_dbus_message_loader_putback_message_link (DBusMessageLoader *loader,
+ DBusList *link)
+{
+ _dbus_list_prepend_link (&loader->messages, link);
+}
+
+/**
* Checks whether the loader is confused due to bad data.
* If messages are received that are invalid, the
* loader gets confused and gives up permanently.
int refcount; /**< Reference count. */
const DBusServerVTable *vtable; /**< Virtual methods for this instance. */
DBusWatchList *watches; /**< Our watches */
- DBusTimeoutList *timeouts; /**< Our timeouts */
-
- DBusCounter *connection_counter; /**< Number of non-finalized DBusConnection
- * to this server
- */
+ DBusTimeoutList *timeouts; /**< Our timeouts */
char *address; /**< Address this server is listening on. */
if (connection == NULL)
return FALSE;
-
- _dbus_connection_set_connection_counter (connection,
- server->connection_counter);
/* See if someone wants to handle this new connection,
* self-referencing for paranoia
server->address = NULL;
server->watches = NULL;
server->timeouts = NULL;
- server->connection_counter = NULL;
if (!_dbus_string_copy_data (address, &server->address))
goto failed;
server->timeouts = _dbus_timeout_list_new ();
if (server->timeouts == NULL)
goto failed;
-
- server->connection_counter = _dbus_counter_new ();
- if (server->connection_counter == NULL)
- goto failed;
_dbus_data_slot_list_init (&server->slot_list);
_dbus_timeout_list_free (server->timeouts);
server->timeouts = NULL;
}
- if (server->connection_counter)
- {
- _dbus_counter_unref (server->connection_counter);
- server->connection_counter = NULL;
- }
if (server->address)
{
dbus_free (server->address);
_dbus_watch_list_free (server->watches);
_dbus_timeout_list_free (server->timeouts);
- _dbus_counter_unref (server->connection_counter);
dbus_free (server->address);
_dbus_verbose ("queueing received message %p\n", message);
- _dbus_message_add_size_counter (message, transport->live_messages_size);
-
- /* pass ownership of link and message ref to connection */
- _dbus_connection_queue_received_message_link (transport->connection,
- link);
+ if (!_dbus_message_add_size_counter (message, transport->live_messages_size))
+ {
+ _dbus_message_loader_putback_message_link (transport->loader,
+ link);
+ status = DBUS_DISPATCH_NEED_MEMORY;
+ break;
+ }
+ else
+ {
+ /* pass ownership of link and message ref to connection */
+ _dbus_connection_queue_received_message_link (transport->connection,
+ link);
+ }
}
if (_dbus_message_loader_get_is_corrupted (transport->loader))
}
/**
- * See dbus_connection_set_max_live_messages_size().
+ * See dbus_connection_set_max_received_size().
*
* @param transport the transport
* @param size the max size of all incoming messages
*/
void
-_dbus_transport_set_max_live_messages_size (DBusTransport *transport,
- long size)
+_dbus_transport_set_max_received_size (DBusTransport *transport,
+ long size)
{
transport->max_live_messages_size = size;
_dbus_counter_set_notify (transport->live_messages_size,
/**
- * See dbus_connection_get_max_live_messages_size().
+ * See dbus_connection_get_max_received_size().
*
* @param transport the transport
* @returns max bytes for all live messages
*/
long
-_dbus_transport_get_max_live_messages_size (DBusTransport *transport)
+_dbus_transport_get_max_received_size (DBusTransport *transport)
{
return transport->max_live_messages_size;
}
typedef struct DBusTransport DBusTransport;
-DBusTransport* _dbus_transport_open (const char *address,
- DBusError *error);
-void _dbus_transport_ref (DBusTransport *transport);
-void _dbus_transport_unref (DBusTransport *transport);
-void _dbus_transport_disconnect (DBusTransport *transport);
-dbus_bool_t _dbus_transport_get_is_connected (DBusTransport *transport);
-dbus_bool_t _dbus_transport_get_is_authenticated (DBusTransport *transport);
-const char* _dbus_transport_get_address (DBusTransport *transport);
-dbus_bool_t _dbus_transport_handle_watch (DBusTransport *transport,
- DBusWatch *watch,
- unsigned int condition);
-dbus_bool_t _dbus_transport_set_connection (DBusTransport *transport,
- DBusConnection *connection);
-void _dbus_transport_messages_pending (DBusTransport *transport,
- int queue_length);
-void _dbus_transport_do_iteration (DBusTransport *transport,
- unsigned int flags,
- int timeout_milliseconds);
-DBusDispatchStatus _dbus_transport_get_dispatch_status (DBusTransport *transport);
-dbus_bool_t _dbus_transport_queue_messages (DBusTransport *transport);
-void _dbus_transport_set_max_message_size (DBusTransport *transport,
- long size);
-long _dbus_transport_get_max_message_size (DBusTransport *transport);
-void _dbus_transport_set_max_live_messages_size (DBusTransport *transport,
- long size);
-long _dbus_transport_get_max_live_messages_size (DBusTransport *transport);
-dbus_bool_t _dbus_transport_get_unix_user (DBusTransport *transport,
- unsigned long *uid);
-void _dbus_transport_set_unix_user_function (DBusTransport *transport,
- DBusAllowUnixUserFunction function,
- void *data,
- DBusFreeFunction free_data_function,
- void **old_data,
- DBusFreeFunction *old_free_data_function);
-dbus_bool_t _dbus_transport_set_auth_mechanisms (DBusTransport *transport,
- const char **mechanisms);
+DBusTransport* _dbus_transport_open (const char *address,
+ DBusError *error);
+void _dbus_transport_ref (DBusTransport *transport);
+void _dbus_transport_unref (DBusTransport *transport);
+void _dbus_transport_disconnect (DBusTransport *transport);
+dbus_bool_t _dbus_transport_get_is_connected (DBusTransport *transport);
+dbus_bool_t _dbus_transport_get_is_authenticated (DBusTransport *transport);
+const char* _dbus_transport_get_address (DBusTransport *transport);
+dbus_bool_t _dbus_transport_handle_watch (DBusTransport *transport,
+ DBusWatch *watch,
+ unsigned int condition);
+dbus_bool_t _dbus_transport_set_connection (DBusTransport *transport,
+ DBusConnection *connection);
+void _dbus_transport_messages_pending (DBusTransport *transport,
+ int queue_length);
+void _dbus_transport_do_iteration (DBusTransport *transport,
+ unsigned int flags,
+ int timeout_milliseconds);
+DBusDispatchStatus _dbus_transport_get_dispatch_status (DBusTransport *transport);
+dbus_bool_t _dbus_transport_queue_messages (DBusTransport *transport);
+void _dbus_transport_set_max_message_size (DBusTransport *transport,
+ long size);
+long _dbus_transport_get_max_message_size (DBusTransport *transport);
+void _dbus_transport_set_max_received_size (DBusTransport *transport,
+ long size);
+long _dbus_transport_get_max_received_size (DBusTransport *transport);
+dbus_bool_t _dbus_transport_get_unix_user (DBusTransport *transport,
+ unsigned long *uid);
+void _dbus_transport_set_unix_user_function (DBusTransport *transport,
+ DBusAllowUnixUserFunction function,
+ void *data,
+ DBusFreeFunction free_data_function,
+ void **old_data,
+ DBusFreeFunction *old_free_data_function);
+dbus_bool_t _dbus_transport_set_auth_mechanisms (DBusTransport *transport,
+ const char **mechanisms);
+
projects / platform / upstream / dbus.git / commitdiff