<commit_before>package org.jetbrains.ktor.servlet

import kotlinx.coroutines.experimental.channels.*
import org.jetbrains.ktor.cio.*
import java.io.*
import java.nio.*
import java.nio.channels.*
import java.util.concurrent.atomic.*
import javax.servlet.*
import kotlin.coroutines.experimental.*
import kotlin.coroutines.experimental.intrinsics.*

class ServletReadChannel(private val servletInputStream: ServletInputStream) : ReadChannel {
    @Volatile
    private var listenerInstalled = 0

    @Volatile
    private var readInProgress = 0

    private val callbackState = ConflatedChannel<Unit>()

    private val readListener = object : ReadListener {
        override fun onAllDataRead() {
            callbackState.close()
        }

        override fun onError(t: Throwable) {
            callbackState.close(t)
        }

        override fun onDataAvailable() {
            callbackState.offer(Unit)
        }
    }

    suspend override fun read(dst: ByteBuffer): Int {
        startReading()

        if (tryPollState()) return -1

        if (listenerInstalled == 0 && ListenerInstalled.compareAndSet(this, 0, 1)) {
            return installAndRead(dst)
        }

        return if (servletInputStream.isFinished) finish()
        else if (servletInputStream.isReady) {
            doRead(dst)
        } else {
            readSuspendUnconfined(dst)
        }
    }

    private suspend fun installAndRead(dst: ByteBuffer): Int {
        servletInputStream.setReadListener(readListener)
        return readSuspendUnconfined(dst)
    }

    private suspend fun readSuspendUnconfined(dst: ByteBuffer): Int {
        return suspendCoroutineOrReturn { cont ->
            val completion = object : Continuation<Int> by cont {
                override val context: CoroutineContext get() = EmptyCoroutineContext
            }

            bufferForLambda = dst
            readSuspendFunction.startCoroutineUninterceptedOrReturn(completion)
        }

        // ~= return run(Unconfined) { readSuspend(dst) }
    }

    private fun <T> suspend(block: suspend () -> T): suspend () -> T = block

    @Volatile
    private var bufferForLambda: ByteBuffer = Empty

    private val readSuspendFunction = suspend {
        readSuspend(bufferForLambda).also { bufferForLambda = Empty }
    }

    private tailrec suspend fun readSuspend(dst: ByteBuffer): Int {
        if (receiveState()) return -1

        return when {
            servletInputStream.isFinished -> finish()
            servletInputStream.isReady -> doRead(dst)
            else -> readSuspend(dst)
        }
    }

    private suspend fun receiveState(): Boolean {
        try {
            if (callbackState.receiveOrNull() == null) {
                endReading()
                return true
            }
        } catch (t: Throwable) {
            endReading()
            throw t
        }

        return false
    }

    private fun tryPollState(): Boolean {
        try {
            if (callbackState.poll() == null && callbackState.isClosedForReceive) {
                endReading()
                return true
            }
        } catch (t: Throwable) {
            endReading()
            throw t
        }

        return false
    }

    private fun startReading() {
        if (!ReadInProgress.compareAndSet(this, 0, 1)) throw IllegalStateException("read() is already in progress")
    }

    private fun endReading() {
        readInProgress = 0
    }

    private fun finish(): Int {
        callbackState.close()
        endReading()
<commit_msg>Servlet: don't rethrow EOFException but return -1 instead<commit_after>package org.jetbrains.ktor.servlet

import kotlinx.coroutines.experimental.channels.*
import org.jetbrains.ktor.cio.*
import java.io.*
import java.nio.*
import java.nio.channels.*
import java.util.concurrent.atomic.*
import javax.servlet.*
import kotlin.coroutines.experimental.*
import kotlin.coroutines.experimental.intrinsics.*

class ServletReadChannel(private val servletInputStream: ServletInputStream) : ReadChannel {
    @Volatile
    private var listenerInstalled = 0

    @Volatile
    private var readInProgress = 0

    private val callbackState = ConflatedChannel<Unit>()

    private val readListener = object : ReadListener {
        override fun onAllDataRead() {
            callbackState.close()
        }

        override fun onError(t: Throwable) {
            callbackState.close(t)
        }

        override fun onDataAvailable() {
            callbackState.offer(Unit)
        }
    }

    suspend override fun read(dst: ByteBuffer): Int {
        startReading()

        if (tryPollState()) return -1

        if (listenerInstalled == 0 && ListenerInstalled.compareAndSet(this, 0, 1)) {
            return installAndRead(dst)
        }

        return if (servletInputStream.isFinished) finish()
        else if (servletInputStream.isReady) {
            doRead(dst)
        } else {
            readSuspendUnconfined(dst)
        }
    }

    private suspend fun installAndRead(dst: ByteBuffer): Int {
        servletInputStream.setReadListener(readListener)
        return readSuspendUnconfined(dst)
    }

    private suspend fun readSuspendUnconfined(dst: ByteBuffer): Int {
        return suspendCoroutineOrReturn { cont ->
            val completion = object : Continuation<Int> by cont {
                override val context: CoroutineContext get() = EmptyCoroutineContext
            }

            bufferForLambda = dst
            readSuspendFunction.startCoroutineUninterceptedOrReturn(completion)
        }

        // ~= return run(Unconfined) { readSuspend(dst) }
    }

    private fun <T> suspend(block: suspend () -> T): suspend () -> T = block

    @Volatile
    private var bufferForLambda: ByteBuffer = Empty

    private val readSuspendFunction = suspend {
        readSuspend(bufferForLambda).also { bufferForLambda = Empty }
    }

    private tailrec suspend fun readSuspend(dst: ByteBuffer): Int {
        if (receiveState()) return -1

        return when {
            servletInputStream.isFinished -> finish()
            servletInputStream.isReady -> doRead(dst)
            else -> readSuspend(dst)
        }
    }

    private suspend fun receiveState(): Boolean {
        try {
            if (callbackState.receiveOrNull() == null) {
                endReading()
                return true
            }
        } catch (eof: EOFException) {
            return false
        } catch (t: Throwable) {
            endReading()
            throw t
        }

        return false
    }

    private fun tryPollState(): Boolean {
        try {
            if (callbackState.poll() == null && callbackState.isClosedForReceive) {
                endReading()
                return true
            }
        } catch (eof: EOFException) {
            return false
        } catch (t: Throwable) {
            endReading()
            throw t
        }

        return false
    }

    private fun startReading() {
        if (!ReadInProgress.compareAndSet(this, 0, 1)) throw IllegalStateException("read() is already in progress")
    }

    private fun endReading() {
        readInProgress = 0
    }

    private fun finish(): Int {
        callbackState.close()
        endReading()

{'fraction_non_alphanumeric': 0.08756205184776614, 'fraction_numerical': 0.0028957528957528956, 'mean_word_length': 2.13304535637149, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '261308', 'n_tokens_mistral': 1991, 'n_tokens_neox': 1899, 'n_words': 473}