001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.mob;
019
020import java.io.IOException;
021import java.util.ArrayList;
022import java.util.Collections;
023import java.util.List;
024import java.util.Map;
025import java.util.concurrent.ConcurrentHashMap;
026import java.util.concurrent.Executors;
027import java.util.concurrent.ScheduledExecutorService;
028import java.util.concurrent.TimeUnit;
029import java.util.concurrent.atomic.AtomicLong;
030import java.util.concurrent.atomic.LongAdder;
031import java.util.concurrent.locks.ReentrantLock;
032import org.apache.hadoop.conf.Configuration;
033import org.apache.hadoop.fs.FileSystem;
034import org.apache.hadoop.fs.Path;
035import org.apache.hadoop.hbase.io.hfile.CacheConfig;
036import org.apache.hadoop.hbase.util.IdLock;
037import org.apache.yetus.audience.InterfaceAudience;
038import org.slf4j.Logger;
039import org.slf4j.LoggerFactory;
040
041import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
042
043/**
044 * The cache for mob files. This cache doesn't cache the mob file blocks. It only caches the
045 * references of mob files. We are doing this to avoid opening and closing mob files all the time.
046 * We just keep references open.
047 */
048@InterfaceAudience.Private
049public class MobFileCache {
050
051  private static final Logger LOG = LoggerFactory.getLogger(MobFileCache.class);
052
053  /*
054   * Eviction and statistics thread. Periodically run to print the statistics and evict the lru
055   * cached mob files when the count of the cached files is larger than the threshold.
056   */
057  static class EvictionThread extends Thread {
058    MobFileCache lru;
059
060    public EvictionThread(MobFileCache lru) {
061      super("MobFileCache.EvictionThread");
062      setDaemon(true);
063      this.lru = lru;
064    }
065
066    @Override
067    public void run() {
068      lru.evict();
069    }
070  }
071
072  // a ConcurrentHashMap, accesses to this map are synchronized.
073  private Map<String, CachedMobFile> map = null;
074  // caches access count
075  private final AtomicLong count = new AtomicLong(0);
076  private long lastAccess = 0;
077  private final LongAdder miss = new LongAdder();
078  private long lastMiss = 0;
079  private final LongAdder evictedFileCount = new LongAdder();
080  private long lastEvictedFileCount = 0;
081
082  // a lock to sync the evict to guarantee the eviction occurs in sequence.
083  // the method evictFile is not sync by this lock, the ConcurrentHashMap does the sync there.
084  private final ReentrantLock evictionLock = new ReentrantLock(true);
085
086  // stripes lock on each mob file based on its hash. Sync the openFile/closeFile operations.
087  private final IdLock keyLock = new IdLock();
088
089  private final ScheduledExecutorService scheduleThreadPool = Executors.newScheduledThreadPool(1,
090    new ThreadFactoryBuilder().setNameFormat("MobFileCache #%d").setDaemon(true).build());
091  private final Configuration conf;
092
093  // the count of the cached references to mob files
094  private final int mobFileMaxCacheSize;
095  private final boolean isCacheEnabled;
096  private float evictRemainRatio;
097
098  public MobFileCache(Configuration conf) {
099    this.conf = conf;
100    this.mobFileMaxCacheSize =
101      conf.getInt(MobConstants.MOB_FILE_CACHE_SIZE_KEY, MobConstants.DEFAULT_MOB_FILE_CACHE_SIZE);
102    isCacheEnabled = (mobFileMaxCacheSize > 0);
103    map = new ConcurrentHashMap<>(mobFileMaxCacheSize);
104    if (isCacheEnabled) {
105      long period = conf.getLong(MobConstants.MOB_CACHE_EVICT_PERIOD,
106        MobConstants.DEFAULT_MOB_CACHE_EVICT_PERIOD); // in seconds
107      evictRemainRatio = conf.getFloat(MobConstants.MOB_CACHE_EVICT_REMAIN_RATIO,
108        MobConstants.DEFAULT_EVICT_REMAIN_RATIO);
109      if (evictRemainRatio < 0.0) {
110        evictRemainRatio = 0.0f;
111        LOG.warn(MobConstants.MOB_CACHE_EVICT_REMAIN_RATIO + " is less than 0.0, 0.0 is used.");
112      } else if (evictRemainRatio > 1.0) {
113        evictRemainRatio = 1.0f;
114        LOG.warn(MobConstants.MOB_CACHE_EVICT_REMAIN_RATIO + " is larger than 1.0, 1.0 is used.");
115      }
116      this.scheduleThreadPool.scheduleAtFixedRate(new EvictionThread(this), period, period,
117        TimeUnit.SECONDS);
118
119      if (LOG.isDebugEnabled()) {
120        LOG.debug("MobFileCache enabled with cacheSize=" + mobFileMaxCacheSize + ", evictPeriods="
121          + period + "sec, evictRemainRatio=" + evictRemainRatio);
122      }
123    } else {
124      LOG.info("MobFileCache disabled");
125    }
126  }
127
128  /**
129   * Evicts the lru cached mob files when the count of the cached files is larger than the
130   * threshold.
131   */
132  public void evict() {
133    if (isCacheEnabled) {
134      // Ensure only one eviction at a time
135      if (!evictionLock.tryLock()) {
136        return;
137      }
138      printStatistics();
139      List<CachedMobFile> evictedFiles = new ArrayList<>();
140      try {
141        if (map.size() <= mobFileMaxCacheSize) {
142          return;
143        }
144        List<CachedMobFile> files = new ArrayList<>(map.values());
145        Collections.sort(files);
146        int start = (int) (mobFileMaxCacheSize * evictRemainRatio);
147        if (start >= 0) {
148          for (int i = start; i < files.size(); i++) {
149            String name = files.get(i).getFileName();
150            CachedMobFile evictedFile = map.remove(name);
151            if (evictedFile != null) {
152              evictedFiles.add(evictedFile);
153            }
154          }
155        }
156      } finally {
157        evictionLock.unlock();
158      }
159      // EvictionLock is released. Close the evicted files one by one.
160      // The closes are sync in the closeFile method.
161      for (CachedMobFile evictedFile : evictedFiles) {
162        closeFile(evictedFile);
163      }
164      evictedFileCount.add(evictedFiles.size());
165    }
166  }
167
168  /**
169   * Evicts the cached file by the name.
170   * @param fileName The name of a cached file.
171   */
172  public void evictFile(String fileName) {
173    if (isCacheEnabled) {
174      IdLock.Entry lockEntry = null;
175      try {
176        // obtains the lock to close the cached file.
177        lockEntry = keyLock.getLockEntry(fileName.hashCode());
178        CachedMobFile evictedFile = map.remove(fileName);
179        if (evictedFile != null) {
180          evictedFile.close();
181          evictedFileCount.increment();
182        }
183      } catch (IOException e) {
184        LOG.error("Failed to evict the file " + fileName, e);
185      } finally {
186        if (lockEntry != null) {
187          keyLock.releaseLockEntry(lockEntry);
188        }
189      }
190    }
191  }
192
193  /**
194   * Opens a mob file.
195   * @param fs        The current file system.
196   * @param path      The file path.
197   * @param cacheConf The current MobCacheConfig
198   * @return A opened mob file. n
199   */
200  public MobFile openFile(FileSystem fs, Path path, CacheConfig cacheConf) throws IOException {
201    if (!isCacheEnabled) {
202      MobFile mobFile = MobFile.create(fs, path, conf, cacheConf);
203      mobFile.open();
204      return mobFile;
205    } else {
206      String fileName = path.getName();
207      CachedMobFile cached = map.get(fileName);
208      IdLock.Entry lockEntry = keyLock.getLockEntry(fileName.hashCode());
209      try {
210        if (cached == null) {
211          cached = map.get(fileName);
212          if (cached == null) {
213            if (map.size() > mobFileMaxCacheSize) {
214              evict();
215            }
216            cached = CachedMobFile.create(fs, path, conf, cacheConf);
217            cached.open();
218            map.put(fileName, cached);
219            miss.increment();
220          }
221        }
222        cached.open();
223        cached.access(count.incrementAndGet());
224      } finally {
225        keyLock.releaseLockEntry(lockEntry);
226      }
227      return cached;
228    }
229  }
230
231  /**
232   * Closes a mob file.
233   * @param file The mob file that needs to be closed.
234   */
235  public void closeFile(MobFile file) {
236    IdLock.Entry lockEntry = null;
237    try {
238      if (!isCacheEnabled) {
239        file.close();
240      } else {
241        lockEntry = keyLock.getLockEntry(file.getFileName().hashCode());
242        file.close();
243      }
244    } catch (IOException e) {
245      LOG.error("MobFileCache, Exception happen during close " + file.getFileName(), e);
246    } finally {
247      if (lockEntry != null) {
248        keyLock.releaseLockEntry(lockEntry);
249      }
250    }
251  }
252
253  public void shutdown() {
254    this.scheduleThreadPool.shutdown();
255    for (int i = 0; i < 100; i++) {
256      if (!this.scheduleThreadPool.isShutdown()) {
257        try {
258          Thread.sleep(10);
259        } catch (InterruptedException e) {
260          LOG.warn("Interrupted while sleeping");
261          Thread.currentThread().interrupt();
262          break;
263        }
264      }
265    }
266
267    if (!this.scheduleThreadPool.isShutdown()) {
268      List<Runnable> runnables = this.scheduleThreadPool.shutdownNow();
269      LOG.debug("Still running " + runnables);
270    }
271  }
272
273  /**
274   * Gets the count of cached mob files.
275   * @return The count of the cached mob files.
276   */
277  public int getCacheSize() {
278    return map == null ? 0 : map.size();
279  }
280
281  /**
282   * Gets the count of accesses to the mob file cache.
283   * @return The count of accesses to the mob file cache.
284   */
285  public long getAccessCount() {
286    return count.get();
287  }
288
289  /**
290   * Gets the count of misses to the mob file cache.
291   * @return The count of misses to the mob file cache.
292   */
293  public long getMissCount() {
294    return miss.sum();
295  }
296
297  /**
298   * Gets the number of items evicted from the mob file cache.
299   * @return The number of items evicted from the mob file cache.
300   */
301  public long getEvictedFileCount() {
302    return evictedFileCount.sum();
303  }
304
305  /**
306   * Gets the hit ratio to the mob file cache.
307   * @return The hit ratio to the mob file cache.
308   */
309  public double getHitRatio() {
310    return count.get() == 0 ? 0 : ((float) (count.get() - miss.sum())) / (float) count.get();
311  }
312
313  /**
314   * Prints the statistics.
315   */
316  public void printStatistics() {
317    long access = count.get() - lastAccess;
318    long missed = miss.sum() - lastMiss;
319    long evicted = evictedFileCount.sum() - lastEvictedFileCount;
320    int hitRatio = access == 0 ? 0 : (int) (((float) (access - missed)) / (float) access * 100);
321    LOG.info("MobFileCache Statistics, access: " + access + ", miss: " + missed + ", hit: "
322      + (access - missed) + ", hit ratio: " + hitRatio + "%, evicted files: " + evicted);
323    lastAccess += access;
324    lastMiss += missed;
325    lastEvictedFileCount += evicted;
326  }
327
328}