Using shared ptr to track iodesc use

src/clib/core/CMakeLists.txt

@@ -6,6 +6,7 @@ message(STATUS "===== Configuring SCORPIO C Core... =====")
 set (spio_core_src
   pio_spmd.cpp
   pioc_support.cpp
+  spio_iodesc_utils.cpp
   pioc.cpp
   pio_nc.cpp
   pio_getput_int.cpp

src/clib/core/iolib/hdf5/spio_async_hdf5_utils.cpp

@@ -79,16 +79,50 @@ struct Hdf5_wcache{
   int nvars;
   int fndims;
   std::vector<int> varids;
-  io_desc_t *iodesc;
+  std::shared_ptr<io_desc_t> iodesc;
   std::vector<int> frame;
 
   bool wr_fillbuf;
   void *iobuf;
   std::size_t iobuf_sz;
   void *fillbuf;
   std::size_t fillbuf_sz;
+
+  Hdf5_wcache(file_desc_t *file, int nvars, int fndims,
+    const int *varids, std::shared_ptr<io_desc_t> iodesc, bool wr_fillbuf, const int *frame);
+  ~Hdf5_wcache();
 };
 
+Hdf5_wcache::Hdf5_wcache(file_desc_t *file, int nvars, int fndims,
+  const int *varids, std::shared_ptr<io_desc_t> iodesc, bool wr_fillbuf, const int *frame):
+    file(file), nvars(nvars), fndims(fndims), iodesc(iodesc),
+    wr_fillbuf(wr_fillbuf), iobuf(NULL), iobuf_sz(0),
+    fillbuf(NULL), fillbuf_sz(0)
+{
+  assert(file && (nvars > 0) && (fndims > 0) && varids && iodesc);
+
+  /* Cache varids and frames (one frame, the frame being written,
+   * for each varid)
+   */
+  this->varids.resize(nvars);
+  std::copy(varids, varids + nvars, this->varids.begin());
+
+  if(frame){
+    this->frame.resize(nvars);
+    std::copy(frame, frame + nvars, this->frame.begin());
+  }
+
+  /* FIXME: Copy/init iobuf and fillbuf here */
+}
+
+Hdf5_wcache::~Hdf5_wcache()
+{
+  /* Don't delete cached iodesc ptr (we don't own it) */
+
+  if(iobuf) { brel(iobuf); iobuf = NULL; }
+  if(fillbuf) { brel(fillbuf); fillbuf = NULL; }
+}
+
 /* Global vars */
 std::atomic<int> SPIO_Util::GVars::npend_hdf5_async_ops;
 
@@ -566,7 +600,7 @@ int pio_iosys_async_op_hdf5_write(void *pdata)
   file_desc_t *file = wcache->file;
   int nvars = wcache->nvars;
   int fndims = wcache->fndims;
-  io_desc_t *iodesc = wcache->iodesc;
+  io_desc_t *iodesc = wcache->iodesc.get();
 
   assert(file && (nvars > 0) && (fndims > 0) && iodesc);
   assert((file->iotype == PIO_IOTYPE_HDF5) || (file->iotype == PIO_IOTYPE_HDF5C));
@@ -787,25 +821,15 @@ void pio_iosys_async_op_hdf5_write_free(void *pdata)
   Hdf5_wcache *wcache = static_cast<struct Hdf5_wcache *>(pdata);
   assert(wcache);
 
-  /* Using swap trick to free vectors
-   * - swap vector with an empty local/temp vector that gets deallocated when func exits
-   */
-  //wcache->varids.clear();
-  std::vector<int>().swap(wcache->varids);
-  //wcache->frame.clear();
-  std::vector<int>().swap(wcache->frame);
-
-  if(wcache->iobuf){ brel(wcache->iobuf); }
-  if(wcache->fillbuf){ brel(wcache->fillbuf); }
 
-  free(wcache);
+  delete(wcache);
 #else // _HDF5
   assert(0);
 #endif // _HDF5
 }
 
 int pio_iosys_async_hdf5_write_op_add(file_desc_t *file, int nvars, int fndims,
-      const int *varids, io_desc_t *iodesc, int fill, const int *frame)
+      const int *varids, std::shared_ptr<io_desc_t> iodesc, int fill, const int *frame)
 {
 #ifdef _HDF5
   int ret = PIO_NOERR;
@@ -819,15 +843,8 @@ int pio_iosys_async_hdf5_write_op_add(file_desc_t *file, int nvars, int fndims,
     return PIO_NOERR;
   }
 
-  std::vector<int> vids(varids, varids + nvars);
-  std::vector<int> frms;
-  if(frame){
-    frms.resize(nvars);
-    std::copy(frame, frame + nvars, frms.begin());
-  }
-
-  Hdf5_wcache *wcache = static_cast<Hdf5_wcache *>(calloc(1, sizeof(Hdf5_wcache)));
-  *wcache = {file, nvars, fndims, vids, iodesc, frms, (fill) ? true : false, NULL, 0, NULL, 0};
+  Hdf5_wcache *wcache = new Hdf5_wcache(file, nvars, fndims, varids, iodesc,
+    (fill) ? true : false, frame);
 
   /* We need to copy the iobuf/fillbuf since the mvcache gets reused for future writes */
   /* Copy iobuf/fillvalue */

src/clib/core/iolib/hdf5/spio_async_hdf5_utils.hpp

@@ -2,6 +2,7 @@
 #include "pio_config.h"
 #include "pio.h"
 #include "pio_internal.h"
+#include <memory>
 
 namespace SPIO_Util{
   namespace GVars{
@@ -23,7 +24,7 @@ int spio_wait_all_hdf5_async_ops(int iosysid);
 int pio_iosys_async_op_hdf5_write(void *pdata);
 void pio_iosys_async_op_hdf5_write_free(void *pdata);
 int pio_iosys_async_hdf5_write_op_add(file_desc_t *file, int nvars, int fndims,
-      const int *varids, io_desc_t *iodesc, int fill, const int *frame);
+      const int *varids, std::shared_ptr<io_desc_t> iodesc, int fill, const int *frame);
 int spio_iosys_async_hdf5_set_frame_op_add(file_desc_t *file, int varid, int frame);
 
 #define __SPIO_ASYNC_HDF5_UTILS_HPP__

src/clib/core/pio_darray.cpp

@@ -24,8 +24,10 @@
 #include "spio_decomp_logger.hpp"
 #include "spio_dt_converter.hpp"
 #include "spio_async_utils.hpp"
+#include "spio_iosys_utils.hpp"
 #include <string>
 #include <memory>
+#include <vector>
 #include <algorithm>
 
 /* uint64_t definition */
@@ -167,11 +169,23 @@ int PIOc_write_darray_multi_impl(int ncid, const int *varids, int ioid, int nvar
                     "Writing multiple variables to file (%s, ncid=%d) failed. Trying to write to a read only file, try reopening the file in write mode (use the PIO_WRITE flag)", pio_get_fname_from_file(file), ncid);
   }
 
-  /* Get iodesc. */
-  if(!(iodesc = pio_get_iodesc_from_id(ioid))){
+  std::shared_ptr<io_desc_t> sp_iodesc = nullptr;
+  /* First try to get cached iodesc. */
+  sp_iodesc = spio_get_iodesc_ref_from_file(file, std::vector<int>(varids, varids+nvars), ioid);
+  if(!sp_iodesc){
+    /* For some cases the I/O desc is not cached right now,
+     * 1) I/O desc is not cached for async I/O - since PIO_write_darray() is not offloaded
+     * 2) The unit tests that call this function directly (PIO_write_darray() is not called) also don't have the iodesc cached
+     * Try to get the iodesc from the global list
+     */
+    sp_iodesc = pio_get_iodesc_sptr_from_id(ioid);
+  }
+  if(!sp_iodesc){
     return pio_err(ios, file, PIO_EBADID, __FILE__, __LINE__,
                     "Writing multiple variables to file (%s, ncid=%d) failed. Invalid arguments, invalid PIO decomposition id (%d) provided", pio_get_fname_from_file(file), ncid, ioid);
   }
+  iodesc = sp_iodesc.get();
+
   pioassert(iodesc->rearranger == PIO_REARR_BOX || iodesc->rearranger == PIO_REARR_SUBSET || iodesc->rearranger == PIO_REARR_CONTIG,
             "unknown rearranger", __FILE__, __LINE__);
 
@@ -413,7 +427,7 @@ int PIOc_write_darray_multi_impl(int ncid, const int *varids, int ioid, int nvar
     case PIO_IOTYPE_HDF5:
     case PIO_IOTYPE_HDF5C:
 #if PIO_USE_ASYNC_WR_THREAD
-        ierr = pio_iosys_async_hdf5_write_op_add(file, nvars, fndims, varids, iodesc,
+        ierr = pio_iosys_async_hdf5_write_op_add(file, nvars, fndims, varids, sp_iodesc,
                                                   DARRAY_DATA, frame);
         if(ierr != PIO_NOERR){
           return pio_err(ios, file, ierr, __FILE__, __LINE__,
@@ -505,7 +519,7 @@ int PIOc_write_darray_multi_impl(int ncid, const int *varids, int ioid, int nvar
       case PIO_IOTYPE_HDF5:
       case PIO_IOTYPE_HDF5C:
   #if PIO_USE_ASYNC_WR_THREAD
-          ierr = pio_iosys_async_hdf5_write_op_add(file, nvars, fndims, varids, iodesc,
+          ierr = pio_iosys_async_hdf5_write_op_add(file, nvars, fndims, varids, sp_iodesc,
                                                     DARRAY_FILL, frame);
           if(ierr != PIO_NOERR){
             return pio_err(ios, file, ierr, __FILE__, __LINE__,
@@ -2398,6 +2412,9 @@ int PIOc_write_darray_impl(int ncid, int varid, int ioid, PIO_Offset arraylen, c
   }
   wmb->num_arrays++;
 
+  /* Cache a ref to the iodesc in the file */
+  spio_add_iodesc_ref_to_file(file, varid, ioid);
+
   LOG((2, "wmb->num_arrays = %d iodesc->maxbytes / iodesc->mpitype_size = %d "
        "iodesc->ndof = %d iodesc->llen = %d", wmb->num_arrays,
        iodesc->maxbytes / iodesc->mpitype_size, iodesc->ndof, iodesc->llen));

src/clib/core/pio_file.cpp

@@ -12,6 +12,8 @@
 #include <thread>
 #include <chrono>
 #include <string>
+#include <algorithm>
+#include <functional>
 #include "spio_hdf5_utils.hpp"
 #include "spio_async_tcomm.hpp"
 
@@ -477,6 +479,61 @@ int spio_wait_on_hard_close(iosystem_desc_t *ios, file_desc_t *file)
   return PIO_NOERR;
 }
 
+void spio_add_iodesc_ref_to_file(file_desc_t *file, int varid, int ioid)
+{
+  assert(file && file->pmtx && file->io_desc_refs);
+  assert((varid >= 0) && (ioid >= 0));
+
+  /* Get the lock before proceeding */
+  std::lock_guard<std::mutex> lg(*(file->pmtx));
+
+  /* Don't add duplicates, just one ref per variable to iodesc used by file.
+   * Note: A variable could however have multiple iodescs associated with it
+   */
+  std::shared_ptr<io_desc_t> iodesc = pio_get_iodesc_sptr_from_id(ioid);
+  /* ioid should be validated before calling this function. e.g. PIOc_write_darray() */
+  assert(iodesc);
+  (*(file->io_desc_refs))[varid][ioid] = iodesc;
+}
+
+std::shared_ptr<io_desc_t> spio_get_iodesc_ref_from_file(file_desc_t *file, const std::vector<int> &varids, int ioid)
+{
+  assert(file && file->pmtx && file->io_desc_refs);
+  assert((varids.size() > 0) && (ioid >= 0));
+
+  /* Get the lock before proceeding */
+  std::lock_guard<std::mutex> lg(*(file->pmtx));
+
+  /* Search for the cached I/O descs for the variable. The variable can have multiple I/O descs cached */
+  std::map<int, std::map<int, std::shared_ptr<io_desc_t> > >::iterator siter = file->io_desc_refs->find(varids[0]);
+
+  /* No I/O descs cached for this variable */
+  if(siter == file->io_desc_refs->end()) { return nullptr; }
+
+  /* Within cached I/O descs for the variable check if we have this I/O desc cached */
+  std::map<int, std::shared_ptr<io_desc_t> >::iterator iter = siter->second.find(ioid);
+
+  if(iter != siter->second.end()){
+    assert(iter->second->ioid == ioid);
+    /* The caller now has ownership of this iodesc */
+    std::shared_ptr<io_desc_t> sp = iter->second;
+    siter->second.erase(iter);
+    return sp;
+  }
+
+  /* When cached iodesc is used, for writing data, its used for all
+   * variables at once. So remove references to the I/O desc for all
+   * the remaining variables
+   */
+  for(std::vector<int>::const_iterator iter = varids.cbegin(); iter != varids.cend(); ++iter){
+    std::map<int, std::map<int, std::shared_ptr<io_desc_t> > >::iterator siter = file->io_desc_refs->find(*iter);
+    if(siter == file->io_desc_refs->end()) { continue; }
+    siter->second.erase(ioid);
+  }
+
+  return nullptr;
+}
+
 /* Close the file ("hard close")
  * @param ios: Pointer to the iosystem_desc
  * @param file: Pointer to the file_desc for the file