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_EXTRA_LOGGING
  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, mig_tail, mb);
 185   GNUNET_PEER_decrement_rcs (mb->target_list, MIGRATION_LIST_SIZE);
 186   mig_size--;
 187   GNUNET_free (mb);
 188 }
 189
 190
 191 /**
 192  * Find content for migration to this peer.
 193  */
 194 static void
 195 find_content (struct MigrationReadyPeer *mrp);
 196
 197
 198 /**
 199  * Transmit the message currently scheduled for
 200  * transmission.
 201  *
 202  * @param cls the 'struct MigrationReadyPeer'
 203  * @param buf_size number of bytes available in buf
 204  * @param buf where to copy the message, NULL on error (peer disconnect)
 205  * @return number of bytes copied to 'buf', can be 0 (without indicating an error)
 206  */
 207 static size_t
 208 transmit_message (void *cls, size_t buf_size, void *buf)
 209 {
 210   struct MigrationReadyPeer *peer = cls;
 211   struct PutMessage *msg;
 212   uint16_t msize;
 213
 214   peer->th = NULL;
 215   msg = peer->msg;
 216   peer->msg = NULL;
 217   if (buf == NULL)
 218   {
 219 #if DEBUG_FS_MIGRATION
 220     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 221                 "Failed to migrate content to another peer (disconnect)\n");
 222 #endif
 223     GNUNET_free (msg);
 224     return 0;
 225   }
 226   msize = ntohs (msg->header.size);
 227   GNUNET_assert (msize <= buf_size);
 228   memcpy (buf, msg, msize);
 229   GNUNET_free (msg);
 230 #if DEBUG_FS_MIGRATION
 231   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Pushing %u bytes to another peer\n",
 232               msize);
 233 #endif
 234   find_content (peer);
 235   return msize;
 236 }
 237
 238
 239 /**
 240  * Send the given block to the given peer.
 241  *
 242  * @param peer target peer
 243  * @param block the block
 244  * @return GNUNET_YES if the block was deleted (!)
 245  */
 246 static int
 247 transmit_content (struct MigrationReadyPeer *peer,
 248                   struct MigrationReadyBlock *block)
 249 {
 250   size_t msize;
 251   struct PutMessage *msg;
 252   unsigned int i;
 253   struct GSF_PeerPerformanceData *ppd;
 254   int ret;
 255
 256   ppd = GSF_get_peer_performance_data_ (peer->peer);
 257   GNUNET_assert (NULL == peer->th);
 258   msize = sizeof (struct PutMessage) + block->size;
 259   msg = GNUNET_malloc (msize);
 260   msg->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
 261   msg->header.size = htons (msize);
 262   msg->type = htonl (block->type);
 263   msg->expiration = GNUNET_TIME_absolute_hton (block->expiration);
 264   memcpy (&msg[1], &block[1], block->size);
 265   peer->msg = msg;
 266   for (i = 0; i < MIGRATION_LIST_SIZE; i++)
 267   {
 268     if (block->target_list[i] == 0)
 269     {
 270       block->target_list[i] = ppd->pid;
 271       GNUNET_PEER_change_rc (block->target_list[i], 1);
 272       break;
 273     }
 274   }
 275   if (MIGRATION_LIST_SIZE == i)
 276   {
 277     delete_migration_block (block);
 278     ret = GNUNET_YES;
 279   }
 280   else
 281   {
 282     ret = GNUNET_NO;
 283   }
 284 #if DEBUG_FS_MIGRATION
 285   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 286               "Asking for transmission of %u bytes for migration\n", msize);
 287 #endif
 288   peer->th = GSF_peer_transmit_ (peer->peer, GNUNET_NO, 0 /* priority */ ,
 289                                  GNUNET_TIME_UNIT_FOREVER_REL, msize,
 290                                  &transmit_message, peer);
 291   return ret;
 292 }
 293
 294
 295 /**
 296  * Count the number of peers this block has
 297  * already been forwarded to.
 298  *
 299  * @param block the block
 300  * @return number of times block was forwarded
 301  */
 302 static unsigned int
 303 count_targets (struct MigrationReadyBlock *block)
 304 {
 305   unsigned int i;
 306
 307   for (i = 0; i < MIGRATION_LIST_SIZE; i++)
 308     if (block->target_list[i] == 0)
 309       return i;
 310   return i;
 311 }
 312
 313
 314 /**
 315  * Check if sending this block to this peer would
 316  * be a good idea.
 317  *
 318  * @param peer target peer
 319  * @param block the block
 320  * @return score (>= 0: feasible, negative: infeasible)
 321  */
 322 static long
 323 score_content (struct MigrationReadyPeer *peer,
 324                struct MigrationReadyBlock *block)
 325 {
 326   unsigned int i;
 327   struct GSF_PeerPerformanceData *ppd;
 328   struct GNUNET_PeerIdentity id;
 329   uint32_t dist;
 330
 331   ppd = GSF_get_peer_performance_data_ (peer->peer);
 332   for (i = 0; i < MIGRATION_LIST_SIZE; i++)
 333     if (block->target_list[i] == ppd->pid)
 334       return -1;
 335   GNUNET_assert (0 != ppd->pid);
 336   GNUNET_PEER_resolve (ppd->pid, &id);
 337   dist = GNUNET_CRYPTO_hash_distance_u32 (&block->query, &id.hashPubKey);
 338   /* closer distance, higher score: */
 339   return UINT32_MAX - dist;
 340 }
 341
 342
 343 /**
 344  * If the migration task is not currently running, consider
 345  * (re)scheduling it with the appropriate delay.
 346  */
 347 static void
 348 consider_gathering (void);
 349
 350
 351 /**
 352  * Find content for migration to this peer.
 353  *
 354  * @param mrp peer to find content for
 355  */
 356 static void
 357 find_content (struct MigrationReadyPeer *mrp)
 358 {
 359   struct MigrationReadyBlock *pos;
 360   long score;
 361   long best_score;
 362   struct MigrationReadyBlock *best;
 363
 364   GNUNET_assert (NULL == mrp->th);
 365   best = NULL;
 366   best_score = -1;
 367   pos = mig_head;
 368   while (NULL != pos)
 369   {
 370     score = score_content (mrp, pos);
 371     if (score > best_score)
 372     {
 373       best_score = score;
 374       best = pos;
 375     }
 376     pos = pos->next;
 377   }
 378   if (NULL == best)
 379   {
 380     if (mig_size < MAX_MIGRATION_QUEUE)
 381     {
 382 #if DEBUG_FS_MIGRATION
 383       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 384                   "No content found for pushing, waiting for queue to fill\n");
 385 #endif
 386       return;                   /* will fill up eventually... */
 387     }
 388 #if DEBUG_FS_MIGRATION
 389     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 390                 "No suitable content found, purging content from full queue\n");
 391 #endif
 392     /* failed to find migration target AND
 393      * queue is full, purge most-forwarded
 394      * block from queue to make room for more */
 395     pos = mig_head;
 396     while (NULL != pos)
 397     {
 398       score = count_targets (pos);
 399       if (score >= best_score)
 400       {
 401         best_score = score;
 402         best = pos;
 403       }
 404       pos = pos->next;
 405     }
 406     GNUNET_assert (NULL != best);
 407     delete_migration_block (best);
 408     consider_gathering ();
 409     return;
 410   }
 411 #if DEBUG_FS_MIGRATION
 412   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 413               "Preparing to push best content to peer %s\n");
 414 #endif
 415   transmit_content (mrp, best);
 416 }
 417
 418
 419 /**
 420  * Task that is run periodically to obtain blocks for content
 421  * migration
 422  *
 423  * @param cls unused
 424  * @param tc scheduler context (also unused)
 425  */
 426 static void
 427 gather_migration_blocks (void *cls,
 428                          const struct GNUNET_SCHEDULER_TaskContext *tc);
 429
 430
 431 /**
 432  * If the migration task is not currently running, consider
 433  * (re)scheduling it with the appropriate delay.
 434  */
 435 static void
 436 consider_gathering ()
 437 {
 438   struct GNUNET_TIME_Relative delay;
 439
 440   if (GSF_dsh == NULL)
 441     return;
 442   if (mig_qe != NULL)
 443     return;
 444   if (mig_task != GNUNET_SCHEDULER_NO_TASK)
 445     return;
 446   if (mig_size >= MAX_MIGRATION_QUEUE)
 447     return;
 448   delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, mig_size);
 449   delay = GNUNET_TIME_relative_divide (delay, MAX_MIGRATION_QUEUE);
 450   delay = GNUNET_TIME_relative_max (delay, min_migration_delay);
 451 #if DEBUG_FS_MIGRATION
 452   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 453               "Scheduling gathering task (queue size: %u)\n", mig_size);
 454 #endif
 455   mig_task =
 456       GNUNET_SCHEDULER_add_delayed (delay, &gather_migration_blocks, NULL);
 457 }
 458
 459
 460 /**
 461  * Process content offered for migration.
 462  *
 463  * @param cls closure
 464  * @param key key for the content
 465  * @param size number of bytes in data
 466  * @param data content stored
 467  * @param type type of the content
 468  * @param priority priority of the content
 469  * @param anonymity anonymity-level for the content
 470  * @param expiration expiration time for the content
 471  * @param uid unique identifier for the datum;
 472  *        maybe 0 if no unique identifier is available
 473  */
 474 static void
 475 process_migration_content (void *cls, const GNUNET_HashCode * key, size_t size,
 476                            const void *data, enum GNUNET_BLOCK_Type type,
 477                            uint32_t priority, uint32_t anonymity,
 478                            struct GNUNET_TIME_Absolute expiration, uint64_t uid)
 479 {
 480   struct MigrationReadyBlock *mb;
 481   struct MigrationReadyPeer *pos;
 482
 483   mig_qe = NULL;
 484   if (key == NULL)
 485   {
 486 #if DEBUG_FS_MIGRATION
 487     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "No content found for migration...\n");
 488 #endif
 489     consider_gathering ();
 490     return;
 491   }
 492   if (GNUNET_TIME_absolute_get_remaining (expiration).rel_value <
 493       MIN_MIGRATION_CONTENT_LIFETIME.rel_value)
 494   {
 495     /* content will expire soon, don't bother */
 496     consider_gathering ();
 497     return;
 498   }
 499   if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
 500   {
 501     if (GNUNET_OK !=
 502         GNUNET_FS_handle_on_demand_block (key, size, data, type, priority,
 503                                           anonymity, expiration, uid,
 504                                           &process_migration_content, NULL))
 505       consider_gathering ();
 506     return;
 507   }
 508 #if DEBUG_FS_MIGRATION
 509   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 510               "Retrieved block `%s' of type %u for migration (queue size: %u/%u)\n",
 511               GNUNET_h2s (key), type, mig_size + 1, MAX_MIGRATION_QUEUE);
 512 #endif
 513   mb = GNUNET_malloc (sizeof (struct MigrationReadyBlock) + size);
 514   mb->query = *key;
 515   mb->expiration = expiration;
 516   mb->size = size;
 517   mb->type = type;
 518   memcpy (&mb[1], data, size);
 519   GNUNET_CONTAINER_DLL_insert_after (mig_head, mig_tail, mig_tail, mb);
 520   mig_size++;
 521   pos = peer_head;
 522   while (pos != NULL)
 523   {
 524     if (NULL == pos->th)
 525     {
 526 #if DEBUG_FS_MIGRATION
 527       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 528                   "Preparing to push best content to peer\n");
 529 #endif
 530       if (GNUNET_YES == transmit_content (pos, mb))
 531         break;                  /* 'mb' was freed! */
 532     }
 533     pos = pos->next;
 534   }
 535   consider_gathering ();
 536 }
 537
 538
 539 /**
 540  * Task that is run periodically to obtain blocks for content
 541  * migration
 542  *
 543  * @param cls unused
 544  * @param tc scheduler context (also unused)
 545  */
 546 static void
 547 gather_migration_blocks (void *cls,
 548                          const struct GNUNET_SCHEDULER_TaskContext *tc)
 549 {
 550   mig_task = GNUNET_SCHEDULER_NO_TASK;
 551   if (mig_size >= MAX_MIGRATION_QUEUE)
 552     return;
 553   if (GSF_dsh != NULL)
 554   {
 555 #if DEBUG_FS_MIGRATION
 556     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 557                 "Asking datastore for content for replication (queue size: %u)\n",
 558                 mig_size);
 559 #endif
 560     mig_qe =
 561         GNUNET_DATASTORE_get_for_replication (GSF_dsh, 0, UINT_MAX,
 562                                               GNUNET_TIME_UNIT_FOREVER_REL,
 563                                               &process_migration_content, NULL);
 564     if (NULL == mig_qe)
 565       consider_gathering ();
 566   }
 567 }
 568
 569
 570 /**
 571  * A peer connected to us.  Start pushing content
 572  * to this peer.
 573  *
 574  * @param peer handle for the peer that connected
 575  */
 576 void
 577 GSF_push_start_ (struct GSF_ConnectedPeer *peer)
 578 {
 579   struct MigrationReadyPeer *mrp;
 580
 581   if (GNUNET_YES != enabled)
 582     return;
 583   mrp = GNUNET_malloc (sizeof (struct MigrationReadyPeer));
 584   mrp->peer = peer;
 585   find_content (mrp);
 586   GNUNET_CONTAINER_DLL_insert (peer_head, peer_tail, mrp);
 587 }
 588
 589
 590 /**
 591  * A peer disconnected from us.  Stop pushing content
 592  * to this peer.
 593  *
 594  * @param peer handle for the peer that disconnected
 595  */
 596 void
 597 GSF_push_stop_ (struct GSF_ConnectedPeer *peer)
 598 {
 599   struct MigrationReadyPeer *pos;
 600
 601   pos = peer_head;
 602   while (pos != NULL)
 603   {
 604     if (pos->peer == peer)
 605     {
 606       GNUNET_CONTAINER_DLL_remove (peer_head, peer_tail, pos);
 607       if (NULL != pos->th)
 608       {
 609         GSF_peer_transmit_cancel_ (pos->th);
 610         pos->th = NULL;
 611       }
 612       if (NULL != pos->msg)
 613       {
 614         GNUNET_free (pos->msg);
 615         pos->msg = NULL;
 616       }
 617       GNUNET_free (pos);
 618       return;
 619     }
 620     pos = pos->next;
 621   }
 622 }
 623
 624
 625 /**
 626  * Setup the module.
 627  */
 628 void
 629 GSF_push_init_ ()
 630 {
 631   enabled =
 632       GNUNET_CONFIGURATION_get_value_yesno (GSF_cfg, "FS", "CONTENT_PUSHING");
 633   if (GNUNET_YES != enabled)
 634     return;
 635
 636   if (GNUNET_OK !=
 637       GNUNET_CONFIGURATION_get_value_time (GSF_cfg, "fs", "MIN_MIGRATION_DELAY",
 638                                            &min_migration_delay))
 639   {
 640     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 641                 _
 642                 ("Invalid value specified for option `%s' in section `%s', content pushing disabled\n"),
 643                 "MIN_MIGRATION_DELAY", "fs");
 644     return;
 645   }
 646   consider_gathering ();
 647 }
 648
 649
 650 /**
 651  * Shutdown the module.
 652  */
 653 void
 654 GSF_push_done_ ()
 655 {
 656   if (GNUNET_SCHEDULER_NO_TASK != mig_task)
 657   {
 658     GNUNET_SCHEDULER_cancel (mig_task);
 659     mig_task = GNUNET_SCHEDULER_NO_TASK;
 660   }
 661   if (NULL != mig_qe)
 662   {
 663     GNUNET_DATASTORE_cancel (mig_qe);
 664     mig_qe = NULL;
 665   }
 666   while (NULL != mig_head)
 667     delete_migration_block (mig_head);
 668   GNUNET_assert (0 == mig_size);
 669 }
 670
 671 /* end of gnunet-service-fs_push.c */