<commit_before>/*
 * JBoss, Home of Professional Open Source
 *
 * Copyright 2013 Red Hat, Inc. and/or its affiliates.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.xnio.ssl;

import static java.lang.Math.max;
import static java.lang.Thread.currentThread;
import static org.xnio.Bits.*;
import static org.xnio._private.Messages.msg;

import java.io.EOFException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;

import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLEngineResult;
import javax.net.ssl.SSLException;
import javax.net.ssl.SSLSession;
import org.xnio.Buffers;
import org.xnio.Pool;
import org.xnio.Pooled;
import org.xnio.XnioIoThread;
import org.xnio.XnioWorker;
import org.xnio.channels.StreamSinkChannel;
import org.xnio.channels.StreamSourceChannel;
import org.xnio.conduits.ConduitReadableByteChannel;
import org.xnio.conduits.ConduitWritableByteChannel;
import org.xnio.conduits.Conduits;
import org.xnio.conduits.ReadReadyHandler;
import org.xnio.conduits.StreamSinkConduit;
import org.xnio.conduits.StreamSourceConduit;
import org.xnio.conduits.WriteReadyHandler;

final class JsseStreamConduit implements StreamSourceConduit, StreamSinkConduit, ReadReadyHandler, WriteReadyHandler, Runnable {

    private static final boolean TRACE_SSL = true;

    //================================================================
    //
    // Immutable state
    //
    //================================================================

    private final JsseSslConnection connection;
    private final SSLEngine engine;
    private final StreamSourceConduit sourceConduit;
    private final StreamSinkConduit sinkConduit;
    /** The buffer into which incoming SSL data is written. */
    private final Pooled<ByteBuffer> receiveBuffer;
    /** The buffer from which outbound SSL data is sent. */
    private final Pooled<ByteBuffer> sendBuffer;
    /** The buffer into which inbound clear data is written. */
    private final Pooled<ByteBuffer> readBuffer;

    //================================================================
    //
    // Mutable state
    //
    //================================================================

    // always inline tasks, for now
    private int state = FLAG_INLINE_TASKS;

    // tasks counter - protected by {@code this}
    private int tasks;

    private ReadReadyHandler readReadyHandler;
    private WriteReadyHandler writeReadyHandler;

    //================================================================
    //
    // Constructors
    //
    //================================================================

    JsseStreamConduit(final JsseSslConnection connection, final SSLEngine engine, final StreamSourceConduit sourceConduit, final StreamSinkConduit sinkConduit, final Pool<ByteBuffer> socketBufferPool, final Pool<ByteBuffer> applicationBufferPool) {
        Pooled<ByteBuffer> receiveBuffer;
        Pooled<ByteBuffer> sendBuffer;
        Pooled<ByteBuffer> readBuffer;
        boolean ok = false;
        final SSLSession session = engine.getSession();
        final int packetBufferSize = session.getPacketBufferSize();
        receiveBuffer = socketBufferPool.allocate();
        try {
            receiveBuffer.getResource().flip();
            sendBuffer = socketBufferPool.allocate();
            try {
                if (receiveBuffer.getResource().capacity() < packetBufferSize || sendBuffer.getResource().capacity() < packetBufferSize) {
                    throw msg.socketBufferTooSmall();
                }
                final int applicationBufferSize = session.getApplicationBufferSize();
                readBuffer = applicationBufferPool.allocate();
                try {
                    if (readBuffer.getResource().capacity() < applicationBufferSize) {
                        throw msg.appBufferTooSmall();
                    }
                    ok = true;
                } finally {
                    if (! ok) readBuffer.free();
                }
            } finally {
                if (! ok) sendBuffer.free();
            }
        } finally {
            if (! ok) receiveBuffer.free();
        }
        this.receiveBuffer = receiveBuffer;
        this.sendBuffer = sendBuffer;
        this.readBuffer = readBuffer;
        receiveBuffer.getResource().clear().limit(0);
        if (sourceConduit.getReadThread() != sinkConduit.getWriteThread()) {
            throw new IllegalArgumentException("Source and sink thread mismatch");
        }
        this.connection = connection;
        this.engine = engine;
        this.sourceConduit = sourceConduit;
        this.sinkConduit = sinkConduit;
        sourceConduit.setReadReadyHandler(this);
        sinkConduit.setWriteReadyHandler(this);
    }

    //================================================================
    //
    // Flags
    //
    //================================================================

    // global flags

    //                                               global-_write----_read----;
    /** TLS is enabled */
    private static final int FLAG_TLS              = 0b00001_000000000_00000000;
    /** Run tasks immediately rather than in the worker */
    private static final int FLAG_INLINE_TASKS     = 0b00010_000000000_00000000;
    /** Set when task is queued; cleared when it runs */
    private static final int FLAG_TASK_QUEUED      = 0b00100_000000000_00000000;
    /** Set when the engine needs the result of a delegated task to continue */
    private static final int FLAG_NEED_ENGINE_TASK = 0b01000_000000000_00000000;
    /** we need a flush to proceed with handshake */
    private static final int FLAG_FLUSH_NEEDED     = 0b10000_000000000_00000000;

    // read state flags

    /** shutdownReads() was called upstream */
    private static final int READ_FLAG_SHUTDOWN    = 0b00000_000000000_00000001;
    /** -1 was read */
    private static final int READ_FLAG_EOF         = 0b00000_000000000_00000010;
    /** resumeReads() called */
    private static final int READ_FLAG_RESUMED     = 0b00000_000000000_00000100;
    /** upstream resumeReads() called */
    private static final int READ_FLAG_UP_RESUMED  = 0b00000_000000000_00001000;
    /** wakeupReads() called */
    private static final int READ_FLAG_WAKEUP      = 0b00000_000000000_00010000;
    /** user read handler should be called if resumed */
    private static final int READ_FLAG_READY       = 0b00000_000000000_00100000;
    /** read needs wrap to proceed */
    private static final int READ_FLAG_NEEDS_WRITE = 0b00000_000000000_01000000;

    // write state flags

    /** shutdownWrites() was called */
    private static final int WRITE_FLAG_SHUTDOWN   = 0b00000_000000001_00000000;
    /** send buffer cleared for final wrap after shutdownWrites() */
    private static final int WRITE_FLAG_SHUTDOWN2  = 0b00000_000000010_00000000;
    /** wrapping was completed after shutdownWrites() was called */
    private static final int WRITE_FLAG_SHUTDOWN3  = 0b00000_000000100_00000000;
    /** flush() returned true after shutdown */
    private static final int WRITE_FLAG_FINISHED   = 0b00000_000001000_00000000;
    /** resumeWrites() called */
    private static final int WRITE_FLAG_RESUMED    = 0b00000_000010000_00000000;
    /** upstream resumeWrites() called */
    private static final int WRITE_FLAG_UP_RESUMED = 0b00000_000100000_00000000;
    /** wakeupWrites() called */
    private static final int WRITE_FLAG_WAKEUP     = 0b00000_001000000_00000000;
    /** user write handler should be called if resumed */
    private static final int WRITE_FLAG_READY      = 0b00000_010000000_00000000;
    /** write needs unwrap to proceed */
    private static final int WRITE_FLAG_NEEDS_READ = 0b00000_100000000_00000000;

    public String getStatus() {
        final StringBuilder b = new StringBuilder();
        b.append("General flags:");
        final int state = this.state;
        if (allAreSet(state, FLAG_TLS)) b.append(" TLS");
        if (allAreSet(state, FLAG_INLINE_TASKS)) b.append(" INLINE_TASKS");
        if (allAreSet(state, FLAG_TASK_QUEUED)) b.append(" TASK_QUEUED");
        if (allAreSet(state, FLAG_NEED_ENGINE_TASK)) b.append(" NEED_ENGINE_TASK");
        if (allAreSet(state, FLAG_FLUSH_NEEDED)) b.append(" FLUSH_NEEDED");
        b.append("\nRead flags:");
        if (allAreSet(state, READ_FLAG_SHUTDOWN)) b.append(" SHUTDOWN");
        if (allAreSet(state, READ_FLAG_EOF)) b.append(" EOF");
        if (allAreSet(state, READ_FLAG_RESUMED)) b.append(" RESUMED");
        if (allAreSet(state, READ_FLAG_UP_RESUMED)) b.append(" UP_RESUMED");
        if (allAreSet(state, READ_FLAG_WAKEUP)) b.append(" WAKEUP");
        if (allAreSet(state, READ_FLAG_READY)) b.append(" READY");
        if (allAreSet(state, READ_FLAG_NEEDS_WRITE)) b.append(" NEEDS_WRITE");
        b.append("\nWrite flags:");
        if (allAreSet(state, WRITE_FLAG_SHUTDOWN)) b.append(" SHUTDOWN");
        if (allAreSet(state, WRITE_FLAG_SHUTDOWN2)) b.append(" SHUTDOWN2");
        if (allAreSet(state, WRITE_FLAG_SHUTDOWN3)) b.append(" SHUTDOWN3");
        if (allAreSet(state, WRITE_FLAG_FINISHED)) b.append(" FINISHED");
        if (allAreSet(state, WRITE_FLAG_RESUMED)) b.append(" RESUMED");
        if (allAreSet(state, WRITE_FLAG_UP_RESUMED)) b.append(" UP_RESUMED");
        if (allAreSet(state, WRITE_FLAG_WAKEUP)) b.append(" WAKEUP");
        if (allAreSet(state, WRITE_FLAG_READY)) b.append(" READY");
        if (allAreSet(state, WRITE_FLAG_NEEDS_READ)) b.append(" NEEDS_READ");
        b.append('\n');
        return b.toString();
    }

    public String toString() {
        return String.format("JSSE Stream Conduit for %s, status:%n%s", connection, getStatus());
    }

    //================================================================
    //
    // Global API
    //
    //================================================================

    public XnioWorker getWorker() {
        return connection.getIoThread().getWorker();
    }

    public XnioIoThread getReadThread() {
        return connection.getIoThread();
    }

    public XnioIoThread getWriteThread() {
        return connection.getIoThread();
    }

    public void forceTermination() {
        markTerminated();
        final ReadReadyHandler readReadyHandler = this.readReadyHandler;
        if (readReadyHandler != null) try {
            readReadyHandler.forceTermination();
        } catch (Throwable ignored) {
        }
        final WriteReadyHandler writeReadyHandler = this.writeReadyHandler;
        if (writeReadyHandler != null) try {
            writeReadyHandler.forceTermination();
        } catch (Throwable ignored) {
        }
    }

    public void terminated() {
        markTerminated();
        final ReadReadyHandler readReadyHandler = this.readReadyHandler;
        if (readReadyHandler != null) try {
            readReadyHandler.terminated();
        } catch (Throwable ignored) {
        }
        final WriteReadyHandler writeReadyHandler = this.writeReadyHandler;
        if (writeReadyHandler != null) try {
            writeReadyHandler.terminated();
        } catch (Throwable ignored) {
        }
    }

    // non-public

    void beginHandshake() throws IOException {
        final int state = this.state;
        if (anyAreSet(state, READ_FLAG_EOF | WRITE_FLAG_SHUTDOWN)) {
            throw new ClosedChannelException();
        }
        if (allAreClear(state, FLAG_TLS)) {
            this.state = state | FLAG_TLS;
        }
        engine.beginHandshake();
    }

    SSLSession getSslSession() {
        return allAreSet(state, FLAG_TLS) ? engine.getSession() : null;
    }

    SSLEngine getEngine() {
        return engine;
    }

    boolean isTls() {
        return allAreSet(state, FLAG_TLS);
    }

    boolean markTerminated() {
        readBuffer.free();
        receiveBuffer.free();
        sendBuffer.free();
        if (anyAreClear(state, READ_FLAG_SHUTDOWN | WRITE_FLAG_FINISHED)) {
            state |= READ_FLAG_SHUTDOWN | WRITE_FLAG_SHUTDOWN | WRITE_FLAG_SHUTDOWN2 | WRITE_FLAG_SHUTDOWN3 | WRITE_FLAG_FINISHED;
            return true;
        } else {
            return false;
        }
    }

    //================================================================
    //
    // Main run task
    //
    //================================================================

    public void run() {
        assert currentThread() == getWriteThread();
        int state = JsseStreamConduit.this.state & ~FLAG_TASK_QUEUED;
        boolean modify = false;
        try {
            // task(s)
            if (allAreSet(state, FLAG_NEED_ENGINE_TASK)) {
                throw new UnsupportedOperationException();
            }
            // write side
            if (anyAreSet(state, WRITE_FLAG_WAKEUP) || allAreSet(state, WRITE_FLAG_RESUMED | WRITE_FLAG_READY)) {
                final WriteReadyHandler writeReadyHandler = JsseStreamConduit.this.writeReadyHandler;
                if (allAreSet(state, WRITE_FLAG_WAKEUP)) {
                    state = state & ~WRITE_FLAG_WAKEUP | WRITE_FLAG_RESUMED;
                    modify = true;
                }
                if (writeReadyHandler != null) {
                    if (allAreSet(state, WRITE_FLAG_RESUMED)) {
                        try {
                            // save flags -------------------------------+
                            if (modify) {                             // |
                                modify = false;                       // |
                                JsseStreamConduit.this.state = state; // |
                            }                                         // |
                            writeReadyHandler.writeReady();           // |
                        } catch (Throwable ignored) {                 // |
                        } finally {                                   // |
                            // restore flags <---------------------------+
                            // it is OK if this is stale
                            state = JsseStreamConduit.this.state;
                        }
                        // Thread safety notice:
                        //---> We must not modify flags unless read and/or write is still resumed; otherwise, the user might
                        //     be doing something in another thread and we could end up overwriting each others' changes.
                        // level-triggering
                        if (allAreSet(state, WRITE_FLAG_RESUMED)) {
                            if (allAreSet(state, WRITE_FLAG_READY)) {
                                if (allAreClear(state, FLAG_TASK_QUEUED)) {
                                    state |= FLAG_TASK_QUEUED;
                                    modify = true;
                                    getWriteThread().execute(this);
                                }
                            } else if (allAreSet(state, WRITE_FLAG_NEEDS_READ) && allAreClear(state, READ_FLAG_UP_RESUMED)) {
                                state |= READ_FLAG_UP_RESUMED;
                                modify = true;
                                sourceConduit.resumeReads();
                            } else if (allAreClear(state, WRITE_FLAG_UP_RESUMED)) {
                                sinkConduit.resumeWrites();
                            }
                        }
                    } else {
                        if (allAreClear(state, READ_FLAG_NEEDS_WRITE | READ_FLAG_RESUMED) && allAreSet(state, WRITE_FLAG_UP_RESUMED)) {
                            state &= ~WRITE_FLAG_UP_RESUMED;
                            modify = true;
                            suspendWrites();
                        }
                    }
                } else {
                    // no handler, we should not be resumed
                    state &= ~WRITE_FLAG_RESUMED;<commit_msg>At JsseStreamConduit.run, do not queue task if write flag is ready<commit_after>/*
 * JBoss, Home of Professional Open Source
 *
 * Copyright 2013 Red Hat, Inc. and/or its affiliates.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.xnio.ssl;

import static java.lang.Math.max;
import static java.lang.Thread.currentThread;
import static org.xnio.Bits.*;
import static org.xnio._private.Messages.msg;

import java.io.EOFException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;

import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLEngineResult;
import javax.net.ssl.SSLException;
import javax.net.ssl.SSLSession;
import org.xnio.Buffers;
import org.xnio.Pool;
import org.xnio.Pooled;
import org.xnio.XnioIoThread;
import org.xnio.XnioWorker;
import org.xnio.channels.StreamSinkChannel;
import org.xnio.channels.StreamSourceChannel;
import org.xnio.conduits.ConduitReadableByteChannel;
import org.xnio.conduits.ConduitWritableByteChannel;
import org.xnio.conduits.Conduits;
import org.xnio.conduits.ReadReadyHandler;
import org.xnio.conduits.StreamSinkConduit;
import org.xnio.conduits.StreamSourceConduit;
import org.xnio.conduits.WriteReadyHandler;

final class JsseStreamConduit implements StreamSourceConduit, StreamSinkConduit, ReadReadyHandler, WriteReadyHandler, Runnable {

    private static final boolean TRACE_SSL = true;

    //================================================================
    //
    // Immutable state
    //
    //================================================================

    private final JsseSslConnection connection;
    private final SSLEngine engine;
    private final StreamSourceConduit sourceConduit;
    private final StreamSinkConduit sinkConduit;
    /** The buffer into which incoming SSL data is written. */
    private final Pooled<ByteBuffer> receiveBuffer;
    /** The buffer from which outbound SSL data is sent. */
    private final Pooled<ByteBuffer> sendBuffer;
    /** The buffer into which inbound clear data is written. */
    private final Pooled<ByteBuffer> readBuffer;

    //================================================================
    //
    // Mutable state
    //
    //================================================================

    // always inline tasks, for now
    private int state = FLAG_INLINE_TASKS;

    // tasks counter - protected by {@code this}
    private int tasks;

    private ReadReadyHandler readReadyHandler;
    private WriteReadyHandler writeReadyHandler;

    //================================================================
    //
    // Constructors
    //
    //================================================================

    JsseStreamConduit(final JsseSslConnection connection, final SSLEngine engine, final StreamSourceConduit sourceConduit, final StreamSinkConduit sinkConduit, final Pool<ByteBuffer> socketBufferPool, final Pool<ByteBuffer> applicationBufferPool) {
        Pooled<ByteBuffer> receiveBuffer;
        Pooled<ByteBuffer> sendBuffer;
        Pooled<ByteBuffer> readBuffer;
        boolean ok = false;
        final SSLSession session = engine.getSession();
        final int packetBufferSize = session.getPacketBufferSize();
        receiveBuffer = socketBufferPool.allocate();
        try {
            receiveBuffer.getResource().flip();
            sendBuffer = socketBufferPool.allocate();
            try {
                if (receiveBuffer.getResource().capacity() < packetBufferSize || sendBuffer.getResource().capacity() < packetBufferSize) {
                    throw msg.socketBufferTooSmall();
                }
                final int applicationBufferSize = session.getApplicationBufferSize();
                readBuffer = applicationBufferPool.allocate();
                try {
                    if (readBuffer.getResource().capacity() < applicationBufferSize) {
                        throw msg.appBufferTooSmall();
                    }
                    ok = true;
                } finally {
                    if (! ok) readBuffer.free();
                }
            } finally {
                if (! ok) sendBuffer.free();
            }
        } finally {
            if (! ok) receiveBuffer.free();
        }
        this.receiveBuffer = receiveBuffer;
        this.sendBuffer = sendBuffer;
        this.readBuffer = readBuffer;
        receiveBuffer.getResource().clear().limit(0);
        if (sourceConduit.getReadThread() != sinkConduit.getWriteThread()) {
            throw new IllegalArgumentException("Source and sink thread mismatch");
        }
        this.connection = connection;
        this.engine = engine;
        this.sourceConduit = sourceConduit;
        this.sinkConduit = sinkConduit;
        sourceConduit.setReadReadyHandler(this);
        sinkConduit.setWriteReadyHandler(this);
    }

    //================================================================
    //
    // Flags
    //
    //================================================================

    // global flags

    //                                               global-_write----_read----;
    /** TLS is enabled */
    private static final int FLAG_TLS              = 0b00001_000000000_00000000;
    /** Run tasks immediately rather than in the worker */
    private static final int FLAG_INLINE_TASKS     = 0b00010_000000000_00000000;
    /** Set when task is queued; cleared when it runs */
    private static final int FLAG_TASK_QUEUED      = 0b00100_000000000_00000000;
    /** Set when the engine needs the result of a delegated task to continue */
    private static final int FLAG_NEED_ENGINE_TASK = 0b01000_000000000_00000000;
    /** we need a flush to proceed with handshake */
    private static final int FLAG_FLUSH_NEEDED     = 0b10000_000000000_00000000;

    // read state flags

    /** shutdownReads() was called upstream */
    private static final int READ_FLAG_SHUTDOWN    = 0b00000_000000000_00000001;
    /** -1 was read */
    private static final int READ_FLAG_EOF         = 0b00000_000000000_00000010;
    /** resumeReads() called */
    private static final int READ_FLAG_RESUMED     = 0b00000_000000000_00000100;
    /** upstream resumeReads() called */
    private static final int READ_FLAG_UP_RESUMED  = 0b00000_000000000_00001000;
    /** wakeupReads() called */
    private static final int READ_FLAG_WAKEUP      = 0b00000_000000000_00010000;
    /** user read handler should be called if resumed */
    private static final int READ_FLAG_READY       = 0b00000_000000000_00100000;
    /** read needs wrap to proceed */
    private static final int READ_FLAG_NEEDS_WRITE = 0b00000_000000000_01000000;

    // write state flags

    /** shutdownWrites() was called */
    private static final int WRITE_FLAG_SHUTDOWN   = 0b00000_000000001_00000000;
    /** send buffer cleared for final wrap after shutdownWrites() */
    private static final int WRITE_FLAG_SHUTDOWN2  = 0b00000_000000010_00000000;
    /** wrapping was completed after shutdownWrites() was called */
    private static final int WRITE_FLAG_SHUTDOWN3  = 0b00000_000000100_00000000;
    /** flush() returned true after shutdown */
    private static final int WRITE_FLAG_FINISHED   = 0b00000_000001000_00000000;
    /** resumeWrites() called */
    private static final int WRITE_FLAG_RESUMED    = 0b00000_000010000_00000000;
    /** upstream resumeWrites() called */
    private static final int WRITE_FLAG_UP_RESUMED = 0b00000_000100000_00000000;
    /** wakeupWrites() called */
    private static final int WRITE_FLAG_WAKEUP     = 0b00000_001000000_00000000;
    /** user write handler should be called if resumed */
    private static final int WRITE_FLAG_READY      = 0b00000_010000000_00000000;
    /** write needs unwrap to proceed */
    private static final int WRITE_FLAG_NEEDS_READ = 0b00000_100000000_00000000;

    public String getStatus() {
        final StringBuilder b = new StringBuilder();
        b.append("General flags:");
        final int state = this.state;
        if (allAreSet(state, FLAG_TLS)) b.append(" TLS");
        if (allAreSet(state, FLAG_INLINE_TASKS)) b.append(" INLINE_TASKS");
        if (allAreSet(state, FLAG_TASK_QUEUED)) b.append(" TASK_QUEUED");
        if (allAreSet(state, FLAG_NEED_ENGINE_TASK)) b.append(" NEED_ENGINE_TASK");
        if (allAreSet(state, FLAG_FLUSH_NEEDED)) b.append(" FLUSH_NEEDED");
        b.append("\nRead flags:");
        if (allAreSet(state, READ_FLAG_SHUTDOWN)) b.append(" SHUTDOWN");
        if (allAreSet(state, READ_FLAG_EOF)) b.append(" EOF");
        if (allAreSet(state, READ_FLAG_RESUMED)) b.append(" RESUMED");
        if (allAreSet(state, READ_FLAG_UP_RESUMED)) b.append(" UP_RESUMED");
        if (allAreSet(state, READ_FLAG_WAKEUP)) b.append(" WAKEUP");
        if (allAreSet(state, READ_FLAG_READY)) b.append(" READY");
        if (allAreSet(state, READ_FLAG_NEEDS_WRITE)) b.append(" NEEDS_WRITE");
        b.append("\nWrite flags:");
        if (allAreSet(state, WRITE_FLAG_SHUTDOWN)) b.append(" SHUTDOWN");
        if (allAreSet(state, WRITE_FLAG_SHUTDOWN2)) b.append(" SHUTDOWN2");
        if (allAreSet(state, WRITE_FLAG_SHUTDOWN3)) b.append(" SHUTDOWN3");
        if (allAreSet(state, WRITE_FLAG_FINISHED)) b.append(" FINISHED");
        if (allAreSet(state, WRITE_FLAG_RESUMED)) b.append(" RESUMED");
        if (allAreSet(state, WRITE_FLAG_UP_RESUMED)) b.append(" UP_RESUMED");
        if (allAreSet(state, WRITE_FLAG_WAKEUP)) b.append(" WAKEUP");
        if (allAreSet(state, WRITE_FLAG_READY)) b.append(" READY");
        if (allAreSet(state, WRITE_FLAG_NEEDS_READ)) b.append(" NEEDS_READ");
        b.append('\n');
        return b.toString();
    }

    public String toString() {
        return String.format("JSSE Stream Conduit for %s, status:%n%s", connection, getStatus());
    }

    //================================================================
    //
    // Global API
    //
    //================================================================

    public XnioWorker getWorker() {
        return connection.getIoThread().getWorker();
    }

    public XnioIoThread getReadThread() {
        return connection.getIoThread();
    }

    public XnioIoThread getWriteThread() {
        return connection.getIoThread();
    }

    public void forceTermination() {
        markTerminated();
        final ReadReadyHandler readReadyHandler = this.readReadyHandler;
        if (readReadyHandler != null) try {
            readReadyHandler.forceTermination();
        } catch (Throwable ignored) {
        }
        final WriteReadyHandler writeReadyHandler = this.writeReadyHandler;
        if (writeReadyHandler != null) try {
            writeReadyHandler.forceTermination();
        } catch (Throwable ignored) {
        }
    }

    public void terminated() {
        markTerminated();
        final ReadReadyHandler readReadyHandler = this.readReadyHandler;
        if (readReadyHandler != null) try {
            readReadyHandler.terminated();
        } catch (Throwable ignored) {
        }
        final WriteReadyHandler writeReadyHandler = this.writeReadyHandler;
        if (writeReadyHandler != null) try {
            writeReadyHandler.terminated();
        } catch (Throwable ignored) {
        }
    }

    // non-public

    void beginHandshake() throws IOException {
        final int state = this.state;
        if (anyAreSet(state, READ_FLAG_EOF | WRITE_FLAG_SHUTDOWN)) {
            throw new ClosedChannelException();
        }
        if (allAreClear(state, FLAG_TLS)) {
            this.state = state | FLAG_TLS;
        }
        engine.beginHandshake();
    }

    SSLSession getSslSession() {
        return allAreSet(state, FLAG_TLS) ? engine.getSession() : null;
    }

    SSLEngine getEngine() {
        return engine;
    }

    boolean isTls() {
        return allAreSet(state, FLAG_TLS);
    }

    boolean markTerminated() {
        readBuffer.free();
        receiveBuffer.free();
        sendBuffer.free();
        if (anyAreClear(state, READ_FLAG_SHUTDOWN | WRITE_FLAG_FINISHED)) {
            state |= READ_FLAG_SHUTDOWN | WRITE_FLAG_SHUTDOWN | WRITE_FLAG_SHUTDOWN2 | WRITE_FLAG_SHUTDOWN3 | WRITE_FLAG_FINISHED;
            return true;
        } else {
            return false;
        }
    }

    //================================================================
    //
    // Main run task
    //
    //================================================================

    public void run() {
        assert currentThread() == getWriteThread();
        int state = JsseStreamConduit.this.state & ~FLAG_TASK_QUEUED;
        boolean modify = false;
        try {
            // task(s)
            if (allAreSet(state, FLAG_NEED_ENGINE_TASK)) {
                throw new UnsupportedOperationException();
            }
            // write side
            if (anyAreSet(state, WRITE_FLAG_WAKEUP) || allAreSet(state, WRITE_FLAG_RESUMED | WRITE_FLAG_READY)) {
                final WriteReadyHandler writeReadyHandler = JsseStreamConduit.this.writeReadyHandler;
                if (allAreSet(state, WRITE_FLAG_WAKEUP)) {
                    state = state & ~WRITE_FLAG_WAKEUP | WRITE_FLAG_RESUMED;
                    modify = true;
                }
                if (writeReadyHandler != null) {
                    if (allAreSet(state, WRITE_FLAG_RESUMED)) {
                        try {
                            // save flags -------------------------------+
                            if (modify) {                             // |
                                modify = false;                       // |
                                JsseStreamConduit.this.state = state; // |
                            }                                         // |
                            writeReadyHandler.writeReady();           // |
                        } catch (Throwable ignored) {                 // |
                        } finally {                                   // |
                            // restore flags <---------------------------+
                            // it is OK if this is stale
                            state = JsseStreamConduit.this.state;
                        }
                        // Thread safety notice:
                        //---> We must not modify flags unless read and/or write is still resumed; otherwise, the user might
                        //     be doing something in another thread and we could end up overwriting each others' changes.
                        // level-triggering
                        if (allAreSet(state, WRITE_FLAG_RESUMED)) {
                            if (!allAreSet(state, WRITE_FLAG_READY) && allAreSet(state, WRITE_FLAG_NEEDS_READ) && allAreClear(state, READ_FLAG_UP_RESUMED)) {
                                state |= READ_FLAG_UP_RESUMED;
                                modify = true;
                                sourceConduit.resumeReads();
                            } else if (allAreClear(state, WRITE_FLAG_UP_RESUMED)) {
                                sinkConduit.resumeWrites();
                            }
                        }
                    } else {
                        if (allAreClear(state, READ_FLAG_NEEDS_WRITE | READ_FLAG_RESUMED) && allAreSet(state, WRITE_FLAG_UP_RESUMED)) {
                            state &= ~WRITE_FLAG_UP_RESUMED;
                            modify = true;
                            suspendWrites();
                        }
                    }
                } else {
                    // no handler, we should not be resumed
                    state &= ~WRITE_FLAG_RESUMED;
{'fraction_non_alphanumeric': 0.13548347873666292, 'fraction_numerical': 0.03045870444113445, 'mean_word_length': 2.2773460848774656, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 62, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6982463', 'n_tokens_mistral': 9703, 'n_tokens_neox': 8593, 'n_words': 2083}