/*
This file is part of GNUnet.
- (C) 2001--2012 Christian Grothoff (and other contributing authors)
+ Copyright (C) 2001--2012 GNUnet e.V.
GNUnet is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published
You should have received a copy of the GNU General Public License
along with GNUnet; see the file COPYING. If not, write to the
- Free Software Foundation, Inc., 59 Temple Place - Suite 330,
- Boston, MA 02111-1307, USA.
+ Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
+ Boston, MA 02110-1301, USA.
*/
/**
static void
start_job (struct GNUNET_FS_QueueEntry *qe)
{
- GNUNET_assert (NULL == qe->client);
- qe->client = GNUNET_CLIENT_connect ("fs", qe->h->cfg);
- if (NULL == qe->client)
- {
- GNUNET_break (0);
- return;
- }
- qe->start (qe->cls, qe->client);
+ qe->active = GNUNET_YES;
+ qe->start (qe->cls);
qe->start_times++;
qe->h->active_blocks += qe->blocks;
qe->h->active_downloads++;
static void
stop_job (struct GNUNET_FS_QueueEntry *qe)
{
- qe->client = NULL;
+ qe->active = GNUNET_NO;
qe->stop (qe->cls);
GNUNET_assert (0 < qe->h->active_downloads);
qe->h->active_downloads--;
"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);
+ 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);
}
* and stopping others.
*
* @param cls the `struct GNUNET_FS_Handle *`
- * @param tc scheduler context
*/
static void
-process_job_queue (void *cls,
- const struct GNUNET_SCHEDULER_TaskContext *tc)
+process_job_queue (void *cls)
{
struct GNUNET_FS_Handle *h = cls;
struct GNUNET_FS_QueueEntry *qe;
int num_downloads_change;
int block_limit_hit;
- h->queue_job = GNUNET_SCHEDULER_NO_TASK;
+ h->queue_job = NULL;
/* restart_at will be set to the time when it makes sense to
re-evaluate the job queue (unless, of course, jobs complete
or are added, then we'll be triggered immediately */
break;
case GNUNET_FS_QUEUE_PRIORITY_NORMAL:
run_time =
- GNUNET_TIME_relative_multiply (h->avg_block_latency,
- qe->blocks * qe->start_times);
+ GNUNET_TIME_relative_saturating_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);
if (0 == rst.rel_value_us)
break;
}
}
+ GNUNET_break (h->active_downloads ==
+ num_downloads_active + num_probes_active);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"PA: %u, PE: %u, PW: %u; DA: %u, DE: %u, DW: %u\n",
num_probes_active,
num_downloads_active,
num_downloads_expired,
num_downloads_waiting);
+ GNUNET_break (h->active_downloads + num_probes_active <=
+ h->max_parallel_downloads);
/* calculate start/stop decisions */
if (h->active_downloads + num_downloads_waiting > h->max_parallel_downloads)
{
/* start all downloads (we can) */
num_downloads_change = num_downloads_waiting;
/* also start probes if there is room, but use a lower cap of (mpd/4) + 1 */
- if (h->max_parallel_downloads / 2 >= (h->active_downloads + num_downloads_change))
+ if (1 + h->max_parallel_downloads / 4 >= (h->active_downloads + num_downloads_change))
num_probes_change = GNUNET_MIN (num_probes_waiting,
(1 + h->max_parallel_downloads / 4) - (h->active_downloads + num_downloads_change));
else
num_probes_change = 0;
}
-
+ GNUNET_break (num_downloads_change <= num_downloads_waiting);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Changing %d probes and %d downloads\n",
+ "Changing %d probes and %d/%u/%u downloads\n",
num_probes_change,
- num_downloads_change);
+ num_downloads_change,
+ (unsigned int) h->active_downloads,
+ (unsigned int) h->max_parallel_downloads);
/* actually stop probes */
next = h->running_head;
while (NULL != (qe = next))
break;
}
}
- GNUNET_break ( (0 == num_downloads_change) || (GNUNET_YES == block_limit_hit) );
+ GNUNET_break ( (0 == num_downloads_change) ||
+ (GNUNET_YES == block_limit_hit) );
GNUNET_break (0 == num_probes_change);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
num_downloads_change,
GNUNET_STRINGS_relative_time_to_string (restart_at, GNUNET_YES));
- /* make sure we run again */
+ /* make sure we run again, callbacks might have
+ already re-scheduled the job, so cancel such
+ an operation (if it exists) */
+ if (NULL != h->queue_job)
+ GNUNET_SCHEDULER_cancel (h->queue_job);
h->queue_job =
GNUNET_SCHEDULER_add_delayed (restart_at, &process_job_queue, h);
}
*/
struct GNUNET_FS_QueueEntry *
GNUNET_FS_queue_ (struct GNUNET_FS_Handle *h,
- GNUNET_FS_QueueStart start,
- GNUNET_FS_QueueStop stop, void *cls,
+ GNUNET_SCHEDULER_TaskCallback start,
+ GNUNET_SCHEDULER_TaskCallback stop,
+ void *cls,
unsigned int blocks,
enum GNUNET_FS_QueuePriority priority)
{
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)
+ if (NULL != h->queue_job)
GNUNET_SCHEDULER_cancel (h->queue_job);
h->queue_job = GNUNET_SCHEDULER_add_now (&process_job_queue, h);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Dequeueing job %p\n",
qe);
- if (NULL != qe->client)
+ if (GNUNET_YES == qe->active)
stop_job (qe);
- GNUNET_CONTAINER_DLL_remove (h->pending_head, h->pending_tail, qe);
+ GNUNET_CONTAINER_DLL_remove (h->pending_head,
+ h->pending_tail,
+ qe);
GNUNET_free (qe);
- if (h->queue_job != GNUNET_SCHEDULER_NO_TASK)
+ if (NULL != h->queue_job)
GNUNET_SCHEDULER_cancel (h->queue_job);
- h->queue_job = GNUNET_SCHEDULER_add_now (&process_job_queue, h);
+ h->queue_job = GNUNET_SCHEDULER_add_now (&process_job_queue,
+ h);
}
ret = GNUNET_new (struct TopLevelActivity);
ret->ssf = ssf;
ret->ssf_cls = ssf_cls;
- GNUNET_CONTAINER_DLL_insert (h->top_head, h->top_tail, ret);
+ GNUNET_CONTAINER_DLL_insert (h->top_head,
+ h->top_tail,
+ ret);
return ret;
}
GNUNET_FS_end_top (struct GNUNET_FS_Handle *h,
struct TopLevelActivity *top)
{
- GNUNET_CONTAINER_DLL_remove (h->top_head, h->top_tail, top);
+ GNUNET_CONTAINER_DLL_remove (h->top_head,
+ h->top_tail,
+ top);
GNUNET_free (top);
}
GNUNET_free_non_null (data);
return 0;
}
- memcpy (buf, &data[offset], max);
+ GNUNET_memcpy (buf, &data[offset], max);
return max;
}
/* re-start publishing (if needed)... */
if (GNUNET_YES != pc->all_done)
{
- GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == pc->upload_task);
+ GNUNET_assert (NULL == pc->upload_task);
pc->upload_task =
GNUNET_SCHEDULER_add_with_priority
(GNUNET_SCHEDULER_PRIORITY_BACKGROUND,
filename, emsg);
GNUNET_free (emsg);
}
- pc->top = GNUNET_FS_make_top (h, &GNUNET_FS_publish_signal_suspend_, pc);
+ pc->top = GNUNET_FS_make_top (h,
+ &GNUNET_FS_publish_signal_suspend_,
+ pc);
return GNUNET_OK;
cleanup:
GNUNET_free_non_null (pc->nid);
GNUNET_assert ((GNUNET_YES == GNUNET_FS_uri_test_ksk (sc->uri)) ||
(GNUNET_YES == GNUNET_FS_uri_test_sks (sc->uri)));
uris = GNUNET_FS_uri_to_string (sc->uri);
- in_pause = (sc->task != GNUNET_SCHEDULER_NO_TASK) ? 'r' : '\0';
+ in_pause = (sc->task != NULL) ? 'r' : '\0';
if ((GNUNET_OK != GNUNET_BIO_write_string (wh, uris)) ||
(GNUNET_OK != write_start_time (wh, sc->start_time)) ||
(GNUNET_OK != GNUNET_BIO_write_string (wh, sc->emsg)) ||
GNUNET_break (0);
goto cleanup;
}
- uc->top = GNUNET_FS_make_top (h, &GNUNET_FS_unindex_signal_suspend_, uc);
+ uc->top = GNUNET_FS_make_top (h,
+ &GNUNET_FS_unindex_signal_suspend_,
+ uc);
pi.status = GNUNET_FS_STATUS_UNINDEX_RESUME;
pi.value.unindex.specifics.resume.message = uc->emsg;
GNUNET_FS_unindex_make_status_ (&pi, uc,
signal_download_resume (dcc);
dcc = dcc->next;
}
- if (NULL != dc->pending_head)
- GNUNET_FS_download_start_downloading_ (dc);
}
GNUNET_FS_compute_depth (GNUNET_FS_uri_chk_get_file_size (dc->uri));
if (GNUNET_FS_uri_test_loc (dc->uri))
GNUNET_assert (GNUNET_OK ==
- GNUNET_FS_uri_loc_get_peer_identity (dc->uri, &dc->target));
+ GNUNET_FS_uri_loc_get_peer_identity (dc->uri,
+ &dc->target));
if (NULL == dc->emsg)
{
dc->top_request = read_download_request (rh);
goto cleanup;
}
}
- dn = get_download_sync_filename (dc, dc->serialization, ".dir");
+ dn = get_download_sync_filename (dc,
+ dc->serialization,
+ ".dir");
if (NULL != dn)
{
- if (GNUNET_YES == GNUNET_DISK_directory_test (dn, GNUNET_YES))
- GNUNET_DISK_directory_scan (dn, &deserialize_subdownload, dc);
+ 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_CONTAINER_DLL_insert (parent->child_head, parent->child_tail, dc);
+ GNUNET_CONTAINER_DLL_insert (parent->child_head,
+ parent->child_tail,
+ dc);
}
if (NULL != search)
{
dc->search = search;
search->download = dc;
}
- if ((NULL == parent) && (NULL == search))
+ if ( (NULL == parent) &&
+ (NULL == search) )
{
- dc->top =
- GNUNET_FS_make_top (dc->h, &GNUNET_FS_download_signal_suspend_, dc);
+ dc->top
+ = GNUNET_FS_make_top (dc->h,
+ &GNUNET_FS_download_signal_suspend_,
+ dc);
signal_download_resume (dc);
}
GNUNET_free (uris);
- dc->task = GNUNET_SCHEDULER_add_now (&GNUNET_FS_download_start_task_, dc);
+ GNUNET_assert (NULL == dc->job_queue);
+ dc->task = GNUNET_SCHEDULER_add_now (&GNUNET_FS_download_start_task_,
+ dc);
return;
cleanup:
GNUNET_free_non_null (uris);
pi.status = GNUNET_FS_STATUS_SEARCH_RESUME;
pi.value.search.specifics.resume.message = sc->emsg;
pi.value.search.specifics.resume.is_paused =
- (NULL == sc->client) ? GNUNET_YES : GNUNET_NO;
+ (NULL == sc->mq) ? GNUNET_YES : GNUNET_NO;
sc->client_info = GNUNET_FS_search_make_status_ (&pi, sc->h, sc);
GNUNET_CONTAINER_multihashmap_iterate (sc->master_result_map,
&signal_result_resume, sc);
* Deserialize informatin about pending operations.
*
* @param master_path which master directory should be scanned
- * @param proc function to call for each entry (will get 'h' for 'cls')
- * @param h the 'struct GNUNET_FS_Handle*'
+ * @param proc function to call for each entry (will get @a h for 'cls')
+ * @param h the `struct GNUNET_FS_Handle *`
*/
static void
-deserialization_master (const char *master_path, GNUNET_FileNameCallback proc,
+deserialization_master (const char *master_path,
+ GNUNET_FileNameCallback proc,
struct GNUNET_FS_Handle *h)
{
char *dn;
/**
* Close our connection with the file-sharing service.
- * The callback given to GNUNET_FS_start will no longer be
+ * The callback given to #GNUNET_FS_start() will no longer be
* called after this function returns.
+ * This function MUST NOT be called from within the
+ * callback itself.
*
* @param h handle that was returned from #GNUNET_FS_start()
*/
void
GNUNET_FS_stop (struct GNUNET_FS_Handle *h)
{
- while (h->top_head != NULL)
+ while (NULL != h->top_head)
h->top_head->ssf (h->top_head->ssf_cls);
- if (h->queue_job != GNUNET_SCHEDULER_NO_TASK)
+ if (NULL != h->queue_job)
GNUNET_SCHEDULER_cancel (h->queue_job);
GNUNET_free (h->client_name);
GNUNET_free (h);