Make hds-v5 thread safe

1) Guard global or static data using mutexes
2) Add object locking via new functions datLock, datUnlock and datLocked.

For HDS-V5 to be thread safe, HDF5 must be configured with --enable-threadsafe.

Makefile.am

@@ -106,6 +106,9 @@ datType.c \
 datUnmap.c \
 datValid.c \
 datVec.c \
+datLock.c \
+datLocked.c \
+datUnlock.c \
 hdsCopy.c \
 hdsErase.c \
 hdsEwild.c \
@@ -143,8 +146,10 @@ dat1CvtChar.c \
 dat1CvtLogical.c \
 dat1DumpLoc.c \
 dat1EncodeSubscript.c \
+dat1EraseHandle.c \
 dat1ExportDims.c \
 dat1FixNameCell.c \
+dat1FreeHandle.c \
 dat1FreeLoc.c \
 dat1GetAttr.c \
 dat1GetAttrBool.c \
@@ -157,6 +162,9 @@ dat1Getenv.c \
 dat1GetFullName.c \
 dat1GetParentID.c \
 dat1GetStructureDims.c \
+dat1Handle.c \
+dat1HandleLock.c \
+dat1HandleMsg.c \
 dat1H5EtoEMS.c \
 dat1ImportDims.c \
 dat1ImportFloc.c \

dat1.h

@@ -138,8 +138,31 @@ typedef enum {
 #define HDS__ATTR_ROOT_NAME "HDS_ROOT_NAME"
 #define HDS__ATTR_ROOT_PRIMITIVE "HDS_ROOT_IS_PRIMITIVE"
 
+/* This structure  contains information about an HDF5 object (group or
+   dataset) that is common to all the locators that refer to the object. */
+typedef struct Handle {
+   char locked;           /* Non-zero if the HDF object is currently locked for
+                             access by a single thread. Zero if the object is
+                             unlocked.  */
+   pthread_t locker;      /* Id for the thread that has locked the object (if
+                             any) */
+   struct Handle *parent; /* Pointer to Handle describing the parent object */
+   struct Handle **children; /* Pointer to array holding pointers to Handles
+                                for any known child objects */
+   int nchild;            /* The length of the "children" array */
+   char *name;            /* Name (cleaned) of the HDF object within its parent */
+   pthread_mutex_t mutex2;   /* Guards access to the values in the handle */
+
+} Handle;
+
+/* Define functions to lock and unlock a Handle's mutex. */
+#define DAT_LOCK_MUTEX2(han) pthread_mutex_lock( &((han)->mutex2) )
+#define DAT_UNLOCK_MUTEX2(han) pthread_mutex_unlock( &((han)->mutex2) )
+
+
 /* Private definition of the HDS locator struct */
 typedef struct LOC {
+  Handle *handle;    /* Structure holding fixed info for the HDF object */
   int hds_version;   /* Implementation version number. Always 5 at the moment. */
   void *pntr;        /* Pointer to memory mapped data array [datMap only] */
   void *regpntr;     /* Pointer that was registered with CNF */
@@ -345,6 +368,9 @@ hds1CountFiles();
 int
 hds1CountLocators( size_t ncomp, char **comps, hdsbool_t skip_scratch_root, int * status );
 
+Handle *
+hds1FindHandle( hid_t file_id, int *status );
+
 void
 dat1SetAttrString( hid_t obj_id, const char * attrname,
                    const char * value, int * status );
@@ -403,7 +429,12 @@ dat1GetStructureDims( const HDSLoc * locator, int maxdims, hdsdim dims[], int *s
 hdsbool_t
 dat1NeedsRootName( hid_t objid, hdsbool_t wantprim, char * rootname, size_t rootnamelen, int * status );
 
-int dat1ValidateLocator( const HDSLoc *loc, int *status );
+Handle *dat1Handle( const HDSLoc *parent_loc, const char *name, int * status );
+Handle *dat1EraseHandle( Handle *parent, const char *name, int * status );
+Handle *dat1FreeHandle( Handle *handle );
+int dat1ValidateLocator( int checklock, const HDSLoc *loc, int *status );
+int dat1HandleLock( Handle *handle, int oper, int recurs, int *status );
+void dat1HandleMsg( const char *token, const Handle *handle );
 
 /* DAT1_H_INCLUDED */
 #endif

dat1DumpLoc.c

@@ -73,19 +73,14 @@
 #include "sae_par.h"
 
 static void dump_dataspace_info( hid_t dataspace_id, const char * label, int *status);
+static void dump_handle( const HDSLoc *loc, int *status );
 
 void dat1DumpLoc( const HDSLoc* locator, int * status ) {
   char * name_str = NULL;
   char * file_str = NULL;
   ssize_t ll;
   hid_t objid = 0;
   hid_t dspace_id = 0;
-  hdsdim lower[DAT__MXDIM];
-  hdsdim upper[DAT__MXDIM];
-  hdsbool_t issubset;
-  int actdim;
-  hssize_t nelem;
-  int i;
 
   if (*status != SAI__OK) return;
 
@@ -107,7 +102,7 @@ void dat1DumpLoc( const HDSLoc* locator, int * status ) {
          (locator->uses_true_mmap ? "file" : "memory"));
   printf("- Is sliced: %d; Primary: %s; Group name: '%s'\n", locator->isslice,
          (locator->isprimary ? "yes" : "no"), locator->grpname);
-  printf("- Is a discontiguous slice: %d\n", (locator->isdiscont ? "yes" : "no") );
+  printf("- Is a discontiguous slice: %s\n", (locator->isdiscont ? "yes" : "no") );
 
   if (locator->dataspace_id > 0) {
     dump_dataspace_info( locator->dataspace_id, "Locator associated", status);
@@ -116,6 +111,8 @@ void dat1DumpLoc( const HDSLoc* locator, int * status ) {
     H5Sclose( dspace_id );
   }
 
+  dump_handle( locator, status );
+
   if (file_str) MEM_FREE(file_str);
   if (name_str) MEM_FREE(name_str);
   return;
@@ -189,3 +186,29 @@ static void dump_dataspace_info( hid_t dataspace_id, const char * label, int *st
   if (blockbuf) MEM_FREE(blockbuf);
   return;
 }
+
+static void dump_handle( const HDSLoc *loc, int *status ){
+   Handle *handle;
+   const char *name;
+
+   if( *status != SAI__OK || !loc ) return;
+
+   printf("Handle: ");
+
+   handle = loc->handle;
+   while( handle ) {
+      name = handle->name ? handle->name : "<>";
+      printf("%s(%p)", name, handle );
+      handle = handle->parent;
+      if( handle ) printf("." );
+   }
+
+   printf("\n" );
+
+}
+
+
+
+
+
+

dat1GetBounds.c

@@ -133,9 +133,7 @@ dat1GetBounds( const HDSLoc * locator, hdsdim lower[DAT__MXDIM],
 
     /* If we are using datSlice then there should be one (and only one) hyperslab
        for the dataspace and we need to handle that. Should be same dimensionality
-       as above. Negative number indicates there were no hyperslabs. We need to
-       check that there is a hyperslab first because H5Sget_select_hyper_nblocks 
-       will report an error if there is no hyperslab selection. */
+       as above. Negative number indicates there were no hyperslabs. */
     if( H5Sget_select_type( locator->dataspace_id ) == H5S_SEL_HYPERSLABS ) {
        nblocks = H5Sget_select_hyper_nblocks( locator->dataspace_id );
     } else {

dat1InitHDF5.c

@@ -77,16 +77,26 @@
 */
 
 #include "hdf5.h"
+#include <pthread.h>
+
+static pthread_mutex_t mutex1 = PTHREAD_MUTEX_INITIALIZER;
 
 void dat1InitHDF5(void) {
   static int DONE = 0;
 
-  if (DONE) return;
+  /* Serialise entry to this function by locking a mutex. */
+  pthread_mutex_lock( &mutex1 );
+
+  /* Return immediately if HDF5 has already been initialised for use with HDS. */
+  if(! DONE ) {
 
   /* Disable automatic printing of HDF5 errors */
-  H5Eset_auto( H5P_DEFAULT, NULL, NULL );
+     H5Eset_auto( H5P_DEFAULT, NULL, NULL );
 
+     DONE = 1;
+  }
 
-  DONE = 1;
+  /* Unlock the mutex so that any waiting threads can call this function. */
+  pthread_mutex_unlock( &mutex1 );
 
 }

dat1New.c

@@ -239,6 +239,7 @@ dat1New( const HDSLoc    *locator,
   if (*status == SAI__OK) {
     HDSLoc *thisloc = dat1AllocLoc( status );
     if (*status == SAI__OK) {
+      thisloc->handle = dat1Handle( locator, cleanname, status );
       thisloc->dataset_id = dataset_id;
       thisloc->group_id = group_id;
       thisloc->dataspace_id = dataspace_id;

dat1TypeInfo.c

@@ -76,6 +76,8 @@
 *-
 */
 
+#include <pthread.h>
+
 #include "hdf5.h"
 
 #include "ems.h"
@@ -86,12 +88,17 @@
 
 #include "prm_par.h"
 
+/* A mutex  used to serialis entry to this function so that multiple
+   threads do not try to acces the static data simultaneously. */
+static pthread_mutex_t mutex1 = PTHREAD_MUTEX_INITIALIZER;
+
 HdsTypeInfo *
 dat1TypeInfo( void ) {
 
   static int FILLED = 0;
   static HdsTypeInfo typeinfo;
 
+  pthread_mutex_lock( &mutex1 );
   if (!FILLED) {
     size_t i;
     unsigned char * ptr;
@@ -118,5 +125,7 @@ dat1TypeInfo( void ) {
 
     FILLED = 1;
   }
+  pthread_mutex_unlock( &mutex1 );
+
   return &typeinfo;
 }

dat1ValidateLocator.c

@@ -13,9 +13,13 @@
 *     Library routine
 
 *  Invocation:
-*     dat1ValidateLocator( const HDSLoc *loc, int * status );
+*     dat1ValidateLocator( int checklock, const HDSLoc *loc, int * status );
 
 *  Arguments:
+*     checklock = int (Given)
+*        If non-zero, an error is reported if the supplied locator is not
+*        locked by the current thread (see datLock). This check is not
+*        performed if "checklock" is zero.
 *     loc = HDSLoc * (Given)
 *        Locator to validate.
 *     status = int* (Given and Returned)
@@ -61,18 +65,20 @@
 *-
 */
 
+#include <pthread.h>
 #include "sae_par.h"
 #include "ems.h"
 #include "hds.h"
 #include "dat1.h"
 #include "dat_err.h"
 
-int dat1ValidateLocator( const HDSLoc *loc, int * status ) {
+int dat1ValidateLocator( int checklock, const HDSLoc *loc, int * status ) {
 
 /* Local Variables; */
    int valid;
 
-/* If the locator is not valid, report an error. */
+/* If the locator has been annulled (e.g. due to the container file being
+   closed when the last primary locator was annulled), report an error. */
    datValid( loc, &valid, status );
    if( !valid && *status == SAI__OK ) {
       *status = DAT__LOCIN;
@@ -81,6 +87,29 @@ int dat1ValidateLocator( const HDSLoc *loc, int * status ) {
              status );
    }
 
+/* Report an error if there is no handle in the locator. */
+   if( loc && !loc->handle && *status == SAI__OK ) {
+      *status = DAT__FATAL;
+      datMsg( "O", loc );
+      emsRep( " ", "The supplied HDS locator for '^O' has no handle (programming error).",
+              status );
+   }
+
+/* If required, check that the object is locked by the current thread.
+   Do not check any child objects as these wil be checked if and when
+   accessed. */
+   if( checklock && *status == SAI__OK ) {
+      if( dat1HandleLock( loc->handle, 1, 0, status ) != 1 &&
+          *status == SAI__OK ) {
+         *status = DAT__THREAD;
+         datMsg( "O", loc );
+         emsRep( " ", "The supplied HDS locator for '^O' cannot be used.",
+                 status );
+         emsRep( " ", "It has not been locked for use by the current thread "
+                 "(programming error).", status );
+      }
+   }
+
    return *status;
 
 }

datAlter.c

@@ -126,7 +126,7 @@ datAlter( HDSLoc *locator, int ndim, const hdsdim dims[], int *status) {
   if (*status != SAI__OK) return *status;
 
   /* Validate input locator. */
-  dat1ValidateLocator( locator, status );
+  dat1ValidateLocator( 1, locator, status );
 
   if (locator->vectorized) {
     *status = DAT__OBJIN;

datBasic.c

@@ -105,7 +105,7 @@ datBasic(const HDSLoc *locator, const char *mode_c, unsigned char **pntr,
   if (*status == SAI__OK) return *status;
 
   /* Validate input locator. */
-  dat1ValidateLocator( locator, status );
+  dat1ValidateLocator( 1, locator, status );
 
   *status = SAI__ERROR;
   emsRep("datBasic", "datBasic is not available in HDF5 interface",

datCcopy.c

@@ -105,8 +105,8 @@ datCcopy(const HDSLoc *locator1, const HDSLoc *locator2, const char *name,
   if (*status != SAI__OK) return *status;
 
   /* Validate input locators. */
-  dat1ValidateLocator( locator1, status );
-  dat1ValidateLocator( locator2, status );
+  dat1ValidateLocator( 1, locator1, status );
+  dat1ValidateLocator( 1, locator2, status );
 
   if (dat1IsStructure(locator1, status)) {
 

datCell.c

@@ -112,7 +112,7 @@ datCell(const HDSLoc *locator1, int ndim, const hdsdim subs[],
   if (*status != SAI__OK) return *status;
 
   /* Validate input locator. */
-  dat1ValidateLocator( locator1, status );
+  dat1ValidateLocator( 1, locator1, status );
 
   datName(locator1, namestr, status );
 
@@ -203,6 +203,9 @@ datCell(const HDSLoc *locator1, int ndim, const hdsdim subs[],
     thisloc = dat1AllocLoc( status );
 
     if (*status == SAI__OK) {
+      /* Handle for cell's data object */
+      thisloc->handle = dat1Handle( locator1, cellname, status );
+
       thisloc->group_id = group_id;
       /* Secondary locator by definition */
       thisloc->file_id = locator1->file_id;

datClen.c

@@ -116,7 +116,7 @@ datClen( const HDSLoc * locator, size_t * clen, int * status ) {
   if (*status != SAI__OK) return *status;
 
   /* Validate input locator. */
-  dat1ValidateLocator( locator, status );
+  dat1ValidateLocator( 1, locator, status );
 
   if (locator->dataset_id <= 0) {
     *status = DAT__OBJIN;

datClone.c

@@ -106,7 +106,7 @@ datClone(const HDSLoc *locator1, HDSLoc **locator2, int *status) {
   if (*status != SAI__OK) return *status;
 
   /* Validate input locator. */
-  dat1ValidateLocator( locator1, status );
+  dat1ValidateLocator( 1, locator1, status );
 
   clonedloc = dat1AllocLoc( status );
   if (*status != SAI__OK) goto CLEANUP;
@@ -146,6 +146,7 @@ datClone(const HDSLoc *locator1, HDSLoc **locator2, int *status) {
   clonedloc->isslice = locator1->isslice;
   clonedloc->iscell = locator1->iscell;
   clonedloc->isdiscont = locator1->isdiscont;
+  clonedloc->handle = locator1->handle;
 
  CLEANUP:
   if (*status != SAI__OK) {

datCoerc.c

@@ -106,7 +106,7 @@ int datCoerc( const HDSLoc *locator1, int ndim, HDSLoc **locator2, int *status)
   if (*status != SAI__OK) return *status;
 
   /* Validate input locator. */
-  dat1ValidateLocator( locator1, status );
+  dat1ValidateLocator( 1, locator1, status );
 
   *status = DAT__FATAL;
   emsRep("datCoerc", "datCoerc: Not yet implemented for HDF5",

datConv.c

@@ -102,7 +102,7 @@ datConv(const HDSLoc *locator, const char *type_str, hdsbool_t *conv,
   if (*status != SAI__OK) return *status;
 
   /* Validate input locator. */
-  dat1ValidateLocator( locator, status );
+  dat1ValidateLocator( 1, locator, status );
 
   *status = DAT__FATAL;
   emsRep("datConv", "datConv: Obsolete routine. Do not use",