浅析dbus-session后台daemon如何处理client们发送过来的mesage
《浅析应用程序dbus_bus_get函数在lib库中的实际执行流程》应用程序首先调用dbus_bus_get()于dbus-session建立连接
dbus_bus_register
==>message = dbus_message_new_method_call (DBUS_SERVICE_DBUS,
DBUS_PATH_DBUS,
DBUS_INTERFACE_DBUS,
"Hello");
// 等待peer发送reply回来
==>reply = dbus_connection_send_with_reply_and_block (connection, message, -1, error);
==>dbus_message_get_args (reply, error,
DBUS_TYPE_STRING, &name, // 获取peer返回回来的unique_name数据
DBUS_TYPE_INVALID);
这时dbus-session后台daemon发现了该应用程序打算与自己建立连接,随后触发socket_handle_watch()
回调函数,为该应用程序建立该应用程序对应的transport和connection存储体[luther.gliethttp]
socket_handle_watch
==>client_fd = _dbus_accept (listen_fd);
==>handle_new_client_fd_and_unlock (server, client_fd);
==*>为client的连接创建transport,该transport的auth_mechanisms等于server->auth_mechanisms,
然后在transport之上继续创建connection,最终位于server中的为了该client建立的独立transport,
也将执行与应用程序端建立的transport同样的_dbus_connection_handle_watch()数据处理函数.
下面是daemon程序dbus-session对来自网络上client们的message进行细化处理的整体代码流程.[luther.gliethttp]
main
==>_dbus_loop_run
==*>_dbus_loop_iterate
1.添加loop->callbacks中DBUS_WATCH_READABLE﹑DBUS_WATCH_WRITABLE和TIMEOUT_CALLBACK到poll的fds中
2.n_ready = _dbus_poll (fds, n_fds, timeout); //等待众多事件之一发生[luther.gliethttp]
3.如果timeout了,那么调用timeout的callback回调函数,如果数据来了,比如:_dbus_connection_handle_watch.
接下来看看_dbus_connection_handle_watch是如何处理来自网络的数据的[luther.gliethttp]
_dbus_connection_handle_watch
==>_dbus_transport_handle_watch (connection->transport, watch, condition);
==>(* transport->vtable->handle_watch) (transport, watch, condition);
即:socket_vtable.socket_handle_watch()处理函数.
==>socket_handle_watch
如果为read,则
==*>do_reading (transport)
将数据读取到transport->loader->data字符串存储空间中.
==**>_dbus_transport_queue_messages
dbus_bool_t
_dbus_transport_queue_messages (DBusTransport *transport)
{
DBusDispatchStatus status;
#if 0
_dbus_verbose ("_dbus_transport_queue_messages()\n");
#endif
/* Queue any messages */
// 将零碎的data片段数据,慢慢堆积到一个message大小,如果已经接收到该message所需的len长度数据
// 即生成了一封完整的message,
// 那么_dbus_transport_get_dispatch_status将始终返回DBUS_DISPATCH_DATA_REMAINS,源码见下面.
while ((status = _dbus_transport_get_dispatch_status (transport)) == DBUS_DISPATCH_DATA_REMAINS)
{
DBusMessage *message;
DBusList *link;
// 从网络上收到了一封完整的message数据.
link = _dbus_message_loader_pop_message_link (transport->loader); // 从transport->loader弹出该封完整的message
_dbus_assert (link != NULL);
message = link->data;
_dbus_verbose ("queueing received message %p\n", message);
if (!_dbus_message_add_size_counter (message, transport->live_messages_size))
// 申请
// link->data->value += message->size_counter_delta;这里link->data指向transport->live_messages_size.
// 所以最终等效于transport->live_messages_size->value += message->size_counter_delta;
// 然后将link添加到message->size_counters链表中,这样作的目的在于,
// 当message处理完毕之后,message释放函数_dbus_message_remove_size_counter将调用
// _dbus_counter_adjust (counter, - message->size_counter_delta);
// 这样当message处理完毕之后,transport->live_messages_size的计数将再减去message->size_counter_delta
// 即: transport->live_messages_size->value -= message->size_counter_delta;[luther.gliethttp]
{
_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 */
// 添加到connection->incoming_messages链表中,源码见下面[luther.gliethttp]
_dbus_connection_queue_received_message_link (transport->connection,
link);
}
}
if (_dbus_message_loader_get_is_corrupted (transport->loader))
{
_dbus_verbose ("Corrupted message stream, disconnecting\n");
_dbus_transport_disconnect (transport);
}
return status != DBUS_DISPATCH_NEED_MEMORY;
}
void
_dbus_connection_queue_received_message_link (DBusConnection *connection,
DBusList *link)
{
DBusPendingCall *pending;
dbus_uint32_t reply_serial;
DBusMessage *message;
_dbus_assert (_dbus_transport_get_is_authenticated (connection->transport));
_dbus_list_append_link (&connection->incoming_messages,
link); // 将link添加到connection->incoming_messages中,link->data就等于message.
message = link->data;
/* If this is a reply we're waiting on, remove timeout for it */
reply_serial = dbus_message_get_reply_serial (message); // 获取刚刚接收到的message的serial id号.
if (reply_serial != 0)
{
pending = _dbus_hash_table_lookup_int (connection->pending_replies,
reply_serial);
// 查看pending_replies哈西表中是否有等待以该serial id号作为reply回应数据的pending操作.
if (pending != NULL)
{
if (_dbus_pending_call_is_timeout_added_unlocked (pending))
// 如果添加了timeout到pending中,那么移除它[luther.gliethttp]
_dbus_connection_remove_timeout_unlocked (connection,
_dbus_pending_call_get_timeout_unlocked (pending));
_dbus_pending_call_set_timeout_added_unlocked (pending, FALSE); // 修改状态为timeout未添加[luther.gliethttp]
}
}
connection->n_incoming += 1; // 来自网络的报文个数加1.
_dbus_connection_wakeup_mainloop (connection); // 唤醒主loop
}
DBusDispatchStatus
_dbus_transport_get_dispatch_status (DBusTransport *transport)
{
if (_dbus_counter_get_value (transport->live_messages_size) >= transport->max_live_messages_size)
return DBUS_DISPATCH_COMPLETE; /* complete for now */
if (!_dbus_transport_get_is_authenticated (transport))
{
if (_dbus_auth_do_work (transport->auth) ==
DBUS_AUTH_STATE_WAITING_FOR_MEMORY)
return DBUS_DISPATCH_NEED_MEMORY;
else if (!_dbus_transport_get_is_authenticated (transport))
return DBUS_DISPATCH_COMPLETE;
}
if (!transport->unused_bytes_recovered &&
!recover_unused_bytes (transport))
return DBUS_DISPATCH_NEED_MEMORY;
transport->unused_bytes_recovered = TRUE;
if (!_dbus_message_loader_queue_messages (transport->loader))
// 尝试从网络数据区transport->loader->data组建一封完整message,源码见下面
return DBUS_DISPATCH_NEED_MEMORY;
if (_dbus_message_loader_peek_message (transport->loader) != NULL) // 查看loader->messages链表是否有完整的message了.
return DBUS_DISPATCH_DATA_REMAINS;
else
return DBUS_DISPATCH_COMPLETE;
}
dbus_bool_t
_dbus_message_loader_queue_messages (DBusMessageLoader *loader)
{
while (!loader->corrupted &&
_dbus_string_get_length (&loader->data) >= DBUS_MINIMUM_HEADER_SIZE) // 当前网络接收到的数据至少保证存放头部控制信息的16字节
{
DBusValidity validity;
int byte_order, fields_array_len, header_len, body_len;
if (_dbus_header_have_message_untrusted (loader->max_message_size,
&validity,
&byte_order,
&fields_array_len,
&header_len,
&body_len,
&loader->data, 0,
_dbus_string_get_length (&loader->data)))
// 当前网络接收到的数据个数如果已经超过了该message下header+body_len长度,那么说明该message已经完整接收
// 所以执行下面的函数进行message重建,最后将重建完成的有效的message追加到loader->messages链表中[luther.gliethttp]
{
DBusMessage *message;
_dbus_assert (validity == DBUS_VALID);
message = dbus_message_new_empty_header ();
if (message == NULL)
return FALSE;
// 进行message重建,最后将重建完成的有效的message追加到loader->messages链表中[luther.gliethttp]
if (!load_message (loader, message,
byte_order, fields_array_len,
header_len, body_len))
{
dbus_message_unref (message);
/* load_message() returns false if corrupted or OOM; if
* corrupted then return TRUE for not OOM
*/
return loader->corrupted;
}
_dbus_assert (loader->messages != NULL);
_dbus_assert (_dbus_list_find_last (&loader->messages, message) != NULL);
}
else
{
_dbus_verbose ("Initial peek at header says we don't have a whole message yet, or data broken with invalid code %d\n",
validity);
if (validity != DBUS_VALID)
{
loader->corrupted = TRUE;
loader->corruption_reason = validity;
}
return TRUE;
}
}
return TRUE;
}
// 上面_dbus_connection_handle_watch将调用
// dispatch_status_function将connection添加到loop->need_dispatch中
_dbus_connection_update_dispatch_status_and_unlock
==>connection->dispatch_status_function
connection->dispatch_status_function = dispatch_status_function;
server->new_connection_function = bus_connections_setup_connection; // 当有新连接时socket_handle_watch将添加新连接对应的transport和对应的connection,这时将调用server->new_connection_function()回调函数.
dispatch_status_function或者bus_connections_setup_connection
==>_dbus_loop_queue_dispatch
==>_dbus_list_append (&loop->need_dispatch, connection)
好了上面_dbus_connection_handle_watch()回调处理函数,已经从网络数据中解析出一封message消息,接下来就是调用
_dbus_loop_dispatch()函数进行message细分处理了[luther.gliethttp].
==>_dbus_loop_dispatch
==*>DBusConnection *connection = _dbus_list_pop_first (&loop->need_dispatch);
dbus_connection_dispatch(connection); // 处理该connection接收到的incoming_messages数据包[luther.gliethttp].
==**>message_link = _dbus_connection_pop_message_link_unlocked (connection); // 从incoming_messages弹出一个message.
reply_serial = dbus_message_get_reply_serial (message); // 读取message的serial id号
// 查找是否pending_replies中有人正在等待该serial id号消息到来[luther.gliethttp]
pending = _dbus_hash_table_lookup_int (connection->pending_replies,
reply_serial);
// 如果有pending,即正在等待reply消息返回,那么执行
// complete_pending_call_and_unlock (connection, pending, message); // 设置pending->reply = message;
// 然后goto out;完成退出.
// 否则继续执行下面的代码[luther.gliethttp]
// 先检查该message是否可被builtin的东西处理,主要是处理interface为"org.freedesktop.DBus.Peer",
// 同时method为"Ping"或"GetMachineId"的消息.
result = _dbus_connection_run_builtin_filters_unlocked_no_update (connection, message);
// 接下来执行connection->filter_list链表上的所有过滤函数[luther.gliethttp]
// 首先,_dbus_message_filter_ref增减引用计数filter->refcount++;
_dbus_list_foreach (&filter_list_copy,
(DBusForeachFunction)_dbus_message_filter_ref,
NULL);
// 接下来执行 connection->filter_list链表上的所有过滤函数
// 所有connection->filter_list链表添加都是由dbus_connection_add_filter()完成.
// 只有一个地方调用了dbus_connection_add_filter()函数,就是bus_dispatch_add_connection()
// 它设置的回调函数为bus_dispatch_message_filter();
//
// 整个流程是这样的
// server_watch_callback
// DBUS_SESSION_BUS_ADDRESS=unix:abstract=/tmp/dbus-0jpvwWTzEb,guid=e4295f4ec1c581bb5a971bfb4a401c5e
// 该unix通道有client尝试连接,将执行下面的一系列函数.
// ==>socket_handle_watch
// ==*>handle_new_client_fd_and_unlock
// ==**>server->new_connection_function = new_connection_callback;
// ==***>new_connection_callback
// ==****>bus_connections_setup_connection
// ==*****>bus_dispatch_add_connection (connection)
// ==******>_dbus_list_append (&connection->filter_list, bus_dispatch_message_filter);
(* filter->function) (connection, message, filter->user_data);
// 这之后数据处理工作就算完成了.
接下来我们就来看看唯一的一个filter_func函数bus_dispatch_message_filter()[luther.gliethttp]
bus_dispatch_message_filter
==>bus_dispatch
// 如果应用程序发送了
// dbus_message_new_method_call (DBUS_SERVICE_DBUS, DBUS_PATH_DBUS, DBUS_INTERFACE_DBUS, "Hello");
service_name = dbus_message_get_destination (message); // 获取目的路径DBUS_SERVICE_DBUS
transaction = bus_transaction_new (context); // 创建一个transaction操作片
// 1.如果service_name等于DBUS_SERVICE_DBUS那么
bus_context_check_security_policy(context, transaction,
connection, NULL, NULL, message, &error); // 做安全检查
bus_driver_handle_message (connection, transaction, message, &error)
// 交由bus总线驱动处理该message[luther.gliethttp]
// 2.如果service_name非DBUS_SERVICE_DBUS,那么dbus-session将作route路由该message消息[luther.gliethttp]
bus_registry_lookup (registry, &service_string);
// 当安全校验全部通过之后,将该transaction操作片route路由到addressed_recipient连接对应的transport上[luther.gliethttp]
bus_transaction_send (transaction, addressed_recipient, message);
我们现在关心的只是应用程序启动注册部分,所以我们下面将着重讨论的是bus_driver_handle_message--这个bus总线
相应interface接口下method方法处理集[luther.gliethttp]
==>bus_driver_handle_message
static struct
{
const char *name;
const char *in_args;
const char *out_args;
dbus_bool_t (* handler) (DBusConnection *connection,
BusTransaction *transaction,
DBusMessage *message,
DBusError *error);
} message_handlers[] = {
{ "Hello",
"",
DBUS_TYPE_STRING_AS_STRING,
bus_driver_handle_hello },
......
};
==>bus_driver_handle_hello
static dbus_bool_t
bus_driver_handle_hello (DBusConnection *connection,
BusTransaction *transaction,
DBusMessage *message,
DBusError *error)
{
DBusString unique_name;
BusService *service;
dbus_bool_t retval;
BusRegistry *registry;
BusConnections *connections;
_DBUS_ASSERT_ERROR_IS_CLEAR (error);
if (bus_connection_is_active (connection))
{
/* We already handled an Hello message for this connection. */
dbus_set_error (error, DBUS_ERROR_FAILED,
"Already handled an Hello message");
return FALSE;
}
/* Note that when these limits are exceeded we don't disconnect the
* connection; we just sort of leave it hanging there until it times
* out or disconnects itself or is dropped due to the max number of
* incomplete connections. It's even OK if the connection wants to
* retry the hello message, we support that.
*/
connections = bus_connection_get_connections (connection);
if (!bus_connections_check_limits (connections, connection,
error))
{
_DBUS_ASSERT_ERROR_IS_SET (error);
return FALSE;
}
if (!_dbus_string_init (&unique_name))
{
BUS_SET_OOM (error);
return FALSE;
}
retval = FALSE;
registry = bus_connection_get_registry (connection);
// 为连接到该dbus-session daemon的应用程序创建unique_name, 源码见下面
// unique_name将等于":1
.12345678"之类[luther.gliethttp]
if (!create_unique_client_name (registry, &unique_name))
{
BUS_SET_OOM (error);
goto out_0;
}
// d = BUS_CONNECTION_DATA (connection);
// 设置_dbus_string_copy_data (name, &d->name)
// 即:connection->slot_list[connection_data_slot]->name = ":1
.12345678";
// 设置 policy
// d->policy = bus_context_create_client_policy (d->connections->context, connection, error);
if (!bus_connection_complete (connection, &unique_name, error))
{
_DBUS_ASSERT_ERROR_IS_SET (error);
goto out_0;
}
// 设置sender域为":1
.12345678"
if (!dbus_message_set_sender (message,
bus_connection_get_name (connection)))
{
BUS_SET_OOM (error);
goto out_0;
}
// 发送带有参数unique_name的message给应用程序,应用程序正等待该unique_name参数到来[luther.gliethttp]
// 源码见下面
if (!bus_driver_send_welcome_message (connection, message, transaction, error))
goto out_0;
/* Create the service */
service = bus_registry_ensure (registry,
&unique_name, connection, 0, transaction, error);
if (service == NULL)
goto out_0;
_dbus_assert (bus_connection_is_active (connection));
retval = TRUE;
out_0:
_dbus_string_free (&unique_name);
return retval;
}
static dbus_bool_t
create_unique_client_name (BusRegistry *registry,
DBusString *str)
{
/* We never want to use the same unique client name twice, because
* we want to guarantee that if you send a message to a given unique
* name, you always get the same application. So we use two numbers
* for INT_MAX * INT_MAX combinations, should be pretty safe against
* wraparound.
*/
/* FIXME these should be in BusRegistry rather than static vars */
static int next_major_number = 0;
static int next_minor_number = 0;
int len;
len = _dbus_string_get_length (str); // 这里len等于0
while (TRUE)
{
/* start out with 1-0, go to 1-1, 1-2, 1-3,
* up to 1-MAXINT, then 2-0, 2-1, etc.
*/
if (next_minor_number <= 0) // minor大到了0x7fffffff,那么major加1,minor归0,进入下一轮.
{
next_major_number += 1;
next_minor_number = 0;
if (next_major_number <= 0)
_dbus_assert_not_reached ("INT_MAX * INT_MAX clients were added");
}
_dbus_assert (next_major_number > 0);
_dbus_assert (next_minor_number >= 0);
/* appname:MAJOR-MINOR */
if (!_dbus_string_append (str, ":"))
return FALSE;
if (!_dbus_string_append_int (str, next_major_number))
return FALSE;
if (!_dbus_string_append (str, "."))
return FALSE;
if (!_dbus_string_append_int (str, next_minor_number))
return FALSE;
next_minor_number += 1;
/* Check if a client with the name exists */
// 务必确认该unique_name在所有连接中确实是唯一的,没有存在过的[luther.gliethttp]
if (bus_registry_lookup (registry, str) == NULL)
break;
/* drop the number again, try the next one. */
_dbus_string_set_length (str, len);
}
return TRUE;
}
static dbus_bool_t
bus_driver_send_welcome_message (DBusConnection *connection,
DBusMessage *hello_message,
BusTransaction *transaction,
DBusError *error)
{
DBusMessage *welcome;
const char *name;
_DBUS_ASSERT_ERROR_IS_CLEAR (error);
name = bus_connection_get_name (connection);
_dbus_assert (name != NULL);
welcome = dbus_message_new_method_return (hello_message); // 生成method方法调用返回message内存单元.
if (welcome == NULL)
{
BUS_SET_OOM (error);
return FALSE;
}
if (!dbus_message_append_args (welcome,
// 向该message追加参数,unique_name内容为":1
.12345678", 调用dbus_bus_get()库函数的应用程序正在
// 饥渴的等着unique_name到来呢[luther.gliethttp].
DBUS_TYPE_STRING, &name,
DBUS_TYPE_INVALID))
{
dbus_message_unref (welcome);
BUS_SET_OOM (error);
return FALSE;
}
_dbus_assert (dbus_message_has_signature (welcome, DBUS_TYPE_STRING_AS_STRING));
// 将数据从dbus-session这个daemon发送到调用dbus_bus_get()库函数的应用程序[luther.gliethttp]
// 在这里会做各种合法性检查,包括该connection是否允许send发送数据,是否允许recive接收数据,selinux安全等[lutehr.gliethttp]
if (!bus_transaction_send_from_driver (transaction, connection, welcome))
{
dbus_message_unref (welcome);
BUS_SET_OOM (error);
return FALSE;
}
else
{
dbus_message_unref (welcome);
return TRUE;
}
}
