Make the CPU driver public.

include/starpu.h

@@ -68,6 +68,7 @@ struct starpu_driver
 	enum starpu_archtype type;
 	union
 	{
+		unsigned cpu_id;
 		unsigned cuda_id;
 #if defined(STARPU_USE_OPENCL) && !defined(__CUDACC__)
 		cl_device_id opencl_id;

src/core/workers.c

@@ -1000,6 +1000,11 @@ starpu_driver_run(struct starpu_driver *d)
 	}
 }
 
+#ifdef STARPU_USE_CPU
+extern int _starpu_cpu_driver_init(struct starpu_driver *);
+extern int _starpu_cpu_driver_run_once(struct starpu_driver *);
+extern int _starpu_cpu_driver_deinit(struct starpu_driver *);
+#endif
 #ifdef STARPU_USE_CUDA
 extern int _starpu_cuda_driver_init(struct starpu_driver *);
 extern int _starpu_cuda_driver_run_once(struct starpu_driver *);
@@ -1018,6 +1023,10 @@ starpu_driver_init(struct starpu_driver *d)
 
 	switch (d->type)
 	{
+#ifdef STARPU_USE_CPU
+	case STARPU_CPU_WORKER:
+		return _starpu_cpu_driver_init(d);
+#endif
 #ifdef STARPU_USE_CUDA
 	case STARPU_CUDA_WORKER:
 		return _starpu_cuda_driver_init(d);
@@ -1026,7 +1035,6 @@ starpu_driver_init(struct starpu_driver *d)
 	case STARPU_OPENCL_WORKER:
 		return _starpu_opencl_driver_init(d);
 #endif
-	case STARPU_CPU_WORKER:    /* Not supported yet */
 	case STARPU_GORDON_WORKER: /* Not supported yet */
 	default:
 		return -EINVAL;
@@ -1040,6 +1048,10 @@ starpu_driver_run_once(struct starpu_driver *d)
 
 	switch (d->type)
 	{
+#ifdef STARPU_USE_CPU
+	case STARPU_CPU_WORKER:
+		return _starpu_cpu_driver_run_once(d);
+#endif
 #ifdef STARPU_USE_CUDA
 	case STARPU_CUDA_WORKER:
 		return _starpu_cuda_driver_run_once(d);
@@ -1048,7 +1060,6 @@ starpu_driver_run_once(struct starpu_driver *d)
 	case STARPU_OPENCL_WORKER:
 		return _starpu_opencl_driver_run_once(d);
 #endif
-	case STARPU_CPU_WORKER:    /* Not supported yet */
 	case STARPU_GORDON_WORKER: /* Not supported yet */
 	default:
 		return -EINVAL;
@@ -1062,6 +1073,10 @@ starpu_driver_deinit(struct starpu_driver *d)
 
 	switch (d->type)
 	{
+#ifdef STARPU_USE_CPU
+	case STARPU_CPU_WORKER:
+		return _starpu_cpu_driver_deinit(d);
+#endif
 #ifdef STARPU_USE_CUDA
 	case STARPU_CUDA_WORKER:
 		return _starpu_cuda_driver_deinit(d);
@@ -1070,7 +1085,6 @@ starpu_driver_deinit(struct starpu_driver *d)
 	case STARPU_OPENCL_WORKER:
 		return _starpu_opencl_driver_deinit(d);
 #endif
-	case STARPU_CPU_WORKER:    /* Not supported yet */
 	case STARPU_GORDON_WORKER: /* Not supported yet */
 	default:
 		return -EINVAL;

src/drivers/cpu/driver_cpu.c

@@ -95,140 +95,184 @@ static int execute_job_on_cpu(struct _starpu_job *j, struct starpu_task *worker_
 	return 0;
 }
 
-void *_starpu_cpu_worker(void *arg)
+static struct _starpu_worker*
+_starpu_get_worker_from_driver(struct starpu_driver *d)
 {
-	struct _starpu_worker *cpu_arg = (struct _starpu_worker *) arg;
-	unsigned memnode = cpu_arg->memory_node;
-	int workerid = cpu_arg->workerid;
-	int devid = cpu_arg->devid;
+#if 1
+	int workers[d->id.cpu_id + 1];
+	int nworkers;
+	nworkers = starpu_worker_get_ids_by_type(STARPU_CPU_WORKER, workers, d->id.cpu_id+1);
+	if (nworkers >= 0 && (unsigned) nworkers < d->id.cpu_id)
+		return NULL; // No device was found.
+	
+	return _starpu_get_worker_struct(workers[d->id.cpu_id]);
+#else
+	int workers[STARPU_NMAXWORKERS];
+	int nworkers;
+	nworkers = starpu_worker_get_ids_by_type(STARPU_CPU_WORKER, workers, STARPU_NMAXWORKERS);
+	STARPU_ASSERT(nworkers > 0);
+	int i;
+	for (i = 0; i < nworkers; i++)
+	{
+		fprintf(stderr, "\tCPU %d\n", i);
+	}
+
+	return _starpu_get_worker_struct(workers[d->id.cpu_id]);
+#endif
+}
+
+int _starpu_cpu_driver_init(struct starpu_driver *d)
+{
+	struct _starpu_worker *cpu_worker;
+	cpu_worker = _starpu_get_worker_from_driver(d);
+	STARPU_ASSERT(cpu_worker);
+
+	unsigned memnode = cpu_worker->memory_node;
+	int devid = cpu_worker->devid;
 
 #ifdef STARPU_USE_FXT
-	_starpu_fxt_register_thread(cpu_arg->bindid);
+	_starpu_fxt_register_thread(cpu_worker->bindid);
 #endif
 	_STARPU_TRACE_WORKER_INIT_START(_STARPU_FUT_CPU_KEY, devid, memnode);
 
-	_starpu_bind_thread_on_cpu(cpu_arg->config, cpu_arg->bindid);
+	_starpu_bind_thread_on_cpu(cpu_worker->config, cpu_worker->bindid);
 
-        _STARPU_DEBUG("cpu worker %d is ready on logical cpu %d\n", devid, cpu_arg->bindid);
+        _STARPU_DEBUG("cpu worker %d is ready on logical cpu %d\n", devid, cpu_worker->bindid);
 
-	_starpu_set_local_memory_node_key(&memnode);
+	_starpu_set_local_memory_node_key(&cpu_worker->memory_node);
 
-	_starpu_set_local_worker_key(cpu_arg);
+	_starpu_set_local_worker_key(cpu_worker);
 
-	snprintf(cpu_arg->name, sizeof(cpu_arg->name), "CPU %d", devid);
-	snprintf(cpu_arg->short_name, sizeof(cpu_arg->short_name), "CPU %d", devid);
+	snprintf(cpu_worker->name, sizeof(cpu_worker->name), "CPU %d", devid);
+	snprintf(cpu_worker->short_name, sizeof(cpu_worker->short_name), "CPU %d", devid);
 
-	cpu_arg->status = STATUS_UNKNOWN;
+	cpu_worker->status = STATUS_UNKNOWN;
 
 	_STARPU_TRACE_WORKER_INIT_END
 
         /* tell the main thread that we are ready */
-	_STARPU_PTHREAD_MUTEX_LOCK(&cpu_arg->mutex);
-	cpu_arg->worker_is_initialized = 1;
-	_STARPU_PTHREAD_COND_SIGNAL(&cpu_arg->ready_cond);
-	_STARPU_PTHREAD_MUTEX_UNLOCK(&cpu_arg->mutex);
+	_STARPU_PTHREAD_MUTEX_LOCK(&cpu_worker->mutex);
+	cpu_worker->worker_is_initialized = 1;
+	_STARPU_PTHREAD_COND_SIGNAL(&cpu_worker->ready_cond);
+	_STARPU_PTHREAD_MUTEX_UNLOCK(&cpu_worker->mutex);
+	return 0;
+}
 
-        struct _starpu_job *j;
-	struct starpu_task *task;
+int _starpu_cpu_driver_run_once(struct starpu_driver *d)
+{
+	struct _starpu_worker *cpu_worker;
+	cpu_worker = _starpu_get_worker_from_driver(d);
+	STARPU_ASSERT(cpu_worker);
 
-	int res;
+	unsigned memnode = cpu_worker->memory_node;
+	int workerid = cpu_worker->workerid;
 
-	while (_starpu_machine_is_running())
-	{
-		_STARPU_TRACE_START_PROGRESS(memnode);
-		_starpu_datawizard_progress(memnode, 1);
-		_STARPU_TRACE_END_PROGRESS(memnode);
+	_STARPU_TRACE_START_PROGRESS(memnode);
+	_starpu_datawizard_progress(memnode, 1);
+	_STARPU_TRACE_END_PROGRESS(memnode);
 
-		_STARPU_PTHREAD_MUTEX_LOCK(cpu_arg->sched_mutex);
+	_STARPU_PTHREAD_MUTEX_LOCK(cpu_worker->sched_mutex);
 
-		task = _starpu_pop_task(cpu_arg);
+        struct _starpu_job *j;
+	struct starpu_task *task;
+	int res;
 
-                if (!task)
-		{
-			if (_starpu_worker_can_block(memnode))
-				_starpu_block_worker(workerid, cpu_arg->sched_cond, cpu_arg->sched_mutex);
+	task = _starpu_pop_task(cpu_worker);
 
-			_STARPU_PTHREAD_MUTEX_UNLOCK(cpu_arg->sched_mutex);
+	if (!task)
+	{
+		if (_starpu_worker_can_block(memnode))
+			_starpu_block_worker(workerid, cpu_worker->sched_cond, cpu_worker->sched_mutex);
 
-			continue;
-		};
+		_STARPU_PTHREAD_MUTEX_UNLOCK(cpu_worker->sched_mutex);
 
-		_STARPU_PTHREAD_MUTEX_UNLOCK(cpu_arg->sched_mutex);
+		return 0;
+	};
 
-		STARPU_ASSERT(task);
-		j = _starpu_get_job_associated_to_task(task);
+	_STARPU_PTHREAD_MUTEX_UNLOCK(cpu_worker->sched_mutex);
 
-		/* can a cpu perform that task ? */
-		if (!_STARPU_CPU_MAY_PERFORM(j))
-		{
-			/* put it and the end of the queue ... XXX */
-			_starpu_push_task(j);
-			continue;
-		}
+	STARPU_ASSERT(task);
+	j = _starpu_get_job_associated_to_task(task);
 
-		int rank = 0;
-		int is_parallel_task = (j->task_size > 1);
+	/* can a cpu perform that task ? */
+	if (!_STARPU_CPU_MAY_PERFORM(j))
+	{
+		/* put it and the end of the queue ... XXX */
+		_starpu_push_task(j);
+		return 0;
+	}
 
-		enum starpu_perf_archtype perf_arch;
+	int rank = 0;
+	int is_parallel_task = (j->task_size > 1);
 
-		/* Get the rank in case it is a parallel task */
-		if (is_parallel_task)
-		{
-			_STARPU_PTHREAD_MUTEX_LOCK(&j->sync_mutex);
-			rank = j->active_task_alias_count++;
-			_STARPU_PTHREAD_MUTEX_UNLOCK(&j->sync_mutex);
+	enum starpu_perf_archtype perf_arch;
+
+	/* Get the rank in case it is a parallel task */
+	if (is_parallel_task)
+	{
+		_STARPU_PTHREAD_MUTEX_LOCK(&j->sync_mutex);
+		rank = j->active_task_alias_count++;
+		_STARPU_PTHREAD_MUTEX_UNLOCK(&j->sync_mutex);
 
-			struct _starpu_combined_worker *combined_worker;
-			combined_worker = _starpu_get_combined_worker_struct(j->combined_workerid);
+		struct _starpu_combined_worker *combined_worker;
+		combined_worker = _starpu_get_combined_worker_struct(j->combined_workerid);
 
-			cpu_arg->combined_workerid = j->combined_workerid;
-			cpu_arg->worker_size = combined_worker->worker_size;
-			cpu_arg->current_rank = rank;
-			perf_arch = combined_worker->perf_arch;
-		}
-		else
-		{
-			cpu_arg->combined_workerid = cpu_arg->workerid;
-			cpu_arg->worker_size = 1;
-			cpu_arg->current_rank = 0;
-			perf_arch = cpu_arg->perf_arch;
-		}
+		cpu_worker->combined_workerid = j->combined_workerid;
+		cpu_worker->worker_size = combined_worker->worker_size;
+		cpu_worker->current_rank = rank;
+		perf_arch = combined_worker->perf_arch;
+	}
+	else
+	{
+		cpu_worker->combined_workerid = cpu_worker->workerid;
+		cpu_worker->worker_size = 1;
+		cpu_worker->current_rank = 0;
+		perf_arch = cpu_worker->perf_arch;
+	}
 
-		_starpu_set_current_task(j->task);
-		cpu_arg->current_task = j->task;
+	_starpu_set_current_task(j->task);
+	cpu_worker->current_task = j->task;
 
-                res = execute_job_on_cpu(j, task, cpu_arg, rank, perf_arch);
+	res = execute_job_on_cpu(j, task, cpu_worker, rank, perf_arch);
 
-		_starpu_set_current_task(NULL);
-		cpu_arg->current_task = NULL;
+	_starpu_set_current_task(NULL);
+	cpu_worker->current_task = NULL;
 
-		if (res)
+	if (res)
+	{
+		switch (res)
 		{
-			switch (res)
-			{
-				case -EAGAIN:
-					_starpu_push_task(j);
-					continue;
-				default:
-					STARPU_ASSERT(0);
-			}
+		case -EAGAIN:
+			_starpu_push_task(j);
+			return 0;
+		default:
+			STARPU_ABORT();
 		}
+	}
 
-		/* In the case of combined workers, we need to inform the
-		 * scheduler each worker's execution is over.
-		 * Then we free the workers' task alias */
-		if (is_parallel_task)
-		{
-			_starpu_sched_post_exec_hook(task);
-			free(task);
-		}
+	/* In the case of combined workers, we need to inform the
+	 * scheduler each worker's execution is over.
+	 * Then we free the workers' task alias */
+	if (is_parallel_task)
+	{
+		_starpu_sched_post_exec_hook(task);
+		free(task);
+	}
 
-		if (rank == 0)
-			_starpu_handle_job_termination(j);
-        }
+	if (rank == 0)
+		_starpu_handle_job_termination(j);
+	return 0;
+}
 
+int _starpu_cpu_driver_deinit(struct starpu_driver *d)
+{
 	_STARPU_TRACE_WORKER_DEINIT_START
 
+	struct _starpu_worker *cpu_worker;
+	cpu_worker = _starpu_get_worker_from_driver(d);
+	STARPU_ASSERT(cpu_worker);
+
+	unsigned memnode = cpu_worker->memory_node;
 	_starpu_handle_all_pending_node_data_requests(memnode);
 
 	/* In case there remains some memory that was automatically
@@ -238,5 +282,23 @@ void *_starpu_cpu_worker(void *arg)
 
 	_STARPU_TRACE_WORKER_DEINIT_END(_STARPU_FUT_CPU_KEY);
 
+	return 0;
+}
+
+void *
+_starpu_cpu_worker(void *arg)
+{
+	struct _starpu_worker *args = arg;
+	struct starpu_driver d =
+	{
+		.type      = STARPU_CPU_WORKER,
+		.id.cpu_id = args->devid
+	};
+
+	_starpu_cpu_driver_init(&d);
+	while (_starpu_machine_is_running())
+		_starpu_cpu_driver_run_once(&d);
+	_starpu_cpu_driver_deinit(&d);
+
 	return NULL;
 }