From 9cf0e753f26488dfde60967ae0184cf986902464 Mon Sep 17 00:00:00 2001 From: lurchi Date: Fri, 24 Nov 2017 21:54:31 +0100 Subject: [PATCH] introduce part request / part ack messages (for leaving multicast groups) and related fixes --- src/include/gnunet_protocols.h | 5 +- src/multicast/gnunet-service-multicast.c | 208 ++++++++++++++--------- src/multicast/multicast_api.c | 51 +++--- 3 files changed, 159 insertions(+), 105 deletions(-) diff --git a/src/include/gnunet_protocols.h b/src/include/gnunet_protocols.h index 09341a242..9cfd00e39 100644 --- a/src/include/gnunet_protocols.h +++ b/src/include/gnunet_protocols.h @@ -2253,15 +2253,16 @@ extern "C" /** * A peer wants to part the group. */ -//#define GNUNET_MESSAGE_TYPE_MULTICAST_PART_REQUEST 754 +#define GNUNET_MESSAGE_TYPE_MULTICAST_PART_REQUEST 754 /** * Acknowledgement sent in response to a part request. * * Unicast message from a group member to the peer wanting to part. */ -//#define GNUNET_MESSAGE_TYPE_MULTICAST_PART_ACK 755 +#define GNUNET_MESSAGE_TYPE_MULTICAST_PART_ACK 755 +// FIXME: this is never used! /** * Group terminated. */ diff --git a/src/multicast/gnunet-service-multicast.c b/src/multicast/gnunet-service-multicast.c index 2f4dc8a14..11f3c08b7 100644 --- a/src/multicast/gnunet-service-multicast.c +++ b/src/multicast/gnunet-service-multicast.c @@ -553,25 +553,36 @@ client_send (struct GNUNET_SERVICE_Client *client, * Send message to all clients connected to the group. */ static void -client_send_group (const struct Group *grp, - const struct GNUNET_MessageHeader *msg) +client_send_group_keep_envelope (const struct Group *grp, + struct GNUNET_MQ_Envelope *env) { - GNUNET_log (GNUNET_ERROR_TYPE_INFO, - "%p Sending message to all clients of the group.\n", grp); + struct ClientList *cli = grp->clients_head; - struct ClientList *cl = grp->clients_head; - while (NULL != cl) + GNUNET_log (GNUNET_ERROR_TYPE_INFO, + "%p Sending message to all clients of the group.\n", + grp); + while (NULL != cli) { - struct GNUNET_MQ_Envelope * - env = GNUNET_MQ_msg_copy (msg); - - GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (cl->client), - env); - cl = cl->next; + GNUNET_MQ_send_copy (GNUNET_SERVICE_client_get_mq (cli->client), + env); + cli = cli->next; } } +/** + * Send message to all clients connected to the group and + * takes care of freeing @env. + */ +static void +client_send_group (const struct Group *grp, + struct GNUNET_MQ_Envelope *env) +{ + client_send_group_keep_envelope (grp, env); + GNUNET_MQ_discard (env); +} + + /** * Iterator callback for sending a message to origin clients. */ @@ -579,10 +590,10 @@ static int client_send_origin_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash, void *origin) { - const struct GNUNET_MessageHeader *msg = cls; + struct GNUNET_MQ_Envelope *env = cls; struct Member *orig = origin; - client_send_group (&orig->group, msg); + client_send_group_keep_envelope (&orig->group, env); return GNUNET_YES; } @@ -594,12 +605,12 @@ static int client_send_member_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash, void *member) { - const struct GNUNET_MessageHeader *msg = cls; + struct GNUNET_MQ_Envelope *env = cls; struct Member *mem = member; if (NULL != mem->join_dcsn) { /* Only send message to admitted members */ - client_send_group (&mem->group, msg); + client_send_group_keep_envelope (&mem->group, env); } return GNUNET_YES; } @@ -615,15 +626,16 @@ client_send_member_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash, */ static int client_send_all (struct GNUNET_HashCode *pub_key_hash, - const struct GNUNET_MessageHeader *msg) + struct GNUNET_MQ_Envelope *env) { int n = 0; n += GNUNET_CONTAINER_multihashmap_get_multiple (origins, pub_key_hash, client_send_origin_cb, - (void *) msg); + (void *) env); n += GNUNET_CONTAINER_multihashmap_get_multiple (members, pub_key_hash, client_send_member_cb, - (void *) msg); + (void *) env); + GNUNET_MQ_discard (env); return n; } @@ -636,14 +648,14 @@ client_send_all (struct GNUNET_HashCode *pub_key_hash, */ static int client_send_random (struct GNUNET_HashCode *pub_key_hash, - const struct GNUNET_MessageHeader *msg) + struct GNUNET_MQ_Envelope *env) { int n = 0; n = GNUNET_CONTAINER_multihashmap_get_random (origins, client_send_origin_cb, - (void *) msg); + (void *) env); if (n <= 0) n = GNUNET_CONTAINER_multihashmap_get_random (members, client_send_member_cb, - (void *) msg); + (void *) env); return n; } @@ -658,12 +670,12 @@ client_send_random (struct GNUNET_HashCode *pub_key_hash, */ static int client_send_origin (struct GNUNET_HashCode *pub_key_hash, - const struct GNUNET_MessageHeader *msg) + struct GNUNET_MQ_Envelope *env) { int n = 0; n += GNUNET_CONTAINER_multihashmap_get_multiple (origins, pub_key_hash, client_send_origin_cb, - (void *) msg); + (void *) env); return n; } @@ -677,17 +689,12 @@ client_send_origin (struct GNUNET_HashCode *pub_key_hash, static void client_send_ack (struct GNUNET_HashCode *pub_key_hash) { + struct GNUNET_MQ_Envelope *env; + GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Sending message ACK to client.\n"); - - static struct GNUNET_MessageHeader *msg = NULL; - if (NULL == msg) - { - msg = GNUNET_malloc (sizeof (*msg)); - msg->type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_FRAGMENT_ACK); - msg->size = htons (sizeof (*msg)); - } - client_send_all (pub_key_hash, msg); + env = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_MULTICAST_FRAGMENT_ACK); + client_send_all (pub_key_hash, env); } @@ -983,7 +990,8 @@ handle_cadet_join_request (void *cls, chn->peer = req->peer; chn->join_status = JOIN_WAITING; - client_send_all (&group_pub_hash, &req->header); + client_send_all (&group_pub_hash, + GNUNET_MQ_msg_copy (&req->header)); } @@ -1102,7 +1110,8 @@ handle_cadet_message (void *cls, { struct Channel *chn = cls; GNUNET_CADET_receive_done (chn->channel); - client_send_all (&chn->group_pub_hash, &msg->header); + client_send_all (&chn->group_pub_hash, + GNUNET_MQ_msg_copy (&msg->header)); } @@ -1153,30 +1162,32 @@ handle_cadet_request (void *cls, { struct Channel *chn = cls; GNUNET_CADET_receive_done (chn->channel); - client_send_origin (&chn->group_pub_hash, &req->header); + client_send_origin (&chn->group_pub_hash, + GNUNET_MQ_msg_copy (&req->header)); } -static int -check_cadet_replay_request (void *cls, - const struct MulticastReplayRequestMessage *req) -{ - uint16_t size = ntohs (req->header.size); - if (size < sizeof (*req)) - { - GNUNET_break_op (0); - return GNUNET_SYSERR; - } - - struct Channel *chn = cls; - if (NULL == chn) - { - GNUNET_break_op (0); - return GNUNET_SYSERR; - } - - return GNUNET_OK; -} +// FIXME: do checks in handle_cadet_replay_request +//static int +//check_cadet_replay_request (void *cls, +// const struct MulticastReplayRequestMessage *req) +//{ +// uint16_t size = ntohs (req->header.size); +// if (size < sizeof (*req)) +// { +// GNUNET_break_op (0); +// return GNUNET_SYSERR; +// } +// +// struct Channel *chn = cls; +// if (NULL == chn) +// { +// GNUNET_break_op (0); +// return GNUNET_SYSERR; +// } +// +// return GNUNET_OK; +//} /** @@ -1187,6 +1198,7 @@ handle_cadet_replay_request (void *cls, const struct MulticastReplayRequestMessage *req) { struct Channel *chn = cls; + GNUNET_CADET_receive_done (chn->channel); struct MulticastReplayRequestMessage rep = *req; @@ -1203,12 +1215,16 @@ handle_cadet_replay_request (void *cls, GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST); } struct GNUNET_HashCode key_hash; - replay_key_hash (rep.fragment_id, rep.message_id, rep.fragment_offset, - rep.flags, &key_hash); + replay_key_hash (rep.fragment_id, + rep.message_id, + rep.fragment_offset, + rep.flags, + &key_hash); GNUNET_CONTAINER_multihashmap_put (grp_replay_req, &key_hash, chn, GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE); - client_send_random (&chn->group_pub_hash, &rep.header); + client_send_random (&chn->group_pub_hash, + GNUNET_MQ_msg_copy (&rep.header)); } @@ -1290,10 +1306,10 @@ cadet_channel_create (struct Group *grp, struct GNUNET_PeerIdentity *peer) struct MulticastJoinDecisionMessageHeader, chn), - GNUNET_MQ_hd_var_size (cadet_replay_request, - GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST, - struct MulticastReplayRequestMessage, - chn), + GNUNET_MQ_hd_fixed_size (cadet_replay_request, + GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST, + struct MulticastReplayRequestMessage, + chn), GNUNET_MQ_hd_var_size (cadet_replay_response, GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE, @@ -1379,10 +1395,10 @@ handle_client_origin_start (void *cls, struct MulticastJoinRequestMessage, grp), - GNUNET_MQ_hd_var_size (cadet_replay_request, - GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST, - struct MulticastReplayRequestMessage, - grp), + GNUNET_MQ_hd_fixed_size (cadet_replay_request, + GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST, + struct MulticastReplayRequestMessage, + grp), GNUNET_MQ_hd_var_size (cadet_replay_response, GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE, @@ -1494,7 +1510,8 @@ handle_client_member_join (void *cls, } GNUNET_CONTAINER_multihashmap_put (grp_mem, &mem->pub_key_hash, mem, GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST); - + + // FIXME: should the members hash map have option UNIQUE_FAST? GNUNET_CONTAINER_multihashmap_put (members, &grp->pub_key_hash, mem, GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE); } @@ -1509,10 +1526,11 @@ handle_client_member_join (void *cls, char *str = GNUNET_CRYPTO_ecdsa_public_key_to_string (&mem->pub_key); GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Client connected to group %s as member %s (%s).\n", + "Client connected to group %s as member %s (%s). size = %d\n", GNUNET_h2s (&grp->pub_key_hash), GNUNET_h2s2 (&mem->pub_key_hash), - str); + str, + GNUNET_CONTAINER_multihashmap_size (members)); GNUNET_free (str); if (NULL != mem->join_dcsn) @@ -1567,7 +1585,9 @@ handle_client_member_join (void *cls, GNUNET_free (mem->join_req); mem->join_req = req; - if (0 == client_send_origin (&grp->pub_key_hash, &mem->join_req->header)) + if (0 == + client_send_origin (&grp->pub_key_hash, + GNUNET_MQ_msg_copy (&mem->join_req->header))) { /* No local origins, send to remote origin */ cadet_send_join_request (mem); } @@ -1580,7 +1600,7 @@ static void client_send_join_decision (struct Member *mem, const struct MulticastJoinDecisionMessageHeader *hdcsn) { - client_send_group (&mem->group, &hdcsn->header); + client_send_group (&mem->group, GNUNET_MQ_msg_copy (&hdcsn->header)); const struct MulticastJoinDecisionMessage * dcsn = (const struct MulticastJoinDecisionMessage *) &hdcsn[1]; @@ -1622,7 +1642,7 @@ handle_client_join_decision (void *cls, return; } GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "%p Got join decision from client for group %s..\n", + "%p got join decision from client for group %s..\n", grp, GNUNET_h2s (&grp->pub_key_hash)); struct GNUNET_CONTAINER_MultiHashMap * @@ -1652,6 +1672,30 @@ handle_client_join_decision (void *cls, } +static void +handle_client_part_request (void *cls, + const struct GNUNET_MessageHeader *msg) +{ + struct Client *c = cls; + struct GNUNET_SERVICE_Client *client = c->client; + struct Group *grp = c->group; + struct GNUNET_MQ_Envelope *env; + + if (NULL == grp) + { + GNUNET_break (0); + GNUNET_SERVICE_client_drop (client); + return; + } + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "%p got part request from client for group %s.\n", + grp, GNUNET_h2s (&grp->pub_key_hash)); + env = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_MULTICAST_PART_ACK); + client_send_group (grp, env); + GNUNET_SERVICE_client_continue (client); +} + + static int check_client_multicast_message (void *cls, const struct GNUNET_MULTICAST_MessageHeader *msg) @@ -1680,6 +1724,7 @@ handle_client_multicast_message (void *cls, GNUNET_assert (GNUNET_YES == grp->is_origin); struct Origin *orig = grp->origin; + // FIXME: use GNUNET_MQ_msg_copy /* FIXME: yucky, should use separate message structs for P2P and CS! */ struct GNUNET_MULTICAST_MessageHeader * out = (struct GNUNET_MULTICAST_MessageHeader *) GNUNET_copy_message (&msg->header); @@ -1696,7 +1741,7 @@ handle_client_multicast_message (void *cls, GNUNET_assert (0); } - client_send_all (&grp->pub_key_hash, &out->header); + client_send_all (&grp->pub_key_hash, GNUNET_MQ_msg_copy (&out->header)); cadet_send_children (&grp->pub_key_hash, &out->header); client_send_ack (&grp->pub_key_hash); GNUNET_free (out); @@ -1751,7 +1796,9 @@ handle_client_multicast_request (void *cls, } uint8_t send_ack = GNUNET_YES; - if (0 == client_send_origin (&grp->pub_key_hash, &out->header)) + if (0 == + client_send_origin (&grp->pub_key_hash, + GNUNET_MQ_msg_copy (&out->header))) { /* No local origins, send to remote origin */ if (NULL != mem->origin_channel) { @@ -1812,7 +1859,9 @@ handle_client_replay_request (void *cls, GNUNET_CONTAINER_multihashmap_put (grp_replay_req, &key_hash, client, GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE); - if (0 == client_send_origin (&grp->pub_key_hash, &rep->header)) + if (0 == + client_send_origin (&grp->pub_key_hash, + GNUNET_MQ_msg_copy (&rep->header))) { /* No local origin, replay from remote members / origin. */ if (NULL != mem->origin_channel) { @@ -1821,6 +1870,7 @@ handle_client_replay_request (void *cls, else { /* FIXME: not yet connected to origin */ + GNUNET_assert (0); GNUNET_SERVICE_client_drop (client); return; } @@ -2033,6 +2083,8 @@ client_notify_disconnect (void *cls, grp, (GNUNET_YES == grp->is_origin) ? "origin" : "member", GNUNET_h2s (&grp->pub_key_hash)); + // FIXME (due to protocol change): here we must not remove all clients, + // only the one we were notified about! struct ClientList *cl = grp->clients_head; while (NULL != cl) { @@ -2119,6 +2171,10 @@ GNUNET_SERVICE_MAIN GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_DECISION, struct MulticastJoinDecisionMessageHeader, NULL), + GNUNET_MQ_hd_fixed_size (client_part_request, + GNUNET_MESSAGE_TYPE_MULTICAST_PART_REQUEST, + struct GNUNET_MessageHeader, + NULL), GNUNET_MQ_hd_var_size (client_multicast_message, GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE, struct GNUNET_MULTICAST_MessageHeader, diff --git a/src/multicast/multicast_api.c b/src/multicast/multicast_api.c index a8b1dee40..3c911f48a 100644 --- a/src/multicast/multicast_api.c +++ b/src/multicast/multicast_api.c @@ -542,31 +542,12 @@ group_cleanup (struct GNUNET_MULTICAST_Group *grp) static void -group_disconnect (struct GNUNET_MULTICAST_Group *grp, - GNUNET_ContinuationCallback cb, - void *cls) +handle_group_part_ack (void *cls, + const struct GNUNET_MessageHeader *msg) { - grp->is_disconnecting = GNUNET_YES; - grp->disconnect_cb = cb; - grp->disconnect_cls = cls; + struct GNUNET_MULTICAST_Group *grp = cls; - if (NULL != grp->mq) - { - struct GNUNET_MQ_Envelope *last = GNUNET_MQ_get_last_envelope (grp->mq); - if (NULL != last) - { - GNUNET_MQ_notify_sent (last, - (GNUNET_SCHEDULER_TaskCallback) group_cleanup, grp); - } - else - { - group_cleanup (grp); - } - } - else - { - group_cleanup (grp); - } + group_cleanup (grp); } @@ -779,6 +760,10 @@ origin_connect (struct GNUNET_MULTICAST_Origin *orig) GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_REQUEST, struct MulticastJoinRequestMessage, grp), + GNUNET_MQ_hd_fixed_size (group_part_ack, + GNUNET_MESSAGE_TYPE_MULTICAST_PART_ACK, + struct GNUNET_MessageHeader, + grp), GNUNET_MQ_hd_fixed_size (group_replay_request, GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST, struct MulticastReplayRequestMessage, @@ -879,8 +864,13 @@ GNUNET_MULTICAST_origin_stop (struct GNUNET_MULTICAST_Origin *orig, void *stop_cls) { struct GNUNET_MULTICAST_Group *grp = &orig->grp; + struct GNUNET_MQ_Envelope *env; - group_disconnect (grp, stop_cb, stop_cls); + grp->is_disconnecting = GNUNET_YES; + grp->disconnect_cb = stop_cb; + grp->disconnect_cls = stop_cls; + env = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_MULTICAST_PART_REQUEST); + GNUNET_MQ_send (grp->mq, env); } @@ -1065,6 +1055,10 @@ member_connect (struct GNUNET_MULTICAST_Member *mem) GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_DECISION, struct MulticastJoinDecisionMessageHeader, mem), + GNUNET_MQ_hd_fixed_size (group_part_ack, + GNUNET_MESSAGE_TYPE_MULTICAST_PART_ACK, + struct GNUNET_MessageHeader, + grp), GNUNET_MQ_hd_fixed_size (group_replay_request, GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST, struct MulticastReplayRequestMessage, @@ -1198,16 +1192,19 @@ GNUNET_MULTICAST_member_part (struct GNUNET_MULTICAST_Member *mem, GNUNET_ContinuationCallback part_cb, void *part_cls) { - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%p Member parting.\n", mem); struct GNUNET_MULTICAST_Group *grp = &mem->grp; + struct GNUNET_MQ_Envelope *env; mem->join_dcsn_cb = NULL; grp->join_req_cb = NULL; grp->message_cb = NULL; grp->replay_msg_cb = NULL; grp->replay_frag_cb = NULL; - - group_disconnect (grp, part_cb, part_cls); + grp->is_disconnecting = GNUNET_YES; + grp->disconnect_cb = part_cb; + grp->disconnect_cls = part_cls; + env = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_MULTICAST_PART_REQUEST); + GNUNET_MQ_send (grp->mq, env); } -- 2.25.1