src/rps/gnunet-service-rps_peers.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C)
   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., 51 Franklin Street, Fifth Floor,
  18      Boston, MA 02110-1301, USA.
  19 */
  20
  21 /**
  22  * @file rps/gnunet-service-rps_peers.c
  23  * @brief utilities for managing (information about) peers
  24  * @author Julius Bünger
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "gnunet_cadet_service.h"
  29 #include <inttypes.h>
  30 #include "rps.h"
  31 #include "gnunet-service-rps_peers.h"
  32
  33
  34
  35 #define LOG(kind, ...) GNUNET_log(kind, __VA_ARGS__)
  36
  37
  38 /**
  39  * Set a peer flag of given peer context.
  40  */
  41 #define set_peer_flag(peer_ctx, mask) ((peer_ctx->peer_flags) |= (mask))
  42
  43 /**
  44  * Get peer flag of given peer context.
  45  */
  46 #define check_peer_flag_set(peer_ctx, mask)\
  47   ((peer_ctx->peer_flags) & (mask) ? GNUNET_YES : GNUNET_NO)
  48
  49 /**
  50  * Unset flag of given peer context.
  51  */
  52 #define unset_peer_flag(peer_ctx, mask) ((peer_ctx->peer_flags) &= ~(mask))
  53
  54 /**
  55  * Set a channel flag of given channel context.
  56  */
  57 #define set_channel_flag(channel_flags, mask) ((*channel_flags) |= (mask))
  58
  59 /**
  60  * Get channel flag of given channel context.
  61  */
  62 #define check_channel_flag_set(channel_flags, mask)\
  63   ((*channel_flags) & (mask) ? GNUNET_YES : GNUNET_NO)
  64
  65 /**
  66  * Unset flag of given channel context.
  67  */
  68 #define unset_channel_flag(channel_flags, mask) ((*channel_flags) &= ~(mask))
  69
  70
  71
  72 /**
  73  * Pending operation on peer consisting of callback and closure
  74  *
  75  * When an operation cannot be executed right now this struct is used to store
  76  * the callback and closure for later execution.
  77  */
  78 struct PeerPendingOp
  79 {
  80   /**
  81    * Callback
  82    */
  83   PeerOp op;
  84
  85   /**
  86    * Closure
  87    */
  88   void *op_cls;
  89 };
  90
  91 /**
  92  * List containing all messages that are yet to be send
  93  *
  94  * This is used to keep track of all messages that have not been sent yet. When
  95  * a peer is to be removed the pending messages can be removed properly.
  96  */
  97 struct PendingMessage
  98 {
  99   /**
 100    * DLL next, prev
 101    */
 102   struct PendingMessage *next;
 103   struct PendingMessage *prev;
 104
 105   /**
 106    * The envelope to the corresponding message
 107    */
 108   struct GNUNET_MQ_Envelope *ev;
 109
 110   /**
 111    * The corresponding context
 112    */
 113   struct PeerContext *peer_ctx;
 114
 115   /**
 116    * The message type
 117    */
 118   const char *type;
 119 };
 120
 121 /**
 122  * Struct used to keep track of other peer's status
 123  *
 124  * This is stored in a multipeermap.
 125  * It contains information such as cadet channels, a message queue for sending,
 126  * status about the channels, the pending operations on this peer and some flags
 127  * about the status of the peer itself. (live, valid, ...)
 128  */
 129 struct PeerContext
 130 {
 131   /**
 132    * Message queue open to client
 133    */
 134   struct GNUNET_MQ_Handle *mq;
 135
 136   /**
 137    * Channel open to client.
 138    */
 139   struct GNUNET_CADET_Channel *send_channel;
 140
 141   /**
 142    * Flags to the sending channel
 143    */
 144   uint32_t *send_channel_flags;
 145
 146   /**
 147    * Channel open from client.
 148    */
 149   struct GNUNET_CADET_Channel *recv_channel; // unneeded?
 150
 151   /**
 152    * Flags to the receiving channel
 153    */
 154   uint32_t *recv_channel_flags;
 155
 156   /**
 157    * Array of pending operations on this peer.
 158    */
 159   struct PeerPendingOp *pending_ops;
 160
 161   /**
 162    * Handle to the callback given to cadet_ntfy_tmt_rdy()
 163    *
 164    * To be canceled on shutdown.
 165    */
 166   struct GNUNET_CADET_TransmitHandle *transmit_handle;
 167
 168   /**
 169    * Number of pending operations.
 170    */
 171   unsigned int num_pending_ops;
 172
 173   /**
 174    * Identity of the peer
 175    */
 176   struct GNUNET_PeerIdentity peer_id;
 177
 178   /**
 179    * Flags indicating status of peer
 180    */
 181   uint32_t peer_flags;
 182
 183   /**
 184    * Last time we received something from that peer.
 185    */
 186   struct GNUNET_TIME_Absolute last_message_recv;
 187
 188   /**
 189    * Last time we received a keepalive message.
 190    */
 191   struct GNUNET_TIME_Absolute last_keepalive;
 192
 193   /**
 194    * DLL with all messages that are yet to be sent
 195    */
 196   struct PendingMessage *pending_messages_head;
 197   struct PendingMessage *pending_messages_tail;
 198
 199   /**
 200    * This is pobably followed by 'statistical' data (when we first saw
 201    * him, how did we get his ID, how many pushes (in a timeinterval),
 202    * ...)
 203    */
 204 };
 205
 206
 207 /**
 208  * Set of all peers to keep track of them.
 209  */
 210 static struct GNUNET_CONTAINER_MultiPeerMap *peer_map;
 211
 212 /**
 213  * Own #GNUNET_PeerIdentity.
 214  */
 215 static const struct GNUNET_PeerIdentity *own_identity;
 216
 217 /**
 218  * Cadet handle.
 219  */
 220 static struct GNUNET_CADET_Handle *cadet_handle;
 221
 222
 223
 224 /**
 225  * @brief Get the #PeerContext associated with a peer
 226  *
 227  * @param peer the peer id
 228  *
 229  * @return the #PeerContext
 230  */
 231 static struct PeerContext *
 232 get_peer_ctx (const struct GNUNET_PeerIdentity *peer)
 233 {
 234   struct PeerContext *ctx;
 235   int ret;
 236
 237   ret = GNUNET_CONTAINER_multipeermap_contains (peer_map, peer);
 238   GNUNET_assert (GNUNET_YES == ret);
 239   ctx = GNUNET_CONTAINER_multipeermap_get (peer_map, peer);
 240   GNUNET_assert (NULL != ctx);
 241   return ctx;
 242 }
 243
 244 /**
 245  * @brief Create a new #PeerContext and insert it into the peer map
 246  *
 247  * @param peer the peer to create the #PeerContext for
 248  *
 249  * @return the #PeerContext
 250  */
 251 static struct PeerContext *
 252 create_peer_ctx (const struct GNUNET_PeerIdentity *peer)
 253 {
 254   struct PeerContext *ctx;
 255   int ret;
 256
 257   GNUNET_assert (GNUNET_NO == Peers_check_peer_known (peer));
 258
 259   ctx = GNUNET_new (struct PeerContext);
 260   ctx->peer_id = *peer;
 261   ctx->send_channel_flags = GNUNET_new (uint32_t);
 262   ctx->recv_channel_flags = GNUNET_new (uint32_t);
 263   ret = GNUNET_CONTAINER_multipeermap_put (peer_map, peer, ctx,
 264       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
 265   GNUNET_assert (GNUNET_OK == ret);
 266   return ctx;
 267 }
 268
 269 /**
 270  * @brief Create or get a #PeerContext
 271  *
 272  * @param peer the peer to get the associated context to
 273  *
 274  * @return the context
 275  */
 276 static struct PeerContext *
 277 create_or_get_peer_ctx (const struct GNUNET_PeerIdentity *peer)
 278 {
 279   if (GNUNET_NO == Peers_check_peer_known (peer))
 280   {
 281     return create_peer_ctx (peer);
 282   }
 283   return get_peer_ctx (peer);
 284 }
 285
 286 /**
 287  * @brief Set the peer flag to living and
 288  *        call the pending operations on this peer.
 289  *
 290  * Also sets the #Peers_VALID flag
 291  *
 292  * @param peer_ctx the #PeerContext of the peer to set live
 293  */
 294 static void
 295 set_peer_live (struct PeerContext *peer_ctx)
 296 {
 297   struct GNUNET_PeerIdentity *peer;
 298   unsigned int i;
 299
 300   /* Cancle cadet transmit_handle if still scheduled */
 301   if (NULL != peer_ctx->transmit_handle)
 302   {
 303     GNUNET_CADET_notify_transmit_ready_cancel (peer_ctx->transmit_handle);
 304     peer_ctx->transmit_handle = NULL;
 305   }
 306
 307   peer = &peer_ctx->peer_id;
 308   set_peer_flag (peer_ctx, Peers_VALID);
 309   set_peer_flag (peer_ctx, Peers_ONLINE);
 310   LOG (GNUNET_ERROR_TYPE_DEBUG,
 311       "Peer %s is live and valid\n",
 312       GNUNET_i2s (peer));
 313
 314   /* Call pending operations */
 315   for (i = 0; i < peer_ctx->num_pending_ops; i++)
 316   {
 317     peer_ctx->pending_ops[i].op (peer_ctx->pending_ops[i].op_cls, peer);
 318   }
 319   GNUNET_array_grow (peer_ctx->pending_ops, peer_ctx->num_pending_ops, 0);
 320 }
 321
 322 /**
 323  * @brief Get the channel of a peer. If not existing, create.
 324  *
 325  * @param peer the peer id
 326  * @return the #GNUNET_CADET_Channel used to send data to @a peer
 327  */
 328 struct GNUNET_CADET_Channel *
 329 get_channel (const struct GNUNET_PeerIdentity *peer)
 330 {
 331   struct PeerContext *peer_ctx;
 332
 333   peer_ctx = get_peer_ctx (peer);
 334   if (NULL == peer_ctx->send_channel)
 335   {
 336     LOG (GNUNET_ERROR_TYPE_DEBUG,
 337          "Trying to establish channel to peer %s\n",
 338          GNUNET_i2s (peer));
 339     peer_ctx->send_channel =
 340       GNUNET_CADET_channel_create (cadet_handle,
 341                                    peer_ctx->send_channel_flags, /* context */
 342                                    peer,
 343                                    GNUNET_RPS_CADET_PORT,
 344                                    GNUNET_CADET_OPTION_RELIABLE);
 345   }
 346   return peer_ctx->send_channel;
 347 }
 348
 349 /**
 350  * Get the message queue (#GNUNET_MQ_Handle) of a specific peer.
 351  *
 352  * If we already have a message queue open to this client,
 353  * simply return it, otherways create one.
 354  *
 355  * @param peer the peer to get the mq to
 356  * @return the #GNUNET_MQ_Handle
 357  */
 358 static struct GNUNET_MQ_Handle *
 359 get_mq (const struct GNUNET_PeerIdentity *peer)
 360 {
 361   struct PeerContext *peer_ctx;
 362
 363   peer_ctx = get_peer_ctx (peer);
 364   GNUNET_assert (NULL == peer_ctx->transmit_handle);
 365
 366   if (NULL == peer_ctx->mq)
 367   {
 368     (void) get_channel (peer);
 369     peer_ctx->mq = GNUNET_CADET_mq_create (peer_ctx->send_channel);
 370   }
 371   return peer_ctx->mq;
 372 }
 373
 374 /**
 375  * @brief Callback that is called when a channel was effectively established.
 376  *
 377  * This is an implementation of #GNUNET_CONNECTION_TransmitReadyNotify and
 378  * given to #GNUNET_CADET_notify_transmit_ready_cancel and called when the
 379  * channel was successfully established.
 380  *
 381  * This function type was originally ment to be called to provide the data to
 382  * be sent. This is called when the connection is ready to queue more data.
 383  * However we use it to get notified about the successful establishement of a
 384  * cadet channel.
 385  *
 386  * @a buf will be NULL and @a size zero if the
 387  * connection was closed for writing in the meantime.
 388  *
 389  * @param cls closure
 390  * @param size number of bytes available in @a buf
 391  * @param buf where the callee should write the message
 392  * @return number of bytes written to @a buf
 393  */
 394 //TODO
 395 static size_t
 396 cadet_notify_transmit_ready_cb (void *cls, size_t size, void *buf)
 397 {
 398   struct PeerContext *peer_ctx = (struct PeerContext *) cls;
 399   // TODO make sure the context is not deleted or the establishing of the
 400   //      channel is cancelled
 401
 402   peer_ctx->transmit_handle = NULL;
 403   LOG (GNUNET_ERROR_TYPE_DEBUG,
 404        "Set ->transmit_handle = NULL for peer %s\n",
 405        GNUNET_i2s (&peer_ctx->peer_id));
 406
 407   if ( (NULL != buf) &&
 408        (0 != size) )
 409   {
 410     set_peer_live (peer_ctx);
 411   }
 412   else
 413   {
 414     LOG (GNUNET_ERROR_TYPE_WARNING,
 415          "Problems establishing a connection to peer %s in order to check liveliness\n",
 416          GNUNET_i2s (&peer_ctx->peer_id));
 417     // TODO reschedule? cleanup?
 418   }
 419   return 0;
 420 }
 421
 422 /**
 423  * Issue a check whether peer is live
 424  *
 425  * @param peer_ctx the context of the peer
 426  */
 427 static void
 428 check_peer_live (struct PeerContext *peer_ctx)
 429 {
 430   LOG (GNUNET_ERROR_TYPE_DEBUG,
 431        "Get informed about peer %s getting live\n",
 432        GNUNET_i2s (&peer_ctx->peer_id));
 433
 434   if (NULL == peer_ctx->transmit_handle &&
 435       NULL == peer_ctx->send_channel)
 436   {
 437     (void) get_channel (&peer_ctx->peer_id);
 438     peer_ctx->transmit_handle =
 439         GNUNET_CADET_notify_transmit_ready (peer_ctx->send_channel,
 440                                             GNUNET_NO,
 441                                             GNUNET_TIME_UNIT_FOREVER_REL,
 442                                             sizeof (struct GNUNET_MessageHeader),
 443                                             cadet_notify_transmit_ready_cb,
 444                                             peer_ctx);
 445   }
 446   else if (NULL != peer_ctx->transmit_handle)
 447     LOG (GNUNET_ERROR_TYPE_DEBUG,
 448          "Already waiting for notification\n");
 449   else if (NULL != peer_ctx->send_channel)
 450     LOG (GNUNET_ERROR_TYPE_DEBUG,
 451          "Already have established channel to peer\n");
 452 }
 453
 454 /**
 455  * @brief Add an envelope to a message passed to mq to list of pending messages
 456  *
 457  * @param peer peer the message was sent to
 458  * @param ev envelope to the message
 459  * @param type type of the message to be sent
 460  * @return pointer to pending message
 461  */
 462 static struct PendingMessage *
 463 insert_pending_message (const struct GNUNET_PeerIdentity *peer,
 464                         struct GNUNET_MQ_Envelope *ev,
 465                         const char *type)
 466 {
 467   struct PendingMessage *pending_msg;
 468   struct PeerContext *peer_ctx;
 469
 470   peer_ctx = get_peer_ctx (peer);
 471   pending_msg = GNUNET_new (struct PendingMessage);
 472   pending_msg->ev = ev;
 473   pending_msg->peer_ctx = peer_ctx;
 474   pending_msg->type = type;
 475   GNUNET_CONTAINER_DLL_insert (peer_ctx->pending_messages_head,
 476                                peer_ctx->pending_messages_tail,
 477                                pending_msg);
 478   return pending_msg;
 479 }
 480
 481 /**
 482  * @brief Remove a pending message from the respective DLL
 483  *
 484  * @param pending_msg the pending message to remove
 485  */
 486 static void
 487 remove_pending_message (struct PendingMessage *pending_msg)
 488 {
 489   struct PeerContext *peer_ctx;
 490
 491   peer_ctx = pending_msg->peer_ctx;
 492   GNUNET_CONTAINER_DLL_remove (peer_ctx->pending_messages_head,
 493                                peer_ctx->pending_messages_tail,
 494                                pending_msg);
 495   /* FIXME We are not able to cancel messages as #GNUNET_CADET_mq_create () does
 496    * not set a #GNUNET_MQ_CancelImpl */
 497   /* GNUNET_MQ_send_cancel (peer_ctx->pending_messages_head->ev); */
 498   GNUNET_free (pending_msg);
 499 }
 500
 501 /**
 502  * @brief Check whether function of type #PeerOp was already scheduled
 503  *
 504  * The array with pending operations will probably never grow really big, so
 505  * iterating over it should be ok.
 506  *
 507  * @param peer the peer to check
 508  * @param peer_op the operation (#PeerOp) on the peer
 509  *
 510  * @return #GNUNET_YES if this operation is scheduled on that peer
 511  *         #GNUNET_NO  otherwise
 512  */
 513 static int
 514 check_operation_scheduled (const struct GNUNET_PeerIdentity *peer,
 515                            const PeerOp peer_op)
 516 {
 517   const struct PeerContext *peer_ctx;
 518   unsigned int i;
 519
 520   peer_ctx = get_peer_ctx (peer);
 521   for (i = 0; i < peer_ctx->num_pending_ops; i++)
 522     if (peer_op == peer_ctx->pending_ops[i].op)
 523       return GNUNET_YES;
 524   return GNUNET_NO;
 525 }
 526
 527 /**
 528  * Iterator over hash map entries. Deletes all contexts of peers.
 529  *
 530  * @param cls closure
 531  * @param key current public key
 532  * @param value value in the hash map
 533  * @return #GNUNET_YES if we should continue to iterate,
 534  *         #GNUNET_NO if not.
 535  */
 536 static int
 537 peermap_clear_iterator (void *cls,
 538                         const struct GNUNET_PeerIdentity *key,
 539                         void *value)
 540 {
 541   Peers_remove_peer (key);
 542   return GNUNET_YES;
 543 }
 544
 545 /**
 546  * @brief This is called once a message is sent.
 547  *
 548  * Removes the pending message
 549  *
 550  * @param cls type of the message that was sent
 551  */
 552 static void
 553 mq_notify_sent_cb (void *cls)
 554 {
 555   struct PendingMessage *pending_msg = (struct PendingMessage *) cls;
 556   LOG (GNUNET_ERROR_TYPE_DEBUG,
 557       "%s was sent.\n",
 558       pending_msg->type);
 559   remove_pending_message (pending_msg);
 560 }
 561
 562
 563 /**
 564  * @brief Initialise storage of peers
 565  *
 566  * @param cadet_h cadet handle
 567  * @param own_id own peer identity
 568  */
 569 void
 570 Peers_initialise (struct GNUNET_CADET_Handle *cadet_h,
 571                   const struct GNUNET_PeerIdentity *own_id)
 572 {
 573   cadet_handle = cadet_h;
 574   own_identity = own_id;
 575   peer_map = GNUNET_CONTAINER_multipeermap_create (4, GNUNET_NO);
 576 }
 577
 578 /**
 579  * @brief Delete storage of peers that was created with #Peers_initialise ()
 580  */
 581 void
 582 Peers_terminate ()
 583 {
 584   if (GNUNET_SYSERR ==
 585       GNUNET_CONTAINER_multipeermap_iterate (peer_map,
 586                                              peermap_clear_iterator,
 587                                              NULL))
 588   {
 589     LOG (GNUNET_ERROR_TYPE_WARNING,
 590         "Iteration destroying peers was aborted.\n");
 591   }
 592   GNUNET_CONTAINER_multipeermap_destroy (peer_map);
 593 }
 594
 595 /**
 596  * @brief Add peer to known peers.
 597  *
 598  * This function is called on new peer_ids from 'external' sources
 599  * (client seed, cadet get_peers(), ...)
 600  *
 601  * @param peer the new #GNUNET_PeerIdentity
 602  *
 603  * @return #GNUNET_YES if peer was inserted
 604  *         #GNUNET_NO  if peer was already known
 605  */
 606 int
 607 Peers_insert_peer (const struct GNUNET_PeerIdentity *peer)
 608 {
 609   if ( (GNUNET_YES == Peers_check_peer_known (peer)) ||
 610        (0 == GNUNET_CRYPTO_cmp_peer_identity (peer, own_identity)) )
 611   {
 612     return GNUNET_NO; /* We already know this peer - nothing to do */
 613   }
 614   (void) create_peer_ctx (peer);
 615   return GNUNET_YES;
 616 }
 617
 618
 619 /**
 620  * @brief Add peer to known peers and check for liveliness.
 621  *
 622  * This function is called on new peer_ids from 'external' sources
 623  * (client seed, cadet get_peers(), ...)
 624  *
 625  * @param peer the new #GNUNET_PeerIdentity
 626  *
 627  * @return #GNUNET_YES if peer was inserted
 628  *         #GNUNET_NO  if peer was already known
 629  */
 630 int
 631 Peers_insert_peer_check_liveliness (const struct GNUNET_PeerIdentity *peer)
 632 {
 633   struct PeerContext *peer_ctx;
 634   int ret;
 635
 636   ret = Peers_insert_peer (peer);
 637   if (0 == GNUNET_CRYPTO_cmp_peer_identity (peer, own_identity))
 638   {
 639     return ret;
 640   }
 641   peer_ctx = get_peer_ctx (peer);
 642   if (GNUNET_NO == check_peer_flag_set (peer_ctx, Peers_VALID))
 643   {
 644     check_peer_live (peer_ctx);
 645   }
 646   return ret;
 647 }
 648
 649 /**
 650  * @brief Remove unecessary data
 651  *
 652  * If the other peer is not intending to send messages, we have messages pending
 653  * to be sent to this peer and we are not waiting for a reply, remove the
 654  * information about it (its #PeerContext).
 655  *
 656  * @param peer the peer to clean
 657  * @return #GNUNET_YES if peer was removed
 658  *         #GNUNET_NO  otherwise
 659  */
 660 int
 661 Peers_clean_peer (const struct GNUNET_PeerIdentity *peer)
 662 {
 663   struct PeerContext *peer_ctx;
 664
 665   // TODO actually remove unnecessary data
 666
 667   if (GNUNET_NO == GNUNET_CONTAINER_multipeermap_contains (peer_map, peer))
 668   {
 669     return GNUNET_NO;
 670   }
 671
 672   peer_ctx = get_peer_ctx (peer);
 673   if ( (NULL != peer_ctx->recv_channel) ||
 674        (NULL != peer_ctx->pending_messages_head) ||
 675        (GNUNET_NO == check_peer_flag_set (peer_ctx, Peers_PULL_REPLY_PENDING)) )
 676   {
 677     return GNUNET_NO;
 678   }
 679   Peers_remove_peer (peer);
 680   return GNUNET_YES;
 681 }
 682
 683 /**
 684  * @brief Remove peer
 685  *
 686  * @param peer the peer to clean
 687  * @return #GNUNET_YES if peer was removed
 688  *         #GNUNET_NO  otherwise
 689  */
 690 int
 691 Peers_remove_peer (const struct GNUNET_PeerIdentity *peer)
 692 {
 693   struct PeerContext *peer_ctx;
 694
 695   if (GNUNET_NO == GNUNET_CONTAINER_multipeermap_contains (peer_map, peer))
 696   {
 697     return GNUNET_NO;
 698   }
 699
 700   peer_ctx = get_peer_ctx (peer);
 701   set_peer_flag (peer_ctx, Peers_TO_DESTROY);
 702   LOG (GNUNET_ERROR_TYPE_DEBUG,
 703        "Going to remove peer %s\n",
 704        GNUNET_i2s (&peer_ctx->peer_id));
 705
 706   GNUNET_array_grow (peer_ctx->pending_ops, peer_ctx->num_pending_ops, 0);
 707   // TODO delete struct GNUNET_TRANSPORT_TransmitHandle *transmit_handle
 708   /* Cancle messages that have not been sent yet */
 709   while (NULL != peer_ctx->pending_messages_head)
 710   {
 711     LOG (GNUNET_ERROR_TYPE_DEBUG,
 712         "Removing unsent %s\n",
 713         peer_ctx->pending_messages_head->type);
 714     remove_pending_message (peer_ctx->pending_messages_head);
 715   }
 716   /* If we are still waiting for notification whether this peer is live
 717    * cancel the according task */
 718   if (NULL != peer_ctx->transmit_handle)
 719   {
 720     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 721          "Trying to cancle transmit_handle for peer %s\n",
 722          GNUNET_i2s (&peer_ctx->peer_id));
 723     GNUNET_CADET_notify_transmit_ready_cancel (peer_ctx->transmit_handle);
 724     peer_ctx->transmit_handle = NULL;
 725   }
 726   if (NULL != peer_ctx->send_channel)
 727   {
 728     GNUNET_CADET_channel_destroy (peer_ctx->send_channel);
 729     peer_ctx->send_channel = NULL;
 730   }
 731   if (NULL != peer_ctx->recv_channel)
 732   {
 733     GNUNET_CADET_channel_destroy (peer_ctx->recv_channel);
 734     peer_ctx->recv_channel = NULL;
 735   }
 736   if (NULL != peer_ctx->mq)
 737   {
 738     GNUNET_MQ_destroy (peer_ctx->mq);
 739     peer_ctx->mq = NULL;
 740   }
 741
 742   GNUNET_free (peer_ctx->send_channel_flags);
 743   GNUNET_free (peer_ctx->recv_channel_flags);
 744
 745   if (GNUNET_YES != GNUNET_CONTAINER_multipeermap_remove_all (peer_map, &peer_ctx->peer_id))
 746   {
 747     LOG (GNUNET_ERROR_TYPE_WARNING, "removing peer from peer_map failed\n");
 748   }
 749   GNUNET_free (peer_ctx);
 750   return GNUNET_YES;
 751 }
 752
 753 /**
 754  * @brief set flags on a given peer.
 755  *
 756  * @param peer the peer to set flags on
 757  * @param flags the flags
 758  */
 759 void
 760 Peers_set_peer_flag (const struct GNUNET_PeerIdentity *peer, enum Peers_PeerFlags flags)
 761 {
 762   struct PeerContext *peer_ctx;
 763
 764   peer_ctx = get_peer_ctx (peer);
 765   set_peer_flag (peer_ctx, flags);
 766 }
 767
 768 /**
 769  * @brief unset flags on a given peer.
 770  *
 771  * @param peer the peer to unset flags on
 772  * @param flags the flags
 773  */
 774 void
 775 Peers_unset_peer_flag (const struct GNUNET_PeerIdentity *peer, enum Peers_PeerFlags flags)
 776 {
 777   struct PeerContext *peer_ctx;
 778
 779   peer_ctx = get_peer_ctx (peer);
 780   unset_peer_flag (peer_ctx, flags);
 781 }
 782
 783 /**
 784  * @brief Check whether flags on a peer are set.
 785  *
 786  * @param peer the peer to check the flag of
 787  * @param flags the flags to check
 788  *
 789  * @return #GNUNET_YES if all given flags are set
 790  *         #GNUNET_NO  otherwise
 791  */
 792 int
 793 Peers_check_peer_flag (const struct GNUNET_PeerIdentity *peer, enum Peers_PeerFlags flags)
 794 {
 795   struct PeerContext *peer_ctx;
 796
 797   peer_ctx = get_peer_ctx (peer);
 798   return check_peer_flag_set (peer_ctx, flags);
 799 }
 800
 801
 802 /**
 803  * @brief set flags on a given channel.
 804  *
 805  * @param channel the channel to set flags on
 806  * @param flags the flags
 807  */
 808 void
 809 Peers_set_channel_flag (uint32_t *channel_flags, enum Peers_ChannelFlags flags)
 810 {
 811   set_channel_flag (channel_flags, flags);
 812 }
 813
 814 /**
 815  * @brief unset flags on a given channel.
 816  *
 817  * @param channel the channel to unset flags on
 818  * @param flags the flags
 819  */
 820 void
 821 Peers_unset_channel_flag (uint32_t *channel_flags, enum Peers_ChannelFlags flags)
 822 {
 823   unset_channel_flag (channel_flags, flags);
 824 }
 825
 826 /**
 827  * @brief Check whether flags on a channel are set.
 828  *
 829  * @param channel the channel to check the flag of
 830  * @param flags the flags to check
 831  *
 832  * @return #GNUNET_YES if all given flags are set
 833  *         #GNUNET_NO  otherwise
 834  */
 835 int
 836 Peers_check_channel_flag (uint32_t *channel_flags, enum Peers_ChannelFlags flags)
 837 {
 838   return check_channel_flag_set (channel_flags, flags);
 839 }
 840
 841 /**
 842  * @brief Check whether we have information about the given peer.
 843  *
 844  * @param peer peer in question
 845  *
 846  * @return #GNUNET_YES if peer is known
 847  *         #GNUNET_NO  if peer is not knwon
 848  */
 849 int
 850 Peers_check_peer_known (const struct GNUNET_PeerIdentity *peer)
 851 {
 852   return GNUNET_CONTAINER_multipeermap_contains (peer_map, peer);
 853 }
 854
 855 /**
 856  * @brief Indicate that we want to send to the other peer
 857  *
 858  * This establishes a sending channel
 859  *
 860  * @param peer the peer to establish channel to
 861  */
 862 void
 863 Peers_indicate_sending_intention (const struct GNUNET_PeerIdentity *peer)
 864 {
 865   GNUNET_assert (GNUNET_YES == Peers_check_peer_known (peer));
 866   (void) get_channel (peer);
 867 }
 868
 869 /**
 870  * @brief Check whether other peer has the intention to send/opened channel
 871  *        towars us
 872  *
 873  * @param peer the peer in question
 874  *
 875  * @return #GNUNET_YES if peer has the intention to send
 876  *         #GNUNET_NO  otherwise
 877  */
 878 int
 879 Peers_check_peer_send_intention (const struct GNUNET_PeerIdentity *peer)
 880 {
 881   const struct PeerContext *peer_ctx;
 882
 883   peer_ctx = get_peer_ctx (peer);
 884   if (NULL != peer_ctx->recv_channel)
 885   {
 886     return GNUNET_YES;
 887   }
 888   return GNUNET_NO;
 889 }
 890
 891 /**
 892  * Handle the channel a peer opens to us.
 893  *
 894  * @param cls The closure
 895  * @param channel The channel the peer wants to establish
 896  * @param initiator The peer's peer ID
 897  * @param port The port the channel is being established over
 898  * @param options Further options
 899  *
 900  * @return initial channel context for the channel
 901  *         (can be NULL -- that's not an error)
 902  */
 903 void *
 904 Peers_handle_inbound_channel (void *cls,
 905                               struct GNUNET_CADET_Channel *channel,
 906                               const struct GNUNET_PeerIdentity *initiator,
 907                               uint32_t port,
 908                               enum GNUNET_CADET_ChannelOption options)
 909 {
 910   struct PeerContext *peer_ctx;
 911
 912   LOG (GNUNET_ERROR_TYPE_DEBUG,
 913       "New channel was established to us (Peer %s).\n",
 914       GNUNET_i2s (initiator));
 915   GNUNET_assert (NULL != channel); /* according to cadet API */
 916   /* Make sure we 'know' about this peer */
 917   peer_ctx = create_or_get_peer_ctx (initiator);
 918   set_peer_live (peer_ctx);
 919   /* We only accept one incoming channel per peer */
 920   if (GNUNET_YES == Peers_check_peer_send_intention (initiator))
 921   {
 922     set_channel_flag (peer_ctx->recv_channel_flags,
 923                       Peers_CHANNEL_ESTABLISHED_TWICE);
 924     GNUNET_CADET_channel_destroy (channel);
 925     /* return the channel context */
 926     return peer_ctx->recv_channel_flags;
 927   }
 928   peer_ctx->recv_channel = channel;
 929   return peer_ctx->recv_channel_flags;
 930 }
 931
 932 /**
 933  * @brief Check whether a sending channel towards the given peer exists
 934  *
 935  * @param peer the peer to check for
 936  *
 937  * @return #GNUNET_YES if a sending channel towards that peer exists
 938  *         #GNUNET_NO  otherwise
 939  */
 940 int
 941 Peers_check_sending_channel_exists (const struct GNUNET_PeerIdentity *peer)
 942 {
 943   struct PeerContext *peer_ctx;
 944
 945   if (GNUNET_NO == Peers_check_peer_known (peer))
 946   { /* If no such peer exists, there is no channel */
 947     return GNUNET_NO;
 948   }
 949   peer_ctx = get_peer_ctx (peer);
 950   if (NULL == peer_ctx->send_channel)
 951   {
 952     return GNUNET_NO;
 953   }
 954   return GNUNET_YES;
 955 }
 956
 957 /**
 958  * @brief check whether the given channel is the sending channel of the given
 959  *        peer
 960  *
 961  * @param peer the peer in question
 962  * @param channel the channel to check for
 963  * @param role either #Peers_CHANNEL_ROLE_SENDING, or
 964  *                    #Peers_CHANNEL_ROLE_RECEIVING
 965  *
 966  * @return #GNUNET_YES if the given chennel is the sending channel of the peer
 967  *         #GNUNET_NO  otherwise
 968  */
 969 int
 970 Peers_check_channel_role (const struct GNUNET_PeerIdentity *peer,
 971                           const struct GNUNET_CADET_Channel *channel,
 972                           enum Peers_ChannelRole role)
 973 {
 974   const struct PeerContext *peer_ctx;
 975
 976   if (GNUNET_NO == Peers_check_peer_known (peer))
 977   {
 978     return GNUNET_NO;
 979   }
 980   peer_ctx = get_peer_ctx (peer);
 981   if ( (Peers_CHANNEL_ROLE_SENDING == role) &&
 982        (channel == peer_ctx->send_channel) )
 983   {
 984     return GNUNET_YES;
 985   }
 986   if ( (Peers_CHANNEL_ROLE_RECEIVING == role) &&
 987        (channel == peer_ctx->recv_channel) )
 988   {
 989     return GNUNET_YES;
 990   }
 991   return GNUNET_NO;
 992 }
 993
 994 /**
 995  * @brief Destroy the send channel of a peer e.g. stop indicating a sending
 996  *        intention to another peer
 997  *
 998  * If there is also no channel to receive messages from that peer, remove it
 999  * from the peermap.
1000  * TODO really?
1001  *
1002  * @peer the peer identity of the peer whose sending channel to destroy
1003  * @return #GNUNET_YES if channel was destroyed
1004  *         #GNUNET_NO  otherwise
1005  */
1006 int
1007 Peers_destroy_sending_channel (const struct GNUNET_PeerIdentity *peer)
1008 {
1009   struct PeerContext *peer_ctx;
1010
1011   if (GNUNET_NO == Peers_check_peer_known (peer))
1012   {
1013     return GNUNET_NO;
1014   }
1015   peer_ctx = get_peer_ctx (peer);
1016   if (NULL != peer_ctx->send_channel)
1017   {
1018     set_channel_flag (peer_ctx->send_channel_flags, Peers_CHANNEL_CLEAN);
1019     GNUNET_CADET_channel_destroy (peer_ctx->send_channel);
1020     peer_ctx->send_channel = NULL;
1021     return GNUNET_YES;
1022   }
1023   return GNUNET_NO;
1024 }
1025
1026 /**
1027  * This is called when a channel is destroyed.
1028  *
1029  * Removes peer completely from our knowledge if the send_channel was destroyed
1030  * Otherwise simply delete the recv_channel
1031  *
1032  * @param cls The closure
1033  * @param channel The channel being closed
1034  * @param channel_ctx The context associated with this channel
1035  */
1036 void
1037 Peers_cleanup_destroyed_channel (void *cls,
1038                                  const struct GNUNET_CADET_Channel *channel,
1039                                  void *channel_ctx)
1040 {
1041   struct GNUNET_PeerIdentity *peer;
1042   struct PeerContext *peer_ctx;
1043
1044   peer = (struct GNUNET_PeerIdentity *) GNUNET_CADET_channel_get_info (
1045       (struct GNUNET_CADET_Channel *) channel, GNUNET_CADET_OPTION_PEER);
1046        // FIXME wait for cadet to change this function
1047
1048   if (GNUNET_NO == Peers_check_peer_known (peer))
1049   {/* We don't want to implicitly create a context that we're about to kill */
1050   LOG (GNUNET_ERROR_TYPE_DEBUG,
1051        "channel (%s) without associated context was destroyed\n",
1052        GNUNET_i2s (peer));
1053     return;
1054   }
1055
1056   peer_ctx = get_peer_ctx (peer);
1057   GNUNET_assert (NULL != peer_ctx); /* It could have been removed by shutdown_task */
1058
1059   /* If our peer issued the destruction of the channel, the #Peers_TO_DESTROY
1060    * flag will be set. In this case simply make sure that the channels are
1061    * cleaned. */
1062   if (Peers_check_peer_flag (peer, Peers_TO_DESTROY))
1063   {/* We initiatad the destruction of this particular peer */
1064     if (channel == peer_ctx->send_channel)
1065       peer_ctx->send_channel = NULL;
1066     else if (channel == peer_ctx->recv_channel)
1067       peer_ctx->recv_channel = NULL;
1068
1069     return;
1070   }
1071
1072   else
1073   { /* We did not initiate the destruction of this peer */
1074     if (channel == peer_ctx->send_channel)
1075     { /* Something (but us) killd the channel - clean up peer */
1076       LOG (GNUNET_ERROR_TYPE_DEBUG,
1077           "send channel (%s) was destroyed - cleaning up\n",
1078           GNUNET_i2s (peer));
1079       peer_ctx->send_channel = NULL;
1080     }
1081     else if (channel == peer_ctx->recv_channel)
1082     { /* Other peer doesn't want to send us messages anymore */
1083       LOG (GNUNET_ERROR_TYPE_DEBUG,
1084            "Peer %s destroyed recv channel - cleaning up channel\n",
1085            GNUNET_i2s (peer));
1086       peer_ctx->recv_channel = NULL;
1087     }
1088     else
1089     {
1090       LOG (GNUNET_ERROR_TYPE_WARNING,
1091            "unknown channel (%s) was destroyed\n",
1092            GNUNET_i2s (peer));
1093     }
1094   }
1095 }
1096
1097 /**
1098  * @brief Issue a check whether peer is live
1099  *
1100  * This tries to establish a channel to the given peer. Once the channel is
1101  * established successfully, we know the peer is live.
1102  *
1103  * @param peer the peer to check liveliness
1104  */
1105 void
1106 Peers_issue_peer_liveliness_check (const struct GNUNET_PeerIdentity *peer)
1107 {
1108   struct PeerContext *peer_ctx;
1109
1110   if (0 == GNUNET_CRYPTO_cmp_peer_identity (peer, own_identity))
1111   {
1112     return; /* We know that we are online */
1113   }
1114
1115   peer_ctx = create_or_get_peer_ctx (peer);
1116   // TODO if LIVE/ONLINE
1117   check_peer_live (peer_ctx);
1118 }
1119
1120 /**
1121  * @brief Send a message to another peer.
1122  *
1123  * Keeps track about pending messages so they can be properly removed when the
1124  * peer is destroyed.
1125  *
1126  * @param peer receeiver of the message
1127  * @param ev envelope of the message
1128  * @param type type of the message
1129  */
1130 void
1131 Peers_send_message (const struct GNUNET_PeerIdentity *peer,
1132                     struct GNUNET_MQ_Envelope *ev,
1133                     const char *type)
1134 {
1135   struct PendingMessage *pending_msg;
1136   struct GNUNET_MQ_Handle *mq;
1137
1138   pending_msg = insert_pending_message (peer, ev, "PULL REPLY");
1139   mq = get_mq (peer);
1140   GNUNET_MQ_notify_sent (ev,
1141                          mq_notify_sent_cb,
1142                          pending_msg);
1143   GNUNET_MQ_send (mq, ev);
1144 }
1145
1146 /**
1147  * @brief Schedule a operation on given peer
1148  *
1149  * Avoids scheduling an operation twice.
1150  *
1151  * @param peer the peer we want to schedule the operation for once it gets live
1152  *
1153  * @return #GNUNET_YES if the operation was scheduled
1154  *         #GNUNET_NO  otherwise
1155  */
1156 int
1157 Peers_schedule_operation (const struct GNUNET_PeerIdentity *peer,
1158                           const PeerOp peer_op)
1159 {
1160   struct PeerPendingOp pending_op;
1161   struct PeerContext *peer_ctx;
1162
1163   if (0 == GNUNET_CRYPTO_cmp_peer_identity (peer, own_identity))
1164   {
1165     return GNUNET_NO;
1166   }
1167   GNUNET_assert (GNUNET_YES == Peers_check_peer_known (peer));
1168
1169   //TODO if LIVE/ONLINE execute immediately
1170
1171   if (GNUNET_NO == check_operation_scheduled (peer, peer_op))
1172   {
1173     peer_ctx = get_peer_ctx (peer);
1174     pending_op.op = peer_op;
1175     pending_op.op_cls = NULL;
1176     GNUNET_array_append (peer_ctx->pending_ops,
1177                          peer_ctx->num_pending_ops,
1178                          pending_op);
1179     return GNUNET_YES;
1180   }
1181   return GNUNET_NO;
1182 }
1183
1184 /* end of gnunet-service-rps_peers.c */