Deterministic expectable matching in receiver hub (instead of multi-thread...

	private void write(Behaviour b, Map<DataUse, String> dataUseVariables, List<FutureInfo> futures,

			Set<String> thrownExceptions) {

		lineComment(b.eClass().getName());

		writeComment(b);

		writeNotification(b, true);

		if (b instanceof CompoundBehaviour) {

			// XX why was this here?

//			write(((CompoundBehaviour) b).getBlock(), null, null, thrownExceptions);

				ComponentInstance c = ((TimerOperation) b).getComponentInstance();

				if (c != null && !isCurrentComponentInstance(c))

					return;

			} else if (b instanceof TimeOperation) {

				ComponentInstance c = ((TimeOperation) b).getComponentInstance();

				if (c != null && !isCurrentComponentInstance(c))

					return;

			} else if (b instanceof Assignment) {

				VariableUse v = ((Assignment) b).getVariable();

				if (!isCurrentComponentInstance(v.getComponentInstance()))

					return;

			} else if (b instanceof Message) {

				if (!isTesterInput(b) && !isCurrentComponentInstance(((Message)b).getSourceGate().getComponent()))

					return;

			}

		}

		lineComment(b.eClass().getName());

		writeComment(b);

		writeNotification(b, true);

		// Set up periodic threads and exceptional behaviours BEFORE the try block so the

		// variables they declare (Thread, Throwable[], ExceptionalBehaviour) are in scope

		// in the finally block below.

					FutureInfo futureInfo = writeTesterInput(b, dataUseVariables);

					myFutures.add(futureInfo);

				} else {

				} else if (isCurrentComponentInstance(((Message)b).getSourceGate().getComponent())) {

					Message m = (Message) b;

					DataUse arg = m.getArgument();

				blockClose();

				append("catch (" + FUTURE_EXECUTION_EXCEPTION + " | " + INTERRUPTED_EXCEPTION + " e) ");

				blockOpen();

				line("if (e.getCause() instanceof AssertionError) throw (AssertionError) e.getCause();");

				line("if (e.getCause() instanceof RuntimeException) throw (RuntimeException) e.getCause();");

				line("throw new RuntimeException(e.getCause());");

				blockClose();

		return null;

	}

	@Override

	public void logUnexpected(Data received, Data expected, Connection connection, NamedElement target) {

		reporter.comment("SystemAdapter | " + target.getName(), "UNEXPECTED: " + describe(received));

		if (expected != null)

			reporter.comment("SystemAdapter | " + target.getName(), "\t vs " + describe(expected));

	}

	private String describe(Data data) {

		Object value = data.getValue();

		Object head;

		Object body;

		if (value instanceof HttpRequestData) {

			head = ((HttpRequestData) value).method;

			body = ((HttpRequestData) value).body;

		} else if (value instanceof HttpResponseData) {

			head = ((HttpResponseData) value).status;

			body = ((HttpResponseData) value).body;

		} else {

			return String.valueOf(value);

		}

		String bodyText = null;

		if (body != null)

			try {

				// An unmatched receive leaves the body as the raw string,

				// matched data contains decoded objects that need encoding for display.

				bodyText = body instanceof String ? (String) body

						: StandardCharsets.UTF_8.decode(ByteBuffer.wrap(encodeBody(body))).toString();

			} catch (IOException e) {

				// Rendering for a log message is best effort.

			}

		return head + " | " + bodyText;

	}

	protected byte[] encodeBody(Object body) throws IOException {

		return mapper.writeValueAsBytes(body);

	}

	private List<Expectable> expecting = Collections.synchronizedList(new ArrayList<>());

	private ExceptionalBehaviour anyReceiver;

	// Per-thread batch state: each thread's entry is {batchSize, batchCount}.

	// Expectations from threads with incomplete batches are skipped by run().

	private Map<Thread, int[]> batches = new HashMap<>();

	/**

	 * 	 The expectable run() is currently blocked on inside systemAdapter.receive (null

	 * 	 when it is not blocked). run() selects an expectable, then blocks OUTSIDE the

	 * 	 `expecting` lock waiting for a message; if that expectable's round is retired in

	 * 	 the meantime (disableBatch), the blocking call must be aborted so run() does not

	 * 	 return the NEXT round's message against a now-stale selection. disableBatch reads

	 * 	 this to decide whether to interrupt the hub thread. Volatile: written by the hub

	 * 	 thread, read by the retiring round's thread.

	 */

	private volatile Expectable blocked;

	/**

	 * 	 Rounds (parallel blocks) currently admitted to this hub, in admission order.

	 * 	 The relative order of independent blocks is not significant. Guarded by the

	 * 	 {@code expecting} monitor.

	 */

	private final List<AlternativeRound> roundOrder = new ArrayList<>();

	/**

	 * 	 Per-round count of expectables still to be registered before that round may be

	 * 	 matched. Set by enableBatch() (which knows the full per-hub round size) and

	 * 	 counted down by receive() as each callable registers its expectable on its own

	 * 	 pool thread. In addition, no anyReceiver may

	 * 	 fire until EVERY entry here is 0 (all queued blocks fully registered), so a block

	 * 	 still wiring up its receives cannot lose its message to a premature "unexpected".

	 */

	private final Map<AlternativeRound, Integer> pendingRegistrations = new HashMap<>();

	/**

	 * Shared per-round exclusivity token. Every expectable submitted as part of a

		this.anyReceiver = anyReceiver;

	}

	public void enableBatch(int size) {

	public void enableBatch(AlternativeRound round, int size) {

		synchronized (expecting) {

			batches.put(Thread.currentThread(), new int[]{size, 0});

			// Admit this round to the hub with its expected registration count. Called

			// once per hub per round, before the round's callables are submitted.

			roundOrder.add(round);

			pendingRegistrations.put(round, size);

		}

	}

	public void disableBatch() {

	public void disableBatch(AlternativeRound round) {

		synchronized (expecting) {

			batches.remove(Thread.currentThread());

			// Retire this round from the hub once its behaviour has completed.

			pendingRegistrations.remove(round);

			roundOrder.remove(round);

			expecting.notifyAll();

		}

		// If run() is currently blocked in systemAdapter.receive on an expectable of the

		// round being retired (e.g. this round's anyReceiver on a connection it never

		// actually reads), abort that blocking call so run() re-selects instead of

		// snatching the next round's message against the stale, now-retired expectable.

		Expectable b = blocked;

		if (b != null && b.round == round)

			thread.interrupt();

	}

	/**

		}

	}

	private synchronized void run() {

	private void run() {

		// Track tried expectables by identity rather than positional index,

		// since the expecting list is mutated concurrently by other threads.

		// Stale entries are harmless — each hub.receive() call creates a new

			synchronized (expecting) {

				currentlyExpecting = null;

				while (currentlyExpecting == null) {

						for (Expectable e : expecting)

							if (!tried.contains(e)

									&& !(e.round != null && e.round.resolved)

									&& !hasPendingBatch(e.ownerThread)) {

						// Match each round's expectables as an isolated group. The odering imposed by precedes()

						//

						// Eligibility: skip already-tried expectables and those of a resolved

						// round. A round's expectables are gated until that round is fully

						// registered (its pending count is 0), so its priority set is complete.

						// 

						// An anyReceiver is additionally held until EVERY queued round is fully

						// registered, so a block still setting up its receives cannot lose its

						// message to a premature "unexpected". Ungated expectables (round == null,

						// the direct receive path) are always eligible.

						boolean allRegistered = true;

						for (int p : pendingRegistrations.values())

							if (p != 0) { allRegistered = false; break; }

						for (Expectable e : expecting) {

							if (tried.contains(e))

								continue;

							if (e.round != null && e.round.resolved)

								continue;

							if (e.round != null && pendingRegistrations.getOrDefault(e.round, -1) != 0)

								continue;

							if (e.anyReceiver && !allRegistered)

								continue;

							if (currentlyExpecting == null || precedes(e, currentlyExpecting))

								currentlyExpecting = e;

								break;

						}

					if (currentlyExpecting == null)

						try {

							expecting.wait();

						} catch (InterruptedException e) {

							// Only a stop() interrupt should end the loop; an interrupt used

							// merely to abort a stale blocking receive (see disableBatch) must

							// just re-evaluate.

							if (stopped)

								return;

						}

				}

				// Publish the selection while still holding the expecting monitor. This

				// orders it against disableBatch's retirement (same monitor): either the

				// round was retired first (the selection above already skipped its

				// expectables) or disableBatch runs after this write and its read of

				// "blocked" sees this expectable and interrupts. No unprotected gap.

				blocked = currentlyExpecting;

			}

			try {

				Data data = systemAdapter.receive(currentlyExpecting.expected, connection, testerComponent);

				if (data != null) {

				if (data == null) {

					// A null return means the adapter had an input for this hub but it

					// did not match this expectable's template (a blocking adapter waits

					// instead of returning null when nothing has arrived).

					if (currentlyExpecting.ignoreUnmatched) {

						// The trigger discards the unmatched head message and matching re-evaluates from the top

						// Lower-priority expectables, defaults and the anyReceiver never see the discarded message.

						tried.clear();

					} else {

						// Mark it tried so the loop advances to the next expectable

						tried.add(currentlyExpecting);

					}

				} else {

					tried.clear();

					if (currentlyExpecting.anyReceiver) {

						// Switch to the first one for reporting

						// Report the unexpected message through the adapter against the

						// highest-priority expected data that was being awaited (if any).

						Data awaited = null;

						synchronized (expecting) {

							Expectable report = null;

							for (Expectable e: expecting)

								if (!e.ignoreUnmatched) {

									currentlyExpecting = e;

									break;

								if (e.round == currentlyExpecting.round && !e.ignoreUnmatched && !e.anyReceiver

										&& (report == null || e.priority < report.priority))

									report = e;

							if (report != null)

								awaited = report.expected;

						}

						systemAdapter.logUnexpected(data, awaited, connection, testerComponent);

						if (anyReceiver != null) {

							anyReceiver.behaviour = () -> {

								throw new StopExceptionImpl(

										"Unexpected message received on " + connection.getName());

							};

						}

					}

					expecting.remove(currentlyExpecting);

				}

			} catch (InterruptedException e) {

			} catch (AssertionError e) {

				tried.add(currentlyExpecting);

				if (!currentlyExpecting.ignoreUnmatched)

					currentlyExpecting.error = e;

			} catch (RuntimeException e) {

			} catch (RuntimeException | AssertionError e) {

				currentlyExpecting.error = e;

				expecting.remove(currentlyExpecting);

				if (currentlyExpecting.round != null)

				synchronized (currentlyExpecting) {

					currentlyExpecting.notifyAll();

				}

			} finally {

				blocked = null;

			}

		}

	}

	}

	public Data receive(Data expected, boolean ignoreUntil) {

		return receive(expected, ignoreUntil, null);

		return receive(expected, ignoreUntil, null, 0);

	}

	public Data receive(Data expected, boolean ignoreUntil, AlternativeRound round) {

		Expectable expectable = new Expectable(expected, ignoreUntil, round);

		return receive(expected, ignoreUntil, round, 0);

	}

	public Data receive(Data expected, boolean ignoreUntil, AlternativeRound round, int priority) {

		Expectable expectable = new Expectable(expected, ignoreUntil, round, priority);

		synchronized (expecting) {

			expecting.add(expectable);

			int[] batch = batches.get(Thread.currentThread());

			if (batch != null) {

				batch[1]++;

				if (batch[1] >= batch[0]) {

					batches.remove(Thread.currentThread());

			// Count this registration against its round's batch. Ungated registrations (no round,

			// or a round with no active batch) wake immediately.

			Integer remaining = round == null ? null : pendingRegistrations.get(round);

			if (remaining != null && remaining > 0) {

				remaining--;

				pendingRegistrations.put(round, remaining);

				if (remaining == 0)

					expecting.notifyAll();

				}

			} else {

				expecting.notifyAll();

			}

		return null;

	}

	/** Must be called while holding the expecting lock. */

	private boolean hasPendingBatch(Thread thread) {

		return batches.containsKey(thread);

	/**

	 * Deterministic match ordering. Returns true if {@code a} should be tried before

	 * {@code b}: real receives before anyReceivers, real receives by round

	 * admission order (never interleaved with another round's) then by within-block priority. 

	 * Must be called while holding the {@code expecting} monitor (reads {@code roundOrder}).

	 */

	private boolean precedes(Expectable a, Expectable b) {

		int aAny = a.anyReceiver ? 1 : 0;

		int bAny = b.anyReceiver ? 1 : 0;

		if (aAny != bAny)

			return aAny < bAny;

		int aBlock = a.round == null ? -1 : roundOrder.indexOf(a.round);

		int bBlock = b.round == null ? -1 : roundOrder.indexOf(b.round);

		if (aBlock != bBlock)

			return aBlock < bBlock;

		return a.priority < b.priority;

	}

	class Expectable {

		Throwable error;

		boolean ignoreUnmatched = false;

		boolean anyReceiver = false;

		Thread ownerThread = Thread.currentThread();

		// Deterministic match priority; lower value wins.

		int priority;

		AlternativeRound round;

		public Expectable(Data expected, boolean ignoreUnmatched, AlternativeRound round) {

		public Expectable(Data expected, boolean ignoreUnmatched, AlternativeRound round, int priority) {

			this.expected = expected;

			this.ignoreUnmatched = ignoreUnmatched;

			this.anyReceiver = expected == null;

			this.round = round;

			this.priority = priority;

		}

	}

}

				@Override

				public ExecutionResult call() throws Exception {

					try {

						Data data = hub.receive(null, false, this.round);

						Data data = hub.receive(null, false, this.round, this.priority);

						return data != null ? new InteractionResult(data) : null;

					} catch (RuntimeException e) {

						// Propagate SA errors through the behaviour execution path:

						// to rethrow the real exception.

						anyReceiver.behaviour = () -> { throw e; };

						return null;

					} catch (AssertionError e) {

						// Same for adapter-reported test failures.

						anyReceiver.behaviour = () -> { throw e; };

						return null;

					}

				}

		 */

		public volatile ReceiverHub.AlternativeRound round;

		/**

		 * Deterministic match priority for this callable's expectable within its round

		 * (lower value = higher priority). Assigned by

		 * {@link #executeAlternativesAndExceptionals(List)} on the main thread, in the

		 * normative submission order, and passed on to the {@link ReceiverHub.Expectable}

		 * so {@link ReceiverHub} selects the winner by this value rather than by the

		 * order pool threads happen to register expectables in. Volatile: written by the

		 * submitting thread, read by the pool thread running {@link #call()}.

		 */

		public volatile int priority;

		public Future<ExecutionResult> execute() {

			return completionService.get().submit(this);

		}

		public final List<Future<ExecutionResult>> alternatives;

		public final List<Future<ExecutionResult>> exceptionals;

		public final Collection<ReceiverHub> batchedHubs;

		public final ReceiverHub.AlternativeRound round;

		public BatchedFutures(List<Future<ExecutionResult>> alternatives,

				List<Future<ExecutionResult>> exceptionals,

				Collection<ReceiverHub> batchedHubs) {

				Collection<ReceiverHub> batchedHubs,

				ReceiverHub.AlternativeRound round) {

			this.alternatives = alternatives;

			this.exceptionals = exceptionals;

			this.batchedHubs = batchedHubs;

			this.round = round;

		}

	}

			hubCounts.merge(hub, 1, Integer::sum);

		}

		// Enable batches BEFORE submitting any callables

		hubCounts.forEach((hub, count) -> hub.enableBatch(count));

		// One exclusivity round shared by every expectable submitted here: the

		// alternatives, the active exceptionals and the anyReceivers. When any of

		// them matches, the round is resolved and the rest are disabled at once.

		// The round is assigned BEFORE execute() so it is visible to the pool

		// thread that creates the Expectable (and before batch gating releases it).

		// It also keys the per-hub FCFS batch admission below.

		ReceiverHub.AlternativeRound round = new ReceiverHub.AlternativeRound();

		// Admit this round's batch on every involved hub BEFORE submitting any

		// callables, so each hub knows how many registrations to await for this round

		// before it may start matching. Hubs serve rounds first-come, first-served.

		hubCounts.forEach((hub, count) -> hub.enableBatch(round, count));

		// Assign an explicit, deterministic priority to every expectable of this round

		// in submission order (lower value = higher priority = matched first). This is

		// the normative order, the per-hub anyReceivers last.

		int priority = 0;

		// Submit alternatives

		List<Future<ExecutionResult>> altFutures = new ArrayList<>();

		for (ExecutionCallable c : alternatives) {

			c.round = round;

			c.priority = priority++;

			altFutures.add(c.execute());

		}

		List<Future<ExecutionResult>> excFutures = new ArrayList<>();

		for (ExceptionalBehaviour exc : enabledExcs) {

			exc.callable.round = round;

			exc.callable.priority = priority++;

			excFutures.add(exc.execute());

		}

		receiverHubs.values().forEach(hub -> {

			hub.getAnyReceiver().callable.round = round;

			excFutures.add(hub.getAnyReceiver().execute());

		});

		for (ReceiverHub hub : receiverHubs.values()) {

			ExceptionalBehaviour any = hub.getAnyReceiver();

			any.callable.round = round;

			any.callable.priority = priority++;

			excFutures.add(any.execute());

		}

		return new BatchedFutures(altFutures, excFutures, hubCounts.keySet());

		return new BatchedFutures(altFutures, excFutures, hubCounts.keySet(), round);

	}

	/**

	 * {@code notifyAll} on hubs that were never batched.

	 */

	public void cleanupBatch(BatchedFutures batch) {

		batch.batchedHubs.forEach(ReceiverHub::disableBatch);

		batch.batchedHubs.forEach(hub -> hub.disableBatch(batch.round));

	}

	public void resumeReceiving() {

	public ExecutionCallable noInput(long period, GateReference gate) {

		// XXX noInput

		return sleep(period);

		return sleep(period * 1000);

	}

	public ExecutionCallable receive(Data expected, Connection connection) {

			}

			@Override

			public ExecutionResult call() throws Exception {

				Data data = getReceiver().receive(expected, isTrigger, this.round);

				Data data = getReceiver().receive(expected, isTrigger, this.round, this.priority);

				return data != null ? new InteractionResult(data) : null;

			}

		};

	 * received data chunk using provided type and match against provided value.

	 * <p>

	 * Implementation should only consider one (the oldest) chunk of incoming data

	 * at a time. If the data matches then it is decoded and returned, otherwise the

	 * method throws <b>AssertionError</b>.

	 * at a time. If the data matches then it is decoded, removed from

	 * the receive queue and returned. If it does not match, the method returns

	 * <b>null</b> and the data remains queued.

	 * <p>

	 * Special <b>null</b> value is used for <code>expected</code> parameter when

	 * the adapter should accept any data and return it (potentially without

	 * @return The data in expected format or null if most recent data didn't match

	 *         the expected one.

	 * @throws InterruptedException If the waiting thread is interrupted.

	 * @throws AssertionError       If the received data didn't match the expected

	 *                              data.

	 * @throws AssertionError       In case of protocol constraint violations,

	 *                              encoding or other related errors.

	 */

	Data receive(Data expected, Connection connection, NamedElement target) throws InterruptedException, AssertionError;

	/**

	 * Logs an unexpected message in an adapter specific format.

	 *

	 * @param received   The (potentially encoded) data, as returned by the {@link #receive(Data, Connection, NamedElement)}.

	 * @param awaited    An expected value, or null if none.

	 * @param connection The connection on which the message was received.

	 * @param target     The component that received the message.

	 */

	void logUnexpected(Data received, Data expected, Connection connection, NamedElement target);

	/**

	 * Calls a remote procedure and blocks until reply is received or exception is

	 * thrown.