diff --git a/h5pyd/_apps/hsload.py b/h5pyd/_apps/hsload.py index 20f3e889..44acf639 100755 --- a/h5pyd/_apps/hsload.py +++ b/h5pyd/_apps/hsload.py @@ -123,6 +123,8 @@ def main(): help="use the given compression algorithm for -z option (lz4 is default)") cfg.setitem("ignorefilters", False, flags=["--ignore-filters"], help="ignore any filters used by source dataset") cfg.setitem("retries", 3, flags=["--retries",], choices=["N",], help="Set number of server retry attempts") + cfg.setitem("no_checks", False, flags=["--no-checks"], help="do not check for existence before creating resources") + cfg.setitem("thread_count", 30, flags=["--thread-count"], choices=["N",] ,help="The number of threads to allocate when making requests in parallel, defaults to 30") cfg.setitem("help", False, flags=["-h", "--help"], help="this message") try: @@ -258,6 +260,7 @@ def main(): "bucket": cfg["hs_bucket"], "mode": mode, "retries": int(cfg["retries"]), + "thread_count": int(cfg["thread_count"]), } fout = h5pyd.File(tgt, **kwargs) @@ -323,6 +326,8 @@ def main(): "extend_offset": cfg["extend_offset"], "ignore_error": cfg["ignore_error"], "no_clobber": no_clobber, + "no_checks": cfg["no_checks"], + "thread_count": int(cfg["thread_count"]), } load_file(fin, fout, **kwargs) diff --git a/h5pyd/_apps/utillib.py b/h5pyd/_apps/utillib.py index 576526f7..144646b5 100755 --- a/h5pyd/_apps/utillib.py +++ b/h5pyd/_apps/utillib.py @@ -12,6 +12,7 @@ import sys import logging +import concurrent.futures try: import h5py @@ -67,7 +68,7 @@ def dump_dtype(dt): def is_h5py(obj): # Return True if objref is a h5py object and False is not - if isinstance(obj, object) and isinstance(obj.id.id, int): + if isinstance(obj, object) and isinstance(obj.id.id, int): # type: ignore return True else: return False @@ -233,7 +234,7 @@ def convert_dtype(srcdt, ctx): tgt_dt = h5pyd.special_dtype(vlen=str) else: tgt_dt = srcdt - return tgt_dt + return tgt_dt # type: ignore def guess_chunk(shape, typesize): @@ -615,7 +616,7 @@ def get_chunktable_dims(dset): table_dims = [] for dim in range(rank): dset_extent = dset.shape[dim] - chunk_extent = chunk_dims[dim] + chunk_extent = chunk_dims[dim] # type: ignore if dset_extent > 0 and chunk_extent > 0: table_extent = -(dset_extent // -chunk_extent) @@ -838,7 +839,7 @@ def create_chunktable(dset, dset_dims, ctx): chunktable_maxshape = [None,] if extend else [] chunktable_maxshape.extend(get_chunktable_dims(dset)) chunk_dims = [1,] if extend else [] - chunk_dims.extend(get_chunk_dims(dset)) + chunk_dims.extend(get_chunk_dims(dset)) # type: ignore fout = ctx["fout"] kwargs = {} @@ -916,7 +917,7 @@ def create_chunktable(dset, dset_dims, ctx): chunk_key = "" for dim in range(rank): - chunk_key += str(index[dim] // chunk_dims[dim]) + chunk_key += str(index[dim] // chunk_dims[dim]) # type: ignore if dim < rank - 1: chunk_key += "_" logging.debug(f"adding chunk_key: {chunk_key}") @@ -1024,15 +1025,15 @@ def update_chunktable(src, tgt, ctx): chunktable = fout[f"datasets/{chunktable_id}"] chunk_arr = get_chunk_locations(src, ctx, include_file_uri=extend) - msg = f"dataset chunk dimensions {chunktable.shape} not compatible with {chunk_arr.shape}" - if len(chunktable.shape) == len(chunk_arr.shape): - if chunktable.shape != chunk_arr.shape: + msg = f"dataset chunk dimensions {chunktable.shape} not compatible with {chunk_arr.shape}" # type: ignore + if len(chunktable.shape) == len(chunk_arr.shape): # type: ignore + if chunktable.shape != chunk_arr.shape: # type: ignore logging.error(msg) if not ctx["ignore_error"]: raise IOError(msg) return - elif len(chunk_arr.shape) + 1 == len(chunktable.shape): - if chunk_arr.shape != chunktable.shape[1:]: + elif len(chunk_arr.shape) + 1 == len(chunktable.shape): # type: ignore + if chunk_arr.shape != chunktable.shape[1:]: # type: ignore logging.error(msg) return else: @@ -1069,7 +1070,7 @@ def update_chunktable(src, tgt, ctx): for i in range(len(chunk_indices)): index.append(int(chunk_indices[i])) index = tuple(index) - chunk_arr[index] = v + chunk_arr[index] = v # type: ignore elif src_layout_class == "H5D_CHUNKED_REF_INDIRECT": file_uri = src_layout["file_uri"] orig_chunktable_id = src_layout["chunk_table"] @@ -1088,7 +1089,7 @@ def update_chunktable(src, tgt, ctx): tgt_index = [0,] tgt_index.extend(it.multi_index) tgt_index = tuple(tgt_index) - chunk_arr[it.multi_index] = e + chunk_arr[it.multi_index] = e # type: ignore else: msg = f"expected chunk ref class but got: {src_layout_class}" logging.error(msg) @@ -1169,7 +1170,7 @@ def create_dataset(dobj, ctx): print(msg) fout = ctx["fout"] - if dobj.name in fout: + if not ctx["no_checks"] and dobj.name in fout: dset = fout[dobj.name] logging.debug(f"{dobj.name} already exists") if ctx["no_clobber"]: @@ -1244,7 +1245,7 @@ def create_dataset(dobj, ctx): rank = 0 else: tgt_shape.extend(dobj.shape) - tgt_maxshape.extend(dobj.maxshape) + tgt_maxshape.extend(dobj.maxshape) # type: ignore rank = len(tgt_shape) if rank > 0 and ctx["extend_dim"]: # set maxshape to unlimited for any dimension that is the extend_dim @@ -1259,8 +1260,8 @@ def create_dataset(dobj, ctx): if ctx["verbose"]: print(msg) for i in range(rank): - tgt_shape[i] = 0 - tgt_maxshape[i] = None + tgt_shape[i] = 0 # type: ignore + tgt_maxshape[i] = None # type: ignore else: # check to see if any dimension scale refers to the extend dim for dim in range(len(dobj.dims)): @@ -1272,8 +1273,8 @@ def create_dataset(dobj, ctx): msg = f"dimscale for dim: {dim}: {dimscale}, type: {type(dimscale)}" logging.debug(msg) if dimscale.name.split("/")[-1] == ctx["extend_dim"]: - tgt_shape[dim] = 0 - tgt_maxshape[dim] = None + tgt_shape[dim] = 0 # type: ignore + tgt_maxshape[dim] = None # type: ignore msg = f"setting dimension {dim} of dataset {dobj.name} to unlimited" logging.info(msg) if ctx["verbose"]: @@ -1596,7 +1597,7 @@ def write_dataset(src, tgt, ctx): if ctx["verbose"]: print(msg) except (IOError, TypeError) as e: - msg = f"ERROR : failed to copy dataset data {src_s}: {e}" + msg = f"ERROR : failed to copy dataset data {src_s}: {e}" # type: ignore logging.error(msg) if not ctx["ignore_error"]: raise @@ -1687,7 +1688,7 @@ def create_group(gobj, ctx): grp = None - if gobj.name in fout: + if not ctx["no_checks"] and gobj.name in fout: grp = fout[gobj.name] logging.debug(f"{gobj.name} already exists") if ctx["no_clobber"]: @@ -1708,7 +1709,7 @@ def create_group(gobj, ctx): if not ctx["ignore_error"]: raise IOError(msg) else: - if ctx["verbose"]: + if not ctx["no_checks"] and ctx["verbose"]: print(f"{gobj.name} not found") grp = fout.create_group(gobj.name) @@ -1799,6 +1800,8 @@ def load_file( extend_dim=None, extend_offset=0, ignore_error=False, + no_checks=False, + thread_count=30, ): logging.info(f"input file: {fin.filename}") @@ -1834,6 +1837,8 @@ def load_file( ctx["extend_offset"] = extend_offset ctx["srcid_desobj_map"] = {} ctx["ignore_error"] = ignore_error + ctx["no_checks"] = no_checks + ctx["thread_count"] = thread_count def copy_attribute_helper(name, obj): logging.info(f"copy attribute - name: {name} obj: {obj.name}") @@ -1875,6 +1880,24 @@ def object_copy_helper(name, obj): else: logging.error(f"no handler for object class: {type(obj)}") + def _visit_in_parallell(func): + logging.info("in parallell...") + jobs = [] + + def _add_to_jobs(name, obj): + jobs.append((name, obj)) + + with concurrent.futures.ThreadPoolExecutor(max_workers=thread_count) as executor: + fin.visititems(_add_to_jobs) + futures = [executor.submit(func, item[0], item[1]) for item in jobs] + + for future in concurrent.futures.as_completed(futures): + try: + future.result() + except Exception as e: + logging.exception(e) + raise + # build a rough map of the file using the internal function above # copy over any attributes # create soft/external links (and hardlinks not already created) @@ -1886,16 +1909,17 @@ def object_copy_helper(name, obj): # copy over any attributes logging.info("creating target attributes") - fin.visititems(copy_attribute_helper) + _visit_in_parallell(copy_attribute_helper) # create soft/external links (and hardlinks not already created) create_links(fin, fout, ctx) # create root soft/external links - fin.visititems(object_link_helper) + + _visit_in_parallell(object_link_helper) if dataload == "ingest" or dataload == "link": # copy dataset data logging.info("copying dataset data") - fin.visititems(object_copy_helper) + _visit_in_parallell(object_copy_helper) else: logging.info("skipping dataset data copy (dataload is None)") diff --git a/h5pyd/_hl/files.py b/h5pyd/_hl/files.py index c9d4be5a..cee73c09 100644 --- a/h5pyd/_hl/files.py +++ b/h5pyd/_hl/files.py @@ -279,6 +279,7 @@ def __init__( track_order=None, retries=10, timeout=180, + thread_count=30, **kwds, ): """Create a new file object. @@ -419,6 +420,7 @@ def __init__( logger=logger, retries=retries, timeout=timeout, + thread_count=thread_count, ) root_json = None diff --git a/h5pyd/_hl/folders.py b/h5pyd/_hl/folders.py index bfdfe672..e8bf9cb3 100644 --- a/h5pyd/_hl/folders.py +++ b/h5pyd/_hl/folders.py @@ -83,6 +83,7 @@ def __init__( owner=None, batch_size=1000, retries=3, + thread_count=30, **kwds, ): """Create a new Folders object. @@ -179,6 +180,7 @@ def __init__( mode=mode, logger=logger, retries=retries, + thread_count=thread_count ) self.log = self._http_conn.logging diff --git a/h5pyd/_hl/httpconn.py b/h5pyd/_hl/httpconn.py index 6745fb99..5d1b4824 100644 --- a/h5pyd/_hl/httpconn.py +++ b/h5pyd/_hl/httpconn.py @@ -166,6 +166,7 @@ def __init__( logger=None, retries=3, timeout=DEFAULT_TIMEOUT, + thread_count=30, **kwds, ): self._domain = domain_name @@ -179,6 +180,8 @@ def __init__( self._api_key = api_key self._s = None # Sessions self._server_info = None + self._thread_count = thread_count + if use_cache: self._cache = {} self._objdb = {} @@ -441,10 +444,12 @@ def GET(self, req, format="json", params=None, headers=None, use_cache=True): if check_cache: self.log.debug("httpcon - checking cache") - if req in self._cache: + try: self.log.debug("httpcon - returning cache result") rsp = self._cache[req] return rsp + except KeyError: + pass self.log.info(f"GET: {self._endpoint + req} [{params['domain']}] timeout: {self._timeout}") @@ -737,11 +742,11 @@ def session(self): s.mount( "http://", - HTTPAdapter(max_retries=retry, pool_connections=16, pool_maxsize=16), + HTTPAdapter(max_retries=retry, pool_connections=self._thread_count, pool_maxsize=self._thread_count), ) s.mount( "https://", - HTTPAdapter(max_retries=retry, pool_connections=16, pool_maxsize=16), + HTTPAdapter(max_retries=retry, pool_connections=self._thread_count, pool_maxsize=self._thread_count), ) self._s = s else: