Skip to content

Request/Response Fragmentation & Assembly #63

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 2 commits into from
Closed

Request/Response Fragmentation & Assembly #63

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5454bb2
Request/Response Fragmentation & Assembly
benjchristensen Sep 22, 2015
b4aa8eb
Bug Fixes
benjchristensen Sep 25, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
304 changes: 287 additions & 17 deletions src/main/java/io/reactivesocket/internal/FragmentedPublisher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,43 +15,313 @@
*/
package io.reactivesocket.internal;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import io.reactivesocket.Frame;
import io.reactivesocket.FrameType;
import io.reactivesocket.Payload;
import io.reactivesocket.internal.frame.PayloadFragmenter;
import io.reactivesocket.internal.rx.QueueDrainHelper;
import io.reactivesocket.internal.rx.SerializedSubscriber;
import io.reactivesocket.internal.rx.SubscriptionArbiter;
import uk.co.real_logic.agrona.concurrent.OneToOneConcurrentArrayQueue;

public class FragmentedPublisher implements Publisher<Frame> {

private final PayloadFragmenter fragmenter = new PayloadFragmenter(Frame.METADATA_MTU, Frame.DATA_MTU);
private final Publisher<Payload> responsePublisher;
private final int streamId;
private final FrameType type;
private volatile Subscriber<Frame> downstream;
private volatile SubscriptionArbiter sa;
private final AtomicInteger wip = new AtomicInteger();

// TODO use better data structures
private final List<InnerSubscriber> subscribers = Collections.synchronizedList(new ArrayList<InnerSubscriber>(16));
private final Queue<InnerSubscriber> toRemove = new ConcurrentLinkedQueue<InnerSubscriber>();

private int index = 0;
private static final Object COMPLETED = new Object();

public FragmentedPublisher(FrameType type, int streamId, Publisher<Payload> responsePublisher) {
this.type = type;
this.streamId = streamId;
this.responsePublisher = responsePublisher;
}

@Override
public void subscribe(Subscriber<? super Frame> child) {
child.onSubscribe(new Subscription() {
SerializedSubscriber<Frame> ssub = new SerializedSubscriber<>(child);
sa = new SubscriptionArbiter();
sa.setSubscription(new Subscription() {

@Override
public void request(long n) {
// TODO Auto-generated method stub

tryEmit();
}

@Override
public void cancel() {
// TODO Auto-generated method stub

}});
sa.cancel();
for (InnerSubscriber is : subscribers) {
is._s.cancel();
}
}

});
ssub.onSubscribe(sa);
downstream = ssub;
}

public void tryEmit() {
QueueDrainHelper.queueDrainLoop(wip,
this::drainInnerSubscribers, /* drain if we get the emission lock */
() -> {
/* nothing to do if another thread has the emission lock */} ,
this::drainInnerSubscribers /* drain if others tried while we were emitting */);
}

private void drainInnerSubscribers() {
long r = sa.getRequested();
int numItems = subscribers.size();
int emitted = 0;
int startIndex = index;
if (subscribers.size() == 0) {
return;
}
while (emitted < r) {
if (index >= subscribers.size()) {
index = 0;
}
InnerSubscriber is = subscribers.get(index++);
numItems--;
if (is == null) {
break;
}
emitted += is.drain(r - emitted);
if (numItems == 0) {
break;
}
if (index == startIndex) {
// looped around, break out so this thread isn't starved
break;
}
}
sa.produced(emitted);
for (InnerSubscriber is : toRemove) {
subscribers.remove(is);
toRemove.remove(is);
}
}

/**
* Horizontally Unbounded submission of Publisher. This means as many "concurrent" outputs as wanted.
* <p>
* This is ultimately controlled by lease semantics which controls how many streams (requests) are in-flight.
*
* @param t
*/
public void submit(Publisher<Frame> t) {
if (downstream == null) {
throw new IllegalStateException("Downstream has not yet subscribed. Please await() subscription");
}

// horizontally no backpressure, we subscribe to all incoming Publishers, and then backpressure their individual streams
InnerSubscriber is = new InnerSubscriber(this);
subscribers.add(is);
t.subscribe(is);
}

/**
* Asynchronously subscribe to Frames, buffer them internally if needed.
*/
private static final class InnerSubscriber implements Subscriber<Frame> {
private final AtomicLong outstanding = new AtomicLong();
// TODO replace this as this is very inefficient
private volatile OneToOneConcurrentArrayQueue<Object> q;
final FragmentedPublisher parent;
private Subscription _s;
static final int BATCH = 128;

public InnerSubscriber(FragmentedPublisher parent) {
this.parent = parent;
}

@Override
public void onSubscribe(Subscription s) {
_s = s;
// we manage our own rate since the transport should go as fast as it can, no application level flow control here
s.request(BATCH);
outstanding.set(BATCH);
}

@Override
public void onNext(Frame frame) {
FrameType type = frame.getType();
int streamId = frame.getStreamId();
if (PayloadFragmenter.requiresFragmenting(Frame.METADATA_MTU, Frame.DATA_MTU, frame)) {
nextFragmented(frame, type, streamId);
} else {
nextUnfragmented(frame);
}
}

private void nextUnfragmented(Frame frame) {
QueueDrainHelper.queueDrainLoop(parent.wip,
/* fast-path if no contention when trying to emit */
() -> {
if (parent.sa.getRequested() == 0) {
createQueueIfNecessary();
q.add(frame);
return;
}
if (q == null || q.size() == 0) {
parent.downstream.onNext(frame);
parent.sa.produced(1);
outstanding.decrementAndGet();
InnerSubscriber.this.requestMoreIfNeeded();
} else {
// enqueue, then drain if there are already things in the queue
q.add(frame);
drainRequestedAndRequestMoreIfNeeded();
}
} ,
/* if contended, then just enqueue */
() -> {
createQueueIfNecessary();
q.add(frame);
} ,
/* if another thread enqueued while emitting above, then this will have a chance to drain after */
parent::drainInnerSubscribers
);
}

private void nextFragmented(Frame frame, FrameType type, int streamId) {
// not reusing each time since I need the Iterator state stored through request(n) and can have several in a queue
PayloadFragmenter fragmenter = new PayloadFragmenter(Frame.METADATA_MTU, Frame.DATA_MTU);
if (FrameType.NEXT_COMPLETE.equals(type)) {
fragmenter.resetForResponseComplete(streamId, frame);
} else {
fragmenter.resetForResponse(streamId, frame);
}

QueueDrainHelper.queueDrainLoop(parent.wip,
/* fast-path if no contention when trying to emit */
() -> {
if (parent.sa.getRequested() == 0) {
createQueueIfNecessary();
q.add(frame);
return;
}
if (q == null || q.size() == 0) {
long r = parent.sa.getRequested();
long emitted = 0;
// emit as much of iterable as requested allows
for (int i = 0; i < r; i++) { // TODO limit so we don't head-of-line block
if (fragmenter.hasNext()) {
parent.downstream.onNext(fragmenter.next());
emitted++;
} else {
break;
}
}
parent.sa.produced(emitted);
outstanding.addAndGet(-emitted);
InnerSubscriber.this.requestMoreIfNeeded();

if (fragmenter.hasNext()) {
// not finished so enqueue
createQueueIfNecessary();
q.add(fragmenter);
}
} else {
// enqueue, then drain if there are already things in the queue
q.add(fragmenter);
drainRequestedAndRequestMoreIfNeeded();
}
} ,
/* if contended, then just enqueue */
() -> {
createQueueIfNecessary();
q.add(fragmenter);
} ,
/* if another thread enqueued while emitting above, then this will have a chance to drain after */
parent::drainInnerSubscribers);
}

private void createQueueIfNecessary() {
if(q == null) {
q = new OneToOneConcurrentArrayQueue<Object>(BATCH);
}
}

private void drainRequestedAndRequestMoreIfNeeded() {
long emitted = drain(parent.sa.getRequested());
parent.sa.produced(emitted);
outstanding.addAndGet(-emitted);
InnerSubscriber.this.requestMoreIfNeeded();
}

public long drain(long maxToDrain) {
if(q == null) {
return 0;
}
long emitted = 0;
while (emitted < maxToDrain) {
Object o = q.peek();
if (o == null) {
break;
}
if (o instanceof Frame) {
parent.downstream.onNext((Frame) o);
emitted++;
q.poll(); // pop it off the queue
} else if (o == COMPLETED) {
parent.toRemove.add(InnerSubscriber.this);
break;
} else {
@SuppressWarnings("unchecked")
Iterator<Frame> ifs = (Iterator<Frame>) o;
while (ifs.hasNext()) {
parent.downstream.onNext(ifs.next());
emitted++;
if (emitted == maxToDrain) {
break;
}
}
if (!ifs.hasNext()) {
// finished, so remove it
q.poll();
}
}
}
return emitted;
}

private void requestMoreIfNeeded() {
long current = outstanding.get();
if (current < 20) {
long d = BATCH - current;
_s.request(d);
outstanding.addAndGet(d);
}
}

@Override
public void onError(Throwable t) {
parent.sa.cancel();
parent.downstream.onError(t);
parent.subscribers.remove(this);
}

@Override
public void onComplete() {
if (q == null || q.size() == 0) {
parent.subscribers.remove(this);
} else {
q.add(COMPLETED);
}
}
}

}
24 changes: 24 additions & 0 deletions src/main/java/io/reactivesocket/internal/Requester.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,8 @@
import io.reactivesocket.exceptions.CancelException;
import io.reactivesocket.exceptions.Exceptions;
import io.reactivesocket.exceptions.Retryable;
import io.reactivesocket.internal.frame.FrameHeaderFlyweight;
import io.reactivesocket.internal.frame.PayloadBuilder;
import io.reactivesocket.internal.frame.RequestFrameFlyweight;
import io.reactivesocket.internal.rx.BackpressureUtils;
import io.reactivesocket.internal.rx.EmptyDisposable;
Expand Down Expand Up @@ -644,6 +646,7 @@ private final static class StreamInputSubscriber implements Subscriber<Frame> {
private final Subscriber<? super Payload> child;
private final Runnable cancelAction;
private final AtomicReference<Subscription> requestStreamSubscription;
private PayloadBuilder payloadBuilder; // created if fragmented

public StreamInputSubscriber(int streamId, long threshold, AtomicLong outstanding, AtomicLong requested, UnicastSubject<Frame> writer, Subscriber<? super Payload> child, Runnable cancelAction) {
this.streamId = streamId;
Expand Down Expand Up @@ -675,6 +678,27 @@ public void onSubscribe(Subscription s) {

@Override
public void onNext(Frame frame) {
if (FrameHeaderFlyweight.FLAGS_RESPONSE_F == (frame.flags() & FrameHeaderFlyweight.FLAGS_RESPONSE_F)) {
// fragment
if(payloadBuilder == null) {
payloadBuilder = new PayloadBuilder();
}
}
if(payloadBuilder != null) {
payloadBuilder.append(frame);
if (FrameHeaderFlyweight.FLAGS_RESPONSE_F != (frame.flags() & FrameHeaderFlyweight.FLAGS_RESPONSE_F)) {
// no more fragments, but we have a PayloadBuilder, so this is the final Frame to be reassembled
Payload payload = payloadBuilder.payload();
Frame assembled = Frame.Response.from(streamId, frame.getType(), payload);
// replace 'frame' with the assembled one
frame = assembled;
payloadBuilder = null;
} else {
// it was a fragment, so return without further processing
return;
}
}

FrameType type = frame.getType();
// convert ERROR messages into terminal events
if (type == FrameType.NEXT_COMPLETE) {
Expand Down
Loading