Chunkinfo

hsds/chunk_dn.py

@@ -366,8 +366,8 @@ async def GET_Chunk(request):
             num_bytes = s3size
         else:
             # list
-            num_bytes = np.prod(s3size)
-        log.debug(f"reading {num_bytes} from {s3path}")
+            num_bytes = np.sum(s3size)
+        log.debug(f"reading {num_bytes} bytes from {s3path}")
         if num_bytes == 0:
             log.warn(f"GET_Chunk for s3path: {s3path} with empty byte range")
             raise HTTPNotFound()

hsds/chunk_sn.py

@@ -279,19 +279,21 @@ def getChunkItem(chunkid):
         default_chunktable_dims = get_chunktable_dims(dims, chunk_dims)
         log.debug(f"default_chunktable_dims: {default_chunktable_dims}")
         table_factors = []
-        hyper_dims = []
+        if "hyper_dims" in layout:
+            hyper_dims = layout["hyper_dims"]
+        else:
+            # assume 1 to 1 matching
+            hyper_dims = chunk_dims
         ref_num_chunks = num_chunks
         for dim in range(rank):
-            if chunktable_dims[dim] % default_chunktable_dims[dim] != 0:
-                msg = f"expected chunktable shape[{dim}] to be a factor"
-                msg += f" of {default_chunktable_dims[dim]}"
+            if chunk_dims[dim] % hyper_dims[dim] != 0:
+                msg = f"expected hyper_dims [{hyper_dims[dim]}] to be a factor"
+                msg += f" of {chunk_dims[dim]}"
                 log.warn(msg)
                 raise HTTPBadRequest(reason=msg)
-            factor = chunktable_dims[dim] // default_chunktable_dims[dim]
+            factor = chunk_dims[dim] // hyper_dims[dim]
             table_factors.append(factor)
             ref_num_chunks *= factor
-            hyper_dim = chunk_dims[dim] // factor
-            hyper_dims.append(hyper_dim)
         log.debug(f"table_factors: {table_factors}")
         log.debug(f"ref_num_chunks: {ref_num_chunks}")
         log.debug(f"hyper_dims: {hyper_dims}")

hsds/datanode_lib.py

@@ -853,7 +853,9 @@ async def get_chunk_bytes(
     # create a buffer for the hsds chunk and arrange h5 chunks within it
     chunk_size = np.prod(chunk_dims) * item_size
     # number of bytes in the hd5 chunk
+    # hyper_dims = [4000,]  # test
     h5_size = np.prod(hyper_dims) * item_size
+    log.debug(f"h5 chunk size: {h5_size}")
     chunk_bytes = bytearray(chunk_size)
     if num_chunks > chunk_size // h5_size:
         # shouldn't have more than this many hyperchunks
@@ -874,27 +876,29 @@ async def get_chunk_bytes(
             "length": length[i],
             "bucket": bucket
         }
-        log.debug(f"get_chunk_bytes - h5_chunk[{i}, offset: {offset[i]}, length: {length[i]}")
+        log.debug(f"get_chunk_bytes - h5_chunk[{i}], offset: {offset[i]}, length: {length[i]}")
         tasks.append(getStorBytes(app, s3key, **kwargs))
 
     log.debug(f"running asyncio.gather on {len(tasks)} tasks")
     results = await asyncio.gather(*tasks)
-    log.debug(f"asyncio.gather got results: {results}")
+    log.debug(f"asyncio.gather got {len(results)} results")
     if len(results) != num_chunks:
         log.error("unexpected number of gather results")
         raise HTTPInternalServerError()
     for i in range(num_chunks):
-        h5_chunk = results[i]
-        if h5_chunk is None:
+        h5_chunk_bytes = results[i]
+        if h5_chunk_bytes is None:
             log.warning(f"get_chunk_bytes - None returned for h5_chunk[{i}]")
             continue
-        if len(h5_chunk) != h5_size:
-            msg = f"get_chunk_bytes - got {len(h5_chunk)} bytes for h5_chunk[{i}], "
+
+        if len(h5_chunk_bytes) != h5_size:
+            msg = f"get_chunk_bytes - got {len(h5_chunk_bytes)} bytes for h5_chunk[{i}], "
             msg += f"expected: {h5_size}"
             log.error(msg)
             continue
         pos = h5_size * i
-        chunk_bytes[pos:(pos + h5_size)] = h5_chunk
+        chunk_bytes[pos:(pos + h5_size)] = h5_chunk_bytes
+        log.debug(f"setting chunk_bytes[{pos}:{(pos+h5_size)}]")
 
     """
     # serial version

hsds/util/chunkUtil.py

@@ -1170,3 +1170,35 @@ def chunkQuery(
         log.debug(f"   {i}: {rsp_arr[i]}")
 
     return rsp_arr
+
+def _find_min_pair(h5chunks, max_gap=None):
+    """ given a dict of chunk_map entries, return the two
+        chunks nearest to each other in the file.
+        If max_gap is set, chunms must be within max_gap bytes   """
+    if len(h5chunks) < 2:
+        return None
+    chunk_indices = list(h5chunks.keys())
+    min_pair = None
+    min_gap = None
+    for index_left in chunk_indices:
+        for index_right in chunk_indices:
+            if index_left == index_right:
+                continue
+            chunk_left = h5chunks[index_left]
+            chunk_right = h5chunks[index_right]
+            if chunk_left["offset"] > chunk_right["offset"]:
+                continue
+            gap = chunk_right["offset"] - (chunk_left["offset"] + chunk_left["length"])
+            if gap == 0:
+                # these two are contiguous
+                return (index_left, index_right)
+            if max_gap is not None  and gap > max_gap:
+                # too far apart
+                continue
+            if min_gap is None or gap < min_gap:
+                min_gap = gap
+                min_pair = (index_left, index_right)
+    return min_pair
+
+
+

tests/integ/link_test.py

@@ -370,7 +370,6 @@ def testGetLinks(self):
                 #
 
             self.assertEqual(ret_names, expected_names)
-            print(expected_names)
 
             # get links with a result limit of 4
             limit = 4

tests/integ/value_test.py

@@ -2998,7 +2998,8 @@ def testIntelligentRangeGet(self):
             "class": "H5D_CHUNKED_REF_INDIRECT",
             "file_uri": file_uri,
             "dims": chunk_dims,
-            "chunk_table": chunkinfo_uuid,
+            "hyper_dims": [chunk_extent,],
+            "chunk_table": chunkinfo_uuid
         }
         # the linked dataset uses gzip, so set it here
         gzip_filter = {