Cleanup data_concurrency.c to properly support SCRATCH and REDUX access types.

src/core/dependencies/data_concurrency.c

@@ -20,22 +20,14 @@
 #include <common/starpu_spinlock.h>
 #include <datawizard/sort_data_handles.h>
 
-static unsigned _submit_job_enforce_data_deps(starpu_job_t j, unsigned start_buffer_index);
-
-static unsigned unlock_one_requester(starpu_data_requester_t r)
-{
-	starpu_job_t j = r->j;
-	unsigned nbuffers = j->task->cl->nbuffers;
-	unsigned buffer_index = r->buffer_index;
-
-	if (buffer_index + 1 < nbuffers)
-	{
-		/* not all buffers are protected yet */
-		return _submit_job_enforce_data_deps(j, buffer_index + 1);
-	}
-	else
-		return 0;
-}
+/*
+ * The different types of data accesses are STARPU_R, STARPU_RW, STARPU_W,
+ * STARPU_SCRATCH and STARPU_REDUX. STARPU_RW is managed as a STARPU_W access.
+ * - A single STARPU_W access is allowed at a time.
+ * - Concurrent STARPU_R accesses are allowed.
+ * - Concurrent STARPU_SCRATCH accesses are allowed.
+ * - Concurrent STARPU_REDUX accesses are allowed.
+ */
 
 /* the header lock must be taken by the caller */
 static unsigned may_unlock_data_req_list_head(starpu_data_handle handle)
@@ -46,22 +38,28 @@ static unsigned may_unlock_data_req_list_head(starpu_data_handle handle)
 
 	/* if there is no reference to the data anymore, we can use it */
 	if (handle->refcnt == 0)
-	{
-		int i;
-		for (i = 0; i < STARPU_MAXNODES; i++)
-			STARPU_ASSERT(!handle->per_node[i]->request);
 		return 1;
-	}
 
 	if (handle->current_mode == STARPU_W)
 		return 0;
 
 	/* data->current_mode == STARPU_R, so we can process more readers */
 	starpu_data_requester_t r = starpu_data_requester_list_front(handle->req_list);
+
+	starpu_access_mode r_mode = r->mode;
+	if (r_mode == STARPU_RW)
+		r_mode = STARPU_W;
 	
-	return (r->mode == STARPU_R);
+	/* If this is a STARPU_R, STARPU_SCRATCH or STARPU_REDUX type of
+	 * access, we only proceed if the cuurrent mode is the same as the
+	 * requested mode. */
+	return (r_mode == handle->current_mode);
 }
 
+/* Try to submit a data request, in case the request can be processed
+ * immediatly, return 0, if there is still a dependency that is not compatible
+ * with the current mode, the request is put in the per-handle list of
+ * "requesters", and this function returns 1. */
 static unsigned _starpu_attempt_to_submit_data_request(unsigned request_from_codelet,
 					starpu_data_handle handle, starpu_access_mode mode,
 					void (*callback)(void *), void *argcb,
@@ -69,9 +67,12 @@ static unsigned _starpu_attempt_to_submit_data_request(unsigned request_from_cod
 {
 	unsigned ret;
 
-	if (mode & STARPU_SCRATCH)
-		return 0;
+	if (mode == STARPU_RW)
+		mode = STARPU_W;
 
+	/* Take the lock protecting the header. We try to do some progression
+	 * in case this is called from a worker, otherwise we just wait for the
+	 * lock to be available. */
 	if (request_from_codelet)
 	{
 		while (_starpu_spin_trylock(&handle->header_lock))
@@ -81,44 +82,35 @@ static unsigned _starpu_attempt_to_submit_data_request(unsigned request_from_cod
 		_starpu_spin_lock(&handle->header_lock);
 	}
 
-	if (handle->refcnt == 0)
+	/* If there is currently nobody accessing the piece of data, or it's
+	 * not another writter and if this is the same type of access as the
+	 * current one, we can proceed. */
+	if ((handle->refcnt == 0) || (!(mode == STARPU_W) && (handle->current_mode == mode)))
 	{
-		/* there is nobody currently about to manipulate the data */
 		handle->refcnt++;
-		handle->current_mode = (mode==STARPU_R)?STARPU_R:STARPU_W;
+		handle->current_mode = mode;
 
 		/* success */
 		ret = 0;
 	}
 	else
 	{
-		/* there is already someone that may access the data */
-		if ( (mode == STARPU_R) && (handle->current_mode == STARPU_R))
-		{
-			handle->refcnt++;
-
-			/* success : there is a new reader */
-			ret = 0;
-		}
-		else
-		{
-			/* there cannot be multiple writers or a new writer
-			 * while the data is in read mode */
-			
-			/* enqueue the request */
-			starpu_data_requester_t r = starpu_data_requester_new();
-				r->mode = mode;
-				r->is_requested_by_codelet = request_from_codelet;
-				r->j = j;
-				r->buffer_index = buffer_index;
-				r->ready_data_callback = callback;
-				r->argcb = argcb;
-
-			starpu_data_requester_list_push_back(handle->req_list, r);
-
-			/* failed */
-			ret = 1;
-		}
+		/* there cannot be multiple writers or a new writer
+		 * while the data is in read mode */
+		
+		/* enqueue the request */
+		starpu_data_requester_t r = starpu_data_requester_new();
+			r->mode = mode;
+			r->is_requested_by_codelet = request_from_codelet;
+			r->j = j;
+			r->buffer_index = buffer_index;
+			r->ready_data_callback = callback;
+			r->argcb = argcb;
+
+		starpu_data_requester_list_push_back(handle->req_list, r);
+
+		/* failed */
+		ret = 1;
 	}
 
 	_starpu_spin_unlock(&handle->header_lock);
@@ -180,21 +172,42 @@ unsigned _starpu_submit_job_enforce_data_deps(starpu_job_t j)
 	return _submit_job_enforce_data_deps(j, 0);
 }
 
+static unsigned unlock_one_requester(starpu_data_requester_t r)
+{
+	starpu_job_t j = r->j;
+	unsigned nbuffers = j->task->cl->nbuffers;
+	unsigned buffer_index = r->buffer_index;
+
+	if (buffer_index + 1 < nbuffers)
+	{
+		/* not all buffers are protected yet */
+		return _submit_job_enforce_data_deps(j, buffer_index + 1);
+	}
+	else
+		return 0;
+}
 
 /* The header lock must already be taken by the caller */
 void _starpu_notify_data_dependencies(starpu_data_handle handle)
 {
+	/* A data access has finished so we remove a reference. */
 	handle->refcnt--;
 
 	while (may_unlock_data_req_list_head(handle))
 	{
-		/* unlock the head of the requester list */
+		/* Grab the head of the requester list and unlock it. */
 		starpu_data_requester_t r = starpu_data_requester_list_pop_front(handle->req_list);
 
+		/* The data is now attributed to that request so we put a
+		 * reference on it. */
 		handle->refcnt++;
 	
-		/* the data is now attributed to that request */
-		handle->current_mode = (r->mode==STARPU_R)?STARPU_R:STARPU_W;
+		/* STARPU_RW accesses are treated as STARPU_W */
+		starpu_access_mode r_mode = r->mode;
+		if (r_mode == STARPU_RW)
+			r_mode = STARPU_W;
+
+		handle->current_mode = r_mode;
 
 		_starpu_spin_unlock(&handle->header_lock);