Add starpu_data_acquire_on_node

ChangeLog

@@ -24,6 +24,7 @@ New features:
   * Performance models measurements can now be provided explicitly by
     applications.
   * Capability to emit communication statistics when running MPI code
+  * Add starpu_data_acquire_on_node
   * New functionnality to wrapper starpu_insert_task to pass a array of
 	data_handles via the parameter STARPU_DATA_ARRAY
 

doc/chapters/basic-api.texi

@@ -489,7 +489,7 @@ be consistent with the access mode specified in the @var{mode} argument.
 access the piece of data anymore.  Note that implicit data
 dependencies are also enforced by @code{starpu_data_acquire}, i.e.
 @code{starpu_data_acquire} will wait for all tasks scheduled to work on
-the data, unless that they have not been disabled explictly by calling
+the data, unless they have been disabled explictly by calling
 @code{starpu_data_set_default_sequential_consistency_flag} or
 @code{starpu_data_set_sequential_consistency_flag}.
 @code{starpu_data_acquire} is a blocking call, so that it cannot be called from
@@ -512,6 +512,16 @@ be called from task callbacks. Upon successful completion, this function
 returns 0.
 @end deftypefun
 
+@deftypefun int starpu_data_acquire_on_node (starpu_data_handle_t @var{handle}, unsigned @var{node}, {enum starpu_access_mode} @var{mode})
+This is the same as @code{starpu_data_acquire}, except that the data will be
+available on the given memory node instead of main memory.
+@end deftypefun
+
+@deftypefun int starpu_data_acquire_on_node_cb (starpu_data_handle_t @var{handle}, unsigned @var{node}, {enum starpu_access_mode} @var{mode}, void (*@var{callback})(void *), void *@var{arg})
+This is the same as @code{starpu_data_acquire_cb}, except that the data will be
+available on the given memory node instead of main memory.
+@end deftypefun
+
 @defmac STARPU_DATA_ACQUIRE_CB (starpu_data_handle_t @var{handle}, {enum starpu_access_mode} @var{mode}, code)
 @code{STARPU_DATA_ACQUIRE_CB} is the same as @code{starpu_data_acquire_cb},
 except that the code to be executed in a callback is directly provided as a

include/starpu_data.h

@@ -1,6 +1,6 @@
 /* StarPU --- Runtime system for heterogeneous multicore architectures.
  *
- * Copyright (C) 2010-2011  Université de Bordeaux 1
+ * Copyright (C) 2010-2012  Université de Bordeaux 1
  * Copyright (C) 2010, 2011, 2012  Centre National de la Recherche Scientifique
  *
  * StarPU is free software; you can redistribute it and/or modify
@@ -64,7 +64,9 @@ void starpu_data_invalidate(starpu_data_handle_t handle);
 void starpu_data_advise_as_important(starpu_data_handle_t handle, unsigned is_important);
 
 int starpu_data_acquire(starpu_data_handle_t handle, enum starpu_access_mode mode);
+int starpu_data_acquire_on_node(starpu_data_handle_t handle, unsigned node, enum starpu_access_mode mode);
 int starpu_data_acquire_cb(starpu_data_handle_t handle, enum starpu_access_mode mode, void (*callback)(void *), void *arg);
+int starpu_data_acquire_on_node_cb(starpu_data_handle_t handle, unsigned node, enum starpu_access_mode mode, void (*callback)(void *), void *arg);
 #ifdef __GCC__
 #  define STARPU_DATA_ACQUIRE_CB(handle, mode, code) do \
 	{ \						\

src/datawizard/user_interactions.c

@@ -91,9 +91,9 @@ static void _starpu_data_acquire_continuation_non_blocking(void *arg)
 
 	STARPU_ASSERT(handle);
 
-	struct _starpu_data_replicate *ram_replicate = &handle->per_node[0];
+	struct _starpu_data_replicate *replicate = &handle->per_node[wrapper->node];
 
-	ret = _starpu_fetch_data_on_node(handle, ram_replicate, wrapper->mode, 0, 1,
+	ret = _starpu_fetch_data_on_node(handle, replicate, wrapper->mode, 0, 1,
 					 _starpu_data_acquire_fetch_data_callback, wrapper);
 	STARPU_ASSERT(!ret);
 }
@@ -114,7 +114,7 @@ static void starpu_data_acquire_cb_pre_sync_callback(void *arg)
 }
 
 /* The data must be released by calling starpu_data_release later on */
-int starpu_data_acquire_cb(starpu_data_handle_t handle,
+int starpu_data_acquire_on_node_cb(starpu_data_handle_t handle, unsigned node,
 			   enum starpu_access_mode mode, void (*callback)(void *), void *arg)
 {
 	STARPU_ASSERT(handle);
@@ -125,6 +125,7 @@ int starpu_data_acquire_cb(starpu_data_handle_t handle,
 	STARPU_ASSERT(wrapper);
 
 	wrapper->handle = handle;
+	wrapper->node = node;
 	wrapper->mode = mode;
 	wrapper->callback = callback;
 	wrapper->callback_arg = arg;
@@ -175,6 +176,12 @@ int starpu_data_acquire_cb(starpu_data_handle_t handle,
 	return 0;
 }
 
+int starpu_data_acquire_cb(starpu_data_handle_t handle,
+			   enum starpu_access_mode mode, void (*callback)(void *), void *arg)
+{
+	return starpu_data_acquire_on_node_cb(handle, 0, mode, callback, arg);
+}
+
 /*
  *	Block data request from application
  */
@@ -186,9 +193,9 @@ static inline void _starpu_data_acquire_continuation(void *arg)
 
 	STARPU_ASSERT(handle);
 
-	struct _starpu_data_replicate *ram_replicate = &handle->per_node[0];
+	struct _starpu_data_replicate *replicate = &handle->per_node[wrapper->node];
 
-	_starpu_fetch_data_on_node(handle, ram_replicate, wrapper->mode, 0, 0, NULL, NULL);
+	_starpu_fetch_data_on_node(handle, replicate, wrapper->mode, 0, 0, NULL, NULL);
 
 	/* continuation of starpu_data_acquire */
 	_STARPU_PTHREAD_MUTEX_LOCK(&wrapper->lock);
@@ -198,7 +205,7 @@ static inline void _starpu_data_acquire_continuation(void *arg)
 }
 
 /* The data must be released by calling starpu_data_release later on */
-int starpu_data_acquire(starpu_data_handle_t handle, enum starpu_access_mode mode)
+int starpu_data_acquire_on_node(starpu_data_handle_t handle, unsigned node, enum starpu_access_mode mode)
 {
 	STARPU_ASSERT(handle);
 	STARPU_ASSERT_MSG(handle->nchildren == 0, "Acquiring a partitioned data is not possible");
@@ -230,7 +237,7 @@ int starpu_data_acquire(starpu_data_handle_t handle, enum starpu_access_mode mod
 	{
 		.handle = handle,
 		.mode = mode,
-		.node = 0, // unused
+		.node = node,
 		.cond = PTHREAD_COND_INITIALIZER,
 		.lock = PTHREAD_MUTEX_INITIALIZER,
 		.finished = 0
@@ -278,8 +285,8 @@ int starpu_data_acquire(starpu_data_handle_t handle, enum starpu_access_mode mod
 	if (!_starpu_attempt_to_submit_data_request_from_apps(handle, mode, _starpu_data_acquire_continuation, &wrapper))
 	{
 		/* no one has locked this data yet, so we proceed immediately */
-		struct _starpu_data_replicate *ram_replicate = &handle->per_node[0];
-		int ret = _starpu_fetch_data_on_node(handle, ram_replicate, mode, 0, 0, NULL, NULL);
+		struct _starpu_data_replicate *replicate = &handle->per_node[node];
+		int ret = _starpu_fetch_data_on_node(handle, replicate, mode, 0, 0, NULL, NULL);
 		STARPU_ASSERT(!ret);
 	}
 	else
@@ -301,19 +308,29 @@ int starpu_data_acquire(starpu_data_handle_t handle, enum starpu_access_mode mod
 	return 0;
 }
 
+int starpu_data_acquire(starpu_data_handle_t handle, enum starpu_access_mode mode)
+{
+	return starpu_data_acquire_on_node(handle, 0, mode);
+}
+
 /* This function must be called after starpu_data_acquire so that the
  * application release the data */
-void starpu_data_release(starpu_data_handle_t handle)
+void starpu_data_release_on_node(starpu_data_handle_t handle, unsigned node)
 {
 	STARPU_ASSERT(handle);
 
 	/* The application can now release the rw-lock */
-	_starpu_release_data_on_node(handle, 0, &handle->per_node[0]);
+	_starpu_release_data_on_node(handle, 0, &handle->per_node[node]);
 
 	/* In case there are some implicit dependencies, unlock the "post sync" tasks */
 	_starpu_unlock_post_sync_tasks(handle);
 }
 
+void starpu_data_release(starpu_data_handle_t handle)
+{
+	starpu_data_release_on_node(handle, 0);
+}
+
 static void _prefetch_data_on_node(void *arg)
 {
 	struct user_interaction_wrapper *wrapper = (struct user_interaction_wrapper *) arg;