| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you 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.apache.cassandra.net; |
| |
| import java.nio.BufferOverflowException; |
| import java.util.Arrays; |
| import java.util.Map; |
| import java.util.concurrent.ConcurrentSkipListMap; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.concurrent.atomic.AtomicReferenceArray; |
| import java.util.concurrent.locks.LockSupport; |
| import java.util.function.Consumer; |
| |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import com.carrotsearch.hppc.LongObjectHashMap; |
| import com.carrotsearch.hppc.predicates.LongObjectPredicate; |
| import com.carrotsearch.hppc.procedures.LongObjectProcedure; |
| import com.carrotsearch.hppc.procedures.LongProcedure; |
| import org.apache.cassandra.net.Verifier.ExpiredMessageEvent.ExpirationType; |
| import org.apache.cassandra.utils.concurrent.WaitQueue; |
| |
| import static java.util.concurrent.TimeUnit.*; |
| import static java.util.concurrent.TimeUnit.NANOSECONDS; |
| import static org.apache.cassandra.net.MessagingService.VERSION_40; |
| import static org.apache.cassandra.net.MessagingService.current_version; |
| import static org.apache.cassandra.net.ConnectionType.LARGE_MESSAGES; |
| import static org.apache.cassandra.net.OutboundConnection.LargeMessageDelivery.DEFAULT_BUFFER_SIZE; |
| import static org.apache.cassandra.net.OutboundConnections.LARGE_MESSAGE_THRESHOLD; |
| import static org.apache.cassandra.net.Verifier.EventCategory.OTHER; |
| import static org.apache.cassandra.net.Verifier.EventCategory.RECEIVE; |
| import static org.apache.cassandra.net.Verifier.EventCategory.SEND; |
| import static org.apache.cassandra.net.Verifier.EventType.ARRIVE; |
| import static org.apache.cassandra.net.Verifier.EventType.CLOSED_BEFORE_ARRIVAL; |
| import static org.apache.cassandra.net.Verifier.EventType.DESERIALIZE; |
| import static org.apache.cassandra.net.Verifier.EventType.ENQUEUE; |
| import static org.apache.cassandra.net.Verifier.EventType.FAILED_CLOSING; |
| import static org.apache.cassandra.net.Verifier.EventType.FAILED_DESERIALIZE; |
| import static org.apache.cassandra.net.Verifier.EventType.FAILED_EXPIRED_ON_SEND; |
| import static org.apache.cassandra.net.Verifier.EventType.FAILED_EXPIRED_ON_RECEIVE; |
| import static org.apache.cassandra.net.Verifier.EventType.FAILED_FRAME; |
| import static org.apache.cassandra.net.Verifier.EventType.FAILED_OVERLOADED; |
| import static org.apache.cassandra.net.Verifier.EventType.FAILED_SERIALIZE; |
| import static org.apache.cassandra.net.Verifier.EventType.FINISH_SERIALIZE_LARGE; |
| import static org.apache.cassandra.net.Verifier.EventType.PROCESS; |
| import static org.apache.cassandra.net.Verifier.EventType.SEND_FRAME; |
| import static org.apache.cassandra.net.Verifier.EventType.SENT_FRAME; |
| import static org.apache.cassandra.net.Verifier.EventType.SERIALIZE; |
| import static org.apache.cassandra.net.Verifier.ExpiredMessageEvent.ExpirationType.ON_SENT; |
| import static org.apache.cassandra.utils.MonotonicClock.approxTime; |
| |
| /** |
| * This class is a single-threaded verifier monitoring a single link, with events supplied by inbound and outbound threads |
| * |
| * By making verification single threaded, it is easier to reason about (and complex enough as is), but also permits |
| * a dedicated thread to monitor timeliness of events, e.g. elapsed time between a given SEND and its corresponding RECEIVE |
| * |
| * TODO: timeliness of events |
| * TODO: periodically stop all activity to/from a given endpoint, until it stops (and verify queues all empty, counters all accurate) |
| * TODO: integrate with proxy that corrupts frames |
| * TODO: test _OutboundConnection_ close |
| */ |
| @SuppressWarnings("WeakerAccess") |
| public class Verifier |
| { |
| private static final Logger logger = LoggerFactory.getLogger(Verifier.class); |
| |
| public enum Destiny |
| { |
| SUCCEED, |
| FAIL_TO_SERIALIZE, |
| FAIL_TO_DESERIALIZE, |
| } |
| |
| enum EventCategory |
| { |
| SEND, RECEIVE, OTHER |
| } |
| |
| enum EventType |
| { |
| FAILED_OVERLOADED(SEND), |
| ENQUEUE(SEND), |
| FAILED_EXPIRED_ON_SEND(SEND), |
| SERIALIZE(SEND), |
| FAILED_SERIALIZE(SEND), |
| FINISH_SERIALIZE_LARGE(SEND), |
| SEND_FRAME(SEND), |
| FAILED_FRAME(SEND), |
| SENT_FRAME(SEND), |
| ARRIVE(RECEIVE), |
| FAILED_EXPIRED_ON_RECEIVE(RECEIVE), |
| DESERIALIZE(RECEIVE), |
| CLOSED_BEFORE_ARRIVAL(RECEIVE), |
| FAILED_DESERIALIZE(RECEIVE), |
| PROCESS(RECEIVE), |
| |
| FAILED_CLOSING(SEND), |
| |
| CONNECT_OUTBOUND(OTHER), |
| SYNC(OTHER), // the connection will stop sending messages, and promptly process any waiting inbound messages |
| CONTROLLER_UPDATE(OTHER); |
| |
| final EventCategory category; |
| |
| EventType(EventCategory category) |
| { |
| this.category = category; |
| } |
| } |
| |
| public static class Event |
| { |
| final EventType type; |
| Event(EventType type) |
| { |
| this.type = type; |
| } |
| } |
| |
| static class SimpleEvent extends Event |
| { |
| final long at; |
| SimpleEvent(EventType type, long at) |
| { |
| super(type); |
| this.at = at; |
| } |
| public String toString() { return type.toString(); } |
| } |
| |
| static class BoundedEvent extends Event |
| { |
| final long start; |
| volatile long end; |
| BoundedEvent(EventType type, long start) |
| { |
| super(type); |
| this.start = start; |
| } |
| public void complete(Verifier verifier) |
| { |
| end = verifier.sequenceId.getAndIncrement(); |
| verifier.events.put(end, this); |
| } |
| } |
| |
| static class SimpleMessageEvent extends SimpleEvent |
| { |
| final long messageId; |
| SimpleMessageEvent(EventType type, long at, long messageId) |
| { |
| super(type, at); |
| this.messageId = messageId; |
| } |
| } |
| |
| static class BoundedMessageEvent extends BoundedEvent |
| { |
| final long messageId; |
| BoundedMessageEvent(EventType type, long start, long messageId) |
| { |
| super(type, start); |
| this.messageId = messageId; |
| } |
| } |
| |
| static class EnqueueMessageEvent extends BoundedMessageEvent |
| { |
| final Message<?> message; |
| final Destiny destiny; |
| EnqueueMessageEvent(EventType type, long start, Message<?> message, Destiny destiny) |
| { |
| super(type, start, message.id()); |
| this.message = message; |
| this.destiny = destiny; |
| } |
| public String toString() { return String.format("%s{%s}", type, destiny); } |
| } |
| |
| static class SerializeMessageEvent extends SimpleMessageEvent |
| { |
| final int messagingVersion; |
| SerializeMessageEvent(EventType type, long at, long messageId, int messagingVersion) |
| { |
| super(type, at, messageId); |
| this.messagingVersion = messagingVersion; |
| } |
| public String toString() { return String.format("%s{ver=%d}", type, messagingVersion); } |
| } |
| |
| static class SimpleMessageEventWithSize extends SimpleMessageEvent |
| { |
| final int messageSize; |
| SimpleMessageEventWithSize(EventType type, long at, long messageId, int messageSize) |
| { |
| super(type, at, messageId); |
| this.messageSize = messageSize; |
| } |
| public String toString() { return String.format("%s{size=%d}", type, messageSize); } |
| } |
| |
| static class FailedSerializeEvent extends SimpleMessageEvent |
| { |
| final int bytesWrittenToNetwork; |
| final Throwable failure; |
| FailedSerializeEvent(long at, long messageId, int bytesWrittenToNetwork, Throwable failure) |
| { |
| super(FAILED_SERIALIZE, at, messageId); |
| this.bytesWrittenToNetwork = bytesWrittenToNetwork; |
| this.failure = failure; |
| } |
| public String toString() { return String.format("FAILED_SERIALIZE{written=%d, failure=%s}", bytesWrittenToNetwork, failure); } |
| } |
| |
| static class ExpiredMessageEvent extends SimpleMessageEvent |
| { |
| enum ExpirationType {ON_SENT, ON_ARRIVED, ON_PROCESSED } |
| final int messageSize; |
| final long timeElapsed; |
| final TimeUnit timeUnit; |
| final ExpirationType expirationType; |
| ExpiredMessageEvent(long at, long messageId, int messageSize, long timeElapsed, TimeUnit timeUnit, ExpirationType expirationType) |
| { |
| super(expirationType == ON_SENT ? FAILED_EXPIRED_ON_SEND : FAILED_EXPIRED_ON_RECEIVE, at, messageId); |
| this.messageSize = messageSize; |
| this.timeElapsed = timeElapsed; |
| this.timeUnit = timeUnit; |
| this.expirationType = expirationType; |
| } |
| public String toString() { return String.format("EXPIRED_%s{size=%d,elapsed=%d,unit=%s}", expirationType, messageSize, timeElapsed, timeUnit); } |
| } |
| |
| static class FrameEvent extends SimpleEvent |
| { |
| final int messageCount; |
| final int payloadSizeInBytes; |
| FrameEvent(EventType type, long at, int messageCount, int payloadSizeInBytes) |
| { |
| super(type, at); |
| this.messageCount = messageCount; |
| this.payloadSizeInBytes = payloadSizeInBytes; |
| } |
| } |
| |
| static class ProcessMessageEvent extends SimpleMessageEvent |
| { |
| final Message<?> message; |
| ProcessMessageEvent(long at, Message<?> message) |
| { |
| super(PROCESS, at, message.id()); |
| this.message = message; |
| } |
| } |
| |
| EnqueueMessageEvent onEnqueue(Message<?> message, Destiny destiny) |
| { |
| EnqueueMessageEvent enqueue = new EnqueueMessageEvent(ENQUEUE, nextId(), message, destiny); |
| events.put(enqueue.start, enqueue); |
| return enqueue; |
| } |
| void onOverloaded(long messageId) |
| { |
| long at = nextId(); |
| events.put(at, new SimpleMessageEvent(FAILED_OVERLOADED, at, messageId)); |
| } |
| void onFailedClosing(long messageId) |
| { |
| long at = nextId(); |
| events.put(at, new SimpleMessageEvent(FAILED_CLOSING, at, messageId)); |
| } |
| void onSerialize(long messageId, int messagingVersion) |
| { |
| long at = nextId(); |
| events.put(at, new SerializeMessageEvent(SERIALIZE, at, messageId, messagingVersion)); |
| } |
| void onFinishSerializeLarge(long messageId) |
| { |
| long at = nextId(); |
| events.put(at, new SimpleMessageEvent(FINISH_SERIALIZE_LARGE, at, messageId)); |
| } |
| void onFailedSerialize(long messageId, int bytesWrittenToNetwork, Throwable failure) |
| { |
| long at = nextId(); |
| events.put(at, new FailedSerializeEvent(at, messageId, bytesWrittenToNetwork, failure)); |
| } |
| void onExpiredBeforeSend(long messageId, int messageSize, long timeElapsed, TimeUnit timeUnit) |
| { |
| onExpired(messageId, messageSize, timeElapsed, timeUnit, ON_SENT); |
| } |
| void onSendFrame(int messageCount, int payloadSizeInBytes) |
| { |
| long at = nextId(); |
| events.put(at, new FrameEvent(SEND_FRAME, at, messageCount, payloadSizeInBytes)); |
| } |
| void onSentFrame(int messageCount, int payloadSizeInBytes) |
| { |
| long at = nextId(); |
| events.put(at, new FrameEvent(SENT_FRAME, at, messageCount, payloadSizeInBytes)); |
| } |
| void onFailedFrame(int messageCount, int payloadSizeInBytes) |
| { |
| long at = nextId(); |
| events.put(at, new FrameEvent(FAILED_FRAME, at, messageCount, payloadSizeInBytes)); |
| } |
| void onArrived(long messageId, int messageSize) |
| { |
| long at = nextId(); |
| events.put(at, new SimpleMessageEventWithSize(ARRIVE, at, messageId, messageSize)); |
| } |
| void onArrivedExpired(long messageId, int messageSize, boolean wasCorrupt, long timeElapsed, TimeUnit timeUnit) |
| { |
| onExpired(messageId, messageSize, timeElapsed, timeUnit, ExpirationType.ON_ARRIVED); |
| } |
| void onDeserialize(long messageId, int messagingVersion) |
| { |
| long at = nextId(); |
| events.put(at, new SerializeMessageEvent(DESERIALIZE, at, messageId, messagingVersion)); |
| } |
| void onClosedBeforeArrival(long messageId, int messageSize) |
| { |
| long at = nextId(); |
| events.put(at, new SimpleMessageEventWithSize(CLOSED_BEFORE_ARRIVAL, at, messageId, messageSize)); |
| } |
| void onFailedDeserialize(long messageId, int messageSize) |
| { |
| long at = nextId(); |
| events.put(at, new SimpleMessageEventWithSize(FAILED_DESERIALIZE, at, messageId, messageSize)); |
| } |
| void process(Message<?> message) |
| { |
| long at = nextId(); |
| events.put(at, new ProcessMessageEvent(at, message)); |
| } |
| void onProcessExpired(long messageId, int messageSize, long timeElapsed, TimeUnit timeUnit) |
| { |
| onExpired(messageId, messageSize, timeElapsed, timeUnit, ExpirationType.ON_PROCESSED); |
| } |
| private void onExpired(long messageId, int messageSize, long timeElapsed, TimeUnit timeUnit, ExpirationType expirationType) |
| { |
| long at = nextId(); |
| events.put(at, new ExpiredMessageEvent(at, messageId, messageSize, timeElapsed, timeUnit, expirationType)); |
| } |
| |
| |
| |
| static class ConnectOutboundEvent extends SimpleEvent |
| { |
| final int messagingVersion; |
| final OutboundConnectionSettings settings; |
| ConnectOutboundEvent(long at, int messagingVersion, OutboundConnectionSettings settings) |
| { |
| super(EventType.CONNECT_OUTBOUND, at); |
| this.messagingVersion = messagingVersion; |
| this.settings = settings; |
| } |
| } |
| |
| // TODO: do we need this? |
| static class ConnectInboundEvent extends SimpleEvent |
| { |
| final int messagingVersion; |
| final InboundMessageHandler handler; |
| ConnectInboundEvent(long at, int messagingVersion, InboundMessageHandler handler) |
| { |
| super(EventType.CONNECT_OUTBOUND, at); |
| this.messagingVersion = messagingVersion; |
| this.handler = handler; |
| } |
| } |
| |
| static class SyncEvent extends SimpleEvent |
| { |
| final Runnable onCompletion; |
| SyncEvent(long at, Runnable onCompletion) |
| { |
| super(EventType.SYNC, at); |
| this.onCompletion = onCompletion; |
| } |
| } |
| |
| static class ControllerEvent extends BoundedEvent |
| { |
| final long minimumBytesInFlight; |
| final long maximumBytesInFlight; |
| ControllerEvent(long start, long minimumBytesInFlight, long maximumBytesInFlight) |
| { |
| super(EventType.CONTROLLER_UPDATE, start); |
| this.minimumBytesInFlight = minimumBytesInFlight; |
| this.maximumBytesInFlight = maximumBytesInFlight; |
| } |
| } |
| |
| void onSync(Runnable onCompletion) |
| { |
| SyncEvent connect = new SyncEvent(nextId(), onCompletion); |
| events.put(connect.at, connect); |
| } |
| |
| void onConnectOutbound(int messagingVersion, OutboundConnectionSettings settings) |
| { |
| ConnectOutboundEvent connect = new ConnectOutboundEvent(nextId(), messagingVersion, settings); |
| events.put(connect.at, connect); |
| } |
| |
| void onConnectInbound(int messagingVersion, InboundMessageHandler handler) |
| { |
| ConnectInboundEvent connect = new ConnectInboundEvent(nextId(), messagingVersion, handler); |
| events.put(connect.at, connect); |
| } |
| |
| private final BytesInFlightController controller; |
| private final AtomicLong sequenceId = new AtomicLong(); |
| private final EventSequence events = new EventSequence(); |
| private final InboundMessageHandlers inbound; |
| private final OutboundConnection outbound; |
| |
| Verifier(BytesInFlightController controller, OutboundConnection outbound, InboundMessageHandlers inbound) |
| { |
| this.controller = controller; |
| this.inbound = inbound; |
| this.outbound = outbound; |
| } |
| |
| private long nextId() |
| { |
| return sequenceId.getAndIncrement(); |
| } |
| |
| public void logFailure(String message, Object ... params) |
| { |
| fail(message, params); |
| } |
| |
| private void fail(String message, Object ... params) |
| { |
| logger.error("{}", String.format(message, params)); |
| logger.error("Connection: {}", currentConnection); |
| } |
| |
| private void fail(String message, Throwable t, Object ... params) |
| { |
| logger.error("{}", String.format(message, params), t); |
| logger.error("Connection: {}", currentConnection); |
| } |
| |
| private void failinfo(String message, Object ... params) |
| { |
| logger.error("{}", String.format(message, params)); |
| } |
| |
| private static class MessageState |
| { |
| final Message<?> message; |
| final Destiny destiny; |
| int messagingVersion; |
| // set initially to message.expiresAtNanos, but if at serialization time we use |
| // an older messaging version we may not be able to serialize expiration |
| long expiresAtNanos; |
| long enqueueStart, enqueueEnd, serialize, arrive, deserialize; |
| boolean processOnEventLoop, processOutOfOrder; |
| Event sendState, receiveState; |
| long lastUpdateAt; |
| long lastUpdateNanos; |
| ConnectionState sentOn; |
| boolean doneSend, doneReceive; |
| |
| int messageSize() |
| { |
| return message.serializedSize(messagingVersion); |
| } |
| |
| MessageState(Message<?> message, Destiny destiny, long enqueueStart) |
| { |
| this.message = message; |
| this.destiny = destiny; |
| this.enqueueStart = enqueueStart; |
| this.expiresAtNanos = message.expiresAtNanos(); |
| } |
| |
| void update(SimpleEvent event, long now) |
| { |
| update(event, event.at, now); |
| } |
| void update(Event event, long at, long now) |
| { |
| lastUpdateAt = at; |
| lastUpdateNanos = now; |
| switch (event.type.category) |
| { |
| case SEND: |
| sendState = event; |
| break; |
| case RECEIVE: |
| receiveState = event; |
| break; |
| default: throw new IllegalStateException(); |
| } |
| } |
| |
| boolean is(EventType type) |
| { |
| switch (type.category) |
| { |
| case SEND: return sendState != null && sendState.type == type; |
| case RECEIVE: return receiveState != null && receiveState.type == type; |
| default: return false; |
| } |
| } |
| |
| boolean is(EventType type1, EventType type2) |
| { |
| return is(type1) || is(type2); |
| } |
| |
| boolean is(EventType type1, EventType type2, EventType type3) |
| { |
| return is(type1) || is(type2) || is(type3); |
| } |
| |
| void require(EventType event, Verifier verifier, EventType type) |
| { |
| if (!is(type)) |
| verifier.fail("Invalid state at %s for %s: expected %s", event, this, type); |
| } |
| |
| void require(EventType event, Verifier verifier, EventType type1, EventType type2) |
| { |
| if (!is(type1) && !is(type2)) |
| verifier.fail("Invalid state at %s for %s: expected %s or %s", event, this, type1, type2); |
| } |
| |
| void require(EventType event, Verifier verifier, EventType type1, EventType type2, EventType type3) |
| { |
| if (!is(type1) && !is(type2) && !is(type3)) |
| verifier.fail("Invalid state %s for %s: expected %s, %s or %s", event, this, type1, type2, type3); |
| } |
| |
| public String toString() |
| { |
| return String.format("{id:%d, state:[%s,%s], upd:%d, ver:%d, enqueue:[%d,%d], ser:%d, arr:%d, deser:%d, expires:%d, sentOn: %d}", |
| message.id(), sendState, receiveState, lastUpdateAt, messagingVersion, enqueueStart, enqueueEnd, serialize, arrive, deserialize, approxTime.translate().toMillisSinceEpoch(expiresAtNanos), sentOn == null ? -1 : sentOn.connectionId); |
| } |
| } |
| |
| private final LongObjectHashMap<MessageState> messages = new LongObjectHashMap<>(); |
| |
| // messages start here, but may enter in a haphazard (non-sequential) fashion; |
| // ENQUEUE_START, ENQUEUE_END both take place here, with the latter imposing bounds on the out-of-order appearance of messages. |
| // note that ENQUEUE_END - being concurrent - may not appear before the message's lifespan has completely ended. |
| private final Queue<MessageState> enqueueing = new Queue<>(); |
| |
| static final class ConnectionState |
| { |
| final long connectionId; |
| final int messagingVersion; |
| // Strict message order will then be determined at serialization time, since this happens on a single thread. |
| // The order in which messages arrive here determines the order they will arrive on the other node. |
| // must follow either ENQUEUE_START or ENQUEUE_END |
| final Queue<MessageState> serializing = new Queue<>(); |
| |
| // Messages sent on the small connection will all be sent in frames; this is a concurrent operation, |
| // so only the sendingFrame MUST be encountered before any future events - |
| // large connections skip this step and goes straight to arriving |
| // we consult the queues in reverse order in arriving, as it is acceptable to find our frame in any of these queues |
| final FramesInFlight framesInFlight = new FramesInFlight(); // unknown if the messages will arrive, accept either |
| |
| // for large messages OR < VERSION_40, arriving can occur BEFORE serializing completes successfully |
| // OR a frame is fully serialized |
| final Queue<MessageState> arriving = new Queue<>(); |
| |
| final Queue<MessageState> deserializingOnEventLoop = new Queue<>(), |
| deserializingOffEventLoop = new Queue<>(); |
| |
| InboundMessageHandler inbound; |
| |
| // TODO |
| long sentCount, sentBytes; |
| long receivedCount, receivedBytes; |
| |
| ConnectionState(long connectionId, int messagingVersion) |
| { |
| this.connectionId = connectionId; |
| this.messagingVersion = messagingVersion; |
| } |
| |
| public String toString() |
| { |
| return String.format("{id: %d, ver: %d, ser: %d, inFlight: %s, arriving: %d, deserOn: %d, deserOff: %d}", |
| connectionId, messagingVersion, serializing.size(), framesInFlight, arriving.size(), deserializingOnEventLoop.size(), deserializingOffEventLoop.size()); |
| } |
| } |
| |
| private final Queue<MessageState> processingOutOfOrder = new Queue<>(); |
| |
| private SyncEvent sync; |
| private long nextMessageId = 0; |
| private long now; |
| private long connectionCounter; |
| private ConnectionState currentConnection = new ConnectionState(connectionCounter++, current_version); |
| |
| private long outboundSentCount, outboundSentBytes; |
| private long outboundSubmittedCount; |
| private long outboundOverloadedCount, outboundOverloadedBytes; |
| private long outboundExpiredCount, outboundExpiredBytes; |
| private long outboundErrorCount, outboundErrorBytes; |
| |
| public void run(Runnable onFailure, long deadlineNanos) |
| { |
| try |
| { |
| long lastEventAt = approxTime.now(); |
| while ((now = approxTime.now()) < deadlineNanos) |
| { |
| Event next = events.await(nextMessageId, 100L, MILLISECONDS); |
| if (next == null) |
| { |
| // decide if we have any messages waiting too long to proceed |
| while (!processingOutOfOrder.isEmpty()) |
| { |
| MessageState m = processingOutOfOrder.get(0); |
| if (now - m.lastUpdateNanos > SECONDS.toNanos(10L)) |
| { |
| fail("Unreasonably long period spent waiting for out-of-order deser/delivery of received message %d", m.message.id()); |
| MessageState v = maybeRemove(m.message.id(), PROCESS); |
| controller.fail(v.message.serializedSize(v.messagingVersion == 0 ? current_version : v.messagingVersion)); |
| processingOutOfOrder.remove(0); |
| } |
| else break; |
| } |
| |
| if (sync != null) |
| { |
| // if we have waited 100ms since beginning a sync, with no events, and ANY of our queues are |
| // non-empty, something is probably wrong; however, let's give ourselves a little bit longer |
| |
| boolean done = |
| currentConnection.serializing.isEmpty() |
| && currentConnection.arriving.isEmpty() |
| && currentConnection.deserializingOnEventLoop.isEmpty() |
| && currentConnection.deserializingOffEventLoop.isEmpty() |
| && currentConnection.framesInFlight.isEmpty() |
| && enqueueing.isEmpty() |
| && processingOutOfOrder.isEmpty() |
| && messages.isEmpty() |
| && controller.inFlight() == 0; |
| |
| //outbound.pendingCount() > 0 ? 5L : 2L |
| if (!done && now - lastEventAt > SECONDS.toNanos(5L)) |
| { |
| // TODO: even 2s or 5s are unreasonable periods of time without _any_ movement on a message waiting to arrive |
| // this seems to happen regularly on MacOS, but we should confirm this does not happen on Linux |
| fail("Unreasonably long period spent waiting for sync (%dms)", NANOSECONDS.toMillis(now - lastEventAt)); |
| messages.<LongObjectProcedure<MessageState>>forEach((k, v) -> { |
| failinfo("%s", v); |
| controller.fail(v.message.serializedSize(v.messagingVersion == 0 ? current_version : v.messagingVersion)); |
| }); |
| currentConnection.serializing.clear(); |
| currentConnection.arriving.clear(); |
| currentConnection.deserializingOnEventLoop.clear(); |
| currentConnection.deserializingOffEventLoop.clear(); |
| enqueueing.clear(); |
| processingOutOfOrder.clear(); |
| messages.clear(); |
| while (!currentConnection.framesInFlight.isEmpty()) |
| currentConnection.framesInFlight.poll(); |
| done = true; |
| } |
| |
| if (done) |
| { |
| ConnectionUtils.check(outbound) |
| .pending(0, 0) |
| .error(outboundErrorCount, outboundErrorBytes) |
| .submitted(outboundSubmittedCount) |
| .expired(outboundExpiredCount, outboundExpiredBytes) |
| .overload(outboundOverloadedCount, outboundOverloadedBytes) |
| .sent(outboundSentCount, outboundSentBytes) |
| .check((message, expect, actual) -> fail("%s: expect %d, actual %d", message, expect, actual)); |
| |
| sync.onCompletion.run(); |
| sync = null; |
| } |
| } |
| continue; |
| } |
| events.clear(nextMessageId); // TODO: simplify collection if we end up using it exclusively as a queue, as we are now |
| lastEventAt = now; |
| |
| switch (next.type) |
| { |
| case ENQUEUE: |
| { |
| MessageState m; |
| EnqueueMessageEvent e = (EnqueueMessageEvent) next; |
| assert nextMessageId == e.start || nextMessageId == e.end; |
| assert e.message != null; |
| if (nextMessageId == e.start) |
| { |
| if (sync != null) |
| fail("Sync in progress - there should be no messages beginning to enqueue"); |
| |
| m = new MessageState(e.message, e.destiny, e.start); |
| messages.put(e.messageId, m); |
| enqueueing.add(m); |
| m.update(e, e.start, now); |
| } |
| else |
| { |
| // warning: enqueueEnd can occur at any time in the future, since it's a different thread; |
| // it could be arbitrarily paused, long enough even for the messsage to be fully processed |
| m = messages.get(e.messageId); |
| if (m != null) |
| m.enqueueEnd = e.end; |
| outboundSubmittedCount += 1; |
| } |
| break; |
| } |
| case FAILED_OVERLOADED: |
| { |
| // TODO: verify that we could have exceeded our memory limits |
| SimpleMessageEvent e = (SimpleMessageEvent) next; |
| assert nextMessageId == e.at; |
| MessageState m = remove(e.messageId, enqueueing, messages); |
| m.require(FAILED_OVERLOADED, this, ENQUEUE); |
| outboundOverloadedBytes += m.message.serializedSize(current_version); |
| outboundOverloadedCount += 1; |
| break; |
| } |
| case FAILED_CLOSING: |
| { |
| // TODO: verify if this is acceptable due to e.g. inbound refusing to process for long enough |
| SimpleMessageEvent e = (SimpleMessageEvent) next; |
| assert nextMessageId == e.at; |
| MessageState m = messages.remove(e.messageId); // definitely cannot have been sent (in theory) |
| enqueueing.remove(m); |
| m.require(FAILED_CLOSING, this, ENQUEUE); |
| fail("Invalid discard of %d: connection was closing for too long", m.message.id()); |
| break; |
| } |
| case SERIALIZE: |
| { |
| // serialize happens serially, so we can compress the asynchronicity of the above enqueue |
| // into a linear sequence of events we expect to occur on arrival |
| SerializeMessageEvent e = (SerializeMessageEvent) next; |
| assert nextMessageId == e.at; |
| MessageState m = get(e); |
| assert m.is(ENQUEUE); |
| m.serialize = e.at; |
| m.messagingVersion = e.messagingVersion; |
| if (current_version != e.messagingVersion) |
| controller.adjust(m.message.serializedSize(current_version), m.message.serializedSize(e.messagingVersion)); |
| |
| m.processOnEventLoop = willProcessOnEventLoop(outbound.type(), m.message, e.messagingVersion); |
| m.expiresAtNanos = expiresAtNanos(m.message, e.messagingVersion); |
| int mi = enqueueing.indexOf(m); |
| for (int i = 0 ; i < mi ; ++i) |
| { |
| MessageState pm = enqueueing.get(i); |
| if (pm.enqueueEnd != 0 && pm.enqueueEnd < m.enqueueStart) |
| { |
| fail("Invalid order of events: %s enqueued strictly before %s, but serialized after", |
| pm, m); |
| } |
| } |
| enqueueing.remove(mi); |
| m.sentOn = currentConnection; |
| currentConnection.serializing.add(m); |
| m.update(e, now); |
| break; |
| } |
| case FINISH_SERIALIZE_LARGE: |
| { |
| // serialize happens serially, so we can compress the asynchronicity of the above enqueue |
| // into a linear sequence of events we expect to occur on arrival |
| SimpleMessageEvent e = (SimpleMessageEvent) next; |
| assert nextMessageId == e.at; |
| MessageState m = maybeRemove(e); |
| outboundSentBytes += m.messageSize(); |
| outboundSentCount += 1; |
| m.sentOn.serializing.remove(m); |
| m.update(e, now); |
| break; |
| } |
| case FAILED_SERIALIZE: |
| { |
| FailedSerializeEvent e = (FailedSerializeEvent) next; |
| assert nextMessageId == e.at; |
| MessageState m = maybeRemove(e); |
| |
| if (outbound.type() == LARGE_MESSAGES) |
| assert e.failure instanceof InvalidSerializedSizeException || e.failure instanceof Connection.IntentionalIOException || e.failure instanceof Connection.IntentionalRuntimeException; |
| else |
| assert e.failure instanceof InvalidSerializedSizeException || e.failure instanceof Connection.IntentionalIOException || e.failure instanceof Connection.IntentionalRuntimeException || e.failure instanceof BufferOverflowException; |
| |
| if (e.bytesWrittenToNetwork == 0) // TODO: use header size |
| messages.remove(m.message.id()); |
| |
| InvalidSerializedSizeException ex; |
| if (outbound.type() != LARGE_MESSAGES |
| || !(e.failure instanceof InvalidSerializedSizeException) |
| || ((ex = (InvalidSerializedSizeException) e.failure).expectedSize <= DEFAULT_BUFFER_SIZE && ex.actualSizeAtLeast <= DEFAULT_BUFFER_SIZE) |
| || (ex.expectedSize > DEFAULT_BUFFER_SIZE && ex.actualSizeAtLeast < DEFAULT_BUFFER_SIZE)) |
| { |
| assert e.bytesWrittenToNetwork == 0; |
| } |
| |
| m.require(FAILED_SERIALIZE, this, SERIALIZE); |
| m.sentOn.serializing.remove(m); |
| if (m.destiny != Destiny.FAIL_TO_SERIALIZE) |
| fail("%s failed to serialize, but its destiny was to %s", m, m.destiny); |
| outboundErrorBytes += m.messageSize(); |
| outboundErrorCount += 1; |
| m.update(e, now); |
| break; |
| } |
| case SEND_FRAME: |
| { |
| FrameEvent e = (FrameEvent) next; |
| assert nextMessageId == e.at; |
| int size = 0; |
| Frame frame = new Frame(); |
| MessageState first = currentConnection.serializing.get(0); |
| int messagingVersion = first.messagingVersion; |
| for (int i = 0 ; i < e.messageCount ; ++i) |
| { |
| MessageState m = currentConnection.serializing.get(i); |
| size += m.message.serializedSize(m.messagingVersion); |
| if (m.messagingVersion != messagingVersion) |
| { |
| fail("Invalid sequence of events: %s encoded to same frame as %s", |
| m, first); |
| } |
| |
| frame.add(m); |
| m.update(e, now); |
| assert !m.doneSend; |
| m.doneSend = true; |
| if (m.doneReceive) |
| messages.remove(m.message.id()); |
| } |
| frame.payloadSizeInBytes = e.payloadSizeInBytes; |
| frame.messageCount = e.messageCount; |
| frame.messagingVersion = messagingVersion; |
| currentConnection.framesInFlight.add(frame); |
| currentConnection.serializing.removeFirst(e.messageCount); |
| if (e.payloadSizeInBytes != size) |
| fail("Invalid frame payload size with %s: expected %d, actual %d", first, size, e.payloadSizeInBytes); |
| break; |
| } |
| case SENT_FRAME: |
| { |
| Frame frame = currentConnection.framesInFlight.supplySendStatus(Frame.Status.SUCCESS); |
| frame.forEach(m -> m.update((SimpleEvent) next, now)); |
| |
| outboundSentBytes += frame.payloadSizeInBytes; |
| outboundSentCount += frame.messageCount; |
| break; |
| } |
| case FAILED_FRAME: |
| { |
| // TODO: is it possible for this to be signalled AFTER our reconnect event? probably, in which case this will be wrong |
| // TODO: verify that this was expected |
| Frame frame = currentConnection.framesInFlight.supplySendStatus(Frame.Status.FAILED); |
| frame.forEach(m -> m.update((SimpleEvent) next, now)); |
| if (frame.messagingVersion >= VERSION_40) |
| { |
| // the contents cannot be delivered without the whole frame arriving, so clear the contents now |
| clear(frame, messages); |
| currentConnection.framesInFlight.remove(frame); |
| } |
| outboundErrorBytes += frame.payloadSizeInBytes; |
| outboundErrorCount += frame.messageCount; |
| break; |
| } |
| case ARRIVE: |
| { |
| SimpleMessageEventWithSize e = (SimpleMessageEventWithSize) next; |
| assert nextMessageId == e.at; |
| MessageState m = get(e); |
| |
| m.arrive = e.at; |
| if (e.messageSize != m.messageSize()) |
| fail("onArrived with invalid size for %s: %d vs %d", m, e.messageSize, m.messageSize()); |
| |
| if (outbound.type() == LARGE_MESSAGES) |
| { |
| m.require(ARRIVE, this, SERIALIZE, FAILED_SERIALIZE, FINISH_SERIALIZE_LARGE); |
| } |
| else |
| { |
| if (!m.is(SEND_FRAME, SENT_FRAME)) |
| { |
| fail("Invalid order of events: %s arrived before being sent in a frame", m); |
| break; |
| } |
| |
| int fi = -1, mi = -1; |
| while (fi + 1 < m.sentOn.framesInFlight.size() && mi < 0) |
| mi = m.sentOn.framesInFlight.get(++fi).indexOf(m); |
| |
| if (fi == m.sentOn.framesInFlight.size()) |
| { |
| fail("Invalid state: %s, but no frame in flight was found to contain it", m); |
| break; |
| } |
| |
| if (fi > 0) |
| { |
| // we have skipped over some frames, meaning these have either failed (and we know it) |
| // or we have not yet heard about them and they have presumably failed, or something |
| // has gone wrong |
| fail("BEGIN: Successfully sent frames were not delivered"); |
| for (int i = 0 ; i < fi ; ++i) |
| { |
| Frame skip = m.sentOn.framesInFlight.get(i); |
| skip.receiveStatus = Frame.Status.FAILED; |
| if (skip.sendStatus == Frame.Status.SUCCESS) |
| { |
| failinfo("Frame %s", skip); |
| for (int j = 0 ; j < skip.size() ; ++j) |
| failinfo("Containing: %s", skip.get(j)); |
| } |
| clear(skip, messages); |
| } |
| m.sentOn.framesInFlight.removeFirst(fi); |
| failinfo("END: Successfully sent frames were not delivered"); |
| } |
| |
| Frame frame = m.sentOn.framesInFlight.get(0); |
| for (int i = 0; i < mi; ++i) |
| fail("Invalid order of events: %s serialized strictly before %s, but arrived after", frame.get(i), m); |
| |
| frame.remove(mi); |
| if (frame.isEmpty()) |
| m.sentOn.framesInFlight.poll(); |
| } |
| m.sentOn.arriving.add(m); |
| m.update(e, now); |
| break; |
| } |
| case DESERIALIZE: |
| { |
| // deserialize may happen in parallel for large messages, but in sequence for small messages |
| // we currently require that this event be issued before any possible error is thrown |
| SimpleMessageEvent e = (SimpleMessageEvent) next; |
| assert nextMessageId == e.at; |
| MessageState m = get(e); |
| m.require(DESERIALIZE, this, ARRIVE); |
| m.deserialize = e.at; |
| // deserialize may be off-loaded, so we can only impose meaningful ordering constraints |
| // on those messages we know to have been processed on the event loop |
| int mi = m.sentOn.arriving.indexOf(m); |
| if (m.processOnEventLoop) |
| { |
| for (int i = 0 ; i < mi ; ++i) |
| { |
| MessageState pm = m.sentOn.arriving.get(i); |
| if (pm.processOnEventLoop) |
| { |
| fail("Invalid order of events: %d (%d, %d) arrived strictly before %d (%d, %d), but deserialized after", |
| pm.message.id(), pm.arrive, pm.deserialize, m.message.id(), m.arrive, m.deserialize); |
| } |
| } |
| m.sentOn.deserializingOnEventLoop.add(m); |
| } |
| else |
| { |
| m.sentOn.deserializingOffEventLoop.add(m); |
| } |
| m.sentOn.arriving.remove(mi); |
| m.update(e, now); |
| break; |
| } |
| case CLOSED_BEFORE_ARRIVAL: |
| { |
| SimpleMessageEventWithSize e = (SimpleMessageEventWithSize) next; |
| assert nextMessageId == e.at; |
| MessageState m = maybeRemove(e); |
| |
| if (e.messageSize != m.messageSize()) |
| fail("onClosedBeforeArrival has invalid size for %s: %d vs %d", m, e.messageSize, m.messageSize()); |
| |
| m.sentOn.deserializingOffEventLoop.remove(m); |
| if (m.destiny == Destiny.FAIL_TO_SERIALIZE && outbound.type() == LARGE_MESSAGES) |
| break; |
| fail("%s closed before arrival, but its destiny was to %s", m, m.destiny); |
| break; |
| } |
| case FAILED_DESERIALIZE: |
| { |
| SimpleMessageEventWithSize e = (SimpleMessageEventWithSize) next; |
| assert nextMessageId == e.at; |
| MessageState m = maybeRemove(e); |
| |
| if (e.messageSize != m.messageSize()) |
| fail("onFailedDeserialize has invalid size for %s: %d vs %d", m, e.messageSize, m.messageSize()); |
| m.require(FAILED_DESERIALIZE, this, ARRIVE, DESERIALIZE); |
| (m.processOnEventLoop ? m.sentOn.deserializingOnEventLoop : m.sentOn.deserializingOffEventLoop).remove(m); |
| switch (m.destiny) |
| { |
| case FAIL_TO_DESERIALIZE: |
| break; |
| case FAIL_TO_SERIALIZE: |
| if (outbound.type() == LARGE_MESSAGES) |
| break; |
| default: |
| fail("%s failed to deserialize, but its destiny was to %s", m, m.destiny); |
| } |
| break; |
| } |
| case PROCESS: |
| { |
| ProcessMessageEvent e = (ProcessMessageEvent) next; |
| assert nextMessageId == e.at; |
| MessageState m = maybeRemove(e); |
| |
| m.require(PROCESS, this, DESERIALIZE); |
| if (!Arrays.equals((byte[]) e.message.payload, (byte[]) m.message.payload)) |
| { |
| fail("Invalid message payload for %d: %s supplied by processor, but %s implied by original message and messaging version", |
| e.messageId, Arrays.toString((byte[]) e.message.payload), Arrays.toString((byte[]) m.message.payload)); |
| } |
| |
| if (m.processOutOfOrder) |
| { |
| assert !m.processOnEventLoop; // will have already been reported small (processOnEventLoop) messages |
| processingOutOfOrder.remove(m); |
| } |
| else if (m.processOnEventLoop) |
| { |
| // we can expect that processing happens sequentially in this case, more specifically |
| // we can actually expect that this event will occur _immediately_ after the deserialize event |
| // so that we have exactly one mess |
| // c |
| int mi = m.sentOn.deserializingOnEventLoop.indexOf(m); |
| for (int i = 0 ; i < mi ; ++i) |
| { |
| MessageState pm = m.sentOn.deserializingOnEventLoop.get(i); |
| fail("Invalid order of events: %s deserialized strictly before %s, but processed after", |
| pm, m); |
| } |
| clearFirst(mi, m.sentOn.deserializingOnEventLoop, messages); |
| m.sentOn.deserializingOnEventLoop.poll(); |
| } |
| else |
| { |
| int mi = m.sentOn.deserializingOffEventLoop.indexOf(m); |
| // process may be off-loaded, so we can only impose meaningful ordering constraints |
| // on those messages we know to have been processed on the event loop |
| for (int i = 0 ; i < mi ; ++i) |
| { |
| MessageState pm = m.sentOn.deserializingOffEventLoop.get(i); |
| pm.processOutOfOrder = true; |
| processingOutOfOrder.add(pm); |
| } |
| m.sentOn.deserializingOffEventLoop.removeFirst(mi + 1); |
| } |
| // this message has been fully validated |
| break; |
| } |
| case FAILED_EXPIRED_ON_SEND: |
| case FAILED_EXPIRED_ON_RECEIVE: |
| { |
| ExpiredMessageEvent e = (ExpiredMessageEvent) next; |
| assert nextMessageId == e.at; |
| MessageState m; |
| switch (e.expirationType) |
| { |
| case ON_SENT: |
| { |
| m = messages.remove(e.messageId); |
| m.require(e.type, this, ENQUEUE); |
| outboundExpiredBytes += m.message.serializedSize(current_version); |
| outboundExpiredCount += 1; |
| messages.remove(m.message.id()); |
| break; |
| } |
| case ON_ARRIVED: |
| m = maybeRemove(e); |
| if (!m.is(ARRIVE)) |
| { |
| if (outbound.type() != LARGE_MESSAGES) m.require(e.type, this, SEND_FRAME, SENT_FRAME, FAILED_FRAME); |
| else m.require(e.type, this, SERIALIZE, FAILED_SERIALIZE, FINISH_SERIALIZE_LARGE); |
| } |
| break; |
| case ON_PROCESSED: |
| m = maybeRemove(e); |
| m.require(e.type, this, DESERIALIZE); |
| break; |
| default: |
| throw new IllegalStateException(); |
| } |
| |
| now = System.nanoTime(); |
| if (m.expiresAtNanos > now) |
| { |
| // we fix the conversion AlmostSameTime for an entire run, which should suffice to guarantee these comparisons |
| fail("Invalid expiry of %d: expiry should occur in %dms; event believes %dms have elapsed, and %dms have actually elapsed", m.message.id(), |
| NANOSECONDS.toMillis(m.expiresAtNanos - m.message.createdAtNanos()), |
| e.timeUnit.toMillis(e.timeElapsed), |
| NANOSECONDS.toMillis(now - m.message.createdAtNanos())); |
| } |
| |
| switch (e.expirationType) |
| { |
| case ON_SENT: |
| enqueueing.remove(m); |
| break; |
| case ON_ARRIVED: |
| if (m.is(ARRIVE)) |
| m.sentOn.arriving.remove(m); |
| switch (m.sendState.type) |
| { |
| case SEND_FRAME: |
| case SENT_FRAME: |
| case FAILED_FRAME: |
| // TODO: this should be robust to re-ordering; should perhaps extract a common method |
| m.sentOn.framesInFlight.get(0).remove(m); |
| if (m.sentOn.framesInFlight.get(0).isEmpty()) |
| m.sentOn.framesInFlight.poll(); |
| break; |
| } |
| break; |
| case ON_PROCESSED: |
| (m.processOnEventLoop ? m.sentOn.deserializingOnEventLoop : m.sentOn.deserializingOffEventLoop).remove(m); |
| break; |
| } |
| |
| if (m.messagingVersion != 0 && e.messageSize != m.messageSize()) |
| fail("onExpired %s with invalid size for %s: %d vs %d", e.expirationType, m, e.messageSize, m.messageSize()); |
| |
| break; |
| } |
| case CONTROLLER_UPDATE: |
| { |
| break; |
| } |
| case CONNECT_OUTBOUND: |
| { |
| ConnectOutboundEvent e = (ConnectOutboundEvent) next; |
| currentConnection = new ConnectionState(connectionCounter++, e.messagingVersion); |
| break; |
| } |
| case SYNC: |
| { |
| sync = (SyncEvent) next; |
| break; |
| } |
| default: |
| throw new IllegalStateException(); |
| } |
| ++nextMessageId; |
| } |
| } |
| catch (InterruptedException e) |
| { |
| } |
| catch (Throwable t) |
| { |
| logger.error("Unexpected error:", t); |
| onFailure.run(); |
| } |
| } |
| |
| private MessageState get(SimpleMessageEvent onEvent) |
| { |
| MessageState m = messages.get(onEvent.messageId); |
| if (m == null) |
| throw new IllegalStateException("Missing " + onEvent + ": " + onEvent.messageId); |
| return m; |
| } |
| private MessageState maybeRemove(SimpleMessageEvent onEvent) |
| { |
| return maybeRemove(onEvent.messageId, onEvent.type, onEvent); |
| } |
| private MessageState maybeRemove(long messageId, EventType onEvent) |
| { |
| return maybeRemove(messageId, onEvent, onEvent); |
| } |
| private MessageState maybeRemove(long messageId, EventType onEvent, Object id) |
| { |
| MessageState m = messages.get(messageId); |
| if (m == null) |
| throw new IllegalStateException("Missing " + id + ": " + messageId); |
| switch (onEvent.category) |
| { |
| case SEND: |
| if (m.doneSend) |
| fail("%s already doneSend %s", onEvent, m); |
| m.doneSend = true; |
| if (m.doneReceive) messages.remove(messageId); |
| break; |
| case RECEIVE: |
| if (m.doneReceive) |
| fail("%s already doneReceive %s", onEvent, m); |
| m.doneReceive = true; |
| if (m.doneSend) messages.remove(messageId); |
| } |
| return m; |
| } |
| |
| |
| private static class Frame extends Queue<MessageState> |
| { |
| enum Status { SUCCESS, FAILED, UNKNOWN } |
| Status sendStatus = Status.UNKNOWN, receiveStatus = Status.UNKNOWN; |
| int messagingVersion; |
| int messageCount; |
| int payloadSizeInBytes; |
| |
| public String toString() |
| { |
| return String.format("{count:%d, size:%d, version:%d, send:%s, receive:%s}", |
| messageCount, payloadSizeInBytes, messagingVersion, sendStatus, receiveStatus); |
| } |
| } |
| |
| private static MessageState remove(long messageId, Queue<MessageState> queue, LongObjectHashMap<MessageState> lookup) |
| { |
| MessageState m = lookup.remove(messageId); |
| queue.remove(m); |
| return m; |
| } |
| |
| private static void clearFirst(int count, Queue<MessageState> queue, LongObjectHashMap<MessageState> lookup) |
| { |
| if (count > 0) |
| { |
| for (int i = 0 ; i < count ; ++i) |
| lookup.remove(queue.get(i).message.id()); |
| queue.removeFirst(count); |
| } |
| } |
| |
| private static void clear(Queue<MessageState> queue, LongObjectHashMap<MessageState> lookup) |
| { |
| if (!queue.isEmpty()) |
| clearFirst(queue.size(), queue, lookup); |
| } |
| |
| private static class EventSequence |
| { |
| static final int CHUNK_SIZE = 1 << 10; |
| static class Chunk extends AtomicReferenceArray<Event> |
| { |
| final long sequenceId; |
| int removed = 0; |
| Chunk(long sequenceId) |
| { |
| super(CHUNK_SIZE); |
| this.sequenceId = sequenceId; |
| } |
| Event get(long sequenceId) |
| { |
| return get((int)(sequenceId - this.sequenceId)); |
| } |
| void set(long sequenceId, Event event) |
| { |
| lazySet((int)(sequenceId - this.sequenceId), event); |
| } |
| } |
| |
| // we use a concurrent skip list to permit efficient searching, even if we always append |
| final ConcurrentSkipListMap<Long, Chunk> chunkList = new ConcurrentSkipListMap<>(); |
| final WaitQueue writerWaiting = new WaitQueue(); |
| |
| volatile Chunk writerChunk = new Chunk(0); |
| Chunk readerChunk = writerChunk; |
| |
| long readerWaitingFor; |
| volatile Thread readerWaiting; |
| |
| EventSequence() |
| { |
| chunkList.put(0L, writerChunk); |
| } |
| |
| public void put(long sequenceId, Event event) |
| { |
| long chunkSequenceId = sequenceId & -CHUNK_SIZE; |
| Chunk chunk = writerChunk; |
| if (chunk.sequenceId != chunkSequenceId) |
| { |
| try |
| { |
| chunk = ensureChunk(chunkSequenceId); |
| } |
| catch (InterruptedException e) |
| { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| chunk.set(sequenceId, event); |
| |
| Thread wake = readerWaiting; |
| long wakeIf = readerWaitingFor; // we are guarded by the above volatile read |
| if (wake != null && wakeIf == sequenceId) |
| LockSupport.unpark(wake); |
| } |
| |
| Chunk ensureChunk(long chunkSequenceId) throws InterruptedException |
| { |
| Chunk chunk = chunkList.get(chunkSequenceId); |
| if (chunk == null) |
| { |
| Map.Entry<Long, Chunk> e; |
| while ( null != (e = chunkList.firstEntry()) && chunkSequenceId - e.getKey() > 1 << 12) |
| { |
| WaitQueue.Signal signal = writerWaiting.register(); |
| if (null != (e = chunkList.firstEntry()) && chunkSequenceId - e.getKey() > 1 << 12) |
| signal.await(); |
| else |
| signal.cancel(); |
| } |
| chunk = chunkList.get(chunkSequenceId); |
| if (chunk == null) |
| { |
| synchronized (this) |
| { |
| chunk = chunkList.get(chunkSequenceId); |
| if (chunk == null) |
| chunkList.put(chunkSequenceId, chunk = new Chunk(chunkSequenceId)); |
| } |
| } |
| } |
| return chunk; |
| } |
| |
| Chunk readerChunk(long readerId) throws InterruptedException |
| { |
| long chunkSequenceId = readerId & -CHUNK_SIZE; |
| if (readerChunk.sequenceId != chunkSequenceId) |
| readerChunk = ensureChunk(chunkSequenceId); |
| return readerChunk; |
| } |
| |
| public Event await(long id, long timeout, TimeUnit unit) throws InterruptedException |
| { |
| return await(id, System.nanoTime() + unit.toNanos(timeout)); |
| } |
| |
| public Event await(long id, long deadlineNanos) throws InterruptedException |
| { |
| Chunk chunk = readerChunk(id); |
| Event result = chunk.get(id); |
| if (result != null) |
| return result; |
| |
| readerWaitingFor = id; |
| readerWaiting = Thread.currentThread(); |
| while (null == (result = chunk.get(id))) |
| { |
| long waitNanos = deadlineNanos - System.nanoTime(); |
| if (waitNanos <= 0) |
| return null; |
| LockSupport.parkNanos(waitNanos); |
| if (Thread.interrupted()) |
| throw new InterruptedException(); |
| } |
| readerWaitingFor = -1; |
| readerWaiting = null; |
| return result; |
| } |
| |
| public Event find(long sequenceId) |
| { |
| long chunkSequenceId = sequenceId & -CHUNK_SIZE; |
| Chunk chunk = readerChunk; |
| if (chunk.sequenceId != chunkSequenceId) |
| { |
| chunk = writerChunk; |
| if (chunk.sequenceId != chunkSequenceId) |
| chunk = chunkList.get(chunkSequenceId); |
| } |
| return chunk.get(sequenceId); |
| } |
| |
| public void clear(long sequenceId) |
| { |
| long chunkSequenceId = sequenceId & -CHUNK_SIZE; |
| Chunk chunk = chunkList.get(chunkSequenceId); |
| chunk.set(sequenceId, null); |
| if (++chunk.removed == CHUNK_SIZE) |
| { |
| chunkList.remove(chunkSequenceId); |
| writerWaiting.signalAll(); |
| } |
| } |
| } |
| |
| static class Queue<T> |
| { |
| private Object[] items = new Object[10]; |
| private int begin, end; |
| |
| int size() |
| { |
| return end - begin; |
| } |
| |
| T get(int i) |
| { |
| //noinspection unchecked |
| return (T) items[i + begin]; |
| } |
| |
| int indexOf(T item) |
| { |
| for (int i = begin ; i < end ; ++i) |
| { |
| if (item == items[i]) |
| return i - begin; |
| } |
| return -1; |
| } |
| |
| void remove(T item) |
| { |
| int i = indexOf(item); |
| if (i >= 0) |
| remove(i); |
| } |
| |
| void remove(int i) |
| { |
| i += begin; |
| assert i < end; |
| |
| if (i == begin || i + 1 == end) |
| { |
| items[i] = null; |
| if (begin + 1 == end) begin = end = 0; |
| else if (i == begin) ++begin; |
| else --end; |
| } |
| else if (i - begin < end - i) |
| { |
| System.arraycopy(items, begin, items, begin + 1, i - begin); |
| items[begin++] = null; |
| } |
| else |
| { |
| System.arraycopy(items, i + 1, items, i, (end - 1) - i); |
| items[--end] = null; |
| } |
| } |
| |
| void add(T item) |
| { |
| if (end == items.length) |
| { |
| Object[] src = items; |
| Object[] trg; |
| if (end - begin < src.length / 2) trg = src; |
| else trg = new Object[src.length * 2]; |
| System.arraycopy(src, begin, trg, 0, end - begin); |
| end -= begin; |
| begin = 0; |
| items = trg; |
| } |
| items[end++] = item; |
| } |
| |
| void clear() |
| { |
| Arrays.fill(items, begin, end, null); |
| begin = end = 0; |
| } |
| |
| void removeFirst(int count) |
| { |
| Arrays.fill(items, begin, begin + count, null); |
| begin += count; |
| if (begin == end) |
| begin = end = 0; |
| } |
| |
| T poll() |
| { |
| if (begin == end) |
| return null; |
| //noinspection unchecked |
| T result = (T) items[begin]; |
| items[begin++] = null; |
| if (begin == end) |
| begin = end = 0; |
| return result; |
| } |
| |
| void forEach(Consumer<T> consumer) |
| { |
| for (int i = 0 ; i < size() ; ++i) |
| consumer.accept(get(i)); |
| } |
| |
| boolean isEmpty() |
| { |
| return begin == end; |
| } |
| |
| public String toString() |
| { |
| StringBuilder result = new StringBuilder(); |
| result.append('['); |
| toString(result); |
| result.append(']'); |
| return result.toString(); |
| } |
| |
| void toString(StringBuilder out) |
| { |
| for (int i = 0 ; i < size() ; ++i) |
| { |
| if (i > 0) out.append(", "); |
| out.append(get(i)); |
| } |
| } |
| } |
| |
| |
| |
| static class FramesInFlight |
| { |
| // this may be negative, indicating we have processed a frame whose status we did not know at the time |
| // TODO: we should verify the status of these frames by logging the inferred status and verifying it matches |
| final Queue<Frame> inFlight = new Queue<>(); |
| final Queue<Frame> retiredWithoutStatus = new Queue<>(); |
| private int withStatus; |
| |
| Frame supplySendStatus(Frame.Status status) |
| { |
| Frame frame; |
| if (withStatus >= 0) frame = inFlight.get(withStatus); |
| else frame = retiredWithoutStatus.poll(); |
| assert frame.sendStatus == Frame.Status.UNKNOWN; |
| frame.sendStatus = status; |
| ++withStatus; |
| return frame; |
| } |
| |
| boolean isEmpty() |
| { |
| return inFlight.isEmpty(); |
| } |
| |
| int size() |
| { |
| return inFlight.size(); |
| } |
| |
| Frame get(int i) |
| { |
| return inFlight.get(i); |
| } |
| |
| void add(Frame frame) |
| { |
| assert frame.sendStatus == Frame.Status.UNKNOWN; |
| inFlight.add(frame); |
| } |
| |
| void remove(Frame frame) |
| { |
| int i = inFlight.indexOf(frame); |
| if (i > 0) throw new IllegalStateException(); |
| if (i == 0) poll(); |
| } |
| |
| void removeFirst(int count) |
| { |
| while (count-- > 0) |
| poll(); |
| } |
| |
| Frame poll() |
| { |
| Frame frame = inFlight.poll(); |
| if (--withStatus < 0) |
| { |
| assert frame.sendStatus == Frame.Status.UNKNOWN; |
| retiredWithoutStatus.add(frame); |
| } |
| else |
| assert frame.sendStatus != Frame.Status.UNKNOWN; |
| return frame; |
| } |
| |
| public String toString() |
| { |
| StringBuilder result = new StringBuilder(); |
| result.append("[withStatus="); |
| result.append(withStatus); |
| result.append("; "); |
| inFlight.toString(result); |
| result.append("; "); |
| retiredWithoutStatus.toString(result); |
| result.append(']'); |
| return result.toString(); |
| } |
| } |
| |
| private static boolean willProcessOnEventLoop(ConnectionType type, Message<?> message, int messagingVersion) |
| { |
| int size = message.serializedSize(messagingVersion); |
| if (type == ConnectionType.SMALL_MESSAGES && messagingVersion >= VERSION_40) |
| return size <= LARGE_MESSAGE_THRESHOLD; |
| else if (messagingVersion >= VERSION_40) |
| return size <= DEFAULT_BUFFER_SIZE; |
| else |
| return size <= LARGE_MESSAGE_THRESHOLD; |
| } |
| |
| private static long expiresAtNanos(Message<?> message, int messagingVersion) |
| { |
| return messagingVersion < VERSION_40 ? message.verb().expiresAtNanos(message.createdAtNanos()) |
| : message.expiresAtNanos(); |
| } |
| |
| } |
