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 org.apache.bookkeeper.bookie;
  
  import static java.lang.Long.max;
  import static org.apache.bookkeeper.bookie.BookKeeperServerStats.LEDGER_CACHE_HIT;
  import static org.apache.bookkeeper.bookie.BookKeeperServerStats.LEDGER_CACHE_MISS;
  import static org.apache.bookkeeper.bookie.BookKeeperServerStats.LEDGER_CACHE_READ_PAGE;
  import static org.apache.bookkeeper.bookie.BookKeeperServerStats.NUM_INDEX_PAGES;
  
  // CHECKSTYLE.OFF: IllegalImport
  import com.google.common.base.Stopwatch;
  import io.netty.util.internal.PlatformDependent;
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.Iterator;
  import java.util.LinkedHashMap;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.ConcurrentLinkedQueue;
  import java.util.concurrent.ConcurrentMap;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicInteger;
  import org.apache.bookkeeper.bookie.stats.IndexInMemPageMgrStats;
  import org.apache.bookkeeper.conf.ServerConfiguration;
  import org.apache.bookkeeper.stats.Counter;
  import org.apache.bookkeeper.stats.Gauge;
  import org.apache.bookkeeper.stats.OpStatsLogger;
  import org.apache.bookkeeper.stats.StatsLogger;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  // CHECKSTYLE.ON: IllegalImport
  
  class IndexInMemPageMgr {
      private static final Logger LOG = LoggerFactory.getLogger(IndexInMemPageMgr.class);
      private static final ConcurrentHashMap<Long, LedgerEntryPage> EMPTY_PAGE_MAP =
              new ConcurrentHashMap<Long, LedgerEntryPage>();
  
      private static class InMemPageCollection implements LEPStateChangeCallback {
  
          final ConcurrentMap<Long, ConcurrentMap<Long, LedgerEntryPage>> pages;
          final Map<EntryKey, LedgerEntryPage> lruCleanPageMap;
          final ConcurrentLinkedQueue<LedgerEntryPage> listOfFreePages;
  
          // Stats
          private final IndexInMemPageMgrStats inMemPageMgrStats;
  
          public InMemPageCollection(StatsLogger statsLogger) {
              pages = new ConcurrentHashMap<>();
              lruCleanPageMap =
                      Collections.synchronizedMap(new LinkedHashMap<EntryKey, LedgerEntryPage>(16, 0.75f, true));
              listOfFreePages = new ConcurrentLinkedQueue<LedgerEntryPage>();
              inMemPageMgrStats = new IndexInMemPageMgrStats(statsLogger);
          }
  
          /**
           * Retrieve the LedgerEntryPage corresponding to the ledger and firstEntry.
           *
           * @param ledgerId Ledger id
           * @param firstEntry Id of the first entry in the page
           * @returns LedgerEntryPage if present
           */
          private LedgerEntryPage getPage(long ledgerId, long firstEntry) {
              ConcurrentMap<Long, LedgerEntryPage> map = pages.get(ledgerId);
              if (null != map) {
                  return map.get(firstEntry);
              }
              return null;
          }
  
          /**
           * Add a LedgerEntryPage to the page map.
           *
           * @param lep Ledger Entry Page object
           */
          private LedgerEntryPage putPage(LedgerEntryPage lep) {
              // Do a get here to avoid too many new ConcurrentHashMaps() as putIntoTable is called frequently.
              ConcurrentMap<Long, LedgerEntryPage> map = pages.get(lep.getLedger());
              if (null == map) {
                  ConcurrentMap<Long, LedgerEntryPage> mapToPut = new ConcurrentHashMap<Long, LedgerEntryPage>();
                  map = pages.putIfAbsent(lep.getLedger(), mapToPut);
                  if (null == map) {
                      map = mapToPut;
                  }
              }
              LedgerEntryPage oldPage = map.putIfAbsent(lep.getFirstEntry(), lep);
              if (null == oldPage) {
                  oldPage = lep;
                  // Also include this in the clean page map if it qualifies.
                  // Note: This is done for symmetry and correctness, however it should never
                  // get exercised since we shouldn't attempt a put without the page being in use
                  addToCleanPagesList(lep);
              }
              return oldPage;
          }
  
          /**
           * Traverse the pages for a given ledger in memory and find the highest entry amongst these pages.
           *
           * @param ledgerId Ledger id
           * @returns last entry in the in memory pages
           */
          private long getLastEntryInMem(long ledgerId) {
              long lastEntry = 0;
              // Find the last entry in the cache
              ConcurrentMap<Long, LedgerEntryPage> map = pages.get(ledgerId);
              if (map != null) {
                  for (LedgerEntryPage lep: map.values()) {
                      if (lep.getMaxPossibleEntry() < lastEntry) {
                          continue;
                      }
                      lep.usePage();
                      long highest = lep.getLastEntry();
                      if (highest > lastEntry) {
                          lastEntry = highest;
                      }
                      lep.releasePage();
                  }
              }
              return lastEntry;
          }
  
          /**
           * Removes ledger entry pages for a given ledger.
           *
           * @param ledgerId Ledger id
           * @returns number of pages removed
           */
          private void removeEntriesForALedger(long ledgerId) {
              // remove pages first to avoid page flushed when deleting file info
              ConcurrentMap<Long, LedgerEntryPage> lPages = pages.remove(ledgerId);
              if (null != lPages) {
                  for (Map.Entry<Long, LedgerEntryPage> pageEntry: lPages.entrySet()) {
                      LedgerEntryPage lep = pageEntry.getValue();
                      lep.usePage();
                      lep.markDeleted();
                      lep.releasePage();
                  }
              }
          }
  
          /**
           * Gets the list of pages in memory that have been changed and hence need to
           * be written as a part of the flush operation that is being issued.
           *
           * @param ledgerId Ledger id
           * @returns last entry in the in memory pages.
           */
          private LinkedList<Long> getFirstEntryListToBeFlushed(long ledgerId) {
              ConcurrentMap<Long, LedgerEntryPage> pageMap = pages.get(ledgerId);
              if (pageMap == null || pageMap.isEmpty()) {
                  return null;
              }
  
              LinkedList<Long> firstEntryList = new LinkedList<Long>();
              for (ConcurrentMap.Entry<Long, LedgerEntryPage> entry: pageMap.entrySet()) {
                  LedgerEntryPage lep = entry.getValue();
                  if (lep.isClean()) {
                      if (!lep.inUse()) {
                          addToCleanPagesList(lep);
                      }
                      if (LOG.isTraceEnabled()) {
                          LOG.trace("Page is clean " + lep);
                      }
                  } else {
                      firstEntryList.add(lep.getFirstEntry());
                  }
              }
              return firstEntryList;
          }
  
          /**
           * Add the LedgerEntryPage to the clean page LRU map.
           *
           * @param lep Ledger Entry Page object
           */
          private void addToCleanPagesList(LedgerEntryPage lep) {
              synchronized (lruCleanPageMap) {
                  if (lep.isClean() && !lep.inUse()) {
                      lruCleanPageMap.put(lep.getEntryKey(), lep);
                  }
              }
          }
  
          /**
           * Remove the LedgerEntryPage from the clean page LRU map.
           *
           * @param lep Ledger Entry Page object
           */
          private void removeFromCleanPageList(LedgerEntryPage lep) {
              synchronized (lruCleanPageMap) {
                  if (!lep.isClean() || lep.inUse()) {
                      lruCleanPageMap.remove(lep.getEntryKey());
                  }
              }
          }
  
          /**
           * Get the set of active ledgers.
           *
           */
          Set<Long> getActiveLedgers() {
              return pages.keySet();
          }
  
          /**
           * Get a clean page and provision it for the specified ledger and firstEntry within the ledger.
           *
           * @param ledgerId Ledger id
           * @param firstEntry Id of the first entry in the page
           * @returns LedgerEntryPage if present
           */
          LedgerEntryPage grabCleanPage(long ledgerId, long firstEntry) {
              LedgerEntryPage lep = listOfFreePages.poll();
              if (null != lep) {
                  lep.resetPage();
                  lep.setLedgerAndFirstEntry(ledgerId, firstEntry);
                  lep.usePage();
                  return lep;
              }
              while (lruCleanPageMap.size() > 0) {
                  lep = null;
                  synchronized (lruCleanPageMap) {
                      Iterator<Map.Entry<EntryKey, LedgerEntryPage>> iterator = lruCleanPageMap.entrySet().iterator();
  
                      Map.Entry<EntryKey, LedgerEntryPage> entry = null;
                      while (iterator.hasNext()) {
                          entry = iterator.next();
                          iterator.remove();
                          if (entry.getValue().isClean() && !entry.getValue().inUse()) {
                              lep = entry.getValue();
                              break;
                          }
                      }
  
                      if (null == lep) {
                          if (LOG.isDebugEnabled()) {
                              LOG.debug("Did not find eligible page in the first pass");
                          }
                          return null;
                      }
                  }
  
                  // We found a candidate page, lets see if we can reclaim it before its re-used
                  ConcurrentMap<Long, LedgerEntryPage> pageMap = pages.get(lep.getLedger());
                  // Remove from map only if nothing has changed since we checked this lep.
                  // Its possible for the ledger to have been deleted or the page to have already
                  // been reclaimed. The page map is the definitive source of information, if anything
                  // has changed we should leave this page along and continue iterating to find
                  // another suitable page.
                  if ((null != pageMap) && (pageMap.remove(lep.getFirstEntry(), lep))) {
                      if (!lep.isClean()) {
                          // Someone wrote to this page while we were reclaiming it.
                          pageMap.put(lep.getFirstEntry(), lep);
                          lep = null;
                      } else {
                          // Do some bookkeeping on the page table
                          pages.remove(lep.getLedger(), EMPTY_PAGE_MAP);
                          // We can now safely reset this lep and return it.
                          lep.usePage();
                          lep.zeroPage();
                          lep.setLedgerAndFirstEntry(ledgerId, firstEntry);
                          return lep;
                      }
                  } else {
                      lep = null;
                  }
              }
              return lep;
          }
  
          public void addToListOfFreePages(LedgerEntryPage lep) {
              if ((null == lep) || lep.inUse()) {
                  inMemPageMgrStats.getIllegalStateResetCounter().inc();
              }
              if (null != lep) {
                  listOfFreePages.add(lep);
              }
          }
  
          @Override
          public void onSetInUse(LedgerEntryPage lep) {
              removeFromCleanPageList(lep);
          }
  
          @Override
          public void onResetInUse(LedgerEntryPage lep) {
              if (!lep.isDeleted()) {
                  addToCleanPagesList(lep);
              } else {
                  addToListOfFreePages(lep);
              }
          }
  
          @Override
          public void onSetClean(LedgerEntryPage lep) {
              addToCleanPagesList(lep);
          }
  
          @Override
          public void onSetDirty(LedgerEntryPage lep) {
              removeFromCleanPageList(lep);
          }
      }
  
      final int pageSize;
      final int entriesPerPage;
      final int pageLimit;
      final InMemPageCollection pageMapAndList;
  
      // The number of pages that have actually been used
      private final AtomicInteger pageCount = new AtomicInteger(0);
  
      // The persistence manager that this page manager uses to
      // flush and read pages
      private final IndexPersistenceMgr indexPersistenceManager;
  
      // Stats
      private final Counter ledgerCacheHitCounter;
      private final Counter ledgerCacheMissCounter;
      private final OpStatsLogger ledgerCacheReadPageStats;
  
      public IndexInMemPageMgr(int pageSize,
                               int entriesPerPage,
                               ServerConfiguration conf,
                               IndexPersistenceMgr indexPersistenceManager,
                               StatsLogger statsLogger) {
          this.pageSize = pageSize;
          this.entriesPerPage = entriesPerPage;
          this.indexPersistenceManager = indexPersistenceManager;
          this.pageMapAndList = new InMemPageCollection(statsLogger);
  
          long maxDirectMemory = PlatformDependent.estimateMaxDirectMemory();
  
          if (conf.getPageLimit() <= 0) {
              // By default, allocate a third of the direct memory to the page cache
              this.pageLimit = (int) ((maxDirectMemory / 3) / this.pageSize);
          } else {
              this.pageLimit = conf.getPageLimit();
          }
          LOG.info("maxDirectMemory = {}, pageSize = {}, pageLimit = {}",
                  maxDirectMemory, pageSize, pageLimit);
          // Expose Stats
          this.ledgerCacheHitCounter = statsLogger.getCounter(LEDGER_CACHE_HIT);
          this.ledgerCacheMissCounter = statsLogger.getCounter(LEDGER_CACHE_MISS);
          this.ledgerCacheReadPageStats = statsLogger.getOpStatsLogger(LEDGER_CACHE_READ_PAGE);
          // Export sampled stats for index pages, ledgers.
          statsLogger.registerGauge(
                  NUM_INDEX_PAGES,
                  new Gauge<Integer>() {
                      @Override
                      public Integer getDefaultValue() {
                          return 0;
                      }
                      @Override
                      public Integer getSample() {
                          return getNumUsedPages();
                      }
                  }
          );
      }
  
      /**
       * @return page size used in ledger cache
       */
      public int getPageSize() {
          return pageSize;
      }
  
      /**
       * @return number of page used in ledger cache
       */
      private int getNumUsedPages() {
          return pageCount.get();
      }
  
      /**
       * Get the ledger entry page for a given <i>pageEntry</i>.
       *
       * @param ledger
       *          ledger id
       * @param pageEntry
       *          first entry id of a given page
       * @return ledger entry page
       * @throws IOException
       */
      LedgerEntryPage getLedgerEntryPage(long ledger,
                                                long pageEntry) throws IOException {
          LedgerEntryPage lep = getLedgerEntryPageFromCache(ledger, pageEntry, false);
          if (lep == null) {
              ledgerCacheMissCounter.inc();
              lep = grabLedgerEntryPage(ledger, pageEntry);
          } else {
              ledgerCacheHitCounter.inc();
          }
          return lep;
      }
  
      LedgerEntryPage getLedgerEntryPageFromCache(long ledger,
                                                         long firstEntry,
                                                         boolean onlyDirty) {
          LedgerEntryPage lep = pageMapAndList.getPage(ledger, firstEntry);
          if (onlyDirty && null != lep && lep.isClean()) {
              return null;
          }
          if (null != lep) {
              lep.usePage();
          }
          return lep;
      }
  
      /**
       * Grab ledger entry page whose first entry is <code>pageEntry</code>.
       * If the page doesn't existed before, we allocate a memory page.
       * Otherwise, we grab a clean page and read it from disk.
       *
       * @param ledger Ledger Id
       * @param pageEntry Start entry of this entry page.
       */
      private LedgerEntryPage grabLedgerEntryPage(long ledger, long pageEntry) throws IOException {
          LedgerEntryPage lep = grabCleanPage(ledger, pageEntry);
          try {
              // should get the up to date page from the persistence manager
              // before we put it into table otherwise we would put
              // an empty page in it
              Stopwatch readPageStopwatch = Stopwatch.createStarted();
              boolean isNewPage = indexPersistenceManager.updatePage(lep);
              if (!isNewPage) {
                  ledgerCacheReadPageStats.registerSuccessfulEvent(
                          readPageStopwatch.elapsed(TimeUnit.MICROSECONDS),
                          TimeUnit.MICROSECONDS);
              }
          } catch (IOException ie) {
              // if we grab a clean page, but failed to update the page
              // we should put this page in the free page list so that it
              // can be reassigned to the next grabPage request
              lep.releasePageNoCallback();
              pageMapAndList.addToListOfFreePages(lep);
              throw ie;
          }
          LedgerEntryPage oldLep;
          if (lep != (oldLep = pageMapAndList.putPage(lep))) {
              // if we grab a clean page, but failed to put it in the cache
              // we should put this page in the free page list so that it
              // can be reassigned to the next grabPage request
              lep.releasePageNoCallback();
              pageMapAndList.addToListOfFreePages(lep);
              // Increment the use count of the old lep because this is unexpected
              oldLep.usePage();
              lep = oldLep;
          }
          return lep;
      }
  
      void removePagesForLedger(long ledgerId) {
          pageMapAndList.removeEntriesForALedger(ledgerId);
      }
  
      long getLastEntryInMem(long ledgerId) {
          return pageMapAndList.getLastEntryInMem(ledgerId);
      }
  
      private LedgerEntryPage grabCleanPage(long ledger, long entry) throws IOException {
          if (entry % entriesPerPage != 0) {
              throw new IllegalArgumentException(entry + " is not a multiple of " + entriesPerPage);
          }
  
          while (true) {
              boolean canAllocate = false;
              if (pageCount.incrementAndGet() <= pageLimit) {
                  canAllocate = true;
              } else {
                  pageCount.decrementAndGet();
              }
  
              if (canAllocate) {
                  LedgerEntryPage lep = new LedgerEntryPage(pageSize, entriesPerPage, pageMapAndList);
                  lep.setLedgerAndFirstEntry(ledger, entry);
                  lep.usePage();
                  return lep;
              }
  
              LedgerEntryPage lep = pageMapAndList.grabCleanPage(ledger, entry);
              if (null != lep) {
                  return lep;
              }
              LOG.info("Could not grab a clean page for ledger {}, entry {}, force flushing dirty ledgers.",
                      ledger, entry);
              flushOneOrMoreLedgers(false);
          }
      }
  
      void flushOneOrMoreLedgers(boolean doAll) throws IOException {
          List<Long> ledgersToFlush = new ArrayList<>(pageMapAndList.getActiveLedgers());
          for (Long potentiallyDirtyLedger : ledgersToFlush) {
              try {
                  flushSpecificLedger(potentiallyDirtyLedger);
              } catch (Bookie.NoLedgerException e) {
                  continue;
              }
              if (!doAll) {
                  break;
              }
          }
      }
  
      /**
       * Flush a specified ledger.
       *
       * @param ledger Ledger Id
       * @throws IOException
       */
      private void flushSpecificLedger(long ledger) throws IOException {
          LinkedList<Long> firstEntryList = pageMapAndList.getFirstEntryListToBeFlushed(ledger);
  
          // flush ledger index file header if necessary
          indexPersistenceManager.flushLedgerHeader(ledger);
  
          if (null == firstEntryList || firstEntryList.size() == 0) {
              if (LOG.isDebugEnabled()) {
                  LOG.debug("Nothing to flush for ledger {}.", ledger);
              }
              // nothing to do
              return;
          }
  
          // Now flush all the pages of a ledger
          List<LedgerEntryPage> entries = new ArrayList<LedgerEntryPage>(firstEntryList.size());
          try {
              for (Long firstEntry: firstEntryList) {
                  LedgerEntryPage lep = getLedgerEntryPageFromCache(ledger, firstEntry, true);
                  if (lep != null) {
                      entries.add(lep);
                  }
              }
              indexPersistenceManager.flushLedgerEntries(ledger, entries);
          } finally {
              for (LedgerEntryPage lep: entries) {
                  lep.releasePage();
              }
          }
      }
  
      void putEntryOffset(long ledger, long entry, long offset) throws IOException {
          int offsetInPage = (int) (entry % entriesPerPage);
          // find the id of the first entry of the page that has the entry
          // we are looking for
          long pageEntry = entry - offsetInPage;
          LedgerEntryPage lep = null;
          try {
              lep = getLedgerEntryPage(ledger, pageEntry);
              assert lep != null;
              lep.setOffset(offset, offsetInPage * LedgerEntryPage.getIndexEntrySize());
          } catch (FileInfo.FileInfoDeletedException e) {
              throw new Bookie.NoLedgerException(ledger);
          } finally {
              if (null != lep) {
                  lep.releasePage();
              }
          }
      }
  
      long getEntryOffset(long ledger, long entry) throws IOException {
          int offsetInPage = (int) (entry % entriesPerPage);
          // find the id of the first entry of the page that has the entry
          // we are looking for
          long pageEntry = entry - offsetInPage;
          LedgerEntryPage lep = null;
          try {
              lep = getLedgerEntryPage(ledger, pageEntry);
              return lep.getOffset(offsetInPage  * LedgerEntryPage.getIndexEntrySize());
          } finally {
              if (lep != null) {
                  lep.releasePage();
              }
          }
      }
  
      /**
       * Represents a page of the index.
       */
      private class PageEntriesImpl implements LedgerCache.PageEntries {
          final long ledgerId;
          final long initEntry;
  
          PageEntriesImpl(long ledgerId, long initEntry) {
              this.ledgerId = ledgerId;
              this.initEntry = initEntry;
          }
  
          @Override
          public LedgerEntryPage getLEP() throws IOException {
              return getLedgerEntryPage(ledgerId, initEntry);
          }
  
          @Override
          public long getFirstEntry() {
              return initEntry;
          }
  
          @Override
          public long getLastEntry() {
              return initEntry + entriesPerPage;
          }
      }
  
      /**
       * Iterable over index pages -- returns PageEntries rather than individual
       * entries because getEntries() above needs to be able to throw an IOException.
       */
      private class PageEntriesIterableImpl implements LedgerCache.PageEntriesIterable {
          final long ledgerId;
          final FileInfoBackingCache.CachedFileInfo fi;
          final long totalEntries;
  
          long curEntry = 0;
  
          PageEntriesIterableImpl(long ledgerId) throws IOException {
              this.ledgerId = ledgerId;
              this.fi = indexPersistenceManager.getFileInfo(ledgerId, null);
              this.totalEntries = max(entriesPerPage * (fi.size() / pageSize), getLastEntryInMem(ledgerId));
          }
  
          @Override
          public Iterator<LedgerCache.PageEntries> iterator() {
              return new Iterator<LedgerCache.PageEntries>() {
                  @Override
                  public boolean hasNext() {
                      return curEntry < totalEntries;
                  }
  
                  @Override
                  public LedgerCache.PageEntries next() {
                      LedgerCache.PageEntries next = new PageEntriesImpl(ledgerId, curEntry);
                      curEntry += entriesPerPage;
                      return next;
                  }
              };
          }
  
          @Override
          public void close() {
              fi.release();
          }
      }
  
      /**
       * Return iterator over pages for mapping entries to entry loggers.
       * @param ledgerId
       * @return Iterator over pages
       * @throws IOException
       */
      public LedgerCache.PageEntriesIterable listEntries(long ledgerId) throws IOException {
          return new PageEntriesIterableImpl(ledgerId);
      }
  }
  