#include "fs_api.h"
#include "fs_tree.h"
+/**
+ * How many block requests can we have outstanding in parallel at a time by default?
+ */
+#define DEFAULT_MAX_PARALLEL_REQUESTS (1024 * 10)
+
+/**
+ * How many downloads can we have outstanding in parallel at a time by default?
+ */
+#define DEFAULT_MAX_PARALLEL_DOWNLOADS 16
/**
* Start the given job (send signal, remove from pending queue, update
struct GNUNET_TIME_Absolute end_time;
h->queue_job = GNUNET_SCHEDULER_NO_TASK;
+ restart_at = GNUNET_TIME_UNIT_FOREVER_REL;
+ /* first, see if we can start all the jobs */
next = h->pending_head;
while (NULL != (qe = next))
{
continue;
}
if ((qe->blocks + h->active_blocks <= h->max_parallel_requests) &&
- (h->active_downloads + 1 <= h->max_parallel_downloads))
+ (h->active_downloads < h->max_parallel_downloads))
{
start_job (qe);
continue;
}
if (h->pending_head == NULL)
return; /* no need to stop anything */
- restart_at = GNUNET_TIME_UNIT_FOREVER_REL;
+ /* then, check if we should stop some jobs */
next = h->running_head;
while (NULL != (qe = next))
{
run_time =
GNUNET_TIME_relative_multiply (h->avg_block_latency,
qe->blocks * qe->start_times);
+ switch (qe->priority)
+ {
+ 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));
+ break;
+ case GNUNET_FS_QUEUE_PRIORITY_NORMAL:
+ break;
+ default:
+ GNUNET_break (0);
+ }
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);
continue;
stop_job (qe);
}
+ /* finally, start some more tasks if we now have empty slots */
+ next = h->pending_head;
+ while (NULL != (qe = next))
+ {
+ next = qe->next;
+ if ((qe->blocks + h->active_blocks <= h->max_parallel_requests) &&
+ (h->active_downloads < h->max_parallel_downloads))
+ {
+ start_job (qe);
+ continue;
+ }
+ }
h->queue_job =
GNUNET_SCHEDULER_add_delayed (restart_at, &process_job_queue, h);
}
* @param stop function to call to pause the job, or on dequeue (if the job was running)
* @param cls closure for start and stop
* @param blocks number of blocks this jobs uses
+ * @param priority how important is this download
* @return queue handle
*/
struct GNUNET_FS_QueueEntry *
GNUNET_FS_queue_ (struct GNUNET_FS_Handle *h, GNUNET_FS_QueueStart start,
- GNUNET_FS_QueueStop stop, void *cls, unsigned int blocks)
+ GNUNET_FS_QueueStop stop, void *cls, unsigned int blocks,
+ enum GNUNET_FS_QueuePriority priority)
{
struct GNUNET_FS_QueueEntry *qe;
qe->cls = cls;
qe->queue_time = GNUNET_TIME_absolute_get ();
qe->blocks = blocks;
+ qe->priority = priority;
GNUNET_CONTAINER_DLL_insert_after (h->pending_head, h->pending_tail,
h->pending_tail, qe);
if (h->queue_job != GNUNET_SCHEDULER_NO_TASK)
* @param cls closure (points to the file information)
* @param offset offset to read from; it is possible
* that the caller might need to go backwards
- * a bit at times
+ * a bit at times; set to UINT64_MAX to tell
+ * the reader that we won't be reading for a while
+ * (used to close the file descriptor but NOT fully
+ * clean up the reader's state); in this case,
+ * a value of '0' for max should be ignored
* @param max maximum number of bytes that should be
* copied to buf; readers are not allowed
* to provide less data unless there is an error;
struct FileInfo *fi = cls;
ssize_t ret;
- if (max == 0)
+ if (UINT64_MAX == offset)
{
- if (fi->fd != NULL)
+ if (NULL != fi->fd)
+ {
+ GNUNET_DISK_file_close (fi->fd);
+ fi->fd = NULL;
+ }
+ return 0;
+ }
+ if (0 == max)
+ {
+ if (NULL != fi->fd)
GNUNET_DISK_file_close (fi->fd);
GNUNET_free (fi->filename);
GNUNET_free (fi);
return 0;
}
- if (fi->fd == NULL)
+ if (NULL == fi->fd)
{
fi->fd =
GNUNET_DISK_file_open (fi->filename, GNUNET_DISK_OPEN_READ,
GNUNET_DISK_PERM_NONE);
- if (fi->fd == NULL)
+ if (NULL == fi->fd)
{
GNUNET_asprintf (emsg, _("Could not open file `%s': %s"), fi->filename,
STRERROR (errno));
}
GNUNET_DISK_file_seek (fi->fd, offset, GNUNET_DISK_SEEK_SET);
ret = GNUNET_DISK_file_read (fi->fd, buf, max);
- if (ret == -1)
+ if (-1 == ret)
{
GNUNET_asprintf (emsg, _("Could not read file `%s': %s"), fi->filename,
STRERROR (errno));
* @param cls closure (points to the buffer)
* @param offset offset to read from; it is possible
* that the caller might need to go backwards
- * a bit at times
+ * a bit at times; set to UINT64_MAX to tell
+ * the reader that we won't be reading for a while
+ * (used to close the file descriptor but NOT fully
+ * clean up the reader's state); in this case,
+ * a value of '0' for max should be ignored
* @param max maximum number of bytes that should be
* copied to buf; readers are not allowed
* to provide less data unless there is an error;
{
char *data = cls;
+ if (UINT64_MAX == offset)
+ return 0;
if (max == 0)
{
GNUNET_free_non_null (data);
filename = NULL;
if ((GNUNET_OK != GNUNET_BIO_read_meta_data (rh, "metadata", &ret->meta)) ||
(GNUNET_OK != GNUNET_BIO_read_string (rh, "ksk-uri", &ksks, 32 * 1024)) ||
- ((ksks != NULL) &&
- (NULL == (ret->keywords = GNUNET_FS_uri_parse (ksks, NULL)))) ||
- (GNUNET_YES != GNUNET_FS_uri_test_ksk (ret->keywords)) ||
+ ( (NULL != ksks) &&
+ ( (NULL == (ret->keywords = GNUNET_FS_uri_parse (ksks, NULL))) ||
+ (GNUNET_YES != GNUNET_FS_uri_test_ksk (ret->keywords)) ) ) ||
(GNUNET_OK != GNUNET_BIO_read_string (rh, "chk-uri", &chks, 1024)) ||
- ((chks != NULL) &&
- ((NULL == (ret->chk_uri = GNUNET_FS_uri_parse (chks, NULL))) ||
- (GNUNET_YES != GNUNET_FS_uri_test_chk (ret->chk_uri)))) ||
+ ( (NULL != chks) &&
+ ( (NULL == (ret->chk_uri = GNUNET_FS_uri_parse (chks, NULL))) ||
+ (GNUNET_YES != GNUNET_FS_uri_test_chk (ret->chk_uri))) ) ||
(GNUNET_OK != read_start_time (rh, &ret->start_time)) ||
(GNUNET_OK != GNUNET_BIO_read_string (rh, "emsg", &ret->emsg, 16 * 1024))
|| (GNUNET_OK !=
{
if (0 == strcmp (srch, pos->serialization))
return pos;
- if (pos->is_directory)
+ if (pos->is_directory == GNUNET_YES)
{
r = find_file_position (pos->data.dir.entries, srch);
if (r != NULL)
struct GNUNET_FS_BlockOptions *bo, int *do_index,
void **client_info)
{
- struct GNUNET_FS_PublishContext *sc = cls;
+ struct GNUNET_FS_PublishContext *pc = cls;
struct GNUNET_FS_ProgressInfo pi;
+ if (GNUNET_YES == pc->skip_next_fi_callback)
+ {
+ pc->skip_next_fi_callback = GNUNET_NO;
+ return GNUNET_OK;
+ }
pi.status = GNUNET_FS_STATUS_PUBLISH_RESUME;
- pi.value.publish.specifics.resume.message = sc->fi->emsg;
- pi.value.publish.specifics.resume.chk_uri = sc->fi->chk_uri;
- *client_info = GNUNET_FS_publish_make_status_ (&pi, sc, fi, 0);
+ pi.value.publish.specifics.resume.message = pc->fi->emsg;
+ pi.value.publish.specifics.resume.chk_uri = pc->fi->chk_uri;
+ *client_info = GNUNET_FS_publish_make_status_ (&pi, pc, fi, 0);
+ if (GNUNET_YES == GNUNET_FS_meta_data_test_for_directory (meta))
+ {
+ /* process entries in directory */
+ pc->skip_next_fi_callback = GNUNET_YES;
+ GNUNET_FS_file_information_inspect (fi, &fip_signal_resume, pc);
+ }
return GNUNET_OK;
}
GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == pc->upload_task);
pc->upload_task =
GNUNET_SCHEDULER_add_with_priority
- (GNUNET_SCHEDULER_PRIORITY_BACKGROUND, &GNUNET_FS_publish_main_, pc);
+ (GNUNET_SCHEDULER_PRIORITY_BACKGROUND,
+ &GNUNET_FS_publish_main_, pc);
}
if (GNUNET_OK != GNUNET_BIO_read_close (rh, &emsg))
{
GNUNET_FS_unindex_sync_ (struct GNUNET_FS_UnindexContext *uc)
{
struct GNUNET_BIO_WriteHandle *wh;
+ char *uris;
if (NULL == uc->serialization)
uc->serialization =
GNUNET_break (0);
goto cleanup;
}
+ if (NULL != uc->ksk_uri)
+ uris = GNUNET_FS_uri_to_string (uc->ksk_uri);
+ else
+ uris = NULL;
if ((GNUNET_OK != GNUNET_BIO_write_string (wh, uc->filename)) ||
(GNUNET_OK != GNUNET_BIO_write_int64 (wh, uc->file_size)) ||
(GNUNET_OK != write_start_time (wh, uc->start_time)) ||
(GNUNET_OK != GNUNET_BIO_write_int32 (wh, (uint32_t) uc->state)) ||
+ (GNUNET_OK !=
+ GNUNET_BIO_write (wh, &uc->chk, sizeof (struct ContentHashKey))) ||
+ (GNUNET_OK != GNUNET_BIO_write_string (wh, uris)) ||
+ (GNUNET_OK != GNUNET_BIO_write_int32 (wh, (uint32_t) uc->ksk_offset)) ||
((uc->state == UNINDEX_STATE_FS_NOTIFY) &&
(GNUNET_OK !=
GNUNET_BIO_write (wh, &uc->file_id, sizeof (GNUNET_HashCode)))) ||
char *fn;
char *dir;
+ if (0 != (dc->options & GNUNET_FS_DOWNLOAD_IS_PROBE))
+ return; /* we don't sync probes */
if (NULL == dc->serialization)
{
dir = get_download_sync_filename (dc, "", "");
|| (GNUNET_OK != GNUNET_BIO_write_int32 (wh, sr->mandatory_missing)) ||
(GNUNET_OK != GNUNET_BIO_write_int32 (wh, sr->optional_support)) ||
(GNUNET_OK != GNUNET_BIO_write_int32 (wh, sr->availability_success)) ||
- (GNUNET_OK != GNUNET_BIO_write_int32 (wh, sr->availability_trials)) ||
- (GNUNET_OK != GNUNET_BIO_write (wh, sr->keyword_bitmap,
- (sr->uri == NULL) ? 0 : (sr->uri->data.ksk.keywordCount + 7) / 8)) )
-
+ (GNUNET_OK != GNUNET_BIO_write_int32 (wh, sr->availability_trials)) )
+ {
+ GNUNET_break (0);
+ goto cleanup;
+ }
+ if ( (sr->uri != NULL) &&
+ (sr->sc->uri->type == ksk) &&
+ (GNUNET_OK != GNUNET_BIO_write (wh, sr->keyword_bitmap,
+ (sr->sc->uri->data.ksk.keywordCount + 7) / 8)) )
{
GNUNET_break (0);
goto cleanup;
struct GNUNET_FS_UnindexContext *uc;
struct GNUNET_FS_ProgressInfo pi;
char *emsg;
+ char *uris;
uint32_t state;
uc = GNUNET_malloc (sizeof (struct GNUNET_FS_UnindexContext));
GNUNET_break (0);
goto cleanup;
}
+ uris = NULL;
if ((GNUNET_OK !=
GNUNET_BIO_read_string (rh, "unindex-fn", &uc->filename, 10 * 1024)) ||
(GNUNET_OK != GNUNET_BIO_read_int64 (rh, &uc->file_size)) ||
(GNUNET_OK != read_start_time (rh, &uc->start_time)) ||
- (GNUNET_OK != GNUNET_BIO_read_int32 (rh, &state)))
+ (GNUNET_OK != GNUNET_BIO_read_int32 (rh, &state)) ||
+ (GNUNET_OK != GNUNET_BIO_read (rh, "uri", &uc->chk, sizeof (struct ContentHashKey))) ||
+ (GNUNET_OK != GNUNET_BIO_read_string (rh, "unindex-kskuri", &uris, 10 * 1024)) ||
+ (GNUNET_OK != GNUNET_BIO_read_int32 (rh, &uc->ksk_offset)) )
{
+ GNUNET_free_non_null (uris);
GNUNET_break (0);
goto cleanup;
}
+ if (NULL != uris)
+ {
+ uc->ksk_uri = GNUNET_FS_uri_parse (uris, &emsg);
+ GNUNET_free (uris);
+ if (NULL == uc->ksk_uri)
+ {
+ GNUNET_break (0);
+ goto cleanup;
+ }
+ }
+ if ( (uc->ksk_offset > 0) &&
+ ( (NULL == uc->ksk_uri) ||
+ (uc->ksk_offset > uc->ksk_uri->data.ksk.keywordCount) ) )
+ {
+ GNUNET_break (0);
+ goto cleanup;
+ }
uc->state = (enum UnindexState) state;
switch (state)
{
}
break;
case UNINDEX_STATE_DS_REMOVE:
+ case UNINDEX_STATE_EXTRACT_KEYWORDS:
+ case UNINDEX_STATE_DS_REMOVE_KBLOCKS:
break;
case UNINDEX_STATE_COMPLETE:
break;
case UNINDEX_STATE_DS_REMOVE:
GNUNET_FS_unindex_do_remove_ (uc);
break;
+ case UNINDEX_STATE_EXTRACT_KEYWORDS:
+ GNUNET_FS_unindex_do_extract_keywords_ (uc);
+ break;
+ case UNINDEX_STATE_DS_REMOVE_KBLOCKS:
+ GNUNET_FS_unindex_do_remove_kblocks_ (uc);
+ break;
case UNINDEX_STATE_COMPLETE:
case UNINDEX_STATE_ERROR:
/* no need to resume any operation, we were done */
GNUNET_break (0);
goto cleanup;
}
- sr->keyword_bitmap = GNUNET_malloc ((sr->uri->data.ksk.keywordCount + 7) / 8); /* round up, count bits */
- if (GNUNET_OK != GNUNET_BIO_read (rh, "keyword-bitmap",
- sr->keyword_bitmap,
- (sr->uri == NULL) ? 0 : (sr->uri->data.ksk.keywordCount + 7) / 8))
+ if (sr->sc->uri->type == ksk)
{
- GNUNET_break (0);
- goto cleanup;
+ 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->keyword_bitmap,
+ (sr->sc->uri->data.ksk.keywordCount + 7) / 8))
+ {
+ GNUNET_break (0);
+ goto cleanup;
+ }
}
GNUNET_free (uris);
if (download != NULL)
char *emsg;
struct GNUNET_BIO_ReadHandle *rh;
struct GNUNET_FS_SearchContext *sc;
+ struct stat buf;
+ if (0 != STAT (filename, &buf))
+ {
+ GNUNET_log_strerror_file (GNUNET_ERROR_TYPE_WARNING, "stat", filename);
+ return GNUNET_OK;
+ }
+ if (S_ISDIR (buf.st_mode))
+ return GNUNET_OK; /* skip directories */
ser = get_serialization_short_name (filename);
rh = GNUNET_BIO_read_open (filename);
if (rh == NULL)
ret->upcb = upcb;
ret->upcb_cls = upcb_cls;
ret->flags = flags;
- ret->max_parallel_downloads = 1;
- ret->max_parallel_requests = 1;
+ ret->max_parallel_downloads = DEFAULT_MAX_PARALLEL_DOWNLOADS;
+ ret->max_parallel_requests = DEFAULT_MAX_PARALLEL_REQUESTS;
ret->avg_block_latency = GNUNET_TIME_UNIT_MINUTES; /* conservative starting point */
va_start (ap, flags);
while (GNUNET_FS_OPTIONS_END != (opt = va_arg (ap, enum GNUNET_FS_OPTIONS)))
projects / oweals / gnunet.git / blobdiff