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