content: /*
   * 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 accord.coordinate;
  
  import java.util.Objects;
  import java.util.function.BiConsumer;
  import java.util.function.BiFunction;
  import java.util.function.Function;
  import java.util.function.Predicate;
  
  import javax.annotation.Nonnull;
  import javax.annotation.Nullable;
  
  import accord.coordinate.tracking.AbstractTracker;
  import accord.local.Node;
  import accord.local.SequentialAsyncExecutor;
  import accord.messages.Callback;
  import accord.messages.Request;
  import accord.primitives.Route;
  import accord.primitives.TxnId;
  import accord.utils.DebugMap;
  import accord.utils.Invariants;
  import accord.utils.SimpleBitSet;
  import accord.utils.SortedArrays.SortedArrayList;
  import accord.utils.SortedList;
  import accord.utils.SortedListMap;
  import accord.utils.SortedListSet;
  import accord.utils.WrappableException;
  import accord.utils.async.AsyncChain;
  import accord.utils.async.Cancellable;
  
  // TODO (expected): move message sending here, so we can invalidate any pending callbacks when we advance the state machine
  public abstract class AbstractCoordination<Result, Reply extends accord.messages.Reply, Ok> extends AbstractSimpleCoordination implements Callback<Reply>
  {
      private BiConsumer<? super Result, Throwable> callback;
      private final SortedArrayList<Node.Id> nodes;
      private final SimpleBitSet expectingReply;
      private Object[] replyState;
      private final @Nullable DebugMap debug;
      private int replyCount;
  
      protected AbstractCoordination(Node node, SequentialAsyncExecutor executor, TxnId txnId, SortedArrayList<Node.Id> nodes, BiConsumer<? super Result, Throwable> callback)
      {
          super(node, executor, txnId);
          this.nodes = nodes;
          this.callback = Invariants.nonNull(callback);
          this.replyState = new Object[nodes.size()];
          this.expectingReply = SimpleBitSet.allocate(nodes.size());
          this.debug = Invariants.debug() ? new DebugMap(nodes) : null;
      }
  
      abstract void onSuccessInternal(Node.Id from, int fromIndex, Reply reply);
      abstract void onFailureInternal(Node.Id from, int fromIndex, Throwable fail);
      void onSlowResponseInternal(Node.Id from) {}
  
      void recordOk(int fromIndex, Ok ok)
      {
          Invariants.require(replyState[fromIndex] == null, "%s", this);
          replyState[fromIndex] = ok;
          replyCount++;
      }
  
      SortedListMap<Node.Id, Ok> finishOks()
      {
          Invariants.require(replyState != null, "%s", this);
          setDoneWithReplies();
          for (int i = expectingReply.nextSetBit(0) ; i >= 0 ; i = expectingReply.nextSetBit(i + 1))
          {
              Object cancel = replyState[i];
              if (cancel != null)
              {
                  ((Cancellable)cancel).cancel();
                  replyState[i] = null;
              }
          }
          expectingReply.clear();
          SortedListMap<Node.Id, Ok> result = new SortedListMap<>(nodes, replyState, replyCount);
          replyState = null;
          return result;
      }
  
      <V> V foldlOks(BiFunction<Ok, V, V> foldl, V zero)
      {
          V result = zero;
          for (int i = 0; i < replyState.length ; ++i)
          {
              if (replyState[i] != null && !expectingReply.get(i))
                  result = foldl.apply((Ok)replyState[i], result);
          }
          return result;
      }
  
      void finishWithSuccess(Result success)
      {
          finishAndInvokeCallback(success, null);
      }
  
      void finishWithFailureOverride(Throwable failure)
      {
          finishAndInvokeCallback(null, FailureAccumulator.append(failure, this.failure()));
      }
  
      void finishOnExaustion()
      {
          finishOnFailure();
      }
  
      void finishOnFailure()
      {
          finishAndInvokeCallback(null, FailureAccumulator.fail(node.agent(), this.failure(), txnId, Route.tryCastToRoute(scope())));
      }
  
      void awaitEpochExactToFinish(long epoch, Runnable runnable)
      {
          setFinishing();
          node.withEpochExact(epoch, executor, (ignore, failure) -> finishWithFailureOverride(failure), WrappableException::wrap, () -> {
              runnable.run();
              Invariants.require(isDone(), "%s", this);
          });
      }
  
      void awaitEpochAtLeastToFinish(long epoch, Runnable runnable)
      {
          setFinishing();
          node.withEpochAtLeast(epoch, executor, (ignore, failure) -> finishWithFailureOverride(failure), WrappableException::wrap, () -> {
              runnable.run();
              Invariants.require(isDone(), "%s", this);
          });
      }
  
      void awaitToFinish(AsyncChain<?> await)
      {
          setFinishing();
          await.begin((success, fail) -> {
              if (fail != null) finishWithFailureOverride(fail);
              else Invariants.require(isDone(), "%s", this);
          });
      }
  
      void markSelfContacted()
      {
          expectingReply.set(nodes.find(node.id()));
      }
  
      void contact(Function<Node.Id, Request> request)
      {
          contact(request, null);
      }
  
      void contact(Function<Node.Id, Request> request, @Nullable Predicate<Node.Id> include)
      {
          executor.executeMaybeImmediately(() -> {
              for (int i = 0; i < nodes.size() ; ++i)
              {
                  Node.Id to = nodes.get(i);
                  if (include == null || include.test(to))
                  {
                      Invariants.require(replyState[i] == null);
                      expectingReply.set(i);
                      replyState[i] = node.send(to, request.apply(to), executor, this);
                      Invariants.require(expectingReply.get(i) || replyState[i] == null);
                  }
              }
          });
      }
  
      @Override
      public void onSuccess(Node.Id from, Reply reply)
      {
          int fromIndex = onReply(from, reply, reply.isFinal());
          if (fromIndex < 0)
              return;
  
          onSuccessInternal(from, fromIndex, reply);
          Invariants.require(!expectingReply.isEmpty() || isFinishing() || isDone(), "%s", this);
      }
  
      @Override
      public final void onSlowResponse(Node.Id from)
      {
          if (!isDoneWithReplies())
              onSlowResponseInternal(from);
      }
  
      @Override
      public void onFailure(Node.Id from, @Nullable Throwable failure)
      {
          int fromIndex = onReply(from, failure, true);
          if (fromIndex < 0)
              return;
  
          onFailureInternal(from, fromIndex, failure);
          Invariants.require(!expectingReply.isEmpty() || isFinishing() || isDone(), "%s", this);
      }
  
      private int onReply(Node.Id from, Object reply, boolean isFinal)
      {
          int fromIndex = nodes.find(from);
          if (isDoneWithReplies())
          {
              if (isFinal && replyState != null)
                  replyState[fromIndex] = null;
              return -1;
          }
  
          if (debug != null)
              debug.debug(from, reply);
  
          if (isFinal)
          {
              boolean expecting = expectingReply.unset(fromIndex);
              Invariants.require(expecting, "%s", this);
              replyState[fromIndex] = null;
          }
          else
          {
              boolean expecting = expectingReply.get(fromIndex);
              Invariants.require(expecting, "%s", this);
          }
          return fromIndex;
      }
  
      @Override
      public boolean onCallbackFailure(Node.Id from, Throwable failure)
      {
          if (isDone())
              return false;
  
          setDone();
          BiConsumer<?, Throwable> callback = tryTakeCallback();
          if (callback != null) callback.accept(null, failure);
          else node.agent().onUncaughtException(failure);
          return true;
      }
  
      void finishAndInvokeCallback(Result success, Throwable failure)
      {
          finishAndTakeCallback().accept(success, failure);
      }
  
      BiConsumer<? super Result, Throwable> finishAndTakeCallback()
      {
          setDone();
          return takeCallback();
      }
  
      private BiConsumer<? super Result, Throwable> takeCallback()
      {
          BiConsumer<? super Result, Throwable> callback = this.callback;
          this.callback = null;
          Invariants.require(callback != null, "%s", this);
          return callback;
      }
  
      private BiConsumer<? super Result, Throwable> tryTakeCallback()
      {
          if (callback == null)
              return null;
          return takeCallback();
      }
  
      @Override
      public boolean abort()
      {
          if (isDone())
              return false;
  
          finishWithFailureOverride(Aborted.aborted(txnId, Route.tryCastToRoute(scope())));
          return true;
      }
  
      @Override
      public final SortedListMap<Node.Id, ?> replies()
      {
          Object[] replyState = this.replyState;
          if (replyState == null)
              return null;
  
          Object[] replies = replyState.clone();
          for (int i = 0 ; i < replies.length ; ++i)
          {
              if (replies[i] != null && !(replies[i] instanceof accord.messages.Reply))
                  replies[i] = null;
          }
          return new SortedListMap<>(nodes, replies, replyCount);
      }
  
      @Override
      public SortedList<Node.Id> inflight()
      {
          SortedListSet<Node.Id> build = SortedListSet.noneOf(nodes);
          for (int i = expectingReply.nextSetBit(0) ; i >= 0 ; i = expectingReply.nextSetBit(i + 1))
              build.addIndex(i);
          return SortedArrayList.copySorted(build, Node.Id[]::new);
      }
  
      @Override
      public String toString()
      {
          String describe = describe();
          AbstractTracker<?> tracker = tracker();
          SortedListMap<Node.Id, ?> replies = replies();
          return kind().name() + ':' + txnId
                 + "scope:" + scope()
                 + " inflight:" + inflight()
                 + (tracker == null ? "" : " tracker:" + tracker.summariseTracker())
                 + (describe.isEmpty() ? "" : ' ' + describe)
                 + (replies == null ? "" : " replies:" + summariseReplies(replies, 60));
      }
  
      public static String summariseReplies(@Nonnull SortedListMap<Node.Id, ?> replies, int maxReplyLength)
      {
          StringBuilder sb = new StringBuilder();
          boolean first = true;
          for (int i = 0 ; i < replies.domainSize() ; ++i)
          {
              Object value = replies.getValue(i);
              if (value == null) continue;
              Node.Id key = replies.getKey(i);
              if (first) first = false;
              else sb.append('\n');
              sb.append(key);
              sb.append('=');
              String v = Objects.toString(value);
              if (v.length() > maxReplyLength)
                  v = v.substring(0, maxReplyLength) + "...";
              sb.append(v);
          }
          return sb.toString();
      }
  }
  