-fixing #2139

[oweals/gnunet.git] / src / fs / gnunet-service-fs_pr.c

diff --git a/src/fs/gnunet-service-fs_pr.c b/src/fs/gnunet-service-fs_pr.c
index dcaba1203e48fafbff03b1b3ed4d596003b098eb..c16f94464e5fd9f54d363de1ef005b366cd0a81d 100644 (file)
--- a/src/fs/gnunet-service-fs_pr.c
+++ b/src/fs/gnunet-service-fs_pr.c
@@ -360,8 +360,8 @@ GSF_pending_request_create_ (enum GSF_PendingRequestOptions options,
         break;                  /* let the request live briefly... */
       if (NULL != dpr->rh)
        dpr->rh (dpr->rh_cls, GNUNET_BLOCK_EVALUATION_REQUEST_VALID, dpr,
-                UINT32_MAX, GNUNET_TIME_UNIT_FOREVER_ABS, GNUNET_BLOCK_TYPE_ANY,
-                NULL, 0);
+                UINT32_MAX, GNUNET_TIME_UNIT_FOREVER_ABS, GNUNET_TIME_UNIT_FOREVER_ABS,
+                 GNUNET_BLOCK_TYPE_ANY, NULL, 0);
       GSF_pending_request_cancel_ (dpr, GNUNET_YES);
     }
   }
@@ -371,7 +371,6 @@ GSF_pending_request_create_ (enum GSF_PendingRequestOptions options,
   return pr;
 }
 
-
 /**
  * Obtain the public data associated with a pending request
  *
@@ -775,6 +774,7 @@ process_reply (void *cls, const GNUNET_HashCode * key, void *value)
   struct ProcessReplyClosure *prq = cls;
   struct GSF_PendingRequest *pr = value;
   GNUNET_HashCode chash;
+  struct GNUNET_TIME_Absolute last_transmission;
 
   if (NULL == pr->rh)
     return GNUNET_YES;
@@ -804,9 +804,11 @@ process_reply (void *cls, const GNUNET_HashCode * key, void *value)
     GNUNET_LOAD_update (GSF_rt_entry_lifetime,
                         GNUNET_TIME_absolute_get_duration (pr->
                                                            public_data.start_time).rel_value);
+    if (!GSF_request_plan_reference_get_last_transmission_ (pr->public_data.rpr_head, prq->sender, &last_transmission))
+      last_transmission.abs_value = GNUNET_TIME_UNIT_FOREVER_ABS.abs_value;
     /* pass on to other peers / local clients */
     pr->rh (pr->rh_cls, prq->eval, pr, prq->anonymity_level, prq->expiration,
-            prq->type, prq->data, prq->size);
+            last_transmission, prq->type, prq->data, prq->size);
     return GNUNET_YES;
   case GNUNET_BLOCK_EVALUATION_OK_DUPLICATE:
     GNUNET_STATISTICS_update (GSF_stats,
@@ -855,8 +857,10 @@ process_reply (void *cls, const GNUNET_HashCode * key, void *value)
   pr->public_data.results_found++;
   prq->request_found = GNUNET_YES;
   /* finally, pass on to other peer / local client */
+  if (!GSF_request_plan_reference_get_last_transmission_ (pr->public_data.rpr_head, prq->sender, &last_transmission))
+    last_transmission.abs_value = GNUNET_TIME_UNIT_FOREVER_ABS.abs_value;
   pr->rh (pr->rh_cls, prq->eval, pr, prq->anonymity_level, prq->expiration,
-          prq->type, prq->data, prq->size);
+          last_transmission, prq->type, prq->data, prq->size);
   return GNUNET_YES;
 }
 
@@ -901,18 +905,66 @@ put_migration_continuation (void *cls, int success,
 {
   struct PutMigrationContext *pmc = cls;
   struct GSF_ConnectedPeer *cp;
+  struct GNUNET_TIME_Relative mig_pause;
+  struct GSF_PeerPerformanceData *ppd;
 
-  cp = GSF_peer_get_ (&pmc->origin);
-  if ((GNUNET_OK != success) && (GNUNET_NO == pmc->requested) && (min_expiration.abs_value > 0)&&
-      (NULL != cp) )
-    GSF_block_peer_migration_ (cp, min_expiration);      
-  GNUNET_free (pmc);
-  /* on failure, increase the put load dramatically */
   if (NULL != datastore_put_load)
-    GNUNET_LOAD_update (datastore_put_load, 
-                       GNUNET_TIME_UNIT_HOURS.rel_value);
+  {
+    if (GNUNET_SYSERR != success)
+    {
+      GNUNET_LOAD_update (datastore_put_load, 
+                         GNUNET_TIME_absolute_get_duration (pmc->start).rel_value);
+    }
+    else
+    {
+      /* on queue failure / timeout, increase the put load dramatically */
+      GNUNET_LOAD_update (datastore_put_load, 
+                         GNUNET_TIME_UNIT_MINUTES.rel_value);
+    }
+  }
+  cp = GSF_peer_get_ (&pmc->origin);
   if (GNUNET_OK == success)
+  {
+    if (NULL != cp)
+    {
+      ppd = GSF_get_peer_performance_data_ (cp);
+      ppd->migration_delay.rel_value /= 2;
+    }
+    GNUNET_free (pmc);
     return;
+  }
+  if ( (GNUNET_NO == success) && 
+       (GNUNET_NO == pmc->requested) && 
+       (NULL != cp) )
+  {
+    ppd = GSF_get_peer_performance_data_ (cp);
+    if (min_expiration.abs_value > 0)
+    {
+#if DEBUG_FS
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 
+                 "Asking to stop migration for %llu ms because datastore is full\n",
+                 (unsigned long long) GNUNET_TIME_absolute_get_remaining (min_expiration).rel_value);
+#endif
+      GSF_block_peer_migration_ (cp, min_expiration);      
+    }
+    else
+    {
+      ppd->migration_delay = GNUNET_TIME_relative_max (GNUNET_TIME_UNIT_SECONDS,
+                                                      ppd->migration_delay);
+      ppd->migration_delay = GNUNET_TIME_relative_min (GNUNET_TIME_UNIT_HOURS,
+                                                      ppd->migration_delay);
+      mig_pause.rel_value = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
+                                                     ppd->migration_delay.rel_value);
+      ppd->migration_delay = GNUNET_TIME_relative_multiply (ppd->migration_delay, 2);
+#if DEBUG_FS
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 
+                 "Replicated content already exists locally, asking to stop migration for %llu ms\n",
+                 (unsigned long long) mig_pause.rel_value);
+#endif
+      GSF_block_peer_migration_ (cp, GNUNET_TIME_relative_to_absolute (mig_pause));
+    }
+  }
+  GNUNET_free (pmc);
   GNUNET_STATISTICS_update (GSF_stats,
                             gettext_noop ("# Datastore `PUT' failures"), 1,
                             GNUNET_NO);
@@ -1006,7 +1058,7 @@ handle_dht_reply (void *cls, struct GNUNET_TIME_Absolute exp,
                               GNUNET_CONSTANTS_SERVICE_TIMEOUT,
                               &put_migration_continuation, pmc))
     {
-      put_migration_continuation (pmc, GNUNET_NO, GNUNET_TIME_UNIT_ZERO_ABS, NULL);
+      put_migration_continuation (pmc, GNUNET_SYSERR, GNUNET_TIME_UNIT_ZERO_ABS, NULL);
     }
   }
 }
@@ -1169,7 +1221,7 @@ process_local_reply (void *cls, const GNUNET_HashCode * key, size_t size,
   if (NULL == key)
   {
 #if DEBUG_FS
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
                 "No further local responses available.\n");
 #endif
     if ((pr->public_data.type == GNUNET_BLOCK_TYPE_FS_DBLOCK) ||
@@ -1524,7 +1576,7 @@ GSF_handle_p2p_content_ (struct GSF_ConnectedPeer *cp,
                               GNUNET_CONSTANTS_SERVICE_TIMEOUT,
                               &put_migration_continuation, pmc))
     {
-      put_migration_continuation (pmc, GNUNET_NO, GNUNET_TIME_UNIT_ZERO_ABS, NULL);
+      put_migration_continuation (pmc, GNUNET_SYSERR, GNUNET_TIME_UNIT_ZERO_ABS, NULL);
     }
   }
   else
@@ -1549,6 +1601,14 @@ GSF_handle_p2p_content_ (struct GSF_ConnectedPeer *cp,
                                        GNUNET_CRYPTO_random_u32
                                        (GNUNET_CRYPTO_QUALITY_WEAK,
                                         (unsigned int) (60000 * putl * putl)));
+#if DEBUG_FS
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 
+               "Asking to stop migration for %llu ms because of load %f and events %d/%d\n",
+               (unsigned long long) block_time.rel_value,
+               putl,
+               active_to_migration,
+               (GNUNET_NO == prq.request_found));
+#endif
     GSF_block_peer_migration_ (cp, GNUNET_TIME_relative_to_absolute (block_time));
   }
   return GNUNET_OK;