Fix and improve efficiency of computing termination times

src/sched_policies/component_fifo.c

@@ -48,10 +48,7 @@ static double fifo_estimated_end(struct starpu_sched_component * component)
 	int card = starpu_bitmap_cardinal(component->workers_in_ctx);
 	STARPU_ASSERT(card != 0);
 	double estimated_end = starpu_sched_component_estimated_end_min(component);
-	STARPU_PTHREAD_MUTEX_LOCK(mutex);
-	fifo->exp_start = STARPU_MAX(fifo->exp_start, starpu_timing_now());
-	estimated_end += fifo->exp_start + fifo->exp_len / card;
-	STARPU_PTHREAD_MUTEX_UNLOCK(mutex);
+	estimated_end += fifo->exp_len / card;
 
 	return estimated_end;
 }

src/sched_policies/component_mct.c

@@ -94,6 +94,9 @@ static int mct_push_task(struct starpu_sched_component * component, struct starp
 
 	best_component = component->children[best_icomponent];
 
+	task->predicted = estimated_lengths[best_icomponent];
+	task->predicted_transfer = estimated_transfer_length[best_icomponent];
+
 	if(starpu_sched_component_is_worker(best_component))
 	{
 		best_component->can_pull(best_component);

src/sched_policies/component_prio.c

@@ -66,10 +66,7 @@ static double prio_estimated_end(struct starpu_sched_component * component)
 	int card = starpu_bitmap_cardinal(component->workers_in_ctx);
 	STARPU_ASSERT(card != 0);
 	double estimated_end = starpu_sched_component_estimated_end_min(component);
-	STARPU_PTHREAD_MUTEX_LOCK(mutex);
-	prio->exp_start = STARPU_MAX(prio->exp_start, starpu_timing_now());
-	estimated_end += prio->exp_start + prio->exp_len / card;
-	STARPU_PTHREAD_MUTEX_UNLOCK(mutex);
+	estimated_end += prio->exp_len / card;
 
 	return estimated_end;
 }

src/sched_policies/component_worker.c

@@ -111,7 +111,6 @@ static void _starpu_worker_task_list_transfer_started(struct _starpu_worker_task
 	l->pipeline_len += transfer_model;
 	l->exp_start = starpu_timing_now() + l->pipeline_len;
 	l->exp_end = l->exp_start + l->exp_len;
-
 }
 
 #ifdef STARPU_DEVEL
@@ -142,6 +141,8 @@ static void _starpu_worker_task_list_finished(struct _starpu_worker_task_list *l
 	if(!isnan(task->predicted))
 		/* The execution is over, remove it from pipelined */
 		l->pipeline_len -= task->predicted;
+	l->exp_start = STARPU_MAX(starpu_timing_now() + l->pipeline_len, l->exp_start);
+	l->exp_end = l->exp_start + l->exp_len;
 }
 
 
@@ -184,6 +185,9 @@ static struct _starpu_worker_task_list * _starpu_worker_task_list_create(void)
 	memset(l, 0, sizeof(*l));
 	l->exp_len = l->pipeline_len = 0.0;
 	l->exp_start = l->exp_end = starpu_timing_now();
+	/* These are only for statistics */
+	STARPU_HG_DISABLE_CHECKING(l->exp_end);
+	STARPU_HG_DISABLE_CHECKING(l->exp_start);
 	STARPU_PTHREAD_MUTEX_INIT(&l->mutex,NULL);
 	return l;
 }
@@ -222,12 +226,12 @@ static inline void _starpu_worker_task_list_add(struct _starpu_worker_task_list
 {
 	double predicted = task->predicted;
 	double predicted_transfer = task->predicted_transfer;
+	double end = l->exp_end;
 
 	/* Sometimes workers didn't take the tasks as early as we expected */
 	l->exp_start = STARPU_MAX(l->exp_start, starpu_timing_now());
-	l->exp_end = l->exp_start + l->exp_len;
 
-	if (starpu_timing_now() + predicted_transfer < l->exp_end)
+	if (starpu_timing_now() + predicted_transfer < end)
 	{
 		/* We may hope that the transfer will be finished by
 		 * the start of the task. */
@@ -237,20 +241,16 @@ static inline void _starpu_worker_task_list_add(struct _starpu_worker_task_list
 	{
 		/* The transfer will not be finished by then, take the
 		 * remainder into account */
-		predicted_transfer = (starpu_timing_now() + predicted_transfer) - l->exp_end;
+		predicted_transfer = (starpu_timing_now() + predicted_transfer) - end;
 	}
 
 	if(!isnan(predicted_transfer))
-	{
-		l->exp_end += predicted_transfer;
 		l->exp_len += predicted_transfer;
-	}
 
 	if(!isnan(predicted))
-	{
-		l->exp_end += predicted;
 		l->exp_len += predicted;
-	}
+
+	l->exp_end = l->exp_start + l->exp_len;
 
 	task->predicted = predicted;
 	task->predicted_transfer = predicted_transfer;
@@ -300,8 +300,8 @@ static inline struct starpu_task * _starpu_worker_task_list_pop(struct _starpu_w
 {
  	if(!l->first)
 	{
+		l->exp_len = l->pipeline_len = 0.0;
 		l->exp_start = l->exp_end = starpu_timing_now();
-		l->exp_len = 0;
 		return NULL;
 	}
 	struct _starpu_task_grid * t = l->first;
@@ -522,9 +522,6 @@ static int simple_worker_push_task(struct starpu_sched_component * component, st
 #endif
 	struct _starpu_worker_task_list * list = data->list;
 	STARPU_PTHREAD_MUTEX_LOCK(&list->mutex);
-        /* Sometimes workers didn't take the tasks as early as we expected */
-	list->exp_start = isnan(list->exp_start) ? starpu_timing_now() + list->pipeline_len : STARPU_MAX(list->exp_start, starpu_timing_now());
-	list->exp_end = list->exp_start + list->exp_len;
 	_starpu_worker_task_list_push(list, t);
 	STARPU_PTHREAD_MUTEX_UNLOCK(&list->mutex);
 	simple_worker_can_pull(component);
@@ -539,14 +536,16 @@ static struct starpu_task * simple_worker_pull_task(struct starpu_sched_componen
 	STARPU_PTHREAD_MUTEX_LOCK(&list->mutex);
 	/* Take the opportunity to update start time */
 	data->list->exp_start = STARPU_MAX(starpu_timing_now(), data->list->exp_start);
+	data->list->exp_end = data->list->exp_start + data->list->exp_len;
 	struct starpu_task * task =  _starpu_worker_task_list_pop(list);
-	STARPU_PTHREAD_MUTEX_UNLOCK(&list->mutex);
 	if(task)
 	{
 		_starpu_worker_task_list_transfer_started(list, task);
+		STARPU_PTHREAD_MUTEX_UNLOCK(&list->mutex);
 		starpu_push_task_end(task);
 		return task;
 	}
+	STARPU_PTHREAD_MUTEX_UNLOCK(&list->mutex);
 	_starpu_sched_component_lock_worker(component->tree->sched_ctx_id, workerid);
 	int i;
 	do
@@ -575,8 +574,10 @@ static struct starpu_task * simple_worker_pull_task(struct starpu_sched_componen
 	{
 		if(!starpu_worker_is_combined_worker(workerid))
 		{
+			STARPU_PTHREAD_MUTEX_LOCK(&list->mutex);
 			_starpu_worker_task_list_add(list, task);
 			_starpu_worker_task_list_transfer_started(list, task);
+			STARPU_PTHREAD_MUTEX_UNLOCK(&list->mutex);
 			starpu_push_task_end(task);
 			return task;
 		}
@@ -588,8 +589,10 @@ static struct starpu_task * simple_worker_pull_task(struct starpu_sched_componen
 	}
 	if(task)
 	{
+		STARPU_PTHREAD_MUTEX_LOCK(&list->mutex);
 		_starpu_worker_task_list_add(list, task);
 		_starpu_worker_task_list_transfer_started(list, task);
+		STARPU_PTHREAD_MUTEX_UNLOCK(&list->mutex);
 		starpu_push_task_end(task);
 	}
 	return task;
@@ -598,10 +601,13 @@ static struct starpu_task * simple_worker_pull_task(struct starpu_sched_componen
 static double simple_worker_estimated_end(struct starpu_sched_component * component)
 {
 	struct _starpu_worker_component_data * data = component->data;
-	STARPU_PTHREAD_MUTEX_LOCK(&data->list->mutex);
-	double tmp = data->list->exp_end = data->list->exp_start + data->list->exp_len;
-	STARPU_PTHREAD_MUTEX_UNLOCK(&data->list->mutex);
-	return tmp;
+	double now = starpu_timing_now();
+	if (now > data->list->exp_start)
+	{
+		data->list->exp_start = now;
+		data->list->exp_end = now + data->list->exp_len;
+	}
+	return data->list->exp_end;
 }
 
 static double simple_worker_estimated_load(struct starpu_sched_component * component)
@@ -805,9 +811,7 @@ static double combined_worker_estimated_end(struct starpu_sched_component * comp
 	for(i = 0; i < combined_worker->worker_size; i++)
 	{
 		data = _worker_components[component->tree->sched_ctx_id][combined_worker->combined_workerid[i]]->data;
-		STARPU_PTHREAD_MUTEX_LOCK(&data->list->mutex);
 		double tmp = data->list->exp_end;
-		STARPU_PTHREAD_MUTEX_UNLOCK(&data->list->mutex);
 		max = tmp > max ? tmp : max;
 	}
 	return max;
@@ -921,8 +925,6 @@ void starpu_sched_component_worker_post_exec_hook(struct starpu_task * task, uns
 	struct _starpu_worker_task_list * list = _worker_get_list(sched_ctx_id);
 	STARPU_PTHREAD_MUTEX_LOCK(&list->mutex);
 	_starpu_worker_task_list_finished(list, task);
-	list->exp_start = STARPU_MAX(starpu_timing_now() + list->pipeline_len, list->exp_start);
-	list->exp_end = list->exp_start + list->exp_len;
 	STARPU_PTHREAD_MUTEX_UNLOCK(&list->mutex);
 }
 

src/sched_policies/deque_modeling_policy_data_aware.c

@@ -177,6 +177,8 @@ static void _starpu_fifo_task_finished(struct _starpu_fifo_taskq *fifo, struct s
 	if(!isnan(task->predicted))
 		/* The execution is over, remove it from pipelined */
 		fifo->pipeline_len -= task->predicted;
+	fifo->exp_start = STARPU_MAX(starpu_timing_now() + fifo->pipeline_len, fifo->exp_start);
+	fifo->exp_end = fifo->exp_start + fifo->exp_len;
 }
 
 
@@ -365,7 +367,6 @@ static int push_task_on_best_worker(struct starpu_task *task, int best_workerid,
 
         /* Sometimes workers didn't take the tasks as early as we expected */
 	fifo->exp_start = isnan(fifo->exp_start) ? starpu_timing_now() + fifo->pipeline_len : STARPU_MAX(fifo->exp_start, starpu_timing_now());
-	fifo->exp_end = fifo->exp_start + fifo->exp_len;
 
 	if ((starpu_timing_now() + predicted_transfer) < fifo->exp_end)
 	{
@@ -382,7 +383,6 @@ static int push_task_on_best_worker(struct starpu_task *task, int best_workerid,
 
 	if(!isnan(predicted_transfer))
 	{
-		fifo->exp_end += predicted_transfer;
 		fifo->exp_len += predicted_transfer;
 		if(dt->num_priorities != -1)
 		{
@@ -396,7 +396,6 @@ static int push_task_on_best_worker(struct starpu_task *task, int best_workerid,
 
 	if(!isnan(predicted))
 	{
-		fifo->exp_end += predicted;
 		fifo->exp_len += predicted;
 		if(dt->num_priorities != -1)
 		{
@@ -407,6 +406,7 @@ static int push_task_on_best_worker(struct starpu_task *task, int best_workerid,
 		}
 
 	}
+	fifo->exp_end = fifo->exp_start + fifo->exp_len;
 
 	STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(sched_mutex);
 
@@ -1209,8 +1209,6 @@ static void dmda_post_exec_hook(struct starpu_task * task, unsigned sched_ctx_id
 	starpu_worker_get_sched_condition(workerid, &sched_mutex, &sched_cond);
 	STARPU_PTHREAD_MUTEX_LOCK_SCHED(sched_mutex);
 	_starpu_fifo_task_finished(fifo, task, dt->num_priorities);
-	fifo->exp_start = STARPU_MAX(starpu_timing_now() + fifo->pipeline_len, fifo->exp_start);
-	fifo->exp_end = fifo->exp_start + fifo->exp_len;
 	STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(sched_mutex);
 }