Fix sync images race condition.

Syncing a set of images using SYNC IMAGES is done by this simple algorithm:

1. Set up asynchronous mpi-receives for all images in the set to sync to,
2. Send the current image's status to all images in the set to sync to, and
3. Wait for the receives from step 1 to finish. When one of the receives
   returns a stopped image state, than abort the sync on the current image
   immediately, else wait until all receives have been answered.

This commit also adds a new testcase: syncimages_ring, where each image
is syncing to its neighbours with wrap around. The testcase is called for
3 (the minimum number of images required to show that the sync works), 13
and 23 images.

This commit fixes the timeout of syncimage_status.f90 mentioned in #298.
This commit superseeds pull request 299 (close #299).

Author: vehre

.travis.yml

@@ -160,11 +160,11 @@ after_success:
   - gcov-6 --version
   - bash <(curl -s https://codecov.io/bash) -x $(which gcov-6)
 
-notifications:
-  webhooks:
-    urls:
-      - https://webhooks.gitter.im/e/93dbafbdf76c1732a623
-      - https://webhooks.gitter.im/e/935231573bf1b9f2fe40
-    on_success: change  # options: [always|never|change]
-    on_failure: always
-    on_start: always
+    #notifications:
+    #webhooks:
+    #urls:
+    #- https://webhooks.gitter.im/e/93dbafbdf76c1732a623
+    #- https://webhooks.gitter.im/e/935231573bf1b9f2fe40
+    #on_success: change  # options: [always|never|change]
+    #on_failure: always
+    #on_start: always

CMakeLists.txt

@@ -391,6 +391,9 @@ if(opencoarrays_aware_compiler)
   add_mpi_test(duplicate_syncimages 8 ${tests_root}/unit/sync/duplicate_syncimages)
   add_mpi_test(co_reduce 4 ${tests_root}/unit/collectives/co_reduce_test)
   add_mpi_test(syncimages_status 32 ${tests_root}/unit/sync/syncimages_status)
+  add_mpi_test(syncimages_ring_np3 3 ${tests_root}/unit/sync/syncimages_ring)
+  add_mpi_test(syncimages_ring_np13 13 ${tests_root}/unit/sync/syncimages_ring)
+  add_mpi_test(syncimages_ring_np23 23 ${tests_root}/unit/sync/syncimages_ring)
   add_mpi_test(simpleatomics 32 ${tests_root}/unit/simple/atomics)
   # possible logic error in the following test
 #  add_mpi_test(increment_my_neighbor 32 ${tests_root}/unit/simple/increment_my_neighbor)

src/mpi/mpi_caf.c

@@ -75,8 +75,8 @@ static int caf_is_finalized = 0;
 /*Sync image part*/
 
 static int *orders;
-MPI_Request *handlers;
 static int *images_full;
+MPI_Request *sync_handles;
 static int *arrived;
 
 /* Pending puts */
@@ -96,8 +96,8 @@ caf_static_t *caf_tot = NULL;
 
 /* Image status variable */
 
-static int *img_status = NULL;
-MPI_Win *stat_tok;
+static int img_status = 0;
+static MPI_Win *stat_tok;
 
 /* Active messages variables */
 
@@ -400,27 +400,31 @@ PREFIX (init) (int *argc, char ***argv)
       orders = calloc (caf_num_images, sizeof (int));
       arrived = calloc (caf_num_images, sizeof (int));
 
-      handlers = malloc(caf_num_images * sizeof(MPI_Request));
+      sync_handles = malloc(caf_num_images * sizeof(MPI_Request));
 
       stat_tok = malloc (sizeof(MPI_Win));
 
 #if MPI_VERSION >= 3
       MPI_Info_create (&mpi_info_same_size);
       MPI_Info_set (mpi_info_same_size, "same_size", "true");
       /* Setting img_status */
-      MPI_Win_allocate(sizeof(int), 1, mpi_info_same_size, CAF_COMM_WORLD, &img_status, stat_tok);
+      MPI_Win_create(&img_status, sizeof(int), 1, mpi_info_same_size, CAF_COMM_WORLD, stat_tok);
 # ifndef CAF_MPI_LOCK_UNLOCK
       MPI_Win_lock_all(MPI_MODE_NOCHECK, *stat_tok);
 # endif // CAF_MPI_LOCK_UNLOCK
 #else
       MPI_Alloc_mem(sizeof(int), MPI_INFO_NULL, &img_status, stat_tok);
       MPI_Win_create(img_status, sizeof(int), 1, MPI_INFO_NULL, CAF_COMM_WORLD, stat_tok);
 #endif // MPI_VERSION
-      *img_status = 0;
     }
   /* MPI_Barrier(CAF_COMM_WORLD); */
 }
 
+/* Forward declaration of sync_images.  */
+
+void
+sync_images_internal (int count, int images[], int *stat, char *errmsg,
+                      int errmsg_len, bool internal);
 
 /* Finalize coarray program.   */
 
@@ -431,10 +435,17 @@ _gfortran_caf_finalize (void)
 PREFIX (finalize) (void)
 #endif
 {
-  *img_status = STAT_STOPPED_IMAGE; /* GFC_STAT_STOPPED_IMAGE = 6000 */
-  MPI_Win_sync(*stat_tok);
+  /* For future security enclose setting img_status in a lock.  */
+  CAF_Win_lock (MPI_LOCK_EXCLUSIVE, caf_this_image - 1, *stat_tok);
+  img_status = STAT_STOPPED_IMAGE; /* GFC_STAT_STOPPED_IMAGE = 6000 */
+  CAF_Win_unlock (caf_this_image - 1, *stat_tok);
 
-  MPI_Barrier(CAF_COMM_WORLD);
+  /* Announce to all other images, that this one is stopped.  */
+  for (int i = 0; i < caf_num_images - 1; ++i)
+    MPI_Send (&img_status, 1, MPI_INT, images_full[i] - 1, 0, CAF_COMM_WORLD);
+
+  /* Add a conventional barrier to prevent images from quitting to early.  */
+  MPI_Barrier (CAF_COMM_WORLD);
 
   while (caf_static_list != NULL)
     {
@@ -477,6 +488,10 @@ PREFIX (finalize) (void)
   MPI_Info_free (&mpi_info_same_size);
 #endif // MPI_VERSION
 
+# ifndef CAF_MPI_LOCK_UNLOCK
+  MPI_Win_unlock_all (*stat_tok);
+# endif // CAF_MPI_LOCK_UNLOCK
+  MPI_Win_free (stat_tok);
   MPI_Comm_free(&CAF_COMM_WORLD);
 
   /* Only call Finalize if CAF runtime Initialized MPI. */
@@ -486,7 +501,7 @@ PREFIX (finalize) (void)
   pthread_mutex_lock(&lock_am);
   caf_is_finalized = 1;
   pthread_mutex_unlock(&lock_am);
-  exit(0);
+  free (sync_handles);
 }
 
 
@@ -2988,7 +3003,14 @@ void
 PREFIX (sync_images) (int count, int images[], int *stat, char *errmsg,
                      int errmsg_len)
 {
-  int ierr = 0, i = 0, remote_stat = 0, j = 0;
+  sync_images_internal (count, images, stat, errmsg, errmsg_len, false);
+}
+
+void
+sync_images_internal (int count, int images[], int *stat, char *errmsg,
+                      int errmsg_len, bool internal)
+{
+  int ierr = 0, i = 0, j = 0, int_zero = 0, done_count = 0;
   MPI_Status s;
 
   if (count == 0 || (count == 1 && images[0] == caf_this_image))
@@ -3042,59 +3064,54 @@ PREFIX (sync_images) (int count, int images[], int *stat, char *errmsg,
       explicit_flush();
 #endif
 
+      /* A rather simple way to synchronice:
+	 - expect all images to sync with receiving an int,
+	 - on the other side, send all processes to sync with an int,
+	 - when the int received is STAT_STOPPED_IMAGE the return immediately,
+	   else wait until all images in the current set of images have send
+	   some data, i.e., synced.
+
+	 This approach as best as possible implements the syncing of different
+	 sets of images and figuring that an image has stopped.  MPI does not
+	 provide any direct means of syncing non-coherent sets of images.
+	 The groups/communicators of MPI always need to be consistent, i.e.,
+	 have the same members on all images participating.  This is
+	 contradictiory to the sync images statement, where syncing, e.g., in a
+	 ring pattern is possible.
+
+	 This implementation guarantees, that as long as no image is stopped
+	 an image only is allowed to continue, when all its images to sync to
+	 also have reached a sync images statement.  This implementation makes
+	 no assumption when the image continues or in which order synced
+	 images continue.  */
       for(i = 0; i < count; ++i)
 	/* Need to have the request handlers contigously in the handlers
 	   array or waitany below will trip about the handler as illegal.  */
 	ierr = MPI_Irecv (&arrived[images[i] - 1], 1, MPI_INT, images[i] - 1, 0,
-	    CAF_COMM_WORLD, &handlers[i]);
+			  CAF_COMM_WORLD, &sync_handles[i]);
       for(i = 0; i < count; ++i)
+	MPI_Send (&int_zero, 1, MPI_INT, images[i] - 1, 0, CAF_COMM_WORLD);
+      done_count = 0;
+      while (done_count < count)
 	{
-# ifdef CAF_MPI_LOCK_UNLOCK
-	  MPI_Win_lock (MPI_LOCK_SHARED, images[i] - 1, 0, *stat_tok);
-# endif // CAF_MPI_LOCK_UNLOCK
-	  ierr = MPI_Get (&remote_stat, 1, MPI_INT,
-			  images[i] - 1, 0, 1, MPI_INT, *stat_tok);
-# ifdef CAF_MPI_LOCK_UNLOCK
-	  MPI_Win_unlock (images[i] - 1, *stat_tok);
-# else // CAF_MPI_LOCK_UNLOCK
-	  MPI_Win_flush (images[i] - 1, *stat_tok);
-# endif // CAF_MPI_LOCK_UNLOCK
-	  if(remote_stat != 0)
+	  ierr = MPI_Waitany (count, sync_handles, &i, &s);
+	  if (i != MPI_UNDEFINED)
 	    {
-	      ierr = STAT_STOPPED_IMAGE;
-	      /* Let the other images know, that at least one image is
-		  stopped by sending STAT_STOPPED_IMAGE instead of our id.  */
-	      for(i = 0; i < count; ++i)
-		MPI_Send (&ierr, 1, MPI_INT, images[i] - 1, 0, CAF_COMM_WORLD);
-	      break;
-	    }
-	}
-      if (ierr == 0)
-	{
-	  int done_count = 0;
-	  for(i = 0; i < count; ++i)
-	    {
-	      if (arrived[images[i] - 1] != STAT_STOPPED_IMAGE)
-		/* Only send, when no stopped images have been found.  */
-		ierr = MPI_Send (&caf_this_image, 1, MPI_INT, images[i] - 1, 0,
-				 CAF_COMM_WORLD);
-	      else
-		ierr = STAT_STOPPED_IMAGE;
-	    }
-
-	  while (ierr != STAT_STOPPED_IMAGE && done_count < count)
-	    {
-	      ierr = MPI_Waitany (count, handlers, &i, &s);
-	      if (i != MPI_UNDEFINED)
-		++done_count;
-	      if (i != MPI_UNDEFINED && arrived[i] == STAT_STOPPED_IMAGE)
-		ierr = STAT_STOPPED_IMAGE;
-	      else if (ierr != MPI_SUCCESS)
-		break;
+	      ++done_count;
+	      if (ierr == MPI_SUCCESS && arrived[i] == STAT_STOPPED_IMAGE)
+		{
+		  /* Possible future extension: Abort pending receives.  At the
+		     moment the receives are discarded by the program
+		     termination.  For the tested mpi-implementation this is ok.
+		   */
+		  ierr = STAT_STOPPED_IMAGE;
+		  break;
+		}
 	    }
+	  else if (ierr != MPI_SUCCESS)
+	    /* Abort receives here, too, when implemented above.  */
+	    break;
 	}
-
-      memset(arrived, 0, sizeof(int) * caf_num_images);
     }
 
 sync_images_err_chk:
@@ -3117,8 +3134,8 @@ PREFIX (sync_images) (int count, int images[], int *stat, char *errmsg,
           if (errmsg_len > len)
             memset (&errmsg[len], ' ', errmsg_len-len);
         }
-      else
-        caf_runtime_error (msg);
+      else if (!internal)
+	caf_runtime_error (msg);
     }
 }
 

src/tests/unit/sync/CMakeLists.txt

@@ -12,3 +12,6 @@ target_link_libraries(duplicate_syncimages OpenCoarrays)
 
 add_executable(syncimages_status syncimages_status.f90)
 target_link_libraries(syncimages_status OpenCoarrays)
+
+add_executable(syncimages_ring syncimages_ring.f90)
+target_link_libraries(syncimages_ring OpenCoarrays)

src/tests/unit/sync/syncimages_ring.f90

@@ -0,0 +1,25 @@
+! SYNC IMAGES([this_image - 1, this_image + 1]) with the STAT=STAT_STOPPED_IMAGE
+! specifier and wrap around of image numbers.  The test is intended to check
+! that syncing in a ring with a stopped image still terminates all images.
+
+program sync_images_ring
+  use, intrinsic:: iso_fortran_env
+  implicit none
+
+  integer :: stat_var = 0
+
+  if (num_images() .lt. 2) error stop "Need at least two images to test."
+
+  associate (me => this_image())
+    if (me /= 1) then
+	  associate (lhs => merge(me - 1, num_images(), me /= 1), &
+				 rhs => merge(me + 1, 1, me /= num_images()))
+		sync images([lhs, rhs], STAT=stat_var)
+		! Only on image 2 and num_images() testing whether a stopped image is
+		! present can be done reliably. All other images could be up ahead.
+		if (stat_var /= STAT_STOPPED_IMAGE .and. me == 2) error stop "Error: stat_var /= STAT_STOPPED_IMAGE: "
+		if(me == 2) print *, 'Test passed.'
+	  end associate
+	end if
+  end associate
+end program sync_images_ring

src/tests/unit/sync/syncimages_status.f90

@@ -10,7 +10,6 @@ program sync_images_stat
   me = this_image()
 
   if (me /= 1 ) then
-     call sleep(1)
      sync images(*,STAT=stat_var)
      if ( stat_var /= STAT_STOPPED_IMAGE) then
         print *, "Error:stat_var /= STAT_STOPPED_IMAGE: ", me