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 ≤ 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
}
}