diff --git a/Readme.md b/Readme.md index 41de28dd..e59e93f5 100644 --- a/Readme.md +++ b/Readme.md @@ -1,5 +1,5 @@ # netchdf -_last updated: 7/22/2025_ +_last updated: 7/25/2025_ This is a rewrite in Kotlin of parts of the devcdm and netcdf-java libraries. @@ -133,7 +133,7 @@ For HDF5 files using deflate filters, the deflate library dominates the read tim are about 2X slower than native code. Unless the deflate libraries get better, there's not much gain in trying to make other parts of the code faster. -We will investigate using Kotlin coroutines to speed up performance bottlenecks. +We are seeing 10x speedup on data reading. see https://github.com/JohnLCaron/netchdf/issues/189. ### Goals and scope diff --git a/core/src/commonMain/kotlin/com/sunya/cdm/api/Netchdf.kt b/core/src/commonMain/kotlin/com/sunya/cdm/api/Netchdf.kt index 7dcb50b3..8806c1b2 100644 --- a/core/src/commonMain/kotlin/com/sunya/cdm/api/Netchdf.kt +++ b/core/src/commonMain/kotlin/com/sunya/cdm/api/Netchdf.kt @@ -20,8 +20,9 @@ interface Netchdf : AutoCloseable { // iterate over all the chunks in section, order is arbitrary. TODO where is intersection with wantSection done ?? fun chunkIterator(v2: Variable, wantSection: SectionPartial? = null, maxElements : Int? = null) : Iterator> + // iterate over all the chunks in section, order is arbitrary, callbacks are in multiple threads. fun readChunksConcurrent(v2: Variable, - lamda : (ArraySection<*>) -> Unit, + lamda : (ArraySection) -> Unit, done : () -> Unit, wantSection: SectionPartial? = null, nthreads: Int? = null) { @@ -30,9 +31,9 @@ interface Netchdf : AutoCloseable { } // the section describes the array chunk reletive to the variable's shape. -data class ArraySection(val array : ArrayTyped, val section : Section) { +data class ArraySection(val array : ArrayTyped, val chunkSection : Section) { fun intersect(wantSection: SectionPartial) : ArrayTyped { - // TODO + // TODO ?? return array } } \ No newline at end of file diff --git a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayByte.kt b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayByte.kt index a0dc0e6d..a7aac2ef 100644 --- a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayByte.kt +++ b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayByte.kt @@ -37,4 +37,10 @@ class ArrayByte(shape : IntArray, val values: ByteArray) : ArrayTyped(Data return dst } + override fun transfer(dst: Any, tc: TransferChunk) { + val src = this.values + val dest = dst as ByteArray + repeat(tc.nelems) { dest[tc.destElem.toInt()+it] = src[tc.srcElem.toInt() + it] } + } + } \ No newline at end of file diff --git a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayDouble.kt b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayDouble.kt index 0be691c0..be6f7333 100644 --- a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayDouble.kt +++ b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayDouble.kt @@ -36,4 +36,10 @@ class ArrayDouble(shape : IntArray, val values: DoubleArray) : ArrayTyped(D return dst } + override fun transfer(dst: Any, tc: TransferChunk) { + val src = this.values + val dest = dst as FloatArray + repeat(tc.nelems) { dest[tc.destElem.toInt()+it] = src[tc.srcElem.toInt() + it] } + } + } \ No newline at end of file diff --git a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayInt.kt b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayInt.kt index 59b89058..e2cd4987 100644 --- a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayInt.kt +++ b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayInt.kt @@ -36,4 +36,10 @@ class ArrayInt(shape : IntArray, val values: IntArray) : ArrayTyped(Datatyp } return dst } + + override fun transfer(dst: Any, tc: TransferChunk) { + val src = this.values + val dest = dst as IntArray + repeat(tc.nelems) { dest[tc.destElem.toInt()+it] = src[tc.srcElem.toInt() + it] } + } } diff --git a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayLong.kt b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayLong.kt index e824590d..a664fb6f 100644 --- a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayLong.kt +++ b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayLong.kt @@ -36,4 +36,10 @@ class ArrayLong(shape : IntArray, val values: LongArray) : ArrayTyped(Data } return dst } + + override fun transfer(dst: Any, tc: TransferChunk) { + val src = this.values + val dest = dst as LongArray + repeat(tc.nelems) { dest[tc.destElem.toInt()+it] = src[tc.srcElem.toInt() + it] } + } } \ No newline at end of file diff --git a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayOpaque.kt b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayOpaque.kt index 17155d01..f5f24710 100644 --- a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayOpaque.kt +++ b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayOpaque.kt @@ -75,5 +75,10 @@ class ArrayOpaque(shape : IntArray, val values : List, val size : Int } } + override fun transfer(dst: Any, tc: TransferChunk) { + val src = this.values + val dest = dst as MutableList + repeat(tc.nelems) { dest[tc.destElem.toInt()+it] = src[tc.srcElem.toInt() + it] } + } } \ No newline at end of file diff --git a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayShort.kt b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayShort.kt index d1cf8807..212115ed 100644 --- a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayShort.kt +++ b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayShort.kt @@ -36,4 +36,10 @@ class ArrayShort(shape : IntArray, val values: ShortArray) : ArrayTyped(D } return dst } + + override fun transfer(dst: Any, tc: TransferChunk) { + val src = this.values + val dest = dst as ShortArray + repeat(tc.nelems) { dest[tc.destElem.toInt()+it] = src[tc.srcElem.toInt() + it] } + } } \ No newline at end of file diff --git a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayString.kt b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayString.kt index df96267b..c5d8f174 100644 --- a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayString.kt +++ b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayString.kt @@ -8,6 +8,7 @@ import com.fleeksoft.charset.decodeToString import com.sunya.cdm.api.* import com.sunya.cdm.layout.IndexND import com.sunya.cdm.layout.IndexSpace +import com.sunya.cdm.layout.TransferChunk // fake ByteBuffer class ArrayString(shape : IntArray, val values : List) : ArrayTyped(Datatype.STRING, shape) { @@ -37,6 +38,12 @@ class ArrayString(shape : IntArray, val values : List) : ArrayTyped + repeat(tc.nelems) { dest[tc.destElem.toInt()+it] = src[tc.srcElem.toInt() + it] } + } } /** diff --git a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayStructureData.kt b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayStructureData.kt index 6f561d24..12f4dc1a 100644 --- a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayStructureData.kt +++ b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayStructureData.kt @@ -3,6 +3,7 @@ package com.sunya.cdm.array import com.sunya.cdm.api.* import com.sunya.cdm.layout.Chunker import com.sunya.cdm.layout.IndexSpace +import com.sunya.cdm.layout.TransferChunk // fixed length data in the ByteBuffer, var length data goes on the heap class ArrayStructureData(shape : IntArray, val ba : ByteArray, val isBE: Boolean, val recsize : Int, val members : List>) @@ -63,6 +64,10 @@ class ArrayStructureData(shape : IntArray, val ba : ByteArray, val isBE: Boolean return ArrayStructureData(section.shape.toIntArray(), sectionBA, isBE, recsize, members) } + override fun transfer(dst: Any, tc: TransferChunk) { + TODO() // maybe nobody chunks structuredata? + } + // structure data is packed into the ByteBuffer starting at the given offset // vlens and strings are on the "heap" stored in the parent ArrayStructureData inner class StructureData(val ba: ByteArray, val offset: Int, val members: List>) { diff --git a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayTyped.kt b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayTyped.kt index d2b8e582..3aed9399 100644 --- a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayTyped.kt +++ b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayTyped.kt @@ -1,6 +1,7 @@ package com.sunya.cdm.array import com.sunya.cdm.api.* +import com.sunya.cdm.layout.TransferChunk // here, shape must be integers, since size cant exceed 32 bits // TODO ArrayTyped is Iterable, but Datatype doesnt have to be T @@ -46,6 +47,8 @@ abstract class ArrayTyped(val datatype: Datatype<*>, val shape: IntArray) : I return dst } */ + abstract fun transfer(dst: Any, tc: TransferChunk) + override fun equals(other: Any?): Boolean { if (this === other) return true if (other !is ArrayTyped<*>) return false @@ -122,6 +125,10 @@ class ArraySingle(shape : IntArray, datatype : Datatype<*>, fillValueAny : An override fun section(section : Section) : ArrayTyped { return ArraySingle(section.shape.toIntArray(), datatype, fillValue as Any) } + + override fun transfer(dst: Any, tc: TransferChunk) { + // hmmmm could be trouble + } } // An empty array of any shape that has no values @@ -130,6 +137,8 @@ class ArrayEmpty(shape : IntArray, datatype : Datatype<*>) : ArrayTyped(da override fun section(section : Section) : ArrayTyped { return ArrayEmpty(section.shape.toIntArray(), datatype) } + + override fun transfer(dst: Any, tc: TransferChunk) {} } diff --git a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayUByte.kt b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayUByte.kt index 7e7b30c8..6b729b60 100644 --- a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayUByte.kt +++ b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayUByte.kt @@ -41,6 +41,12 @@ class ArrayUByte(shape: IntArray, datatype: Datatype<*>, val values: UByteArray) return dst } + override fun transfer(dst: Any, tc: TransferChunk) { + val src = this.values + val dest = dst as UByteArray + repeat(tc.nelems) { dest[tc.destElem.toInt()+it] = src[tc.srcElem.toInt() + it] } + } + companion object { fun fromByteArray(shape: IntArray, values: ByteArray): ArrayUByte = ArrayUByte(shape, UByteArray(values.size) { values[it].toUByte() }) diff --git a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayUInt.kt b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayUInt.kt index 151e5175..4226c86e 100644 --- a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayUInt.kt +++ b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayUInt.kt @@ -40,6 +40,12 @@ class ArrayUInt(shape : IntArray, datatype : Datatype<*>, val values: UIntArray) return dst } + override fun transfer(dst: Any, tc: TransferChunk) { + val src = this.values + val dest = dst as UIntArray + repeat(tc.nelems) { dest[tc.destElem.toInt()+it] = src[tc.srcElem.toInt() + it] } + } + companion object { fun fromIntArray(shape : IntArray, values : IntArray): ArrayUInt = ArrayUInt(shape, UIntArray(values.size) { values[it].toUInt() } ) diff --git a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayULong.kt b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayULong.kt index 4d077246..1db82830 100644 --- a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayULong.kt +++ b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayULong.kt @@ -40,6 +40,12 @@ class ArrayULong(shape : IntArray, datatype : Datatype<*>, val values: ULongArra return dst } + override fun transfer(dst: Any, tc: TransferChunk) { + val src = this.values + val dest = dst as ULongArray + repeat(tc.nelems) { dest[tc.destElem.toInt()+it] = src[tc.srcElem.toInt() + it] } + } + companion object { fun fromLongArray(shape : IntArray, values : LongArray): ArrayULong = ArrayULong(shape, Datatype.ULONG, ULongArray(values.size) { values[it].toULong() } ) diff --git a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayUShort.kt b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayUShort.kt index 280ce91e..3297df2a 100644 --- a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayUShort.kt +++ b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayUShort.kt @@ -40,6 +40,12 @@ class ArrayUShort(shape : IntArray, datatype : Datatype<*>, val values: UShortAr return dst } + override fun transfer(dst: Any, tc: TransferChunk) { + val src = this.values + val dest = dst as UShortArray + repeat(tc.nelems) { dest[tc.destElem.toInt()+it] = src[tc.srcElem.toInt() + it] } + } + companion object { fun fromShortArray(shape : IntArray, values : ShortArray): ArrayUShort = ArrayUShort(shape, UShortArray(values.size) { values[it].toUShort() } ) diff --git a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayVlen.kt b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayVlen.kt index 105233c3..1c6fed9e 100644 --- a/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayVlen.kt +++ b/core/src/commonMain/kotlin/com/sunya/cdm/array/ArrayVlen.kt @@ -3,6 +3,7 @@ package com.sunya.cdm.array import com.sunya.cdm.api.* import com.sunya.cdm.layout.IndexND import com.sunya.cdm.layout.IndexSpace +import com.sunya.cdm.layout.TransferChunk // maybe should just return primitive array if only one ?? class ArrayVlen(shape : IntArray, val values : List>, val baseType : Datatype) @@ -34,6 +35,12 @@ class ArrayVlen(shape : IntArray, val values : List>, val baseType : return ArrayVlen(section.shape.toIntArray(), sectionList, baseType) } + override fun transfer(dst: Any, tc: TransferChunk) { + val src = this.values + val dest = dst as MutableList> + repeat(tc.nelems) { dest[tc.destElem.toInt()+it] = src[tc.srcElem.toInt() + it] } + } + override fun equals(other: Any?): Boolean { if (this === other) return true if (other !is ArrayVlen<*>) return false diff --git a/core/src/commonMain/kotlin/com/sunya/cdm/layout/Chunker.kt b/core/src/commonMain/kotlin/com/sunya/cdm/layout/Chunker.kt index 969bce1a..f6978d4d 100644 --- a/core/src/commonMain/kotlin/com/sunya/cdm/layout/Chunker.kt +++ b/core/src/commonMain/kotlin/com/sunya/cdm/layout/Chunker.kt @@ -102,21 +102,6 @@ internal class Chunker(val dataChunk: IndexSpace, val wantSpace: IndexSpace, mer return "Chunker(nelems=$nelems, totalNelems=$totalNelems, dataChunk=$dataChunk, wantSpace=$wantSpace)" } - /* transfer from src to dst buffer, using my computed chunks - internal fun transfer(src: ByteBuffer, elemSize : Int, dst: ByteBuffer) { - for (chunk in this) { - System.arraycopy( - src.array(), - src.arrayOffset() + elemSize * chunk.srcElem.toInt(), - dst.array(), - dst.arrayOffset() + elemSize * chunk.destElem.toInt(), - elemSize * chunk.nelems, - ) - } - } - - */ - // the chunker tracks the dst offset internal fun transferBA(src: ByteArray, srcOffset: Int, elemSize : Int, dst: ByteArray, dstOffset: Int) { for (chunk in this) { diff --git a/core/src/commonMain/kotlin/com/sunya/cdm/layout/TransferChunk.kt b/core/src/commonMain/kotlin/com/sunya/cdm/layout/TransferChunk.kt index 706d25f5..1f23dee5 100644 --- a/core/src/commonMain/kotlin/com/sunya/cdm/layout/TransferChunk.kt +++ b/core/src/commonMain/kotlin/com/sunya/cdm/layout/TransferChunk.kt @@ -5,7 +5,7 @@ package com.sunya.cdm.layout * Everything here is in elements, not bytes. * Read nelems from src at srcElem, store in destination at destElem. */ -internal data class TransferChunk( +data class TransferChunk( val srcElem : Long, // start reading here in the source val nelems: Int, // read these many contiguous elements val destElem: Long // start transferring to here in destination diff --git a/core/src/commonMain/kotlin/com/sunya/netchdf/hdf4/ODLparser.kt b/core/src/commonMain/kotlin/com/sunya/netchdf/hdf4/ODLparser.kt index b45828c2..b0ee43b7 100644 --- a/core/src/commonMain/kotlin/com/sunya/netchdf/hdf4/ODLparser.kt +++ b/core/src/commonMain/kotlin/com/sunya/netchdf/hdf4/ODLparser.kt @@ -7,6 +7,7 @@ import com.sunya.cdm.api.Group import com.sunya.cdm.util.Indent import com.sunya.cdm.util.InternalLibraryApi import com.sunya.cdm.util.makeValidCdmObjectName +import io.github.oshai.kotlinlogging.KotlinLogging /* * http://newsroom.gsfc.nasa.gov/sdptoolkit/hdfeosfaq.html @@ -332,10 +333,6 @@ private fun stripQuotes(name: String): String { //////////////////////////////////////////////////////////////////////////////////////////////// -private const val showDetail = false -private const val showProblems = false -private const val showValidationFailures = false - @InternalLibraryApi class ODLparser(val rootGroup: Group.Builder, val show : Boolean = false) { @@ -398,11 +395,11 @@ class ODLparser(val rootGroup: Group.Builder, val show : Boolean = false) { try { val dimLength = att.component2().toInt() if (dimLength == 0) { - println(" *** ODL has zero dimension length for ${att.component1()}") + logger.warn{" *** ODL has zero dimension length for ${att.component1()}"} return false } } catch (ex : Exception) { - println(" *** ODL cant parse dimension ${att.component1()} length ${att.component2()}") + logger.warn{" *** ODL cant parse dimension ${att.component1()} length ${att.component2()}"} return false } } @@ -435,4 +432,12 @@ class ODLparser(val rootGroup: Group.Builder, val show : Boolean = false) { } return true } + + companion object { + private val logger = KotlinLogging.logger("ODLParser") + + private const val showDetail = false + private const val showProblems = false + private const val showValidationFailures = false + } } \ No newline at end of file diff --git a/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/BTree1data.kt b/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/BTree1data.kt index 0f3f8f49..4f115c02 100644 --- a/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/BTree1data.kt +++ b/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/BTree1data.kt @@ -22,6 +22,7 @@ internal class BTree1data( rootNode = BTreeNode(rootNodeAddress, null) } + // if other layouts like BTree2data had this interface we could use in chunkConcurrent fun asSequence(): Sequence> = sequence { repeat( tiling.nelems) { //val startingIndex = tiling.orderToIndex(it.toLong()) @@ -80,7 +81,7 @@ internal class BTree1data( } // this does not have missing data. Use iterator on the Btree1data class - // return only the leaf nodes, in depth-first order + /* return only the leaf nodes, in depth-first order fun asSequence(): Sequence> = sequence { // Handle child nodes recursively (in-order traversal) if (children.isNotEmpty()) { @@ -90,7 +91,7 @@ internal class BTree1data( } else { // If it's a leaf node (no children) keyValues.forEach { yield(it) } } - } + } */ fun findDataChunk(wantOrder: Int): DataChunk? { if (children.isNotEmpty()) { // search tree; assumes that chunks are ordered diff --git a/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/BTree2data.kt b/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/BTree2data.kt index bab1967b..90495989 100644 --- a/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/BTree2data.kt +++ b/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/BTree2data.kt @@ -326,20 +326,6 @@ internal class BTree2data(private val h5: H5builder, owner: String, address: Lon } } - fun makeMissingDataChunkEntry(rootNode: BTree1.Node, wantKey: LongArray): DataChunkIF { - return MissingDataChunk() - } - - class MissingDataChunk() : DataChunkIF { - override fun childAddress() = -1L - override fun offsets() = longArrayOf() - override fun isMissing() = true - override fun chunkSize() = 0 - override fun filterMask() = 0 - - override fun show(tiling : Tiling) : String = "missing" - } - companion object { internal fun findRecord1byId(records: List, hugeObjectID: Int): Record1? { for (record in records) { diff --git a/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/H5chunkConcurrent.kt b/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/H5chunkConcurrent.kt index 7a0432e0..707110e6 100644 --- a/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/H5chunkConcurrent.kt +++ b/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/H5chunkConcurrent.kt @@ -10,6 +10,7 @@ import com.sunya.cdm.api.Variable import com.sunya.cdm.api.computeSize import com.sunya.cdm.api.toIntArray import com.sunya.cdm.api.toLongArray +import com.sunya.cdm.array.ArrayTyped import com.sunya.cdm.iosp.OpenFileState import com.sunya.cdm.layout.Chunker import com.sunya.cdm.layout.IndexSpace @@ -18,6 +19,7 @@ import com.sunya.cdm.layout.transferMissingNelems import com.sunya.cdm.util.InternalLibraryApi import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.Job import kotlinx.coroutines.channels.ReceiveChannel import kotlinx.coroutines.channels.produce @@ -26,8 +28,8 @@ import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking import kotlinx.coroutines.yield -class H5chunkConcurrent(h5file: Hdf5File, val v2: Variable<*>, wantSection: SectionPartial?) { - val h5 = h5file.header +@ExperimentalCoroutinesApi +class H5chunkConcurrent(val h5: H5builder, val v2: Variable, wantSection: SectionPartial?) { val rafext: OpenFileExtended = h5.openFileExtended() internal val bTree: BTree1data @@ -35,8 +37,6 @@ class H5chunkConcurrent(h5file: Hdf5File, val v2: Variable<*>, wantSection: Sect val chunkShape: IntArray val tiling: Tiling val nchunks: Long - // internal val rootNode: BTree1data.BTreeNode - // val rootAddress: Long val wantSpace: IndexSpace val allData : Boolean @@ -55,10 +55,9 @@ class H5chunkConcurrent(h5file: Hdf5File, val v2: Variable<*>, wantSection: Sect // its not obvious you actually need a seperate raf bTree = BTree1data(rafext, mdl.btreeAddress, varShape, chunkShape.toLongArray()) - // rootAddress = mdl.btreeAddress } - fun readChunks(nthreads: Int, lamda: (ArraySection<*>) -> Unit, done: () -> Unit) { + fun readChunks(nthreads: Int, lamda: (ArraySection) -> Unit, done: () -> Unit) { runBlocking { val jobs = mutableListOf() @@ -92,7 +91,7 @@ class H5chunkConcurrent(h5file: Hdf5File, val v2: Variable<*>, wantSection: Sect private fun CoroutineScope.launchJob( worker: Worker, input: ReceiveChannel>, - lamda: (ArraySection<*>) -> Unit, + lamda: (ArraySection) -> Unit, ) = launch(Dispatchers.Default) { for (pair: Pair in input) { val arraySection = worker.work(pair.second) @@ -121,7 +120,7 @@ class H5chunkConcurrent(h5file: Hdf5File, val v2: Variable<*>, wantSection: Sect state = OpenFileState(0L, h5type.isBE) } - fun work(dataChunk : DataChunkIF) : ArraySection<*>? { + fun work(dataChunk : DataChunkIF) : ArraySection? { val dataSpace = IndexSpace(v2.rank, dataChunk.offsets(), vinfo.storageDims) if (!allData && !wantSpace.intersects(dataSpace)) { return null @@ -156,7 +155,7 @@ class H5chunkConcurrent(h5file: Hdf5File, val v2: Variable<*>, wantSection: Sect h5.processDataIntoArray(ba, h5type.isBE, datatype, intersectSpace.shape.toIntArray(), h5type, elemSize) } - return ArraySection(array, intersectSpace.section(v2.shape)) + return ArraySection(array as ArrayTyped, intersectSpace.section(v2.shape)) } } val debugChunking = false diff --git a/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/H5chunkReader.kt b/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/H5chunkReader.kt index 4f8d8bba..c1dac0de 100644 --- a/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/H5chunkReader.kt +++ b/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/H5chunkReader.kt @@ -1,156 +1,215 @@ -@file:OptIn(InternalLibraryApi::class) +@file:OptIn(InternalLibraryApi::class, ExperimentalUnsignedTypes::class) package com.sunya.netchdf.hdf5 +import com.fleeksoft.charset.decodeToString import com.sunya.cdm.api.* import com.sunya.cdm.array.* import com.sunya.cdm.iosp.OpenFileState import com.sunya.cdm.layout.Chunker import com.sunya.cdm.layout.IndexSpace +import com.sunya.cdm.layout.TransferChunk import com.sunya.cdm.util.InternalLibraryApi +import kotlin.collections.iterator -internal class H5chunkReader(val h5 : H5builder) { - private val debugChunking = false +private val debugChunking = false - internal fun readChunkedData(v2: Variable, wantSection: Section, index: Iterator): ArrayTyped { - val vinfo = v2.spObject as DataContainerVariable - val h5type = vinfo.h5type +// DataLayoutSingleChunk4, DataLayoutImplicit4, DataLayoutFixedArray4, DataLayoutExtensibleArray4, DataLayoutBtreeVer2 +internal fun H5builder.readChunkedData(v2: Variable, wantSection: Section, index: Iterator): ArrayTyped { + val vinfo = v2.spObject as DataContainerVariable + val h5type = vinfo.h5type - val elemSize = vinfo.storageDims[vinfo.storageDims.size - 1].toInt() // last one is always the elements size - val datatype = vinfo.h5type.datatype() + val elemSize = vinfo.storageDims[vinfo.storageDims.size - 1].toInt() // last one is always the elements size + val datatype = vinfo.h5type.datatype() - val wantSpace = IndexSpace(wantSection) - val sizeBytes = wantSpace.totalElements * elemSize - if (sizeBytes <= 0 || sizeBytes >= Int.MAX_VALUE) { - throw RuntimeException("Illegal nbytes to read = $sizeBytes") - } - val ba = ByteArray(sizeBytes.toInt()) - - // just reading into memory the entire index for now - // val index = BTree2j(h5, v2.name, vinfo.dataPos, vinfo.storageDims) - - val filters = FilterPipeline(v2.name, vinfo.mfp, vinfo.h5type.isBE) - val state = OpenFileState(0L, vinfo.h5type.isBE) - - // just run through all the chunks, we wont read any that we dont want - for (dataChunk : ChunkImpl in index) { - val dataSection = IndexSpace(v2.rank, dataChunk.chunkOffset.toLongArray(), vinfo.storageDims) - val chunker = Chunker(dataSection, wantSpace) // each DataChunkEntry has its own Chunker iteration - if (chunker.nelems > 0) { // TODO efficient enough ?? - /* if (dataChunk.isMissing()) { - if (debugChunking) println(" missing ${dataChunk.show(tiledData.tiling)}") - chunker.transferMissing(vinfo.fillValue, elemSize, ba) - } else { */ - state.pos = dataChunk.address - val rawdata = h5.raf.readByteArray(state, dataChunk.size) - val filteredData = if (vinfo.mfp == null || dataChunk.filterMask == null) rawdata - else filters.apply(rawdata, dataChunk.filterMask) - chunker.transferBA(filteredData, 0, elemSize, ba, 0) - } - // } + val wantSpace = IndexSpace(wantSection) + val sizeBytes = wantSpace.totalElements * elemSize + if (sizeBytes <= 0 || sizeBytes >= Int.MAX_VALUE) { + throw RuntimeException("Illegal nbytes to read = $sizeBytes") + } + val ba = ByteArray(sizeBytes.toInt()) + + // just reading into memory the entire index for now + // val index = BTree2j(h5, v2.name, vinfo.dataPos, vinfo.storageDims) + + val filters = FilterPipeline(v2.name, vinfo.mfp, vinfo.h5type.isBE) + val state = OpenFileState(0L, vinfo.h5type.isBE) + + // just run through all the chunks, we wont read any that we dont want + for (dataChunk : ChunkImpl in index) { + val dataSection = IndexSpace(v2.rank, dataChunk.chunkOffset.toLongArray(), vinfo.storageDims) + val chunker = Chunker(dataSection, wantSpace) // each DataChunkEntry has its own Chunker iteration + if (chunker.nelems > 0) { // TODO efficient enough ?? + /* if (dataChunk.isMissing()) { + if (debugChunking) println(" missing ${dataChunk.show(tiledData.tiling)}") + chunker.transferMissing(vinfo.fillValue, elemSize, ba) + } else { */ + state.pos = dataChunk.address + val rawdata = this.raf.readByteArray(state, dataChunk.size) + val filteredData = if (vinfo.mfp == null || dataChunk.filterMask == null) rawdata + else filters.apply(rawdata, dataChunk.filterMask) + chunker.transferBA(filteredData, 0, elemSize, ba, 0) } + // } + } - val shape = wantSpace.shape.toIntArray() + val shape = wantSpace.shape.toIntArray() - return if (h5type.datatype5 == Datatype5.Vlen) { - h5.processVlenIntoArray(h5type, shape, ba, wantSpace.totalElements.toInt(), elemSize) - } else { - h5.processDataIntoArray(ba, vinfo.h5type.isBE, datatype, shape, h5type, elemSize) as ArrayTyped - } + return if (h5type.datatype5 == Datatype5.Vlen) { + this.processVlenIntoArray(h5type, shape, ba, wantSpace.totalElements.toInt(), elemSize) + } else { + this.processDataIntoArray(ba, vinfo.h5type.isBE, datatype, shape, h5type, elemSize) as ArrayTyped } +} - // TODO can we use concurrent reading ?? - internal fun readBtreeVer1(v2: Variable, wantSection: Section): ArrayTyped { - val vinfo = v2.spObject as DataContainerVariable - val h5type = vinfo.h5type +/* TODO can we use concurrent reading ?? +internal fun readBtree1data(v2: Variable, wantSection: Section): ArrayTyped { + val vinfo = v2.spObject as DataContainerVariable + val h5type = vinfo.h5type - val elemSize = vinfo.storageDims[vinfo.storageDims.size - 1].toInt() // last one is always the elements size - val datatype = vinfo.h5type.datatype() + val elemSize = vinfo.storageDims[vinfo.storageDims.size - 1].toInt() // last one is always the elements size + val datatype = vinfo.h5type.datatype() - val wantSpace = IndexSpace(wantSection) - val sizeBytes = wantSpace.totalElements * elemSize - if (sizeBytes <= 0 || sizeBytes >= Int.MAX_VALUE) { - throw RuntimeException("Illegal nbytes to read = $sizeBytes") - } - val ba = ByteArray(sizeBytes.toInt()) - - val btree1 = if (vinfo.mdl is DataLayoutBTreeVer1) - BTree1(h5, vinfo.dataPos, 1, vinfo.storageDims.size) - else - throw RuntimeException("Unsupprted mdl ${vinfo.mdl}") - - val tiledData = H5TiledData1(btree1, v2.shape, vinfo.storageDims) - val filters = FilterPipeline(v2.name, vinfo.mfp, vinfo.h5type.isBE) - if (debugChunking) println(" readChunkedData tiles=${tiledData.tiling}") - - var transferChunks = 0 - val state = OpenFileState(0L, vinfo.h5type.isBE) - for (dataChunk: DataChunkIF in tiledData.dataChunks(wantSpace)) { // : Iterable - val dataSection = IndexSpace(v2.rank, dataChunk.offsets(), vinfo.storageDims) - val chunker = Chunker(dataSection, wantSpace) // each DataChunkEntry has its own Chunker iteration - if (dataChunk.isMissing()) { - if (debugChunking) println(" missing ${dataChunk.show(tiledData.tiling)}") - chunker.transferMissing(vinfo.fillValue, elemSize, ba) - } else { - if (debugChunking) println(" chunk=${dataChunk.show(tiledData.tiling)}") - state.pos = dataChunk.childAddress() - val chunkData = h5.raf.readByteArray(state, dataChunk.chunkSize()) - val filteredData = if (dataChunk.filterMask() == null) chunkData - else filters.apply(chunkData, dataChunk.filterMask()!!) - chunker.transferBA(filteredData, 0, elemSize, ba, 0) - transferChunks += chunker.transferChunks - } - } + val wantSpace = IndexSpace(wantSection) + val sizeBytes = wantSpace.totalElements * elemSize + if (sizeBytes <= 0 || sizeBytes >= Int.MAX_VALUE) { + throw RuntimeException("Illegal nbytes to read = $sizeBytes") + } + val ba = ByteArray(sizeBytes.toInt()) + + val reader = H5chunkConcurrent(h5, v2, wantSection) + val availableProcessors = com.sunya.netchdf.util.getAvailableProcessors() + reader.readChunks(availableProcessors, lamda = { asection: ArraySection<*> -> + val (array, section) = asection + println(" section = ${section}") + val dataSpace = IndexSpace(section) + + val useEntireChunk = wantSpace.contains(dataSpace) + val intersectSpace = if (useEntireChunk) dataSpace else wantSpace.intersect(dataSpace) - val shape = wantSpace.shape.toIntArray() + val chunker = Chunker(dataSpace, wantSpace) // each DataChunkEntry has its own Chunker iteration + chunker.transferBA(array, 0, elemSize, ba, 0) - return if (h5type.datatype5 == Datatype5.Vlen) { - h5.processVlenIntoArray(h5type, shape, ba, wantSpace.totalElements.toInt(), elemSize) + if (h5type.datatype5 == Datatype5.Vlen) { + // internal fun H5builder.processVlenIntoArray(h5type: H5TypeInfo, shape: IntArray, ba: ByteArray, nelems: Int, elemSize : Int): ArrayTyped { + this.processVlenIntoArray(h5type, intersectSpace.shape.toIntArray(), ba, intersectSpace.totalElements.toInt(), elemSize) } else { - h5.processDataIntoArray(ba, vinfo.h5type.isBE, datatype, shape, h5type, elemSize) as ArrayTyped + this.processDataIntoArray(ba, h5type.isBE, datatype, intersectSpace.shape.toIntArray(), h5type, elemSize) } - } -} -// Chunked data apparently has heapIds directly, not addresses of heapIds. Go figure. -internal fun H5builder.processVlenIntoArray(h5type: H5TypeInfo, shape: IntArray, ba: ByteArray, nelems: Int, elemSize : Int): ArrayTyped { - val h5heap = H5heap(this) + }, done = { }) + + return ArraySection(array, intersectSpace.section(v2.shape)) +} */ - if (h5type.isVlenString) { - val sarray = mutableListOf() - for (i in 0 until nelems) { - val sval = h5heap.readHeapString(ba, i * elemSize) - sarray.add(sval ?: "") +// DataLayoutBTreeVer1 +internal fun H5builder.readBtreeVer1(v2: Variable, wantSection: Section): ArrayTyped { + val vinfo = v2.spObject as DataContainerVariable + val h5type = vinfo.h5type + + val elemSize = vinfo.storageDims[vinfo.storageDims.size - 1].toInt() // last one is always the elements size + val datatype = vinfo.h5type.datatype() + + val wantSpace = IndexSpace(wantSection) + val sizeBytes = wantSpace.totalElements * elemSize + if (sizeBytes <= 0 || sizeBytes >= Int.MAX_VALUE) { + throw RuntimeException("Illegal nbytes to read = $sizeBytes") + } + val ba = ByteArray(sizeBytes.toInt()) + + val btree1 = if (vinfo.mdl is DataLayoutBTreeVer1) + BTree1(this, vinfo.dataPos, 1, vinfo.storageDims.size) + else + throw RuntimeException("Unsupprted mdl ${vinfo.mdl}") + + val tiledData = H5TiledData1(btree1, v2.shape, vinfo.storageDims) + val filters = FilterPipeline(v2.name, vinfo.mfp, vinfo.h5type.isBE) + if (debugChunking) println(" readChunkedData tiles=${tiledData.tiling}") + + var transferChunks = 0 + val state = OpenFileState(0L, vinfo.h5type.isBE) + for (dataChunk: DataChunkIF in tiledData.dataChunks(wantSpace)) { // : Iterable + val dataSection = IndexSpace(v2.rank, dataChunk.offsets(), vinfo.storageDims) + val chunker = Chunker(dataSection, wantSpace) // each DataChunkEntry has its own Chunker iteration + if (dataChunk.isMissing()) { + if (debugChunking) println(" missing ${dataChunk.show(tiledData.tiling)}") + chunker.transferMissing(vinfo.fillValue, elemSize, ba) + } else { + if (debugChunking) println(" chunk=${dataChunk.show(tiledData.tiling)}") + state.pos = dataChunk.childAddress() + val chunkData = this.raf.readByteArray(state, dataChunk.chunkSize()) + val filteredData = if (dataChunk.filterMask() == null) chunkData + else filters.apply(chunkData, dataChunk.filterMask()!!) + chunker.transferBA(filteredData, 0, elemSize, ba, 0) + transferChunks += chunker.transferChunks } - return ArrayString(shape, sarray) as ArrayTyped + } + + val shape = wantSpace.shape.toIntArray() + return if (h5type.datatype5 == Datatype5.Vlen) { + this.processVlenIntoArray(h5type, shape, ba, wantSpace.totalElements.toInt(), elemSize) } else { - val base = h5type.base!! - if (base.datatype5 == Datatype5.Reference) { - val refsList = mutableListOf() - for (i in 0 until nelems) { - val heapId = h5heap.readHeapIdentifier(ba, i * elemSize) - val vlenArray = h5heap.getHeapDataArray(heapId, Datatype.LONG, base.isBE) as Array - // LOOK require vlenArray is Array - // TODO val refsArray = this.convertReferencesToDataObjectName(vlenArray.asIterable()) - val refsArray = this.convertReferencesToDataObjectName(vlenArray) - for (s in refsArray) { - refsList.add(s) - } - } - return ArrayString(shape, refsList) as ArrayTyped - } + this.processDataIntoArray(ba, vinfo.h5type.isBE, datatype, shape, h5type, elemSize) as ArrayTyped + } +} + +// DataLayoutBTreeVer1 +internal fun readBtree1data(hdf5: Hdf5File, v2: Variable, wantSection: SectionPartial?): ArrayTyped { + val vinfo = v2.spObject as DataContainerVariable + val h5type = vinfo.h5type + val datatype = vinfo.h5type.datatype() + val elemSize = vinfo.storageDims[vinfo.storageDims.size - 1].toInt() // last one is always the elements size + + val useSection = SectionPartial.fill(wantSection, v2.shape) + val wantSpace = IndexSpace(useSection) + val nelems = wantSpace.totalElements.toInt() + + val values = when (datatype) { + Datatype.BYTE -> ByteArray(nelems) + Datatype.CHAR, Datatype.UBYTE, Datatype.ENUM1 -> UByteArray(nelems) + Datatype.SHORT -> ShortArray(nelems) + Datatype.USHORT, Datatype.ENUM2 -> UShortArray(nelems) + Datatype.INT -> IntArray(nelems) + Datatype.UINT, Datatype.ENUM4 -> UIntArray(nelems) + Datatype.LONG -> LongArray(nelems) + Datatype.ULONG, Datatype.ENUM8 -> ULongArray(nelems) + Datatype.DOUBLE -> DoubleArray(nelems) + Datatype.FLOAT -> FloatArray(nelems) + Datatype.STRING -> MutableList(nelems) {""} + else -> throw IllegalArgumentException("datatype ${datatype}") + } - // general case is to read an array of vlen objects - // each vlen generates an Array of type baseType - val listOfArrays = mutableListOf>() - val readDatatype = base.datatype() - for (i in 0 until nelems) { - val heapId = h5heap.readHeapIdentifier(ba, i * elemSize) - val vlenArray = h5heap.getHeapDataArray(heapId, readDatatype, base.isBE) - // LOOK require vlenArray is Array - listOfArrays.add(vlenArray) + // so instead of type byte array we have the actual ArrayTyped + // we have to transfer the chunk into the approprate places in the array + val chunkIter = hdf5.chunkIterator(v2, wantSection) + chunkIter.forEach { dataChunk : ArraySection -> + val dataSection = IndexSpace(dataChunk.chunkSection) + val chunker = Chunker(dataSection, wantSpace) // each DataChunkEntry has its own Chunker iteration + chunker.forEach { + // println(it) + dataChunk.array.transfer(values, it) } - return ArrayVlen.fromArray(shape, listOfArrays, readDatatype) as ArrayTyped } + + val shape = wantSpace.shape.toIntArray() + val result = when (datatype) { + Datatype.BYTE -> ArrayByte(shape, values as ByteArray) + Datatype.CHAR, Datatype.UBYTE, Datatype.ENUM1 -> ArrayUByte(shape, values as UByteArray) + Datatype.SHORT -> ArrayShort(shape, values as ShortArray) + Datatype.USHORT, Datatype.ENUM2 -> ArrayUShort(shape, values as UShortArray) + Datatype.INT -> ArrayInt(shape, values as IntArray) + Datatype.UINT, Datatype.ENUM4 -> ArrayUInt(shape, values as UIntArray) + Datatype.LONG -> ArrayLong(shape, values as LongArray) + Datatype.ULONG, Datatype.ENUM8 -> ArrayULong(shape, values as ULongArray) + Datatype.DOUBLE -> ArrayDouble(shape, values as DoubleArray) + Datatype.FLOAT -> ArrayFloat(shape, values as FloatArray) + Datatype.STRING -> ArrayString(shape, values as List) + else -> throw IllegalArgumentException("datatype ${datatype}") + } + return result as ArrayTyped +} + +internal fun transfer(src: FloatArray, dest: FloatArray, tc: TransferChunk) { + repeat(tc.nelems) { dest[tc.destElem.toInt()+it] = src[tc.srcElem.toInt() + it] } } \ No newline at end of file diff --git a/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/H5reader.kt b/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/H5dataReader.kt similarity index 77% rename from core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/H5reader.kt rename to core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/H5dataReader.kt index 94ea46e4..b1405bb6 100644 --- a/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/H5reader.kt +++ b/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/H5dataReader.kt @@ -11,7 +11,7 @@ import com.sunya.cdm.util.InternalLibraryApi private const val debugLayout = false -// Handles reading attributes and non-chunked Variables +// Handles reading attributes and non-chunked Variables (DataLayoutContiguous, DataLayoutContiguous3) internal fun H5builder.readRegularData(dc: DataContainer, datatype: Datatype, section : Section?): ArrayTyped { if (dc.mds.type == DataspaceType.Null) { return ArrayString(intArrayOf(), listOf()) as ArrayTyped @@ -30,7 +30,8 @@ internal fun H5builder.readRegularData(dc: DataContainer, datatype: Datatype return readDataWithLayout(state, layout, datatype, wantSection.shape, h5type) } -// LOOK: not subsetting +// TODO: not subsetting +// DataLayoutCompact, DataLayoutCompact3 internal fun H5builder.readCompactData(v2 : Variable, shape : IntArray): ArrayTyped { val vinfo = v2.spObject as DataContainerVariable val ba = when (vinfo.mdl) { @@ -46,7 +47,7 @@ internal fun H5builder.readCompactData(v2 : Variable, shape : IntArray): } } -// handles reading data with a Layout. LOOK: Fill Value ?? +// handles reading data with a Layout. TODO: are we using Fill Value ?? internal fun H5builder.readDataWithLayout(state: OpenFileState, layout: Layout, datatype: Datatype, shape : LongArray, h5type : H5TypeInfo): ArrayTyped { val sizeBytes = layout.totalNelems * layout.elemSize if (sizeBytes <= 0 || sizeBytes >= Int.MAX_VALUE) { @@ -178,3 +179,46 @@ internal fun H5builder.readVlenDataWithLayout(dc: DataContainer, layout : Layout } } +// Chunked data apparently has heapIds directly, not addresses of heapIds. Go figure. +internal fun H5builder.processVlenIntoArray(h5type: H5TypeInfo, shape: IntArray, ba: ByteArray, nelems: Int, elemSize : Int): ArrayTyped { + val h5heap = H5heap(this) + + if (h5type.isVlenString) { + val sarray = mutableListOf() + for (i in 0 until nelems) { + val sval = h5heap.readHeapString(ba, i * elemSize) + sarray.add(sval ?: "") + } + return ArrayString(shape, sarray) as ArrayTyped + + } else { + val base = h5type.base!! + if (base.datatype5 == Datatype5.Reference) { + val refsList = mutableListOf() + for (i in 0 until nelems) { + val heapId = h5heap.readHeapIdentifier(ba, i * elemSize) + val vlenArray = h5heap.getHeapDataArray(heapId, Datatype.LONG, base.isBE) as Array + // LOOK require vlenArray is Array + // TODO val refsArray = this.convertReferencesToDataObjectName(vlenArray.asIterable()) + val refsArray = this.convertReferencesToDataObjectName(vlenArray) + for (s in refsArray) { + refsList.add(s) + } + } + return ArrayString(shape, refsList) as ArrayTyped + } + + // general case is to read an array of vlen objects + // each vlen generates an Array of type baseType + val listOfArrays = mutableListOf>() + val readDatatype = base.datatype() + for (i in 0 until nelems) { + val heapId = h5heap.readHeapIdentifier(ba, i * elemSize) + val vlenArray = h5heap.getHeapDataArray(heapId, readDatatype, base.isBE) + // LOOK require vlenArray is Array + listOfArrays.add(vlenArray) + } + return ArrayVlen.fromArray(shape, listOfArrays, readDatatype) as ArrayTyped + } +} + diff --git a/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/Hdf5File.kt b/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/Hdf5File.kt index 265477f0..2d80e7d6 100644 --- a/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/Hdf5File.kt +++ b/core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/Hdf5File.kt @@ -12,6 +12,8 @@ import com.sunya.cdm.array.ArrayTyped import com.sunya.cdm.array.TypedByteArray import com.sunya.cdm.iosp.* import com.sunya.cdm.util.InternalLibraryApi +import com.sunya.netchdf.util.Deque +import com.sunya.netchdf.util.useDefaultNThreads import kotlin.String /** @@ -39,6 +41,11 @@ class Hdf5File(val filename : String, strict : Boolean = false) : Netchdf { return vinfo.mdl.javaClass.simpleName } + var useNThreads : Int? = null + fun useNThreads(): Int { + return useNThreads ?: com.sunya.netchdf.util.useDefaultNThreads() + } + override fun readArrayData(v2: Variable, section: SectionPartial?): ArrayTyped { if (v2.nelems == 0L) { return ArrayEmpty(v2.shape.toIntArray(), v2.datatype) @@ -70,35 +77,38 @@ class Hdf5File(val filename : String, strict : Boolean = false) : Netchdf { header.readRegularData(vinfo, v2.datatype, wantSection) } else if (vinfo.mdl is DataLayoutBTreeVer1) { - H5chunkReader(header).readBtreeVer1(v2, wantSection) + if (v2.datatype == Datatype.COMPOUND) + header.readBtreeVer1(v2, wantSection) + else + readBtree1data(this, v2, section) } else if (vinfo.mdl is DataLayoutSingleChunk4) { - // H5chunkReader(header).readSingleChunk(v2, wantSection) + // header.readSingleChunk(v2, wantSection) // internal data class DataLayoutSingleChunk4(val flags: Byte, val chunkDimensions: IntArray, val chunkSize: Int, val heapAddress: Long, val filterMask: Int?) : DataLayoutMessage() { val offset = IntArray(v2.rank) val chunk = ChunkImpl(vinfo.mdl.heapAddress, vinfo.mdl.chunkSize, offset, vinfo.mdl.filterMask) - H5chunkReader(header).readChunkedData(v2, wantSection, listOf(chunk).iterator()) + header.readChunkedData(v2, wantSection, listOf(chunk).iterator()) } else if (vinfo.mdl is DataLayoutImplicit4) { - // H5chunkReader(header).readImplicit4(v2, wantSection) + // header.readImplicit4(v2, wantSection) val index = ImplicitChunkIndex(header, varShape=v2.shape.toIntArray(), vinfo.mdl) - H5chunkReader(header).readChunkedData(v2, wantSection, index.chunkIterator()) + header.readChunkedData(v2, wantSection, index.chunkIterator()) } else if (vinfo.mdl is DataLayoutFixedArray4) { - // H5chunkReader(header).readFixedArray4(v2, wantSection) + // header.readFixedArray4(v2, wantSection) val index = FixedArrayIndex(header, varShape=v2.shape.toIntArray(), vinfo.mdl) // mdl.fixedArrayIndex - H5chunkReader(header).readChunkedData(v2, wantSection, index.chunkIterator()) + header.readChunkedData(v2, wantSection, index.chunkIterator()) } else if (vinfo.mdl is DataLayoutExtensibleArray4) { val index = ExtensibleArrayIndex(header, vinfo.mdl.indexAddress, v2.shape.toIntArray(), vinfo.mdl.chunkDimensions) - H5chunkReader(header).readChunkedData(v2, wantSection, index.chunkIterator()) + header.readChunkedData(v2, wantSection, index.chunkIterator()) } else if (vinfo.mdl is DataLayoutBtreeVer2) { - // H5chunkReader(header).readBtreeVer2j(v2, wantSection) + // header.readBtreeVer2j(v2, wantSection) val index = BTree2data(header, v2.name, vinfo.dataPos, vinfo.storageDims) - H5chunkReader(header).readChunkedData(v2, wantSection, index.chunkIterator()) + header.readChunkedData(v2, wantSection, index.chunkIterator()) } else { throw RuntimeException("Unsupported data layer type ${vinfo.mdl}") @@ -126,43 +136,42 @@ class Hdf5File(val filename : String, strict : Boolean = false) : Netchdf { // TODO can we use concurrent reading ?? return if (this.layoutName(v2) == "DataLayoutBTreeVer1") { - H5chunkIterator(header, v2, wantSection) + // H5chunkIterator(header, v2, wantSection) + H5chunkIterator2(this, v2, section) } else { H5maxIterator(this, v2, wantSection, maxElements ?: 100_000) } } - override fun readChunksConcurrent(v2: Variable, lamda : (ArraySection<*>) -> Unit, done : () -> Unit, - wantSection: SectionPartial?, nthreads: Int?) { - val reader = H5chunkConcurrent(this, v2, wantSection) - // TODO default nthreads ?? - reader.readChunks(nthreads ?: 20, lamda, done = { done() }) - } - - /* - class Btree1chunkIterator(val hdfFile: Hdf5File, val v2: Variable<*>, val wantSection: SectionPartial?): AbstractIterator>() { - val reader = H5readConcurrent(hdfFile, v2) - val nthreads = 20 - var currElement: ArraySection<*>? = null // could be a queue ? or a stack with a limit - var done: Boolean = false + class H5chunkIterator2(hdfFile: Hdf5File, val v2: Variable, val wantSection: SectionPartial?): AbstractIterator>() { + val reader = H5chunkConcurrent(hdfFile.header, v2, wantSection) + val nthreads = hdfFile.useNThreads() + val deque = Deque>(10) init { - reader.readChunks(nthreads, - lamda = { it -> - if (currElement == null) currElement = it - }, - done = { done = true } + reader.readChunks( + nthreads, + lamda = { deque.add(it) }, + done = { deque.done() } ) } override fun computeNext() { - if (currElement != null) { - setNext( currElement!! ) - currElement = null + val firstElement = deque.next() + if (firstElement != null) { + setNext(firstElement) } else { - wait() + done() } - if (done) done() } - } */ + } + + override fun readChunksConcurrent(v2: Variable, lamda : (ArraySection) -> Unit, done : () -> Unit, + wantSection: SectionPartial?, nthreads: Int?) { + val reader = H5chunkConcurrent(header, v2, wantSection) + val availableProcessors = this.useNThreads() + // println("availableProcessors = $availableProcessors") + reader.readChunks(nthreads ?: availableProcessors, lamda, done = { done() }) + } + } \ No newline at end of file diff --git a/core/src/commonMain/kotlin/com/sunya/netchdf/util/Kmp.kt b/core/src/commonMain/kotlin/com/sunya/netchdf/util/Kmp.kt new file mode 100644 index 00000000..df2f8a17 --- /dev/null +++ b/core/src/commonMain/kotlin/com/sunya/netchdf/util/Kmp.kt @@ -0,0 +1,11 @@ +package com.sunya.netchdf.util + + +expect fun useDefaultNThreads(): Int + +// add to end, take from head +expect class Deque(initialCapacity: Int) { + fun add(item: T) + fun next(): T? + fun done() +} diff --git a/core/src/commonTest/data/netcdf4/tiling.nc4 b/core/src/commonTest/data/netcdf4/tiling.nc4 new file mode 100755 index 00000000..e5479e2d Binary files /dev/null and b/core/src/commonTest/data/netcdf4/tiling.nc4 differ diff --git a/core/src/commonTest/kotlin/com/sunya/netchdf/hdf5/Btree1dataTest.kt b/core/src/commonTest/kotlin/com/sunya/netchdf/hdf5/Btree1dataTest.kt index d0f689c5..7e57375e 100644 --- a/core/src/commonTest/kotlin/com/sunya/netchdf/hdf5/Btree1dataTest.kt +++ b/core/src/commonTest/kotlin/com/sunya/netchdf/hdf5/Btree1dataTest.kt @@ -60,9 +60,9 @@ class Btree1dataTest { for (nthreads in listOf(1, 2, 4, 8, 10, 16, 20, 24, 32, 40, 48)) { val time = measureNanoTime { // fun readChunks(nthreads: Int, lamda: (ArraySection<*>) -> Unit, done: () -> Unit) { - val reader = H5chunkConcurrent(myfile, myvar, null) - reader.readChunks(nthreads, { asect: ArraySection<*> -> - println(" section = ${asect.section}") + val reader = H5chunkConcurrent(myfile.header, myvar, null) + reader.readChunks(nthreads, lamda = { asect: ArraySection<*> -> + // println(" section = ${asect.chunkSection}") }, { }, ) } println("$nthreads, ${time * nano}") diff --git a/core/src/jvmMain/kotlin/com/sunya/netchdf/util/Kmp.kt b/core/src/jvmMain/kotlin/com/sunya/netchdf/util/Kmp.kt new file mode 100644 index 00000000..1a86618a --- /dev/null +++ b/core/src/jvmMain/kotlin/com/sunya/netchdf/util/Kmp.kt @@ -0,0 +1,48 @@ +package com.sunya.netchdf.util + +import java.lang.Runtime.* +import java.util.concurrent.ConcurrentLinkedDeque + +// todo: i have 2 threads per processer, these dont help the IO i think. +actual fun useDefaultNThreads(): Int { + return getRuntime().availableProcessors() / 2 +} + +actual class Deque actual constructor(initialCapacity: Int) { + val delegate = ConcurrentLinkedDeque() + var done = false + + actual fun add(item: T) { + delegate.add(item) + } + + //actual fun next(): T? { + // return delegate.poll() // can i block until available ?? + //} + + actual fun next(): T? { + var countWaits = 0 + while (true) { + val firstElement = delegate.poll() + if (firstElement != null) { + return firstElement + } else if (done) { + done() + return null + } else if (countWaits > 100) { + println("takes too long (10 sec)") + done() + return null + } else { + // wait 100 msecs + Thread.sleep(100) // java only + countWaits++ + } + } + + } + + actual fun done() { + done = true + } +} \ No newline at end of file diff --git a/core/src/nativeMain/kotlin/com/sunya/netchdf/util/Kmp.kt b/core/src/nativeMain/kotlin/com/sunya/netchdf/util/Kmp.kt new file mode 100644 index 00000000..2830305c --- /dev/null +++ b/core/src/nativeMain/kotlin/com/sunya/netchdf/util/Kmp.kt @@ -0,0 +1,24 @@ +package com.sunya.netchdf.util + +actual fun getAvailableProcessors(): Int { + return Platform.availableProcessors() / 2 +} + +// TODO +actual class Deque actual constructor(initialCapacity: Int) { + private val delegate = ArrayDeque>(initialCapacity) // could be a queue ? or a stack with a limit + private val mutex = Mutex() + + actual fun add(item: T) = runBlocking { + mutex.withLock { + delegate.add(item) + } + } + + actual fun next(): T? = runBlocking { + mutex.withLock { + return deque.removeFirstOrNull() + } + return delegate.poll() // can i block until available ?? + } +} \ No newline at end of file diff --git a/core/src/nativeMain/kotlin/com/sunya/netchdfc/NetchdfCApi.kt b/core/src/nativeMain/kotlin/com/sunya/netchdfc/NetchdfCApi.kt index 9241ae57..aaf3d759 100644 --- a/core/src/nativeMain/kotlin/com/sunya/netchdfc/NetchdfCApi.kt +++ b/core/src/nativeMain/kotlin/com/sunya/netchdfc/NetchdfCApi.kt @@ -41,3 +41,4 @@ class VariableData(val varName: String, dataShape: IntArray, val nelems: Int, da val pinnedData: CPointer = data.pin().addressOf(0) } + diff --git a/testclibs/src/test/kotlin/com/sunya/netchdf/NetchdfClibTest.kt b/testclibs/src/test/kotlin/com/sunya/netchdf/NetchdfClibTest.kt index ce8a7538..8c77e006 100644 --- a/testclibs/src/test/kotlin/com/sunya/netchdf/NetchdfClibTest.kt +++ b/testclibs/src/test/kotlin/com/sunya/netchdf/NetchdfClibTest.kt @@ -738,7 +738,7 @@ fun compareOneVarIterate(myvar: Variable<*>, myfile: Netchdf, cvar : Variable<*> val time1 = measureNanoTime { val chunkIter = myfile.chunkIterator(myvar) for (pair in chunkIter) { - if (debugIter) println(" compareOneVarIterate myvar=${myvar.name} ${pair.section} = ${pair.array.shape.contentToString()}") + if (debugIter) println(" compareOneVarIterate myvar=${myvar.name} ${pair.chunkSection} = ${pair.array.shape.contentToString()}") sum1 += sumValues(pair.array) countChunks++ } @@ -750,7 +750,7 @@ fun compareOneVarIterate(myvar: Variable<*>, myfile: Netchdf, cvar : Variable<*> val time2 = measureNanoTime { val chunkIter = cfile.chunkIterator(cvar) for (pair in chunkIter) { - if (debugIter) println(" compareOneVarIterate cvar=${cvar.name} ${pair.section} = ${pair.array.shape.contentToString()}") + if (debugIter) println(" compareOneVarIterate cvar=${cvar.name} ${pair.chunkSection} = ${pair.array.shape.contentToString()}") sum2 += sumValues(pair.array) countChunks++ } diff --git a/testfiles/src/test/kotlin/com/sunya/netchdf/CountVersions.kt b/testfiles/src/test/kotlin/com/sunya/netchdf/CountVersions.kt index 22472182..268d2613 100644 --- a/testfiles/src/test/kotlin/com/sunya/netchdf/CountVersions.kt +++ b/testfiles/src/test/kotlin/com/sunya/netchdf/CountVersions.kt @@ -119,7 +119,7 @@ class CountVersions { ncfile.rootGroup().allVariables().forEach { v -> val layout = hdf5File?.layoutName(v) ?: "" - varSizes["$filename#${v.name}#$layout"] = v.nelems + varSizes["$filename#${v.name}#$layout#${ncfile.size/(1000*1000)}"] = v.nelems } } } diff --git a/testfiles/src/test/kotlin/com/sunya/netchdf/hdf5/H5readConcurrentTest.kt b/testfiles/src/test/kotlin/com/sunya/netchdf/hdf5/H5readConcurrentTest.kt index 83e68cf1..f334d282 100644 --- a/testfiles/src/test/kotlin/com/sunya/netchdf/hdf5/H5readConcurrentTest.kt +++ b/testfiles/src/test/kotlin/com/sunya/netchdf/hdf5/H5readConcurrentTest.kt @@ -22,7 +22,7 @@ class H5readConcurrentTest { @Test fun sanity() { - compareChunkReading(testData + "cdmUnitTest/formats/netcdf4/hiig_forec_20140208.nc", "salt") + compareChunkReading(testData + "cdmUnitTest/formats/netcdf4/hiig_forec_20140208.nc", "salt", showStats = true) } // array reading is failing, btree address == -1 @@ -48,24 +48,74 @@ class H5readConcurrentTest { } @Test - fun timeH5readConcurrentThreads() { - val filename = "/home/all/testdata/cdmUnitTest/formats/netcdf4/hiig_forec_20140208.nc" - val varname = "salt" + fun timeH5compareReading() { + val filename = "../core/src/commonTest/data/netcdf4/tiling.nc4" + // val filename = "/home/all/testdata/cdmUnitTest/formats/netcdf4/hiig_forec_20140208.nc" + val varname = "Turbulence_SIGMET_AIRMET" // "salt" Hdf5File(filename).use { myfile : Hdf5File -> println("${myfile.type()} $filename ${myfile.size / 1000.0 / 1000.0} Mbytes") val myvar = myfile.rootGroup().allVariables().find { it.fullname() == varname } ?: throw RuntimeException("cant find $varname") + println(" ${myvar.fullname()}") + val timing = mutableMapOf>() + println("readArrayData") println("nthreads, time in secs") + for (nthreads in listOf(1, 2, 4, 8, 10, 16, 20, 24, 32, 40, 48)) { + myfile.useNThreads = nthreads + val time = measureNanoTime { + myfile.readArrayData(myvar) + } + println("$nthreads, ${time * nano}") + val map1 = timing.getOrPut(nthreads) { mutableMapOf() } + map1["readArrayData"] = time * nano + } + + println("\nchunkIterator") + println("nthreads, time in secs") + for (nthreads in listOf(1, 2, 4, 8, 10, 16, 20, 24, 32, 40, 48)) { + myfile.useNThreads = nthreads + val time = measureNanoTime { + myfile.chunkIterator(myvar) + } + println("$nthreads, ${time * nano}") + val map1 = timing.getOrPut(nthreads) { mutableMapOf() } + map1["chunkIterator"] = time * nano + } + println("\nchunksConcurrent") + println("nthreads, time in secs") for (nthreads in listOf(1, 2, 4, 8, 10, 16, 20, 24, 32, 40, 48)) { + myfile.useNThreads = nthreads val time = measureNanoTime { // fun readChunksConcurrent(v2: Variable, lamda : (ArraySection<*>) -> Unit, done : () -> Unit, nthreads: Int?) { - myfile.readChunksConcurrent(myvar, lamda = { it -> println(" section = ${it.section}") }, { }, wantSection = null, nthreads) + myfile.readChunksConcurrent(myvar, lamda = { }, { }, wantSection = null, nthreads) } println("$nthreads, ${time * nano}") + val map1 = timing.getOrPut(nthreads) { mutableMapOf() } + map1["chunksConcurrent"] = time * nano + } + + val table = buildString { + appendLine("Read chunked data") + appendLine("nthreads, time in secs") + val categories = timing[1]!!.keys + append(" ,") + categories.forEach { append(" $it, ") } + appendLine() + timing.keys.forEach { nthread -> + append(" $nthread, ") + val catTime = timing[nthread]!! + catTime.forEach { cat, time -> + append("$time,") + } + appendLine() + } } + + println() + println(table) } } } diff --git a/testfiles/src/test/kotlin/com/sunya/netchdf/testutils/CompareChunkReading.kt b/testfiles/src/test/kotlin/com/sunya/netchdf/testutils/CompareChunkReading.kt index fb178988..ce76dcb5 100644 --- a/testfiles/src/test/kotlin/com/sunya/netchdf/testutils/CompareChunkReading.kt +++ b/testfiles/src/test/kotlin/com/sunya/netchdf/testutils/CompareChunkReading.kt @@ -16,7 +16,7 @@ import kotlin.test.assertTrue ////////////////////////////////////////////////////////////////////////////////////// // compare reading data regular and through the chunkIterate API -fun compareChunkReading(filename: String, varname : String? = null) { +fun compareChunkReading(filename: String, varname : String? = null, showStats: Boolean = false) { openNetchdfFile(filename).use { myfile -> if (myfile == null) { println("*** not a netchdf file = $filename") @@ -26,10 +26,10 @@ fun compareChunkReading(filename: String, varname : String? = null) { var countChunks = 0 if (varname != null) { val myvar = myfile.rootGroup().allVariables().find { it.fullname() == varname } ?: throw RuntimeException("cant find $varname") - countChunks += compareChunkReadingForVar(myfile, myvar) + countChunks += compareChunkReadingForVar(myfile, myvar, showStats = showStats) } else { myfile.rootGroup().allVariables().forEach { it -> - countChunks += compareChunkReadingForVar(myfile, it) + countChunks += compareChunkReadingForVar(myfile, it, showStats = showStats) } } if (countChunks > 0) { @@ -38,11 +38,18 @@ fun compareChunkReading(filename: String, varname : String? = null) { } } -fun compareChunkReadingForVar(myfile: Netchdf, myvar: Variable<*>): Int { +fun compareChunkReadingForVar(myfile: Netchdf, myvar: Variable<*>, showStats: Boolean): Int { val filename = myfile.location().substringAfterLast('/') println(" ${myvar.nameAndShape()}") Stats.clear() + var sumArrayRead = 0.0 + val time3 = measureNanoTime { + val arrayData = myfile.readArrayData(myvar, null) + sumArrayRead += sumValues(arrayData) + } + Stats.of("readArrayData", filename, "chunk").accum(time3, 1) + var sumChunkIterator = 0.0 var countChunkIterator = 0 val time1 = measureNanoTime { @@ -56,17 +63,9 @@ fun compareChunkReadingForVar(myfile: Netchdf, myvar: Variable<*>): Int { /* if (pair.section.toString().contains("[0:0][0:17][0:97][148:295]")) { println(pair) } */ - } } Stats.of("chunkIterator", filename, "chunk").accum(time1, countChunkIterator) - - var sumArrayRead = 0.0 - val time3 = measureNanoTime { - val arrayData = myfile.readArrayData(myvar, null) - sumArrayRead += sumValues(arrayData) - } - Stats.of("readArrayData", filename, "chunk").accum(time3, 1) assertTrue(nearlyEquals(sumChunkIterator, sumArrayRead), "sumChunkIterator $sumChunkIterator != $sumArrayRead sumArrayRead") if (myfile is Hdf5File) { @@ -87,7 +86,7 @@ fun compareChunkReadingForVar(myfile: Netchdf, myvar: Variable<*>): Int { }, done = { }) } val countConcurrentRead = counta.load() - Stats.of("concurrentSum", filename, "chunk").accum(time2,countConcurrentRead ) + Stats.of("concurrentChunks", filename, "chunk").accum(time2,countConcurrentRead ) val sumConcurrentRead = suma.get() assertTrue( nearlyEquals(sumConcurrentRead, sumArrayRead), @@ -96,7 +95,7 @@ fun compareChunkReadingForVar(myfile: Netchdf, myvar: Variable<*>): Int { } } - // Stats.show() + if (showStats) Stats.show() return countChunkIterator }