ArrayIndexOutOfBoundsException in LeftRight.waitForReaders

[kkris]

First of all, thanks for a nice concurrent hashmap implementation, was glad to find one I could use in my multiplatform app!

Today I got a crash report from production which I narrowed down to a simple reproducing example attached and pasted below. I am not quite sure what is going on, but it reproduces every time so I guess this is some kind of edge case and not a race condition.
There are two examples which I think are related:

• using suspendingWorker an exception is thrown (see below)
• using blockingWorker the workers never finish even though one would expect them to do

Exception:

Exception in thread "DefaultDispatcher-worker-48" java.lang.ArrayIndexOutOfBoundsException: Index 64 out of bounds for length 64
	at io.github.charlietap.leftright.LeftRight.waitForReaders(LeftRight.kt:60)
	at io.github.charlietap.cachemap.InternalCacheMap.put(InternalCacheMap.kt:165)
	at org.example.AppKt.suspendingWorker(App.kt:42)
	at org.example.AppKt$suspendingWorker$1.invokeSuspend(App.kt)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:101)
	at kotlinx.coroutines.internal.LimitedDispatcher$Worker.run(LimitedDispatcher.kt:113)
	at kotlinx.coroutines.scheduling.TaskImpl.run(Tasks.kt:89)
	at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:589)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:823)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:720)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:707)
	Suppressed: kotlinx.coroutines.internal.DiagnosticCoroutineContextException: [StandaloneCoroutine{Cancelling}@1b38f3d3, Dispatchers.Default]

Example project to reproduce the issue. Use ./gradlew run to execute.
cachemap-bug-reproducer.zip

@OptIn(DelicateCoroutinesApi::class)
fun main(args: Array<String>) {
    val cache: CacheMap<String, String> = cacheMapOf()

    repeat(64) {
        GlobalScope.launch {
            withContext(Dispatchers.IO) {
                suspendingWorker(cache)

                // Note: Alternatively, using this worker seems to unexpectedly block at some point
                // blockingWorker(cache)
            }
        }
    }

    Thread.sleep(100_000)
}

suspend fun suspendingWorker(cache: CacheMap<String, String>) {
    while (true) {
        val key = Random.nextInt(1000).toString()

        val existing = cache.get(key)
        if (existing == null) {
            val value = GlobalScope.async {
                UUID.randomUUID().toString()
            }.await()

            cache.put(key, value)
        }

        if (cache.size >= 998) {
            break
        }
    }
}

fun blockingWorker(cache: CacheMap<String, String>) {
    while (true) {
        val key = Random.nextInt(1000).toString()
        val existing = cache.get(key)
        if (existing == null) {
            cache.put(key, UUID.randomUUID().toString())
        }

        if (cache.size >= 998) {
            // Note: code never reaches this break even though one would expect the cache to fill up at some point
            break
        }

        println(cache.size)
    }
}

[CharlieTap]

Hey @kkris,

Apologies for the slow reply to this, the reason you're seeing this is because chasm has a default around the amount of distinct reader threads that can access the hashmap.

The default logic is roughly this:

maxOf(64, (4* The number of cpu cores available))

Given you're seeing the issue at 64 threads, it sounds like you're running this on 16 core chip or less. My first thought is that it seems a little odd that you're spawning so many threads. Given this is cpu bound work normally I would say this is overkill but I don't know the details so I'm just going to provide you with solutions:

Option 1:

Try using only Dispatcher.Default when accessing cachemap

Option 2:

You could create your own dispatcher just for cachemap

val limitedDispatcher = Executors.newFixedThreadPool(64).asCoroutineDispatcher()

Option 3:

If you know the exact amount of threads involved ahead of time cachemap has a constructor which takes the readerMaxParallism

fun <K, V> suspendCacheMapOf(readerParallelism: Int, initialCapacity: Int, initialPopulation: Map<K, V>): SuspendCacheMap<K, V> {
    return InternalSuspendCacheMap(readerParallelism, initialCapacity, initialPopulation)
}