ronantakizawa/codereview-bench huggingface dataset:
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/native/src/MultithreadedDispatchers.kt (Kotlin)

Code before:
    override fun dispatch(context: CoroutineContext, block: Runnable) {
        if (activeWorkers.value != workersCount) {
            tryAddWorker()
        }

        tasksQueue.trySend(block).onClosed {
            throw IllegalStateException("Dispatcher $name was closed, attempted to schedule : $block")
        }
    }

    private fun tryAddWorker() {
        activeWorkers.loop { curWorkers ->
            if (curWorkers >= workersCount) return
            if (activeWorkers.compareAndSet(curWorkers, curWorkers + 1)) {
                val worker = Worker.start(name = "$name-$curWorkers")
                worker.executeAfter { workerRunLoop() }
                workers[curWorkers].value = worker
                return
            }
        }
    }

    override fun close() {
        tasksQueue.close()
        /*
         * Here we cannot avoid not to wait on `.result`, otherwise it will lead
         * to a native memory leak, including pthread handle.
         */
        val requests = Array(workersCount) {
            workers[it].value?.requestTermination()
        }
        requests.map { it?.result }
    }
}


Review comment (suggestion):
```suggestion
         * Here we cannot avoid waiting on `.result`, otherwise it will lead
```

Code after:
        // .receive() without proper evaluation
        for (task in tasksQueue) {
            /**
             * Any unhandled exception here will pass through worker's boundary and will be properly reported.
             */
            task.run()
        }
    }

    override fun dispatch(context: CoroutineContext, block: Runnable) {
        fun throwClosed(block: Runnable) {
            throw IllegalStateException("Dispatcher $name was closed, attempted to schedule: $block")
        }

        if (!workerPool.allocate()) throwClosed(block) // Do not even try to send to avoid race

        tasksQueue.trySend(block).onClosed {
            throwClosed(block)
        }
    }

    override fun close() {
        val workers = workerPool.close()
        tasksQueue.close()
        /*
         * Here we cannot avoid waiting on `.result`, otherwise it will lead
         * to a native memory leak, including a pthread handle.
         */
        val requests = workers.map { it.requestTermination() }
        requests.map { it.result }
    }
}
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/native/src/MultithreadedDispatchers.kt (Kotlin)

Code before:
        if (activeWorkers.value != workersCount) {
            tryAddWorker()
        }

        tasksQueue.trySend(block).onClosed {
            throw IllegalStateException("Dispatcher $name was closed, attempted to schedule : $block")
        }
    }

    private fun tryAddWorker() {
        activeWorkers.loop { curWorkers ->
            if (curWorkers >= workersCount) return
            if (activeWorkers.compareAndSet(curWorkers, curWorkers + 1)) {
                val worker = Worker.start(name = "$name-$curWorkers")
                worker.executeAfter { workerRunLoop() }
                workers[curWorkers].value = worker
                return
            }
        }
    }

    override fun close() {
        tasksQueue.close()
        /*
         * Here we cannot avoid not to wait on `.result`, otherwise it will lead
         * to a native memory leak, including pthread handle.
         */
        val requests = Array(workersCount) {
            workers[it].value?.requestTermination()
        }
        requests.map { it?.result }
    }
}


Review comment (performance):
```suggestion
         * to a native memory leak, including a pthread handle.
```

Code after:
        for (task in tasksQueue) {
            /**
             * Any unhandled exception here will pass through worker's boundary and will be properly reported.
             */
            task.run()
        }
    }

    override fun dispatch(context: CoroutineContext, block: Runnable) {
        fun throwClosed(block: Runnable) {
            throw IllegalStateException("Dispatcher $name was closed, attempted to schedule: $block")
        }

        if (!workerPool.allocate()) throwClosed(block) // Do not even try to send to avoid race

        tasksQueue.trySend(block).onClosed {
            throwClosed(block)
        }
    }

    override fun close() {
        val workers = workerPool.close()
        tasksQueue.close()
        /*
         * Here we cannot avoid waiting on `.result`, otherwise it will lead
         * to a native memory leak, including a pthread handle.
         */
        val requests = workers.map { it.requestTermination() }
        requests.map { it.result }
    }
}
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-test/jvm/test/MemoryLeakTest.kt (Kotlin)

Code before:
/*
 * Copyright 2016-2022 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
 */
import kotlinx.coroutines.*
import kotlinx.coroutines.test.*
import java.lang.ref.*
import kotlin.test.*

class MemoryLeakTest {

    @Test
    fun testCancellationLeakInTestCoroutineScheduler() = runTest {
        lateinit var weakRef: WeakReference<*>
        val job = launch(start = CoroutineStart.UNDISPATCHED) {
            val leakingObject = Any()
            weakRef = WeakReference(leakingObject)
            delay(3)
            // This code is needed to hold a reference to `leakingObject` until the job itself is weakly reachable.
            leakingObject.hashCode()
        }
        job.cancel()
        System.gc()
        assertNotNull(weakRef.get()) // the cancellation handler has not run yet
        runCurrent()
        System.gc()
        assertNull(weakRef.get()) // the cancellation handler has run, disposing of the job
    }
}


Review comment (suggestion):
`System.gc()` is not the most reliable tool.

We have deterministic tests on that matter, see `FieldWalker` usages.

Code after:
/*
 * Copyright 2016-2022 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
 */
import kotlinx.coroutines.*
import kotlinx.coroutines.test.*
import kotlin.test.*

class MemoryLeakTest {

    @Test
    fun testCancellationLeakInTestCoroutineScheduler() = runTest {
        val leakingObject = Any()
        val job = launch(start = CoroutineStart.UNDISPATCHED) {
            delay(1)
            // This code is needed to hold a reference to `leakingObject` until the job itself is weakly reachable.
            leakingObject.hashCode()
        }
        job.cancel()
        FieldWalker.assertReachableCount(1, testScheduler) { it === leakingObject }
        runCurrent()
        FieldWalker.assertReachableCount(0, testScheduler) { it === leakingObject }
    }
}
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/jvm/test/TestBase.kt (Kotlin)

Code before:
            ex = e
            if (expected != null) {
                if (!expected(e))
                    error("Unexpected exception: $e", e)
            } else {
                throw e
            }
        } finally {
            if (ex == null && expected != null) error("Exception was expected but none produced")
        }
        if (exCount < unhandled.size)
            error("Too few unhandled exceptions $exCount, expected ${unhandled.size}")
    }

    protected inline fun <reified T: Throwable> assertFailsWith(block: () -> Unit): T {
        val result = runCatching(block)
        assertTrue(result.exceptionOrNull() is T, "Expected ${T::class}, but had $result")
        return result.exceptionOrNull()!! as T
    }

    protected suspend fun currentDispatcher() = coroutineContext[ContinuationInterceptor]!!
}

/*
 * We ignore tests that test **real** non-virtualized tests with time on Windows, because
 * our TC Windows is virtualized itself (oh, the irony) and its clock resolution is dozens of ms,
 * which makes such tests flaky.
 */
public actual val isJavaAndWindows: Boolean = System.getProperty("os.name")!!.contains("Windows")


Review comment (suggestion):
```suggestion
 * our CI Windows is virtualized itself (oh, the irony) and its clock resolution is dozens of ms,
```
Should be more clear to external contributors.

Code after:
            ex = e
            if (expected != null) {
                if (!expected(e))
                    error("Unexpected exception: $e", e)
            } else {
                throw e
            }
        } finally {
            if (ex == null && expected != null) error("Exception was expected but none produced")
        }
        if (exCount < unhandled.size)
            error("Too few unhandled exceptions $exCount, expected ${unhandled.size}")
    }

    protected inline fun <reified T: Throwable> assertFailsWith(block: () -> Unit): T {
        val result = runCatching(block)
        assertTrue(result.exceptionOrNull() is T, "Expected ${T::class}, but had $result")
        return result.exceptionOrNull()!! as T
    }

    protected suspend fun currentDispatcher() = coroutineContext[ContinuationInterceptor]!!
}

/*
 * We ignore tests that test **real** non-virtualized tests with time on Windows, because
 * our CI Windows is virtualized itself (oh, the irony) and its clock resolution is dozens of ms,
 * which makes such tests flaky.
 */
public actual val isJavaAndWindows: Boolean = System.getProperty("os.name")!!.contains("Windows")
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/jvm/test/knit/ClosedAfterGuideTestExecutor.kt (Kotlin)

Code before:
private class ClosedAfterGuideTestDispatcher(
    private val nThreads: Int,
    private val name: String
) : ExecutorCoroutineDispatcher() {
    private val threadNo = AtomicInteger()

    override val executor: Executor =
        Executors.newScheduledThreadPool(nThreads, object : ThreadFactory {
            override fun newThread(target: java.lang.Runnable): Thread {
                return PoolThread(
                    this@ClosedAfterGuideTestDispatcher,
                    target,
                    if (nThreads == 1) name else name + "-" + threadNo.incrementAndGet()
                )
            }
        })

    override fun dispatch(context: CoroutineContext, block: Runnable) {
        executor.execute(wrapTask(block))
    }

    override fun close() {
        (executor as ExecutorService).apply {
            shutdown()
            awaitTermination(1, TimeUnit.MINUTES)
        }
    }

    override fun toString(): String = "ThreadPoolDispatcher[$nThreads, $name]"
}


Review comment (suggestion):
No strong opinion about this change, I just found it severely surprising that there's no way to wait for the tasks to finish executing. Can undo this, move the corresponding test to native-only, and change the doc to state that the previously-submitted tasks may execute *in parallel* to the caller of `close()`.

Code after:
private class ClosedAfterGuideTestDispatcher(
    private val nThreads: Int,
    private val name: String
) : ExecutorCoroutineDispatcher() {
    private val threadNo = AtomicInteger()

    override val executor: Executor =
        Executors.newScheduledThreadPool(nThreads, object : ThreadFactory {
            override fun newThread(target: java.lang.Runnable): Thread {
                return PoolThread(
                    this@ClosedAfterGuideTestDispatcher,
                    target,
                    if (nThreads == 1) name else name + "-" + threadNo.incrementAndGet()
                )
            }
        })

    override fun dispatch(context: CoroutineContext, block: Runnable) {
        executor.execute(wrapTask(block))
    }

    override fun close() {
        (executor as ExecutorService).shutdown()
    }

    override fun toString(): String = "ThreadPoolDispatcher[$nThreads, $name]"
}
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/native/test/MultithreadedDispatchersTest.kt (Kotlin)

Code before:
/*
 * Copyright 2016-2023 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
 */

package kotlinx.coroutines

import kotlin.native.concurrent.*
import kotlin.test.*

class MultithreadedDispatchersTest {
    /**
     * Test that [newFixedThreadPoolContext] does not allocate more dispatchers than it needs to.
     * Incidentally also tests that it will allocate enough workers for its needs. Otherwise, the test will hang.
     */
    @Test
    fun testNotAllocatingExtraDispatchers() {
        suspend fun spin(set: MutableSet<Worker>) {
            repeat(100) {
                set.add(Worker.current)
                delay(1)
            }
        }
        val dispatcher = newFixedThreadPoolContext(64, "test")
        try {
            runBlocking {
                val encounteredWorkers = mutableSetOf<Worker>()
                var canStart1 = false
                var canStart2 = false
                val coroutine1 = launch(dispatcher) {
                    while (!canStart1) {
                        // intentionally empty
                    }
                    canStart2 = true
                    spin(encounteredWorkers)
                }
                val coroutine2 = launch(dispatcher) {
                    canStart1 = true
                    while (!canStart2) {
                        // intentionally empty
                    }
                    spin(encounteredWorkers)
                }
                listOf(coroutine1, coroutine2).joinAll()
                assertEquals(2, encounteredWorkers.size)
            }
        } finally {
            dispatcher.close()
        }
    }
}


Review comment (bug):
This test failed in the CI with `encounteredWorkers.size == 1`, but I fail to see how it's possible. Can different workers be equal to each other?

Code after:
        val barrier = BlockingBarrier(2)
        val lock = SynchronizedObject()
        suspend fun spin(set: MutableSet<Worker>) {
            repeat(100) {
                synchronized(lock) { set.add(Worker.current) }
                delay(1)
            }
        }
        val dispatcher = newFixedThreadPoolContext(64, "test")
        try {
            runBlocking {
                val encounteredWorkers = mutableSetOf<Worker>()
                val coroutine1 = launch(dispatcher) {
                    barrier.await()
                    spin(encounteredWorkers)
                }
                val coroutine2 = launch(dispatcher) {
                    barrier.await()
                    spin(encounteredWorkers)
                }
                listOf(coroutine1, coroutine2).joinAll()
                assertEquals(2, encounteredWorkers.size)
            }
        } finally {
            dispatcher.close()
        }
    }
}
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/src/channels/Channels.common.kt (Kotlin)

Code before:
/*
 * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
 */
@file:JvmMultifileClass
@file:JvmName("ChannelsKt")
@file:Suppress("DEPRECATION_ERROR", "DEPRECATION")
@file:OptIn(ExperimentalContracts::class)

package kotlinx.coroutines.channels

import kotlinx.coroutines.*
import kotlinx.coroutines.selects.*
import kotlin.contracts.*
import kotlin.jvm.*

internal const val DEFAULT_CLOSE_MESSAGE = "Channel was closed"


// -------- Operations on BroadcastChannel --------

/**
 * Opens subscription to this [BroadcastChannel] and makes sure that the given [block] consumes all elements
 * from it by always invoking [cancel][ReceiveChannel.cancel] after the execution of the block.
 *
 * **Note: This API is obsolete since 1.5.0 and deprecated for removal since 1.7.0**
 * It is replaced with [SharedFlow][kotlinx.coroutines.flow.SharedFlow].
 *
 * Safe to remove in 1.9.0 as was inline before.
 */
@ObsoleteCoroutinesApi
@Deprecated(level = DeprecationLevel.WARNING, message = "BroadcastChannel is deprecated in the favour of SharedFlow and is no longer supported")

Review comment (suggestion):
I don't like this `Suppress`, it could obscure some real issues eventually. I checked, and the compiler seems pleased if we just add the `Suppress` to each deprecated method separately: https://gist.github.com/dkhalanskyjb/de271eff5aa3286fd680769e080696f8

Code after:
/*
 * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
 */
@file:JvmMultifileClass
@file:JvmName("ChannelsKt")
@file:OptIn(ExperimentalContracts::class)

package kotlinx.coroutines.channels

import kotlinx.coroutines.*
import kotlinx.coroutines.selects.*
import kotlin.contracts.*
import kotlin.jvm.*

internal const val DEFAULT_CLOSE_MESSAGE = "Channel was closed"


// -------- Operations on BroadcastChannel --------

/**
 * Opens subscription to this [BroadcastChannel] and makes sure that the given [block] consumes all elements
 * from it by always invoking [cancel][ReceiveChannel.cancel] after the execution of the block.
 *
 * **Note: This API is obsolete since 1.5.0 and deprecated for removal since 1.7.0**
 * It is replaced with [SharedFlow][kotlinx.coroutines.flow.SharedFlow].
 *
 * Safe to remove in 1.9.0 as was inline before.
 */
@ObsoleteCoroutinesApi
@Suppress("DEPRECATION")
@Deprecated(level = DeprecationLevel.WARNING, message = "BroadcastChannel is deprecated in the favour of SharedFlow and is no longer supported")
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/src/CoroutineContext.common.kt (Kotlin)

Code before:
/*
 * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
 */

package kotlinx.coroutines

import kotlin.coroutines.*

/**
 * Creates a context for a new coroutine. It installs [Dispatchers.Default] when no other dispatcher or
 * [ContinuationInterceptor] is specified and adds optional support for debugging facilities (when turned on)
 * and copyable-thread-local facilities on JVM.
 */
public expect fun CoroutineScope.newCoroutineContext(context: CoroutineContext): CoroutineContext

/**
 * Creates a context for coroutine builder functions that do not launch a new coroutine, e.g. [withContext].
 * @suppress
 */
@InternalCoroutinesApi
public expect fun CoroutineContext.newCoroutineContext(addedContext: CoroutineContext): CoroutineContext

@Suppress("PropertyName")
internal expect val DefaultDelay: Delay

// countOrElement -- pre-cached value for ThreadContext.kt
internal expect inline fun <T> withCoroutineContext(context: CoroutineContext, countOrElement: Any?, block: () -> T): T
internal expect inline fun <T> withContinuationContext(continuation: Continuation<*>, countOrElement: Any?, block: () -> T): T
internal expect fun Continuation<*>.toDebugString(): String
internal expect val CoroutineContext.coroutineName: String?


Review comment (suggestion):
Please leave it on actuals as well.
It is used not for the sake of inlining but in order to access it with the help of `@Suppress` from a different KLib. `@PublishedApi` guarantees a stable unmangled name here

Code after:
/*
 * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
 */

package kotlinx.coroutines

import kotlin.coroutines.*

/**
 * Creates a context for a new coroutine. It installs [Dispatchers.Default] when no other dispatcher or
 * [ContinuationInterceptor] is specified and adds optional support for debugging facilities (when turned on)
 * and copyable-thread-local facilities on JVM.
 */
public expect fun CoroutineScope.newCoroutineContext(context: CoroutineContext): CoroutineContext

/**
 * Creates a context for coroutine builder functions that do not launch a new coroutine, e.g. [withContext].
 * @suppress
 */
@InternalCoroutinesApi
public expect fun CoroutineContext.newCoroutineContext(addedContext: CoroutineContext): CoroutineContext

@PublishedApi // to have unmangled name when using from other modules via suppress
@Suppress("PropertyName")
internal expect val DefaultDelay: Delay

// countOrElement -- pre-cached value for ThreadContext.kt
internal expect inline fun <T> withCoroutineContext(context: CoroutineContext, countOrElement: Any?, block: () -> T): T
internal expect inline fun <T> withContinuationContext(continuation: Continuation<*>, countOrElement: Any?, block: () -> T): T
internal expect fun Continuation<*>.toDebugString(): String
internal expect val CoroutineContext.coroutineName: String?
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/src/CoroutineContext.common.kt (Kotlin)

Code before:
/*
 * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
 */

package kotlinx.coroutines

import kotlin.coroutines.*

/**
 * Creates a context for a new coroutine. It installs [Dispatchers.Default] when no other dispatcher or
 * [ContinuationInterceptor] is specified and adds optional support for debugging facilities (when turned on)
 * and copyable-thread-local facilities on JVM.
 */
public expect fun CoroutineScope.newCoroutineContext(context: CoroutineContext): CoroutineContext

/**
 * Creates a context for coroutine builder functions that do not launch a new coroutine, e.g. [withContext].
 * @suppress
 */
@InternalCoroutinesApi
public expect fun CoroutineContext.newCoroutineContext(addedContext: CoroutineContext): CoroutineContext

@PublishedApi // to have unmangled name when using from other modules
@Suppress("PropertyName")
internal expect val DefaultDelay: Delay

// countOrElement -- pre-cached value for ThreadContext.kt
internal expect inline fun <T> withCoroutineContext(context: CoroutineContext, countOrElement: Any?, block: () -> T): T
internal expect inline fun <T> withContinuationContext(continuation: Continuation<*>, countOrElement: Any?, block: () -> T): T
internal expect fun Continuation<*>.toDebugString(): String
internal expect val CoroutineContext.coroutineName: String?


Review comment (suggestion):
```suggestion
@PublishedApi // to have unmangled name when using from other modules via suppress
```

Code after:
/*
 * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
 */

package kotlinx.coroutines

import kotlin.coroutines.*

/**
 * Creates a context for a new coroutine. It installs [Dispatchers.Default] when no other dispatcher or
 * [ContinuationInterceptor] is specified and adds optional support for debugging facilities (when turned on)
 * and copyable-thread-local facilities on JVM.
 */
public expect fun CoroutineScope.newCoroutineContext(context: CoroutineContext): CoroutineContext

/**
 * Creates a context for coroutine builder functions that do not launch a new coroutine, e.g. [withContext].
 * @suppress
 */
@InternalCoroutinesApi
public expect fun CoroutineContext.newCoroutineContext(addedContext: CoroutineContext): CoroutineContext

@PublishedApi // to have unmangled name when using from other modules via suppress
@Suppress("PropertyName")
internal expect val DefaultDelay: Delay

// countOrElement -- pre-cached value for ThreadContext.kt
internal expect inline fun <T> withCoroutineContext(context: CoroutineContext, countOrElement: Any?, block: () -> T): T
internal expect inline fun <T> withContinuationContext(continuation: Continuation<*>, countOrElement: Any?, block: () -> T): T
internal expect fun Continuation<*>.toDebugString(): String
internal expect val CoroutineContext.coroutineName: String?