diff --git a/core/src/main/java/com/softwaremill/jox/Channel.java b/core/src/main/java/com/softwaremill/jox/Channel.java index 956f6d4..fd4eb52 100644 --- a/core/src/main/java/com/softwaremill/jox/Channel.java +++ b/core/src/main/java/com/softwaremill/jox/Channel.java @@ -6,6 +6,7 @@ import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.LockSupport; +import java.util.function.Function; import java.util.stream.Stream; import static com.softwaremill.jox.CellState.*; @@ -121,7 +122,9 @@ public Channel() { * Capacity cannot be negative. */ public Channel(int capacity) { - assert capacity >= 0 : "Capacity must be non-negative."; + if (capacity < 0) { + throw new IllegalArgumentException("Capacity must be non-negative."); + } this.capacity = capacity; isRendezvous = capacity == 0L; @@ -277,7 +280,18 @@ private SendResult updateCellSend(Segment segment, int i, long s, T value) throw segment.setCell(i, DONE); return SendResult.RESUMED; } else { - // cell interrupted -> trying with a new one + // when cell interrupted -> trying with a new one + // when close in progress -> subsequent cells are already closed, this will be detected in the next iteration + return SendResult.FAILED; + } + } + case StoredSelect ss -> { + // a select clause is waiting -> trying to resume + if (ss.getSelect().trySelect(ss, value)) { + segment.setCell(i, DONE); + return SendResult.RESUMED; + } else { + // select unsuccessful -> trying with a new one return SendResult.FAILED; } } @@ -318,6 +332,14 @@ public T receive() throws InterruptedException { * @return Either a value of type {@code T}, or {@link ChannelClosed}, when the channel is closed. */ public Object receiveSafe() throws InterruptedException { + return doReceive(null, null); + } + + /** + * @return If {@code select} & {@code selectClause} is {@code null}: the received value, or {@link ChannelClosed}, + * when the channel is closed. Otherwise, might also return {@link StoredSelect}. + */ + private Object doReceive(SelectInstance select, SelectClause selectClause) throws InterruptedException { while (true) { // reading the segment before the counter increment - this is needed to find the required segment later var segment = receiveSegment.get(); @@ -344,23 +366,27 @@ public Object receiveSafe() throws InterruptedException { } } - var result = updateCellReceive(segment, i, r); + var result = updateCellReceive(segment, i, r, select, selectClause); if (result == ReceiveResult.CLOSED) { // not cleaning the previous segments - the close procedure might still need it return closedReason.get(); } else { /* - After `updateCellReceive` completes and the channel isn't closed, we can be sure that S > r: + After `updateCellReceive` completes and the channel isn't closed, we can be sure that S > r, unless + we stored the given select instance: - if we stored and awaited a continuation, and it was resumed, then a sender must have appeared - if we marked the cell as broken, then a sender is in progress in that cell - if a continuation was present, then the sender must have been there - if the cell was interrupted, that could have been only because of a sender - if a value was buffered, that's because there was/is a matching sender - The only case when S < r is when awaiting on the continuation is interrupted, in which case the - exception propagates outside of this method. + The only cases when S <= r are when: + - awaiting on the continuation is interrupted, in which case the exception propagates outside of this method + - we stored the given select instance (in an empty / in-buffer cell) */ - segment.cleanPrev(); + if (!(result instanceof StoredSelect)) { + segment.cleanPrev(); + } if (result != ReceiveResult.FAILED) { return result; } @@ -371,13 +397,16 @@ public Object receiveSafe() throws InterruptedException { /** * Invariant maintained by receive + expandBuffer: between R and B the number of cells that are empty / IN_BUFFER should be equal * to the buffer size. These are the cells that can accept a sender without suspension. + *

+ * This method might suspend (and be interrupted) only if {@code select} is {@code null}. * * @param segment The segment which stores the cell's state. * @param i The index within the {@code segment}. * @param r Index of the reserved cell. - * @return Either a state-result ({@link ReceiveResult}), or the received value. + * @param select The select instance of which this receive is part of, or {@code null} (along with {@code selectClause}) if this is a direct receive call. + * @return Either a state-result ({@link ReceiveResult}), {@link StoredSelect} in case {@code select} is not {@code null}, or the received value. */ - private Object updateCellReceive(Segment segment, int i, long r) throws InterruptedException { + private Object updateCellReceive(Segment segment, int i, long r, SelectInstance select, SelectClause selectClause) throws InterruptedException { while (true) { var state = segment.getCell(i); // reading the current state of the cell; we'll try to update it atomically var switchState = state == null ? IN_BUFFER : state; // we can't combine null+IN_BUFFER in the switch statement, hence cheating a bit here @@ -385,19 +414,30 @@ private Object updateCellReceive(Segment segment, int i, long r) throws Interrup switch (switchState) { case IN_BUFFER -> { // means that state == null || state == IN_BUFFER if (r >= getSendersCounter(sendersAndClosedFlag.get())) { // reading the sender's counter - // cell is empty, and no sender -> suspend - // not using any payload - var c = new Continuation(null); - if (segment.casCell(i, state, c)) { - expandBuffer(); - var result = c.await(segment, i); - if (result == ChannelClosedMarker.CLOSED) { - return ReceiveResult.CLOSED; - } else { - return result; + if (select != null) { + // cell is empty, no sender, and we are in a select -> store the select instance + // and await externally + var storedSelect = new StoredSelect(select, segment, i, false, selectClause); + if (segment.casCell(i, state, storedSelect)) { + expandBuffer(); + return storedSelect; } + // else: CAS unsuccessful, repeat + } else { + // cell is empty, and no sender -> suspend + // not using any payload + var c = new Continuation(null); + if (segment.casCell(i, state, c)) { + expandBuffer(); + var result = c.await(segment, i); + if (result == ChannelClosedMarker.CLOSED) { + return ReceiveResult.CLOSED; + } else { + return result; + } + } + // else: CAS unsuccessful, repeat } - // else: CAS unsuccessful, repeat } else { // sender in progress, receiver changed state first -> restart if (segment.casCell(i, state, BROKEN)) { @@ -408,6 +448,7 @@ private Object updateCellReceive(Segment segment, int i, long r) throws Interrup } } case Continuation c -> { + // resolving a potential race with `expandBuffer` if (segment.casCell(i, state, RESUMING)) { // a sender is waiting -> trying to resume if (c.tryResume(0)) { @@ -415,13 +456,16 @@ private Object updateCellReceive(Segment segment, int i, long r) throws Interrup expandBuffer(); return c.getPayload(); } else { - // cell interrupted -> trying with a new one + // when cell interrupted -> trying with a new one // the state will be set to INTERRUPTED_SEND by the continuation, meanwhile everybody else will observe RESUMING + // when close in progress -> the cell state will be updated to CLOSED, subsequent cells are already closed, + // which will be detected in the next iteration return ReceiveResult.FAILED; } } // else: CAS unsuccessful, repeat } + // TODO: StoredSelect case Buffered b -> { segment.setCell(i, DONE); expandBuffer(); @@ -487,7 +531,7 @@ private void expandBuffer() { // the cell is already counted as processed by the close procedure return; } - // else, the cell must have been an interrupted sender; `Continuation` the properly notifies the segment + // else, the cell must have been an interrupted sender; `Continuation` properly notifies the segment } } @@ -503,8 +547,10 @@ private ExpandBufferResult updateCellExpandBuffer(Segment segment, int i) { segment.setCell(i, new Buffered(c.getPayload())); return ExpandBufferResult.DONE; } else { - // cell interrupted -> trying with a new one + // when cell interrupted -> trying with a new one // the state will be set to INTERRUPTED_SEND by the continuation, meanwhile everybody else will observe RESUMING + // when close in progress -> the cell state will be updated to CLOSED, subsequent cells are already closed, + // which will be detected in the next iteration return ExpandBufferResult.FAILED; } } @@ -514,6 +560,26 @@ private ExpandBufferResult updateCellExpandBuffer(Segment segment, int i) { // must be a receiver continuation - another buffer expansion already happened return ExpandBufferResult.DONE; } + case StoredSelect ss when ss.isSender() -> { + // TODO + throw new UnsupportedOperationException(); +// if (segment.casCell(i, state, RESUMING)) { +// // a sender is waiting -> trying to resume +// if (ss.getSelect().trySelect(ss,0)) { +// segment.setCell(i, new Buffered(c.getPayload())); +// return ExpandBufferResult.DONE; +// } else { +// // select unsuccessful -> trying with a new one +// // the state will be set to INTERRUPTED_SEND by the continuation, meanwhile everybody else will observe RESUMING +// return ExpandBufferResult.FAILED; +// } +// } +// // else: CAS unsuccessful, repeat + } + case StoredSelect ss -> { + // must be a receiver clause of the select - another buffer expansion already happened + return ExpandBufferResult.DONE; + } case Buffered b -> { // an element is already buffered; if the ordering of operations was different, we would put IN_BUFFER in that cell and finish return ExpandBufferResult.DONE; @@ -697,17 +763,30 @@ private void updateCellClose(Segment segment, int i) { } } case Continuation c -> { - if (segment.casCell(i, state, RESUMING)) { - if (c.tryResume(ChannelClosedMarker.CLOSED)) { - segment.setCell(i, CLOSED); - segment.cellInterruptedSender_orClosed(); - return; - } else { - // cell interrupted - the segment counters will be appropriately decremented from the - // continuation, depending if this is a sender or receiver; moreover, the cell is already - // processed in case this is a receiver - return; - } + // potential race wih sender/receiver resuming the continuation - resolved by synchronizing on + // `Continuation.data`: only one thread will successfully change its value from `null` + if (c.tryResume(ChannelClosedMarker.CLOSED)) { + segment.setCell(i, CLOSED); + segment.cellInterruptedSender_orClosed(); + return; + } else { + // when cell interrupted - the segment counters will be appropriately decremented from the + // continuation, depending on if this is a sender or receiver; moreover, the cell is already + // processed in case this is a receiver + // otherwise, the cell might be completed with a value, and the result is processed normally + // on another thread + return; + } + } + case StoredSelect ss -> { + if (ss.getSelect().channelClosed(closedReason.get())) { + segment.setCell(i, CLOSED); + segment.cellInterruptedSender_orClosed(); + return; + } else { + // the select hasn't been closed; instead, it will clean up all cells as part of handling the + // non-closed state in its main loop + return; } } case DONE, BROKEN -> { @@ -742,6 +821,63 @@ public Throwable isError() { } } + // ************** + // Select clauses + // ************** + + private static final Function IDENTITY = Function.identity(); + + public SelectClause receiveClause() { + //noinspection unchecked + return receiveClauseMap((Function) IDENTITY); + } + + public SelectClause receiveClauseMap(Function callback) { + return new SelectClause<>() { + @Override + Channel getChannel() { + return Channel.this; + } + + @Override + Object register(SelectInstance select) { + try { + return doReceive(select, this); + } catch (InterruptedException e) { + // not possible, as we provide a select, so no suspension should happen + throw new IllegalStateException(e); + } + } + + @Override + U transformedRawValue() { + //noinspection unchecked + return callback.apply((T) rawValue); + } + }; + } +// +// public SelectClause sendClause(T value) { +// return sendClauseMap(value, () -> null); +// } +// +// public SelectClause sendClauseMap(T value, Supplier callback) {} + + void disposeStoredSelect(Segment segment, int i, boolean isSender) { + // We treat the cell as if it was interrupted - the code is same as in `Continuation.await`; + // there's no need to resolve races with `SelectInstance.trySelect`, as disposal is called either when a clause + // is selected, a channel is closed, or during re-registration. In all cases `trySelect` would fail. + // In other words, the races are resolved by synchronizing on `SelectInstance.state`. + segment.setCell(i, isSender ? INTERRUPTED_SEND : INTERRUPTED_RECEIVE); + + // notifying the segment - if all cells become interrupted, the segment can be removed + if (isSender) { + segment.cellInterruptedSender_orClosed(); + } else { + segment.cellInterruptedReceiver(); + } + } + // **** // Misc // **** @@ -795,6 +931,8 @@ public String toString() { case Buffered b -> sb.append("V(").append(b.value()).append(")"); case Continuation c when c.isSender() -> sb.append("WS(").append(c.getPayload()).append(")"); case Continuation c -> sb.append("WR"); + case StoredSelect ss when ss.isSender() -> sb.append("SS"); + case StoredSelect ss -> sb.append("SR"); default -> throw new IllegalStateException("Unexpected value: " + state); } if (i != Segment.SEGMENT_SIZE - 1) sb.append(","); @@ -818,13 +956,15 @@ enum SendResult { } /** - * Possible return values of {@code Channel#updateCellReceive}: one of the enum constants below, or the received value. + * Possible return values of {@code Channel#updateCellReceive}: one of the enum constants below, the received value or {@link SelectStored} (used for disposal). */ enum ReceiveResult { FAILED, CLOSED } +record SelectStored(Segment segment, int i) {} + /** * Possible return values of {@code Channel#expandBuffer}: one of the enum constants below, or the received value. */ @@ -834,7 +974,7 @@ enum ExpandBufferResult { CLOSED } -// possible states of a cell: one of the enum constants below, Buffered, or Continuation +// possible states of a cell: one of the enum constants below, Buffered, Continuation or SelectInstance enum CellState { DONE, @@ -853,7 +993,7 @@ final class Continuation { * The number of busy-looping iterations before yielding, during {@link Continuation#await(Segment, int)}. * {@code 0}, if there's a single CPU. */ - private static final int SPINS = Runtime.getRuntime().availableProcessors() == 1 ? 0 : 10000; + static final int SPINS = Runtime.getRuntime().availableProcessors() == 1 ? 0 : 10000; private final Thread creatingThread; private volatile Object data; // set using DATA var handle diff --git a/core/src/main/java/com/softwaremill/jox/Select.java b/core/src/main/java/com/softwaremill/jox/Select.java new file mode 100644 index 0000000..3c983e6 --- /dev/null +++ b/core/src/main/java/com/softwaremill/jox/Select.java @@ -0,0 +1,372 @@ +package com.softwaremill.jox; + +import java.util.*; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.LockSupport; + +public class Select { + /* + Inspired by Kotlin's implementation: https://github.com/Kotlin/kotlinx.coroutines/blob/master/kotlinx-coroutines-core/common/src/selects/Select.kt + + Each select invocation proceeds through a couple of states. The state is stored in an internal representation + of the select, a `SelectInstance`. + + First, the select starts in the `REGISTERING` state. For each clause, we call it's `register` method, which in turn + reserves a send/receive cell in the channel, and stores a `StoredSelect` instance. This instance, apart from the + `SelectInstance`, also holds the segment & index of the cell, so that we can clean up later, when another clause is + selected. + + During registration, if another thread tries to select a clause concurrently, it's prevented from doing so; + instead, we collect the clauses for which this happened in a list, and re-register them. Such a thread treats + such invocations as if the cell was interrupted, and retries with a new cell. Later, we clean up the stored + selects, which are re-registered, so that there are no memory leaks. + + It's possible that a clause is completed immediately after registration. If that's the case, we overwrite the state + (including a potentially concurrently set closed state). + + Any of the states set during registration are acted up in `checkStateAndWait`. The method properly cleans up in + case a clause was selected, or a channel becomes closed. If it sees a `REGISTERING` state, the state is changed + to the current `Thread`, and the computation is suspended. Then, another thread running `trySelect` or + `channelClosed`, after changing the state appropriately, has to wake up the thread to let the select know, that it + should inspect the state again. + */ + + /** + * Select exactly one clause to complete. Each clause should be created for a different channel. + * + * @param clauses The clauses, from which one will be selected. Not {@code null}. + * @return The value returned by the selected clause. + * @throws ChannelClosedException When any of the channels is closed. + */ + @SafeVarargs + public static U select(SelectClause... clauses) throws InterruptedException { + var r = selectSafe(clauses); + if (r instanceof ChannelClosed c) { + throw c.toException(); + } else { + //noinspection unchecked + return (U) r; + } + } + + /** + * Select exactly one clause to complete. Each clause should be created for a different channel. + * Doesn't throw exceptions when the channel is closed, but returns a value. + * + * @param clauses The clauses, from which one will be selected. Not {@code null}. + * @return Either the value returned by the selected clause, or {@link ChannelClosed}, when any of the channels is closed. + */ + @SafeVarargs + public static Object selectSafe(SelectClause... clauses) throws InterruptedException { + // check that the clause doesn't refer to a channel that is already used in a different clause + verifyChannelsUnique(clauses); + + var si = new SelectInstance(clauses.length); + for (var clause : clauses) { + if (!si.register(clause)) { + break; // channel is closed, or a clause was selected - in both cases, no point in further registrations + } + } + + return si.checkStateAndWait(); + } + + private static void verifyChannelsUnique(SelectClause[] clauses) { + // we expect the number of clauses to be small, so that this n^2 double-loop is faster than allocating a set + for (int i = 0; i < clauses.length; i++) { + for (int j = i + 1; j < clauses.length; j++) { + if (clauses[i].getChannel() == clauses[j].getChannel()) { + throw new IllegalArgumentException("Channel " + clauses[i].getChannel() + " is used in multiple clauses"); + } + } + } + } +} + +class SelectInstance { + private final AtomicReference state = new AtomicReference<>(SelectState.REGISTERING); + + /** + * The content of the list will be written & read only by the main select thread. Hence, no synchronization is necessary. + */ + private final List storedSelects; + + SelectInstance(int clausesCount) { + storedSelects = new ArrayList<>(clausesCount); + } + + // registration + + /** + * Register a clause in this select instance. Only one clause for each channel should be registered. + * + * @return {@code true}, if the registration was successful, and the clause has been stored. {@code false}, if the + * channel is closed, or the clause has been immediately selected. + */ + boolean register(SelectClause clause) { + // register the clause + var result = clause.register(this); + switch (result) { + case StoredSelect ss -> { + // keeping the stored select to later call dispose() + storedSelects.add(ss); + return true; + } + case ChannelClosed cc -> { + // when setting the state, we might override another state: + // - a list of clauses to re-register - there's no point in doing that anyway (since the channel is closed) + // - another closed state (set concurrently) + state.set(cc); + return false; + } + default -> { + // else: the clause was selected + clause.setRawValue(result); + // when setting the state, we might override another state: + // - a list of clauses to re-register - there's no point in doing that anyway (since we already selected a clause) + // - a closed state - the closure must have happened concurrently with registration; we give priority to immediate selects then + state.set(clause); + return false; + } + } + } + + // main loop + + /** + * @return Either the value returned by the selected clause, or {@link ChannelClosed}, when any of the channels is closed. + */ + Object checkStateAndWait() throws InterruptedException { + while (true) { + var currentState = state.get(); + switch (currentState) { + case SelectState.REGISTERING -> { + // registering done, waiting until a clause is selected - setting the thread to wake up as the state + // we won't leave this case until the state is changed from Thread + var currentThread = Thread.currentThread(); + if (state.compareAndSet(SelectState.REGISTERING, currentThread)) { + var spinIterations = Continuation.SPINS; + while (state.get() == currentThread) { + // same logic as in Continuation + if (spinIterations > 0) { + Thread.onSpinWait(); + spinIterations -= 1; + } else { + LockSupport.park(); + + if (Thread.interrupted()) { + if (state.compareAndSet(currentThread, SelectState.INTERRUPTED)) { + // since we changed the state, we know that none of the clauses will become completed + cleanup(null); + throw new InterruptedException(); + } else { + // another thread already changed the state; setting the interrupt status (so that + // the next blocking operation throws), and continuing + Thread.currentThread().interrupt(); + } + } + } + } + // inspect the updated state in next iteration + } + // else: CAS unsuccessful, retry + } + case ChannelClosed cc -> { + cleanup(null); + return cc; + } + case List clausesToReRegister -> { + // moving the state back to registering + if (state.compareAndSet(currentState, SelectState.REGISTERING)) { + //noinspection unchecked + for (var clause : (List>) clausesToReRegister) { + // disposing & removing the stored select for the clause which we'll re-register + var storedSelectsIterator = storedSelects.iterator(); + while (storedSelectsIterator.hasNext()) { + var stored = storedSelectsIterator.next(); + if (stored.getClause() == clause) { + stored.dispose(); + storedSelectsIterator.remove(); + break; + } + } + + if (!register(clause)) { + // channel is closed, or clause was selected - in both cases, no point in further + // re-registrations; the state should be appropriately updated + break; + } + } + // inspect the updated state in next iteration + } + // else: CAS unsuccessful, retry + } + case SelectClause selectedClause -> { + cleanup(selectedClause); + // running the transformation at the end, after the cleanup is done, in case this throws any exceptions + return selectedClause.transformedRawValue(); + } + default -> throw new IllegalStateException("Unknown state: " + currentState); + } + } + } + + private void cleanup(SelectClause selected) { + // disposing of all the clauses that were registered, except for the selected one + for (var stored : storedSelects) { + if (stored.getClause() != selected) { + stored.dispose(); + } + } + storedSelects.clear(); + } + + // callbacks from select, that a clause is selected / the channel is closed + + /** + * @return {@code true} if the given clause was successfully selected, {@code false} otherwise (a channel is closed, + * another clause is selected, registration is in progress, select is interrupted). + */ + boolean trySelect(StoredSelect storedSelect, Object rawValue) { + while (true) { + var currentState = state.get(); + switch (currentState) { + case SelectState.REGISTERING -> { + if (state.compareAndSet(currentState, Collections.singletonList(storedSelect.getClause()))) { + return false; // concurrent clause selection is not possible during registration + } + // else: CAS unsuccessful, retry + } + case List clausesToReRegister -> { + // we need a new object for CAS + var newClausesToReRegister = new ArrayList>(clausesToReRegister.size() + 1); + //noinspection unchecked + newClausesToReRegister.addAll((Collection>) clausesToReRegister); + newClausesToReRegister.add(storedSelect.getClause()); + if (state.compareAndSet(currentState, newClausesToReRegister)) { + return false; // concurrent clause selection is not possible during registration + } + // else: CAS unsuccessful, retry + } + case SelectState.INTERRUPTED -> { + // already interrupted, will be cleaned up soon + return false; + } + case ChannelClosed cc -> { + // closed, will be cleaned up soon + return false; + } + case SelectClause selectedClause -> { + // already selected, will be cleaned up soon + return false; + } + case Thread t -> { + // setting the value first, before the memory barrier created by setting (and in the main loop + // thread, reading) the state. + storedSelect.getClause().setRawValue(rawValue); + if (state.compareAndSet(currentState, storedSelect.getClause())) { + LockSupport.unpark(t); + return true; + } + // else: CAS unsuccessful, retry + } + default -> throw new IllegalStateException("Unknown state: " + currentState); + } + } + } + + /** + * @return {@code true}, if the state of the select has been changed to closed; otherwise, a clause might have been + * already selected, another closure reason provided, or the select might have been interrupted. + */ + boolean channelClosed(ChannelClosed channelClosed) { + while (true) { + var currentState = state.get(); + switch (currentState) { + case SelectState.REGISTERING -> { + // the channel closed state will be discovered when there's a call to `checkStateAndWait` after registration completes + if (state.compareAndSet(currentState, channelClosed)) { + return true; + } + // else: CAS unsuccessful, retry + } + case List clausesToReRegister -> { + // same as above + if (state.compareAndSet(currentState, channelClosed)) { + return true; + } + // else: CAS unsuccessful, retry + } + case SelectState.INTERRUPTED -> { + // already interrupted + return false; + } + case ChannelClosed cc -> { + // already closed + return false; + } + case SelectClause selectedClause -> { + // already selected + return false; + } + case Thread t -> { + if (state.compareAndSet(currentState, channelClosed)) { + LockSupport.unpark(t); + return true; + } + // else: CAS unsuccessful, retry + } + default -> throw new IllegalStateException("Unknown state: " + currentState); + } + } + } +} + +// possible states of a select instance: +// - one of the enumeration values below +// - Thread to wake up +// - ChannelClosed +// - a List of clauses to re-register +// - SelectClause (the selected clause) + +enum SelectState { + REGISTERING, + INTERRUPTED +} + +// + +/** + * Used to keep information about a select instance that is stored in a channel, awaiting completion. + */ +class StoredSelect { + private final SelectInstance select; + private final Segment segment; + private final int i; + private final boolean isSender; + private final SelectClause clause; + + public StoredSelect(SelectInstance select, Segment segment, int i, boolean isSender, SelectClause clause) { + this.select = select; + this.segment = segment; + this.i = i; + this.isSender = isSender; + this.clause = clause; + } + + public SelectInstance getSelect() { + return select; + } + + public boolean isSender() { + return isSender; + } + + SelectClause getClause() { + return clause; + } + + void dispose() { + clause.getChannel().disposeStoredSelect(segment, i, isSender); + } +} diff --git a/core/src/main/java/com/softwaremill/jox/SelectClause.java b/core/src/main/java/com/softwaremill/jox/SelectClause.java new file mode 100644 index 0000000..016891d --- /dev/null +++ b/core/src/main/java/com/softwaremill/jox/SelectClause.java @@ -0,0 +1,30 @@ +package com.softwaremill.jox; + +/** + * A clause to use as part of {@link Select#select(SelectClause[])}. Clauses can be created having a channel instance, + * using {@link Channel#receiveClause()}. + */ +public abstract class SelectClause { + /** + * The value might be set from different threads, but there's always a memory barrier between write & read, due to + * synchronization on {@link SelectInstance}'s {@code state} field. + */ + protected Object rawValue; + + abstract Channel getChannel(); + + /** + * @return Either a {@link StoredSelect}, {@link ChannelClosed} when the channel is already closed, or the selected value. + */ + abstract Object register(SelectInstance select); + + /** + * Should be called only after {@link #setRawValue(Object)}. Transforms the raw value with the transformation function + * provided when creating the clause. + */ + abstract T transformedRawValue(); + + void setRawValue(Object rawValue) { + this.rawValue = rawValue; + } +} diff --git a/core/src/test/java/com/softwaremill/jox/SelectTest.java b/core/src/test/java/com/softwaremill/jox/SelectTest.java new file mode 100644 index 0000000..633f55b --- /dev/null +++ b/core/src/test/java/com/softwaremill/jox/SelectTest.java @@ -0,0 +1,22 @@ +package com.softwaremill.jox; + +import org.junit.jupiter.api.Test; + +import static com.softwaremill.jox.Select.select; +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class SelectTest { + @Test + public void testSimpleSelect() throws InterruptedException { + // given + Channel ch1 = new Channel<>(1); + Channel ch2 = new Channel<>(1); + ch2.send("v"); + + // when + String received = select(ch1.receiveClause(), ch2.receiveClause()); + + // then + assertEquals("v", received); + } +}