Line data    Source code

       1             : // Copyright (C) 2016 The Android Open Source Project
       2             : //
       3             : // Licensed under the Apache License, Version 2.0 (the "License");
       4             : // you may not use this file except in compliance with the License.
       5             : // You may obtain a copy of the License at
       6             : //
       7             : // http://www.apache.org/licenses/LICENSE-2.0
       8             : //
       9             : // Unless required by applicable law or agreed to in writing, software
      10             : // distributed under the License is distributed on an "AS IS" BASIS,
      11             : // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
      12             : // See the License for the specific language governing permissions and
      13             : // limitations under the License.
      14             : 
      15             : package com.google.gerrit.index;
      16             : 
      17             : import static java.nio.charset.StandardCharsets.UTF_8;
      18             : import static java.util.Objects.requireNonNull;
      19             : 
      20             : import com.google.auto.value.AutoValue;
      21             : import com.google.common.base.Stopwatch;
      22             : import com.google.common.flogger.FluentLogger;
      23             : import com.google.common.util.concurrent.ListenableFuture;
      24             : import com.google.common.util.concurrent.MoreExecutors;
      25             : import java.io.OutputStream;
      26             : import java.io.OutputStreamWriter;
      27             : import java.io.PrintWriter;
      28             : import java.util.concurrent.ExecutionException;
      29             : import java.util.concurrent.RejectedExecutionException;
      30             : import java.util.concurrent.TimeUnit;
      31             : import java.util.concurrent.atomic.AtomicBoolean;
      32             : import org.eclipse.jgit.lib.ProgressMonitor;
      33             : import org.eclipse.jgit.util.io.NullOutputStream;
      34             : 
      35             : /** Base class for implementations that can index all entities of a given type. */
      36         151 : public abstract class SiteIndexer<K, V, I extends Index<K, V>> {
      37         151 :   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
      38             : 
      39             :   /** Result of an operation to index a subset or all of the entities of a given type. */
      40             :   @AutoValue
      41          16 :   public abstract static class Result {
      42             :     public abstract long elapsedNanos();
      43             : 
      44             :     public abstract boolean success();
      45             : 
      46             :     public abstract int doneCount();
      47             : 
      48             :     public abstract int failedCount();
      49             : 
      50             :     public static Result create(Stopwatch sw, boolean success, int done, int failed) {
      51          16 :       return new AutoValue_SiteIndexer_Result(
      52          16 :           sw.elapsed(TimeUnit.NANOSECONDS), success, done, failed);
      53             :     }
      54             : 
      55             :     public long elapsed(TimeUnit timeUnit) {
      56          15 :       return timeUnit.convert(elapsedNanos(), TimeUnit.NANOSECONDS);
      57             :     }
      58             :   }
      59             : 
      60         151 :   protected int totalWork = -1;
      61         151 :   protected OutputStream progressOut = NullOutputStream.INSTANCE;
      62         151 :   protected PrintWriter verboseWriter = newPrintWriter(NullOutputStream.INSTANCE);
      63             : 
      64             :   public void setTotalWork(int num) {
      65          15 :     totalWork = num;
      66          15 :   }
      67             : 
      68             :   public void setProgressOut(OutputStream out) {
      69          15 :     progressOut = requireNonNull(out);
      70          15 :   }
      71             : 
      72             :   public void setVerboseOut(OutputStream out) {
      73          16 :     verboseWriter = newPrintWriter(requireNonNull(out));
      74          16 :   }
      75             : 
      76             :   /** Indexes all entities for the provided index. */
      77             :   public abstract Result indexAll(I index);
      78             : 
      79             :   protected final void addErrorListener(
      80             :       ListenableFuture<?> future, String desc, ProgressMonitor progress, AtomicBoolean ok) {
      81          16 :     future.addListener(
      82          16 :         new ErrorListener(future, desc, progress, ok), MoreExecutors.directExecutor());
      83          16 :   }
      84             : 
      85             :   protected PrintWriter newPrintWriter(OutputStream out) {
      86         151 :     return new PrintWriter(new OutputStreamWriter(out, UTF_8), true);
      87             :   }
      88             : 
      89             :   private static class ErrorListener implements Runnable {
      90             :     private final ListenableFuture<?> future;
      91             :     private final String desc;
      92             :     private final ProgressMonitor progress;
      93             :     private final AtomicBoolean ok;
      94             : 
      95             :     private ErrorListener(
      96          16 :         ListenableFuture<?> future, String desc, ProgressMonitor progress, AtomicBoolean ok) {
      97          16 :       this.future = future;
      98          16 :       this.desc = desc;
      99          16 :       this.progress = progress;
     100          16 :       this.ok = ok;
     101          16 :     }
     102             : 
     103             :     @Override
     104             :     public void run() {
     105             :       try {
     106          16 :         future.get();
     107           0 :       } catch (RejectedExecutionException e) {
     108             :         // Server shutdown, don't spam the logs.
     109           0 :         failSilently();
     110           0 :       } catch (ExecutionException | InterruptedException e) {
     111           0 :         fail(e);
     112           0 :       } catch (RuntimeException e) {
     113           0 :         failAndThrow(e);
     114           0 :       } catch (Error e) {
     115             :         // Can't join with RuntimeException because "RuntimeException |
     116             :         // Error" becomes Throwable, which messes with signatures.
     117           0 :         failAndThrow(e);
     118             :       } finally {
     119          16 :         synchronized (progress) {
     120          16 :           progress.update(1);
     121          16 :         }
     122             :       }
     123          16 :     }
     124             : 
     125             :     private void failSilently() {
     126           0 :       ok.set(false);
     127           0 :     }
     128             : 
     129             :     private void fail(Throwable t) {
     130           0 :       logger.atSevere().withCause(t).log("Failed to index %s", desc);
     131           0 :       ok.set(false);
     132           0 :     }
     133             : 
     134             :     private void failAndThrow(RuntimeException e) {
     135           0 :       fail(e);
     136           0 :       throw e;
     137             :     }
     138             : 
     139             :     private void failAndThrow(Error e) {
     140           0 :       fail(e);
     141           0 :       throw e;
     142             :     }
     143             :   }
     144             : }