<commit_before>package org.jetbrains.ktor.servlet

import org.jetbrains.ktor.nio.*
import org.jetbrains.ktor.util.*
import java.nio.*
import java.nio.channels.*
import java.util.concurrent.*
import java.util.concurrent.atomic.*
import javax.servlet.*

internal class ServletWriteChannel(val servletOutputStream: ServletOutputStream) : WriteChannel {
    private val listenerInstalled = AtomicBoolean()
    private val currentHandler = AtomicReference<AsyncHandler?>()
    private val pendingFlush = AtomicInteger(0)
    private val writeSemaphore = Semaphore(1)

    private val currentBuffer = AtomicReference<ByteBuffer?>()

    @Volatile
    private var lastWriteSize: Int = 0

    private val writeReadyListener = object : WriteListener {
        override fun onWritePossible() {
            writeAndNotify()
        }

        override fun onError(t: Throwable?) {
            fireHandler { handler, buffer ->
                handler.failed(t ?: RuntimeException("ServletWriteChannel.onError(null)"))
            }
        }
    }

    override fun write(src: ByteBuffer, handler: AsyncHandler) {
        if (!currentHandler.compareAndSet(null, handler)) {
            handler.failed(IllegalStateException("Write operation is already in progress"))
            return
        }
        currentBuffer.set(src)
        lastWriteSize = 0

        setupOrWrite()
    }

    override fun flush(handler: AsyncHandler) {
        if (!currentHandler.compareAndSet(null, handler)) {
            handler.failed(IllegalStateException("Write operation is already in progress"))
            return
        }
        currentBuffer.set(null)
        lastWriteSize = 0

        requestFlush()
    }

    override fun requestFlush() {
        pendingFlush.incrementAndGet()
        setupOrWrite()
    }

    private fun setupOrWrite() {
        if (listenerInstalled.compareAndSet(false, true)) {
            servletOutputStream.setWriteListener(writeReadyListener)
        } else {
            writeAndNotify()
        }
    }

    private fun writeAndNotify() {
        val done = try {
            doPendingWriteFlush()
        } catch (e: Exception) {
            fireHandler { handler, buffer ->
                handler.failed(e)
            }

            false
        }

        if (done) {
            fireHandler { handler, buffer ->
                if (buffer != null) {
                    handler.success(lastWriteSize)
                } else {
                    handler.successEnd()
                }
            }
        }
    }

    private fun doPendingWriteFlush(): Boolean {
        var writeRequested = false
        var written = false

        var flushRequested = false
        var flushed = false

        writeSemaphore.tryUse {
            val handler = currentHandler.get()
            val buffer = this.currentBuffer.get()

            if (handler != null && buffer != null) {
                writeRequested = true
                written = doWrite(buffer)
                if (!written) {
                    return false
                }
            }

            if (pendingFlush.getAndSet(0) > 0) {
                flushRequested = true
                flushed = doFlush()
                if (!flushed) {
                    pendingFlush.incrementAndGet()
                }
            }
        }

        return (!writeRequested || written) && (!flushRequested || flushed) && (writeRequested || flushRequested)
    }

    private fun doWrite(buffer: ByteBuffer): Boolean {
        if (servletOutputStream.isReady) {
            writeBuffer(buffer)
            return true
        }

        return false
    }

    private fun writeBuffer(buffer: ByteBuffer) {
        lastWriteSize = buffer.remaining()
        if (buffer.hasArray()) {
//            servletOutputStream.write(buffer.array(), buffer.arrayOffset() + buffer.position(), buffer.remaining())
            servletOutputStream.write(buffer.getAll()) // TODO: we should do like this because of possible concurrent array access
            buffer.position(buffer.limit())
        } else {
            val heapBuffer = ByteBuffer.allocate(buffer.remaining())
            heapBuffer.put(buffer)
            writeBuffer(heapBuffer)
        }
    }

    private fun doFlush(): Boolean {
        if (servletOutputStream.isReady) {
            servletOutputStream.flush()
            return true
        }
        return false
    }

    override fun close() {
        try {
            servletOutputStream.close()
        } finally {
            fireHandler { handler, byteBuffer ->
                handler.failed(ClosedChannelException())
            }
        }
    }<commit_msg>Servlets: temporary workaround for Jetty issue to eliminate corrupted long response<commit_after>package org.jetbrains.ktor.servlet

import org.jetbrains.ktor.nio.*
import org.jetbrains.ktor.util.*
import java.nio.*
import java.nio.channels.*
import java.util.concurrent.*
import java.util.concurrent.atomic.*
import javax.servlet.*

internal class ServletWriteChannel(val servletOutputStream: ServletOutputStream) : WriteChannel {
    private val listenerInstalled = AtomicBoolean()
    private val currentHandler = AtomicReference<AsyncHandler?>()
    private val pendingFlush = AtomicInteger(0)
    private val writeSemaphore = Semaphore(1)

    private val currentBuffer = AtomicReference<ByteBuffer?>()

    @Volatile
    private var lastWriteSize: Int = 0

    private val writeReadyListener = object : WriteListener {
        override fun onWritePossible() {
            writeAndNotify()
        }

        override fun onError(t: Throwable?) {
            fireHandler { handler, buffer ->
                handler.failed(t ?: RuntimeException("ServletWriteChannel.onError(null)"))
            }
        }
    }

    override fun write(src: ByteBuffer, handler: AsyncHandler) {
        if (!currentHandler.compareAndSet(null, handler)) {
            handler.failed(IllegalStateException("Write operation is already in progress"))
            return
        }
        currentBuffer.set(src)
        lastWriteSize = 0

        setupOrWrite()
    }

    override fun flush(handler: AsyncHandler) {
        if (!currentHandler.compareAndSet(null, handler)) {
            handler.failed(IllegalStateException("Write operation is already in progress"))
            return
        }
        currentBuffer.set(null)
        lastWriteSize = 0

        requestFlush()
    }

    override fun requestFlush() {
        pendingFlush.incrementAndGet()
        setupOrWrite()
    }

    private fun setupOrWrite() {
        if (listenerInstalled.compareAndSet(false, true)) {
            servletOutputStream.setWriteListener(writeReadyListener)
        } else {
            writeAndNotify()
        }
    }

    private fun writeAndNotify() {
        val done = try {
            doPendingWriteFlush()
        } catch (e: Exception) {
            fireHandler { handler, buffer ->
                handler.failed(e)
            }

            false
        }

        if (done) {
            fireHandler { handler, buffer ->
                if (buffer != null) {
                    handler.success(lastWriteSize)
                } else {
                    handler.successEnd()
                }
            }
        }
    }

    private fun doPendingWriteFlush(): Boolean {
        var writeRequested = false
        var written = false

        var flushRequested = false
        var flushed = false

        writeSemaphore.tryUse {
            val handler = currentHandler.get()
            val buffer = this.currentBuffer.get()

            if (handler != null && buffer != null) {
                writeRequested = true
                written = doWrite(buffer)
                if (!written) {
                    return false
                }
            }

            if (pendingFlush.getAndSet(0) > 0) {
                flushRequested = true
                flushed = doFlush()
                if (!flushed) {
                    pendingFlush.incrementAndGet()
                }
            }
        }

        return (!writeRequested || written) && (!flushRequested || flushed) && (writeRequested || flushRequested)
    }

    private fun doWrite(buffer: ByteBuffer): Boolean {
        if (servletOutputStream.isReady) {
            writeBuffer(buffer)
            return true
        }

        return false
    }

    private fun writeBuffer(buffer: ByteBuffer) {
        lastWriteSize = buffer.remaining()
        if (buffer.hasArray()) {
//            servletOutputStream.write(buffer.array(), buffer.arrayOffset() + buffer.position(), buffer.remaining())
            servletOutputStream.write(buffer.getAll()) // TODO: we should do like this because of possible concurrent array access
//            buffer.position(buffer.limit())
        } else {
            val heapBuffer = ByteBuffer.allocate(buffer.remaining())
            heapBuffer.put(buffer)
            writeBuffer(heapBuffer)
        }
    }

    private fun doFlush(): Boolean {
        if (servletOutputStream.isReady) {
            servletOutputStream.flush()
            return true
        }
        return false
    }

    override fun close() {
        try {
            servletOutputStream.close()
        } finally {
            fireHandler { handler, byteBuffer ->
                handler.failed(ClosedChannelException())
            }
        }
    }
{'fraction_non_alphanumeric': 0.08890545144804089, 'fraction_numerical': 0.0014906303236797275, 'mean_word_length': 1.826662654228107, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 17, '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': '5713869', 'n_tokens_mistral': 2391, 'n_tokens_neox': 2269, 'n_words': 534}