src/fs/gnunet-service-fs_push.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2011 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 fs/gnunet-service-fs_push.c
  23  * @brief API to push content from our datastore to other peers
  24  *            ('anonymous'-content P2P migration)
  25  * @author Christian Grothoff
  26  */
  27 #include "platform.h"
  28 #include "gnunet-service-fs.h"
  29 #include "gnunet-service-fs_cp.h"
  30 #include "gnunet-service-fs_indexing.h"
  31 #include "gnunet-service-fs_push.h"
  32
  33
  34 #define DEBUG_FS_MIGRATION GNUNET_NO
  35
  36 /**
  37  * How long must content remain valid for us to consider it for migration?
  38  * If content will expire too soon, there is clearly no point in pushing
  39  * it to other peers.  This value gives the threshold for migration.  Note
  40  * that if this value is increased, the migration testcase may need to be
  41  * adjusted as well (especially the CONTENT_LIFETIME in fs_test_lib.c).
  42  */
  43 #define MIN_MIGRATION_CONTENT_LIFETIME GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 30)
  44
  45
  46 /**
  47  * Block that is ready for migration to other peers.  Actual data is at the end of the block.
  48  */
  49 struct MigrationReadyBlock
  50 {
  51
  52   /**
  53    * This is a doubly-linked list.
  54    */
  55   struct MigrationReadyBlock *next;
  56
  57   /**
  58    * This is a doubly-linked list.
  59    */
  60   struct MigrationReadyBlock *prev;
  61
  62   /**
  63    * Query for the block.
  64    */
  65   GNUNET_HashCode query;
  66
  67   /**
  68    * When does this block expire?
  69    */
  70   struct GNUNET_TIME_Absolute expiration;
  71
  72   /**
  73    * Peers we already forwarded this
  74    * block to.  Zero for empty entries.
  75    */
  76   GNUNET_PEER_Id target_list[MIGRATION_LIST_SIZE];
  77
  78   /**
  79    * Size of the block.
  80    */
  81   size_t size;
  82
  83   /**
  84    *  Number of targets already used.
  85    */
  86   unsigned int used_targets;
  87
  88   /**
  89    * Type of the block.
  90    */
  91   enum GNUNET_BLOCK_Type type;
  92 };
  93
  94
  95 /**
  96  * Information about a peer waiting for
  97  * migratable data.
  98  */
  99 struct MigrationReadyPeer
 100 {
 101   /**
 102    * This is a doubly-linked list.
 103    */
 104   struct MigrationReadyPeer *next;
 105
 106   /**
 107    * This is a doubly-linked list.
 108    */
 109   struct MigrationReadyPeer *prev;
 110
 111   /**
 112    * Handle to peer.
 113    */
 114   struct GSF_ConnectedPeer *peer;
 115
 116   /**
 117    * Handle for current transmission request,
 118    * or NULL for none.
 119    */
 120   struct GSF_PeerTransmitHandle *th;
 121
 122   /**
 123    * Message we are trying to push right now (or NULL)
 124    */
 125   struct PutMessage *msg;
 126 };
 127
 128
 129 /**
 130  * Head of linked list of blocks that can be migrated.
 131  */
 132 static struct MigrationReadyBlock *mig_head;
 133
 134 /**
 135  * Tail of linked list of blocks that can be migrated.
 136  */
 137 static struct MigrationReadyBlock *mig_tail;
 138
 139 /**
 140  * Head of linked list of peers.
 141  */
 142 static struct MigrationReadyPeer *peer_head;
 143
 144 /**
 145  * Tail of linked list of peers.
 146  */
 147 static struct MigrationReadyPeer *peer_tail;
 148
 149 /**
 150  * Request to datastore for migration (or NULL).
 151  */
 152 static struct GNUNET_DATASTORE_QueueEntry *mig_qe;
 153
 154 /**
 155  * ID of task that collects blocks for migration.
 156  */
 157 static GNUNET_SCHEDULER_TaskIdentifier mig_task;
 158
 159 /**
 160  * What is the maximum frequency at which we are allowed to
 161  * poll the datastore for migration content?
 162  */
 163 static struct GNUNET_TIME_Relative min_migration_delay;
 164
 165 /**
 166  * Size of the doubly-linked list of migration blocks.
 167  */
 168 static unsigned int mig_size;
 169
 170 /**
 171  * Is this module enabled?
 172  */
 173 static int enabled;
 174
 175
 176 /**
 177  * Delete the given migration block.
 178  *
 179  * @param mb block to delete
 180  */
 181 static void
 182 delete_migration_block (struct MigrationReadyBlock *mb)
 183 {
 184   GNUNET_CONTAINER_DLL_remove (mig_head,
 185                                mig_tail,
 186                                mb);
 187   GNUNET_PEER_decrement_rcs (mb->target_list,
 188                              MIGRATION_LIST_SIZE);
 189   mig_size--;
 190   GNUNET_free (mb);
 191 }
 192
 193
 194 /**
 195  * Find content for migration to this peer.
 196  */
 197 static void
 198 find_content (struct MigrationReadyPeer *mrp);
 199
 200
 201 /**
 202  * Transmit the message currently scheduled for
 203  * transmission.
 204  *
 205  * @param cls the 'struct MigrationReadyPeer'
 206  * @param buf_size number of bytes available in buf
 207  * @param buf where to copy the message, NULL on error (peer disconnect)
 208  * @return number of bytes copied to 'buf', can be 0 (without indicating an error)
 209  */
 210 static size_t
 211 transmit_message (void *cls,
 212                   size_t buf_size,
 213                   void *buf)
 214 {
 215   struct MigrationReadyPeer *peer = cls;
 216   struct PutMessage *msg;
 217   uint16_t msize;
 218
 219   peer->th = NULL;
 220   msg = peer->msg;
 221   peer->msg = NULL;
 222   if (buf == NULL)
 223     {
 224 #if DEBUG_FS_MIGRATION
 225       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 226                   "Failed to migrate content to another peer (disconnect)\n");
 227 #endif
 228       GNUNET_free (msg);
 229       return 0;
 230     }
 231   msize = ntohs (msg->header.size);
 232   GNUNET_assert (msize <= buf_size);
 233   memcpy (buf, msg, msize);
 234   GNUNET_free (msg);
 235 #if DEBUG_FS_MIGRATION
 236   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 237               "Pushing %u bytes to another peer\n",
 238               msize);
 239 #endif
 240   find_content (peer);
 241   return msize;
 242 }
 243
 244
 245 /**
 246  * Send the given block to the given peer.
 247  *
 248  * @param peer target peer
 249  * @param block the block
 250  * @return GNUNET_YES if the block was deleted (!)
 251  */
 252 static int
 253 transmit_content (struct MigrationReadyPeer *peer,
 254                   struct MigrationReadyBlock *block)
 255 {
 256   size_t msize;
 257   struct PutMessage *msg;
 258   unsigned  int i;
 259   struct GSF_PeerPerformanceData *ppd;
 260   int ret;
 261
 262   ppd = GSF_get_peer_performance_data_ (peer->peer);
 263   GNUNET_assert (NULL == peer->th);
 264   msize = sizeof (struct PutMessage) + block->size;
 265   msg = GNUNET_malloc (msize);
 266   msg->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
 267   msg->header.size = htons (msize);
 268   msg->type = htonl (block->type);
 269   msg->expiration = GNUNET_TIME_absolute_hton (block->expiration);
 270   memcpy (&msg[1],
 271           &block[1],
 272           block->size);
 273   peer->msg = msg;
 274   for (i=0;i<MIGRATION_LIST_SIZE;i++)
 275     {
 276       if (block->target_list[i] == 0)
 277         {
 278           block->target_list[i] = ppd->pid;
 279           GNUNET_PEER_change_rc (block->target_list[i], 1);
 280           break;
 281         }
 282     }
 283   if (MIGRATION_LIST_SIZE == i)
 284     {
 285       delete_migration_block (block);
 286       ret = GNUNET_YES;
 287     }
 288   else
 289     {
 290       ret = GNUNET_NO;
 291     }
 292 #if DEBUG_FS_MIGRATION
 293   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 294               "Asking for transmission of %u bytes for migration\n",
 295               msize);
 296 #endif
 297   peer->th = GSF_peer_transmit_ (peer->peer,
 298                                  GNUNET_NO,
 299                                  0 /* priority */,
 300                                  GNUNET_TIME_UNIT_FOREVER_REL,
 301                                  msize,
 302                                  &transmit_message,
 303                                  peer);
 304   return ret;
 305 }
 306
 307
 308 /**
 309  * Count the number of peers this block has
 310  * already been forwarded to.
 311  *
 312  * @param block the block
 313  * @return number of times block was forwarded
 314  */
 315 static unsigned int
 316 count_targets (struct MigrationReadyBlock *block)
 317 {
 318   unsigned int i;
 319
 320   for (i=0;i<MIGRATION_LIST_SIZE;i++)
 321     if (block->target_list[i] == 0)
 322       return i;
 323   return i;
 324 }
 325
 326
 327 /**
 328  * Check if sending this block to this peer would
 329  * be a good idea.
 330  *
 331  * @param peer target peer
 332  * @param block the block
 333  * @return score (>= 0: feasible, negative: infeasible)
 334  */
 335 static long
 336 score_content (struct MigrationReadyPeer *peer,
 337                struct MigrationReadyBlock *block)
 338 {
 339   unsigned int i;
 340   struct GSF_PeerPerformanceData *ppd;
 341   struct GNUNET_PeerIdentity id;
 342   uint32_t dist;
 343
 344   ppd = GSF_get_peer_performance_data_ (peer->peer);
 345   for (i=0;i<MIGRATION_LIST_SIZE;i++)
 346     if (block->target_list[i] == ppd->pid)
 347       return -1;
 348   GNUNET_PEER_resolve (ppd->pid,
 349                        &id);
 350   dist = GNUNET_CRYPTO_hash_distance_u32 (&block->query,
 351                                           &id.hashPubKey);
 352   /* closer distance, higher score: */
 353   return UINT32_MAX - dist;
 354 }
 355
 356
 357 /**
 358  * If the migration task is not currently running, consider
 359  * (re)scheduling it with the appropriate delay.
 360  */
 361 static void
 362 consider_gathering (void);
 363
 364
 365 /**
 366  * Find content for migration to this peer.
 367  *
 368  * @param mrp peer to find content for
 369  */
 370 static void
 371 find_content (struct MigrationReadyPeer *mrp)
 372 {
 373   struct MigrationReadyBlock *pos;
 374   long score;
 375   long best_score;
 376   struct MigrationReadyBlock *best;
 377
 378   GNUNET_assert (NULL == mrp->th);
 379   best = NULL;
 380   best_score = -1;
 381   pos = mig_head;
 382   while (NULL != pos)
 383     {
 384       score = score_content (mrp, pos);
 385       if (score > best_score)
 386         {
 387           best_score = score;
 388           best = pos;
 389         }
 390       pos = pos->next;
 391     }
 392   if (NULL == best)
 393     {
 394       if (mig_size < MAX_MIGRATION_QUEUE)
 395         {
 396 #if DEBUG_FS_MIGRATION
 397           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 398                       "No content found for pushing, waiting for queue to fill\n");
 399 #endif
 400           return; /* will fill up eventually... */
 401         }
 402 #if DEBUG_FS_MIGRATION
 403       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 404                   "No suitable content found, purging content from full queue\n");
 405 #endif
 406       /* failed to find migration target AND
 407          queue is full, purge most-forwarded
 408          block from queue to make room for more */
 409       pos = mig_head;
 410       while (NULL != pos)
 411         {
 412           score = count_targets (pos);
 413           if (score >= best_score)
 414             {
 415               best_score = score;
 416               best = pos;
 417             }
 418           pos = pos->next;
 419         }
 420       GNUNET_assert (NULL != best);
 421       delete_migration_block (best);
 422       consider_gathering ();
 423       return;
 424     }
 425 #if DEBUG_FS_MIGRATION
 426   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 427               "Preparing to push best content to peer\n");
 428 #endif
 429   transmit_content (mrp, best);
 430 }
 431
 432
 433 /**
 434  * Task that is run periodically to obtain blocks for content
 435  * migration
 436  *
 437  * @param cls unused
 438  * @param tc scheduler context (also unused)
 439  */
 440 static void
 441 gather_migration_blocks (void *cls,
 442                          const struct GNUNET_SCHEDULER_TaskContext *tc);
 443
 444
 445 /**
 446  * If the migration task is not currently running, consider
 447  * (re)scheduling it with the appropriate delay.
 448  */
 449 static void
 450 consider_gathering ()
 451 {
 452   struct GNUNET_TIME_Relative delay;
 453
 454   if (GSF_dsh == NULL)
 455     return;
 456   if (mig_qe != NULL)
 457     return;
 458   if (mig_task != GNUNET_SCHEDULER_NO_TASK)
 459     return;
 460   if (mig_size >= MAX_MIGRATION_QUEUE)
 461     return;
 462   delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
 463                                          mig_size);
 464   delay = GNUNET_TIME_relative_divide (delay,
 465                                        MAX_MIGRATION_QUEUE);
 466   delay = GNUNET_TIME_relative_max (delay,
 467                                     min_migration_delay);
 468 #if DEBUG_FS_MIGRATION
 469   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 470               "Scheduling gathering task (queue size: %u)\n",
 471               mig_size);
 472 #endif
 473   mig_task = GNUNET_SCHEDULER_add_delayed (delay,
 474                                            &gather_migration_blocks,
 475                                            NULL);
 476 }
 477
 478
 479 /**
 480  * Process content offered for migration.
 481  *
 482  * @param cls closure
 483  * @param key key for the content
 484  * @param size number of bytes in data
 485  * @param data content stored
 486  * @param type type of the content
 487  * @param priority priority of the content
 488  * @param anonymity anonymity-level for the content
 489  * @param expiration expiration time for the content
 490  * @param uid unique identifier for the datum;
 491  *        maybe 0 if no unique identifier is available
 492  */
 493 static void
 494 process_migration_content (void *cls,
 495                            const GNUNET_HashCode *key,
 496                            size_t size,
 497                            const void *data,
 498                            enum GNUNET_BLOCK_Type type,
 499                            uint32_t priority,
 500                            uint32_t anonymity,
 501                            struct GNUNET_TIME_Absolute
 502                            expiration, uint64_t uid)
 503 {
 504   struct MigrationReadyBlock *mb;
 505   struct MigrationReadyPeer *pos;
 506
 507   mig_qe = NULL;
 508   if (key == NULL)
 509     {
 510 #if DEBUG_FS_MIGRATION
 511       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 512                   "No content found for migration...\n");
 513 #endif
 514       consider_gathering ();
 515       return;
 516     }
 517   if (GNUNET_TIME_absolute_get_remaining (expiration).rel_value <
 518       MIN_MIGRATION_CONTENT_LIFETIME.rel_value)
 519     {
 520       /* content will expire soon, don't bother */
 521       consider_gathering ();
 522       return;
 523     }
 524   if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
 525     {
 526       if (GNUNET_OK !=
 527           GNUNET_FS_handle_on_demand_block (key, size, data,
 528                                             type, priority, anonymity,
 529                                             expiration, uid,
 530                                             &process_migration_content,
 531                                             NULL))
 532         consider_gathering ();
 533       return;
 534     }
 535 #if DEBUG_FS_MIGRATION
 536   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 537               "Retrieved block `%s' of type %u for migration (queue size: %u/%u)\n",
 538               GNUNET_h2s (key),
 539               type,
 540               mig_size + 1,
 541               MAX_MIGRATION_QUEUE);
 542 #endif
 543   mb = GNUNET_malloc (sizeof (struct MigrationReadyBlock) + size);
 544   mb->query = *key;
 545   mb->expiration = expiration;
 546   mb->size = size;
 547   mb->type = type;
 548   memcpy (&mb[1], data, size);
 549   GNUNET_CONTAINER_DLL_insert_after (mig_head,
 550                                      mig_tail,
 551                                      mig_tail,
 552                                      mb);
 553   mig_size++;
 554   pos = peer_head;
 555   while (pos != NULL)
 556     {
 557       if (NULL == pos->th)
 558         {
 559 #if DEBUG_FS_MIGRATION
 560           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 561                       "Preparing to push best content to peer\n");
 562 #endif
 563           if (GNUNET_YES == transmit_content (pos, mb))
 564             break; /* 'mb' was freed! */
 565         }
 566       pos = pos->next;
 567     }
 568   consider_gathering ();
 569 }
 570
 571
 572 /**
 573  * Task that is run periodically to obtain blocks for content
 574  * migration
 575  *
 576  * @param cls unused
 577  * @param tc scheduler context (also unused)
 578  */
 579 static void
 580 gather_migration_blocks (void *cls,
 581                          const struct GNUNET_SCHEDULER_TaskContext *tc)
 582 {
 583   mig_task = GNUNET_SCHEDULER_NO_TASK;
 584   if (mig_size >= MAX_MIGRATION_QUEUE)
 585     return;
 586   if (GSF_dsh != NULL)
 587     {
 588 #if DEBUG_FS_MIGRATION
 589       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 590                   "Asking datastore for content for replication (queue size: %u)\n",
 591                   mig_size);
 592 #endif
 593       mig_qe = GNUNET_DATASTORE_get_for_replication (GSF_dsh,
 594                                                      0, UINT_MAX,
 595                                                      GNUNET_TIME_UNIT_FOREVER_REL,
 596                                                      &process_migration_content, NULL);
 597       if (NULL == mig_qe)
 598         consider_gathering ();
 599     }
 600 }
 601
 602
 603 /**
 604  * A peer connected to us.  Start pushing content
 605  * to this peer.
 606  *
 607  * @param peer handle for the peer that connected
 608  */
 609 void
 610 GSF_push_start_ (struct GSF_ConnectedPeer *peer)
 611 {
 612   struct MigrationReadyPeer *mrp;
 613
 614   if (GNUNET_YES != enabled)
 615     return;
 616   mrp = GNUNET_malloc (sizeof (struct MigrationReadyPeer));
 617   mrp->peer = peer;
 618   find_content (mrp);
 619   GNUNET_CONTAINER_DLL_insert  (peer_head,
 620                                 peer_tail,
 621                                 mrp);
 622 }
 623
 624
 625 /**
 626  * A peer disconnected from us.  Stop pushing content
 627  * to this peer.
 628  *
 629  * @param peer handle for the peer that disconnected
 630  */
 631 void
 632 GSF_push_stop_ (struct GSF_ConnectedPeer *peer)
 633 {
 634   struct MigrationReadyPeer *pos;
 635
 636   pos = peer_head;
 637   while (pos != NULL)
 638     {
 639       if (pos->peer == peer)
 640         {
 641           GNUNET_CONTAINER_DLL_remove (peer_head,
 642                                        peer_tail,
 643                                        pos);
 644           if (NULL != pos->th)
 645             GSF_peer_transmit_cancel_ (pos->th);
 646           GNUNET_free (pos);
 647           return;
 648         }
 649       pos = pos->next;
 650     }
 651 }
 652
 653
 654 /**
 655  * Setup the module.
 656  */
 657 void
 658 GSF_push_init_ ()
 659 {
 660   enabled = GNUNET_CONFIGURATION_get_value_yesno (GSF_cfg,
 661                                                   "FS",
 662                                                   "CONTENT_PUSHING");
 663   if (GNUNET_YES != enabled)
 664     return;
 665
 666   if (GNUNET_OK !=
 667       GNUNET_CONFIGURATION_get_value_time (GSF_cfg,
 668                                            "fs",
 669                                            "MIN_MIGRATION_DELAY",
 670                                            &min_migration_delay))
 671     {
 672       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 673                   _("Invalid value specified for option `%s' in section `%s', content pushing disabled\n"),
 674                   "MIN_MIGRATION_DELAY",
 675                   "fs");
 676       return;
 677     }
 678   consider_gathering ();
 679 }
 680
 681
 682 /**
 683  * Shutdown the module.
 684  */
 685 void
 686 GSF_push_done_ ()
 687 {
 688   if (GNUNET_SCHEDULER_NO_TASK != mig_task)
 689     {
 690       GNUNET_SCHEDULER_cancel (mig_task);
 691       mig_task = GNUNET_SCHEDULER_NO_TASK;
 692     }
 693   if (NULL != mig_qe)
 694     {
 695       GNUNET_DATASTORE_cancel (mig_qe);
 696       mig_qe = NULL;
 697     }
 698   while (NULL != mig_head)
 699     delete_migration_block (mig_head);
 700   GNUNET_assert (0 == mig_size);
 701 }
 702
 703 /* end of gnunet-service-fs_push.c */