Add starpu_idle_prefetch_task_input_on_node, and call it at task submission for tasks which execute on a specific node

ChangeLog

@@ -83,7 +83,8 @@ New features:
   * New performance model format to better represent parallel tasks.
     Used to provide estimations for the execution times of the
     parallel tasks on scheduling contexts or combined workers.
-  * starpu_data_idle_prefetch_on_node allows to queue prefetches to be done
+  * starpu_data_idle_prefetch_on_node and
+    starpu_idle_prefetch_task_input_on_node allow to queue prefetches to be done
     only when the bus is idle.
 
 Small features:

doc/doxygen/chapters/api/scheduling_policy.doxy

@@ -187,6 +187,10 @@ Whether \ref STARPU_PREFETCH was set
 \ingroup API_Scheduling_Policy
 Prefetch data for a given task on a given node
 
+\fn int starpu_idle_prefetch_task_input_on_node(struct starpu_task *task, unsigned node)
+\ingroup API_Scheduling_Policy
+Prefetch data for a given task on a given node when the bus is idle
+
 \fn void starpu_sched_ctx_worker_shares_tasks_lists(int workerid, int sched_ctx_id)
 \ingroup API_Scheduling_Policy
 The scheduling policies indicates if the worker may pop tasks from the list of other workers

include/starpu_scheduler.h

@@ -1,6 +1,6 @@
 /* StarPU --- Runtime system for heterogeneous multicore architectures.
  *
- * Copyright (C) 2010-2014  Université de Bordeaux
+ * Copyright (C) 2010-2015  Université de Bordeaux
  * Copyright (C) 2011  Télécom-SudParis
  *
  * StarPU is free software; you can redistribute it and/or modify
@@ -70,6 +70,7 @@ int starpu_combined_worker_can_execute_task(unsigned workerid, struct starpu_tas
 
 int starpu_get_prefetch_flag(void);
 int starpu_prefetch_task_input_on_node(struct starpu_task *task, unsigned node);
+int starpu_idle_prefetch_task_input_on_node(struct starpu_task *task, unsigned node);
 
 uint32_t starpu_task_footprint(struct starpu_perfmodel *model, struct starpu_task *task, struct starpu_perfmodel_arch *arch, unsigned nimpl);
 uint32_t starpu_task_data_footprint(struct starpu_task *task);

src/core/task.c

@@ -590,6 +590,11 @@ int starpu_task_submit(struct starpu_task *task)
 			return -ENODEV;
 		}
 
+		if (task->execute_on_a_specific_worker)
+			/* We already know where it will go, so already push
+			 * idle requests to get the data there */
+			starpu_idle_prefetch_task_input_on_node(task, starpu_worker_get_memory_node(task->workerid));
+
 		/* If this is a continuation, we don't modify the implicit data dependencies detected earlier. */
 		if (!continuation)
 		{

src/datawizard/coherency.c

@@ -661,6 +661,11 @@ int _starpu_fetch_data_on_node(starpu_data_handle_t handle, struct _starpu_data_
         return ret;
 }
 
+static int idle_prefetch_data_on_node(starpu_data_handle_t handle, struct _starpu_data_replicate *replicate, enum starpu_data_access_mode mode)
+{
+	return _starpu_fetch_data_on_node(handle, replicate, mode, 1, 2, 1, NULL, NULL);
+}
+
 static int prefetch_data_on_node(starpu_data_handle_t handle, struct _starpu_data_replicate *replicate, enum starpu_data_access_mode mode)
 {
 	return _starpu_fetch_data_on_node(handle, replicate, mode, 1, 1, 1, NULL, NULL);
@@ -769,6 +774,27 @@ int starpu_prefetch_task_input_on_node(struct starpu_task *task, unsigned node)
 	return 0;
 }
 
+int starpu_idle_prefetch_task_input_on_node(struct starpu_task *task, unsigned node)
+{
+	unsigned nbuffers = STARPU_TASK_GET_NBUFFERS(task);
+	unsigned index;
+
+	for (index = 0; index < nbuffers; index++)
+	{
+		starpu_data_handle_t handle = STARPU_TASK_GET_HANDLE(task, index);
+		enum starpu_data_access_mode mode = STARPU_TASK_GET_MODE(task, index);
+
+		if (mode & (STARPU_SCRATCH|STARPU_REDUX))
+			continue;
+
+		struct _starpu_data_replicate *replicate = &handle->per_node[node];
+		idle_prefetch_data_on_node(handle, replicate, mode);
+	}
+
+	return 0;
+}
+
+
 static struct _starpu_data_replicate *get_replicate(starpu_data_handle_t handle, enum starpu_data_access_mode mode, int workerid, unsigned node)
 {
 	if (mode & (STARPU_SCRATCH|STARPU_REDUX))