reenable native KMP.

Readme.md

@@ -59,10 +59,10 @@ Also see:
 We use the latest LTS (long term support) Java version, and will not be explicitly supporting older versions.
 Currently that is Java 21.
 
-We also use the latest stable version of Kotlin that is compatible with the Java version. Currently that is Kotlin 2.1.
+We also use the latest stable version of Kotlin that is compatible with the Java version. Currently that is Kotlin 2.1.21.
 
 Gradle is our build system. We will use the latest stable version of Gradle compatible with our Java and Kotlin versions.
-Currently that is Gradle 8.14.
+Currently that is Gradle 8.14.3.
 
 For now, you must download and build the library yourself. Eventually we will publish it to Maven Central.
 The IntelliJ IDE is highly recommended for all JVM development.
@@ -265,7 +265,7 @@ For example, a Variable of datatype Float will return an ArrayFloat, which is Ar
     return data as ArrayUByte. 
   * Netcdf-4 encodes CHAR values as HDF5 string type with elemSize = 1, so we use that convention to detect 
     legacy CHAR variables in HDF5 format. (NC_CHAR should not be used in new Netcdf-4 files, use NC_UBYTE or NC_STRING.) 
-    Variables of type CHAR return data as STRING, since users can use UBYTE if thats what they intend.
+    Variables of type CHAR return data as STRING, since users can use UBYTE if that's what they intend.
   * Netcdf-4/HDF5 String variables may be fixed or variable length. For fixed Strings, we set the size of Datatype.STRING to 
     the fixed size. For both fixed and variable length Strings, the string will be truncated at the first zero byte, if any.
   * HDF4 does not have a STRING type, but does have signed and unsigned CHAR, and signed and unsigned BYTE. 
@@ -318,7 +318,7 @@ isNetcdf4 = true.
     1.  If a group or variable has an attribute with name "_NCProperties", "_Netcdf4Coordinates", "_Netcdf4Dimid" or "_nc3_strict".
     2.  If a variable name starts with "_nc4_non_coord_".
     3.  If a variable has an attrinute named "DIMENSION_LIST with type vlen of reference.
-    4.  If a dimenson name starts with "This is a netCDF dimension but not a netCDF variable"
+    4.  If a dimension name starts with "This is a netCDF dimension but not a netCDF variable"
 
 Other than trying to identify which library wrote the file, Netchdf does not do any special processing for Netcdf4 files,
 except:

core/build.gradle.kts

@@ -9,7 +9,6 @@ version = libs.versions.netchdf.get()
 kotlin {
     jvm()
 
-    /*
     val hostOs = System.getProperty("os.name")
     val arch = System.getProperty("os.arch")
     when {
@@ -29,8 +28,6 @@ kotlin {
         else -> throw GradleException("Host OS is not supported.")
     }
 
-     */
-
     sourceSets {
         val commonMain by getting {
             dependencies {

core/src/commonMain/kotlin/com/sunya/cdm/api/Dimension.kt

@@ -12,10 +12,9 @@ data class Dimension(val orgName : String, val length : Long, val isShared : Boo
     constructor(len : Int) : this("", len.toLong(), false)
     constructor(len : Long) : this("", len, false)
 
-    // orgName may be different
     override fun equals(other: Any?): Boolean {
         if (this === other) return true
-        if (javaClass != other?.javaClass) return false
+        if (other == null || this::class != other::class) return false
 
         other as Dimension
 
@@ -33,5 +32,4 @@ data class Dimension(val orgName : String, val length : Long, val isShared : Boo
         return result
     }
 
-
 }

core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/BTree2.kt

@@ -210,7 +210,9 @@ fun getSizeOfNumberOfRecords(
 
 // jhdf
 internal fun bytesNeededToHoldNumber(number: Int): Int {
-    return (Integer.numberOfTrailingZeros(Integer.highestOneBit(number)) + 8) / 8
+    //     return (Integer.numberOfTrailingZeros(Integer.highestOneBit(number)) + 8) / 8
+    val p1 = number.takeHighestOneBit()
+    return (p1.countTrailingZeroBits() + 8) / 8
 }
 
 /* private fun getSizeOfTotalNumberOfChildRecords(nodeSize: Int, depth: Int, recordSize: Int): Int {

core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/BTree2data.kt

@@ -63,6 +63,8 @@ internal class BTree2data(
         }
     }
 
+    fun countChunks() = asSequence().count()
+
     fun chunkIterator(): Iterator<DataChunk> = asSequence().iterator()
 
     internal fun findDataChunk(order: Int): DataChunk? {

core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/ExtensibleArrayIndex.kt

@@ -5,7 +5,6 @@ package com.sunya.netchdf.hdf5
 import com.sunya.cdm.api.computeSize
 import com.sunya.cdm.iosp.OpenFileState
 import com.sunya.cdm.util.InternalLibraryApi
-import java.util.*
 
 // DataLayoutExtensibleArray4
 class ExtensibleArrayIndex(val h5: H5builder, address: Long, datasetDimensions: IntArray, chunkDimensions: IntArray) {

core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/FractalHeap.kt

@@ -7,6 +7,8 @@ import com.sunya.cdm.iosp.OpenFileIF
 import com.sunya.cdm.iosp.OpenFileState
 import com.sunya.cdm.util.*
 
+val UNDEFINED_ADDRESS = -1L
+
 /** Level 1G - Fractal Heap  */
 // TODO rewrite
 internal class FractalHeap(private val h5: H5builder, forWho: String, address: Long) {

core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/H5readChunkedConcurrent.kt

@@ -18,6 +18,7 @@ import com.sunya.cdm.util.InternalLibraryApi
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.channels.produce
@@ -44,14 +45,38 @@ class H5readChunkedConcurrent<T>(val h5: H5builder, val v2: Variable<T>, wantSec
         if (vinfo.mdl is DataLayoutBTreeVer1) {
             val mdl = vinfo.mdl
             chunks = BTree1data(rafext, mdl.btreeAddress, varShape, mdl.chunkDims.toLongArray())
+            println(" BTree1data nchunks = ${chunks.countChunks()}")
         } else if (vinfo.mdl is DataLayoutBtreeVer2) {
             chunks = BTree2data(rafext, v2.name, vinfo.dataPos, v2.shape, vinfo.storageDims)
+            println(" BTree2data nchunks = ${chunks.countChunks()}")
         } else {
-            throw RuntimeException("H5chunkConcurrent cant read ${vinfo.mdl.javaClass.simpleName}")
+            throw RuntimeException("H5chunkConcurrent cant read ${vinfo.mdl::class}")
         }
     }
 
+    // Start the backgroup threads then return immediately
+
     fun readChunks(nthreads: Int, lamda: (ArraySection<T>) -> Unit, done: () -> Unit) {
+        println(" readChunks with  nthreads = $nthreads")
+
+        GlobalScope.launch(Dispatchers.Default) {
+            val jobs = mutableListOf<Job>()
+            val workers = mutableListOf<Worker>()
+            val chunkProducer = produceChunks(chunks.asSequence())
+            repeat(nthreads) {
+                val worker = Worker()
+                jobs.add(launchJob(worker, chunkProducer, lamda))
+                workers.add(worker)
+            }
+
+            // wait for all jobs to be done, then close everything
+            joinAll(*jobs.toTypedArray())
+            workers.forEach { it.rafext.close() }
+            done()
+        }
+    }
+
+    fun readChunksBlocking(nthreads: Int, lamda: (ArraySection<T>) -> Unit, done: () -> Unit) {
 
         runBlocking {
             val jobs = mutableListOf<Job>()

core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/Hdf5File.kt

@@ -39,7 +39,7 @@ class Hdf5File(val filename : String, strict : Boolean = false) : Netchdf {
             return("DataContainerAttribute")
         }
         val vinfo = (v.spObject as DataContainerVariable)
-        return vinfo.mdl.javaClass.simpleName
+        return vinfo.mdl::class.simpleName ?: "none"
     }
 
     var useNThreads : Int? = null
@@ -166,10 +166,11 @@ class Hdf5File(val filename : String, strict : Boolean = false) : Netchdf {
         val deque = Deque<ArraySection<T>>(10)
 
         init {
+            // we start up the concurrent read, but its blocking here
             reader.readChunks(
                 nthreads,
                 lamda = { deque.add(it) },
-                done = { deque.done() }
+                done = { deque.complete() }
             )
         }
 
@@ -187,8 +188,7 @@ class Hdf5File(val filename : String, strict : Boolean = false) : Netchdf {
                                           wantSection: SectionPartial?, nthreads: Int?) {
         val reader = H5readChunkedConcurrent(header, v2, wantSection)
         val availableProcessors = this.useNThreads()
-        // println("availableProcessors = $availableProcessors")
-        reader.readChunks(nthreads ?: availableProcessors, lamda, done = { done() })
+        reader.readChunksBlocking(nthreads ?: availableProcessors, lamda, done)
     }
 
 }

core/src/commonMain/kotlin/com/sunya/netchdf/util/Kmp.kt

@@ -1,11 +1,10 @@
 package com.sunya.netchdf.util
 
-
 expect fun useDefaultNThreads(): Int
 
 // add to end, take from head
 expect class Deque<T>(initialCapacity: Int) {
     fun add(item: T)
     fun next(): T?
-    fun done()
+    fun complete()
 }

core/src/commonTest/kotlin/com/sunya/netchdf/hdf5/Btree1dataTest.kt

@@ -5,9 +5,7 @@ package com.sunya.netchdf.hdf5
 import com.sunya.cdm.api.ArraySection
 import com.sunya.cdm.api.toLongArray
 import com.sunya.cdm.util.InternalLibraryApi
-import com.sunya.netchdf.testutil.nano
 import com.sunya.netchdf.testutil.testData
-import kotlin.system.measureNanoTime
 import kotlin.test.Test
 
 class Btree1dataTest {
@@ -58,15 +56,15 @@ class Btree1dataTest {
             println("nthreads,   time in secs")
 
             for (nthreads in listOf(1, 2, 4, 8, 10, 16, 20, 24, 32, 40, 48)) {
-                val time = measureNanoTime {
+               // val time = measureNanoTime {
                     //     fun readChunks(nthreads: Int, lamda: (ArraySection<*>) -> Unit, done: () -> Unit) {
                     val reader = H5readChunkedConcurrent(myfile.header, myvar, null)
                     reader.readChunks(nthreads, lamda = { asect: ArraySection<*> ->
                         // println(" section = ${asect.chunkSection}")
                     }, { }, )
                 }
-                println("$nthreads, ${time * nano}")
-            }
+                //println("$nthreads, ${time * nano}")
+            //}
         }
     }
 

core/src/commonTest/kotlin/com/sunya/netchdf/hdf5/JhdfReadTest.kt

@@ -7,7 +7,7 @@ import com.sunya.netchdf.testutil.testData
 
 class JhdfReadTest {
     companion object {
-        @JvmStatic
+        // @JvmStatic
         fun files(): Iterator<String> {
             return JhdfFiles.files()
         }

core/src/commonTest/kotlin/com/sunya/netchdf/testutil/TestFiles.kt

@@ -3,6 +3,7 @@ package com.sunya.netchdf.testutil
 import okio.FileSystem
 import okio.Path
 import okio.Path.Companion.toPath
+import okio.SYSTEM
 
 const val testData = "../core/src/commonTest/data/"
 

core/src/jvmMain/kotlin/com/sunya/netchdf/util/Kmp.kt

@@ -1,48 +1,51 @@
 package com.sunya.netchdf.util
 
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
 import java.lang.Runtime.*
 import java.util.concurrent.ConcurrentLinkedDeque
+import kotlin.RuntimeException
 
 // todo: i have 2 threads per processer, these dont help the IO i think.
 actual fun useDefaultNThreads(): Int {
     return getRuntime().availableProcessors() / 2
 }
 
+private const val debug = false
+
 actual class Deque<T> actual constructor(initialCapacity: Int) {
     val delegate = ConcurrentLinkedDeque<T>()
-    var done = false
+    var complete = false
 
     actual fun add(item: T) {
+        if (debug) println(" Deque add")
         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()
+        while (countWaits < 1000) {
+            val firstElement = delegate.pollFirst()
             if (firstElement != null) {
+                if (debug) println(" Deque got element")
                 return firstElement
-            } else if (done) {
-                done()
-                return null
-            } else if (countWaits > 100) {
-                println("takes too long (10 sec)")
-                done()
+            } else if (complete) {
+                if (debug) println(" Deque complete")
                 return null
             } else {
-                // wait 100 msecs
-                Thread.sleep(100) // java only
+                if (debug) println(" Deque wait")
+                runBlocking {
+                    delay(10)
+                }
                 countWaits++
             }
         }
-
+        throw RuntimeException("Deque next timed out (10 sec)")
     }
 
-    actual fun done() {
-        done = true
+    actual fun complete() {
+        if (debug) println(" Deque complete was called")
+        complete = true
     }
 }

core/src/commonMain/kotlin/com/sunya/netchdf/hdf5/FractalHeapJ.kt → core/src/jvmTest/kotlin/com/sunya/netchdf/hdf5/FractalHeapJ.kt

@@ -312,5 +312,5 @@ fun bytesNeededToHoldNumber(number: Long): Int {
     if (number == 0L) {
         return 1
     }
-    return ceil(java.math.BigInteger.valueOf(number).bitLength() / 8.0).toInt()
+    return ceil(BigInteger.valueOf(number).bitLength() / 8.0).toInt()
 }