<commit_before>package io.ktor.network.sockets

import io.ktor.network.selector.*
import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.io.*
import kotlinx.coroutines.experimental.io.ByteChannel
import kotlinx.io.pool.*
import java.nio.channels.*
import java.util.concurrent.atomic.*

internal abstract class NIOSocketImpl<out S>(override val channel: S, val selector: SelectorManager, val pool: ObjectPool<ByteBuffer>?) : ReadWriteSocket, SelectableBase(channel)
        where S : java.nio.channels.ByteChannel, S : java.nio.channels.SelectableChannel {

    protected val closeFlag = AtomicBoolean()
    private val readerJob = AtomicReference<ReaderJob?>()
    private val writerJob = AtomicReference<WriterJob?>()

    override val closed = CompletableDeferred<Unit>()

    // NOTE: it is important here to use different versions of attachForReadingImpl
    // because it is not always valid to use channel's internal buffer for NIO read/write:
    //  at least UDP datagram reading MUST use bigger byte buffer otherwise datagram could be truncated
    //  that will cause broken data
    // however it is not the case for attachForWriting this is why we use direct writing in any case

    final override fun attachForReading(channel: ByteChannel): WriterJob {
        return attachFor("reading", channel, writerJob) {
            if (pool != null) {
                attachForReadingImpl(channel, this.channel, this, selector, pool)
            } else {
                attachForReadingDirectImpl(channel, this.channel, this, selector)
            }
        }
    }

    final override fun attachForWriting(channel: ByteChannel): ReaderJob {
        return attachFor("writing", channel, readerJob) {
            attachForWritingDirectImpl(channel, this.channel, this, selector)
        }
    }

    override fun dispose() {
        close()
    }

    override fun close() {
        if (closeFlag.compareAndSet(false, true)) {
            readerJob.get()?.channel?.close()
            writerJob.get()?.cancel()
            checkChannels()
            super.close()
        }
    }

    private fun <J : Job> attachFor(name: String, channel: ByteChannel, ref: AtomicReference<J?>, producer: () -> J): J {
        if (closeFlag.get()) {
            val e = ClosedChannelException()
            channel.close(e)
            throw e
        }

        val j = producer()

        if (!ref.compareAndSet(null, j)) {
            val e = IllegalStateException("$name channel has been already set")
            j.cancel(e)
            throw e
        }
        if (closeFlag.get()) {
            val e = ClosedChannelException()
            j.cancel(e)
            channel.close(e)
            throw e
        }

        j.invokeOnCompletion {
            checkChannels()
        }

        return j
    }

    private fun actualClose(): Throwable? {
        return try {
            channel.close()
            null
        } catch (t: Throwable) {
            t
        } finally {
            selector.notifyClosed(this)
        }
    }

    private fun checkChannels() {
        if (closeFlag.get() && readerJob.completedOrNotStarted && writerJob.completedOrNotStarted) {
            val e1 = readerJob.exception
            val e2 = writerJob.exception
            val e3 = actualClose()

            val combined = combine(combine(e1, e2), e3)

            if (combined == null) closed.complete(Unit) else closed.completeExceptionally(combined)
        }
    }

    private fun combine(e1: Throwable?, e2: Throwable?): Throwable? = when {
        e1 == null -> e2
        e2 == null -> e1
        e1 === e2 -> e1
        else -> {
            e1.addSuppressed(e2)
            e1
        }
    }

    private val AtomicReference<out Job?>.completedOrNotStarted: Boolean
        get() = get().let { it == null || it.isCompleted }

    private val AtomicReference<out Job?>.exception: Throwable?
        get() = get()?.takeUnless { it.isActive || it.isCancelled }?.getCompletionException()?.takeUnless { it is CancellationException }
}<commit_msg>CIO: avoid old getCompletionException() usage<commit_after>package io.ktor.network.sockets

import io.ktor.network.selector.*
import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.io.*
import kotlinx.coroutines.experimental.io.ByteChannel
import kotlinx.io.pool.*
import java.nio.channels.*
import java.util.concurrent.atomic.*

internal abstract class NIOSocketImpl<out S>(override val channel: S, val selector: SelectorManager, val pool: ObjectPool<ByteBuffer>?) : ReadWriteSocket, SelectableBase(channel)
        where S : java.nio.channels.ByteChannel, S : java.nio.channels.SelectableChannel {

    protected val closeFlag = AtomicBoolean()
    private val readerJob = AtomicReference<ReaderJob?>()
    private val writerJob = AtomicReference<WriterJob?>()

    override val closed = CompletableDeferred<Unit>()

    // NOTE: it is important here to use different versions of attachForReadingImpl
    // because it is not always valid to use channel's internal buffer for NIO read/write:
    //  at least UDP datagram reading MUST use bigger byte buffer otherwise datagram could be truncated
    //  that will cause broken data
    // however it is not the case for attachForWriting this is why we use direct writing in any case

    final override fun attachForReading(channel: ByteChannel): WriterJob {
        return attachFor("reading", channel, writerJob) {
            if (pool != null) {
                attachForReadingImpl(channel, this.channel, this, selector, pool)
            } else {
                attachForReadingDirectImpl(channel, this.channel, this, selector)
            }
        }
    }

    final override fun attachForWriting(channel: ByteChannel): ReaderJob {
        return attachFor("writing", channel, readerJob) {
            attachForWritingDirectImpl(channel, this.channel, this, selector)
        }
    }

    override fun dispose() {
        close()
    }

    override fun close() {
        if (closeFlag.compareAndSet(false, true)) {
            readerJob.get()?.channel?.close()
            writerJob.get()?.cancel()
            checkChannels()
            super.close()
        }
    }

    private fun <J : Job> attachFor(name: String, channel: ByteChannel, ref: AtomicReference<J?>, producer: () -> J): J {
        if (closeFlag.get()) {
            val e = ClosedChannelException()
            channel.close(e)
            throw e
        }

        val j = producer()

        if (!ref.compareAndSet(null, j)) {
            val e = IllegalStateException("$name channel has been already set")
            j.cancel(e)
            throw e
        }
        if (closeFlag.get()) {
            val e = ClosedChannelException()
            j.cancel(e)
            channel.close(e)
            throw e
        }

        j.invokeOnCompletion {
            checkChannels()
        }

        return j
    }

    private fun actualClose(): Throwable? {
        return try {
            channel.close()
            null
        } catch (t: Throwable) {
            t
        } finally {
            selector.notifyClosed(this)
        }
    }

    private fun checkChannels() {
        if (closeFlag.get() && readerJob.completedOrNotStarted && writerJob.completedOrNotStarted) {
            val e1 = readerJob.exception
            val e2 = writerJob.exception
            val e3 = actualClose()

            val combined = combine(combine(e1, e2), e3)

            if (combined == null) closed.complete(Unit) else closed.completeExceptionally(combined)
        }
    }

    private fun combine(e1: Throwable?, e2: Throwable?): Throwable? = when {
        e1 == null -> e2
        e2 == null -> e1
        e1 === e2 -> e1
        else -> {
            e1.addSuppressed(e2)
            e1
        }
    }

    private val AtomicReference<out Job?>.completedOrNotStarted: Boolean
        get() = get().let { it == null || it.isCompleted }

    private val AtomicReference<out Job?>.exception: Throwable?
        get() = get()?.takeUnless { it.isActive || it.isCancelled }
                ?.getCancellationException()
                ?.let { (it as? JobCancellationException)?.cause }
}
{'fraction_non_alphanumeric': 0.10026770503772207, 'fraction_numerical': 0.004380627889997567, 'mean_word_length': 2.235826771653543, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 31, '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': '3829419', 'n_tokens_mistral': 2296, 'n_tokens_neox': 2193, 'n_words': 609}