hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTableResultScanner.java

   1 /**
   2  * Licensed to the Apache Software Foundation (ASF) under one
   3  * or more contributor license agreements.  See the NOTICE file
   4  * distributed with this work for additional information
   5  * regarding copyright ownership.  The ASF licenses this file
   6  * to you under the Apache License, Version 2.0 (the
   7  * "License"); you may not use this file except in compliance
   8  * with the License.  You may obtain a copy of the License at
   9  *
  10  *     http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  */
  18 package org.apache.hadoop.hbase.client;
  19
  20 import static org.apache.hadoop.hbase.client.ConnectionUtils.calcEstimatedSize;
  21
  22 import java.io.IOException;
  23 import java.io.InterruptedIOException;
  24 import java.util.ArrayDeque;
  25 import java.util.Queue;
  26 import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
  27 import org.apache.hadoop.hbase.util.FutureUtils;
  28 import org.apache.yetus.audience.InterfaceAudience;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31
  32 /**
  33  * The {@link ResultScanner} implementation for {@link AsyncTable}. It will fetch data automatically
  34  * in background and cache it in memory. Typically the {@link #maxCacheSize} will be
  35  * {@code 2 * scan.getMaxResultSize()}.
  36  */
  37 @InterfaceAudience.Private
  38 class AsyncTableResultScanner implements ResultScanner, AdvancedScanResultConsumer {
  39
  40   private static final Logger LOG = LoggerFactory.getLogger(AsyncTableResultScanner.class);
  41
  42   private final AsyncTable<AdvancedScanResultConsumer> rawTable;
  43
  44   private final long maxCacheSize;
  45
  46   private final Scan scan;
  47
  48   private final Queue<Result> queue = new ArrayDeque<>();
  49
  50   private ScanMetrics scanMetrics;
  51
  52   private long cacheSize;
  53
  54   private boolean closed = false;
  55
  56   private Throwable error;
  57
  58   private ScanResumer resumer;
  59
  60   public AsyncTableResultScanner(AsyncTable<AdvancedScanResultConsumer> table, Scan scan,
  61       long maxCacheSize) {
  62     this.rawTable = table;
  63     this.maxCacheSize = maxCacheSize;
  64     this.scan = scan;
  65     table.scan(scan, this);
  66   }
  67
  68   private void addToCache(Result result) {
  69     queue.add(result);
  70     cacheSize += calcEstimatedSize(result);
  71   }
  72
  73   private void stopPrefetch(ScanController controller) {
  74     if (LOG.isDebugEnabled()) {
  75       LOG.debug(String.format("0x%x", System.identityHashCode(this)) +
  76         " stop prefetching when scanning " + rawTable.getName() + " as the cache size " +
  77         cacheSize + " is greater than the maxCacheSize " + maxCacheSize);
  78     }
  79     resumer = controller.suspend();
  80   }
  81
  82   @Override
  83   public synchronized void onNext(Result[] results, ScanController controller) {
  84     assert results.length > 0;
  85     if (closed) {
  86       controller.terminate();
  87       return;
  88     }
  89     for (Result result : results) {
  90       addToCache(result);
  91     }
  92     notifyAll();
  93     if (cacheSize >= maxCacheSize) {
  94       stopPrefetch(controller);
  95     }
  96   }
  97
  98   @Override
  99   public synchronized void onHeartbeat(ScanController controller) {
 100     if (closed) {
 101       controller.terminate();
 102       return;
 103     }
 104     if (scan.isNeedCursorResult()) {
 105       controller.cursor().ifPresent(c -> queue.add(Result.createCursorResult(c)));
 106     }
 107   }
 108
 109   @Override
 110   public synchronized void onError(Throwable error) {
 111     this.error = error;
 112     notifyAll();
 113   }
 114
 115   @Override
 116   public synchronized void onComplete() {
 117     closed = true;
 118     notifyAll();
 119   }
 120
 121   @Override
 122   public void onScanMetricsCreated(ScanMetrics scanMetrics) {
 123     this.scanMetrics = scanMetrics;
 124   }
 125
 126   private void resumePrefetch() {
 127     if (LOG.isDebugEnabled()) {
 128       LOG.debug(String.format("0x%x", System.identityHashCode(this)) + " resume prefetching");
 129     }
 130     resumer.resume();
 131     resumer = null;
 132   }
 133
 134   @Override
 135   public synchronized Result next() throws IOException {
 136     while (queue.isEmpty()) {
 137       if (closed) {
 138         return null;
 139       }
 140       if (error != null) {
 141         FutureUtils.rethrow(error);
 142       }
 143       try {
 144         wait();
 145       } catch (InterruptedException e) {
 146         throw new InterruptedIOException();
 147       }
 148     }
 149     Result result = queue.poll();
 150     if (!result.isCursor()) {
 151       cacheSize -= calcEstimatedSize(result);
 152       if (resumer != null && cacheSize <= maxCacheSize / 2) {
 153         resumePrefetch();
 154       }
 155     }
 156     return result;
 157   }
 158
 159   @Override
 160   public synchronized void close() {
 161     closed = true;
 162     queue.clear();
 163     cacheSize = 0;
 164     if (resumer != null) {
 165       resumePrefetch();
 166     }
 167     notifyAll();
 168   }
 169
 170   @Override
 171   public boolean renewLease() {
 172     // we will do prefetching in the background and if there is no space we will just suspend the
 173     // scanner. The renew lease operation will be handled in the background.
 174     return false;
 175   }
 176
 177   // used in tests to test whether the scanner has been suspended
 178   synchronized boolean isSuspended() {
 179     return resumer != null;
 180   }
 181
 182   @Override
 183   public ScanMetrics getScanMetrics() {
 184     return scanMetrics;
 185   }
 186
 187   int getCacheSize() {
 188     return queue.size();
 189   }
 190 }