[svn-r13690] Description:

    Fully enable readers/writer locking in metadata cache and correct errors
in cache clients which were modifying a cache entry while only holding a
read lock on it.

Tested on:
    Mac OS X/32 10.4.9 (amazon)
    Linux/32 2.6 (chicago)
    Linux/64 2.6 (chicago2)
    FreeBSD/32 6.2 (duty)
    FreeBSD/64 6.2 (liberty)

11 files changed, 27 insertions, 53 deletions

diff --git a/src/H5B2.c b/src/H5B2.c
index b7f372c..5a2d5ee 100644
--- a/src/H5B2.c
+++ b/src/H5B2.c
@@ -1137,7 +1137,7 @@ H5B2_modify(H5F_t *f, hid_t dxpl_id, const H5B2_class_t *type, haddr_t addr,
         hbool_t changed;        /* Whether the 'modify' callback changed the record */
 
         /* Lock B-tree leaf node */
-        if (NULL == (leaf = H5AC_protect(f, dxpl_id, H5AC_BT2_LEAF, curr_node_ptr.addr, &(curr_node_ptr.node_nrec), bt2_shared, H5AC_READ)))
+        if (NULL == (leaf = H5AC_protect(f, dxpl_id, H5AC_BT2_LEAF, curr_node_ptr.addr, &(curr_node_ptr.node_nrec), bt2_shared, H5AC_WRITE)))
             HGOTO_ERROR(H5E_BTREE, H5E_CANTPROTECT, FAIL, "unable to load B-tree internal node")
 
         /* Locate record */
@@ -1173,7 +1173,7 @@ H5B2_modify(H5F_t *f, hid_t dxpl_id, const H5B2_class_t *type, haddr_t addr,
         } /* end else */
 
         /* Mark the node as dirty if it changed */
-        leaf_flags |= changed ? H5AC__DIRTIED_FLAG : 0;
+        leaf_flags |= (changed ? H5AC__DIRTIED_FLAG : 0);
 
         /* Unlock current node */
         if (H5AC_unprotect(f, dxpl_id, H5AC_BT2_LEAF, curr_node_ptr.addr, leaf, leaf_flags) < 0)
diff --git a/src/H5C.c b/src/H5C.c
index 08ef464..d9951fe 100644
--- a/src/H5C.c
+++ b/src/H5C.c
@@ -2996,7 +2996,6 @@ H5C_create(size_t		      max_cache_size,
         ((cache_ptr->epoch_markers)[i]).is_protected	 = FALSE;
 	((cache_ptr->epoch_markers)[i]).is_read_only	 = FALSE;
 	((cache_ptr->epoch_markers)[i]).ro_ref_count	 = 0;
-	((cache_ptr->epoch_markers)[i]).max_ro_ref_count = 0;
         ((cache_ptr->epoch_markers)[i]).is_pinned	 = FALSE;
         ((cache_ptr->epoch_markers)[i]).in_slist	 = FALSE;
         ((cache_ptr->epoch_markers)[i]).ht_next		 = NULL;
@@ -4569,7 +4568,6 @@ H5C_insert_entry(H5F_t * 	     f,
     entry_ptr->is_protected = FALSE;
     entry_ptr->is_read_only = FALSE;
     entry_ptr->ro_ref_count = 0;
-    entry_ptr->max_ro_ref_count = 0; /* JRM - delete this when possible */
 
     entry_ptr->is_pinned = insert_pinned;
 
@@ -5730,11 +5728,6 @@ H5C_protect(H5F_t *	        f,
 
 	    (entry_ptr->ro_ref_count)++;
 
-	    /* JRM - delete this when possible */
-	    if ( entry_ptr->ro_ref_count > entry_ptr->max_ro_ref_count ) {
-
-		entry_ptr->max_ro_ref_count = entry_ptr->ro_ref_count;
-	    }
 	} else {
 
             HGOTO_ERROR(H5E_CACHE, H5E_CANTPROTECT, NULL, \
@@ -5750,7 +5743,6 @@ H5C_protect(H5F_t *	        f,
 
 	    entry_ptr->is_read_only = TRUE;
 	    entry_ptr->ro_ref_count = 1;
-	    entry_ptr->max_ro_ref_count = 1;
 	}
 
     	entry_ptr->dirtied = FALSE;
@@ -7098,9 +7090,7 @@ H5C_unprotect(H5F_t *		  f,
 
 	    HDassert( entry_ptr->ro_ref_count == 1 );
 
-	    if ( ( dirtied ) &&
-	     /* JRM - delete the following line when possible */ 
-	     ( entry_ptr->max_ro_ref_count > 1 ) ) {
+	    if ( dirtied ) {
 
                 HGOTO_ERROR(H5E_CACHE, H5E_CANTUNPROTECT, FAIL, \
                             "Read only entry modified(2)??")
@@ -7108,7 +7098,6 @@ H5C_unprotect(H5F_t *		  f,
 
 	    entry_ptr->is_read_only = FALSE;
 	    entry_ptr->ro_ref_count = 0;
-	    entry_ptr->max_ro_ref_count = 0;
 	}
 
 #ifdef H5_HAVE_PARALLEL
@@ -9709,7 +9698,6 @@ H5C_load_entry(H5F_t *             f,
     entry_ptr->is_protected = FALSE;
     entry_ptr->is_read_only = FALSE;
     entry_ptr->ro_ref_count = 0;
-    entry_ptr->max_ro_ref_count = 0; /* JRM - delete this when possible */
     entry_ptr->in_slist = FALSE;
     entry_ptr->flush_marker = FALSE;
 #ifdef H5_HAVE_PARALLEL
diff --git a/src/H5Cprivate.h b/src/H5Cprivate.h
index dc71e16..d3fcbeb 100644
--- a/src/H5Cprivate.h
+++ b/src/H5Cprivate.h
@@ -283,20 +283,6 @@ typedef herr_t (*H5C_log_flush_func_t)(H5C_t * cache_ptr,
  * 		must be zero whenever either is_protected or is_read_only
  * 		are TRUE.
  *
- * max_ro_ref_count:  Integer field used to track the maximum value of 
- * 		ro_ref_count in the current protection of this entry.
- * 		The field must be reset to zero when the entry is 
- * 		unprotected.  
- *
- * 		This field exist to allow us to refrain from flagging 
- * 		an error if an entry is protected read only, and then
- * 		unprotected dirtied if the ro_ref_count has not exceeded
- * 		1.  
- *
- * 		It is a temporary fix which should be removed once we
- * 		have corrected all the instances of this behaviour in the
- *		code that calls the metadata cache.
- *
  * is_pinned:	Boolean flag indicating whether the entry has been pinned
  * 		in the cache.
  *
@@ -466,7 +452,6 @@ typedef struct H5C_cache_entry_t
     hbool_t		is_protected;
     hbool_t		is_read_only;
     int			ro_ref_count;
-    int			max_ro_ref_count; /* delete this when possible */
     hbool_t		is_pinned;
     hbool_t		in_slist;
     hbool_t		flush_marker;
diff --git a/src/H5Gent.c b/src/H5Gent.c
index 55a1032..9aeb91b 100644
--- a/src/H5Gent.c
+++ b/src/H5Gent.c
@@ -496,7 +496,7 @@ H5G_ent_debug(H5F_t UNUSED *f, hid_t dxpl_id, const H5G_entry_t *ent, FILE * str
             if(heap_addr > 0 && H5F_addr_defined(heap_addr)) {
                 H5HL_t *heap;
 
-                heap = H5HL_protect(ent->file, dxpl_id, heap_addr);
+                heap = H5HL_protect(ent->file, dxpl_id, heap_addr, H5AC_READ);
                 lval = H5HL_offset_into(ent->file, heap, ent->cache.slink.lval_offset);
                 HDfprintf(stream, "%*s%-*s %s\n", nested_indent, "", nested_fwidth,
                            "Link value:",
diff --git a/src/H5Glink.c b/src/H5Glink.c
index 8eb4329..d63a7ad 100644
--- a/src/H5Glink.c
+++ b/src/H5Glink.c
@@ -247,7 +247,7 @@ H5G_ent_to_link(H5F_t *f, hid_t dxpl_id, H5O_link_t *lnk, haddr_t lheap_addr,
         /* Check if the heap pointer was passed in */
         if(!heap) {
             /* Lock the local heap */
-            if(NULL == (heap = H5HL_protect(f, dxpl_id, lheap_addr)))
+            if(NULL == (heap = H5HL_protect(f, dxpl_id, lheap_addr, H5AC_READ)))
                 HGOTO_ERROR(H5E_SYM, H5E_NOTFOUND, FAIL, "unable to protect local heap")
         } /* end if */
 
@@ -314,7 +314,7 @@ H5G_ent_to_info(H5F_t *f, hid_t dxpl_id, H5L_info_t *info, haddr_t lheap_addr,
         H5HL_t *heap;           /* Pointer to local heap for group */
 
         /* Lock the local heap */
-        if(NULL == (heap = H5HL_protect(f, dxpl_id, lheap_addr)))
+        if(NULL == (heap = H5HL_protect(f, dxpl_id, lheap_addr, H5AC_READ)))
             HGOTO_ERROR(H5E_SYM, H5E_NOTFOUND, FAIL, "unable to protect local heap")
 
         s = H5HL_offset_into(f, heap, ent->cache.slink.lval_offset);
diff --git a/src/H5Gnode.c b/src/H5Gnode.c
index 02d57df..d37f8da 100644
--- a/src/H5Gnode.c
+++ b/src/H5Gnode.c
@@ -284,7 +284,7 @@ H5G_node_debug_key(FILE *stream, H5F_t *f, hid_t dxpl_id, int indent, int fwidth
 
         HDfprintf(stream, "%*s%-*s ", indent, "", fwidth, "Name:");
 
-        if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->heap_addr)))
+        if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->heap_addr, H5AC_READ)))
             HGOTO_ERROR(H5E_SYM, H5E_NOTFOUND, FAIL, "unable to protect symbol name")
 
         s = H5HL_offset_into(f, heap, key->offset);
@@ -794,7 +794,7 @@ H5G_node_cmp2(H5F_t *f, hid_t dxpl_id, void *_lt_key, void *_udata, void *_rt_ke
     HDassert(rt_key);
 
     /* Get base address of heap */
-    if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->heap_addr)))
+    if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->heap_addr, H5AC_READ)))
 	HGOTO_ERROR(H5E_SYM, H5E_NOTFOUND, FAIL, "unable to protect symbol name")
 
     base = H5HL_offset_into(f, heap, (size_t)0);
@@ -855,7 +855,7 @@ H5G_node_cmp3(H5F_t *f, hid_t dxpl_id, void *_lt_key, void *_udata, void *_rt_ke
     FUNC_ENTER_NOAPI_NOINIT(H5G_node_cmp3)
 
     /* Get base address of heap */
-    if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->heap_addr)))
+    if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->heap_addr, H5AC_READ)))
 	HGOTO_ERROR(H5E_SYM, H5E_NOTFOUND, FAIL, "unable to protect symbol name")
 
     base = H5HL_offset_into(f, heap, (size_t)0);
@@ -933,7 +933,7 @@ H5G_node_found(H5F_t *f, hid_t dxpl_id, haddr_t addr, const void UNUSED *_lt_key
 	HGOTO_ERROR(H5E_SYM, H5E_CANTLOAD, FAIL, "unable to protect symbol table node")
 
     /* Get base address of heap */
-    if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->common.heap_addr)))
+    if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->common.heap_addr, H5AC_READ)))
 	HGOTO_ERROR(H5E_SYM, H5E_NOTFOUND, FAIL, "unable to protect symbol name")
 
     base = H5HL_offset_into(f, heap, (size_t)0);
@@ -1046,7 +1046,7 @@ H5G_node_insert(H5F_t *f, hid_t dxpl_id, haddr_t addr,
 	HGOTO_ERROR(H5E_SYM, H5E_CANTLOAD, H5B_INS_ERROR, "unable to protect symbol table node")
 
     /* Get base address of heap */
-    if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->common.heap_addr)))
+    if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->common.heap_addr, H5AC_READ)))
 	HGOTO_ERROR(H5E_SYM, H5E_NOTFOUND, H5B_INS_ERROR, "unable to protect symbol name")
 
     base = H5HL_offset_into(f, heap, (size_t)0);
@@ -1228,7 +1228,7 @@ H5G_node_remove(H5F_t *f, hid_t dxpl_id, haddr_t addr, void *_lt_key/*in,out*/,
         const char *s;          /* Pointer to string in local heap */
 
         /* Lock the heap down */
-        if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->common.heap_addr)))
+        if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->common.heap_addr, H5AC_WRITE)))
             HGOTO_ERROR(H5E_SYM, H5E_CANTPROTECT, H5B_INS_ERROR, "unable to protect local heap")
 
         /* Get base address of heap */
@@ -1461,7 +1461,7 @@ H5G_node_iterate(H5F_t *f, hid_t dxpl_id, const void UNUSED *_lt_key, haddr_t ad
             size_t		n;      /* Length of link name */
             const char		*name;  /* Pointer to link name in heap */
 
-            if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->heap_addr)))
+            if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->heap_addr, H5AC_READ)))
                 HGOTO_ERROR(H5E_SYM, H5E_NOTFOUND, H5_ITER_ERROR, "unable to protect symbol name")
 
             name = H5HL_offset_into(f, heap, ents[u].name_off);
@@ -1816,7 +1816,7 @@ H5G_node_copy(H5F_t *f, hid_t dxpl_id, const void UNUSED *_lt_key, haddr_t addr,
 	HGOTO_ERROR(H5E_SYM, H5E_CANTLOAD, H5_ITER_ERROR, "unable to load symbol table node")
 
     /* get the base address of the heap */
-    if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->src_heap_addr)))
+    if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->src_heap_addr, H5AC_READ)))
        HGOTO_ERROR(H5E_SYM, H5E_NOTFOUND, H5_ITER_ERROR, "unable to protect symbol name")
 
     /* copy object in this node one by one */
@@ -1952,7 +1952,7 @@ H5G_node_build_table(H5F_t *f, hid_t dxpl_id, const void UNUSED *_lt_key, haddr_
 	HGOTO_ERROR(H5E_SYM, H5E_CANTLOAD, H5_ITER_ERROR, "unable to load symbol table node")
 
     /* Lock down the heap for this group */
-    if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->heap_addr)))
+    if(NULL == (heap = H5HL_protect(f, dxpl_id, udata->heap_addr, H5AC_READ)))
         HGOTO_ERROR(H5E_SYM, H5E_NOTFOUND, H5_ITER_ERROR, "unable to protect symbol name")
 
     /* Check if the link table needs to be extended */
@@ -2058,7 +2058,7 @@ H5G_node_debug(H5F_t *f, hid_t dxpl_id, haddr_t addr, FILE * stream, int indent,
 	fprintf(stream, "%*sSymbol %u:\n", indent - 3, "", u);
 
 	if(heap_addr > 0 && H5F_addr_defined(heap_addr)) {
-            if(NULL == (heap = H5HL_protect(f, dxpl_id, heap_addr)))
+            if(NULL == (heap = H5HL_protect(f, dxpl_id, heap_addr, H5AC_READ)))
                 HGOTO_ERROR(H5E_SYM, H5E_NOTFOUND, FAIL, "unable to protect symbol name");
 
             s = H5HL_offset_into(f, heap, sn->entry[u].name_off);
@@ -2066,7 +2066,7 @@ H5G_node_debug(H5F_t *f, hid_t dxpl_id, haddr_t addr, FILE * stream, int indent,
             if(s)
                 fprintf(stream, "%*s%-*s `%s'\n", indent, "", fwidth, "Name:", s);
 
-            if(H5HL_unprotect(f, dxpl_id, heap, heap_addr, FALSE) < 0)
+            if(H5HL_unprotect(f, dxpl_id, heap, heap_addr, H5AC__NO_FLAGS_SET) < 0)
                 HGOTO_ERROR(H5E_SYM, H5E_PROTECT, FAIL, "unable to unprotect symbol name")
             heap = NULL; s = NULL;
 	}
diff --git a/src/H5Gstab.c b/src/H5Gstab.c
index 431571b..7bb07aa 100644
--- a/src/H5Gstab.c
+++ b/src/H5Gstab.c
@@ -582,7 +582,7 @@ H5G_stab_get_name_by_idx_cb(const H5G_entry_t *ent, void *_udata)
     name_off = ent->name_off;
 
     /* Pin the heap down in memory */
-    if(NULL == (heap = H5HL_protect(udata->common.f, udata->common.dxpl_id, udata->heap_addr)))
+    if(NULL == (heap = H5HL_protect(udata->common.f, udata->common.dxpl_id, udata->heap_addr, H5AC_READ)))
         HGOTO_ERROR(H5E_SYM, H5E_NOTFOUND, FAIL, "unable to protect symbol name")
 
     name = H5HL_offset_into(udata->common.f, heap, name_off);
@@ -905,7 +905,7 @@ H5G_stab_lookup_by_idx_cb(const H5G_entry_t *ent, void *_udata)
     HDassert(udata);
 
     /* Pin the heap down in memory */
-    if(NULL == (heap = H5HL_protect(udata->common.f, udata->common.dxpl_id, udata->heap_addr)))
+    if(NULL == (heap = H5HL_protect(udata->common.f, udata->common.dxpl_id, udata->heap_addr, H5AC_READ)))
         HGOTO_ERROR(H5E_SYM, H5E_NOTFOUND, FAIL, "unable to protect symbol name")
 
     /* Duplicate the link name */
diff --git a/src/H5HFsection.c b/src/H5HFsection.c
index 3f59035..e031ed2 100644
--- a/src/H5HFsection.c
+++ b/src/H5HFsection.c
@@ -833,7 +833,7 @@ HDfprintf(stderr, "%s: hdr->man_dtable.curr_root_rows = %u\n", FUNC, hdr->man_dt
 #ifdef QAK
 HDfprintf(stderr, "%s: dblock_addr = %a\n", FUNC, dblock_addr);
 #endif /* QAK */
-        if(NULL == (dblock = H5HF_man_dblock_protect(hdr, dxpl_id, dblock_addr, dblock_size, sect->u.single.parent, sect->u.single.par_entry, H5AC_READ)))
+        if(NULL == (dblock = H5HF_man_dblock_protect(hdr, dxpl_id, dblock_addr, dblock_size, sect->u.single.parent, sect->u.single.par_entry, H5AC_WRITE)))
             HGOTO_ERROR(H5E_HEAP, H5E_CANTPROTECT, FAIL, "unable to load fractal heap direct block")
         HDassert(H5F_addr_eq(dblock->block_off + dblock_overhead, sect->sect_info.addr));
 
@@ -1186,7 +1186,7 @@ HDfprintf(stderr, "%s: (*sect).sect_info = {%a, %Hu, %u}\n", FUNC, (*sect)->sect
 HDfprintf(stderr, "%s: dblock_addr = %a\n", FUNC, dblock_addr);
 #endif /* QAK */
     if(NULL == (dblock = H5HF_man_dblock_protect(hdr, dxpl_id, dblock_addr, 
-            dblock_size, (*sect)->u.single.parent, (*sect)->u.single.par_entry, H5AC_READ)))
+            dblock_size, (*sect)->u.single.parent, (*sect)->u.single.par_entry, H5AC_WRITE)))
         HGOTO_ERROR(H5E_HEAP, H5E_CANTPROTECT, FAIL, "unable to load fractal heap direct block")
     HDassert(H5F_addr_eq(dblock->block_off + dblock_size, (*sect)->sect_info.addr + (*sect)->sect_info.size));
 
diff --git a/src/H5HL.c b/src/H5HL.c
index 7f0e638..44d7a53 100644
--- a/src/H5HL.c
+++ b/src/H5HL.c
@@ -813,7 +813,7 @@ done:
  *-------------------------------------------------------------------------
  */
 H5HL_t *
-H5HL_protect(H5F_t *f, hid_t dxpl_id, haddr_t addr)
+H5HL_protect(H5F_t *f, hid_t dxpl_id, haddr_t addr, H5AC_protect_t rw)
 {
     H5HL_t *ret_value;
 
@@ -823,7 +823,7 @@ H5HL_protect(H5F_t *f, hid_t dxpl_id, haddr_t addr)
     HDassert(f);
     HDassert(H5F_addr_defined(addr));
 
-    if(NULL == (ret_value = H5AC_protect(f, dxpl_id, H5AC_LHEAP, addr, NULL, NULL, H5AC_READ)))
+    if(NULL == (ret_value = H5AC_protect(f, dxpl_id, H5AC_LHEAP, addr, NULL, NULL, rw)))
         HGOTO_ERROR(H5E_HEAP, H5E_CANTLOAD, NULL, "unable to load heap")
 
 done:
diff --git a/src/H5HLprivate.h b/src/H5HLprivate.h
index 22786f3..eab55e3 100644
--- a/src/H5HLprivate.h
+++ b/src/H5HLprivate.h
@@ -33,6 +33,7 @@
 
 /* Private headers needed by this file. */
 #include "H5private.h"		/* Generic Functions			*/
+#include "H5ACprivate.h"	/* Metadata cache			*/
 #include "H5Fprivate.h"		/* File access				*/
 
 /*
@@ -63,7 +64,7 @@ typedef struct H5HL_t H5HL_t;
  * Library prototypes...
  */
 H5_DLL herr_t H5HL_create(H5F_t *f, hid_t dxpl_id, size_t size_hint, haddr_t *addr/*out*/);
-H5_DLL H5HL_t *H5HL_protect(H5F_t *f, hid_t dxpl_id, haddr_t addr);
+H5_DLL H5HL_t *H5HL_protect(H5F_t *f, hid_t dxpl_id, haddr_t addr, H5AC_protect_t rw);
 H5_DLL void *H5HL_offset_into(H5F_t *f, const H5HL_t *heap, size_t offset);
 H5_DLL herr_t H5HL_remove(H5F_t *f, hid_t dxpl_id, H5HL_t *heap, size_t offset,
     size_t size, unsigned *heap_flags);
diff --git a/src/H5Oefl.c b/src/H5Oefl.c
index 44efe0b..9c03b20 100644
--- a/src/H5Oefl.c
+++ b/src/H5Oefl.c
@@ -121,7 +121,7 @@ H5O_efl_decode(H5F_t *f, hid_t dxpl_id, unsigned UNUSED mesg_flags,
 #ifndef NDEBUG
     HDassert(H5F_addr_defined(mesg->heap_addr));
 
-    if(NULL == (heap = H5HL_protect(f, dxpl_id, mesg->heap_addr)))
+    if(NULL == (heap = H5HL_protect(f, dxpl_id, mesg->heap_addr, H5AC_READ)))
         HGOTO_ERROR(H5E_SYM, H5E_NOTFOUND, NULL, "unable to read protect link value")
 
     s = H5HL_offset_into(f, heap, 0);
@@ -138,7 +138,7 @@ H5O_efl_decode(H5F_t *f, hid_t dxpl_id, unsigned UNUSED mesg_flags,
     if(NULL == mesg->slot)
 	HGOTO_ERROR(H5E_RESOURCE, H5E_NOSPACE, NULL, "memory allocation failed")
 
-    if(NULL == (heap = H5HL_protect(f, dxpl_id, mesg->heap_addr)))
+    if(NULL == (heap = H5HL_protect(f, dxpl_id, mesg->heap_addr, H5AC_READ)))
         HGOTO_ERROR(H5E_SYM, H5E_NOTFOUND, NULL, "unable to read protect link value")
     for(u = 0; u < mesg->nused; u++) {
 	/* Name */