aboutsummaryrefslogtreecommitdiff
path: root/src/experimentation/gnunet-daemon-experimentation_nodes.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/experimentation/gnunet-daemon-experimentation_nodes.c')
-rw-r--r--src/experimentation/gnunet-daemon-experimentation_nodes.c1157
1 files changed, 0 insertions, 1157 deletions
diff --git a/src/experimentation/gnunet-daemon-experimentation_nodes.c b/src/experimentation/gnunet-daemon-experimentation_nodes.c
deleted file mode 100644
index 1c4152793..000000000
--- a/src/experimentation/gnunet-daemon-experimentation_nodes.c
+++ /dev/null
@@ -1,1157 +0,0 @@
1/*
2 This file is part of GNUnet.
3 (C) 2012-2013 Christian Grothoff (and other contributing authors)
4
5 GNUnet is free software; you can redistribute it and/or modify
6 it under the terms of the GNU General Public License as published
7 by the Free Software Foundation; either version 3, or (at your
8 option) any later version.
9
10 GNUnet is distributed in the hope that it will be useful, but
11 WITHOUT ANY WARRANTY; without even the implied warranty of
12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 General Public License for more details.
14
15 You should have received a copy of the GNU General Public License
16 along with GNUnet; see the file COPYING. If not, write to the
17 Free Software Foundation, Inc., 59 Temple Place - Suite 330,
18 Boston, MA 02111-1307, USA.
19*/
20
21/**
22 * @file experimentation/gnunet-daemon-experimentation_nodes.c
23 * @brief experimentation daemon: node management
24 * @author Christian Grothoff
25 * @author Matthias Wachs
26 */
27#include "platform.h"
28#include "gnunet_util_lib.h"
29#include "gnunet_core_service.h"
30#include "gnunet_statistics_service.h"
31#include "gnunet-daemon-experimentation.h"
32
33
34#define FAST_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5)
35
36/**
37 * Core handle
38 */
39static struct GNUNET_CORE_Handle *ch;
40
41/**
42 * Peer's own identity
43 */
44static struct GNUNET_PeerIdentity me;
45
46/**
47 * Nodes with a pending request
48 */
49static struct GNUNET_CONTAINER_MultiPeerMap *nodes_requested;
50
51/**
52 * Active experimentation nodes
53 */
54static struct GNUNET_CONTAINER_MultiPeerMap *nodes_active;
55
56/**
57 * Inactive experimentation nodes
58 * To be excluded from future requests
59 */
60static struct GNUNET_CONTAINER_MultiPeerMap *nodes_inactive;
61
62
63struct NodeComCtx
64{
65 struct NodeComCtx *prev;
66 struct NodeComCtx *next;
67
68 struct Node *n;
69 struct Experiment *e;
70
71 size_t size;
72 GNUNET_CONNECTION_TransmitReadyNotify notify;
73 void *notify_cls;
74};
75
76
77/**
78 * Update statistics
79 *
80 * @param m peermap to update values from
81 */
82static void
83update_stats (struct GNUNET_CONTAINER_MultiPeerMap *m)
84{
85 GNUNET_assert (NULL != m);
86 GNUNET_assert (NULL != GED_stats);
87
88 if (m == nodes_active)
89 {
90 GNUNET_STATISTICS_set (GED_stats, "# nodes active",
91 GNUNET_CONTAINER_multipeermap_size(m), GNUNET_NO);
92 }
93 else if (m == nodes_inactive)
94 {
95 GNUNET_STATISTICS_set (GED_stats, "# nodes inactive",
96 GNUNET_CONTAINER_multipeermap_size(m), GNUNET_NO);
97 }
98 else if (m == nodes_requested)
99 {
100 GNUNET_STATISTICS_set (GED_stats, "# nodes requested",
101 GNUNET_CONTAINER_multipeermap_size(m), GNUNET_NO);
102 }
103 else
104 GNUNET_break (0);
105}
106
107
108/**
109 * Clean up node
110 *
111 * @param cls the peermap to clean up
112 * @param key key of the current node
113 * @param value related node object
114 * @return always #GNUNET_OK
115 */
116static int
117cleanup_node (void *cls,
118 const struct GNUNET_PeerIdentity * key,
119 void *value)
120{
121 struct Node *n;
122 struct NodeComCtx *e_cur;
123 struct NodeComCtx *e_next;
124 struct GNUNET_CONTAINER_MultiPeerMap *cur = cls;
125
126 n = value;
127 if (GNUNET_SCHEDULER_NO_TASK != n->timeout_task)
128 {
129 GNUNET_SCHEDULER_cancel (n->timeout_task);
130 n->timeout_task = GNUNET_SCHEDULER_NO_TASK;
131 }
132
133 if (NULL != n->cth)
134 {
135 GNUNET_CORE_notify_transmit_ready_cancel (n->cth);
136 n->cth = NULL;
137 }
138 e_next = n->e_req_head;
139 while (NULL != (e_cur = e_next))
140 {
141 e_next = e_cur->next;
142 GNUNET_CONTAINER_DLL_remove (n->e_req_head, n->e_req_tail, e_cur);
143 GNUNET_free (e_cur);
144 }
145 GNUNET_break (0 == GNUNET_CONTAINER_multipeermap_remove (cur, key, value));
146 GNUNET_free (value);
147 return GNUNET_OK;
148}
149
150
151/**
152 * Check if id passed is my id
153 *
154 * @param id the id to check
155 * @return GNUNET_YES or GNUNET_NO
156 */
157static int
158is_me (const struct GNUNET_PeerIdentity *id)
159{
160 if (0 == memcmp (&me, id, sizeof (me)))
161 return GNUNET_YES;
162 else
163 return GNUNET_NO;
164}
165
166
167/**
168 * Core startup callback
169 *
170 * @param cls unused
171 * @param my_identity my id
172 */
173static void
174core_startup_handler (void *cls,
175 const struct GNUNET_PeerIdentity *my_identity)
176{
177 me = *my_identity;
178}
179
180
181static void
182schedule_transmisson (struct NodeComCtx *e_ctx);
183
184
185static size_t
186transmit_read_wrapper (void *cls, size_t bufsize, void *buf)
187{
188 struct NodeComCtx *e_ctx = cls;
189 struct NodeComCtx *next;
190
191 size_t res = e_ctx->notify (e_ctx->notify_cls, bufsize, buf);
192 e_ctx->n->cth = NULL;
193
194 GNUNET_CONTAINER_DLL_remove (e_ctx->n->e_req_head, e_ctx->n->e_req_tail, e_ctx);
195 next = e_ctx->n->e_req_head;
196 GNUNET_free (e_ctx);
197
198 if (NULL != next)
199 {
200 /* Schedule next message */
201 schedule_transmisson (next);
202 }
203 return res;
204}
205
206
207static void
208schedule_transmisson (struct NodeComCtx *e_ctx)
209{
210 if (NULL != e_ctx->n->cth)
211 return;
212
213 e_ctx->n->cth = GNUNET_CORE_notify_transmit_ready (ch, GNUNET_NO,
214 GNUNET_CORE_PRIO_BEST_EFFORT,
215 FAST_TIMEOUT,
216 &e_ctx->n->id, e_ctx->size,
217 transmit_read_wrapper, e_ctx);
218 if (NULL == e_ctx->n->cth)
219 {
220 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
221 _("Cannot send message to peer `%s' for experiment `%s'\n"),
222 GNUNET_i2s(&e_ctx->n->id), e_ctx->e->name);
223 GNUNET_free (e_ctx);
224 }
225}
226
227
228/**
229 * Remove experimentation request due to timeout
230 *
231 * @param cls the related node
232 * @param tc scheduler's task context
233 */
234static void
235remove_request (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
236{
237 struct Node *n = cls;
238
239 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
240 "Removing request for peer %s due to timeout\n",
241 GNUNET_i2s (&n->id));
242 if (GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (nodes_requested, &n->id))
243 {
244 GNUNET_break (0 == GNUNET_CONTAINER_multipeermap_remove (nodes_requested, &n->id, n));
245 update_stats (nodes_requested);
246 GNUNET_CONTAINER_multipeermap_put (nodes_inactive, &n->id, n,
247 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
248 update_stats (nodes_inactive);
249 }
250 n->timeout_task = GNUNET_SCHEDULER_NO_TASK;
251}
252
253
254static int
255append_public_key (void *cls,
256 const struct GNUNET_HashCode *key,
257 void *value)
258{
259 struct GNUNET_CRYPTO_EddsaPublicKey **issuers = cls;
260 struct Issuer *issuer = value;
261
262 *issuers[0] = issuer->pubkey;
263 *issuers = &((*issuers)[1]);
264 return GNUNET_OK;
265}
266
267
268/**
269 * Core's transmit notify callback to send request
270 *
271 * @param cls the related node
272 * @param bufsize buffer size
273 * @param buf the buffer to copy to
274 * @return bytes passed
275 */
276static size_t
277send_experimentation_request_cb (void *cls, size_t bufsize, void *buf)
278{
279 struct Node *n = cls;
280 struct Experimentation_Request msg;
281 unsigned int my_issuer_count = GNUNET_CONTAINER_multihashmap_size (valid_issuers);
282 size_t msg_size = sizeof (msg);
283 size_t ri_size = sizeof (struct GNUNET_CRYPTO_EddsaPublicKey) * my_issuer_count;
284 size_t total_size = msg_size + ri_size;
285 struct GNUNET_CRYPTO_EddsaPublicKey *issuers;
286
287 n->cth = NULL;
288 if (NULL == buf)
289 {
290 /* client disconnected */
291 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
292 "Client disconnected\n");
293 if (GNUNET_SCHEDULER_NO_TASK != n->timeout_task)
294 GNUNET_SCHEDULER_cancel (n->timeout_task);
295 GNUNET_SCHEDULER_add_now (&remove_request, n);
296 return 0;
297 }
298 GNUNET_assert (bufsize >= total_size);
299 msg.msg.size = htons (total_size);
300 msg.msg.type = htons (GNUNET_MESSAGE_TYPE_EXPERIMENTATION_REQUEST);
301 msg.capabilities = htonl (GSE_node_capabilities);
302 msg.issuer_count = htonl (my_issuer_count);
303 memcpy (buf, &msg, msg_size);
304 issuers = (struct GNUNET_CRYPTO_EddsaPublicKey *) buf + msg_size;
305 GNUNET_CONTAINER_multihashmap_iterate (valid_issuers,
306 &append_public_key,
307 &issuers);
308 GNUNET_log (GNUNET_ERROR_TYPE_INFO,
309 _("Sending experimentation request to peer %s\n"),
310 GNUNET_i2s (&n->id));
311 return total_size;
312}
313
314
315/**
316 * Send request to peer to start add him to to the set of experimentation nodes
317 *
318 * @param peer the peer to send to
319 */
320static void
321send_experimentation_request (const struct GNUNET_PeerIdentity *peer)
322{
323 struct Node *n;
324 struct NodeComCtx *e_ctx;
325 size_t size;
326 size_t c_issuers;
327
328 c_issuers = GNUNET_CONTAINER_multihashmap_size (valid_issuers);
329 size = sizeof (struct Experimentation_Request) +
330 c_issuers * sizeof (struct GNUNET_CRYPTO_EddsaPublicKey);
331 n = GNUNET_new (struct Node);
332 n->id = *peer;
333 n->timeout_task = GNUNET_SCHEDULER_add_delayed (EXP_RESPONSE_TIMEOUT, &remove_request, n);
334 n->capabilities = NONE;
335
336 e_ctx = GNUNET_new (struct NodeComCtx);
337 e_ctx->n = n;
338 e_ctx->e = NULL;
339 e_ctx->size = size;
340 e_ctx->notify = &send_experimentation_request_cb;
341 e_ctx->notify_cls = n;
342 GNUNET_CONTAINER_DLL_insert_tail(n->e_req_head, n->e_req_tail, e_ctx);
343 schedule_transmisson (e_ctx);
344
345 GNUNET_assert (GNUNET_OK ==
346 GNUNET_CONTAINER_multipeermap_put (nodes_requested,
347 peer, n,
348 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
349 update_stats (nodes_requested);
350}
351
352
353/**
354 * Core's transmit notify callback to send response
355 *
356 * @param cls the related node
357 * @param bufsize buffer size
358 * @param buf the buffer to copy to
359 * @return bytes passed
360 */
361static size_t
362send_response_cb (void *cls, size_t bufsize, void *buf)
363{
364 struct Node *n = cls;
365 struct Experimentation_Response msg;
366 size_t c_issuers = GNUNET_CONTAINER_multihashmap_size (valid_issuers);
367 size_t ri_size = c_issuers * sizeof (struct GNUNET_CRYPTO_EddsaPublicKey);
368 size_t msg_size = sizeof (msg);
369 size_t total_size = msg_size + ri_size;
370 struct GNUNET_CRYPTO_EddsaPublicKey *issuers;
371
372 n->cth = NULL;
373 if (buf == NULL)
374 {
375 /* client disconnected */
376 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
377 "Client disconnected\n");
378 return 0;
379 }
380 GNUNET_assert (bufsize >= total_size);
381
382 msg.msg.size = htons (total_size);
383 msg.msg.type = htons (GNUNET_MESSAGE_TYPE_EXPERIMENTATION_RESPONSE);
384 msg.capabilities = htonl (GSE_node_capabilities);
385 msg.issuer_count = htonl (c_issuers);
386 memcpy (buf, &msg, msg_size);
387 issuers = (struct GNUNET_CRYPTO_EddsaPublicKey *) buf + msg_size;
388 GNUNET_CONTAINER_multihashmap_iterate (valid_issuers,
389 &append_public_key,
390 &issuers);
391 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
392 "Sending response to peer %s\n",
393 GNUNET_i2s (&n->id));
394 return total_size;
395}
396
397
398static void
399get_experiments_cb (struct Node *n, struct Experiment *e)
400{
401 static int counter = 0;
402 if (NULL == e)
403 return; /* Done */
404
405 /* Tell the scheduler to add a node with an experiment */
406 GED_scheduler_add (n, e, GNUNET_YES);
407 counter ++;
408}
409
410
411struct Node *
412get_node (const struct GNUNET_PeerIdentity *id)
413{
414 struct Node * res;
415 struct Node * tmp;
416
417 res = NULL;
418 tmp = NULL;
419 tmp = GNUNET_CONTAINER_multipeermap_get (nodes_active, id);
420 if (res == NULL)
421 res = tmp;
422
423 tmp = GNUNET_CONTAINER_multipeermap_get (nodes_inactive, id);
424 if (res == NULL)
425 res = tmp;
426 else
427 GNUNET_break (0); /* Multiple instances */
428
429 tmp = GNUNET_CONTAINER_multipeermap_get (nodes_requested, id);
430 if (res == NULL)
431 res = tmp;
432 else
433 GNUNET_break (0); /* Multiple instances */
434
435 return res;
436}
437
438
439/**
440 * Set a specific node as active
441 *
442 * @param n the node
443 */
444static void
445node_make_active (struct Node *n)
446{
447 int c1;
448
449 GNUNET_CONTAINER_multipeermap_put (nodes_active,
450 &n->id, n, GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
451 update_stats (nodes_active);
452 GNUNET_log (GNUNET_ERROR_TYPE_INFO,
453 _("Added peer `%s' as active node\n"),
454 GNUNET_i2s (&n->id));
455 /* Request experiments for this node to start them */
456 for (c1 = 0; c1 < n->issuer_count; c1++)
457 {
458 GED_experiments_get (n, &n->issuer_id[c1], &get_experiments_cb);
459 }
460}
461
462
463/**
464 * Handle a request and send a response
465 *
466 * @param peer the source
467 * @param message the message
468 */
469static void
470handle_request (const struct GNUNET_PeerIdentity *peer,
471 const struct GNUNET_MessageHeader *message)
472{
473 struct Node *n;
474 struct NodeComCtx *e_ctx;
475 const struct Experimentation_Request *rm = (const struct Experimentation_Request *) message;
476 const struct GNUNET_CRYPTO_EddsaPublicKey *rmi = (const struct GNUNET_CRYPTO_EddsaPublicKey *) &rm[1];
477 unsigned int my_issuer_count = GNUNET_CONTAINER_multihashmap_size (valid_issuers);
478 int c1;
479 int c2;
480 uint32_t ic;
481 uint32_t ic_accepted;
482 int make_active;
483
484 if (ntohs (message->size) < sizeof (struct Experimentation_Request))
485 {
486 GNUNET_break (0);
487 return;
488 }
489 ic = ntohl (rm->issuer_count);
490 if (ntohs (message->size) !=
491 sizeof (struct Experimentation_Request) + ic * sizeof (struct GNUNET_CRYPTO_EddsaPublicKey))
492 {
493 GNUNET_break (0);
494 return;
495 }
496
497 make_active = GNUNET_NO;
498 if (NULL != (n = GNUNET_CONTAINER_multipeermap_get (nodes_active, peer)))
499 {
500 /* Nothing to do */
501 }
502 else if (NULL != (n = GNUNET_CONTAINER_multipeermap_get (nodes_requested, peer)))
503 {
504 GNUNET_assert (GNUNET_OK == GNUNET_CONTAINER_multipeermap_remove (nodes_requested, peer, n));
505 if (GNUNET_SCHEDULER_NO_TASK != n->timeout_task)
506 {
507 GNUNET_SCHEDULER_cancel (n->timeout_task);
508 n->timeout_task = GNUNET_SCHEDULER_NO_TASK;
509 }
510 update_stats (nodes_requested);
511 make_active = GNUNET_YES;
512 }
513 else if (NULL != (n = GNUNET_CONTAINER_multipeermap_get (nodes_inactive, peer)))
514 {
515 GNUNET_break (0 == GNUNET_CONTAINER_multipeermap_remove (nodes_inactive, peer, n));
516 update_stats (nodes_inactive);
517 make_active = GNUNET_YES;
518 }
519 else
520 {
521 /* Create new node */
522 n = GNUNET_new (struct Node);
523 n->id = *peer;
524 n->capabilities = NONE;
525 make_active = GNUNET_YES;
526 }
527
528 /* Update node */
529 n->capabilities = ntohl (rm->capabilities);
530
531 /* Filter accepted issuer */
532 ic_accepted = 0;
533 for (c1 = 0; c1 < ic; c1++)
534 {
535 if (GNUNET_YES == GED_experiments_issuer_accepted(&rmi[c1]))
536 ic_accepted ++;
537 }
538 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
539 "Request from peer `%s' with %u issuers, we accepted %u issuer \n",
540 GNUNET_i2s (peer), ic, ic_accepted);
541 GNUNET_free_non_null (n->issuer_id);
542 n->issuer_id = GNUNET_malloc (ic_accepted * sizeof (struct GNUNET_CRYPTO_EddsaPublicKey));
543 c2 = 0;
544 for (c1 = 0; c1 < ic; c1++)
545 {
546 if (GNUNET_YES == GED_experiments_issuer_accepted (&rmi[c1]))
547 {
548 n->issuer_id[c2] = rmi[c1];
549 c2 ++;
550 }
551 }
552 n->issuer_count = ic_accepted;
553
554 if (GNUNET_YES == make_active)
555 node_make_active (n);
556
557 /* Send response */
558 e_ctx = GNUNET_new (struct NodeComCtx);
559 e_ctx->n = n;
560 e_ctx->e = NULL;
561 e_ctx->size = sizeof (struct Experimentation_Response) +
562 my_issuer_count * sizeof (struct GNUNET_CRYPTO_EddsaPublicKey);
563 e_ctx->notify = &send_response_cb;
564 e_ctx->notify_cls = n;
565
566 GNUNET_CONTAINER_DLL_insert_tail(n->e_req_head, n->e_req_tail, e_ctx);
567 schedule_transmisson (e_ctx);
568}
569
570
571/**
572 * Handle a response
573 *
574 * @param peer the source
575 * @param message the message
576 */
577static void handle_response (const struct GNUNET_PeerIdentity *peer,
578 const struct GNUNET_MessageHeader *message)
579{
580 struct Node *n;
581 const struct Experimentation_Response *rm = (const struct Experimentation_Response *) message;
582 const struct GNUNET_CRYPTO_EddsaPublicKey *rmi = (const struct GNUNET_CRYPTO_EddsaPublicKey *) &rm[1];
583 uint32_t ic;
584 uint32_t ic_accepted;
585 int make_active;
586 unsigned int c1;
587 unsigned int c2;
588
589 if (ntohs (message->size) < sizeof (struct Experimentation_Response))
590 {
591 GNUNET_break (0);
592 return;
593 }
594 ic = ntohl (rm->issuer_count);
595 if (ntohs (message->size) != sizeof (struct Experimentation_Response) + ic * sizeof (struct GNUNET_CRYPTO_EddsaPublicKey))
596 {
597 GNUNET_break (0);
598 return;
599 }
600
601 make_active = GNUNET_NO;
602 if (NULL != (n = GNUNET_CONTAINER_multipeermap_get (nodes_active, peer)))
603 {
604 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
605 "Received %s from %s peer `%s'\n",
606 "RESPONSE", "active", GNUNET_i2s (peer));
607 }
608 else if (NULL != (n = GNUNET_CONTAINER_multipeermap_get (nodes_requested, peer)))
609 {
610 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Received %s from %s peer `%s'\n",
611 "RESPONSE", "requested", GNUNET_i2s (peer));
612 GNUNET_assert (GNUNET_OK == GNUNET_CONTAINER_multipeermap_remove (nodes_requested, peer, n));
613 if (GNUNET_SCHEDULER_NO_TASK != n->timeout_task)
614 {
615 GNUNET_SCHEDULER_cancel (n->timeout_task);
616 n->timeout_task = GNUNET_SCHEDULER_NO_TASK;
617 }
618 update_stats (nodes_requested);
619 make_active = GNUNET_YES;
620 }
621 else if (NULL != (n = GNUNET_CONTAINER_multipeermap_get (nodes_inactive, peer)))
622 {
623 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
624 "Received %s from peer `%s'\n",
625 "RESPONSE", "inactive", GNUNET_i2s (peer));
626 GNUNET_break (0 == GNUNET_CONTAINER_multipeermap_remove (nodes_inactive, peer, n));
627 update_stats (nodes_inactive);
628 make_active = GNUNET_YES;
629 }
630 else
631 {
632 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Received %s from %s peer `%s'\n",
633 "RESPONSE", "unknown", GNUNET_i2s (peer));
634 return;
635 }
636
637 /* Update */
638 n->capabilities = ntohl (rm->capabilities);
639
640 /* Filter accepted issuer */
641 ic_accepted = 0;
642 for (c1 = 0; c1 < ic; c1++)
643 {
644 if (GNUNET_YES == GED_experiments_issuer_accepted(&rmi[c1]))
645 ic_accepted ++;
646 }
647 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
648 "Response from peer `%s' with %u issuers, we accepted %u issuer \n",
649 GNUNET_i2s (peer), ic, ic_accepted);
650 GNUNET_free_non_null (n->issuer_id);
651 n->issuer_id = GNUNET_malloc (ic_accepted * sizeof (struct GNUNET_PeerIdentity));
652 c2 = 0;
653 for (c1 = 0; c1 < ic; c1++)
654 {
655 if (GNUNET_YES == GED_experiments_issuer_accepted(&rmi[c1]))
656 {
657 n->issuer_id[c2] = rmi[c1];
658 c2 ++;
659 }
660 }
661 n->issuer_count = ic_accepted;
662
663 if (GNUNET_YES == make_active)
664 node_make_active (n);
665}
666
667
668/**
669 * Handle a response
670 *
671 * @param peer the source
672 * @param message the message
673 */
674static void
675handle_start (const struct GNUNET_PeerIdentity *peer,
676 const struct GNUNET_MessageHeader *message)
677{
678 uint16_t size;
679 uint32_t name_len;
680 const struct GED_start_message *msg;
681 const char *name;
682 struct Node *n;
683 struct Experiment *e;
684
685 if (NULL == peer)
686 {
687 GNUNET_break (0);
688 return;
689 }
690 if (NULL == message)
691 {
692 GNUNET_break (0);
693 return;
694 }
695
696 size = ntohs (message->size);
697 if (size < sizeof (struct GED_start_message))
698 {
699 GNUNET_break (0);
700 return;
701 }
702 msg = (const struct GED_start_message *) message;
703 name_len = ntohl (msg->len_name);
704 if (size != sizeof (struct GED_start_message) + name_len)
705 {
706 GNUNET_break (0);
707 return;
708 }
709
710 n = get_node (peer);
711 if (NULL == n)
712 {
713 GNUNET_break (0);
714 return;
715 }
716 name = (const char *) &msg[1];
717 if (name[name_len-1] != '\0')
718 {
719 GNUNET_break (0);
720 return;
721 }
722 if (name_len != strlen (name) + 1)
723 {
724 GNUNET_break (0);
725 return;
726 }
727 e = GED_experiments_find (&msg->issuer, name, GNUNET_TIME_absolute_ntoh(msg->version_nbo));
728 if (NULL == e)
729 {
730 GNUNET_break (0);
731 return;
732 }
733 GED_scheduler_handle_start (n, e);
734}
735
736
737/**
738 * Handle a response
739 *
740 * @param peer the source
741 * @param message the message
742 */
743static void
744handle_start_ack (const struct GNUNET_PeerIdentity *peer,
745 const struct GNUNET_MessageHeader *message)
746{
747 uint16_t size;
748 uint32_t name_len;
749 const struct GED_start_ack_message *msg;
750 const char *name;
751 struct Node *n;
752 struct Experiment *e;
753
754 if (NULL == peer)
755 {
756 GNUNET_break (0);
757 return;
758 }
759 if (NULL == message)
760 {
761 GNUNET_break (0);
762 return;
763 }
764
765 size = ntohs (message->size);
766 if (size < sizeof (struct GED_start_ack_message))
767 {
768 GNUNET_break (0);
769 return;
770 }
771 msg = (const struct GED_start_ack_message *) message;
772 name_len = ntohl (msg->len_name);
773 if (size != sizeof (struct GED_start_message) + name_len)
774 {
775 GNUNET_break (0);
776 return;
777 }
778
779 n = get_node (peer);
780 if (NULL == n)
781 {
782 GNUNET_break (0);
783 return;
784 }
785 name = (const char *) &msg[1];
786 if (name[name_len-1] != '\0')
787 {
788 GNUNET_break (0);
789 return;
790 }
791 if (name_len != strlen (name) + 1)
792 {
793 GNUNET_break (0);
794 return;
795 }
796
797 e = GED_experiments_find (&msg->issuer, name, GNUNET_TIME_absolute_ntoh(msg->version_nbo));
798 if (NULL == e)
799 {
800 GNUNET_break (0);
801 return;
802 }
803 GED_scheduler_handle_start_ack (n, e);
804}
805
806
807/**
808 * Handle a response
809 *
810 * @param peer the source
811 * @param message the message
812 */
813static void
814handle_stop (const struct GNUNET_PeerIdentity *peer,
815 const struct GNUNET_MessageHeader *message)
816{
817 uint16_t size;
818 uint32_t name_len;
819 const struct GED_stop_message *msg;
820 const char *name;
821 struct Node *n;
822 struct Experiment *e;
823
824 if (NULL == peer)
825 {
826 GNUNET_break (0);
827 return;
828 }
829 if (NULL == message)
830 {
831 GNUNET_break (0);
832 return;
833 }
834
835 size = ntohs (message->size);
836 if (size < sizeof (struct GED_stop_message))
837 {
838 GNUNET_break (0);
839 return;
840 }
841 msg = (const struct GED_stop_message *) message;
842 name_len = ntohl (msg->len_name);
843 if (size != sizeof (struct GED_start_message) + name_len)
844 {
845 GNUNET_break (0);
846 return;
847 }
848
849 n = get_node (peer);
850 if (NULL == n)
851 {
852 GNUNET_break (0);
853 return;
854 }
855 name = (const char *) &msg[1];
856 if (name[name_len-1] != '\0')
857 {
858 GNUNET_break (0);
859 return;
860 }
861
862 if (name_len != strlen (name) + 1)
863 {
864 GNUNET_break (0);
865 return;
866 }
867
868 e = GED_experiments_find (&msg->issuer, name, GNUNET_TIME_absolute_ntoh(msg->version_nbo));
869 if (NULL == e)
870 {
871 GNUNET_break (0);
872 return;
873 }
874 GED_scheduler_handle_stop (n, e);
875}
876
877
878/**
879 * Method called whenever a given peer connects.
880 *
881 * @param cls closure
882 * @param peer peer identity this notification is about
883 */
884static void
885core_connect_handler (void *cls,
886 const struct GNUNET_PeerIdentity *peer)
887{
888 if (GNUNET_YES == is_me(peer))
889 return;
890
891 GNUNET_log (GNUNET_ERROR_TYPE_INFO, _("Connected to peer %s\n"),
892 GNUNET_i2s (peer));
893
894 if (GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (nodes_requested, peer))
895 return; /* We already sent a request */
896
897 if (GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (nodes_active, peer))
898 return; /* This peer is known as active */
899
900 if (GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (nodes_inactive, peer))
901 return; /* This peer is known as inactive */
902
903 send_experimentation_request (peer);
904}
905
906
907/**
908 * Method called whenever a given peer disconnects.
909 *
910 * @param cls closure
911 * @param peer peer identity this notification is about
912 */
913static void
914core_disconnect_handler (void *cls,
915 const struct GNUNET_PeerIdentity * peer)
916{
917 struct Node *n;
918 if (GNUNET_YES == is_me(peer))
919 return;
920
921 GNUNET_log (GNUNET_ERROR_TYPE_INFO, _("Disconnected from peer %s\n"),
922 GNUNET_i2s (peer));
923
924 if (NULL != (n = GNUNET_CONTAINER_multipeermap_get (nodes_requested, peer)))
925 cleanup_node (nodes_requested, peer, n);
926
927 if (NULL != (n = GNUNET_CONTAINER_multipeermap_get (nodes_active, peer)))
928 cleanup_node (nodes_active, peer, n);
929
930 if (NULL != (n = GNUNET_CONTAINER_multipeermap_get (nodes_inactive, peer)))
931 cleanup_node (nodes_inactive, peer, n);
932}
933
934
935/**
936 * Handle a request and send a response
937 *
938 * @param cls unused
939 * @param other the sender
940 * @param message the message
941 * @return GNUNET_OK to keep connection, GNUNET_SYSERR on error
942 */
943static int
944core_receive_handler (void *cls,
945 const struct GNUNET_PeerIdentity *other,
946 const struct GNUNET_MessageHeader *message)
947{
948 if (ntohs (message->size) < sizeof (struct GNUNET_MessageHeader))
949 {
950 GNUNET_break (0);
951 return GNUNET_SYSERR;
952 }
953
954 switch (ntohs (message->type)) {
955 case GNUNET_MESSAGE_TYPE_EXPERIMENTATION_REQUEST:
956 handle_request (other, message);
957 break;
958 case GNUNET_MESSAGE_TYPE_EXPERIMENTATION_RESPONSE:
959 handle_response (other, message);
960 break;
961 case GNUNET_MESSAGE_TYPE_EXPERIMENTATION_START:
962 handle_start (other, message);
963 break;
964 case GNUNET_MESSAGE_TYPE_EXPERIMENTATION_START_ACK:
965 handle_start_ack (other, message);
966 break;
967 case GNUNET_MESSAGE_TYPE_EXPERIMENTATION_STOP:
968 handle_stop (other, message);
969 break;
970 default:
971 break;
972 }
973
974 return GNUNET_OK;
975}
976
977
978static size_t
979node_experiment_start_cb (void *cls, size_t bufsize, void *buf)
980{
981 struct NodeComCtx *e_ctx = cls;
982 struct GED_start_message *msg;
983 size_t name_len;
984 size_t size;
985
986 if (NULL == buf)
987 return 0;
988
989 name_len = strlen(e_ctx->e->name) + 1;
990 size = sizeof (struct GED_start_message) + name_len;
991
992 msg = GNUNET_malloc (size);
993 msg->header.size = htons (size);
994 msg->header.type = htons (GNUNET_MESSAGE_TYPE_EXPERIMENTATION_START);
995 msg->issuer = e_ctx->e->issuer;
996 msg->version_nbo = GNUNET_TIME_absolute_hton(e_ctx->e->version);
997 msg->len_name = htonl (name_len);
998 memcpy (&msg[1], e_ctx->e->name, name_len);
999
1000 memcpy (buf, msg, size);
1001 GNUNET_free (msg);
1002 return size;
1003}
1004
1005
1006static size_t
1007node_experiment_start_ack_cb (void *cls, size_t bufsize, void *buf)
1008{
1009 struct NodeComCtx *e_ctx = cls;
1010 struct GED_start_ack_message *msg;
1011 size_t name_len;
1012 size_t size;
1013 if (NULL == buf)
1014 return 0;
1015
1016 name_len = strlen(e_ctx->e->name) + 1;
1017 size = sizeof (struct GED_start_ack_message) + name_len;
1018
1019 msg = GNUNET_malloc (size);
1020 msg->header.size = htons (size);
1021 msg->header.type = htons (GNUNET_MESSAGE_TYPE_EXPERIMENTATION_START_ACK);
1022 msg->issuer = e_ctx->e->issuer;
1023 msg->version_nbo = GNUNET_TIME_absolute_hton(e_ctx->e->version);
1024 msg->len_name = htonl (name_len);
1025 memcpy (&msg[1], e_ctx->e->name, name_len);
1026
1027 memcpy (buf, msg, size);
1028 GNUNET_free (msg);
1029 return size;
1030}
1031
1032
1033
1034
1035/**
1036 * Confirm a experiment START with a node
1037 *
1038 * @return GNUNET_NO if core was busy with sending, GNUNET_OK otherwise
1039 */
1040int
1041GED_nodes_send_start_ack (struct Node *n, struct Experiment *e)
1042{
1043 struct NodeComCtx *e_ctx;
1044
1045 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1046 "Sending %s for experiment request to peer `%s' for experiment `%s'\n",
1047 "START_ACK" ,GNUNET_i2s(&n->id), e->name);
1048
1049 e_ctx = GNUNET_new (struct NodeComCtx);
1050 e_ctx->n = n;
1051 e_ctx->e = e;
1052 e_ctx->size = sizeof (struct GED_start_ack_message) + strlen (e->name) + 1;
1053 e_ctx->notify = &node_experiment_start_ack_cb;
1054 e_ctx->notify_cls = e_ctx;
1055
1056 GNUNET_CONTAINER_DLL_insert_tail (n->e_req_head, n->e_req_tail, e_ctx);
1057 schedule_transmisson (e_ctx);
1058 return GNUNET_OK;
1059}
1060
1061
1062/**
1063 * Request a experiment to start with a node
1064 *
1065 * @return GNUNET_NO if core was busy with sending, GNUNET_OK otherwise
1066 */
1067int
1068GED_nodes_send_start (struct Node *n, struct Experiment *e)
1069{
1070 struct NodeComCtx *e_ctx;
1071
1072 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1073 "Sending %s for experiment request to peer `%s' for experiment `%s'\n",
1074 "START", GNUNET_i2s(&n->id), e->name);
1075
1076 e_ctx = GNUNET_new (struct NodeComCtx);
1077 e_ctx->n = n;
1078 e_ctx->e = e;
1079 e_ctx->size = sizeof (struct GED_start_message) + strlen (e->name) + 1;
1080 e_ctx->notify = &node_experiment_start_cb;
1081 e_ctx->notify_cls = e_ctx;
1082
1083 GNUNET_CONTAINER_DLL_insert_tail (n->e_req_head, n->e_req_tail, e_ctx);
1084 schedule_transmisson (e_ctx);
1085 return GNUNET_OK;
1086}
1087
1088
1089/**
1090 * Start the nodes management
1091 */
1092void
1093GED_nodes_start ()
1094{
1095 /* Connecting to core service to find partners */
1096 ch = GNUNET_CORE_connect (GED_cfg, NULL,
1097 &core_startup_handler,
1098 &core_connect_handler,
1099 &core_disconnect_handler,
1100 &core_receive_handler,
1101 GNUNET_NO, NULL, GNUNET_NO, NULL);
1102 if (NULL == ch)
1103 {
1104 GNUNET_log (GNUNET_ERROR_TYPE_INFO, _("Failed to connect to CORE service!\n"));
1105 return;
1106 }
1107
1108 nodes_requested = GNUNET_CONTAINER_multipeermap_create (10, GNUNET_NO);
1109 nodes_active = GNUNET_CONTAINER_multipeermap_create (10, GNUNET_NO);
1110 nodes_inactive = GNUNET_CONTAINER_multipeermap_create (10, GNUNET_NO);
1111}
1112
1113
1114/**
1115 * Stop the nodes management
1116 */
1117void
1118GED_nodes_stop ()
1119{
1120 if (NULL != ch)
1121 {
1122 GNUNET_CORE_disconnect (ch);
1123 ch = NULL;
1124 }
1125
1126 if (NULL != nodes_requested)
1127 {
1128 GNUNET_CONTAINER_multipeermap_iterate (nodes_requested,
1129 &cleanup_node,
1130 nodes_requested);
1131 update_stats (nodes_requested);
1132 GNUNET_CONTAINER_multipeermap_destroy (nodes_requested);
1133 nodes_requested = NULL;
1134 }
1135
1136 if (NULL != nodes_active)
1137 {
1138 GNUNET_CONTAINER_multipeermap_iterate (nodes_active,
1139 &cleanup_node,
1140 nodes_active);
1141 update_stats (nodes_active);
1142 GNUNET_CONTAINER_multipeermap_destroy (nodes_active);
1143 nodes_active = NULL;
1144 }
1145
1146 if (NULL != nodes_inactive)
1147 {
1148 GNUNET_CONTAINER_multipeermap_iterate (nodes_inactive,
1149 &cleanup_node,
1150 nodes_inactive);
1151 update_stats (nodes_inactive);
1152 GNUNET_CONTAINER_multipeermap_destroy (nodes_inactive);
1153 nodes_inactive = NULL;
1154 }
1155}
1156
1157/* end of gnunet-daemon-experimentation_nodes.c */
generated by cgit v1.2.3 (git 2.39.1) at 2024-04-19 21:36:29 +0000