[svn-r18638] Description:

	Lock the dataspace message into object header chunk #0 (so that the
flush dependency between the extensible array and the object header works and
has the correct semantics).

Tested on:
        FreeBSD/32 6.3 (duty) in debug mode
        FreeBSD/64 6.3 (liberty) w/C++ & FORTRAN, in debug mode
        Linux/32 2.6 (jam) w/PGI compilers, w/default API=1.8.x,
                w/C++ & FORTRAN, w/threadsafe, in debug mode
        Linux/64-amd64 2.6 (amani) w/Intel compilers, w/default API=1.6.x,
                w/C++ & FORTRAN, in production mode
        Solaris/32 2.10 (linew) w/deprecated symbols disabled, w/C++ & FORTRAN,
                w/szip filter, in production mode
        Linux/64-ia64 2.6 (cobalt) w/Intel compilers, w/C++ & FORTRAN,
                in production mode
        Linux/64-ia64 2.4 (tg-login3) w/parallel, w/FORTRAN, in debug mode
        Linux/64-amd64 2.6 (abe) w/parallel, w/FORTRAN, in production mode
        Mac OS X/32 10.6.3 (amazon) in debug mode
        Mac OS X/32 10.6.3 (amazon) w/C++ & FORTRAN, w/threadsafe,
                in production mode

6 files changed, 146 insertions, 20 deletions

diff --git a/src/H5Dchunk.c b/src/H5Dchunk.c
index 68d8781..7087166 100644
--- a/src/H5Dchunk.c
+++ b/src/H5Dchunk.c
@@ -2458,7 +2458,7 @@ H5D_chunk_flush_entry(const H5D_t *dset, hid_t dxpl_id, const H5D_dxpl_cache_t *
         ent->dirty = FALSE;
 
         /* Check for SWMR writes to the file */
-        if(dset->shared->layout.storage.u.chunk.ops->can_swim && H5F_INTENT(dset->oloc.file) & H5F_ACC_SWMR_WRITE) {
+        if(dset->shared->layout.storage.u.chunk.ops->can_swim && (H5F_INTENT(dset->oloc.file) & H5F_ACC_SWMR_WRITE)) {
             /* Mark the proxy entry in the cache as clean */
             if(H5D_chunk_proxy_mark(ent, FALSE) < 0)
                 HGOTO_ERROR(H5E_DATASET, H5E_CANTMARKDIRTY, FAIL, "can't mark proxy for chunk from metadata cache as clean")
@@ -2537,7 +2537,7 @@ H5D_chunk_cache_evict(const H5D_t *dset, hid_t dxpl_id, const H5D_dxpl_cache_t *
     } /* end else */
 
     /* Check for SWMR writes to the file */
-    if(dset->shared->layout.storage.u.chunk.ops->can_swim && H5F_INTENT(dset->oloc.file) & H5F_ACC_SWMR_WRITE) {
+    if(dset->shared->layout.storage.u.chunk.ops->can_swim && (H5F_INTENT(dset->oloc.file) & H5F_ACC_SWMR_WRITE)) {
         /* Remove the proxy entry in the cache */
         if(H5D_chunk_proxy_remove(dset, dxpl_id, ent) < 0)
             HGOTO_ERROR(H5E_DATASET, H5E_CANTREMOVE, FAIL, "can't remove proxy for chunk from metadata cache")
@@ -2893,7 +2893,7 @@ H5D_chunk_lock(const H5D_io_info_t *io_info, H5D_chunk_ud_t *udata,
 
         /* Check for SWMR writes to the file */
         if(io_info->dset->shared->layout.storage.u.chunk.ops->can_swim
-                && H5F_INTENT(io_info->dset->oloc.file) & H5F_ACC_SWMR_WRITE) {
+                && (H5F_INTENT(io_info->dset->oloc.file) & H5F_ACC_SWMR_WRITE)) {
             /* Insert a proxy entry in the cache, to make certain that the
              *  flush dependencies are maintained in the proper way for SWMR
              *  access to work.
@@ -3041,7 +3041,7 @@ H5D_chunk_unlock(const H5D_io_info_t *io_info, const H5D_chunk_ud_t *udata,
 
             /* Check for SWMR writes to the file */
             if(io_info->dset->shared->layout.storage.u.chunk.ops->can_swim
-                    && H5F_INTENT(io_info->dset->oloc.file) & H5F_ACC_SWMR_WRITE) {
+                    && (H5F_INTENT(io_info->dset->oloc.file) & H5F_ACC_SWMR_WRITE)) {
                 /* Mark the proxy entry in the cache as dirty */
                 if(H5D_chunk_proxy_mark(ent, TRUE) < 0)
                     HGOTO_ERROR(H5E_DATASET, H5E_CANTMARKDIRTY, FAIL, "can't mark proxy for chunk from metadata cache as dirty")
diff --git a/src/H5Dint.c b/src/H5Dint.c
index 9f6f93f..425b243 100644
--- a/src/H5Dint.c
+++ b/src/H5Dint.c
@@ -65,6 +65,8 @@ static H5D_shared_t *H5D_new(hid_t dcpl_id, hbool_t creating,
 static herr_t H5D_init_type(H5F_t *file, const H5D_t *dset, hid_t type_id,
     const H5T_t *type);
 static herr_t H5D_init_space(H5F_t *file, const H5D_t *dset, const H5S_t *space);
+static herr_t H5D_swmr_setup(const H5D_t *dset, hid_t dxpl_id);
+static herr_t H5D_swmr_teardown(const H5D_t *dataset, hid_t dxpl_id);
 static herr_t H5D_update_oh_info(H5F_t *file, hid_t dxpl_id, H5D_t *dset,
     hid_t dapl_id);
 static herr_t H5D_open_oid(H5D_t *dataset, hid_t dapl_id, hid_t dxpl_id);
@@ -692,6 +694,105 @@ done:
 
 
 /*-------------------------------------------------------------------------
+ * Function:	H5D_swmr_setup
+ *
+ * Purpose:	Set up SWMR access for a chunked dataset, if possible
+ *
+ * Return:	Success:    SUCCEED
+ *		Failure:    FAIL
+ *
+ * Programmer:	Quincey Koziol
+ *		Tuesday, April 27, 2010
+ *
+ *-------------------------------------------------------------------------
+ */
+static herr_t
+H5D_swmr_setup(const H5D_t *dataset, hid_t dxpl_id)
+{
+    herr_t ret_value = SUCCEED;         /* Return value */
+
+    FUNC_ENTER_NOAPI_NOINIT(H5D_swmr_setup)
+
+    /* Sanity checking */
+    HDassert(dataset);
+
+    /* Check if it's possible to enable swmr access to this dataset */
+    if(dataset->shared->layout.type == H5D_CHUNKED &&
+            dataset->shared->layout.storage.u.chunk.ops->can_swim &&
+            (H5F_INTENT(dataset->oloc.file) & H5F_ACC_SWMR_WRITE)) {
+        int chunkno;    /* Object header chunk index for message */
+
+        /* Get object header chunk index for dataspace message */
+        if((chunkno = H5O_msg_get_chunkno(&dataset->oloc, H5O_SDSPACE_ID, dxpl_id)) < 0)
+            HGOTO_ERROR(H5E_DATASET, H5E_CANTGET, FAIL, "unable to query dataspace chunk index")
+
+        /* Fail currently, if the dataspace message is not in chunk #0 */
+        /* (Note that this could be addressed by moving the dataspace message
+         *      into chunk #0, but that can be hard and we're deferring that
+         *      work for now. -QAK)
+         */
+        if(chunkno > 0)
+            HGOTO_ERROR(H5E_DATASET, H5E_BADVALUE, FAIL, "dataspace chunk index must be 0 for SWMR access, chunkno = %d", chunkno)
+
+        /* Pin the object header */
+        if(NULL == (dataset->shared->oh = H5O_pin(&dataset->oloc, dxpl_id)))
+            HGOTO_ERROR(H5E_DATASET, H5E_CANTPIN, FAIL, "unable to pin dataset object header")
+
+        /* Lock dataspace message into chunk #0 */
+        if(H5O_msg_lock(&dataset->oloc, H5O_SDSPACE_ID, dxpl_id) < 0)
+            HGOTO_ERROR(H5E_DATASET, H5E_CANTLOCK, FAIL, "can't lock dataspace message into object header chunk #0")
+
+        /* Indicate that dataset is set up for SWMR access */
+        dataset->shared->is_swimming = TRUE;
+    } /* end if */
+
+done:
+    FUNC_LEAVE_NOAPI(ret_value)
+} /* end H5D_swmr_setup() */
+
+
+/*-------------------------------------------------------------------------
+ * Function:	H5D_swmr_teardown
+ *
+ * Purpose:	Tear down SWMR access for a chunked dataset.
+ *
+ * Return:	Success:    SUCCEED
+ *		Failure:    FAIL
+ *
+ * Programmer:	Quincey Koziol
+ *		Tuesday, April 27, 2010
+ *
+ *-------------------------------------------------------------------------
+ */
+static herr_t
+H5D_swmr_teardown(const H5D_t *dataset, hid_t dxpl_id)
+{
+    herr_t ret_value = SUCCEED;         /* Return value */
+
+    FUNC_ENTER_NOAPI_NOINIT(H5D_swmr_teardown)
+
+    /* Sanity checking */
+    HDassert(dataset);
+    HDassert(dataset->shared->is_swimming);
+    HDassert(dataset->shared->oh);
+
+    /* Unlock dataspace message from chunk #0 */
+    if(H5O_msg_unlock(&dataset->oloc, H5O_SDSPACE_ID, dxpl_id) < 0)
+        HGOTO_ERROR(H5E_DATASET, H5E_CANTUNLOCK, FAIL, "can't unlock dataspace message from object header chunk #0")
+
+    /* Release pointer to object header */
+    if(H5O_unpin(dataset->shared->oh) < 0)
+        HGOTO_ERROR(H5E_DATASET, H5E_CANTUNPIN, FAIL, "unable to unpin dataset object header")
+
+    /* Indicate that dataset is NOT set up for SWMR access now */
+    dataset->shared->is_swimming = FALSE;
+
+done:
+    FUNC_LEAVE_NOAPI(ret_value)
+} /* end H5D_swmr_teardown() */
+
+
+/*-------------------------------------------------------------------------
  * Function:	H5D_update_oh_info
  *
  * Purpose:	Create and fill object header for dataset
@@ -797,9 +898,11 @@ H5D_update_oh_info(H5F_t *file, hid_t dxpl_id, H5D_t *dset, hid_t dapl_id)
     if(NULL == (oh = H5O_pin(oloc, dxpl_id)))
         HGOTO_ERROR(H5E_DATASET, H5E_CANTPIN, FAIL, "unable to pin dataset object header")
 
-    /* Update the datatype and dataspace header messages */
+    /* Write the dataspace header message */
     if(H5S_append(file, dxpl_id, oh, dset->shared->space) < 0)
         HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL, "unable to update dataspace header message")
+
+    /* Write the datatype header message */
     if(H5O_msg_append_oh(file, dxpl_id, oh, H5O_DTYPE_ID, H5O_MSG_FLAG_CONSTANT, 0, type) < 0)
         HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL, "unable to update datatype header message")
 
@@ -862,6 +965,10 @@ H5D_update_oh_info(H5F_t *file, hid_t dxpl_id, H5D_t *dset, hid_t dapl_id)
         if(H5O_touch_oh(file, dxpl_id, oh, TRUE) < 0)
             HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL, "unable to update modification time message")
 
+    /* Set up SWMR writes to the dataset, if possible */
+    if(H5D_swmr_setup(dset, dxpl_id) < 0)
+        HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL, "unable to set up SWMR access for dataset")
+
 done:
     /* Release pointer to object header itself */
     if(oh != NULL)
@@ -1312,6 +1419,10 @@ H5D_open_oid(H5D_t *dataset, hid_t dapl_id, hid_t dxpl_id)
             HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL, "unable to initialize file storage")
     } /* end if */
 
+    /* Set up SWMR writes to the dataset, if possible */
+    if(H5D_swmr_setup(dataset, dxpl_id) < 0)
+        HGOTO_ERROR(H5E_DATASET, H5E_CANTINIT, FAIL, "unable to set up SWMR access for dataset")
+
 done:
     if(ret_value < 0) {
         if(H5F_addr_defined(dataset->oloc.addr) && H5O_close(&(dataset->oloc)) < 0)
@@ -1424,6 +1535,11 @@ H5D_close(H5D_t *dataset)
 #endif /* NDEBUG */
         } /* end switch */ /*lint !e788 All appropriate cases are covered */
 
+        /* If the dataset is opened for SWMR access, shut that down */
+        if(dataset->shared->is_swimming)
+            if(H5D_swmr_teardown(dataset, H5AC_dxpl_id) < 0)
+                HGOTO_ERROR(H5E_DATASET, H5E_CLOSEERROR, FAIL, "unable to shut down SWMR access")
+        
         /*
         * Release datatype, dataspace and creation property list -- there isn't
         * much we can do if one of these fails, so we just continue.
diff --git a/src/H5Dlayout.c b/src/H5Dlayout.c
index bebe1dc..acfae3f 100644
--- a/src/H5Dlayout.c
+++ b/src/H5Dlayout.c
@@ -305,7 +305,7 @@ H5D_layout_set_latest_version(H5O_layout_t *layout, const H5S_t *space)
             /* Set the chunk index type to an extensible array */
             layout->u.chunk.idx_type = H5D_CHUNK_IDX_EARRAY;
             layout->storage.u.chunk.idx_type = H5D_CHUNK_IDX_EARRAY;
-            layout->storage.u.chunk.ops = H5D_COPS_EARRAY;;
+            layout->storage.u.chunk.ops = H5D_COPS_EARRAY;
 
             /* Set the extensible array creation parameters */
             /* (use hard-coded defaults for now, until we give applications
@@ -322,7 +322,7 @@ H5D_layout_set_latest_version(H5O_layout_t *layout, const H5S_t *space)
             /* Set the chunk index type to a fixed array */
 	    layout->u.chunk.idx_type = H5D_CHUNK_IDX_FARRAY;
             layout->storage.u.chunk.idx_type = H5D_CHUNK_IDX_FARRAY;
-            layout->storage.u.chunk.ops = H5D_COPS_FARRAY;;
+            layout->storage.u.chunk.ops = H5D_COPS_FARRAY;
 
             /* Set the fixed array creation parameters */
             /* (use hard-coded defaults for now, until we give applications
diff --git a/src/H5Dpkg.h b/src/H5Dpkg.h
index 68b5835..0788637 100644
--- a/src/H5Dpkg.h
+++ b/src/H5Dpkg.h
@@ -432,6 +432,8 @@ typedef struct H5D_shared_t {
     H5D_dcpl_cache_t    dcpl_cache;     /* Cached DCPL values */
     H5O_layout_t        layout;         /* Data layout                  */
     hbool_t             checked_filters;/* TRUE if dataset passes can_apply check */
+    H5O_t              *oh;             /* Pointer to dataset's object header, pinned */
+    hbool_t             is_swimming;    /* TRUE if dataset has SWMR access enabled */
 
     /* Buffered/cached information for types of raw data storage*/
     struct {
diff --git a/test/swmr_reader.c b/test/swmr_reader.c
index 9c9cd1e..394a46e 100644
--- a/test/swmr_reader.c
+++ b/test/swmr_reader.c
@@ -214,19 +214,19 @@ static void
 usage(void)

 {

     printf("Usage error!\n");

-    printf("Usage: swmr_reader [-q] [-s <# of seconds to sleep between polling>] [-h <# of common symbols to poll>] [-l <# of random symbols to poll>] <# of seconds to test>\n");

+    printf("Usage: swmr_reader [-q] [-s <# of seconds to sleep between polling>] [-h <# of common symbols to poll>] [-l <# of random symbols to poll>] [-r <random # seed>] <# of seconds to test>\n");

     printf("Defaults to verbose (no '-q' given), 1 second between polling ('-s 1'), 5 common symbols to poll ('-h 5') and 10 random symbols to poll ('-l 10')\n");

     exit(1);

 } 

 

 int main(int argc, const char *argv[])

 {

-    time_t curr_time;   /* Current time, for seeding random number generator */

     long nseconds = 0;  /* # of seconds to test */

     int poll_time = 1;  /* # of seconds between polling */

     int ncommon = 5;    /* # of common symbols to poll */

     int nrandom = 10;   /* # of random symbols to poll */

     unsigned verbose = 1;       /* Whether to emit some informational messages */

+    int random_seed = 0; /* Random # seed */

     unsigned u;         /* Local index variables */

 

     /* Parse command line options */

@@ -237,14 +237,6 @@ int main(int argc, const char *argv[])
         while(u < (unsigned)argc) {

             if(argv[u][0] == '-') {

                 switch(argv[u][1]) {

-                    /* # of seconds between polling */

-                    case 's':

-                        poll_time = atoi(argv[u + 1]);

-                        if(poll_time < 0)

-                            usage();

-                        u += 2;

-                        break;

-

                     /* # of common symbols to poll */

                     case 'h':

                         ncommon = atoi(argv[u + 1]);

@@ -267,6 +259,22 @@ int main(int argc, const char *argv[])
                         u++;

                         break;

 

+                    /* Random # seed */

+                    case 'r':

+                        random_seed = atoi(argv[u + 1]);

+                        if(random_seed < 0)

+                            usage();

+                        u += 2;

+                        break;

+

+                    /* # of seconds between polling */

+                    case 's':

+                        poll_time = atoi(argv[u + 1]);

+                        if(poll_time < 0)

+                            usage();

+                        u += 2;

+                        break;

+

                     default:

                         usage();

                         break;

@@ -297,8 +305,8 @@ int main(int argc, const char *argv[])
     } /* end if */

 

     /* Create randomized set of numbers */

-    curr_time = time(NULL);

-    srandom((unsigned)curr_time);

+    random_seed += (int)time(NULL);

+    srandom((unsigned)random_seed);

 

     /* Emit informational message */

     if(verbose)

diff --git a/test/testswmr.sh b/test/testswmr.sh
index bec97bb..c0d2e08 100755
--- a/test/testswmr.sh
+++ b/test/testswmr.sh
@@ -81,7 +81,7 @@ $DPRINT pid_writer=$pid_writer
 n=0
 echo launch $Nreaders swmr_readers
 while [ $n -lt $Nreaders ]; do
-    ./swmr_reader $Nsecs &
+    ./swmr_reader -r $n $Nsecs &
     pid_readers="$pid_readers $!"
     n=`expr $n + 1`
 done