ronantakizawa/codereview-bench huggingface dataset:
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/jvm/src/internal/ResizableAtomicArray.kt (Kotlin)

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

package kotlinx.coroutines.internal

import kotlinx.coroutines.*
import java.util.concurrent.atomic.*

/**
 * Atomic array with lock-free reads and synchronized modifications. It logically has an unbounded size,
 * is implicitly filled with nulls, and is resized on updates as needed to grow.
 */
internal class ResizableAtomicArray<T>(initialLength: Int) {
    @Volatile
    private var array = AtomicReferenceArray<T>(initialLength)

    // for debug output
    public fun currentLength(): Int = array.length()

    public operator fun get(index: Int): T? {
        assert { index >= 0 }
        val array = this.array // volatile read
        return if (index < array.length()) array[index] else null
    }

    @Synchronized
    operator fun set(index: Int, value: T?) {
        assert { index >= 0 }
        val curArray = this.array
        val curLen = curArray.length()
        if (index < curLen) {
            curArray[index] = value
        } else {
            val newArray = AtomicReferenceArray<T>((index + 1).coerceAtLeast(2 * curLen))
            for (i in 0 until curLen) newArray[i] = curArray[i]
            newArray[index] = value
            array = newArray // copy done
        }
    }
}


Review comment (suggestion):
`set` is always guarded by `synchronized(workers)`.

It's better to avoid excessive synchronization; alternatively, you can do `@GuardedBy("this")` or `assert { Thread.holdsLock(this) }`

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

package kotlinx.coroutines.internal

import java.util.concurrent.atomic.*

/**
 * Atomic array with lock-free reads and synchronized modifications. It logically has an unbounded size,
 * is implicitly filled with nulls, and is resized on updates as needed to grow.
 */
internal class ResizableAtomicArray<T>(initialLength: Int) {
    @Volatile
    private var array = AtomicReferenceArray<T>(initialLength)

    // for debug output
    public fun currentLength(): Int = array.length()

    public operator fun get(index: Int): T? {
        val array = this.array // volatile read
        return if (index < array.length()) array[index] else null
    }

    // Must not be called concurrently, e.g. always use synchronized(this) to call this function
    fun setSynchronized(index: Int, value: T?) {
        val curArray = this.array
        val curLen = curArray.length()
        if (index < curLen) {
            curArray[index] = value
        } else {
            val newArray = AtomicReferenceArray<T>((index + 1).coerceAtLeast(2 * curLen))
            for (i in 0 until curLen) newArray[i] = curArray[i]
            newArray[index] = value
            array = newArray // copy done
        }
    }
}
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/jvm/src/internal/ResizableAtomicArray.kt (Kotlin)

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

package kotlinx.coroutines.internal

import kotlinx.coroutines.*
import java.util.concurrent.atomic.*

/**
 * Atomic array with lock-free reads and synchronized modifications. It logically has an unbounded size,
 * is implicitly filled with nulls, and is resized on updates as needed to grow.
 */
internal class ResizableAtomicArray<T>(initialLength: Int) {
    @Volatile
    private var array = AtomicReferenceArray<T>(initialLength)

    // for debug output
    public fun currentLength(): Int = array.length()

    public operator fun get(index: Int): T? {
        assert { index >= 0 }
        val array = this.array // volatile read
        return if (index < array.length()) array[index] else null
    }

    @Synchronized
    operator fun set(index: Int, value: T?) {
        assert { index >= 0 }
        val curArray = this.array
        val curLen = curArray.length()
        if (index < curLen) {
            curArray[index] = value
        } else {
            val newArray = AtomicReferenceArray<T>((index + 1).coerceAtLeast(2 * curLen))
            for (i in 0 until curLen) newArray[i] = curArray[i]
            newArray[index] = value
            array = newArray // copy done
        }
    }
}


Review comment (suggestion):
This assert is redundant, `ARA` will perform it for you with better error message

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

package kotlinx.coroutines.internal

import java.util.concurrent.atomic.*

/**
 * Atomic array with lock-free reads and synchronized modifications. It logically has an unbounded size,
 * is implicitly filled with nulls, and is resized on updates as needed to grow.
 */
internal class ResizableAtomicArray<T>(initialLength: Int) {
    @Volatile
    private var array = AtomicReferenceArray<T>(initialLength)

    // for debug output
    public fun currentLength(): Int = array.length()

    public operator fun get(index: Int): T? {
        val array = this.array // volatile read
        return if (index < array.length()) array[index] else null
    }

    // Must not be called concurrently, e.g. always use synchronized(this) to call this function
    fun setSynchronized(index: Int, value: T?) {
        val curArray = this.array
        val curLen = curArray.length()
        if (index < curLen) {
            curArray[index] = value
        } else {
            val newArray = AtomicReferenceArray<T>((index + 1).coerceAtLeast(2 * curLen))
            for (i in 0 until curLen) newArray[i] = curArray[i]
            newArray[index] = value
            array = newArray // copy done
        }
    }
}
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/src/flow/operators/Delay.kt (Kotlin)

Code before:
 *     delay(1000)
 *     emit(4)
 * }.timeout(100.milliseconds) {
 *     emit(-1) // Item to emit on timeout
 * }.onEach {
 *     delay(300) // This will not cause a timeout
 * }
 * ```
 * <!--- KNIT example-timeout-duration-01.kt -->
 *
 * produces the following emissions
 *
 * ```text
 * 1, 2, 3, -1
 * ```
 * <!--- TEST -->
 *
 * Note that delaying on the downstream doesn't trigger the timeout.
 *
 * @param timeout Timeout period
 * @param action Action to invoke on timeout. Default is to throw [FlowTimeoutException]
 */
@FlowPreview
public fun <T> Flow<T>.timeout(
    timeout: Duration,
    @BuilderInference action: suspend FlowCollector<T>.() -> Unit = { throw FlowTimeoutException(timeout.toDelayMillis()) }
): Flow<T> = timeout(timeout.toDelayMillis(), action)

/**
 * Returns a flow that will timeout if the upstream takes too long to emit.
 *
 * Example:
 *
 * ```kotlin
 * flow {
 *     emit(1)
 *     delay(100)
 *     emit(2)
 *     delay(100)
 *     emit(3)
 *     delay(1000)
 *     emit(4)
 * }.timeout(100) {
 *     emit(-1) // Item to emit on timeout
 * }.onEach {
 *     delay(300) // This will not cause a timeout
 * }
 * ```
 * <!--- KNIT example-timeout-duration-02.kt -->
 *
 * produces the following emissions

Review comment (suggestion):
This is the non-orthogonality I was talking in https://github.com/Kotlin/kotlinx.coroutines/pull/2745#pullrequestreview-681969186

Luckily, it seems like this one can be safely removed without affecting the intention of the API.
For example, instead of 
```
.timeout(...) { emit(-1) }
```

timeout operator can throw timeout exception unconditionally, and then the same pattern can be expressed in already familiar manner:
```
.timeout(...).catch {  if (it is Timeout) emit(-1) else throw it }
```

Code after:
 *     emit(-1) // Item to emit on timeout
 * }.onEach {
 *     delay(300) // This will not cause a timeout
 * }
 * ```
 * <!--- KNIT example-timeout-duration-01.kt -->
 *
 * produces the following emissions
 *
 * ```text
 * 1, 2, 3, -1
 * ```
 * <!--- TEST -->
 *
 * Note that delaying on the downstream doesn't trigger the timeout.
 *
 * @param timeout period. If non-positive, the flow is timed out immediately
 */
@FlowPreview
public fun <T> Flow<T>.timeout(
    timeout: Duration
): Flow<T> = timeoutInternal(timeout)

private fun <T> Flow<T>.timeoutInternal(
    timeout: Duration
): Flow<T> = scopedFlow { downStream ->
    if (timeout <= Duration.ZERO) throw TimeoutCancellationException("Timed out immediately")
    val values = buffer(Channel.RENDEZVOUS).produceIn(this)
    whileSelect {
        values.onReceiveCatching { value ->
            value.onSuccess {
                downStream.emit(it)
            }.onClosed {
                return@onReceiveCatching false
            }
            return@onReceiveCatching true
        }
        onTimeout(timeout) {
            throw TimeoutCancellationException("Timed out waiting for $timeout")
        }
    }
}
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/src/flow/operators/Delay.kt (Kotlin)

Code before:
 * flow {
 *     emit(1)
 *     delay(100)
 *     emit(2)
 *     delay(100)
 *     emit(3)
 *     delay(1000)
 *     emit(4)
 * }.timeout(100.milliseconds).catch {
 *     emit(-1) // Item to emit on timeout
 * }.onEach {
 *     delay(300) // This will not cause a timeout
 * }
 * ```
 * <!--- KNIT example-timeout-duration-01.kt -->
 *
 * produces the following emissions
 *
 * ```text
 * 1, 2, 3, -1
 * ```
 * <!--- TEST -->
 *
 * Note that delaying on the downstream doesn't trigger the timeout.
 *
 * @param timeout Timeout period
 */
@FlowPreview
public fun <T> Flow<T>.timeout(
    timeout: Duration
): Flow<T> = timeoutInternal(timeout.toDelayMillis())

private fun <T> Flow<T>.timeoutInternal(
    timeoutMillis: Long
): Flow<T> = scopedFlow { downStream ->
    if (timeoutMillis <= 0L) throw FlowTimeoutException("Timed out immediately")

    // Produce the values using the default (rendezvous) channel
    // Similar to [debounceInternal]
    val values = produce<Any?> {
        var timeoutJob = launch { // Emits timeout unless cancelled
            delay(timeoutMillis)
            send(TIMEOUT)
        }
        try {
            collect {
                timeoutJob.cancel() // Upstream emitted, so cancel the job

                send(it ?: NULL)

                // We reset the job here. The reason being is that the `flow.emit()` suspends, which in turn suspends `send()`.

Review comment (suggestion):
```suggestion
 * @param timeout period. If non-positive, the flow is timed out immediately
```

Code after:
 *     delay(100)
 *     emit(2)
 *     delay(100)
 *     emit(3)
 *     delay(1000)
 *     emit(4)
 * }.timeout(100.milliseconds).catch {
 *     emit(-1) // Item to emit on timeout
 * }.onEach {
 *     delay(300) // This will not cause a timeout
 * }
 * ```
 * <!--- KNIT example-timeout-duration-01.kt -->
 *
 * produces the following emissions
 *
 * ```text
 * 1, 2, 3, -1
 * ```
 * <!--- TEST -->
 *
 * Note that delaying on the downstream doesn't trigger the timeout.
 *
 * @param timeout period. If non-positive, the flow is timed out immediately
 */
@FlowPreview
public fun <T> Flow<T>.timeout(
    timeout: Duration
): Flow<T> = timeoutInternal(timeout)

private fun <T> Flow<T>.timeoutInternal(
    timeout: Duration
): Flow<T> = scopedFlow { downStream ->
    if (timeout <= Duration.ZERO) throw TimeoutCancellationException("Timed out immediately")
    val values = buffer(Channel.RENDEZVOUS).produceIn(this)
    whileSelect {
        values.onReceiveCatching { value ->
            value.onSuccess {
                downStream.emit(it)
            }.onClosed {
                return@onReceiveCatching false
            }
            return@onReceiveCatching true
        }
        onTimeout(timeout) {
            throw TimeoutCancellationException("Timed out waiting for $timeout")
        }
    }
}
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/test/flow/operators/TimeoutTest.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.flow.operators

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.test.*
import kotlin.time.Duration.Companion.milliseconds

class TimeoutTest : TestBase() {
    @Test
    fun testBasic() = withVirtualTime {
        expect(1)
        val flow = flow {
            expect(3)
            emit("A")
            delay(100)
            emit("B")
            delay(100)
            emit("C")
            expect(4)
            delay(400)
            expectUnreached()
        }

        expect(2)
        val list = mutableListOf<String>()
        assertFailsWith<FlowTimeoutException>(flow.timeout(300.milliseconds).onEach { list.add(it) })
        assertEquals(listOf("A", "B", "C"), list)
        finish(5)
    }

    @Test
    fun testBasicCustomAction() = withVirtualTime {
        expect(1)

Review comment (bug):
Please add a test that uses `null` value and fails if you replace `send(it ?: NULL)` with `send(it)` in your implementation

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.flow.operators

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.test.*
import kotlin.time.Duration.Companion.milliseconds

class TimeoutTest : TestBase() {
    @Test
    fun testBasic() = withVirtualTime {
        expect(1)
        val flow = flow {
            expect(3)
            emit("A")
            delay(100)
            emit("B")
            delay(100)
            emit("C")
            expect(4)
            delay(400)
            expectUnreached()
        }

        expect(2)
        val list = mutableListOf<String>()
        assertFailsWith<TimeoutCancellationException>(flow.timeout(300.milliseconds).onEach { list.add(it) })
        assertEquals(listOf("A", "B", "C"), list)
        finish(5)
    }

    @Test
    fun testSingleNull() = withVirtualTime {
        val flow = flow<Int?> {
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/test/flow/operators/TimeoutTest.kt (Kotlin)

Code before:
    fun testDelayedFirst() = withVirtualTime {
        expect(1)
        val flow = flow {
            expect(3)
            delay(100)
            emit(1)
            expect(4)
        }.timeout(250.milliseconds)
        expect(2)
        assertEquals(1, flow.singleOrNull())
        finish(5)
    }

    // Test may rarely fail due to timeout being reached before test has completed
    @Test
    fun testEmpty() = runTest {
        val flow = emptyFlow<Any?>().timeout(1.milliseconds)
        assertNull(flow.singleOrNull())
    }

    // Test may rarely fail due to timeout being reached before test has completed
    @Test
    fun testScalar() = runTest {
        val flow = flowOf(1, 2, 3).timeout(1.milliseconds)
        assertEquals(listOf(1, 2, 3), flow.toList())
    }

    @Test
    fun testUpstreamError() = testUpstreamError(TestException())

    @Test
    fun testUpstreamErrorTimeoutException() = testUpstreamError(TimeoutCancellationException(0, Job()))

    private inline fun <reified T: Throwable> testUpstreamError(cause: T) = runTest {
        val flow = flow {
            emit(1)
            throw cause
        }.timeout(1.milliseconds)

        assertFailsWith<T>(flow)
    }

    @Test
    fun testDownstreamError() = runTest {
        val flow = flow {
            expect(1)
            emit(1)
            hang { expect(3) }
            expectUnreached()
        }.timeout(100.milliseconds).map {
            expect(2)

Review comment (suggestion):
```suggestion
    @Test
    fun testScalar() = withVirtualTime {
        val flow = flowOf(1, 2, 3).timeout(1.milliseconds)
        assertEquals(listOf(1, 2, 3), flow.toList())
        finish(1)
    }
```

Code after:
    fun testDelayedFirst() = withVirtualTime {
        expect(1)
        val flow = flow {
            expect(3)
            delay(100)
            emit(1)
            expect(4)
        }.timeout(250.milliseconds)
        expect(2)
        assertEquals(1, flow.singleOrNull())
        finish(5)
    }

    @Test
    fun testEmpty() = withVirtualTime {
        val flow = emptyFlow<Any?>().timeout(1.milliseconds)
        assertNull(flow.singleOrNull())
        finish(1)
    }

    @Test
    fun testScalar() = withVirtualTime {
        val flow = flowOf(1, 2, 3).timeout(1.milliseconds)
        assertEquals(listOf(1, 2, 3), flow.toList())
        finish(1)
    }

    @Test
    fun testUpstreamError() = testUpstreamError(TestException())

    @Test
    fun testUpstreamErrorTimeoutException() = testUpstreamError(TimeoutCancellationException(0, Job()))

    private inline fun <reified T: Throwable> testUpstreamError(cause: T) = runTest {
        try {
            // Workaround for JS legacy bug
            flow {
                emit(1)
                throw cause
            }.timeout(1.milliseconds).collect()
            expectUnreached()
        } catch (e: Throwable) {
            assertTrue { e is T }
            finish(1)
        }
    }

    @Test
    fun testDownstreamError() = runTest {
        val flow = flow {
            expect(1)
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/src/flow/operators/Delay.kt (Kotlin)

Code before:
 * <!--- KNIT example-timeout-duration-01.kt -->
 *
 * produces the following emissions
 *
 * ```text
 * 1, 2, 3, -1
 * ```
 * <!--- TEST -->
 *
 * Note that delaying on the downstream doesn't trigger the timeout.
 *
 * @param timeout period. If non-positive, the flow is timed out immediately
 */
@FlowPreview
public fun <T> Flow<T>.timeout(
    timeout: Duration
): Flow<T> = timeoutInternal(timeout)

private fun <T> Flow<T>.timeoutInternal(
    timeout: Duration
): Flow<T> = scopedFlow { downStream ->
    if (timeout <= Duration.ZERO) throw TimeoutCancellationException("Timed out immediately")

    val values = buffer(Channel.RENDEZVOUS).produceIn(this)

    whileSelect {
        values.onReceiveCatching { value ->
            value.onSuccess {
                downStream.emit(it)
            }.onClosed {
                return@onReceiveCatching false
            }
            return@onReceiveCatching true
        }
        onTimeout(timeout) {
            throw TimeoutCancellationException("Timed out waiting for $timeout")
        }
    }
}


Review comment (suggestion):
```suggestion
    if (timeout <= Duration.ZERO) throw TimeoutCancellationException("Timed out immediately")
    val values = buffer(Channel.RENDEZVOUS).produceIn(this)
    whileSelect {
```

Code after:
 * <!--- KNIT example-timeout-duration-01.kt -->
 *
 * produces the following emissions
 *
 * ```text
 * 1, 2, 3, -1
 * ```
 * <!--- TEST -->
 *
 * Note that delaying on the downstream doesn't trigger the timeout.
 *
 * @param timeout period. If non-positive, the flow is timed out immediately
 */
@FlowPreview
public fun <T> Flow<T>.timeout(
    timeout: Duration
): Flow<T> = timeoutInternal(timeout)

private fun <T> Flow<T>.timeoutInternal(
    timeout: Duration
): Flow<T> = scopedFlow { downStream ->
    if (timeout <= Duration.ZERO) throw TimeoutCancellationException("Timed out immediately")
    val values = buffer(Channel.RENDEZVOUS).produceIn(this)
    whileSelect {
        values.onReceiveCatching { value ->
            value.onSuccess {
                downStream.emit(it)
            }.onClosed {
                return@onReceiveCatching false
            }
            return@onReceiveCatching true
        }
        onTimeout(timeout) {
            throw TimeoutCancellationException("Timed out waiting for $timeout")
        }
    }
}
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/test/LimitedParallelismSharedTest.kt (Kotlin)

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

package kotlinx.coroutines

import kotlin.test.*

class LimitedParallelismSharedTest : TestBase() {
    @Test
    fun testTaskFairness() = runTest {
        val view = Dispatchers.Default.limitedParallelism(1)
        val view2 = Dispatchers.Default.limitedParallelism(1)
        val j1 = launch(view) {
            while (true) {
                yield()
            }
        }
        val j2 = launch(view2) { j1.cancel() }
        joinAll(j1, j2)
    }

    @Test
    fun testParallelismSpec() {
        assertFailsWith<IllegalArgumentException> { Dispatchers.Default.limitedParallelism(0) }
        assertFailsWith<IllegalArgumentException> { Dispatchers.Default.limitedParallelism(-1) }
        assertFailsWith<IllegalArgumentException> { Dispatchers.Default.limitedParallelism(Int.MIN_VALUE) }
        Dispatchers.Default.limitedParallelism(Int.MAX_VALUE)
    }
}


Review comment (refactor):
Suggestion: move this test to the JS source-set, and move the `testTaskFairness` test from the JVM source-set to `concurrent`. This way, there will still be two versions of the test, but each one will be the most specific one for each target.

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

package kotlinx.coroutines

import kotlin.test.*

class LimitedParallelismSharedTest : TestBase() {

    @Test
    fun testLimitedDefault() = runTest {
        // Test that evaluates the very basic completion of tasks in limited dispatcher
        // for all supported platforms.
        // For more specific and concurrent tests, see 'concurrent' package.
        val view = Dispatchers.Default.limitedParallelism(1)
        val view2 = Dispatchers.Default.limitedParallelism(1)
        val j1 = launch(view) {
            while (true) {
                yield()
            }
        }
        val j2 = launch(view2) { j1.cancel() }
        joinAll(j1, j2)
    }

    @Test
    fun testParallelismSpec() {
        assertFailsWith<IllegalArgumentException> { Dispatchers.Default.limitedParallelism(0) }
        assertFailsWith<IllegalArgumentException> { Dispatchers.Default.limitedParallelism(-1) }
        assertFailsWith<IllegalArgumentException> { Dispatchers.Default.limitedParallelism(Int.MIN_VALUE) }
        Dispatchers.Default.limitedParallelism(Int.MAX_VALUE)
    }
}