src/experimentation/gnunet-daemon-experimentation_nodes.c

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