qe->h->active_blocks += qe->blocks;
qe->h->active_downloads++;
qe->start_time = GNUNET_TIME_absolute_get ();
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Starting job %p (%u active)\n",
+ qe,
+ qe->h->active_downloads);
GNUNET_CONTAINER_DLL_remove (qe->h->pending_head, qe->h->pending_tail, qe);
GNUNET_CONTAINER_DLL_insert_after (qe->h->running_head, qe->h->running_tail,
qe->h->running_tail, qe);
GNUNET_TIME_relative_add (qe->run_time,
GNUNET_TIME_absolute_get_duration
(qe->start_time));
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Stopping job %p (%u active)\n",
+ qe,
+ qe->h->active_downloads);
GNUNET_CONTAINER_DLL_remove (qe->h->running_head, qe->h->running_tail, qe);
GNUNET_CONTAINER_DLL_insert_after (qe->h->pending_head, qe->h->pending_tail,
qe->h->pending_tail, qe);
num_probes_expired = 0;
num_downloads_active = 0;
num_downloads_expired = 0;
- for (qe = h->running_head; NULL != qe; qe = qe->next)
+ next = h->running_head;
+ while (NULL != (qe = next))
{
- run_time =
- GNUNET_TIME_relative_multiply (h->avg_block_latency,
- qe->blocks * qe->start_times);
+ next = qe->next;
switch (qe->priority)
{
case GNUNET_FS_QUEUE_PRIORITY_PROBE:
- num_probes_active++;
- /* run probes for at most 1s * number-of-restarts; note that
- as the total runtime of a probe is limited to 2m, we don't
- need to additionally limit the total time of a probe to
- strictly limit its lifetime. */
- run_time = GNUNET_TIME_relative_min (run_time,
- GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
- 1 + qe->start_times));
+ run_time = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 2);
end_time = GNUNET_TIME_absolute_add (qe->start_time, run_time);
rst = GNUNET_TIME_absolute_get_remaining (end_time);
- restart_at = GNUNET_TIME_relative_min (rst, restart_at);
if (0 == rst.rel_value)
+ {
num_probes_expired++;
+ stop_job (qe);
+ }
+ else
+ {
+ num_probes_active++;
+ restart_at = GNUNET_TIME_relative_min (rst, restart_at);
+ }
break;
case GNUNET_FS_QUEUE_PRIORITY_NORMAL:
- num_downloads_active++;
+ run_time =
+ GNUNET_TIME_relative_multiply (h->avg_block_latency,
+ qe->blocks * qe->start_times);
end_time = GNUNET_TIME_absolute_add (qe->start_time, run_time);
rst = GNUNET_TIME_absolute_get_remaining (end_time);
- restart_at = GNUNET_TIME_relative_min (rst, restart_at);
if (0 == rst.rel_value)
+ {
num_downloads_expired++;
+ stop_job (qe);
+ }
+ else
+ {
+ num_downloads_active++;
+ restart_at = GNUNET_TIME_relative_min (rst, restart_at);
+ }
break;
default:
GNUNET_break (0);
num_downloads_active,
num_downloads_expired,
num_downloads_waiting);
-
- /* calculate stop decisions */
- num_probes_change = 0;
- num_downloads_change = 0;
+ /* calculate start/stop decisions */
if (h->active_downloads + num_downloads_waiting > h->max_parallel_requests)
{
- if (num_probes_active > 0)
- num_probes_change = - GNUNET_MIN (num_probes_active,
- h->max_parallel_requests - (h->active_downloads + num_downloads_waiting));
- else if (h->active_downloads + num_downloads_waiting > h->max_parallel_requests)
- num_downloads_change = - GNUNET_MIN (num_downloads_expired,
- h->max_parallel_requests - (h->active_downloads + num_downloads_waiting));
+ /* stop probes if possible */
+ num_probes_change = - num_probes_active;
+ num_downloads_change = h->max_parallel_requests - h->active_downloads;
+ }
+ else
+ {
+ /* start all downloads */
+ num_downloads_change = num_downloads_waiting;
+ /* start as many probes as we can */
+ num_probes_change = GNUNET_MIN (num_probes_waiting,
+ h->max_parallel_requests - (h->active_downloads + num_downloads_waiting));
}
-
+
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Stopping %d probes and %d downloads\n",
+ "Changing %d probes and %d downloads\n",
num_probes_change,
num_downloads_change);
- /* then, check if we should stop some jobs */
+ /* actually stop probes */
next = h->running_head;
while (NULL != (qe = next))
{
next = qe->next;
- run_time =
- GNUNET_TIME_relative_multiply (h->avg_block_latency,
- qe->blocks * qe->start_times);
- switch (qe->priority)
+ if (GNUNET_FS_QUEUE_PRIORITY_PROBE != qe->priority)
+ continue;
+ if (num_probes_change < 0)
{
- case GNUNET_FS_QUEUE_PRIORITY_PROBE:
- /* run probes for at most 1s * number-of-restarts; note that
- as the total runtime of a probe is limited to 2m, we don't
- need to additionally limit the total time of a probe to
- strictly limit its lifetime. */
- run_time = GNUNET_TIME_relative_min (run_time,
- GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
- 1 + qe->start_times));
- end_time = GNUNET_TIME_absolute_add (qe->start_time, run_time);
- rst = GNUNET_TIME_absolute_get_remaining (end_time);
- restart_at = GNUNET_TIME_relative_min (rst, restart_at);
- if ( (num_probes_change < 0) &&
- ( (num_probes_expired < - num_probes_change) ||
- (0 == rst.rel_value) ) )
- {
- stop_job (qe);
- num_probes_change++;
- if (0 == rst.rel_value)
- num_probes_expired--;
- }
- break;
- case GNUNET_FS_QUEUE_PRIORITY_NORMAL:
- end_time = GNUNET_TIME_absolute_add (qe->start_time, run_time);
- rst = GNUNET_TIME_absolute_get_remaining (end_time);
- restart_at = GNUNET_TIME_relative_min (rst, restart_at);
- if ( (num_downloads_change < 0) &&
- ( (num_downloads_expired < - num_downloads_change) ||
- (0 == rst.rel_value) ) )
- {
- stop_job (qe);
- num_downloads_change++;
- if (0 == rst.rel_value)
- num_downloads_expired--;
- }
- break;
- default:
- GNUNET_break (0);
- break;
+ stop_job (qe);
+ num_probes_change++;
+ if (0 == num_probes_change)
+ break;
}
}
- GNUNET_break (0 == num_downloads_change);
- GNUNET_break (0 == num_probes_change);
-
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "AD: %u, MP: %u\n",
- h->active_downloads,
- h->max_parallel_requests);
-
- /* calculate start decisions */
- if (h->active_downloads + num_downloads_waiting < h->max_parallel_requests)
- {
- /* can start all downloads, fill rest with probes */
- num_downloads_change = num_downloads_waiting;
- num_probes_change = GNUNET_MIN (num_probes_waiting,
- h->max_parallel_requests - (h->active_downloads + num_downloads_waiting));
- }
- else
- {
- num_downloads_change = h->max_parallel_requests - h->active_downloads;
- num_probes_change = 0;
- }
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Starting %d probes and %d downloads\n",
- num_probes_change,
- num_downloads_change);
+ GNUNET_break (0 <= num_probes_change);
- /* finally, start some more tasks if we now have empty slots */
+ /* start some more tasks if we now have empty slots */
block_limit_hit = GNUNET_NO;
next = h->pending_head;
- while (NULL != (qe = next))
+ while ( (NULL != (qe = next)) &&
+ ( (num_probes_change > 0) ||
+ (num_downloads_change > 0) ) )
{
next = qe->next;
switch (qe->priority)
{
start_job (qe);
num_probes_change--;
+ run_time = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 2);
+ restart_at = GNUNET_TIME_relative_min (run_time, restart_at);
}
break;
case GNUNET_FS_QUEUE_PRIORITY_NORMAL:
}
GNUNET_break ( (0 == num_downloads_change) || (GNUNET_YES == block_limit_hit) );
GNUNET_break (0 == num_probes_change);
+
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Left with %d probes and %d downloads to start\n",
+ "AD: %u, MP: %u; %d probes and %d downloads to start, will run again in %s\n",
+ h->active_downloads,
+ h->max_parallel_requests,
num_probes_change,
- num_downloads_change);
+ num_downloads_change,
+ GNUNET_STRINGS_relative_time_to_string (restart_at, GNUNET_YES));
/* make sure we run again */
h->queue_job =
if (h->queue_job != GNUNET_SCHEDULER_NO_TASK)
GNUNET_SCHEDULER_cancel (h->queue_job);
h->queue_job = GNUNET_SCHEDULER_add_now (&process_job_queue, h);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Queueing job %p\n",
+ qe);
return qe;
}
/**
* Dequeue a job from the queue.
- * @param qh handle for the job
+ *
+ * @param qe handle for the job
*/
void
-GNUNET_FS_dequeue_ (struct GNUNET_FS_QueueEntry *qh)
+GNUNET_FS_dequeue_ (struct GNUNET_FS_QueueEntry *qe)
{
struct GNUNET_FS_Handle *h;
- h = qh->h;
- if (NULL != qh->client)
- stop_job (qh);
- GNUNET_CONTAINER_DLL_remove (h->pending_head, h->pending_tail, qh);
- GNUNET_free (qh);
+ h = qe->h;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Dequeueing job %p\n",
+ qe);
+ if (NULL != qe->client)
+ stop_job (qe);
+ GNUNET_CONTAINER_DLL_remove (h->pending_head, h->pending_tail, qe);
+ GNUNET_free (qe);
if (h->queue_job != GNUNET_SCHEDULER_NO_TASK)
GNUNET_SCHEDULER_cancel (h->queue_job);
h->queue_job = GNUNET_SCHEDULER_add_now (&process_job_queue, h);
dn = get_serialization_file_name_in_dir (h, ext, uni, "");
if (NULL == dn)
return;
- if ((GNUNET_OK == GNUNET_DISK_directory_test (dn)) &&
+ if ((GNUNET_YES == GNUNET_DISK_directory_test (dn, GNUNET_YES)) &&
(GNUNET_OK != GNUNET_DISK_directory_remove (dn)))
GNUNET_log_strerror_file (GNUNET_ERROR_TYPE_WARNING, "rmdir", dn);
GNUNET_free (dn);
struct GNUNET_FS_PublishContext *pc;
int32_t options;
int32_t all_done;
+ int32_t have_ns;
char *fi_root;
- char *ns;
+ struct GNUNET_CRYPTO_EccPrivateKey ns;
char *fi_pos;
char *emsg;
- pc = GNUNET_malloc (sizeof (struct GNUNET_FS_PublishContext));
+ pc = GNUNET_new (struct GNUNET_FS_PublishContext);
pc->h = h;
pc->serialization = get_serialization_short_name (filename);
fi_root = NULL;
fi_pos = NULL;
- ns = NULL;
rh = GNUNET_BIO_read_open (filename);
if (NULL == rh)
{
GNUNET_BIO_read_string (rh, "publish-nuid", &pc->nuid, 1024)) ||
(GNUNET_OK != GNUNET_BIO_read_int32 (rh, &options)) ||
(GNUNET_OK != GNUNET_BIO_read_int32 (rh, &all_done)) ||
+ (GNUNET_OK != GNUNET_BIO_read_int32 (rh, &have_ns)) ||
(GNUNET_OK !=
GNUNET_BIO_read_string (rh, "publish-firoot", &fi_root, 128)) ||
(GNUNET_OK != GNUNET_BIO_read_string (rh, "publish-fipos", &fi_pos, 128))
- || (GNUNET_OK != GNUNET_BIO_read_string (rh, "publish-ns", &ns, 1024)))
+ || ( (GNUNET_YES == have_ns) &&
+ (GNUNET_OK != GNUNET_BIO_read (rh, "publish-ns", &ns, sizeof (ns)))) )
{
GNUNET_break (0);
goto cleanup;
GNUNET_break (0);
goto cleanup;
}
- if (NULL != ns)
+ if (GNUNET_YES == have_ns)
{
- pc->ns = GNUNET_FS_namespace_create (h, ns);
- if (NULL == pc->ns)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- _
- ("Failed to recover namespace `%s', cannot resume publishing operation.\n"),
- ns);
- goto cleanup;
- }
+ pc->ns = GNUNET_new (struct GNUNET_CRYPTO_EccPrivateKey);
+ *pc->ns = ns;
}
if ((0 == (pc->options & GNUNET_FS_PUBLISH_OPTION_SIMULATE_ONLY)) &&
(GNUNET_YES != pc->all_done))
filename, emsg);
GNUNET_free (emsg);
}
- GNUNET_free_non_null (ns);
pc->top = GNUNET_FS_make_top (h, &GNUNET_FS_publish_signal_suspend_, pc);
return GNUNET_OK;
cleanup:
GNUNET_free_non_null (pc->nuid);
GNUNET_free_non_null (fi_root);
GNUNET_free_non_null (fi_pos);
- GNUNET_free_non_null (ns);
if ((NULL != rh) && (GNUNET_OK != GNUNET_BIO_read_close (rh, &emsg)))
{
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
GNUNET_FS_publish_sync_ (struct GNUNET_FS_PublishContext *pc)
{
struct GNUNET_BIO_WriteHandle *wh;
+ int32_t have_ns;
if (NULL == pc->serialization)
pc->serialization =
GNUNET_break (0);
goto cleanup;
}
+ have_ns = (NULL != pc->ns) ? GNUNET_YES : GNUNET_NO;
if ((GNUNET_OK != GNUNET_BIO_write_string (wh, pc->nid)) ||
(GNUNET_OK != GNUNET_BIO_write_string (wh, pc->nuid)) ||
(GNUNET_OK != GNUNET_BIO_write_int32 (wh, pc->options)) ||
(GNUNET_OK != GNUNET_BIO_write_int32 (wh, pc->all_done)) ||
+ (GNUNET_OK != GNUNET_BIO_write_int32 (wh, have_ns)) ||
(GNUNET_OK != GNUNET_BIO_write_string (wh, pc->fi->serialization)) ||
(GNUNET_OK !=
GNUNET_BIO_write_string (wh,
(NULL == pc->fi_pos) ? NULL : pc->fi_pos->serialization)) ||
- (GNUNET_OK !=
- GNUNET_BIO_write_string (wh,
- (NULL == pc->ns) ? NULL : pc->ns->name)))
+ ( (NULL != pc->ns) &&
+ GNUNET_BIO_write (wh,
+ pc->ns,
+ sizeof (struct GNUNET_CRYPTO_EccPrivateKey)) ) )
{
GNUNET_break (0);
goto cleanup;
/**
* Read a download request tree.
*
- * @param rh stream to read from
+ * @param rh mesh to read from
* @return value the 'struct DownloadRequest', NULL on error
*/
static struct DownloadRequest *
struct GNUNET_BIO_WriteHandle *wh;
char *uris;
+ if (NULL == sr->sc)
+ return;
uris = NULL;
if (NULL == sr->serialization)
sr->serialization =
- make_serialization_file_name_in_dir (sr->sc->h,
+ make_serialization_file_name_in_dir (sr->h,
(sr->sc->psearch_result ==
NULL) ?
GNUNET_FS_SYNC_PATH_MASTER_SEARCH :
sr->sc->serialization);
if (NULL == sr->serialization)
return;
- wh = get_write_handle_in_dir (sr->sc->h,
+ wh = get_write_handle_in_dir (sr->h,
(sr->sc->psearch_result ==
NULL) ? GNUNET_FS_SYNC_PATH_MASTER_SEARCH :
GNUNET_FS_SYNC_PATH_CHILD_SEARCH,
goto cleanup;
}
if ( (NULL != sr->uri) &&
- (ksk == sr->sc->uri->type) &&
+ (GNUNET_FS_URI_KSK == sr->sc->uri->type) &&
(GNUNET_OK != GNUNET_BIO_write (wh, sr->keyword_bitmap,
(sr->sc->uri->data.ksk.keywordCount + 7) / 8)) )
{
GNUNET_free_non_null (uris);
if (NULL != wh)
(void) GNUNET_BIO_write_close (wh);
- remove_sync_file_in_dir (sr->sc->h,
+ remove_sync_file_in_dir (sr->h,
(NULL == sr->sc->psearch_result)
? GNUNET_FS_SYNC_PATH_MASTER_SEARCH
: GNUNET_FS_SYNC_PATH_CHILD_SEARCH,
if (NULL == uc->ksk_uri)
{
GNUNET_break (0);
+ GNUNET_free_non_null (emsg);
goto cleanup;
}
}
download = NULL;
update_srch = NULL;
sr = GNUNET_malloc (sizeof (struct GNUNET_FS_SearchResult));
+ sr->h = sc->h;
sr->sc = sc;
sr->serialization = ser;
if ((GNUNET_OK != GNUNET_BIO_read_string (rh, "result-uri", &uris, 10 * 1024))
GNUNET_break (0);
goto cleanup;
}
- if (ksk == sr->sc->uri->type)
+ if (GNUNET_FS_URI_KSK == sr->sc->uri->type)
{
sr->keyword_bitmap = GNUNET_malloc ((sr->sc->uri->data.ksk.keywordCount + 7) / 8); /* round up, count bits */
if (GNUNET_OK != GNUNET_BIO_read (rh, "keyword-bitmap",
sr->availability_trials;
pi.value.search.specifics.resume_result.applicability_rank =
sr->optional_support;
- sr->client_info = GNUNET_FS_search_make_status_ (&pi, sc);
+ sr->client_info = GNUNET_FS_search_make_status_ (&pi, sc->h, sc);
}
if (NULL != sr->download)
{
}
dc->options = (enum GNUNET_FS_DownloadOptions) options;
dc->active =
- GNUNET_CONTAINER_multihashmap_create (1 + 2 * (dc->length / DBLOCK_SIZE));
+ GNUNET_CONTAINER_multihashmap_create (1 + 2 * (dc->length / DBLOCK_SIZE), GNUNET_NO);
dc->has_finished = (int) status;
dc->treedepth =
GNUNET_FS_compute_depth (GNUNET_FS_uri_chk_get_file_size (dc->uri));
dn = get_download_sync_filename (dc, dc->serialization, ".dir");
if (NULL != dn)
{
- if (GNUNET_YES == GNUNET_DISK_directory_test (dn))
+ if (GNUNET_YES == GNUNET_DISK_directory_test (dn, GNUNET_YES))
GNUNET_DISK_directory_scan (dn, &deserialize_subdownload, dc);
GNUNET_free (dn);
}
if (NULL != parent)
{
- GNUNET_abort (); // for debugging for now - FIXME
GNUNET_CONTAINER_DLL_insert (parent->child_head, parent->child_tail, dc);
}
if (NULL != search)
pi.value.search.specifics.resume.message = sc->emsg;
pi.value.search.specifics.resume.is_paused =
(NULL == sc->client) ? GNUNET_YES : GNUNET_NO;
- sc->client_info = GNUNET_FS_search_make_status_ (&pi, sc);
+ sc->client_info = GNUNET_FS_search_make_status_ (&pi, sc->h, sc);
GNUNET_CONTAINER_multihashmap_iterate (sc->master_result_map,
&signal_result_resume, sc);
goto cleanup;
}
sc->options = (enum GNUNET_FS_SearchOptions) options;
- sc->master_result_map = GNUNET_CONTAINER_multihashmap_create (16);
+ sc->master_result_map = GNUNET_CONTAINER_multihashmap_create (16, GNUNET_NO);
dn = get_serialization_file_name_in_dir (h,
(sc->psearch_result ==
NULL) ?
sc->serialization, "");
if (NULL != dn)
{
- if (GNUNET_YES == GNUNET_DISK_directory_test (dn))
+ if (GNUNET_YES == GNUNET_DISK_directory_test (dn, GNUNET_YES))
GNUNET_DISK_directory_scan (dn, &deserialize_search_result, sc);
GNUNET_free (dn);
}
dn = get_serialization_file_name (h, master_path, "");
if (NULL == dn)
return;
- if (GNUNET_YES == GNUNET_DISK_directory_test (dn))
+ if (GNUNET_YES == GNUNET_DISK_directory_test (dn, GNUNET_YES))
GNUNET_DISK_directory_scan (dn, proc, h);
GNUNET_free (dn);
}