// Copyright 2017 The Bazel Authors. All rights reserved.
//
// Licensed 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 com.google.devtools.build.lib.query2.engine;

import com.google.common.base.Function;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
import com.google.common.util.concurrent.AsyncFunction;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.devtools.build.lib.query2.engine.QueryEnvironment.QueryTaskCallable;
import com.google.devtools.build.lib.query2.engine.QueryEnvironment.QueryTaskFuture;
import com.google.devtools.build.lib.util.Preconditions;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * A partial implementation of {@link QueryEnvironment} that has trivial in-thread implementations
 * of all the {@link QueryTaskFuture}/{@link QueryTaskCallable} helper methods.
 */
public abstract class AbstractQueryEnvironment<T> implements QueryEnvironment<T> {
  /** Concrete implementation of {@link QueryTaskFuture}. */
  protected static final class QueryTaskFutureImpl<T>
      extends QueryTaskFutureImplBase<T> implements ListenableFuture<T> {
    private final ListenableFuture<T> delegate;

    private QueryTaskFutureImpl(ListenableFuture<T> delegate) {
      this.delegate = delegate;
    }

    public static <R> QueryTaskFutureImpl<R> ofDelegate(ListenableFuture<R> delegate) {
      return (delegate instanceof QueryTaskFutureImpl)
          ? (QueryTaskFutureImpl<R>) delegate
          : new QueryTaskFutureImpl<>(delegate);
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
      return delegate.cancel(mayInterruptIfRunning);
    }

    @Override
    public boolean isCancelled() {
      return delegate.isCancelled();
    }

    @Override
    public boolean isDone() {
      return delegate.isDone();
    }

    @Override
    public T get() throws InterruptedException, ExecutionException {
      return delegate.get();
    }

    @Override
    public T get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
      return delegate.get(timeout, unit);
    }

    @Override
    public void addListener(Runnable listener, Executor executor) {
      delegate.addListener(listener, executor);
    }

    @Override
    public T getIfSuccessful() {
      Preconditions.checkState(delegate.isDone());
      try {
        return delegate.get();
      } catch (CancellationException | InterruptedException | ExecutionException e) {
        throw new IllegalStateException(e);
      }
    }

    public T getChecked() throws InterruptedException, QueryException {
      try {
        return get();
      } catch (CancellationException e) {
        throw new InterruptedException();
      } catch (ExecutionException e) {
        Throwable cause = e.getCause();
        Throwables.propagateIfPossible(cause, QueryException.class);
        Throwables.propagateIfPossible(cause, InterruptedException.class);
        throw new IllegalStateException(e.getCause());
      }
    }
  }

  @Override
  public <R> QueryTaskFuture<R> immediateSuccessfulFuture(R value) {
    return new QueryTaskFutureImpl<>(Futures.immediateFuture(value));
  }

  @Override
  public <R> QueryTaskFuture<R> immediateFailedFuture(QueryException e) {
    return new QueryTaskFutureImpl<>(Futures.<R>immediateFailedFuture(e));
  }

  @Override
  public <R> QueryTaskFuture<R> immediateCancelledFuture() {
    return new QueryTaskFutureImpl<>(Futures.<R>immediateCancelledFuture());
  }

  @Override
  public QueryTaskFuture<Void> eval(
      QueryExpression expr, VariableContext<T> context, Callback<T> callback) {
    return expr.eval(this, context, callback);
  }

  @Override
  public <R> QueryTaskFuture<R> executeAsync(QueryTaskCallable<R> callable) {
    try {
      return immediateSuccessfulFuture(callable.call());
    } catch (QueryException e) {
      return immediateFailedFuture(e);
    } catch (InterruptedException e) {
      return immediateCancelledFuture();
    }
  }

  @Override
  public <R> QueryTaskFuture<R> whenSucceedsCall(
      QueryTaskFuture<?> future, QueryTaskCallable<R> callable) {
    return whenAllSucceedCall(ImmutableList.of(future), callable);
  }

  private static class Dummy implements QueryTaskCallable<Void> {
    public static final Dummy INSTANCE = new Dummy();

    private Dummy() {}

    @Override
    public Void call() {
      return null;
    }
  }

  @Override
  public QueryTaskFuture<Void> whenAllSucceed(Iterable<? extends QueryTaskFuture<?>> futures) {
    return whenAllSucceedCall(futures, Dummy.INSTANCE);
  }

  @Override
  public <R> QueryTaskFuture<R> whenAllSucceedCall(
      Iterable<? extends QueryTaskFuture<?>> futures, QueryTaskCallable<R> callable) {
    return QueryTaskFutureImpl.ofDelegate(
        Futures.whenAllSucceed(cast(futures)).call(callable));
  }

  @Override
  public <T1, T2> QueryTaskFuture<T2> transformAsync(
      QueryTaskFuture<T1> future,
      final Function<T1, QueryTaskFuture<T2>> function) {
    return QueryTaskFutureImpl.ofDelegate(
        Futures.transformAsync(
            (QueryTaskFutureImpl<T1>) future,
            new AsyncFunction<T1, T2>() {
              @Override
              public ListenableFuture<T2> apply(T1 input) throws Exception {
                return (QueryTaskFutureImpl<T2>) function.apply(input);
              }
            }));
  }

  protected static Iterable<QueryTaskFutureImpl<?>> cast(
      Iterable<? extends QueryTaskFuture<?>> futures) {
    return Iterables.transform(
        futures,
        new Function<QueryTaskFuture<?>, QueryTaskFutureImpl<?>>() {
          @Override
          public QueryTaskFutureImpl<?> apply(QueryTaskFuture<?> future) {
            return (QueryTaskFutureImpl<?>) future;
          }
        });
  }
}

@@ -4412,16 +4412,22 @@
 context,
+ final
  Callbac
@@ -4451,48 +4451,967 @@
     
-return expr.eval(this, context, callback
+// Not all QueryEnvironment implementations embrace the async+streaming evaluation framework. In%0A    // particular, the streaming callbacks employed by functions like 'deps' use%0A    // QueryEnvironment#buildTransitiveClosure. So if the implementation of that method does some%0A    // heavyweight blocking work, then it's best to do this blocking work in a single batch.%0A    // Importantly, the callback we pass in needs to maintain order.%0A    final QueryUtil.AggregateAllCallback%3CT%3E aggregateAllCallback =%0A        QueryUtil.newOrderedAggregateAllOutputFormatterCallback();%0A    QueryTaskFuture%3CVoid%3E evalAllFuture = expr.eval(this, context, aggregateAllCallback);%0A    return whenSucceedsCall(%0A        evalAllFuture,%0A        new QueryTaskCallable%3CVoid%3E() %7B%0A          @Override%0A          public Void call() throws QueryException, InterruptedException %7B%0A            callback.process(aggregateAllCallback.getResult());%0A            return null;%0A          %7D%0A        %7D
 );%0A 

Fix inadvertent performance regression introduced by the recent rewrite of 'blaze query'.