/*
* Copyright (c) 2017, 2018, Oracle and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License version 2 only, as
* published by the Free Software Foundation. Oracle designates this
* particular file as subject to the "Classpath" exception as provided
* by Oracle in the LICENSE file that accompanied this code.
*
* This code is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
* version 2 for more details (a copy is included in the LICENSE file that
* accompanied this code).
*
* You should have received a copy of the GNU General Public License version
* 2 along with this work; if not, write to the Free Software Foundation,
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
* or visit www.oracle.com if you need additional information or have any
* questions.
*/
package jdk.internal.net.http;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.Flow;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.function.Consumer;
import java.util.function.Supplier;
import jdk.internal.net.http.common.BufferSupplier;
import jdk.internal.net.http.common.Demand;
import jdk.internal.net.http.common.FlowTube;
import jdk.internal.net.http.common.Log;
import jdk.internal.net.http.common.Logger;
import jdk.internal.net.http.common.SequentialScheduler;
import jdk.internal.net.http.common.SequentialScheduler.DeferredCompleter;
import jdk.internal.net.http.common.SequentialScheduler.RestartableTask;
import jdk.internal.net.http.common.Utils;
/**
* A SocketTube is a terminal tube plugged directly into the socket.
* The read subscriber should call {@code subscribe} on the SocketTube before
* the SocketTube is subscribed to the write publisher.
*/
final class SocketTube implements FlowTube {
final Logger debug = Utils.getDebugLogger(this::dbgString, Utils.DEBUG);
static final AtomicLong IDS = new AtomicLong();
private final HttpClientImpl client;
private final SocketChannel channel;
private final SliceBufferSource sliceBuffersSource;
private final Object lock = new Object();
private final AtomicReference<Throwable> errorRef = new AtomicReference<>();
private final InternalReadPublisher readPublisher;
private final InternalWriteSubscriber writeSubscriber;
private final long id = IDS.incrementAndGet();
public SocketTube(HttpClientImpl client, SocketChannel channel,
Supplier<ByteBuffer> buffersFactory) {
this.client = client;
this.channel = channel;
this.sliceBuffersSource = new SliceBufferSource(buffersFactory);
this.readPublisher = new InternalReadPublisher();
this.writeSubscriber = new InternalWriteSubscriber();
}
/**
* Returns {@code true} if this flow is finished.
* This happens when this flow internal read subscription is completed,
* either normally (EOF reading) or exceptionally (EOF writing, or
* underlying socket closed, or some exception occurred while reading or
* writing to the socket).
*
* @return {@code true} if this flow is finished.
*/
public boolean isFinished() {
InternalReadPublisher.InternalReadSubscription subscription =
readPublisher.subscriptionImpl;
return subscription != null && subscription.completed
|| subscription == null && errorRef.get() != null;
}
// ===================================================================== //
// Flow.Publisher //
// ======================================================================//
/**
* {@inheritDoc }
* @apiNote This method should be called first. In particular, the caller
* must ensure that this method must be called by the read
* subscriber before the write publisher can call {@code onSubscribe}.
* Failure to adhere to this contract may result in assertion errors.
*/
@Override
public void subscribe(Flow.Subscriber<? super List<ByteBuffer>> s) {
Objects.requireNonNull(s);
assert s instanceof TubeSubscriber : "Expected TubeSubscriber, got:" + s;
readPublisher.subscribe(s);
}
// ===================================================================== //
// Flow.Subscriber //
// ======================================================================//
/**
* {@inheritDoc }
* @apiNote The caller must ensure that {@code subscribe} is called by
* the read subscriber before {@code onSubscribe} is called by
* the write publisher.
* Failure to adhere to this contract may result in assertion errors.
*/
@Override
public void onSubscribe(Flow.Subscription subscription) {
writeSubscriber.onSubscribe(subscription);
}
@Override
public void onNext(List<ByteBuffer> item) {
writeSubscriber.onNext(item);
}
@Override
public void onError(Throwable throwable) {
writeSubscriber.onError(throwable);
}
@Override
public void onComplete() {
writeSubscriber.onComplete();
}
// ===================================================================== //
// Events //
// ======================================================================//
void signalClosed() {
// Ensures that the subscriber will be terminated and that future
// subscribers will be notified when the connection is closed.
if (Log.channel()) {
Log.logChannel("Connection close signalled: connection closed locally ({0})",
channelDescr());
}
readPublisher.subscriptionImpl.signalError(
new IOException("connection closed locally"));
}
/**
* A restartable task used to process tasks in sequence.
*/
private static class SocketFlowTask implements RestartableTask {
final Runnable task;
private final Object monitor = new Object();
SocketFlowTask(Runnable task) {
this.task = task;
}
@Override
public final void run(DeferredCompleter taskCompleter) {
try {
// non contentious synchronized for visibility.
synchronized(monitor) {
task.run();
}
} finally {
taskCompleter.complete();
}
}
}
// This is best effort - there's no guarantee that the printed set of values
// is consistent. It should only be considered as weakly accurate - in
// particular in what concerns the events states, especially when displaying
// a read event state from a write event callback and conversely.
void debugState(String when) {
if (debug.on()) {
StringBuilder state = new StringBuilder();
InternalReadPublisher.InternalReadSubscription sub =
readPublisher.subscriptionImpl;
InternalReadPublisher.ReadEvent readEvent =
sub == null ? null : sub.readEvent;
Demand rdemand = sub == null ? null : sub.demand;
InternalWriteSubscriber.WriteEvent writeEvent =
writeSubscriber.writeEvent;
Demand wdemand = writeSubscriber.writeDemand;
int rops = readEvent == null ? 0 : readEvent.interestOps();
long rd = rdemand == null ? 0 : rdemand.get();
int wops = writeEvent == null ? 0 : writeEvent.interestOps();
long wd = wdemand == null ? 0 : wdemand.get();
state.append(when).append(" Reading: [ops=")
.append(rops).append(", demand=").append(rd)
.append(", stopped=")
.append((sub == null ? false : sub.readScheduler.isStopped()))
.append("], Writing: [ops=").append(wops)
.append(", demand=").append(wd)
.append("]");
debug.log(state.toString());
}
}
/**
* A repeatable event that can be paused or resumed by changing its
* interestOps. When the event is fired, it is first paused before being
* signaled. It is the responsibility of the code triggered by
* {@code signalEvent} to resume the event if required.
*/
private static abstract class SocketFlowEvent extends AsyncEvent {
final SocketChannel channel;
final int defaultInterest;
volatile int interestOps;
volatile boolean registered;
SocketFlowEvent(int defaultInterest, SocketChannel channel) {
super(AsyncEvent.REPEATING);
this.defaultInterest = defaultInterest;
this.channel = channel;
}
final boolean registered() {return registered;}
final void resume() {
interestOps = defaultInterest;
registered = true;
}
final void pause() {interestOps = 0;}
@Override
public final SelectableChannel channel() {return channel;}
@Override
public final int interestOps() {return interestOps;}
@Override
public final void handle() {
pause(); // pause, then signal
signalEvent(); // won't be fired again until resumed.
}
@Override
public final void abort(IOException error) {
debug().log(() -> "abort: " + error);
pause(); // pause, then signal
signalError(error); // should not be resumed after abort (not checked)
}
protected abstract void signalEvent();
protected abstract void signalError(Throwable error);
abstract Logger debug();
}
// ===================================================================== //
// Writing //
// ======================================================================//
// This class makes the assumption that the publisher will call onNext
// sequentially, and that onNext won't be called if the demand has not been
// incremented by request(1).
// It has a 'queue of 1' meaning that it will call request(1) in
// onSubscribe, and then only after its 'current' buffer list has been
// fully written and current set to null;
private final class InternalWriteSubscriber
implements Flow.Subscriber<List<ByteBuffer>> {
volatile WriteSubscription subscription;
volatile List<ByteBuffer> current;
volatile boolean completed;
final AsyncTriggerEvent startSubscription =
new AsyncTriggerEvent(this::signalError, this::startSubscription);
final WriteEvent writeEvent = new WriteEvent(channel, this);
final Demand writeDemand = new Demand();
@Override
public void onSubscribe(Flow.Subscription subscription) {
WriteSubscription previous = this.subscription;
if (debug.on()) debug.log("subscribed for writing");
try {
boolean needEvent = current == null;
if (needEvent) {
if (previous != null && previous.upstreamSubscription != subscription) {
previous.dropSubscription();
}
}
this.subscription = new WriteSubscription(subscription);
if (needEvent) {
if (debug.on())
debug.log("write: registering startSubscription event");
client.registerEvent(startSubscription);
}
} catch (Throwable t) {
signalError(t);
}
}
@Override
public void onNext(List<ByteBuffer> bufs) {
assert current == null : dbgString() // this is a queue of 1.
+ "w.onNext current: " + current;
assert subscription != null : dbgString()
+ "w.onNext: subscription is null";
current = bufs;
tryFlushCurrent(client.isSelectorThread()); // may be in selector thread
// For instance in HTTP/2, a received SETTINGS frame might trigger
// the sending of a SETTINGS frame in turn which might cause
// onNext to be called from within the same selector thread that the
// original SETTINGS frames arrived on. If rs is the read-subscriber
// and ws is the write-subscriber then the following can occur:
// ReadEvent -> rs.onNext(bytes) -> process server SETTINGS -> write
// client SETTINGS -> ws.onNext(bytes) -> tryFlushCurrent
debugState("leaving w.onNext");
}
// Don't use a SequentialScheduler here: rely on onNext() being invoked
// sequentially, and not being invoked if there is no demand, request(1).
// onNext is usually called from within a user / executor thread.
// Initial writing will be performed in that thread. If for some reason,
// not all the data can be written, a writeEvent will be registered, and
// writing will resume in the selector manager thread when the
// writeEvent is fired.
//
// If this method is invoked in the selector manager thread (because of
// a writeEvent), then the executor will be used to invoke request(1),
// ensuring that onNext() won't be invoked from within the selector
// thread. If not in the selector manager thread, then request(1) is
// invoked directly.
void tryFlushCurrent(boolean inSelectorThread) {
List<ByteBuffer> bufs = current;
if (bufs == null) return;
try {
assert inSelectorThread == client.isSelectorThread() :
"should " + (inSelectorThread ? "" : "not ")
+ " be in the selector thread";
long remaining = Utils.remaining(bufs);
if (debug.on()) debug.log("trying to write: %d", remaining);
long written = writeAvailable(bufs);
if (debug.on()) debug.log("wrote: %d", written);
assert written >= 0 : "negative number of bytes written:" + written;
assert written <= remaining;
if (remaining - written == 0) {
current = null;
if (writeDemand.tryDecrement()) {
Runnable requestMore = this::requestMore;
if (inSelectorThread) {
assert client.isSelectorThread();
client.theExecutor().execute(requestMore);
} else {
assert !client.isSelectorThread();
requestMore.run();
}
}
} else {
resumeWriteEvent(inSelectorThread);
}
} catch (Throwable t) {
signalError(t);
}
}
// Kick off the initial request:1 that will start the writing side.
// Invoked in the selector manager thread.
void startSubscription() {
try {
if (debug.on()) debug.log("write: starting subscription");
if (Log.channel()) {
Log.logChannel("Start requesting bytes for writing to channel: {0}",
channelDescr());
}
assert client.isSelectorThread();
// make sure read registrations are handled before;
readPublisher.subscriptionImpl.handlePending();
if (debug.on()) debug.log("write: offloading requestMore");
// start writing;
client.theExecutor().execute(this::requestMore);
} catch(Throwable t) {
signalError(t);
}
}
void requestMore() {
WriteSubscription subscription = this.subscription;
subscription.requestMore();
}
@Override
public void onError(Throwable throwable) {
signalError(throwable);
}
@Override
public void onComplete() {
completed = true;
// no need to pause the write event here: the write event will
// be paused if there is nothing more to write.
List<ByteBuffer> bufs = current;
long remaining = bufs == null ? 0 : Utils.remaining(bufs);
if (debug.on())
debug.log( "write completed, %d yet to send", remaining);
debugState("InternalWriteSubscriber::onComplete");
}
void resumeWriteEvent(boolean inSelectorThread) {
if (debug.on()) debug.log("scheduling write event");
resumeEvent(writeEvent, this::signalError);
}
void signalWritable() {
if (debug.on()) debug.log("channel is writable");
tryFlushCurrent(true);
}
void signalError(Throwable error) {
debug.log(() -> "write error: " + error);
if (Log.channel()) {
Log.logChannel("Failed to write to channel ({0}: {1})",
channelDescr(), error);
}
completed = true;
readPublisher.signalError(error);
Flow.Subscription subscription = this.subscription;
if (subscription != null) subscription.cancel();
}
// A repeatable WriteEvent which is paused after firing and can
// be resumed if required - see SocketFlowEvent;
final class WriteEvent extends SocketFlowEvent {
final InternalWriteSubscriber sub;
WriteEvent(SocketChannel channel, InternalWriteSubscriber sub) {
super(SelectionKey.OP_WRITE, channel);
this.sub = sub;
}
@Override
protected final void signalEvent() {
try {
client.eventUpdated(this);
sub.signalWritable();
} catch(Throwable t) {
sub.signalError(t);
}
}
@Override
protected void signalError(Throwable error) {
sub.signalError(error);
}
@Override
Logger debug() { return debug; }
}
final class WriteSubscription implements Flow.Subscription {
final Flow.Subscription upstreamSubscription;
volatile boolean cancelled;
WriteSubscription(Flow.Subscription subscription) {
this.upstreamSubscription = subscription;
}
@Override
public void request(long n) {
if (cancelled) return;
upstreamSubscription.request(n);
}
@Override
public void cancel() {
if (cancelled) return;
if (debug.on()) debug.log("write: cancel");
if (Log.channel()) {
Log.logChannel("Cancelling write subscription");
}
dropSubscription();
upstreamSubscription.cancel();
}
void dropSubscription() {
synchronized (InternalWriteSubscriber.this) {
cancelled = true;
if (debug.on()) debug.log("write: resetting demand to 0");
writeDemand.reset();
}
}
void requestMore() {
try {
if (completed || cancelled) return;
boolean requestMore;
long d;
// don't fiddle with demand after cancel.
// see dropSubscription.
synchronized (InternalWriteSubscriber.this) {
if (cancelled) return;
d = writeDemand.get();
requestMore = writeDemand.increaseIfFulfilled();
}
if (requestMore) {
if (debug.on()) debug.log("write: requesting more...");
upstreamSubscription.request(1);
} else {
if (debug.on())
debug.log("write: no need to request more: %d", d);
}
} catch (Throwable t) {
if (debug.on())
debug.log("write: error while requesting more: " + t);
signalError(t);
} finally {
debugState("leaving requestMore: ");
}
}
}
}
// ===================================================================== //
// Reading //
// ===================================================================== //
// The InternalReadPublisher uses a SequentialScheduler to ensure that
// onNext/onError/onComplete are called sequentially on the caller's
// subscriber.
// However, it relies on the fact that the only time where
// runOrSchedule() is called from a user/executor thread is in signalError,
// right after the errorRef has been set.
// Because the sequential scheduler's task always checks for errors first,
// and always terminate the scheduler on error, then it is safe to assume
// that if it reaches the point where it reads from the channel, then
// it is running in the SelectorManager thread. This is because all
// other invocation of runOrSchedule() are triggered from within a
// ReadEvent.
//
// When pausing/resuming the event, some shortcuts can then be taken
// when we know we're running in the selector manager thread
// (in that case there's no need to call client.eventUpdated(readEvent);
//
private final class InternalReadPublisher
implements Flow.Publisher<List<ByteBuffer>> {
private final InternalReadSubscription subscriptionImpl
= new InternalReadSubscription();
AtomicReference<ReadSubscription> pendingSubscription = new AtomicReference<>();
private volatile ReadSubscription subscription;
@Override
public void subscribe(Flow.Subscriber<? super List<ByteBuffer>> s) {
Objects.requireNonNull(s);
TubeSubscriber sub = FlowTube.asTubeSubscriber(s);
ReadSubscription target = new ReadSubscription(subscriptionImpl, sub);
ReadSubscription previous = pendingSubscription.getAndSet(target);
if (previous != null && previous != target) {
if (debug.on())
debug.log("read publisher: dropping pending subscriber: "
+ previous.subscriber);
previous.errorRef.compareAndSet(null, errorRef.get());
previous.signalOnSubscribe();
if (subscriptionImpl.completed) {
previous.signalCompletion();
} else {
previous.subscriber.dropSubscription();
}
}
if (debug.on()) debug.log("read publisher got subscriber");
subscriptionImpl.signalSubscribe();
debugState("leaving read.subscribe: ");
}
void signalError(Throwable error) {
if (debug.on()) debug.log("error signalled " + error);
if (!errorRef.compareAndSet(null, error)) {
return;
}
if (Log.channel()) {
Log.logChannel("Error signalled on channel {0}: {1}",
channelDescr(), error);
}
subscriptionImpl.handleError();
}
final class ReadSubscription implements Flow.Subscription {
final InternalReadSubscription impl;
final TubeSubscriber subscriber;
final AtomicReference<Throwable> errorRef = new AtomicReference<>();
final BufferSource bufferSource;
volatile boolean subscribed;
volatile boolean cancelled;
volatile boolean completed;
public ReadSubscription(InternalReadSubscription impl,
TubeSubscriber subscriber) {
this.impl = impl;
this.bufferSource = subscriber.supportsRecycling()
? new SSLDirectBufferSource(client)
: SocketTube.this.sliceBuffersSource;
this.subscriber = subscriber;
}
@Override
public void cancel() {
cancelled = true;
}
@Override
public void request(long n) {
if (!cancelled) {
impl.request(n);
} else {
if (debug.on())
debug.log("subscription cancelled, ignoring request %d", n);
}
}
void signalCompletion() {
assert subscribed || cancelled;
if (completed || cancelled) return;
synchronized (this) {
if (completed) return;
completed = true;
}
Throwable error = errorRef.get();
if (error != null) {
if (debug.on())
debug.log("forwarding error to subscriber: " + error);
subscriber.onError(error);
} else {
if (debug.on()) debug.log("completing subscriber");
subscriber.onComplete();
}
}
void signalOnSubscribe() {
if (subscribed || cancelled) return;
synchronized (this) {
if (subscribed || cancelled) return;
subscribed = true;
}
subscriber.onSubscribe(this);
if (debug.on()) debug.log("onSubscribe called");
if (errorRef.get() != null) {
signalCompletion();
}
}
}
final class InternalReadSubscription implements Flow.Subscription {
private final Demand demand = new Demand();
final SequentialScheduler readScheduler;
private volatile boolean completed;
private final ReadEvent readEvent;
private final AsyncEvent subscribeEvent;
InternalReadSubscription() {
readScheduler = new SequentialScheduler(new SocketFlowTask(this::read));
subscribeEvent = new AsyncTriggerEvent(this::signalError,
this::handleSubscribeEvent);
readEvent = new ReadEvent(channel, this);
}
/*
* This method must be invoked before any other method of this class.
*/
final void signalSubscribe() {
if (readScheduler.isStopped() || completed) {
// if already completed or stopped we can handle any
// pending connection directly from here.
if (debug.on())
debug.log("handling pending subscription while completed");
handlePending();
} else {
try {
if (debug.on()) debug.log("registering subscribe event");
client.registerEvent(subscribeEvent);
} catch (Throwable t) {
signalError(t);
handlePending();
}
}
}
final void handleSubscribeEvent() {
assert client.isSelectorThread();
debug.log("subscribe event raised");
if (Log.channel()) Log.logChannel("Start reading from {0}", channelDescr());
readScheduler.runOrSchedule();
if (readScheduler.isStopped() || completed) {
// if already completed or stopped we can handle any
// pending connection directly from here.
if (debug.on())
debug.log("handling pending subscription when completed");
handlePending();
}
}
/*
* Although this method is thread-safe, the Reactive-Streams spec seems
* to not require it to be as such. It's a responsibility of the
* subscriber to signal demand in a thread-safe manner.
*
* See Reactive Streams specification, rules 2.7 and 3.4.
*/
@Override
public final void request(long n) {
if (n > 0L) {
boolean wasFulfilled = demand.increase(n);
if (wasFulfilled) {
if (debug.on()) debug.log("got some demand for reading");
resumeReadEvent();
// if demand has been changed from fulfilled
// to unfulfilled register read event;
}
} else {
signalError(new IllegalArgumentException("non-positive request"));
}
debugState("leaving request("+n+"): ");
}
@Override
public final void cancel() {
pauseReadEvent();
if (Log.channel()) {
Log.logChannel("Read subscription cancelled for channel {0}",
channelDescr());
}
readScheduler.stop();
}
private void resumeReadEvent() {
if (debug.on()) debug.log("resuming read event");
resumeEvent(readEvent, this::signalError);
}
private void pauseReadEvent() {
if (debug.on()) debug.log("pausing read event");
pauseEvent(readEvent, this::signalError);
}
final void handleError() {
assert errorRef.get() != null;
readScheduler.runOrSchedule();
}
final void signalError(Throwable error) {
if (!errorRef.compareAndSet(null, error)) {
return;
}
if (debug.on()) debug.log("got read error: " + error);
if (Log.channel()) {
Log.logChannel("Read error signalled on channel {0}: {1}",
channelDescr(), error);
}
readScheduler.runOrSchedule();
}
final void signalReadable() {
readScheduler.runOrSchedule();
}
/** The body of the task that runs in SequentialScheduler. */
final void read() {
// It is important to only call pauseReadEvent() when stopping
// the scheduler. The event is automatically paused before
// firing, and trying to pause it again could cause a race
// condition between this loop, which calls tryDecrementDemand(),
// and the thread that calls request(n), which will try to resume
// reading.
try {
while(!readScheduler.isStopped()) {
if (completed) return;
// make sure we have a subscriber
if (handlePending()) {
if (debug.on())
debug.log("pending subscriber subscribed");
return;
}
// If an error was signaled, we might not be in the
// the selector thread, and that is OK, because we
// will just call onError and return.
ReadSubscription current = subscription;
Throwable error = errorRef.get();
if (current == null) {
assert error != null;
if (debug.on())
debug.log("error raised before subscriber subscribed: %s",
(Object)error);
return;
}
TubeSubscriber subscriber = current.subscriber;
if (error != null) {
completed = true;
// safe to pause here because we're finished anyway.
pauseReadEvent();
if (debug.on())
debug.log("Sending error " + error
+ " to subscriber " + subscriber);
if (Log.channel()) {
Log.logChannel("Raising error with subscriber for {0}: {1}",
channelDescr(), error);
}
current.errorRef.compareAndSet(null, error);
current.signalCompletion();
readScheduler.stop();
debugState("leaving read() loop with error: ");
return;
}
// If we reach here then we must be in the selector thread.
assert client.isSelectorThread();
if (demand.tryDecrement()) {
// we have demand.
try {
List<ByteBuffer> bytes = readAvailable(current.bufferSource);
if (bytes == EOF) {
if (!completed) {
if (debug.on()) debug.log("got read EOF");
if (Log.channel()) {
Log.logChannel("EOF read from channel: {0}",
channelDescr());
}
completed = true;
// safe to pause here because we're finished
// anyway.
pauseReadEvent();
current.signalCompletion();
readScheduler.stop();
}
debugState("leaving read() loop after EOF: ");
return;
} else if (Utils.remaining(bytes) > 0) {
// the subscriber is responsible for offloading
// to another thread if needed.
if (debug.on())
debug.log("read bytes: " + Utils.remaining(bytes));
assert !current.completed;
subscriber.onNext(bytes);
// we could continue looping until the demand
// reaches 0. However, that would risk starving
// other connections (bound to other socket
// channels) - as other selected keys activated
// by the selector manager thread might be
// waiting for this event to terminate.
// So resume the read event and return now...
resumeReadEvent();
debugState("leaving read() loop after onNext: ");
return;
} else {
// nothing available!
if (debug.on()) debug.log("no more bytes available");
// re-increment the demand and resume the read
// event. This ensures that this loop is
// executed again when the socket becomes
// readable again.
demand.increase(1);
resumeReadEvent();
debugState("leaving read() loop with no bytes");
return;
}
} catch (Throwable x) {
signalError(x);
continue;
}
} else {
if (debug.on()) debug.log("no more demand for reading");
// the event is paused just after firing, so it should
// still be paused here, unless the demand was just
// incremented from 0 to n, in which case, the
// event will be resumed, causing this loop to be
// invoked again when the socket becomes readable:
// This is what we want.
// Trying to pause the event here would actually
// introduce a race condition between this loop and
// request(n).
debugState("leaving read() loop with no demand");
break;
}
}
} catch (Throwable t) {
if (debug.on()) debug.log("Unexpected exception in read loop", t);
signalError(t);
} finally {
if (readScheduler.isStopped()) {
if (debug.on()) debug.log("Read scheduler stopped");
if (Log.channel()) {
Log.logChannel("Stopped reading from channel {0}", channelDescr());
}
}
handlePending();
}
}
boolean handlePending() {
ReadSubscription pending = pendingSubscription.getAndSet(null);
if (pending == null) return false;
if (debug.on())
debug.log("handling pending subscription for %s",
pending.subscriber);
ReadSubscription current = subscription;
if (current != null && current != pending && !completed) {
current.subscriber.dropSubscription();
}
if (debug.on()) debug.log("read demand reset to 0");
subscriptionImpl.demand.reset(); // subscriber will increase demand if it needs to.
pending.errorRef.compareAndSet(null, errorRef.get());
if (!readScheduler.isStopped()) {
subscription = pending;
} else {
if (debug.on()) debug.log("socket tube is already stopped");
}
if (debug.on()) debug.log("calling onSubscribe");
pending.signalOnSubscribe();
if (completed) {
pending.errorRef.compareAndSet(null, errorRef.get());
pending.signalCompletion();
}
return true;
}
}
// A repeatable ReadEvent which is paused after firing and can
// be resumed if required - see SocketFlowEvent;
final class ReadEvent extends SocketFlowEvent {
final InternalReadSubscription sub;
ReadEvent(SocketChannel channel, InternalReadSubscription sub) {
super(SelectionKey.OP_READ, channel);
this.sub = sub;
}
@Override
protected final void signalEvent() {
try {
client.eventUpdated(this);
sub.signalReadable();
} catch(Throwable t) {
sub.signalError(t);
}
}
@Override
protected final void signalError(Throwable error) {
sub.signalError(error);
}
@Override
Logger debug() { return debug; }
}
}
// ===================================================================== //
// Buffer Management //
// ===================================================================== //
// This interface is used by readAvailable(BufferSource);
public interface BufferSource {
/**
* Returns a buffer to read data from the socket.
*
* @implNote
* Different implementation can have different strategies, as to
* which kind of buffer to return, or whether to return the same
* buffer. The only constraints are that:
* a. the buffer returned must not be null
* b. the buffer position indicates where to start reading
* c. the buffer limit indicates where to stop reading.
* d. the buffer is 'free' - that is - it is not used
* or retained by anybody else
*
* @return A buffer to read data from the socket.
*/
ByteBuffer getBuffer();
/**
* Appends the read-data in {@code buffer} to the list of buffer to
* be sent downstream to the subscriber. May return a new
* list, or append to the given list.
*
* @implNote
* Different implementation can have different strategies, but
* must obviously be consistent with the implementation of the
* getBuffer() method. For instance, an implementation could
* decide to add the buffer to the list and return a new buffer
* next time getBuffer() is called, or could decide to add a buffer
* slice to the list and return the same buffer (if remaining
* space is available) next time getBuffer() is called.
*
* @param list The list before adding the data. Can be null.
* @param buffer The buffer containing the data to add to the list.
* @param start The start position at which data were read.
* The current buffer position indicates the end.
* @return A possibly new list where a buffer containing the
* data read from the socket has been added.
*/
List<ByteBuffer> append(List<ByteBuffer> list, ByteBuffer buffer, int start);
/**
* Returns the given unused {@code buffer}, previously obtained from
* {@code getBuffer}.
*
* @implNote This method can be used, if necessary, to return
* the unused buffer to the pull.
*
* @param buffer The unused buffer.
*/
default void returnUnused(ByteBuffer buffer) { }
}
// An implementation of BufferSource used for unencrypted data.
// This buffer source uses heap buffers and avoids wasting memory
// by forwarding read-only buffer slices downstream.
// Buffers allocated through this source are simply GC'ed when
// they are no longer referenced.
private static final class SliceBufferSource implements BufferSource {
private final Supplier<ByteBuffer> factory;
private volatile ByteBuffer current;
public SliceBufferSource() {
this(Utils::getBuffer);
}
public SliceBufferSource(Supplier<ByteBuffer> factory) {
this.factory = Objects.requireNonNull(factory);
}
// Reuses the same buffer if some space remains available.
// Otherwise, returns a new heap buffer.
@Override
public final ByteBuffer getBuffer() {
ByteBuffer buf = current;
buf = (buf == null || !buf.hasRemaining())
? (current = factory.get()) : buf;
assert buf.hasRemaining();
return buf;
}
// Adds a read-only slice to the list, potentially returning a
// new list with that slice at the end.
@Override
public final List<ByteBuffer> append(List <ByteBuffer> list, ByteBuffer buf, int start) {
// creates a slice to add to the list
int limit = buf.limit();
buf.limit(buf.position());
buf.position(start);
ByteBuffer slice = buf.slice();
// restore buffer state to what it was before creating the slice
buf.position(buf.limit());
buf.limit(limit);
// add the buffer to the list
return SocketTube.listOf(list, slice.asReadOnlyBuffer());
}
}
// An implementation of BufferSource used for encrypted data.
// This buffer source uses direct byte buffers that will be
// recycled by the SocketTube subscriber.
//
private static final class SSLDirectBufferSource implements BufferSource {
private final BufferSupplier factory;
private final HttpClientImpl client;
private ByteBuffer current;
public SSLDirectBufferSource(HttpClientImpl client) {
this.client = Objects.requireNonNull(client);
this.factory = Objects.requireNonNull(client.getSSLBufferSupplier());
}
// Obtains a 'free' byte buffer from the pool, or returns
// the same buffer if nothing was read at the previous cycle.
// The subscriber will be responsible for recycling this
// buffer into the pool (see SSLFlowDelegate.Reader)
@Override
public final ByteBuffer getBuffer() {
assert client.isSelectorThread();
ByteBuffer buf = current;
if (buf == null) {
buf = current = factory.get();
}
assert buf.hasRemaining();
assert buf.position() == 0;
return buf;
}
// Adds the buffer to the list. The buffer will be later returned to the
// pool by the subscriber (see SSLFlowDelegate.Reader).
// The next buffer returned by getBuffer() will be obtained from the
// pool. It might be the same buffer or another one.
// Because socket tube can read up to MAX_BUFFERS = 3 buffers, and because
// recycling will happen in the flow before onNext returns, then the
// pool can not grow larger than MAX_BUFFERS = 3 buffers, even though
// it's shared by all SSL connections opened on that client.
@Override
public final List<ByteBuffer> append(List <ByteBuffer> list, ByteBuffer buf, int start) {
assert client.isSelectorThread();
assert buf.isDirect();
assert start == 0;
assert current == buf;
current = null;
buf.limit(buf.position());
buf.position(start);
// add the buffer to the list
return SocketTube.listOf(list, buf);
}
@Override
public void returnUnused(ByteBuffer buffer) {
// if current is null, then the buffer will have been added to the
// list, through append. Otherwise, current is not null, and needs
// to be returned to prevent the buffer supplier pool from growing
// to more than MAX_BUFFERS.
assert buffer == current;
ByteBuffer buf = current;
if (buf != null) {
assert buf.position() == 0;
current = null;
// the supplier assert if buf has remaining
buf.limit(buf.position());
factory.recycle(buf);
}
}
}
// ===================================================================== //
// Socket Channel Read/Write //
// ===================================================================== //
static final int MAX_BUFFERS = 3;
static final List<ByteBuffer> EOF = List.of();
static final List<ByteBuffer> NOTHING = List.of(Utils.EMPTY_BYTEBUFFER);
// readAvailable() will read bytes into the 'current' ByteBuffer until
// the ByteBuffer is full, or 0 or -1 (EOF) is returned by read().
// When that happens, a slice of the data that has been read so far
// is inserted into the returned buffer list, and if the current buffer
// has remaining space, that space will be used to read more data when
// the channel becomes readable again.
private List<ByteBuffer> readAvailable(BufferSource buffersSource) throws IOException {
ByteBuffer buf = buffersSource.getBuffer();
assert buf.hasRemaining();
int read;
int pos = buf.position();
List<ByteBuffer> list = null;
while (buf.hasRemaining()) {
try {
while ((read = channel.read(buf)) > 0) {
if (!buf.hasRemaining())
break;
}
} catch (IOException x) {
if (buf.position() == pos && list == null) {
// make sure that the buffer source will recycle
// 'buf' if needed
buffersSource.returnUnused(buf);
// no bytes have been read, just throw...
throw x;
} else {
// some bytes have been read, return them and fail next time
errorRef.compareAndSet(null, x);
read = 0; // ensures outer loop will exit
}
}
// nothing read;
if (buf.position() == pos) {
// An empty list signals the end of data, and should only be
// returned if read == -1. If some data has already been read,
// then it must be returned. -1 will be returned next time
// the caller attempts to read something.
buffersSource.returnUnused(buf);
if (list == null) {
// nothing read - list was null - return EOF or NOTHING
list = read == -1 ? EOF : NOTHING;
}
break;
}
// check whether this buffer has still some free space available.
// if so, we will keep it for the next round.
list = buffersSource.append(list, buf, pos);
if (read <= 0 || list.size() == MAX_BUFFERS) {
break;
}
buf = buffersSource.getBuffer();
pos = buf.position();
assert buf.hasRemaining();
}
return list;
}
private static <T> List<T> listOf(List<T> list, T item) {
int size = list == null ? 0 : list.size();
switch (size) {
case 0: return List.of(item);
case 1: return List.of(list.get(0), item);
case 2: return List.of(list.get(0), list.get(1), item);
default: // slow path if MAX_BUFFERS > 3
List<T> res = list instanceof ArrayList ? list : new ArrayList<>(list);
res.add(item);
return res;
}
}
private long writeAvailable(List<ByteBuffer> bytes) throws IOException {
ByteBuffer[] srcs = bytes.toArray(Utils.EMPTY_BB_ARRAY);
final long remaining = Utils.remaining(srcs);
long written = 0;
while (remaining > written) {
try {
long w = channel.write(srcs);
assert w >= 0 : "negative number of bytes written:" + w;
if (w == 0) {
break;
}
written += w;
} catch (IOException x) {
if (written == 0) {
// no bytes were written just throw
throw x;
} else {
// return how many bytes were written, will fail next time
break;
}
}
}
return written;
}
private void resumeEvent(SocketFlowEvent event,
Consumer<Throwable> errorSignaler) {
boolean registrationRequired;
synchronized(lock) {
registrationRequired = !event.registered();
event.resume();
}
try {
if (registrationRequired) {
client.registerEvent(event);
} else {
client.eventUpdated(event);
}
} catch(Throwable t) {
errorSignaler.accept(t);
}
}
private void pauseEvent(SocketFlowEvent event,
Consumer<Throwable> errorSignaler) {
synchronized(lock) {
event.pause();
}
try {
client.eventUpdated(event);
} catch(Throwable t) {
errorSignaler.accept(t);
}
}
@Override
public void connectFlows(TubePublisher writePublisher,
TubeSubscriber readSubscriber) {
if (debug.on()) debug.log("connecting flows");
this.subscribe(readSubscriber);
writePublisher.subscribe(this);
}
@Override
public String toString() {
return dbgString();
}
final String dbgString() {
return "SocketTube("+id+")";
}
final String channelDescr() {
return String.valueOf(channel);
}
}