Async search context and and state management framework #6

eirsep · 2020-11-09T09:56:33Z

Issue #3

Description of changes

Contains the components surrounding the Async Search Context, both active and persisted :

The active context, represents an ongoing and memory-resident search, keeps track of the underlying SearchTask and the SearchProgressActionListener.
The persisted context represents a persisted version of AsyncSearchContext through a backing AsyncSearchPersistenceModel.

Async Search Stage and Async Search Context Permits are also introduced with the changes.

TODO

Multi threaded tests for Async Search Active Context

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

…veContextId

itiyamas

Some comments below. Not reviewed the tests yet.
It is generally a good practice to add assertions for invariant conditions. That goes a long way in protecting your system against bugs that violate those invariants, given you have the right tests. Another advantage that you get is the invariants are clearly documented in the code. I noticed that there is scope for adding invariants in a lot of places- especially the transitions and permits classes.

itiyamas · 2020-11-11T03:33:58Z

...in/java/com/amazon/opendistroforelasticsearch/search/async/context/AsyncSearchContextId.java

+import java.io.IOException;
+import java.util.Objects;
+
+public class AsyncSearchContextId implements Writeable {


Some explanation of why you use this as the id would help? Was a UUID not enough for uniqueness. What is the purpose of the long id?

The long id is used as a key for a MAP of AsyncSearchContext objects stored on every node (AsyncSearchActiveStore class which will be introduced in an upcoming PR contains said map).
The UUID injects a randomness to the asyncsearch id string which is used to access an existing async search

UUID is for randomness(maynot be good candidate for hash keys/uniqueness guarantees) and id for better hashing and uniqueness guarantees

itiyamas · 2020-11-11T03:38:43Z

src/main/java/com/amazon/opendistroforelasticsearch/search/async/AsyncSearchId.java

+            out.writeLong(asyncSearchId.getTaskId());
+            out.writeString(asyncSearchId.getAsyncSearchContextId().getContextId());
+            out.writeLong(asyncSearchId.getAsyncSearchContextId().getId());
+            return Base64.getUrlEncoder().encodeToString(BytesReference.toBytes(out.bytes()));


withoutPadding() ?

changed this to have withoutPadding()

itiyamas · 2020-11-11T03:39:29Z

src/main/java/com/amazon/opendistroforelasticsearch/search/async/AsyncSearchId.java

+import java.util.Base64;
+import java.util.Objects;
+
+public class AsyncSearchId {


An explanation on why you chose this as the id would be helpful.

+1. Lets add this across other newly introduced constructs.

Missed adding a java doc on this class. Will add one.

Added java docs

We'll change it to be Converter

itiyamas · 2020-11-11T03:51:21Z

...in/java/com/amazon/opendistroforelasticsearch/search/async/response/AsyncSearchResponse.java

@@ -111,8 +110,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
        builder.field(START_TIME_IN_MILLIS.getPreferredName(), startTimeMillis);
        builder.field(EXPIRATION_TIME_IN_MILLIS.getPreferredName(), expirationTimeMillis);
        if (searchResponse != null) {
-            builder.field(RESPONSE.getPreferredName());


Why was this not caught in tests in the earlier change? Do we need to add more tests?

AsyncSearchResponseTests.java has been added in this PR.

It verifies serialization of AsyncSearchResponse.

itiyamas · 2020-11-11T03:52:59Z

...in/java/com/amazon/opendistroforelasticsearch/search/async/response/AsyncSearchResponse.java

-                expirationTimeMillis == other.expirationTimeMillis &&
-                searchResponse == other.searchResponse &&
-                error == other.error;
+        try {


Where is the equals method called in the request flow?

It is used only in tests.

itiyamas · 2020-11-11T06:15:53Z

.../amazon/opendistroforelasticsearch/search/async/context/active/AsyncSearchActiveContext.java

+     *
+     * @param nextAsyncSearchStage asyncSearchStage to advance
+     */
+    public synchronized void advanceStage(AsyncSearchStage nextAsyncSearchStage) {


This should not be public. The events should be public and this class should decide the stage based on the event.

making it package-private for testing.

As discussed, we moved most of this to the state machine

itiyamas · 2020-11-11T06:19:11Z

.../amazon/opendistroforelasticsearch/search/async/context/active/AsyncSearchActiveContext.java

+        if (currentAsyncSearchStage != nextAsyncSearchStage) {
+            try {
+                asyncSearchStage.onTransition(contextListener, getContextId());
+            } catch (Exception ex) {


Swallowing all exceptions! Would you get a consistent behavior if listener is not executed. Should you not fail the context in some tragic way or atleast close it if this happens?

Agreed. Removing the surrounding try-catch.

Listeners in ES are almost always surrounded by parent catch all

itiyamas · 2020-11-11T06:20:19Z

.../amazon/opendistroforelasticsearch/search/async/context/active/AsyncSearchActiveContext.java

+        }
+    }
+
+    public void acquireContextPermit(final ActionListener<Releasable> onPermitAcquired, TimeValue timeout, String reason) {


Who would be the callers? Permit seems like an internal construct of context. Maybe send the event to context and let context decide about the permit? Maybe I need to see more classes here to understand the behavior.

The AsyncSearchService (which will be introduced in an upcoming PR) will be invoking the methods surround the permits acquisition.

itiyamas · 2020-11-11T06:24:31Z

...main/java/com/amazon/opendistroforelasticsearch/search/async/context/AsyncSearchContext.java

+
+    protected final AsyncSearchContextId asyncSearchContextId;
+    protected final LongSupplier currentTimeSupplier;
+    protected volatile AsyncSearchProgressListener asyncSearchProgressListener;


You are redefining the listener in the child class.

It has not been redefined in the child classes - AsyncSearchActiveContext and AsyncSearchPersistenceContext

itiyamas · 2020-11-11T06:27:01Z

...main/java/com/amazon/opendistroforelasticsearch/search/async/context/AsyncSearchContext.java

+    }
+
+    public @Nullable
+    AsyncSearchProgressListener getAsyncSearchProgressListener() {


Why nullable? Maybe define this method only in subclasses that are expected to have a listener object?

As discussed offline, we need to provide a single interface to the consumers. Retaining it for now as we don't have a better/cleaner alternative

getsaurabh02

Thanks for making the changes. Reviewed 11/16 files (excluding test) till now. Overall looks good to me. Added few comments for clarification and considerations.

getsaurabh02 · 2020-11-11T07:00:11Z

src/main/java/com/amazon/opendistroforelasticsearch/search/async/AsyncSearchId.java

+import java.util.Base64;
+import java.util.Objects;
+
+public class AsyncSearchId {


+1. Lets add this across other newly introduced constructs.

getsaurabh02 · 2020-11-11T07:02:01Z

src/main/java/com/amazon/opendistroforelasticsearch/search/async/AsyncSearchId.java

+        return "[" + node + "][" + taskId + "][" + asyncSearchContextId + "]";
+    }
+
+    public static String buildAsyncId(AsyncSearchId asyncSearchId) {


nit: shall we keep this as final for clear intent.

Declared the class final.

getsaurabh02 · 2020-11-11T07:24:52Z

src/main/java/com/amazon/opendistroforelasticsearch/search/async/AsyncSearchId.java

+import java.util.Base64;
+import java.util.Objects;
+
+public class AsyncSearchId {


Should we declare this class as final to prevent extensions?

getsaurabh02 · 2020-11-11T08:22:00Z

...main/java/com/amazon/opendistroforelasticsearch/search/async/context/AsyncSearchContext.java

+        return Collections.unmodifiableSet(
+                Sets.newHashSet(
+                        AsyncSearchStage.INIT, AsyncSearchStage.RUNNING, AsyncSearchStage.SUCCEEDED, AsyncSearchStage.FAILED));


From async search response perspective only RUNNING is needed. What is the purpose of maintaining these stages in a set here?

This is a method which will be invoked by AsyncSearchService (which will be introduced in upcoming PRs) to scan for expired searches which need to be cleaned up.

[Question] Can it be moved to constant, rather than a method call?

getsaurabh02 · 2020-11-11T20:15:41Z

.../amazon/opendistroforelasticsearch/search/async/context/active/AsyncSearchActiveContext.java

+            throw new IllegalStateException("only next asyncSearchStage " + INIT +
+                    " is allowed when the current asyncSearchStage is null");


Are these real production scenario we can hit?

Concurrent Delete Async Search requests need to be handled. That is a valid production scenario.

Multiple delete async search requests can get hold of the context object but only one should be allowed to advance context stage to DELETED

getsaurabh02 · 2020-11-11T20:25:37Z

.../amazon/opendistroforelasticsearch/search/async/context/active/AsyncSearchActiveContext.java

+    public void acquireAllContextPermits(final ActionListener<Releasable> onPermitAcquired, TimeValue timeout, String reason) {
+        asyncSearchContextPermits.asyncAcquireAllPermits(onPermitAcquired, timeout, reason);
+    }


Do we worry about starvation here, which could result in lingering state transition? How about restricting fresh permits once aquireAll has been issued, to ensure fairness?

If allPermits have been aquired that means the search is in post processing state and will leave the lock soon.
And if that happens within the acquire timeout, the request will proceed, else it will fail with TimeoutException and the we wont run into a starvation state.
@eirsep Can clarify

If acquireAll is invoked and queued for acquiring permits, fresh permits are queued behind it as Semaphore has fair setting on TRUE. Fairness ensure FIFO basis for permit acquisition.

FIFO will be applicable only on the waiting requests. However, here the single permit requests will not wait and will be immediately allowed. Maybe the use case for allPermits contention is rare and can be handled later based on need.

getsaurabh02 · 2020-11-11T20:41:40Z

...mazon/opendistroforelasticsearch/search/async/context/permits/AsyncSearchContextPermits.java

+
+    private Releasable acquirePermits(int permits, TimeValue timeout, final String details) throws RuntimeException {
+        try {
+            if (semaphore.tryAcquire(permits, timeout.getMillis(), TimeUnit.MILLISECONDS)) {


What is the advantage of keeping a timeout here, considering the total permits is integer max. There are definite gains of not using timeout though. One use case I can think is for mutating operations contending for permit while all is acquired by post processor?

The timeout is because there is also an asyncAcquireAllPermits which takes integer max number of permits and hence is a blocking call (think persisting completed async search response in index). We need to timeout mutating operations which are blocked beyond a waiting time.

getsaurabh02 · 2020-11-11T20:54:48Z

...mazon/opendistroforelasticsearch/search/async/context/permits/AsyncSearchContextPermits.java

+                                "previous lock details: [" + lockDetails + "] trying to lock for [" + details + "]");
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();


Possibly we don't think we need to worry about releasing permit here, as one is not acquired yet.

getsaurabh02 · 2020-11-11T20:57:59Z

...mazon/opendistroforelasticsearch/search/async/context/permits/AsyncSearchContextPermits.java

+
+    private void asyncAcquirePermit(
+            int permits, final ActionListener<Releasable> onAcquired, final TimeValue timeout, String reason) {
+        threadPool.executor(AsyncSearchPlugin.OPEN_DISTRO_ASYNC_SEARCH_GENERIC_THREAD_POOL_NAME).execute(new AbstractRunnable() {


Do we plan to introduce a separate threadpool going forward?

Not sure which thread pool you mean, while we might add few based on our performance tests

Dedicated threadpool for different asyc search operations. Agree that can wait for more concrete results based on perf testing.

…cement and initialized stage in active context constructor.

psychbot · 2020-11-13T09:49:35Z

src/main/java/com/amazon/opendistroforelasticsearch/search/async/AsyncSearchId.java

+            out.writeLong(asyncSearchId.getAsyncSearchContextId().getId());
+            return Base64.getUrlEncoder().withoutPadding().encodeToString(BytesReference.toBytes(out.bytes()));
+        } catch (Exception e) {
+            throw new IllegalArgumentException("Cannot build async search id", e);


[NitPick] Should we have a separate exception handling and throw appropriate errors instead of a generic one.
Will that be helpful for the end user?

As mentioned in AsyncSearchIdTests.java, this exception will be caught and in turn a ResourceNotFoundException or a wrapper on RNF will be thrown to indicate that no Async Search Execution exists with the illegal Id. That responsibility will be delegated to the class handling the request.

psychbot · 2020-11-13T09:51:10Z

src/main/java/com/amazon/opendistroforelasticsearch/search/async/AsyncSearchId.java

+            out.writeString(asyncSearchId.getAsyncSearchContextId().getContextId());
+            out.writeLong(asyncSearchId.getAsyncSearchContextId().getId());
+            return Base64.getUrlEncoder().withoutPadding().encodeToString(BytesReference.toBytes(out.bytes()));
+        } catch (Exception e) {


[NitPick] We are catching all the exception which it too broad, Will it be possible to narrow it down?

Changing this to the expected IOException. Thank you.

psychbot · 2020-11-13T09:56:00Z

src/main/java/com/amazon/opendistroforelasticsearch/search/async/AsyncSearchId.java

+    }
+
+    @Override
+    public boolean equals(Object o) {


[Question] Will deepEquals suffice the use case here? Applicable in other classes.

No. I think deepEquals targets Arrays. If the objects being compared aren't arrays of any sort, it simply check equals

psychbot · 2020-11-13T10:01:16Z

...main/java/com/amazon/opendistroforelasticsearch/search/async/context/AsyncSearchContext.java

+    protected volatile AsyncSearchProgressListener asyncSearchProgressListener;
+
+    public AsyncSearchContext(AsyncSearchContextId asyncSearchContextId, LongSupplier currentTimeSupplier) {
+        Objects.requireNonNull(asyncSearchContextId);


[NitPick]

This will raise a NPE is that expected and handled at caller.

We can use @NonNull annotation in the method definition itself. You can explore it and apply at other places as well.

Thanks for pointing this out. Had meant to add an assert.

psychbot · 2020-11-13T10:05:48Z

...main/java/com/amazon/opendistroforelasticsearch/search/async/context/AsyncSearchContext.java

+        return Collections.unmodifiableSet(
+                Sets.newHashSet(
+                        AsyncSearchStage.INIT, AsyncSearchStage.RUNNING, AsyncSearchStage.SUCCEEDED, AsyncSearchStage.FAILED));


[Question] Can it be moved to constant, rather than a method call?

psychbot · 2020-11-13T10:18:18Z

.../amazon/opendistroforelasticsearch/search/async/context/active/AsyncSearchActiveContext.java

+    }
+
+    public boolean shouldPersist() {
+        return keepOnCompletion && isExpired() == false && asyncSearchStage != DELETED;


[NitPick] We can use !isExpired() and save on additional comparison operation.

psychbot · 2020-11-13T10:22:33Z

.../amazon/opendistroforelasticsearch/search/async/context/active/AsyncSearchActiveContext.java

+        assert nextAsyncSearchStage != null : "Next async search stage cannot bu null!";
+        assert this.asyncSearchStage != null : "async search stage cannot be null!";
+        AsyncSearchStage currentAsyncSearchStage = this.asyncSearchStage;
+        if (this.asyncSearchStage.nextTransitions().contains(nextAsyncSearchStage) == false) {


[NitPick] Can we can avoid the comparison.

psychbot · 2020-11-13T10:24:22Z

.../amazon/opendistroforelasticsearch/search/async/context/active/AsyncSearchActiveContext.java

+    private volatile AsyncSearchStage asyncSearchStage;
+    private final AsyncSearchContextListener contextListener;
+
+    public AsyncSearchActiveContext(AsyncSearchContextId asyncSearchContextId, String nodeId,


A small java doc on top of every public class and method will be helpful.
Applicable at other places as well.

There is a java doc for the class and any methods which need explanation. I don't think one is required here, as it is the sole constructor for the class.

psychbot · 2020-11-13T10:41:17Z

...mazon/opendistroforelasticsearch/search/async/context/permits/AsyncSearchContextPermits.java

+     * @param reason the reason for acquiring the permit
+     */
+    public void asyncAcquireAllPermits(final ActionListener<Releasable> onAcquired, final TimeValue timeout, String reason) {
+        asyncAcquirePermit(TOTAL_PERMITS, onAcquired, timeout, reason);


[Question] I don't really get the logic of Integer.MAX_VALUE.
We are saying that for any mutation operation we will first acquire lock on the permit value 1(asyncAcquirePermit method) and release it at the end, why to we need to consider the MAX_VALUE space?
Shouldn't acquiring a lock on the same permit value(1) will suffice?

When we acquire a single permit out of Integer.MAX_VALUE number of available permits, we imply that its a non
blocking operation. We support concurrent operations of this sort.(asyncAcquirePermits) For ex. updating expiry time is non blocking.

When we acquire all available permits, we imply that its a blocking operation.(asyncAcquireAllPermits). All other operations trying to acquire 1 or all permits will have to wait on this or timeout. For ex. storing completed async search response in index is blocking operation.

psychbot · 2020-11-13T10:43:16Z

.../amazon/opendistroforelasticsearch/search/async/context/active/AsyncSearchActiveContext.java

+    public void acquireAllContextPermits(final ActionListener<Releasable> onPermitAcquired, TimeValue timeout, String reason) {
+        asyncSearchContextPermits.asyncAcquireAllPermits(onPermitAcquired, timeout, reason);
+    }


If allPermits have been aquired that means the search is in post processing state and will leave the lock soon.
And if that happens within the acquire timeout, the request will proceed, else it will fail with TimeoutException and the we wont run into a starvation state.
@eirsep Can clarify

psychbot

Thanks for doing these changes, Added a few minor comment and questions.

psychbot · 2020-11-13T10:47:58Z

...opendistroforelasticsearch/search/async/context/persistence/AsyncSearchPersistenceModel.java

+        if (response == null) {
+            return emptyMap();
+        }
+        return convertToMap(response, false, Requests.INDEX_CONTENT_TYPE).v2();


[NitPick] Can use ternary operator. Applicable at other places.

…earchStateMachine

…r events

itiyamas · 2020-11-19T18:55:01Z

src/main/java/com/amazon/opendistroforelasticsearch/search/async/plugin/AsyncSearchPlugin.java

+        return Collections.singletonList(stateMachine);
+    }
+
+    private AsyncSearchStateMachine getAsyncSearchStateMachineDefinition() {


Thanks for making these changes. Looks much cleaner now!

itiyamas · 2020-11-19T18:58:07Z

...a/com/amazon/opendistroforelasticsearch/search/async/context/state/AbstractStateMachine.java

+    }
+
+    @Override
+    public AsyncSearchState trigger(AsyncSearchContextEvent event) {


Could there be more than 2 events triggered on state machine? If yes, how do you ensure locking/event accumulation?

added synchronized block on context object

Thanks for pointing this out, got missed with the state machine refactor

itiyamas · 2020-11-19T18:59:20Z

...a/com/amazon/opendistroforelasticsearch/search/async/context/state/AbstractStateMachine.java

+        if (getFinalStates().contains(currentState)) {
+            result = currentState;
+        } else {
+            for (AsyncSearchTransition<? extends AsyncSearchContextEvent> transition : getTransitions()) {


Could have used a map instead of iterating through the list?

agreed. changed transitions to be a map.

itiyamas · 2020-11-19T19:01:12Z

...a/com/amazon/opendistroforelasticsearch/search/async/context/state/AbstractStateMachine.java

+                        eventListener.accept(event.asyncSearchContext().getContextId(), event.asyncSearchContext().getContextListener());
+                    } catch (Exception ex) {
+                        logger.error(() -> new ParameterizedMessage("Failed to execute listener for async search id : {}",
+                                event.asyncSearchContext.getAsyncSearchId()), ex);


Is this blanket level handling intentional? What happens if the event listener is not executed- is that a best effort operation and its failure does not impact async search in any way?

The event listener will be used to collects node level stats about async searches. Its not an operation which impacts any ongoing async search requests.

Yes, its intentional given its a non-critical listener consumed for stats

itiyamas · 2020-11-19T19:11:59Z

...a/com/amazon/opendistroforelasticsearch/search/async/context/state/AbstractStateMachine.java

+                if (currentState.equals(transition.sourceState()) && transition.eventType().equals(event.getClass())) {
+                    execute(transition.onEvent(), event, currentState);
+                    event.asyncSearchContext().setState(transition.targetState());
+                    logger.debug("Executed event {} for async event {} ", event.getClass().getName(),
+                            event.asyncSearchContext.getAsyncSearchId());
+                    BiConsumer<AsyncSearchContextId, AsyncSearchContextListener> eventListener = transition.eventListener();
+                    try {
+                        eventListener.accept(event.asyncSearchContext().getContextId(), event.asyncSearchContext().getContextListener());
+                    } catch (Exception ex) {


I am not very clear on how synchronization is handled. This machine is stateless, but the context is stateful. However, the operations sent to context are not syncronized. Am I missing something here?

Added synchronization block with lock on context. Had just pushed a commit
57010e4

Thanks for pointing this out, got missed with the state machine refactor

…condensed and added synchronized block. added asyncsearchstatemachineException for illegal transition triggers

itiyamas · 2020-11-23T03:39:09Z

...mazon/opendistroforelasticsearch/search/async/context/permits/AsyncSearchContextPermits.java


    private static final int TOTAL_PERMITS = Integer.MAX_VALUE;

-    //visible for testing
    final Semaphore semaphore;
    private final AsyncSearchContextId asyncSearchContextId;
    private volatile String lockDetails;


remove this?

Will check on this. Will revisit with more tests

itiyamas · 2020-11-23T03:52:51Z

...mazon/opendistroforelasticsearch/search/async/context/active/AsyncSearchActiveContextIT.java

+import static com.amazon.opendistroforelasticsearch.search.async.context.state.AsyncSearchState.RUNNING;
+import static com.amazon.opendistroforelasticsearch.search.async.context.state.AsyncSearchState.SUCCEEDED;
+
+public class AsyncSearchActiveContextIT extends AsyncSearchSingleNodeTestCase {


test cases missing for async search state machine.

functional tests for asyncsearch state machine can be found in AsyncSearchStateMachineTests.java.

This class is verifying the state machine definition used for active context states and transitions

itiyamas · 2020-11-23T03:56:07Z

.../opendistroforelasticsearch/search/async/context/permits/AsyncSearchContextPermitsTests.java

+import static org.hamcrest.Matchers.hasToString;
+import static org.hamcrest.Matchers.instanceOf;
+
+public class AsyncSearchContextPermitsTests extends ESTestCase {


Missing test cases.

Can you plz elaborate on the specific scenarios which you feel we haven't handled?

itiyamas · 2020-11-23T03:59:24Z

...a/com/amazon/opendistroforelasticsearch/search/async/state/AsyncSearchStateMachineTests.java

+import static com.amazon.opendistroforelasticsearch.search.async.context.state.AsyncSearchState.PERSIST_FAILED;
+import static com.amazon.opendistroforelasticsearch.search.async.context.state.AsyncSearchState.SUCCEEDED;
+
+public class AsyncSearchStateMachineTests extends ESTestCase {


Missing test cases

itiyamas · 2020-11-23T04:01:16Z

...om/amazon/opendistroforelasticsearch/search/async/context/state/AsyncSearchStateMachine.java

+                asyncSearchContext.setState(transition.targetState());
+                logger.debug("Executed event {} for async event {} ", event.getClass().getName(),
+                        event.asyncSearchContext.getAsyncSearchId());
+                BiConsumer<AsyncSearchContextId, AsyncSearchContextListener> eventListener = transition.eventListener();


Does this code block need to be in synchronized block?

Do you mean the event listener in the synchronised block. I don't see an issue, given that we today only intend to update stats which is pretty light weight. If we don't synchronise the stats can mess up in some unexpected ways too. I plan on retaining it for now and change later as needed

itiyamas · 2020-11-23T04:11:12Z

.../amazon/opendistroforelasticsearch/search/async/context/active/AsyncSearchActiveContext.java

+    public void setTask(SearchTask searchTask) {
+        assert currentStage == INIT;
+        Objects.requireNonNull(searchTask);
+        searchTask.setProgressListener(asyncSearchProgressListener);
+        this.searchTask.set(searchTask);
+        this.startTimeMillis = searchTask.getStartTime();
+        this.expirationTimeMillis = startTimeMillis + keepAlive.getMillis();
+        this.asyncSearchId.set(AsyncSearchIdConverter.buildAsyncId(new AsyncSearchId(nodeId, searchTask.getId(), getContextId())));
+    }
+
+    public void processSearchFailure(Exception e) {
+        assert currentStage != DELETED : "cannot process search failure. Async search context is already DELETED";
+        if (completed.compareAndSet(false, true)) {


I am assuming these methods, even though public, are not called from anywhere except state machine execution and are protected under synchronized blocks.

Yes. They are protected under synchronized blocks and only called in the state machine trigger() method.

setTask cannot be invoked concurrently by any means as it is a part if the async search submission and hence guarded by SetOnce construct

We are using CAS in cases where it makes more sense rather than synchronized eg : ProcessSearchFailure processSearchResponse

itiyamas · 2020-11-23T04:15:08Z

.../amazon/opendistroforelasticsearch/search/async/context/active/AsyncSearchActiveContext.java

+        return asyncSearchId.get();
+    }
+
+    public boolean shouldPersist() {


Add synchronized?
Who is the caller?

Wondering if volatile isn't good enough

Are you leaking anything if stage is returned, but the actions post modifying the stage is not executed? e.g. listener execution. It seems fine for now given listeners are independent as you mentioned.

itiyamas

As discussed offline, missing tests will be added in a separate PR.

getsaurabh02 · 2020-11-23T06:30:33Z

.../amazon/opendistroforelasticsearch/search/async/context/active/AsyncSearchActiveContext.java

+    public void acquireAllContextPermits(final ActionListener<Releasable> onPermitAcquired, TimeValue timeout, String reason) {
+        asyncSearchContextPermits.asyncAcquireAllPermits(onPermitAcquired, timeout, reason);
+    }


FIFO will be applicable only on the waiting requests. However, here the single permit requests will not wait and will be immediately allowed. Maybe the use case for allPermits contention is rare and can be handled later based on need.

getsaurabh02 · 2020-11-23T06:33:28Z

...mazon/opendistroforelasticsearch/search/async/context/permits/AsyncSearchContextPermits.java

+
+    private void asyncAcquirePermit(
+            int permits, final ActionListener<Releasable> onAcquired, final TimeValue timeout, String reason) {
+        threadPool.executor(AsyncSearchPlugin.OPEN_DISTRO_ASYNC_SEARCH_GENERIC_THREAD_POOL_NAME).execute(new AbstractRunnable() {


Dedicated threadpool for different asyc search operations. Agree that can wait for more concrete results based on perf testing.

Surya Nistala added 4 commits November 9, 2020 15:01

Async search response tests

2b8bbfd

async search context and context permit. async search stage

db91f3b

move context related components into a dedicated context directory

a5f9580

persistence context

073d4a9

eirsep changed the title ~~Async search context~~ Async search context and and state management framework Nov 9, 2020

eirsep requested review from psychbot and getsaurabh02 November 9, 2020 10:49

Bukhtawar requested a review from itiyamas November 9, 2020 11:35

Surya Nistala added 3 commits November 9, 2020 17:20

active context

3c692c1

async search id tests. rename AsyncSearchContextId to AsyncSearchActi…

4e9663e

…veContextId

rename AsyncSearchActiveContextId to AsyncSearchContextId

5fff307

itiyamas suggested changes Nov 11, 2020

View reviewed changes

getsaurabh02 reviewed Nov 11, 2020

View reviewed changes

Surya Nistala added 3 commits November 12, 2020 12:17

added missing java docs. added assertions to async search stage advan…

8ab0c79

…cement and initialized stage in active context constructor.

minor fixes to async search context Permit

2faec54

revert releasable try-with-resources

1fc0a1e

psychbot reviewed Nov 13, 2020

View reviewed changes

Surya Nistala and others added 3 commits November 13, 2020 19:04

added asserts in advance stage method. misc. minor fixes

7a6800d

State machine changes

b11868f

added more async search transitions and events. pending fix in AsyncS…

ac9674d

…earchStateMachine

Bukhtawar requested a review from itiyamas November 18, 2020 10:41

Surya Nistala and others added 3 commits November 19, 2020 11:11

async search state machine fix for trigger method. Java docs added fo…

5863650

…r events

State machine wildcard capture changes

2212a49

Minor state machine changes

e69d4a5

itiyamas suggested changes Nov 20, 2020

View reviewed changes

Surya Nistala added 2 commits November 20, 2020 11:26

async search active context tests. asyn search state machine classes …

57010e4

…condensed and added synchronized block. added asyncsearchstatemachineException for illegal transition triggers

refactored state machine transitions to map

c42a15b

eirsep requested a review from getsaurabh02 November 20, 2020 08:42

Releasable Action Listener changes

0438193

Bukhtawar added 2 commits November 20, 2020 19:07

Async search converter and Closeable changes

d6358f2

Async search converter and Closeable changes tests

1a16c0a

Bukhtawar requested a review from itiyamas November 20, 2020 13:53

itiyamas reviewed Nov 23, 2020

View reviewed changes

itiyamas approved these changes Nov 23, 2020

View reviewed changes

getsaurabh02 approved these changes Nov 23, 2020

View reviewed changes

add license on files

5a3204b

Bukhtawar merged commit 51ee4ed into master Nov 23, 2020

Bukhtawar mentioned this pull request Nov 23, 2020

Asynchronous Search component breakdown #3

Closed

8 tasks

Bukhtawar deleted the async_search_context branch January 2, 2021 18:35

		throw new IllegalStateException("only next asyncSearchStage " + INIT +
		" is allowed when the current asyncSearchStage is null");

Async search context and and state management framework #6

Async search context and and state management framework #6

Conversation

eirsep commented Nov 9, 2020 • edited Loading

Issue #3

Description of changes

TODO

itiyamas left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eirsep Nov 11, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eirsep Nov 11, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

getsaurabh02 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

psychbot Nov 13, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

psychbot Nov 13, 2020 • edited Loading

Choose a reason for hiding this comment

psychbot left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eirsep commented Nov 9, 2020 •

edited

Loading

eirsep Nov 11, 2020 •

edited

Loading

eirsep Nov 11, 2020 •

edited

Loading

psychbot Nov 13, 2020 •

edited

Loading

psychbot Nov 13, 2020 •

edited

Loading