[svn-r8875] Purpose:

    Bug fix

Description:
    Correct problems with "resurrecting" a dataset in a file.  (This occurs
when a dataset which is open gets unlinked from the group hierarchy (making it
"dead" and marked for deletion in the file) and then is re-linked to the group
hierarchy).  Note that the current solution applies only to datasets, further
work will fix this for groups and named datatypes also.
    Also, fix the "debug" routines to be a little more helpful in certain
situations.
    Additionally, fix a locking bug in the symbol table node splitting routine
which could be [one of] the cause[s] of the file corruption in flexible
parallel operation.

Platforms tested:
    FreeBSD 4.10 (sleipnir) w/parallel
    h5committested

1 files changed, 93 insertions, 70 deletions

diff --git a/src/H5O.c b/src/H5O.c
index 42a97a6..b7ef425 100644
--- a/src/H5O.c
+++ b/src/H5O.c
@@ -28,23 +28,25 @@
 #define H5F_PACKAGE		/*suppress error about including H5Fpkg	  */
 #define H5O_PACKAGE		/*suppress error about including H5Opkg	  */
 
-#include "H5private.h"
-#include "H5ACprivate.h"
-#include "H5Eprivate.h"
-#include "H5Fpkg.h"
-#include "H5FLprivate.h"	/*Free Lists	  */
-#include "H5Iprivate.h"
-#include "H5MFprivate.h"
-#include "H5MMprivate.h"
-#include "H5Opkg.h"             /* Object header functions                 */
-#include "H5Pprivate.h"
+/* Pablo information */
+/* (Put before include files to avoid problems with inline functions) */
+#define PABLO_MASK	H5O_mask
+
+#include "H5private.h"		/* Generic Functions			*/
+#include "H5ACprivate.h"	/* Metadata cache			*/
+#include "H5Eprivate.h"		/* Error handling		  	*/
+#include "H5Fpkg.h"             /* File access				*/
+#include "H5FLprivate.h"	/* Free lists                           */
+#include "H5Iprivate.h"		/* IDs			  		*/
+#include "H5MFprivate.h"	/* File memory management		*/
+#include "H5MMprivate.h"	/* Memory management			*/
+#include "H5Opkg.h"             /* Object headers			*/
+#include "H5Pprivate.h"		/* Property lists			*/
 
 #ifdef H5_HAVE_GETTIMEOFDAY
 #include <sys/time.h>
 #endif /* H5_HAVE_GETTIMEOFDAY */
 
-#define PABLO_MASK	H5O_mask
-
 /* PRIVATE PROTOTYPES */
 static herr_t H5O_init(H5F_t *f, hid_t dxpl_id, size_t size_hint,
                        H5G_entry_t *ent/*out*/, haddr_t header);
@@ -60,10 +62,10 @@ static herr_t H5O_share(H5F_t *f, hid_t dxpl_id, const H5O_class_t *type, const
 			 H5HG_t *hobj/*out*/);
 #endif /* NOT_YET */
 static unsigned H5O_find_in_ohdr(H5F_t *f, hid_t dxpl_id, H5O_t *oh,
-			     const H5O_class_t **type_p, int sequence);
+                                 const H5O_class_t **type_p, int sequence);
 static int H5O_modify_real(H5G_entry_t *ent, const H5O_class_t *type,
     int overwrite, unsigned flags, unsigned update_time, const void *mesg,
-hid_t dxpl_id);
+    hid_t dxpl_id);
 static int H5O_append_real(H5F_t *f, hid_t dxpl_id, H5O_t *oh,
     const H5O_class_t *type, unsigned flags, const void *mesg);
 static herr_t H5O_remove_real(H5G_entry_t *ent, const H5O_class_t *type,
@@ -140,13 +142,13 @@ static void *(*H5O_fast_g[H5G_NCACHED]) (const H5G_cache_t *,
 /* Declare a free list to manage the H5O_t struct */
 H5FL_DEFINE_STATIC(H5O_t);
 
-/* Declare a PQ free list to manage the H5O_mesg_t array information */
+/* Declare a free list to manage the H5O_mesg_t sequence information */
 H5FL_SEQ_DEFINE_STATIC(H5O_mesg_t);
 
-/* Declare a PQ free list to manage the H5O_chunk_t array information */
+/* Declare a free list to manage the H5O_chunk_t sequence information */
 H5FL_SEQ_DEFINE_STATIC(H5O_chunk_t);
 
-/* Declare a PQ free list to manage the chunk image information */
+/* Declare a free list to manage the chunk image information */
 H5FL_BLK_DEFINE_STATIC(chunk_image);
 
 /* Declare external the free list for time_t's */
@@ -231,7 +233,7 @@ H5O_create(H5F_t *f, hid_t dxpl_id, size_t size_hint, H5G_entry_t *ent/*out*/)
 
     /* initialize the object header */
     if (H5O_init(f, dxpl_id, size_hint, ent, header) != SUCCEED)
-	HGOTO_ERROR(H5E_OHDR, H5E_CANTINIT, FAIL, "unable to initialize object header");
+        HGOTO_ERROR(H5E_OHDR, H5E_CANTINIT, FAIL, "unable to initialize object header");
 
 done:
     FUNC_LEAVE_NOAPI(ret_value);
@@ -474,7 +476,6 @@ H5O_load(H5F_t *f, hid_t dxpl_id, haddr_t addr, const void UNUSED * _udata1,
     unsigned	chunkno;
     haddr_t	chunk_addr;
     size_t	chunk_size;
-    H5O_cont_t	*cont = NULL;
     uint8_t	flags;
 
     FUNC_ENTER_NOAPI(H5O_load, NULL);
@@ -521,7 +522,6 @@ H5O_load(H5F_t *f, hid_t dxpl_id, haddr_t addr, const void UNUSED * _udata1,
 
     /* read each chunk from disk */
     while (H5F_addr_defined(chunk_addr)) {
-
 	/* increase chunk array size */
 	if (oh->nchunks >= oh->alloc_nchunks) {
 	    unsigned na = oh->alloc_nchunks + H5O_NCHUNKS;
@@ -545,9 +545,7 @@ H5O_load(H5F_t *f, hid_t dxpl_id, haddr_t addr, const void UNUSED * _udata1,
 	    HGOTO_ERROR(H5E_OHDR, H5E_READERROR, NULL, "unable to read object header data");
 	
 	/* load messages from this chunk */
-	for (p = oh->chunk[chunkno].image;
-	     p < oh->chunk[chunkno].image + chunk_size;
-	     p += mesg_size) {
+	for (p = oh->chunk[chunkno].image; p < oh->chunk[chunkno].image + chunk_size; p += mesg_size) {
 	    UINT16DECODE(p, id);
 	    UINT16DECODE(p, mesg_size);
 	    assert (mesg_size==H5O_ALIGN (mesg_size));
@@ -583,22 +581,21 @@ H5O_load(H5F_t *f, hid_t dxpl_id, haddr_t addr, const void UNUSED * _udata1,
 		oh->mesg[mesgno].chunkno = chunkno;
 	    }
 	}
-	assert(p == oh->chunk[chunkno].image + chunk_size);
-
-	/* decode next object header continuation message */
-	for (chunk_addr=HADDR_UNDEF;
-	     !H5F_addr_defined(chunk_addr) && curmesg < oh->nmesgs;
-	     curmesg++) {
-	    if (H5O_CONT_ID == oh->mesg[curmesg].type->id) {
-		uint8_t *p2 = oh->mesg[curmesg].raw;
-
-		cont = (H5O_CONT->decode) (f, dxpl_id, p2, NULL);
-		oh->mesg[curmesg].native = cont;
-		chunk_addr = cont->addr;
-		chunk_size = cont->size;
-		cont->chunkno = oh->nchunks;	/*the next chunk to allocate */
-	    }
-	}
+
+        assert(p == oh->chunk[chunkno].image + chunk_size);
+
+        /* decode next object header continuation message */
+        for (chunk_addr = HADDR_UNDEF; !H5F_addr_defined(chunk_addr) && curmesg < oh->nmesgs; ++curmesg) {
+            if (H5O_CONT_ID == oh->mesg[curmesg].type->id) {
+                H5O_cont_t *cont;
+
+                cont = (H5O_CONT->decode) (f, dxpl_id, oh->mesg[curmesg].raw, NULL);
+                oh->mesg[curmesg].native = cont;
+                chunk_addr = cont->addr;
+                chunk_size = cont->size;
+                cont->chunkno = oh->nchunks;	/*the next chunk to allocate */
+            }
+        }
     }
 
     /* Set return value */
@@ -1190,7 +1187,7 @@ H5O_link(const H5G_entry_t *ent, int adjust, hid_t dxpl_id)
             /* Check if the object is still open by the user */
             if(H5FO_opened(ent->file,ent->header)>=0) {
                 /* Flag the object to be deleted when it's closed */
-                if(H5FO_mark(ent->file,ent->header)<0)
+                if(H5FO_mark(ent->file,ent->header,TRUE)<0)
                     HGOTO_ERROR(H5E_OHDR, H5E_CANTDELETE, FAIL, "can't mark object for deletion");
             } /* end if */
             else {
@@ -1203,6 +1200,16 @@ H5O_link(const H5G_entry_t *ent, int adjust, hid_t dxpl_id)
             } /* end else */
         } /* end if */
     } else if (adjust>0) {
+        /* A new object, or one that will be deleted */
+        if(oh->nlink==0) {
+            /* Check if the object is current open, but marked for deletion */
+            if(H5FO_marked(ent->file,ent->header)>0) {
+                /* Remove "delete me" flag on the object */
+                if(H5FO_mark(ent->file,ent->header,FALSE)<0)
+                    HGOTO_ERROR(H5E_OHDR, H5E_CANTDELETE, FAIL, "can't mark object for deletion");
+            } /* end if */
+        } /* end if */
+
 	oh->nlink += adjust;
 	oh->cache_info.dirty = TRUE;
     }
@@ -1211,7 +1218,7 @@ H5O_link(const H5G_entry_t *ent, int adjust, hid_t dxpl_id)
     ret_value = oh->nlink;
 
 done:
-    if (oh && H5AC_unprotect(ent->file, dxpl_id, H5AC_OHDR, ent->header, oh, deleted) < 0 && ret_value>=0)
+    if (oh && H5AC_unprotect(ent->file, dxpl_id, H5AC_OHDR, ent->header, oh, deleted) < 0)
 	HDONE_ERROR(H5E_OHDR, H5E_PROTECT, FAIL, "unable to release object header");
 
     FUNC_LEAVE_NOAPI(ret_value);
@@ -1298,18 +1305,21 @@ H5O_count_real (H5G_entry_t *ent, const H5O_class_t *type, hid_t dxpl_id)
     assert (type);
 
     /* Load the object header */
-    if (NULL==(oh=H5AC_find(ent->file, dxpl_id, H5AC_OHDR, ent->header, NULL, NULL)))
+    if (NULL == (oh = H5AC_protect(ent->file, dxpl_id, H5AC_OHDR, ent->header, NULL, NULL)))
 	HGOTO_ERROR (H5E_OHDR, H5E_CANTLOAD, FAIL, "unable to load object header");
 
-    for (u=acc=0; u<oh->nmesgs; u++) {
+    /* Loop over all messages, counting the ones of the type looked for */
+    for (u=acc=0; u<oh->nmesgs; u++)
 	if (oh->mesg[u].type==type)
             acc++;
-    }
 
     /* Set return value */
     ret_value=acc;
 
 done:
+    if (oh && H5AC_unprotect(ent->file, dxpl_id, H5AC_OHDR, ent->header, oh, FALSE) != SUCCEED)
+	HDONE_ERROR(H5E_OHDR, H5E_PROTECT, FAIL, "unable to release object header");
+
     FUNC_LEAVE_NOAPI(ret_value);
 } /* end H5O_count_real() */
 
@@ -1397,7 +1407,7 @@ H5O_exists_real(H5G_entry_t *ent, const H5O_class_t *type, int sequence, hid_t d
     assert(sequence>=0);
 
     /* Load the object header */
-    if (NULL==(oh=H5AC_find(ent->file, dxpl_id, H5AC_OHDR, ent->header, NULL, NULL)))
+    if (NULL == (oh = H5AC_protect(ent->file, dxpl_id, H5AC_OHDR, ent->header, NULL, NULL)))
 	HGOTO_ERROR(H5E_OHDR, H5E_CANTLOAD, FAIL, "unable to load object header");
 
     /* Scan through the messages looking for the right one */
@@ -1412,6 +1422,9 @@ H5O_exists_real(H5G_entry_t *ent, const H5O_class_t *type, int sequence, hid_t d
     ret_value=(sequence<0);
 
 done:
+    if (oh && H5AC_unprotect(ent->file, dxpl_id, H5AC_OHDR, ent->header, oh, FALSE) != SUCCEED)
+	HDONE_ERROR(H5E_OHDR, H5E_PROTECT, FAIL, "unable to release object header");
+
     FUNC_LEAVE_NOAPI(ret_value);
 } /* end H5O_exists_real() */
 
@@ -1493,17 +1506,22 @@ done:
  *		Aug  6 1997
  *
  * Modifications:
- *
+ *      Bill Wendling, 2003-09-30
+ *      Protect the object header and pass it into the H5O_find_in_ohdr
+ *      function. This is done because the H5O_find_in_ohdr used to
+ *      protect the ohdr, find the message, and then unprotect it. This
+ *      saves time and also helps the FPHDF5 stuff, where unprotecting
+ *      actually destroys the object in the cache.
  *-------------------------------------------------------------------------
  */
 void *
 H5O_read_real(H5G_entry_t *ent, const H5O_class_t *type, int sequence, void *mesg, hid_t dxpl_id)
 {
-    H5O_t		*oh = NULL;
-    int		idx;
-    H5G_cache_t		*cache = NULL;
-    H5G_type_t		cache_type;
-    void		*ret_value = NULL;
+    H5O_t          *oh = NULL;
+    int             idx;
+    H5G_cache_t    *cache = NULL;
+    H5G_type_t      cache_type;
+    void           *ret_value = NULL;
 
     FUNC_ENTER_NOAPI_NOINIT(H5O_read_real);
 
@@ -1516,6 +1534,7 @@ H5O_read_real(H5G_entry_t *ent, const H5O_class_t *type, int sequence, void *mes
 
     /* can we get it from the symbol table entry? */
     cache = H5G_ent_cache(ent, &cache_type);
+
     if (H5O_fast_g[cache_type]) {
 	ret_value = (H5O_fast_g[cache_type]) (cache, type, mesg);
 	if (ret_value)
@@ -1553,7 +1572,7 @@ H5O_read_real(H5G_entry_t *ent, const H5O_class_t *type, int sequence, void *mes
     }
 
 done:
-    if (oh && H5AC_unprotect(ent->file, dxpl_id, H5AC_OHDR, ent->header, oh, FALSE) < 0 && ret_value!=NULL)
+    if (oh && H5AC_unprotect(ent->file, dxpl_id, H5AC_OHDR, ent->header, oh, FALSE) < 0)
 	HDONE_ERROR(H5E_OHDR, H5E_PROTECT, NULL, "unable to release object header");
 
     FUNC_LEAVE_NOAPI(ret_value);
@@ -1563,25 +1582,27 @@ done:
 /*-------------------------------------------------------------------------
  * Function:	H5O_find_in_ohdr
  *
- * Purpose:	Find a message in the object header without consulting
- *		a symbol table entry.
+ * Purpose:     Find a message in the object header without consulting
+ *              a symbol table entry.
  *
- * Return:	Success:	Index number of message.
+ * Return:      Success:    Index number of message.
+ *              Failure:    Negative
  *
- *		Failure:	Negative
- *
- * Programmer:	Robb Matzke
- *		matzke@llnl.gov
- *		Aug  6 1997
+ * Programmer:  Robb Matzke
+ *              matzke@llnl.gov
+ *              Aug  6 1997
  *
  * Modifications:
- *		Robb Matzke, 1999-07-28
- *		The ADDR argument is passed by value.
+ *      Robb Matzke, 1999-07-28
+ *      The ADDR argument is passed by value.
+ *
+ *      Bill Wendling, 2003-09-30
+ *      Modified so that the object header needs to be AC_protected
+ *      before calling this function.
  *-------------------------------------------------------------------------
  */
 static unsigned
-H5O_find_in_ohdr(H5F_t *f, hid_t dxpl_id, H5O_t *oh, const H5O_class_t **type_p,
-		 int sequence)
+H5O_find_in_ohdr(H5F_t *f, hid_t dxpl_id, H5O_t *oh, const H5O_class_t **type_p, int sequence)
 {
     unsigned		u;
     const H5O_class_t	*type = NULL;
@@ -1601,6 +1622,7 @@ H5O_find_in_ohdr(H5F_t *f, hid_t dxpl_id, H5O_t *oh, const H5O_class_t **type_p,
 	if (--sequence < 0)
             break;
     }
+
     if (sequence >= 0)
 	HGOTO_ERROR(H5E_OHDR, H5E_NOTFOUND, UFAIL, "unable to find object header message");
 
@@ -1613,6 +1635,7 @@ H5O_find_in_ohdr(H5F_t *f, hid_t dxpl_id, H5O_t *oh, const H5O_class_t **type_p,
     } else {
 	type = oh->mesg[u].type;
     }
+
     if (NULL == oh->mesg[u].native) {
 	assert(type->decode);
 	oh->mesg[u].native = (type->decode) (f, dxpl_id, oh->mesg[u].raw, NULL);
@@ -1820,7 +1843,7 @@ H5O_modify_real(H5G_entry_t *ent, const H5O_class_t *type, int overwrite,
     ret_value = sequence;
 
 done:
-    if (oh && H5AC_unprotect(ent->file, dxpl_id, H5AC_OHDR, ent->header, oh, FALSE) < 0 && ret_value!=FAIL)
+    if (oh && H5AC_unprotect(ent->file, dxpl_id, H5AC_OHDR, ent->header, oh, FALSE) < 0)
 	HDONE_ERROR(H5E_OHDR, H5E_PROTECT, FAIL, "unable to release object header");
     
     FUNC_LEAVE_NOAPI(ret_value);
@@ -1903,7 +1926,7 @@ H5O_unprotect(H5G_entry_t *ent, H5O_t *oh, hid_t dxpl_id)
     assert(oh);
 
     if (H5AC_unprotect(ent->file, dxpl_id, H5AC_OHDR, ent->header, oh, FALSE) < 0)
-	HDONE_ERROR(H5E_OHDR, H5E_PROTECT, FAIL, "unable to release object header");
+	HGOTO_ERROR(H5E_OHDR, H5E_PROTECT, FAIL, "unable to release object header");
 
 done:
     FUNC_LEAVE_NOAPI(ret_value);
@@ -2243,7 +2266,7 @@ H5O_touch(H5G_entry_t *ent, hbool_t force, hid_t dxpl_id)
 	HGOTO_ERROR(H5E_OHDR, H5E_CANTINIT, FAIL, "unable to update object modificaton time");
 
 done:
-    if (oh && H5AC_unprotect(ent->file, dxpl_id, H5AC_OHDR, ent->header, oh, FALSE)<0 && ret_value>=0)
+    if (oh && H5AC_unprotect(ent->file, dxpl_id, H5AC_OHDR, ent->header, oh, FALSE)<0)
 	HDONE_ERROR(H5E_OHDR, H5E_PROTECT, FAIL, "unable to release object header");
 
     FUNC_LEAVE_NOAPI(ret_value);
@@ -2495,7 +2518,7 @@ H5O_remove_real(H5G_entry_t *ent, const H5O_class_t *type, int sequence, hid_t d
 	HGOTO_ERROR(H5E_OHDR, H5E_CANTINIT, FAIL, "unable to remove constant message(s)");
 
 done:
-    if (oh && H5AC_unprotect(ent->file, dxpl_id, H5AC_OHDR, ent->header, oh, FALSE) < 0 && ret_value>=0)
+    if (oh && H5AC_unprotect(ent->file, dxpl_id, H5AC_OHDR, ent->header, oh, FALSE) < 0)
 	HDONE_ERROR(H5E_OHDR, H5E_PROTECT, FAIL, "unable to release object header");
 
     FUNC_LEAVE_NOAPI(ret_value);
@@ -3135,7 +3158,7 @@ H5O_delete(H5F_t *f, hid_t dxpl_id, haddr_t addr)
         HGOTO_ERROR(H5E_OHDR, H5E_CANTDELETE, FAIL, "can't delete object from file");
 
 done:
-    if (oh && H5AC_unprotect(f, dxpl_id, H5AC_OHDR, addr, oh, TRUE)<0 && ret_value>=0)
+    if (oh && H5AC_unprotect(f, dxpl_id, H5AC_OHDR, addr, oh, TRUE)<0)
 	HDONE_ERROR(H5E_OHDR, H5E_PROTECT, FAIL, "unable to release object header");
 
     FUNC_LEAVE_NOAPI(ret_value);
@@ -3546,7 +3569,7 @@ H5O_debug(H5F_t *f, hid_t dxpl_id, haddr_t addr, FILE *stream, int indent, int f
 	HDfprintf(stream, "*** TOTAL SIZE DOES NOT MATCH ALLOCATED SIZE!\n");
 
 done:
-    if (oh && H5AC_unprotect(f, dxpl_id, H5AC_OHDR, addr, oh, FALSE) < 0 && ret_value>=0)
+    if (oh && H5AC_unprotect(f, dxpl_id, H5AC_OHDR, addr, oh, FALSE) < 0)
 	HDONE_ERROR(H5E_OHDR, H5E_PROTECT, FAIL, "unable to release object header");
 
     FUNC_LEAVE_NOAPI(ret_value);