# IdentityPool.java

    /*
    ** Module   : IdentityPool.java
    ** Abstract : Pool of reusable primary keys.
    **
    ** Copyright (c) 2004-2023, Golden Code Development Corporation.
    **
    ** -#- -I- --Date-- --JPRM-- -----------------------------------Description-----------------------------------
    ** 001 SVL 20080901   @39777 Created initial version. The pool performs
    **                           periodical database scanning and allows to
    **                           reclaim keys from deletes and create rollbacks.
    ** 002 SVL 20080922   @39893 Fixed NPE during logging.
    ** 003 ECF 20081107   @40403 Changed default scan settings. They are less
    **                           intrusive now (a larger band is scanned, but
    **                           only in 60 second intervals).
    ** 004 ECF 20090107   @41012 Fixed server shutdown. Scanning thread needed
    **                           to be a daemon, it was holding up a clean
    **                           shutdown of the server.
    ** 005 SVL 20090715   @43194 Added OOME protection into scanning thread.
    ** 006 ECF 20090729   @43452 Adapted Commitable API change. Added validate()
    **                           method to IdentityContext.
    ** 007 SVL 20090929   @44051 Fixed reclaimedKeysPool and keysPool after-scan
    **                           merging and edge case when scanThreshold = 0.
    ** 008 SVL 20090930   @44057 Added stop() function.
    ** 009 OM  20140603          Replaced Apache logging. Fixed session close based on log level.
    ** 010 EVL 20160223          Javadoc fixes to make compatible with Oracle Java 8 for Solaris 10.
    ** 011 ECF 20160825          Disabled key scanning by default, as key reclamation has been
    **                           determined to be the source of record lock deadlocks.
    ** 012 ECF 20160907          Do not notify scanning thread that pool is low if scanning is
    **                           disabled. Format cleanup.
    ** 128 ECF 20190714          Added support for aggressive buffer flushing mode.
    ** 129 ECF 20200906          New ORM implementation.
    **     CA  20220210          Removed Commitable.rollbackPending and its notifications, as there is no actual
    **                           implementation of this method.
    **     TJD 20220331          Removed unused/deprecated KeyScanner
    ** 130 GBB 20230512          Logging methods replaced by CentralLogger/ConversionStatus.
    */

    /*
    ** This program is free software: you can redistribute it and/or modify
    ** it under the terms of the GNU Affero General Public License as
    ** published by the Free Software Foundation, either version 3 of the
    ** License, or (at your option) any later version.
    **
    ** This program is distributed in the hope that it will be useful,
    ** but WITHOUT ANY WARRANTY; without even the implied warranty of
    ** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    ** GNU Affero General Public License for more details.
    **
    ** You may find a copy of the GNU Affero GPL version 3 at the following
    ** location: https://www.gnu.org/licenses/agpl-3.0.en.html
    **
    ** Additional terms under GNU Affero GPL version 3 section 7:
    **
    **   Under Section 7 of the GNU Affero GPL version 3, the following additional
    **   terms apply to the works covered under the License.  These additional terms
    **   are non-permissive additional terms allowed under Section 7 of the GNU
    **   Affero GPL version 3 and may not be removed by you.
    **
    **   0. Attribution Requirement.
    **
    **     You must preserve all legal notices or author attributions in the covered
    **     work or Appropriate Legal Notices displayed by works containing the covered
    **     work.  You may not remove from the covered work any author or developer
    **     credit already included within the covered work.
    **
    **   1. No License To Use Trademarks.
    **
    **     This license does not grant any license or rights to use the trademarks
    **     Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks
    **     of Golden Code Development Corporation. You are not authorized to use the
    **     name Golden Code, FWD, or the names of any author or contributor, for
    **     publicity purposes without written authorization.
    **
    **   2. No Misrepresentation of Affiliation.
    **
    **     You may not represent yourself as Golden Code Development Corporation or FWD.
    **
    **     You may not represent yourself for publicity purposes as associated with
    **     Golden Code Development Corporation, FWD, or any author or contributor to
    **     the covered work, without written authorization.
    **
    **   3. No Misrepresentation of Source or Origin.
    **
    **     You may not represent the covered work as solely your work.  All modified
    **     versions of the covered work must be marked in a reasonable way to make it
    **     clear that the modified work is not originating from Golden Code Development
    **     Corporation or FWD.  All modified versions must contain the notices of
    **     attribution required in this license.
    */

    package com.goldencode.p2j.persist.id;

    import com.goldencode.p2j.persist.*;
    import com.goldencode.p2j.security.*;
    import com.goldencode.p2j.util.*;
    import com.goldencode.p2j.util.logging.*;

    import java.util.*;
    import java.util.logging.*;

    /**
     * Represents a pool of reusable primary keys. Built up every time the server
     * is started. Keys can be obtained in two ways:
     * <ul>
     * <li>1. By detecting "gaps" of unused keys in all tables in a background
     * thread. Current implementation periodically scans tables (with an optional
     * delay between queries) in descending direction from highest key band to
     * lowest key band.
     * <li>2. By reclaiming keys from deletes and create rollbacks.
     * </ul>
     */
    public class IdentityPool
    {
       /** Logger */
       private static final CentralLogger LOG = CentralLogger.get(IdentityManager.class.getName());

       /**
        * Available keys stored in ranges. Contains both scanned and reclaimed
        * keys which are more than maxKey.
        */
       private final SortedSet<Range> keysPool = new TreeSet<>();

       /**
        * Available reclaimed keys which are &le; maxKey. Together with keysPool
        * it forms the whole pool of keys. This set of reclaimed keys was
        * separated for convenience considerations.
        */
       private final SortedSet<Long> reclaimedKeysPool = new TreeSet<>();

       /** Keys which were returned by nextPrimaryKey() and were not committed or rolled back. */
       private final SortedSet<Long> handedOutKeys = new TreeSet<>();

       /**
        * Keys which were returned by nextPrimaryKey(), were committed or
        * rolled back but included in the range currently being scanned and
        * pending for scanning to be completed in order to exclude this set
        * of keys from newly found keys.
        */
       private final SortedSet<Long> pendingHandedOutKeys = new TreeSet<>();

       /** Number of available keys (total of keysPool and reclaimedKeysPool). */
       private long poolSize = 0;

       /** The number of IDs we will try to reach in the pool, before we stop scanning. */
       private int targetSize;

       /**
        * The minimum size of available IDs being tracked before a database scan
        * is triggered to find more IDs.
        */
       private int scanThreshold;

       /**
        * The minimum time in seconds between database scans triggered by
        * scanThreshold. In other words, if the pool size stays below
        * scanThreshold, we don't want to trigger another scan until scanInterval
        * seconds have passed since the last scan, so we are not constantly
        * slamming the database looking for IDs that are not there. After all, the
        * point of reclamation is the remove all the "air" between IDs, so the
        * theoretical target is no available IDs. This value doubles each time a
        * threshold-triggered scan occurs, until it reaches maximumScanInterval.
        */
       private int scanInterval;

       /** Maximum number of seconds to which scanInterval can grow. */
       private int maximumScanInterval;

       /** Number of records to involve in each query. */
       private int scanRange;

       /**
        * Number of seconds between range queries, in case an admin wants to
        * "spread out" the impact of reclamation scans.
        */
       private int scanDelay;

       /** Current upper bound of key range for scanning. */
       private long maxKey = 0L;

       /** The range currently being scanned. Empty if no scanning is performed at this moment. */
       private final Range currentlyScannedRange = new Range();

       /**
        * Maximum time between scans (in seconds) for which we can hold the
        * session in order to not to obtain a new session each time.
        */
       private int sessionHoldTime = 2;

       /** Method of available keys searching */
       private ScanMethod scanMethod;

       /** Synchronization object between the scanning thread and identity pool itself. */
       private final Object waitLock = new Object();

       /** Database for which this identity pool has been initialized. */
       private Database database;

       /**
        * Indicates that the pool is stopped (probably because of an error). In
        * the this state the pool does not perform scanning for new keys, does not
        * accept any reclaimed keys and does not hand out any keys.
        */
       private boolean stopped = false;

       /**
        * Context-local container for the keys which were returned by
        * nextPrimaryKey() for this context and were not committed or rolled back.
        */
       private final ContextLocal<IdentityContext> context =
       new ContextLocal<IdentityContext>()
       {
          protected IdentityContext initialValue()
          {
             return (new IdentityContext());
          }
       };

       /**
        * Create the identity pool for the given database.
        *
        * @param database
        *        Database for which the identity pool will be created.
        */
       public IdentityPool(Database database)
       {
          this.database = database;

          // Read identity pool configuration.
          String basePath =
                String.format("database/%s/p2j/identity_pool/", database.getName()) + "%s";

          targetSize =
             Utils.getDirectoryNodeInt(null,
                                       String.format(basePath, "target_size"),
                                       10000,
                                       false);
          scanThreshold =
             Utils.getDirectoryNodeInt(null,
                                       String.format(basePath, "scan_threshold"),
                                       1000,
                                       false);

          if (scanThreshold > targetSize)
          {
             if (LOG.isLoggable(Level.WARNING))
             {
                LOG.log(Level.WARNING,
                        "Scan_threshold should be less or equal target_size. " +
                        "Using scan_threshold=target_size instead.");
             }
             scanThreshold = targetSize;
          }

          scanInterval =
             Utils.getDirectoryNodeInt(null,
                                       String.format(basePath, "initial_scan_interval"),
                                       60 * 60,
                                       false);

          maximumScanInterval =
             Utils.getDirectoryNodeInt(null,
                                       String.format(basePath, "maximum_scan_interval"),
                                       24 * 60 * 60,
                                       false);

          if (maximumScanInterval < scanInterval)
          {
             if (LOG.isLoggable(Level.WARNING))
             {
                LOG.log(Level.WARNING,
                        "Maximum_scan_interval should be more or equal initial_scan_interval. " +
                        "Using maximum_scan_interval=initial_scan_interval instead.");
             }
             maximumScanInterval = scanInterval;
          }

          scanRange =
             Utils.getDirectoryNodeInt(null, String.format(basePath, "scan_range"), 100000, false);
          scanDelay =
             Utils.getDirectoryNodeInt(null, String.format(basePath, "scan_delay"), 60, false);
          String scanMethodString =
             Utils.getDirectoryNodeString(null, String.format(basePath, "scan_method"), null, false);
          if (scanMethodString == null)
          {
             // 20160825 ECF: disabling primary key reclamation, which has been the source of
             // deadlocks, due to insertion of out-of-order keys (they are used as a sorting
             // tie-breaker for non-unique indexes.
             // scanMethod = ScanMethod.JDBC;
             scanMethod = ScanMethod.NONE;
          }
          else
          {
             try
             {
                scanMethod = ScanMethod.valueOf(scanMethodString.toUpperCase());
             }
             catch (IllegalArgumentException e)
             {
                if (LOG.isLoggable(Level.WARNING))
                {
                   LOG.log(Level.WARNING,
                           "Unrecognized scan_method option. Should be " +
                           "'jdbc', 'stored_procedure' or 'none'. Using default instead.");
                }
                // 20160825 ECF: disabling primary key reclamation, which has been the source of
                // deadlocks, due to insertion of out-of-order keys (they are used as a sorting
                // tie-breaker for non-unique indexes.
                // scanMethod = ScanMethod.JDBC;
                scanMethod = ScanMethod.NONE;
             }
          }

          if (scanMethod != ScanMethod.NONE)
          {
             LOG.info("Identity pool initialized for " + database +
                      " target_size = "           + targetSize +
                      " scan_threshold = "        + scanThreshold +
                      " initial_scan_interval = " + scanInterval +
                      " maximum_scan_interval = " + maximumScanInterval +
                      " scan_range = "            + scanRange +
                      " scan_delay = "            + scanDelay +
                      " scan_method = "           + scanMethod);
          }
       }

       /**
        * Return the next primary key from the pool (for the database which is
        * served by this pool).
        *
        * @return  The next primary key or <code>null</code> if no key is
        *          available.
        */
       public Long nextPrimaryKey()
       {
          if (TransactionManager.isTransaction() && !stopped)
          {
             Long key = null;

             synchronized (keysPool)
             {
                if (poolSize > 0)
                {
                   // First try to take a key from the reclaimedKeysPool.
                   if (!reclaimedKeysPool.isEmpty())
                   {
                      key = reclaimedKeysPool.first();
                      reclaimedKeysPool.remove(key);
                   }
                   else if (!keysPool.isEmpty())
                   {
                      // Get the first key from the first range in the pool.
                      Range range = keysPool.first();
                      key = range.extractFirstNumber();
                      if (key == null)
                      {
                         throw new IllegalStateException("Empty range in the pool.");
                      }

                      // If the range become empty, remove it from the pool.
                      if (range.isEmpty())
                      {
                         keysPool.remove(range);
                      }
                   }
                   else
                   {
                      LOG.log(Level.SEVERE,
                              "Inconsistent pool state. Stored pool size " +
                              "does not match actual pool size. Recovered.");
                      poolSize = 0;
                   }

                   if (key != null)
                   {
                      poolSize--;

                      // Add the returned key to the global set of handed out keys.
                      handedOutKeys.add(key);
                   }
                }

                // Notify the scanning thread that it should perform scanning
                // since we are running out of keys.
                if (isKeyScanningEnabled() && poolSize <= scanThreshold)
                {
                   synchronized (waitLock)
                   {
                      waitLock.notify();
                   }
                }
             }

             if (key != null)
             {
                // Add the returned key to the local set of handed out keys.
                context.get().keyHandedOut(key);
                return key;
             }
          }

          return null;
       }

       /**
        * Indicate whether key scanning is enabled.
        *
        * @return  {@code true} if scan method is not {@code NONE}, else {@code false}.
        */
       public boolean isKeyScanningEnabled()
       {
          return scanMethod != ScanMethod.NONE;
       }

       /**
        * Tells identity pool the starting high end of keys to check in its
        * descending range scans. This function should be called only once.
        *
        * @param key
        *        High end of keys.
        */
       public void setScanStartKey(Long key)
       {
          if (maxKey != 0)
          {
             throw new IllegalStateException("Scan start key can be set only once.");
          }

          if (!(key instanceof Long))
          {
             throw new IllegalArgumentException(
                   "This identity pool implementation supports only Long keys.");
          }

          if ((Long) key < 1)
          {
             LOG.log(Level.WARNING,
                     "Scan start key < 1 has been set. The pool has not been started.");
             return;
          }

          maxKey = (Long) key;

          if (LOG.isLoggable(Level.FINE))
          {
             LOG.log(Level.FINE, "Start key has been set up to " + maxKey);
          }

          if (scanMethod != ScanMethod.NONE)
          {
             LOG.log(Level.WARNING, "The only supported ScanMethod is NONE");
          }
       }

       /**
        * Returns the array of keys to the pool (presumably recovered by deletes
        * and create rollbacks).
        *
        * @param keys
        *        The array of keys to be returned to the pool.
        */
       public void reclaimKeys(Long[] keys)
       {
          if (stopped || keys == null || keys.length == 0)
          {
             return;
          }

          boolean fine = LOG.isLoggable(Level.FINE);

          synchronized (keysPool)
          {
             for (Long key : keys)
             {
                if (!(key instanceof Long))
                {
                   throw new IllegalArgumentException(
                      "This identity pool implementation supports only Long keys.");
                }

                Long id = (Long) key;

                if (id > maxKey)
                {
                   // Add the key to the keysPool (as a separate range)...
                   keysPool.add(new Range(id));

                   if (fine)
                   {
                      LOG.log(Level.FINE, "Reclaimed key " + id + " was added to the keysPool");
                   }
                }
                else
                {
                   // ... or to the reclaimedKeysPool.
                   reclaimedKeysPool.add(id);

                   if (fine)
                   {
                      LOG.log(Level.FINE,
                              "Reclaimed key " + id + " was added to the reclaimedKeysPool");
                   }
                }

                poolSize++;
             }
          }
       }

       /**
        * Stops the pool, i.e. scanning thread will be stopped, pool will not
        * accept any reclaimed keys or hand out any keys.
        */
       public void stop()
       {
          if (!stopped)
          {
             boolean stopPerformed = false;

             synchronized (waitLock)
             {
                if (!stopped)
                {
                   stopped = true;
                   stopPerformed = true;
                   waitLock.notify();
                }
             }

             if (stopPerformed)
             {
                LOG.info("Identity pool was forced to stop.");
             }
          }
       }

       /**
        * Notifies the pool that the following handed out keys were committed or
        * rolled back and we can remove them from the global handed out set.
        *
        * @param keys
        *        Previously handed out keys which were committed or rolled back.
        */
       private void clearKeysInHands(Collection<Long> keys)
       {
          synchronized (currentlyScannedRange)
          {
             synchronized (keysPool)
             {
                // No scanning performed meanwhile.
                if (currentlyScannedRange.isEmpty())
                {
                   handedOutKeys.removeAll(keys);
                }
                else
                {
                   // Move all keys which are in the range being scanned to
                   // the pendingHandedOutKeys and remove all given keys from handedOutKeys.
                   Iterator<Long> iter = keys.iterator();
                   while (iter.hasNext())
                   {
                      Long key = iter.next();

                      if (currentlyScannedRange.contains(key))
                      {
                         pendingHandedOutKeys.add(key);
                      }

                      iter.remove();
                   }
                }
             }
          }
       }

       /**
        * Context-local container which manages the keys which were returned by
        * nextPrimaryKey() for this context and were not committed or rolled back.
        */
       private class IdentityContext
       implements Commitable
       {
          /** Keys handed out for this context. */
          private Set<Long> localKeysHandedOut = new HashSet<>();

          /**
           * Informs the IdentityContext that we hand out one more key for it.
           *
           * @param key
           *        Handed out key.
           */
          public void keyHandedOut(Long key)
          {
             if (localKeysHandedOut.isEmpty())
             {
                // Register as listener for the "master" commit/rollback calls.
                TransactionManager.registerTransactionCommit(this, true);
             }

             localKeysHandedOut.add(key);

             if (LOG.isLoggable(Level.FINE))
             {
                 LOG.log(Level.FINE, "Key was handed out to " + this + ": " + key);
             }
          }

          /**
           * Called by TransactionManager after the "master" commit was performed.
           * Removes the keys handed out to this context from local and global
           * keys handed out keys.
           *
           * @param transaction
           *        Always true.
           */
          public void commit(boolean transaction)
          {
             clearKeysInHands(localKeysHandedOut);
             localKeysHandedOut.clear();

             if (LOG.isLoggable(Level.FINE))
             {
                LOG.log(Level.FINE, "Commit was called for " + this);
             }
          }

          /**
           * Called by TransactionManager after the "master" rollback was
           * performed. Removes the keys handed out to this context from local
           * and global keys handed out keys.
           *
           * @param transaction
           *        Always true.
           */
          public void rollback(boolean transaction)
          {
             clearKeysInHands(localKeysHandedOut);
             localKeysHandedOut.clear();

             if (LOG.isLoggable(Level.FINE))
             {
                LOG.log(Level.FINE, "Rollback was called for " + this);
             }
          }

          /**
           * No-op
           */
          public void validate(boolean transaction, boolean aggressiveFlush)
          {
          }
       }

       /**
        * Enumeration which defines a set of available methods for retrieval of
        * available keys.
        */
       private enum ScanMethod
       {

          /** Do not retrieve keys */
          NONE
       }
    }