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 static size_t
 264 send_experimentation_request_cb (void *cls, size_t bufsize, void *buf)
 265 {
 266   struct Node *n = cls;
 267   struct Experimentation_Request msg;
 268   size_t msg_size = sizeof (msg);
 269   size_t ri_size = sizeof (struct Experimentation_Issuer) * GSE_my_issuer_count;
 270   size_t total_size = msg_size + ri_size;
 271
 272   n->cth = NULL;
 273   if (NULL == buf)
 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   memset (buf, '\0', bufsize);
 283   GNUNET_assert (bufsize >= total_size);
 284
 285         msg.msg.size = htons (total_size);
 286         msg.msg.type = htons (GNUNET_MESSAGE_TYPE_EXPERIMENTATION_REQUEST);
 287         msg.capabilities = htonl (GSE_node_capabilities);
 288         msg.issuer_count = htonl (GSE_my_issuer_count);
 289         memcpy (buf, &msg, msg_size);
 290         memcpy (&((char *) buf)[msg_size], GSE_my_issuer, ri_size);
 291
 292         GNUNET_log (GNUNET_ERROR_TYPE_INFO, _("Sending experimentation request to peer %s\n"),
 293                         GNUNET_i2s (&n->id));
 294         return total_size;
 295 }
 296
 297
 298 /**
 299  * Send request to peer to start add him to to the set of experimentation nodes
 300  *
 301  * @param peer the peer to send to
 302  */
 303 static void
 304 send_experimentation_request (const struct GNUNET_PeerIdentity *peer)
 305 {
 306         struct Node *n;
 307         struct NodeComCtx *e_ctx;
 308         size_t size;
 309         size_t c_issuers;
 310
 311         c_issuers = GSE_my_issuer_count;
 312
 313         size = sizeof (struct Experimentation_Request) +
 314                                  c_issuers * sizeof (struct Experimentation_Issuer);
 315         n = GNUNET_malloc (sizeof (struct Node));
 316         n->id = *peer;
 317         n->timeout_task = GNUNET_SCHEDULER_add_delayed (EXP_RESPONSE_TIMEOUT, &remove_request, n);
 318         n->capabilities = NONE;
 319
 320         e_ctx = GNUNET_malloc (sizeof (struct NodeComCtx));
 321         e_ctx->n = n;
 322         e_ctx->e = NULL;
 323         e_ctx->size = size;
 324         e_ctx->notify = &send_experimentation_request_cb;
 325         e_ctx->notify_cls = n;
 326         GNUNET_CONTAINER_DLL_insert_tail(n->e_req_head, n->e_req_tail, e_ctx);
 327         schedule_transmisson (e_ctx);
 328
 329         GNUNET_assert (GNUNET_OK == GNUNET_CONTAINER_multihashmap_put (nodes_requested,
 330                         &peer->hashPubKey, n, GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
 331         update_stats (nodes_requested);
 332 }
 333
 334
 335 /**
 336  * Core's transmit notify callback to send response
 337  *
 338  * @param cls the related node
 339  * @param bufsize buffer size
 340  * @param buf the buffer to copy to
 341  * @return bytes passed
 342  */
 343 size_t send_response_cb (void *cls, size_t bufsize, void *buf)
 344 {
 345         struct Node *n = cls;
 346         struct Experimentation_Response msg;
 347         size_t ri_size = GSE_my_issuer_count * sizeof (struct Experimentation_Issuer);
 348         size_t msg_size = sizeof (msg);
 349         size_t total_size = msg_size + ri_size;
 350
 351         n->cth = NULL;
 352   if (buf == NULL)
 353   {
 354     /* client disconnected */
 355     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Client disconnected\n");
 356     return 0;
 357   }
 358   GNUNET_assert (bufsize >= total_size);
 359
 360         msg.msg.size = htons (total_size);
 361         msg.msg.type = htons (GNUNET_MESSAGE_TYPE_EXPERIMENTATION_RESPONSE);
 362         msg.capabilities = htonl (GSE_node_capabilities);
 363         msg.issuer_count = htonl (GSE_my_issuer_count);
 364         memcpy (buf, &msg, msg_size);
 365         memcpy (&((char *) buf)[msg_size], GSE_my_issuer, ri_size);
 366
 367         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending response to peer %s\n",
 368                         GNUNET_i2s (&n->id));
 369         return total_size;
 370 }
 371
 372
 373 static void
 374 get_experiments_cb (struct Node *n, struct Experiment *e)
 375 {
 376         static int counter = 0;
 377         if (NULL == e)
 378                         return; /* Done */
 379
 380         /* Tell the scheduler to add a node with an experiment */
 381         GED_scheduler_add (n, e, GNUNET_YES);
 382         counter ++;
 383 }
 384
 385 struct Node *
 386 get_node (const struct GNUNET_PeerIdentity *id)
 387 {
 388         struct Node * res;
 389         struct Node * tmp;
 390
 391         res = NULL;
 392         tmp = NULL;
 393         tmp = GNUNET_CONTAINER_multihashmap_get (nodes_active, &id->hashPubKey);
 394         if (res == NULL)
 395                 res = tmp;
 396
 397         tmp = GNUNET_CONTAINER_multihashmap_get (nodes_inactive, &id->hashPubKey);
 398         if (res == NULL)
 399                 res = tmp;
 400         else
 401                 GNUNET_break (0); /* Multiple instances */
 402
 403         tmp = GNUNET_CONTAINER_multihashmap_get (nodes_requested, &id->hashPubKey);
 404         if (res == NULL)
 405                 res = tmp;
 406         else
 407                 GNUNET_break (0); /* Multiple instances */
 408
 409         return res;
 410 }
 411
 412
 413 /**
 414  * Set a specific node as active
 415  *
 416  * @param n the node
 417  */
 418 static void node_make_active (struct Node *n)
 419 {
 420         int c1;
 421   GNUNET_CONTAINER_multihashmap_put (nodes_active,
 422                         &n->id.hashPubKey, n, GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
 423         update_stats (nodes_active);
 424         GNUNET_log (GNUNET_ERROR_TYPE_INFO, _("Added peer `%s' as active node\n"),
 425                         GNUNET_i2s (&n->id));
 426         /* Request experiments for this node to start them */
 427         for (c1 = 0; c1 < n->issuer_count; c1++)
 428         {
 429
 430                 GED_experiments_get (n, &n->issuer_id[c1], &get_experiments_cb);
 431         }
 432 }
 433
 434
 435 /**
 436  * Handle a request and send a response
 437  *
 438  * @param peer the source
 439  * @param message the message
 440  */
 441 static void handle_request (const struct GNUNET_PeerIdentity *peer,
 442                                                                                                                 const struct GNUNET_MessageHeader *message)
 443 {
 444         struct Node *n;
 445         struct NodeComCtx *e_ctx;
 446         struct Experimentation_Request *rm = (struct Experimentation_Request *) message;
 447         struct Experimentation_Issuer *rmi = (struct Experimentation_Issuer *) &rm[1];
 448         int c1;
 449         int c2;
 450         uint32_t ic;
 451         uint32_t ic_accepted;
 452         int make_active;
 453
 454         if (ntohs (message->size) < sizeof (struct Experimentation_Request))
 455         {
 456                 GNUNET_break (0);
 457                 return;
 458         }
 459         ic = ntohl (rm->issuer_count);
 460         if (ntohs (message->size) != sizeof (struct Experimentation_Request) + ic * sizeof (struct Experimentation_Issuer))
 461         {
 462                 GNUNET_break (0);
 463                 return;
 464         }
 465
 466         make_active = GNUNET_NO;
 467         if (NULL != (n = GNUNET_CONTAINER_multihashmap_get (nodes_active, &peer->hashPubKey)))
 468         {
 469                         /* Nothing to do */
 470         }
 471         else if (NULL != (n = GNUNET_CONTAINER_multihashmap_get (nodes_requested, &peer->hashPubKey)))
 472         {
 473                         GNUNET_CONTAINER_multihashmap_remove (nodes_requested, &peer->hashPubKey, n);
 474                         if (GNUNET_SCHEDULER_NO_TASK != n->timeout_task)
 475                         {
 476                                 GNUNET_SCHEDULER_cancel (n->timeout_task);
 477                                 n->timeout_task = GNUNET_SCHEDULER_NO_TASK;
 478                         }
 479                         update_stats (nodes_requested);
 480                         make_active = GNUNET_YES;
 481         }
 482         else if (NULL != (n = GNUNET_CONTAINER_multihashmap_get (nodes_inactive, &peer->hashPubKey)))
 483         {
 484                   GNUNET_break (0 == GNUNET_CONTAINER_multihashmap_remove (nodes_inactive, &peer->hashPubKey, n));
 485                         update_stats (nodes_inactive);
 486                         make_active = GNUNET_YES;
 487         }
 488         else
 489         {
 490                         /* Create new node */
 491                         n = GNUNET_malloc (sizeof (struct Node));
 492                         n->id = *peer;
 493                         n->capabilities = NONE;
 494                         make_active = GNUNET_YES;
 495         }
 496
 497         /* Update node */
 498         n->capabilities = ntohl (rm->capabilities);
 499
 500         /* Filter accepted issuer */
 501         ic_accepted = 0;
 502         for (c1 = 0; c1 < ic; c1++)
 503         {
 504                 if (GNUNET_YES == GED_experiments_issuer_accepted(&rmi[c1].issuer_id))
 505                         ic_accepted ++;
 506         }
 507         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Request from peer `%s' with %u issuers, we accepted %u issuer \n",
 508                         GNUNET_i2s (peer), ic, ic_accepted);
 509         GNUNET_free_non_null (n->issuer_id);
 510         n->issuer_id = GNUNET_malloc (ic_accepted * sizeof (struct GNUNET_PeerIdentity));
 511         c2 = 0;
 512         for (c1 = 0; c1 < ic; c1++)
 513         {
 514                         if (GNUNET_YES == GED_experiments_issuer_accepted(&rmi[c1].issuer_id))
 515                         {
 516                                 n->issuer_id[c2] = rmi[c1].issuer_id;
 517                                 c2 ++;
 518                         }
 519         }
 520         n->issuer_count = ic_accepted;
 521
 522         if (GNUNET_YES == make_active)
 523                 node_make_active (n);
 524
 525         /* Send response */
 526         e_ctx = GNUNET_malloc (sizeof (struct NodeComCtx));
 527         e_ctx->n = n;
 528         e_ctx->e = NULL;
 529         e_ctx->size = sizeof (struct Experimentation_Response) + GSE_my_issuer_count * sizeof (struct Experimentation_Issuer);
 530         e_ctx->notify = &send_response_cb;
 531         e_ctx->notify_cls = n;
 532
 533         GNUNET_CONTAINER_DLL_insert_tail(n->e_req_head, n->e_req_tail, e_ctx);
 534         schedule_transmisson (e_ctx);
 535 }
 536
 537
 538 /**
 539  * Handle a response
 540  *
 541  * @param peer the source
 542  * @param message the message
 543  */
 544 static void handle_response (const struct GNUNET_PeerIdentity *peer,
 545                                                                                                                  const struct GNUNET_MessageHeader *message)
 546 {
 547         struct Node *n;
 548         struct Experimentation_Response *rm = (struct Experimentation_Response *) message;
 549         struct Experimentation_Issuer *rmi = (struct Experimentation_Issuer *) &rm[1];
 550         uint32_t ic;
 551         uint32_t ic_accepted;
 552         int make_active;
 553         unsigned int c1;
 554         unsigned int c2;
 555
 556         if (ntohs (message->size) < sizeof (struct Experimentation_Response))
 557         {
 558                 GNUNET_break (0);
 559                 return;
 560         }
 561         ic = ntohl (rm->issuer_count);
 562         if (ntohs (message->size) != sizeof (struct Experimentation_Response) + ic * sizeof (struct Experimentation_Issuer))
 563         {
 564                 GNUNET_break (0);
 565                 return;
 566         }
 567
 568         make_active = GNUNET_NO;
 569         if (NULL != (n = GNUNET_CONTAINER_multihashmap_get (nodes_active, &peer->hashPubKey)))
 570         {
 571                         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Received %s from %s peer `%s'\n",
 572                                         "RESPONSE", "active", GNUNET_i2s (peer));
 573         }
 574         else if (NULL != (n = GNUNET_CONTAINER_multihashmap_get (nodes_requested, &peer->hashPubKey)))
 575         {
 576                         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Received %s from %s peer `%s'\n",
 577                                         "RESPONSE", "requested", GNUNET_i2s (peer));
 578                         GNUNET_CONTAINER_multihashmap_remove (nodes_requested, &peer->hashPubKey, n);
 579                         if (GNUNET_SCHEDULER_NO_TASK != n->timeout_task)
 580                         {
 581                                 GNUNET_SCHEDULER_cancel (n->timeout_task);
 582                                 n->timeout_task = GNUNET_SCHEDULER_NO_TASK;
 583                         }
 584                         update_stats (nodes_requested);
 585                         make_active = GNUNET_YES;
 586         }
 587         else if (NULL != (n = GNUNET_CONTAINER_multihashmap_get (nodes_inactive, &peer->hashPubKey)))
 588         {
 589                         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Received %s from peer `%s'\n",
 590                                         "RESPONSE", "inactive", GNUNET_i2s (peer));
 591                         GNUNET_break (0 == GNUNET_CONTAINER_multihashmap_remove (nodes_inactive, &peer->hashPubKey, n));
 592                         update_stats (nodes_inactive);
 593                         make_active = GNUNET_YES;
 594         }
 595         else
 596         {
 597                         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Received %s from %s peer `%s'\n",
 598                                         "RESPONSE", "unknown", GNUNET_i2s (peer));
 599                         return;
 600         }
 601
 602         /* Update */
 603         n->capabilities = ntohl (rm->capabilities);
 604
 605         /* Filter accepted issuer */
 606         ic_accepted = 0;
 607         for (c1 = 0; c1 < ic; c1++)
 608         {
 609                 if (GNUNET_YES == GED_experiments_issuer_accepted(&rmi[c1].issuer_id))
 610                         ic_accepted ++;
 611         }
 612         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Response from peer `%s' with %u issuers, we accepted %u issuer \n",
 613                         GNUNET_i2s (peer), ic, ic_accepted);
 614         GNUNET_free_non_null (n->issuer_id);
 615         n->issuer_id = GNUNET_malloc (ic_accepted * sizeof (struct GNUNET_PeerIdentity));
 616         c2 = 0;
 617         for (c1 = 0; c1 < ic; c1++)
 618         {
 619                         if (GNUNET_YES == GED_experiments_issuer_accepted(&rmi[c1].issuer_id))
 620                         {
 621                                 n->issuer_id[c2] = rmi[c1].issuer_id;
 622                                 c2 ++;
 623                         }
 624         }
 625         n->issuer_count = ic_accepted;
 626
 627         if (GNUNET_YES == make_active)
 628                 node_make_active (n);
 629 }
 630
 631 /**
 632  * Handle a response
 633  *
 634  * @param peer the source
 635  * @param message the message
 636  */
 637 static void handle_start (const struct GNUNET_PeerIdentity *peer,
 638                                                                                                                  const struct GNUNET_MessageHeader *message)
 639 {
 640         uint16_t size;
 641         uint32_t name_len;
 642         const struct GED_start_message *msg;
 643         const char *name;
 644         struct Node *n;
 645         struct Experiment *e;
 646
 647         if (NULL == peer)
 648         {
 649                 GNUNET_break (0);
 650                 return;
 651         }
 652         if (NULL == message)
 653         {
 654                 GNUNET_break (0);
 655                 return;
 656         }
 657
 658         size = ntohs (message->size);
 659         if (size < sizeof (struct GED_start_message))
 660         {
 661                 GNUNET_break (0);
 662                 return;
 663         }
 664         msg = (const struct GED_start_message *) message;
 665         name_len = ntohl (msg->len_name);
 666         if (size != sizeof (struct GED_start_message) + name_len)
 667         {
 668                 GNUNET_break (0);
 669                 return;
 670         }
 671
 672         n = get_node (peer);
 673         if (NULL == n)
 674         {
 675                 GNUNET_break (0);
 676                 return;
 677         }
 678         name = (const char *) &msg[1];
 679         if (name[name_len-1] != '\0')
 680         {
 681                 GNUNET_break (0);
 682                 return;
 683         }
 684
 685         if (name_len != strlen (name) + 1)
 686         {
 687                 GNUNET_break (0);
 688                 return;
 689         }
 690
 691         e = GED_experiments_find (&msg->issuer, name, GNUNET_TIME_absolute_ntoh(msg->version_nbo));
 692         if (NULL == e)
 693         {
 694                 GNUNET_break (0);
 695                 return;
 696         }
 697
 698         GED_scheduler_handle_start (n, e);
 699 }
 700
 701 /**
 702  * Handle a response
 703  *
 704  * @param peer the source
 705  * @param message the message
 706  */
 707 static void handle_start_ack (const struct GNUNET_PeerIdentity *peer,
 708                                                                                                                  const struct GNUNET_MessageHeader *message)
 709 {
 710         uint16_t size;
 711         uint32_t name_len;
 712         const struct GED_start_ack_message *msg;
 713         const char *name;
 714         struct Node *n;
 715         struct Experiment *e;
 716
 717         if (NULL == peer)
 718         {
 719                 GNUNET_break (0);
 720                 return;
 721         }
 722         if (NULL == message)
 723         {
 724                 GNUNET_break (0);
 725                 return;
 726         }
 727
 728         size = ntohs (message->size);
 729         if (size < sizeof (struct GED_start_ack_message))
 730         {
 731                 GNUNET_break (0);
 732                 return;
 733         }
 734         msg = (const struct GED_start_ack_message *) message;
 735         name_len = ntohl (msg->len_name);
 736         if (size != sizeof (struct GED_start_message) + name_len)
 737         {
 738                 GNUNET_break (0);
 739                 return;
 740         }
 741
 742         n = get_node (peer);
 743         if (NULL == n)
 744         {
 745                 GNUNET_break (0);
 746                 return;
 747         }
 748         name = (const char *) &msg[1];
 749         if (name[name_len-1] != '\0')
 750         {
 751                 GNUNET_break (0);
 752                 return;
 753         }
 754
 755         if (name_len != strlen (name) + 1)
 756         {
 757                 GNUNET_break (0);
 758                 return;
 759         }
 760
 761         e = GED_experiments_find (&msg->issuer, name, GNUNET_TIME_absolute_ntoh(msg->version_nbo));
 762         if (NULL == e)
 763         {
 764                 GNUNET_break (0);
 765                 return;
 766         }
 767         GED_scheduler_handle_start_ack (n, e);
 768 }
 769
 770 /**
 771  * Handle a response
 772  *
 773  * @param peer the source
 774  * @param message the message
 775  */
 776 static void handle_stop (const struct GNUNET_PeerIdentity *peer,
 777                                                                                                  const struct GNUNET_MessageHeader *message)
 778 {
 779         uint16_t size;
 780         uint32_t name_len;
 781         const struct GED_stop_message *msg;
 782         const char *name;
 783         struct Node *n;
 784         struct Experiment *e;
 785
 786         if (NULL == peer)
 787         {
 788                 GNUNET_break (0);
 789                 return;
 790         }
 791         if (NULL == message)
 792         {
 793                 GNUNET_break (0);
 794                 return;
 795         }
 796
 797         size = ntohs (message->size);
 798         if (size < sizeof (struct GED_stop_message))
 799         {
 800                 GNUNET_break (0);
 801                 return;
 802         }
 803         msg = (const struct GED_stop_message *) message;
 804         name_len = ntohl (msg->len_name);
 805         if (size != sizeof (struct GED_start_message) + name_len)
 806         {
 807                 GNUNET_break (0);
 808                 return;
 809         }
 810
 811         n = get_node (peer);
 812         if (NULL == n)
 813         {
 814                 GNUNET_break (0);
 815                 return;
 816         }
 817         name = (const char *) &msg[1];
 818         if (name[name_len-1] != '\0')
 819         {
 820                 GNUNET_break (0);
 821                 return;
 822         }
 823
 824         if (name_len != strlen (name) + 1)
 825         {
 826                 GNUNET_break (0);
 827                 return;
 828         }
 829
 830         e = GED_experiments_find (&msg->issuer, name, GNUNET_TIME_absolute_ntoh(msg->version_nbo));
 831         if (NULL == e)
 832         {
 833                 GNUNET_break (0);
 834                 return;
 835         }
 836         GED_scheduler_handle_stop (n, e);
 837 }
 838
 839 /**
 840  * Method called whenever a given peer connects.
 841  *
 842  * @param cls closure
 843  * @param peer peer identity this notification is about
 844  */
 845 void core_connect_handler (void *cls,
 846                            const struct GNUNET_PeerIdentity *peer)
 847 {
 848         if (GNUNET_YES == is_me(peer))
 849                 return;
 850
 851         GNUNET_log (GNUNET_ERROR_TYPE_INFO, _("Connected to peer %s\n"),
 852                         GNUNET_i2s (peer));
 853
 854         if (GNUNET_YES == GNUNET_CONTAINER_multihashmap_contains (nodes_requested, &peer->hashPubKey))
 855                 return; /* We already sent a request */
 856
 857         if (GNUNET_YES == GNUNET_CONTAINER_multihashmap_contains (nodes_active, &peer->hashPubKey))
 858                 return; /* This peer is known as active  */
 859
 860         if (GNUNET_YES == GNUNET_CONTAINER_multihashmap_contains (nodes_inactive, &peer->hashPubKey))
 861                 return; /* This peer is known as inactive  */
 862
 863         send_experimentation_request (peer);
 864 }
 865
 866
 867 /**
 868  * Method called whenever a given peer disconnects.
 869  *
 870  * @param cls closure
 871  * @param peer peer identity this notification is about
 872  */
 873 void core_disconnect_handler (void *cls,
 874                            const struct GNUNET_PeerIdentity * peer)
 875 {
 876         struct Node *n;
 877         if (GNUNET_YES == is_me(peer))
 878                 return;
 879
 880         GNUNET_log (GNUNET_ERROR_TYPE_INFO, _("Disconnected from peer %s\n"),
 881                         GNUNET_i2s (peer));
 882
 883         if (NULL != (n = GNUNET_CONTAINER_multihashmap_get (nodes_requested, &peer->hashPubKey)))
 884                 cleanup_node (nodes_requested, &peer->hashPubKey, n);
 885
 886         if (NULL != (n = GNUNET_CONTAINER_multihashmap_get (nodes_active, &peer->hashPubKey)))
 887                 cleanup_node (nodes_active, &peer->hashPubKey, n);
 888
 889         if (NULL != (n = GNUNET_CONTAINER_multihashmap_get (nodes_inactive, &peer->hashPubKey)))
 890                 cleanup_node (nodes_inactive, &peer->hashPubKey, n);
 891 }
 892
 893
 894 /**
 895  * Handle a request and send a response
 896  *
 897  * @param cls unused
 898  * @param other the sender
 899  * @param message the message
 900  * @return GNUNET_OK to keep connection, GNUNET_SYSERR on error
 901  */
 902 static int
 903 core_receive_handler (void *cls,
 904                                                                                         const struct GNUNET_PeerIdentity *other,
 905                                                                                         const struct GNUNET_MessageHeader *message)
 906 {
 907         if (ntohs (message->size) < sizeof (struct GNUNET_MessageHeader))
 908         {
 909                         GNUNET_break (0);
 910                         return GNUNET_SYSERR;
 911         }
 912
 913         switch (ntohs (message->type)) {
 914                 case GNUNET_MESSAGE_TYPE_EXPERIMENTATION_REQUEST:
 915                         handle_request (other, message);
 916                         break;
 917                 case GNUNET_MESSAGE_TYPE_EXPERIMENTATION_RESPONSE:
 918                         handle_response (other, message);
 919                         break;
 920                 case GNUNET_MESSAGE_TYPE_EXPERIMENTATION_START:
 921                         handle_start (other, message);
 922                         break;
 923                 case GNUNET_MESSAGE_TYPE_EXPERIMENTATION_START_ACK:
 924                         handle_start_ack (other, message);
 925                         break;
 926                 case GNUNET_MESSAGE_TYPE_EXPERIMENTATION_STOP:
 927                         handle_stop (other, message);
 928                         break;
 929                 default:
 930                         break;
 931         }
 932
 933         return GNUNET_OK;
 934 }
 935
 936
 937 size_t node_experiment_start_cb (void *cls, size_t bufsize, void *buf)
 938 {
 939         struct NodeComCtx *e_ctx = cls;
 940         struct GED_start_message *msg;
 941         size_t name_len;
 942         size_t size;
 943
 944         if (NULL == buf)
 945                 return 0;
 946
 947         name_len = strlen(e_ctx->e->name) + 1;
 948         size = sizeof (struct GED_start_message) + name_len;
 949
 950         msg = GNUNET_malloc (size);
 951         msg->header.size = htons (size);
 952         msg->header.type = htons (GNUNET_MESSAGE_TYPE_EXPERIMENTATION_START);
 953         msg->issuer = e_ctx->e->issuer;
 954         msg->version_nbo = GNUNET_TIME_absolute_hton(e_ctx->e->version);
 955         msg->len_name = htonl (name_len);
 956         memcpy (&msg[1], e_ctx->e->name, name_len);
 957
 958         memcpy (buf, msg, size);
 959         GNUNET_free (msg);
 960         return size;
 961 }
 962
 963 size_t node_experiment_start_ack_cb (void *cls, size_t bufsize, void *buf)
 964 {
 965         struct NodeComCtx *e_ctx = cls;
 966         struct GED_start_ack_message *msg;
 967         size_t name_len;
 968         size_t size;
 969         if (NULL == buf)
 970                 return 0;
 971
 972         name_len = strlen(e_ctx->e->name) + 1;
 973         size = sizeof (struct GED_start_ack_message) + name_len;
 974
 975         msg = GNUNET_malloc (size);
 976         msg->header.size = htons (size);
 977         msg->header.type = htons (GNUNET_MESSAGE_TYPE_EXPERIMENTATION_START_ACK);
 978         msg->issuer = e_ctx->e->issuer;
 979         msg->version_nbo = GNUNET_TIME_absolute_hton(e_ctx->e->version);
 980         msg->len_name = htonl (name_len);
 981         memcpy (&msg[1], e_ctx->e->name, name_len);
 982
 983         memcpy (buf, msg, size);
 984         GNUNET_free (msg);
 985         return size;
 986 }
 987
 988
 989
 990
 991 /**
 992  * Confirm a experiment START with a node
 993  *
 994  * @return GNUNET_NO if core was busy with sending, GNUNET_OK otherwise
 995  */
 996 int
 997 GED_nodes_send_start_ack (struct Node *n, struct Experiment *e)
 998 {
 999         struct NodeComCtx *e_ctx;
1000
1001         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1002                         "Sending %s for experiment request to peer `%s' for experiment `%s'\n",
1003                         "START_ACK" ,GNUNET_i2s(&n->id), e->name);
1004
1005         e_ctx = GNUNET_malloc (sizeof (struct NodeComCtx));
1006         e_ctx->n = n;
1007         e_ctx->e = e;
1008         e_ctx->size = sizeof (struct GED_start_ack_message) + strlen (e->name) + 1;
1009         e_ctx->notify = &node_experiment_start_ack_cb;
1010         e_ctx->notify_cls = e_ctx;
1011
1012         GNUNET_CONTAINER_DLL_insert_tail (n->e_req_head, n->e_req_tail, e_ctx);
1013         schedule_transmisson (e_ctx);
1014         return GNUNET_OK;
1015 }
1016
1017
1018 /**
1019  * Request a experiment to start with a node
1020  *
1021  * @return GNUNET_NO if core was busy with sending, GNUNET_OK otherwise
1022  */
1023 int
1024 GED_nodes_send_start (struct Node *n, struct Experiment *e)
1025 {
1026         struct NodeComCtx *e_ctx;
1027
1028         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1029                         "Sending %s for experiment request to peer `%s' for experiment `%s'\n",
1030                         "START", GNUNET_i2s(&n->id), e->name);
1031
1032         e_ctx = GNUNET_malloc (sizeof (struct NodeComCtx));
1033         e_ctx->n = n;
1034         e_ctx->e = e;
1035         e_ctx->size = sizeof (struct GED_start_message) + strlen (e->name) + 1;
1036         e_ctx->notify = &node_experiment_start_cb;
1037         e_ctx->notify_cls = e_ctx;
1038
1039         GNUNET_CONTAINER_DLL_insert_tail (n->e_req_head, n->e_req_tail, e_ctx);
1040         schedule_transmisson (e_ctx);
1041         return GNUNET_OK;
1042 }
1043
1044
1045 /**
1046  * Start the nodes management
1047  */
1048 void
1049 GED_nodes_start ()
1050 {
1051         /* Connecting to core service to find partners */
1052         ch = GNUNET_CORE_connect (GED_cfg, NULL,
1053                                                                                                                 &core_startup_handler,
1054                                                                                                                 &core_connect_handler,
1055                                                                                                                 &core_disconnect_handler,
1056                                                                                                                 &core_receive_handler,
1057                                                                                                                 GNUNET_NO, NULL, GNUNET_NO, NULL);
1058         if (NULL == ch)
1059         {
1060                         GNUNET_log (GNUNET_ERROR_TYPE_INFO, _("Failed to connect to CORE service!\n"));
1061                         return;
1062         }
1063
1064         nodes_requested = GNUNET_CONTAINER_multihashmap_create (10, GNUNET_NO);
1065         nodes_active = GNUNET_CONTAINER_multihashmap_create (10, GNUNET_NO);
1066         nodes_inactive = GNUNET_CONTAINER_multihashmap_create (10, GNUNET_NO);
1067 }
1068
1069
1070 /**
1071  * Stop the nodes management
1072  */
1073 void
1074 GED_nodes_stop ()
1075 {
1076   if (NULL != ch)
1077   {
1078                 GNUNET_CORE_disconnect (ch);
1079                 ch = NULL;
1080   }
1081
1082   if (NULL != nodes_requested)
1083   {
1084                 GNUNET_CONTAINER_multihashmap_iterate (nodes_requested,
1085                                                                                                                                                                          &cleanup_node,
1086                                                                                                                                                                          nodes_requested);
1087                 update_stats (nodes_requested);
1088                 GNUNET_CONTAINER_multihashmap_destroy (nodes_requested);
1089                 nodes_requested = NULL;
1090   }
1091
1092   if (NULL != nodes_active)
1093   {
1094                 GNUNET_CONTAINER_multihashmap_iterate (nodes_active,
1095                                                                                                                                                                          &cleanup_node,
1096                                                                                                                                                                          nodes_active);
1097                 update_stats (nodes_active);
1098                 GNUNET_CONTAINER_multihashmap_destroy (nodes_active);
1099                 nodes_active = NULL;
1100   }
1101
1102   if (NULL != nodes_inactive)
1103   {
1104                 GNUNET_CONTAINER_multihashmap_iterate (nodes_inactive,
1105                                                                                                                                                                          &cleanup_node,
1106                                                                                                                                                                          nodes_inactive);
1107                 update_stats (nodes_inactive);
1108                 GNUNET_CONTAINER_multihashmap_destroy (nodes_inactive);
1109                 nodes_inactive = NULL;
1110   }
1111 }
1112
1113 /* end of gnunet-daemon-experimentation_nodes.c */