InMemoryLockManager.java
/*
** Module : InMemoryLockManager.java
** Abstract : In-memory, record-level lock manager; synchronizes concurrent
** access to records within a single process
**
** Copyright (c) 2004-2025, Golden Code Development Corporation.
**
** -#- -I- --Date-- --JPRM-- ----------------------------------Description-----------------------------------
** 001 ECF 20051010 @23076 Created initial version. An in-memory lock
** manager implementation. Manages pessimistic,
** row-level locking in support of the Progress
** locking semantic. Locks are limited in scope
** to the current server process.
** 002 ECF 20051104 @23241 Added queryLock method. Queries lock type
** held by the current context on a specific
** record.
** 003 ECF 20051110 @23381 Removed database from lock and queryLock
** signatures. This reflects a change in the
** LockManager API. Added setDatabase method to
** allow database name to be preserved for
** logging. It was unnecessary to track database
** name with each lock, since a LockManager
** instance is now database-specific.
** 004 ECF 20051207 @23755 Use more specific exception type if lock is
** unavailable. PersistenceException replaced
** with LockUnavailableException.
** 005 ECF 20060117 @23961 Report lock conflict to user. A message is
** sent via the LogicalTerminal before the
** current thread waits.
** 006 ECF 20060120 @24041 Replaced ThreadLocal with ContextLocal.
** Implemented cleanup routine to release all
** locks held by an expired context.
** 007 ECF 20060223 @24718 Clear conflict message. Added call to
** LogicalTerminal.hideMessage if session was
** blocked waiting for lock and is freed.
** 008 ECF 20060610 @27104 Fixed offline mode. NPEs had been introduced
** when this class was integrated with the P2J
** server environment.
** 009 ECF 20060728 @28267 Minor changes. Prevent setting user wait
** message more than once. Improved javadoc.
** 010 ECF 20060918 @29666 Minor logging improvement during context
** cleanup.
** 011 ECF 20070412 @32980 Retrofitted LockManager interface changes.
** Extracted LockRecord to top level class
** RecordIdentifier. Also improved logging
** during context cleanup.
** 012 ECF 20070911 @35288 Replaced database name string with Database
** information object. Required for non-local
** database connections.
** 013 ECF 20071213 @36376 Added lock contention debug logging. When
** LockManager logger is set to FINEST, each
** lock acquisition includes stack trace data.
** Upon contention for a record lock, these data
** are logged, so we can see who is requesting a
** lock who is holding it, and where in the code
** the lock was requested and acquired.
** 014 ECF 20071214 @36379 Throw EndConditionException when user aborts
** lock request. Previously we were throwing
** StopConditionException, which was never
** caught by business logic. Note: this appears
** to be a more correct approach, but further
** testing is required.
** 015 ECF 20071220 @36577 Fixed lock leak detection false positive. We
** were not removing an entry from the map of
** locally locked records in certain release
** cases. Also refactored context locals into an
** inner class (Context) to minimize context
** local variable lookups. Integrated generics.
** 016 ECF 20080221 @37140 Fixed NPE in contextCleanup().
** 017 ECF 20080506 @38218 Fixed core algorithm, which was not properly
** allowing up/downgrade transitions between
** SHARE and EXCLUSIVE lock types. Also added
** isHeadless(). Used to determine whether a UI
** exists to enable a message to be sent to the
** user in lock contention situations. Improved
** built-in test harness to test up/downgrades.
** 017 ECF 20080506 @38219 The current implementation of UI detection is
** flawed, in that it is server-based, rather
** than client-based. A problem will occur when
** a non-standard client (which does not support
** a UI) is attached to a standard server.
** 018 ECF 20080806 @39323 Added support for 64-bit primary keys. Only
** applies to test harness in main(). Core code
** is agnostic to data type of record ID.
** 019 ECF 20081106 @40353 Added logging of locking above a configurable
** time threshold. If a lock is held for more
** than the threshold (set in the directory), a
** warning and stack trace are logged.
** 020 ECF 20081113 @40452 Fixed lock leak. Newly created locks were not
** being tracked by the local context, so they
** could not be cleaned up in the event of an
** abrupt end of the session. Improved logging.
** 021 ECF 20081204 @40802 Changed warning threshold logging. We now use
** the same approach as the Persistence class
** uses.
** 022 ECF 20090128 @41241 Fixed ConcurrentModificationException during
** cleanup. This exception could be raised as
** leaked locks were being released.
** 023 ECF 20090801 @43498 Added support for lock administration. Added
** inner class which implements LockAdministrator,
** and new getLockAdministrator() method to main
** class. Modified internal data structures to
** report lock information at a moment in time.
** 024 CA 20091119 @44430 If a record is locked by another subject, now the
** status message contains the subject name too.
** Also, the status line which contains the record
** locked message is no longer trimmed to 63 chars.
** Instead of session IDs, now it uses a session
** token to describe a locker (so subject name will
** be provided too).
** 025 CA 20091119 @44437 The record lock info was modified to receive
** session tokens instead of session IDs.
** 026 CA 20091203 @44474 Rolled back H014 (@36379) - correct behavior is
** to throw a StopConditionException when user
** aborts lock request.
** 027 VMN 20130830 Added lockTable method.
** 028 ECF 20131028 Added support for lock metadata. Moved to lock sub-package.
** 029 ECF 20150111 Replaced Apache commons logging with J2SE logging.
** 030 ECF 20150130 Ensure table lock info is not collected by lock admin.
** 031 ECF 20150305 Fixed implementation of automatic SHARE locking of tables during
** record locking; we were leaking table locks, which were only being
** removed at context cleanup.
** 032 ECF 20160120 Fixed synchronization within lockTable method.
** 033 ECF 20160819 Added APIs to permit wait timeouts to be specified. Fixed lock
** notification algorithm to correctly determine old lock types.
** 034 SBI 20170225 Moved administration profile type, access control list type and
** record lock type to their own modules respectively.
** 035 ECF 20200906 Use parameterized record identifiers.
** 036 IAS 20200906 Fixed type parameters.
** CA 20201003 Use an identity HashSet where possible.
** ECF 20210228 Minor format cleanup.
** TJD 20220504 Upgrade do Java 11 minor changes
** SVL 20230113 Improved performance by replacing some "for-each" loops with indexed "for" loops.
** 037 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 038 GBB 20230825 SecurityManager session methods calls updated.
** 039 OM 20240909 Improved Database API. Concurrent multitenancy functionality implementation.
** 040 CA 20241030 99% of the time a record will be locked in a single session; in such cases, keep
** that single LockStatus reference and move to a set only when another session
** establishes a lock.
** 041 OM 20241128 Multi-tenant runtime support: selected the proper persistence context, eventually
** based on [sharedDb] parameter.
** 042 ES 20250407 Handle StopConditionException throw ErrorManager.
*/
/*
** 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.lock;
import java.io.*;
import java.util.*;
import java.util.logging.*;
import com.goldencode.p2j.admin.*;
import com.goldencode.p2j.main.*;
import com.goldencode.p2j.persist.*;
import com.goldencode.p2j.security.*;
import com.goldencode.p2j.security.SecurityManager;
import com.goldencode.p2j.ui.*;
import com.goldencode.p2j.util.*;
import com.goldencode.p2j.util.ErrorManager;
import com.goldencode.p2j.util.logging.*;
import com.goldencode.util.*;
/**
* An implementation of the Progress row-level, pessimistic locking semantic.
* Database level locking is not performed by this class; that is left to
* clients of this class if they so choose. This class is used to grant or
* deny permission to a particular lock requester for a given record in a
* given database table at a particular point in time. Depending upon the
* parameters of a request to obtain a lock, that request may return
* normally, indicating the requested lock has been acquired, or it may
* result in an <code>ErrorConditionException</code> being thrown. A
* successful lock acquisition does not necessarily return immediately;
* rather, the current thread may be blocked until such time as the lock
* becomes available.
* <p>
* It is important to note that this level of lock management occurs outside
* of the database server. There is a fundamental assumption that <i>all
* attempts to lock database records pessimistically are routed through this
* layer <b>before</b> client code attempts to obtain (or release) any
* database-level pessimistic locks.</i> If this fundamental assumption is
* violated, the Progress locking semantic cannot be guaranteed. While this
* is a significantly limiting factor for concurrent, external access to the
* database, it is necessary for the P2J environment to accurately mimic an
* application's pre-conversion behavior, since Progress locking behavior is
* different than that which is provided by SQL semantics.
* <p>
* A further assumption of this implementation is that changes to the lock
* status of a record for a particular context are only made within that
* context, and in fact, within that context, it is further assumed that
* changes are only made on a single thread. Otherwise, work continuing on
* a thread with an assumption of a certain lock status could be corrupted if
* a separate thread makes a change to that lock status.
* <p>
* Because this implementation manages all locks in memory and does not
* persist lock state, an abnormal or abrupt end of the server process will
* result in the loss of the current state of all record locks.
* <p>
* <b>Implementation Notes:</b><br>
* The locking algorithm implemented by this class is naturally biased toward
* share locks over exclusive locks. In a share-lock intensive environment,
* exclusive locks may be starved out, as multiple threads may acquire and
* release share locks. As long as any one thread holds a share lock, an
* exclusive lock cannot be acquired. There is no fairness algorithm built
* into this implementation to prevent such a scenario. When a lock is
* released (or downgraded) in a contention situation, all blocked threads are
* notified of the lock status change. The first thread to be run after a lock
* status change is given the first opportunity to acquire its lock. It is up
* to the JVM implementation (which typically will defer to the operating
* system thread scheduler) as to the order in which waiting threads are
* notified and run.
* @param <T>
* table reference type
*/
public class InMemoryLockManager<T extends Serializable>
implements LockManager<T>
{
/** Logger */
private static final CentralLogger LOG = CentralLogger.get(LockManager.class.getName());
/** Is warn level logging enabled? */
private static final boolean warn = LOG.isLoggable(Level.WARNING);
/** Is debug level logging enabled? */
private static final boolean debug = LOG.isLoggable(Level.FINE);
/** Is trace level logging enabled? */
private static final boolean trace = LOG.isLoggable(Level.FINEST);
/** Is offline mode? */
private final boolean offline = (SecurityManager.getInstance() == null);
/** Master lock table of lock status by {@code RecordIdentifier} */
private final Map<RecordIdentifier<T>, LockStatus<T>> lockTable = new HashMap<>();
/** Map of reference counts for records in the process of being locked */
private final Map<RecordIdentifier<T>, Integer> inPlay = new HashMap<>();
/** Lock administrator */
private final Admin admin = new Admin();
/** Current session context */
private final ContextLocal<Context<T>> context = new ContextLocal<Context<T>>()
{
protected Context<T> initialValue() { return (new Context<>(offline)); }
protected void cleanup(Context<T> ctx) { contextCleanup(ctx); }
};
/** Database whose record locks are to be managed */
private Database database = null;
/** Optional lock listener */
private LockListener listener = null;
/** Next lock event ID, guaranteed to be sequential and ascending */
private long nextEventID = 0L;
/**
* Threshold in milliseconds above which a lock hold time is reported;
* must be non-negative to enable reporting.
*/
private long warnThreshold = -1L;
/**
* Constructor.
*/
public InMemoryLockManager()
{
}
/**
* Helper method to format stack trace data into a string.
*
* @param buf
* Buffer into which to format data.
* @param throwable
* Object which contains stack trace element array.
*/
private static void printStackTrace(StringBuilder buf, Throwable throwable)
{
for (StackTraceElement elem : throwable.getStackTrace())
{
buf.append("\n ");
buf.append(elem);
}
}
/**
* Write a log entry to reflect a lock transition from one lock state to another.
*
* @param lockType
* The new type of lock; <code>LockType.NONE</code> if lock was released.
* @param database
* Database.
* @param ident
* Record descriptor.
*/
private static <T extends Serializable> void logTransition(LockType lockType,
Database database,
RecordIdentifier<T> ident)
{
logTransition(lockType, null, database, ident, 0L);
}
/**
* Write a log entry to reflect a lock transition from one lock state to
* another. If an old lock type is provided, it is logged as well. If a
* positive, elapsed time is provided, that time is logged, along with a
* stack trace.
*
* @param lockType
* The new type of lock; <code>LockType.NONE</code> if lock was
* released.
* @param oldLockType
* The lock type before the transition.
* @param database
* Database.
* @param ident
* Record descriptor.
* @param elapsed
* For a downgrade or release transition, the amount of time which
* has elapsed, in milliseconds, since <code>oldLockType</code>
* was acquired.
*/
private static <T extends Serializable> void logTransition(LockType lockType,
LockType oldLockType,
Database database,
RecordIdentifier<T> ident,
long elapsed)
{
StringBuilder buf = new StringBuilder("[");
buf.append(Utils.describeContext());
buf.append("] --> ");
buf.append(database);
buf.append(".");
buf.append(ident);
if (lockType != LockType.NONE)
{
buf.append(" LOCK ").append(lockType);
}
else
{
buf.append(" UNLOCK");
}
if (oldLockType != null)
{
buf.append(" (from ");
buf.append(oldLockType);
buf.append(")");
}
if (elapsed > 0L)
{
buf.append(" after ");
buf.append(elapsed);
buf.append(" milliseconds");
LOG.log(Level.WARNING, buf.toString(), new Throwable());
}
else
{
LOG.log(Level.FINEST, buf.toString());
}
}
/**
* Set the name of the physical database with which this lock manager is
* associated.
*
* @param database
* Database whose record locks are to be managed.
*/
public void setDatabase(Database database)
{
this.database = database;
// retrieve report threshold value from directory
if (!offline)
{
String dbID = database.getId();
String spec = "database/%s/p2j/warn_threshold/enable";
String path = String.format(spec, dbID);
boolean warn = Utils.getDirectoryNodeBoolean(null, path, false, false);
if (warn)
{
String dbType = (database.isDirty() ? "dirty" : "primary");
spec = "database/%s/p2j/warn_threshold/lock_manager/%s";
path = String.format(spec, dbID, dbType);
warnThreshold = Utils.getDirectoryNodeInt(null, path, -1, false);
if (LOG.isLoggable(Level.INFO))
{
StringBuilder buf = new StringBuilder("Warn threshold for ");
buf.append(database);
buf.append(": ");
buf.append(warnThreshold);
LOG.log(Level.INFO, buf.toString());
}
}
}
}
/**
* Register a lock listener, which will receive notifications of changes in lock status for any
* record managed by this object.
*
* @param listener
* Listener to be registered or <code>null</code> to deregister the current listener.
*/
public void setLockListener(LockListener listener)
{
this.listener = listener;
}
/**
* Retrieve the {@link LockAdministrator} object associated with this
* <code>LockManager</code> instance.
* <p>
* To ensure a caller has sufficient rights to access administrative lock
* functions, the implementation of this method confirms with the
* administrative server before returning any lock information.
*
* @return <code>LockAdministrator</code> instance, or <code>null</code>
* if the caller does not have administrative rights.
*/
public LockAdministrator getLockAdministrator()
{
if (AdminServerImpl.getSecurityAdmin().adminAccess())
{
return admin;
}
return null;
}
/**
* Lock or release a single database record in the current context. The record is uniquely
* identified by a table name and record ID (i.e., primary key). Whether the record is locked
* or released depends upon the lock type requested: {@code LockType.NONE} is used to release
* a record; any other lock type is used to lock a record. Specifying {@code LockType.NONE}
* for a record not locked in the current context is logically a no-op, though some overhead
* is required for the check, and the caller may block waiting on the lock table monitor.
* <p>
* Before obtaining any record lock, {@code lockTable(LockType.SHARE, table, true)} is
* invoked to acquire a share lock on the table. After releasing any record lock, {@code
* lockTable(LockType.NONE, table, true)} is invoked to release that lock.
* <p>
* Any normal return from this method indicates the requested lock was acquired (or released)
* successfully. Any error will result in an exceptional return. In the event a
* record cannot be locked due to a conflicting lock held by another context, the current
* thread will block until the lock is acquired, unless the requested lock is a "no-wait"
* variant, in which case an exception is thrown.
* <p>
* This method also operates in a check-only mode (if {@code update} is set to {@code false})
* where it goes through the motions of obtaining the specified lock type, but does not
* actually change the lock status. This is at best an unreliable check, since another
* context can change the lock status as soon as the lock status monitor is released, rendering
* the check stale. Nevertheless, it exists to mimic the effect of the lock type option to the
* Progress can-find function.
*
* @param lockType
* The type of lock requested; {@code LockType.NONE} to release a lock.
* @param ident
* ID which uniquely identifies the record being queried.
* @param update
* {@code true} if the lock state for the current record should be updated; else
* {@code false}. This is set to {@code false} if the caller just wants to know
* whether a lock is available, without actually changing its status.
*
* @throws LockUnavailableException
* if a "no-wait" lock cannot be acquired immediately.
*/
public void lock(LockType lockType, RecordIdentifier<T> ident, boolean update)
throws LockUnavailableException
{
lock(lockType, ident, update, 0L);
}
/**
* Lock or release a single database record in the current context. The record is uniquely
* identified by a table name and record ID (i.e., primary key). Whether the record is locked
* or released depends upon the lock type requested: {@code LockType.NONE} is used to release
* a record; any other lock type is used to lock a record. Specifying {@code LockType.NONE}
* for a record not locked in the current context is logically a no-op, though some overhead
* is required for the check, and the caller may block waiting on the lock table monitor.
* <p>
* Before obtaining any record lock, {@code lockTable(LockType.SHARE, table, true)} is
* invoked to acquire a share lock on the table. After releasing any record lock, {@code
* lockTable(LockType.NONE, table, true)} is invoked to release that lock.
* <p>
* Any normal return from this method indicates the requested lock was acquired (or released)
* successfully. Any error or timeout will result in an exceptional return. In the event a
* record cannot be locked due to a conflicting lock held by another context, the current
* thread will block until the lock is acquired, unless either (a) the requested lock is a
* "no-wait" variant; or (b) a positive timeout value was provided and at least that number
* of milliseconds has elapsed without acquiring the lock. In either case, an exception is
* thrown as described below.
* <p>
* This method also operates in a check-only mode (if {@code update} is set to {@code false})
* where it goes through the motions of obtaining the specified lock type, but does not
* actually change the lock status. This is at best an unreliable check, since another
* context can change the lock status as soon as the lock status monitor is released, rendering
* the check stale. Nevertheless, it exists to mimic the effect of the lock type option to the
* Progress can-find function.
*
* @param lockType
* The type of lock requested; {@code LockType.NONE} to release a lock.
* @param ident
* ID which uniquely identifies the record being queried.
* @param update
* {@code true} if the lock state for the current record should be updated; else
* {@code false}. This is set to {@code false} if the caller just wants to know
* whether a lock is available, without actually changing its status.
* @param timeout
* Number of milliseconds to wait before throwing {@link LockTimeoutException},
* or 0L to wait indefinitely.
*
* @throws LockUnavailableException
* if a "no-wait" lock cannot be acquired immediately.
* @throws LockTimeoutException
* if a non-zero timeout was provided and the requested lock has not been acquired
* by the time that period has elapsed.
*/
public void lock(LockType lockType, RecordIdentifier<T> ident, boolean update, long timeout)
throws LockUnavailableException,
LockTimeoutException
{
boolean noLock = LockType.NONE.equals(lockType);
// if no update to the lock state is requested and the lock type is NONE, no need to check
// further
if (noLock && !update)
{
return;
}
T table = ident.getTable();
LockType previousTableLock = null;
try
{
// if we are acquiring (or testing whether we could acquire) any record lock, we have to
// get a share lock on the table first
if (!noLock)
{
LockType prev = lockTable(LockType.SHARE, table, update);
if (update)
{
previousTableLock = prev;
}
}
// call lockImpl for normal record lock processing
lockImpl(lockType, ident, update, timeout);
}
catch (LockUnavailableException exc)
{
// if record lock acquisition failed, but table lock was modified, we have to set table
// lock back to what it was previously
if (previousTableLock != null && !previousTableLock.isShare())
{
try
{
lockTable(previousTableLock, table, update);
}
catch (LockUnavailableException exc2)
{
// should never get here, since NO-WAIT variant will not have been requested
}
}
throw exc;
}
finally
{
// if lock type is NONE (the record lock has been released) and this context has only 1
// remaining lock for the current table, it represents the table lock itself; release it
if (noLock)
{
// update must be true, or we wouldn't be here
Context<T> local = context.get();
Map<RecordIdentifier<T>, LockStatus<T>> statusMap = local.locked.get(table);
if (statusMap != null && statusMap.size() == 1)
{
try
{
lockTable(LockType.NONE, table, update);
}
catch (LockUnavailableException exc)
{
// will not be thrown for lock release
}
}
}
}
}
/**
* Attempt to obtain the specified lock type on a table. If the lock cannot be obtained
* immediately, the current thread will block until the lock becomes available and is
* obtained, unless the requested lock type is a "no-wait" lock. In the latter case,
* an exception is raised, indicating that the lock is not available. The current thread
* continues once the lock is obtained.
* <p>
* A lock currently held can be released by specifying <code>LockType.NONE</code> for the
* <code>lockType</code> parameter.
* <p>
* The normal return of this method indicates that the lock request (or lack thereof)
* completed successfully, and that the current context now holds a lock of the requested
* type.
*
* @param lockType
* Type of lock to be obtained. <code>LockType.NONE</code> is used to release an
* existing lock, and to continue with no lock.
* @param table
* Table identifier.
* @param update
* <code>true</code> if the lock state for the current table should be updated;
* else <code>false</code>.
*
* @return The lock type held by the current context on the given table before this method
* was invoked.
*
* @throws LockUnavailableException
* if a "no-wait" lock cannot be acquired immediately.
*/
public LockType lockTable(LockType lockType, T table, boolean update)
throws LockUnavailableException
{
boolean noLock = LockType.NONE.equals(lockType);
// if no update to the lock state is requested and the lock type is NONE, no need to check
// further
if (noLock && !update)
{
return LockType.NONE;
}
TableIdentifier<T> ident = new TableIdentifier<>(table);
LockStatus<T> status;
synchronized (lockTable)
{
status = lockTable.get(ident);
}
// remember current lock type for this context
LockType currentLockType = (status == null) ? LockType.NONE : status.lockType;
// attempt to change lock if requested type is different than current type
if (!currentLockType.equals(lockType))
{
lockImpl(lockType, ident, update, 0L);
}
// return previous lock type (so that calling code knows what to set it back to after bulk
// delete or if record lock attempt fails)
return currentLockType;
}
/**
* Query the lock type currently held by the current context for the
* specified database record.
* <p>
* This method will always return the <b>non-</b><code>NO_WAIT</code>
* variants of a lock type. That is, even if the actual lock type is
* <code>LockType.EXCLUSIVE_NO_WAIT</code> or
* <code>LockType.SHARE_NO_WAIT</code>, the simpler types of
* <code>LockType.EXCLUSIVE</code> and <code>LockType.SHARE</code>,
* respectively, are returned. The lack of a lock returns
* <code>LockType.NONE</code> rather than <code>null</code>.
*
* @param ident
* ID which uniquely identifies the record being queried.
*
* @return The lock type currently held by the current context, never
* <code>null</code>.
*/
public LockType queryLock(RecordIdentifier<T> ident)
{
LockStatus<T> status = null;
synchronized (lockTable)
{
// Check whether record is locked (by anyone).
status = lockTable.get(ident);
}
// Check whether record is locked by current context. If so, return
// lock type. No need to synchronize on status here, because we are
// only querying the status of the current context, which cannot be
// changed by a concurrently running context.
if (status != null)
{
Context<T> local = context.get();
if (status.isLockedBy(local.locker))
{
return status.getLockType().toWaitVariant();
}
}
// Record is not locked by current context.
return LockType.NONE;
}
/**
* Release all locks represented by the specified set of lock status
* objects. This routine is called during context cleanup. Generally,
* if exiting the context cleanly, <code>locks</code> will be empty.
* However, if the context is aborted abnormally, due to a dropped client,
* for instance, the set may contain some entries which must be cleaned
* up to prevent ghost record locks.
*
* @param local
* Current context's state.
*/
private void contextCleanup(Context<T> local)
{
boolean clean = local.locked.isEmpty();
if (clean && !warn)
{
// nothing to clean up and nothing to report
return;
}
String prefix = null;
if (debug || (warn && !clean))
{
StringBuilder buf = new StringBuilder("[");
buf.append(Utils.describeContext());
buf.append("] --> ");
buf.append(database);
buf.append(": ");
prefix = buf.toString();
}
if (clean)
{
if (debug)
{
StringBuilder buf = new StringBuilder(prefix);
buf.append("no record lock leak detected");
LOG.log(Level.FINE, buf.toString());
}
return;
}
// locks were orphaned in this context, so we must release them
int leakCount = 0;
ArrayList<LockStatus<T>> statusList = new ArrayList<>();
for (Map<RecordIdentifier<T>, LockStatus<T>> map : local.locked.values())
{
leakCount += map.size();
statusList.addAll(map.values());
}
if (warn)
{
StringBuilder buf = new StringBuilder(prefix);
buf.append("cleaning up ");
buf.append(leakCount);
buf.append(" leaked record lock(s) for exiting context (");
buf.append(local.locked);
buf.append(")");
LOG.log(Level.WARNING, buf.toString());
}
if (debug)
{
StringBuilder buf = new StringBuilder(prefix);
buf.append("releasing all record locks held by context...");
LOG.log(Level.FINE, buf.toString());
}
// release locks held within this context
for (int i = 0; i < statusList.size(); i++)
{
LockStatus<T> ls = statusList.get(i);
release(ls, local);
}
if (debug)
{
StringBuilder buf = new StringBuilder(prefix);
buf.append("...context record lock release complete");
LOG.log(Level.FINE, buf.toString());
}
}
/**
* Release the current context's lock, if any, on the record represented
* by <code>status</code>. If after doing so, no lockers remain in
* <code>status</code>, and no updates are pending, <code>status</code> is
* removed from the lock table. If no lock is held by the current
* context, this method is effectively a no-op.
*
* @param status
* Object which represents the lock status of the target database
* record.
* @param local
* Current session context.
*/
private void release(LockStatus<T> status, Context<T> local)
{
SessionToken locker = local.locker;
synchronized (status)
{
if (status.isLockedBy(locker))
{
RecordIdentifier<T> ident = status.getRecordIdentifier();
LockType oldType = status.getLockType();
// Remove current context from the set of lockers for this record.
status.removeLocker(locker);
// Remove the lock status object from the lock table, if it is not
// "in-play" (being modified by other threads).
if (status.getLockerCount() == 0)
{
synchronized (lockTable)
{
if (!inPlay.containsKey(ident))
{
lockTable.remove(ident);
}
}
}
// Remove the corresponding entry from the map of locally locked records.
T table = ident.getTable();
Map<RecordIdentifier<T>, LockStatus<T>> statusMap = local.locked.get(table);
statusMap.remove(ident);
if (statusMap.isEmpty())
{
local.locked.remove(table);
}
notifyLockChange(locker, ident, oldType, LockType.NONE, local.osUserName);
if (trace)
{
logTransition(LockType.NONE, database, ident);
}
}
}
}
/**
* Indicate if we are operating in headless mode, meaning no UI is
* available to give feedback to users.
*
* @return <code>true</code> if headless, else <code>false</code>.
*/
protected boolean isHeadless()
{
return offline;
}
/**
* Attempt to acquire a share lock on the record represented by
* <code>status</code>. If the record is available for a share lock, the
* method returns immediately. Otherwise, the outcome depends upon the
* value of the "no-wait" designation of the requested lock type: if
* <code>true</code>, an exception is thrown; if <code>false</code>, the
* method blocks until the record becomes available and the share lock is
* acquired.
* <p>
* A share lock can only be acquired if the record is currently unlocked,
* is share locked by any context, or is exclusively locked by the current
* context. If another context holds an exclusive lock on the record, it
* cannot be share locked by this context until that lock is released.
*
* @param status
* Object which represents the lock status of the target database
* record.
* @param locker
* the current locking context.
* @param noWait
* If <code>true</code>, the current thread will not block in the
* event the record cannot be locked immediately; if
* <code>false</code>, the method will block to wait for a
* conflicting lock to be released.
* @param update
* <code>true</code> if the lock state for the current record
* should be updated; else <code>false</code>.
* @param timeout
* Number of milliseconds to wait before throwing {@link LockTimeoutException},
* or 0L to wait indefinitely.
*
* @throws LockUnavailableException
* if a "no-wait" lock cannot be acquired immediately.
* @throws LockTimeoutException
* if a non-zero timeout was provided and the requested lock has not been acquired
* by the time that period has elapsed.
*/
private void acquireShare(LockStatus<T> status,
SessionToken locker,
boolean noWait,
boolean update,
long timeout)
throws LockUnavailableException
{
// Can acquire share lock if:
// 1) the record is unlocked; or
// 2) other users have a share lock; or
// 3) we are the only holder of a share or exclusive lock.
LockType lockType = status.getLockType();
boolean exclusive = lockType.isExclusive();
boolean onlyLocker = status.isOnlyLocker(locker);
if (!exclusive || onlyLocker)
{
if (update)
{
boolean typeChanged = status.setLockType(LockType.SHARE);
boolean lockerAdded = status.addLocker(locker);
if (listener != null && (typeChanged || lockerAdded))
{
LockType oldType;
if (onlyLocker)
{
if (exclusive)
{
// downgraded lock from exclusive to share
oldType = LockType.EXCLUSIVE;
}
else
{
// already had share lock (alone), still do
oldType = LockType.SHARE;
}
}
else if (lockerAdded)
{
// acquired share lock from no lock
oldType = LockType.NONE;
}
else
{
// already had share lock (with others), still do
oldType = LockType.SHARE;
}
// only notify of lock change if we didn't already hold share lock
if (oldType != LockType.SHARE)
{
RecordIdentifier<T> ident = status.getRecordIdentifier();
notifyLockChange(locker, ident, oldType, LockType.SHARE, context.get().osUserName);
}
}
// If downgrading from exclusive to share, notify other threads
// which may be waiting to acquire share locks.
if (exclusive)
{
status.notifyAll();
}
}
return;
}
// Handle no-wait condition. If we got here, the record could not
// be locked immediately.
if (noWait)
{
RecordIdentifier<T> ident = status.getRecordIdentifier();
String msg = "Could not acquire share lock: " + ident;
// Debug logging to report lock contention.
if (debug)
{
status.logContention(LockType.SHARE_NO_WAIT);
}
throw new LockUnavailableException(msg, LockType.SHARE_NO_WAIT, ident);
}
// Block waiting on an exclusive lock; obtain share lock when record becomes available.
lockWhenAvailable(LockType.SHARE, status, locker, update, timeout);
}
/**
* Attempt to acquire an exclusive lock on the record represented by
* <code>status</code>. If the record is available for such a lock, the
* method returns immediately. Otherwise, the outcome depends upon the
* value of the <code>noWait</code> parameter: if <code>true</code>, an
* exception is thrown; if <code>false</code>, the method blocks until
* the record becomes available and the exclusive lock is acquired.
* <p>
* An exclusive lock can only be acquired if the record is currently
* unlocked, or is share locked by the current context. If another
* context holds either an exclusive lock or a share lock on the record,
* it cannot be exclusively locked by this context until all such locks
* are released.
*
* @param status
* Object which represents the lock status of the target database
* record.
* @param locker
* the current locking context.
* @param noWait
* If <code>true</code>, the current thread will not block in the
* event the record cannot be locked immediately; if
* <code>false</code>, the method will block to wait for a
* conflicting lock to be released.
* @param update
* <code>true</code> if the lock state for the current record
* should be updated; else <code>false</code>.
* @param timeout
* Number of milliseconds to wait before throwing {@link LockTimeoutException},
* or 0L to wait indefinitely.
*
* @throws LockUnavailableException
* if a "no-wait" lock cannot be acquired immediately.
* @throws LockTimeoutException
* if a non-zero timeout was provided and the requested lock has not been acquired
* by the time that period has elapsed.
*/
private void acquireExclusive(LockStatus<T> status,
SessionToken locker,
boolean noWait,
boolean update,
long timeout)
throws LockUnavailableException
{
// Can only acquire exclusive lock immediately if we are the only holder of a share or
// exclusive lock, or if there is no lock at all.
LockType lockType = status.getLockType();
if (status.isOnlyLocker(locker) || LockType.NONE.equals(lockType))
{
if (update)
{
boolean typeChanged = status.setLockType(LockType.EXCLUSIVE);
boolean lockerAdded = status.addLocker(locker);
if (listener != null && (typeChanged || lockerAdded))
{
LockType oldType;
if (lockerAdded)
{
// acquired exclusive lock from no lock
oldType = LockType.NONE;
}
else
{
// had share lock (alone), upgraded to exclusive
oldType = LockType.SHARE;
}
RecordIdentifier<T> ident = status.getRecordIdentifier();
notifyLockChange(locker, ident, oldType, LockType.EXCLUSIVE, context.get().osUserName);
}
}
return;
}
// Handle no-wait condition. If we got here, the record could not be locked immediately.
if (noWait)
{
RecordIdentifier<T> ident = status.getRecordIdentifier();
String msg = "Could not acquire exclusive lock: " + ident;
// Debug logging to report lock contention.
if (debug)
{
status.logContention(LockType.EXCLUSIVE_NO_WAIT);
}
throw new LockUnavailableException(msg, LockType.EXCLUSIVE_NO_WAIT, ident);
}
// Block waiting on an exclusive or share lock; obtain exclusive
// lock when record becomes available.
lockWhenAvailable(LockType.EXCLUSIVE, status, locker, update, timeout);
}
/**
* Worker method to handle single database record or table as whole in the current context.
*
* The record is uniquely identified by a table name and record ID (i.e., primary key).
*
* Whether the lock object (record or table as whole) is locked or released depends upon the
* lock type requested: <code>LockType.NONE</code> is used to release a object; any other
* lock type is used to lock a object. Specifying <code>LockType.NONE</code> for a object
* not locked in the current context is logically a no-op, though some overhead is required
* for the check, and the caller may block waiting on the lock table monitor.
* <p>
* Any normal return from this method indicates the requested lock was acquired (or released)
* successfully. Any error will result in an exceptional return. In the event a object
* cannot be locked, due to a conflicting lock held by another context, the current thread
* will block until the lock is acquired, unless the requested lock is a "no-wait" variant,
* in which case an exception is thrown.
* <p>
* This method also operates in a check-only mode (if <code>update</code> is set to
* <code>false</code>) where it goes through the motions of obtaining the specified lock
* type, but does not actually change the lock status. This is at best an unreliable check,
* since another context can change the lock status as soon as the lock status monitor is
* released, rendering the check stale. Nevertheless, it exists to mimic the effect of the
* lock type option to the Progress can-find function.
*
* @param lockType
* The type of lock requested; <code>LockType.NONE</code> to
* release a lock.
* @param ident
* ID which uniquely identifies the record or table being queried.
* @param update
* <code>true</code> if the lock state for the current record or table should be
* updated; else <code>false</code>. This is set to <code>false</code> if the
* caller just wants to know whether a lock is available, without actually changing
* its status.
* @param timeout
* Number of milliseconds to wait before throwing {@link LockTimeoutException},
* or 0L to wait indefinitely.
*
* @throws LockUnavailableException
* if a "no-wait" lock cannot be acquired immediately.
* @throws LockTimeoutException
* if a non-zero timeout was provided and the requested lock has not been acquired
* by the time that period has elapsed.
*/
private void lockImpl(LockType lockType, RecordIdentifier<T> ident, boolean update, long timeout)
throws LockUnavailableException,
LockTimeoutException
{
boolean noLock = LockType.NONE.equals(lockType);
// if no update to the lock state is requested and the lock type is NONE, no need to check
// further
if (noLock && !update)
{
return;
}
Context<T> local = context.get();
SessionToken locker = local.locker;
T table = ident.getTable();
LockStatus<T> status;
synchronized (lockTable)
{
// check whether record is already locked
status = lockTable.get(ident);
if (!noLock)
{
// check whether this is the uncontended, "fast" case where no other context holds
// the lock
if (status == null)
{
// if an update was requested, acquire the lock at the specified level; otherwise,
// return immediately.
if (update)
{
status = (debug
? new TraceLockStatus<>(database, ident, lockType, locker, warnThreshold)
: new LockStatus<>(database, ident, lockType, locker, warnThreshold));
// add status to map of locks held by this context
Map<RecordIdentifier<T>, LockStatus<T>> statusMap = local.locked.get(table);
if (statusMap == null)
{
statusMap = new HashMap<>();
local.locked.put(table, statusMap);
}
statusMap.put(ident, status);
// add status to main lock table
lockTable.put(ident, status);
notifyLockChange(locker, ident, LockType.NONE, lockType, local.osUserName);
if (trace)
{
logTransition(lockType, database, ident);
}
}
return;
}
// another context already holds a lock; mark the target record as "in-play" so that
// it will not be removed from the lock table while this thread is using it
Integer count = inPlay.get(ident);
if (count == null)
{
// establish reference count
count = 1;
}
else
{
// increment reference count
count++;
}
inPlay.put(ident, count);
}
}
// release a lock (or a no-op if record is not currently locked by the current context)
if (noLock)
{
if (status != null)
{
release(status, local);
}
return;
}
// If we got this far, the lock is already held by another context (or
// by the current context). Attempt to acquire it, which may or may not
// lead to contention, depending on the current and requested lock
// types, and who holds them.
try
{
synchronized (status)
{
boolean noWait = lockType.isNoWait();
if (lockType.isShare())
{
acquireShare(status, locker, noWait, update, timeout);
}
else if (lockType.isExclusive())
{
acquireExclusive(status, locker, noWait, update, timeout);
}
if (update)
{
// add status to map of locks held by this context
Map<RecordIdentifier<T>, LockStatus<T>> statusMap = local.locked.get(table);
if (statusMap == null)
{
statusMap = new HashMap<>();
local.locked.put(table, statusMap);
}
statusMap.put(ident, status);
// log the change
if (trace)
{
logTransition(lockType, database, ident);
}
}
}
}
finally
{
synchronized (lockTable)
{
// update "in-play" status to indicate this thread is finished with this record
int count = inPlay.get(ident).intValue();
if (count == 1)
{
// remove reference count for this lock record altogether
inPlay.remove(ident);
}
else
{
// decrement reference count
inPlay.put(ident, Integer.valueOf(--count));
}
}
}
}
/**
* Attempt to lock the record represented by {@code status} with the given lock type as
* soon as the record is available, blocking the current thread for {@code timeout}
* milliseconds, or optionally indefinitely. Multiple threads may end up blocked in this
* method waiting on a particular record. When that record becomes available, all threads are
* notified, such that they become runnable.
* <p>
* There is no fairness guarantee, FIFO or otherwise, as to which thread will execute first;
* this is up to the platform-specific thread scheduler. The first of the released threads to
* execute sets the new lock type of the {@code status} object, which may prevent other
* contexts from locking (if set to {@code LockType.EXCLUSIVE}) or it may not (if set to
* {@code LockType.SHARE}). Threads which were unable to acquire a lock upon release will
* re-enter a wait state, blocking until another chance occurs.
* <p>
* This method must only be invoked when the current thread holds the monitor for the {@code
* status} object.
*
* @param requested
* The type of lock being acquired. The only valid types are
* {@code LockType.EXCLUSIVE} and {@code LockType.SHARE}.
* @param status
* Object which represents the lock status of the target database record.
* @param locker
* The current locking context.
* @param update
* {@code true} if the lock state for the current record should be updated; else
* {@code false}.
* @param timeout
* Number of milliseconds to wait before throwing {@link LockTimeoutException},
* or 0L to wait indefinitely.
*
* @throws LockTimeoutException
* if a non-zero timeout was provided and the requested lock has not been acquired
* by the time that period has elapsed.
* @throws StopConditionException
* if the current thread is interrupted while blocked.
*/
private void lockWhenAvailable(LockType requested,
LockStatus<T> status,
SessionToken locker,
boolean update,
long timeout)
throws LockTimeoutException
{
boolean clearMsg = false;
try
{
long waitUntil = timeout > 0L ? System.nanoTime() + (timeout * 1000000) : 0L;
long start = (warnThreshold > -1L ? System.currentTimeMillis() : -1L);
LockType current;
// Loop indefinitely; we will break only when the lock is acquired
// or if we are interrupted.
while (true)
{
// Debug logging to report lock contention.
if (debug)
{
status.logContention(requested);
}
// Report the conflict to the user.
if (!isHeadless() && !clearMsg)
{
Iterator<SessionToken> iter = status.lockers();
T table = status.getRecordIdentifier().getTable();
StringBuilder buf = new StringBuilder(table.toString());
buf.append(" in use by ");
for (int i = 0; iter.hasNext(); i++)
{
if (i > 0)
{
buf.append(", ");
}
SessionToken next = iter.next();
String subject = next.getSubject();
buf.append(subject);
}
buf.append(". Wait or press CTRL-C to stop. (121)");
LogicalTerminal.statusDefault(buf.toString(), true);
clearMsg = true;
}
// We are here because another context holds either a share or an
// exclusive lock. Wait for such lock to be released or downgraded...
status.wait(timeout);
// Determine current lock type. In a lock transition, it should only be NONE or
// SHARE at this point, since to get here, we had to have released or downgraded
// the lock (unless we timed out or had a spurious wake-up).
current = status.getLockType();
LockType oldType = current;
// if current lock is exclusive, we timed out or spuriously woke up
if (!current.isExclusive())
{
// Set the lock type if we're the first one in...
if (LockType.NONE.equals(current) || status.isOnlyLocker(locker))
{
if (!update)
{
// We only wanted to know *if* we could get the lock, so we're done.
break;
}
// Set the current lock type to the requested lock type.
status.setLockType(requested);
current = requested;
}
// ...then acquire the lock, if possible. This test presumes that
// the current lock type will only be EXCLUSIVE if the previous
// block has set it to such on behalf of the requesting context.
if (current.equals(requested) &&
(LockType.SHARE.equals(current) ||
status.getLockerCount() == 0 ||
status.isOnlyLocker(locker)))
{
if (update && (status.addLocker(locker) || oldType != requested))
{
notifyLockChange(locker,
status.getRecordIdentifier(),
oldType,
requested,
context.get().osUserName);
}
// Got it; we're done waiting.
break;
}
}
// check whether timeout (if any) has elapsed; if so, throw LockTimeoutException
if (waitUntil > 0L && System.nanoTime() >= waitUntil)
{
throw new LockTimeoutException(requested, status.getRecordIdentifier());
}
// otherwise, continue waiting (includes possible spurious wake-up)...
}
if (start > -1L)
{
// Report how long we had to wait to acquire the lock, if it was
// beyond the report threshold.
long elapsed = System.currentTimeMillis() - start;
if (elapsed > warnThreshold)
{
RecordIdentifier<T> ident = status.getRecordIdentifier();
logTransition(requested, current, database, ident, elapsed);
}
}
}
catch (InterruptedException exc)
{
ErrorManager.handleStopException(new LockConflictConditionException(exc));
}
finally
{
if (clearMsg)
{
LogicalTerminal.statusDefault();
}
}
}
/**
* Notify a listener, if one is set, of a lock change event. The listener should return as
* quickly as possible to minimize any performance impact. If no listener is set, this method
* returns immediately.
*
* @param locker
* The current locking context.
* @param ident
* Identifier of record whose lock status has changed.
* @param oldType
* Lock type before the change.
* @param newType
* Lock type after the change.
*/
private void notifyLockChange(SessionToken locker,
RecordIdentifier<T> ident,
LockType oldType,
LockType newType,
String osUserName)
{
if (listener == null)
{
return;
}
// cast to RecordIdentifier<String> is ok; the only listener for this event expects this type
@SuppressWarnings("unchecked")
RecordLockEvent evt = new RecordLockEvent(nextEventID++,
locker,
(RecordIdentifier<String>) ident,
oldType,
newType);
evt.setOsUserName(osUserName);
listener.lockChanged(evt);
}
/**
* Context local lock manager state. Stores locker's session ID and tracks
* the record locks held by the current context.
*/
private static class Context<T extends Serializable>
{
/** Session token of the current context */
private final SessionToken locker;
/** IDs of all records currently locked by this context, organized by table */
private final Map<T, Map<RecordIdentifier<T>, LockStatus<T>>> locked;
/** The OS name of the user who generated this lock event. */
private final String osUserName;
/**
* Constructor.
*
* @param offline
* <code>true</code> for offline mode; <code>false</code> if
* running in the context of a P2J server process.
*/
Context(boolean offline)
{
if (offline)
{
locker = new SessionToken(-1, Thread.currentThread().getName());
}
else
{
locker = SecurityManager.getInstance().sessionSm.getSessionToken();
}
ClientParameters cp = StandardServer.getClientParameters();
osUserName = (cp == null) ? null : cp.osUserName;
locked = new HashMap<>();
}
}
/**
* Helper class which indicates the status of a locked record. This
* includes:
* <ul>
* <li>the type of lock;
* <li>the set of locking context IDs (may be multiple for a share lock,
* must be singular for an exclusive lock);
* <li>a <code>RecordIdentifier</code> object which uniquely identifies
* the locked record.
* </ul>
* If a lock request must block waiting for a lock to be released, the
* current thread synchronizes on instances of this class. These objects
* are stored in the lock manager's master lock table, keyed by
* <code>RecordIdentifier</code> objects.
*/
private static class LockStatus<T extends Serializable>
{
/** Database */
private final Database database;
/** Unique identifier of record associated with this status object */
private final RecordIdentifier<T> ident;
/** This field is used as long as a single session locks this record, to avoid a set instantiation. */
private SessionToken singleLocker = null;
/** The set of IDs of the lockers of the record */
private Set<SessionToken> lockers = null;
/** Threshold in milliseconds above which a lock hold time is reported */
private final long warnThreshold;
/** The type of lock being held on the record */
private LockType lockType = LockType.NONE;
/** Timestamp at which lock was first obtained, for admin purposes */
private long lockAdminTime = -1L;
/** Timestamp at which lock was first obtained, for warning purposes */
private long lockWarnTime = -1L;
/**
* Constructor.
*
* @param ident
* Unique identifier of record associated with this status
* object.
* @param lockType
* Type of lock initially held on the record.
*
* @throws NullPointerException
* if <code>locker</code> is <code>null</code>.
*/
LockStatus(Database database, RecordIdentifier<T> ident, LockType lockType, long warnThreshold)
{
this.database = database;
this.ident = ident;
this.warnThreshold = warnThreshold;
setLockType(lockType);
}
/**
* Constructor.
*
* @param ident
* Unique identifier of record associated with this status
* object.
* @param lockType
* Type of lock initially held on the record.
* @param locker
* the initial locking context.
*
* @throws NullPointerException
* if <code>locker</code> is <code>null</code>.
*/
LockStatus(Database database,
RecordIdentifier<T> ident,
LockType lockType,
SessionToken locker,
long warnThreshold)
{
this(database, ident, lockType, warnThreshold);
if (locker == null)
{
throw new NullPointerException("Locker ID must be non-null");
}
addLocker(locker);
}
/**
* Compose a string representation of this object for assistance in
* debugging and testing.
*
* @return String representation of this object's internal state.
*/
public String toString()
{
StringBuilder buf = new StringBuilder(lockType.toString());
buf.append(" [");
Iterator<SessionToken> iter = lockers();
for (int i = 0; iter.hasNext(); i++)
{
SessionToken next = iter.next();
if (i > 0)
{
buf.append(' ');
}
buf.append(next);
}
buf.append("]");
return buf.toString();
}
/**
* Remove a locking context from the set of contexts which currently
* hold locks on the backing record. If this was the last locking
* context in the set, reset the lock type to <code>LockType.NONE</code>
* and notify any threads which are blocked on this object to resume
* processing.
*
* @param locker
* Unique identifier of locking context to be removed.
*/
protected void removeLocker(SessionToken locker)
{
boolean setToNone = false;
if (lockers == null)
{
if (singleLocker == locker)
{
singleLocker = null;
setToNone = true;
}
else
{
LOG.log(Level.SEVERE, "Inconsistent locking? Trying to remove locker which was not registered.");
}
}
else
{
lockers.remove(locker);
if (lockers.isEmpty())
{
setToNone = true;
lockers = null;
}
}
if (setToNone)
{
// Reset lock type to none.
setLockType(LockType.NONE);
}
// Notify all waiters to resume processing.
notifyAll();
}
/**
* Compose a string representation of this object for assistance in
* debugging and testing.
*
* @return String representation of this object's internal state.
*/
protected String toStringVerbose()
{
return toString();
}
/**
* Store this object's information within a {@link
* com.goldencode.p2j.admin.RecordLockInfo RecordLockInfo} object for administrative purposes.
*
* @return Lock information for this lock.
*/
RecordLockInfo getLockInfo()
{
RecordLockType type = null;
if (lockType.isExclusive())
{
type = RecordLockType.EXCLUSIVE;
}
else if (lockType.isShare())
{
type = RecordLockType.SHARE;
}
else
{
type = RecordLockType.NONE;
}
SessionToken[] tokens = lockers != null
? lockers.toArray(new SessionToken[0])
: singleLocker == null
? new SessionToken[0]
: new SessionToken[] { singleLocker } ;
return new RecordLockInfo(type,
lockAdminTime,
database.getName(),
ident.getTable().toString(),
ident.getRecordID(),
tokens,
!database.isTenant());
}
/**
* Log debugging information about a lock acquisition request which
* causes contention with one or more existing lock holders.
*
* @param requested
* Lock type which was requested.
*/
void logContention(LockType requested)
{
StringBuilder buf = new StringBuilder("[");
buf.append(Utils.describeContext());
buf.append("] --> ");
buf.append(database);
buf.append(".");
buf.append(ident);
buf.append(" LOCK CONTENTION; requested ");
buf.append(requested);
buf.append(" at location:");
printStackTrace(buf, new Throwable());
buf.append(System.lineSeparator());
buf.append("current status: ");
buf.append(toStringVerbose());
String msg = buf.toString();
LOG.log(Level.FINE, msg);
}
/**
* Retrieve the object which uniquely identifies the record represented
* by this lock status object.
*
* @return Lock record object.
*/
RecordIdentifier<T> getRecordIdentifier()
{
return ident;
}
/**
* Get an iterator on the lockers currently holding this lock.
*
* @return Iterator of <code>SessionToken</code>s.
*/
Iterator<SessionToken> lockers()
{
if (lockers != null)
{
return lockers.iterator();
}
return singleLocker == null ? Collections.emptyIterator() : Arrays.asList(singleLocker).iterator();
}
/**
* Add a locking context to the set of contexts which currently hold
* locks on the backing record. This may be multiple for share locks,
* and must be a single context for exclusive locks (though this
* restriction is not enforced here).
*
* @param locker
* Unique identifier of locking context to be added.
*
* @return <code>true</code> if <code>locker</code> was added; <code>false</code> if
* <code>locker</code> already was in the set of lockers.
*/
boolean addLocker(SessionToken locker)
{
if (lockers == null)
{
if (singleLocker == null)
{
singleLocker = locker;
return true;
}
else if (singleLocker == locker)
{
return false;
}
else
{
lockers = Collections.newSetFromMap(new IdentityHashMap<>());
lockers.add(singleLocker);
lockers.add(locker);
singleLocker = null;
return true;
}
}
else
{
return lockers.add(locker);
}
}
/**
* Retrieve the current lock type applied to the backing record.
*
* @return Lock type.
*/
LockType getLockType()
{
return lockType;
}
/**
* Set the current lock type applied to the backing record.
*
* @param lockType
* Lock type.
*
* @return <code>true</code> to indicate lock type was changed; <code>false</code> to
* indicate <code>lockType</code> was the same as the existing lock type.
*/
boolean setLockType(LockType lockType)
{
if (LockType.NONE.equals(this.lockType) && !LockType.NONE.equals(lockType))
{
lockAdminTime = System.currentTimeMillis();
}
if (warn && warnThreshold >= 0L)
{
recordOrReportTime(this.lockType, lockType);
}
LockType newType = lockType.toWaitVariant();
boolean different = (!this.lockType.equals(newType));
this.lockType = newType;
return different;
}
/**
* Indicate whether the specified locking context is among those which
* currently lock the backing record.
*
* @param locker
* Unique context identifier to check.
*
* @return <code>true</code> if <code>locker</code> is among the
* records current lockers, else <code>false</code>.
*/
boolean isLockedBy(SessionToken locker)
{
return singleLocker == locker || (lockers != null && lockers.contains(locker));
}
/**
* Determine whether the specified locker is the only context which
* currently holds a lock on the backing record.
*
* @param locker
* Unique identifier of the locking context to be checked.
*
* @return <code>true</code> if the given locker is the only entity
* holding a lock on the backing record, else
* <code>false</code>.
*/
boolean isOnlyLocker(SessionToken locker)
{
return singleLocker == locker || (lockers != null && lockers.size() == 1 && lockers.contains(locker));
}
/**
* Determine the number of contexts which currently hold a lock on the
* backing record.
*
* @return 0 for an unlocked record; 1 for an exclusive lock; 1 or
* more for a share lock.
*/
int getLockerCount()
{
return lockers == null ? singleLocker == null ? 0 : 1 : lockers.size();
}
/**
* Given an old lock type and a new lock type, either record the current
* time, or check whether the elapsed time since the last recorded time
* exceeds the reporting threshold, and write a warning to the log. If
* the old and new lock types are the same, do nothing.
* <p>
* If the new lock type is more restrictive than the old, we record the
* current time. If the new lock type is less restrictive than the old,
* we check the threshold and possibly report the warning. The idea is
* to report when a lock is potentially blocking other access for longer
* than expected.
*
* @param oldType
* Old lock type.
* @param newType
* New lock type.
*/
private void recordOrReportTime(LockType oldType, LockType newType)
{
int comp = oldType.compareTo(newType);
if (comp < 0)
{
// New type is more restrictive.
lockWarnTime = lockAdminTime;
}
else if (comp > 0)
{
// New type is less restrictive.
long elapsed = System.currentTimeMillis() - lockWarnTime;
if (elapsed > warnThreshold)
{
logTransition(newType, oldType, database, ident, elapsed);
}
}
}
}
/**
* An extension of <code>LockStatus</code> used for trace-level debug
* logging. In addition to the information tracked by the superclass, this
* class maps stack trace information for the point of lock acquisition to
* each lock holder. This information is used to debug lock contention
* issues. It is only used when logging for the <code>LockManager</code>
* is set to FINEST.
*/
private class TraceLockStatus<R extends Serializable>
extends LockStatus<R>
{
/** Map of locker tokens to Throwables containing stack data */
private final Map<SessionToken, Throwable> traces = new LinkedHashMap<>();
/**
* Constructor.
*
* @param ident
* Unique identifier of record associated with this status
* object.
* @param lockType
* Type of lock initially held on the record.
* @param locker
* the initial locking context.
*
* @throws NullPointerException
* if <code>locker</code> is <code>null</code>.
*/
TraceLockStatus(Database database,
RecordIdentifier<R> ident,
LockType lockType,
SessionToken locker,
long warnThreshold)
{
super(database, ident, lockType, warnThreshold);
if (locker == null)
{
throw new NullPointerException("Locker ID must be non-null");
}
addLocker(locker);
}
/**
* Compose a string representation of this object for assistance in debugging and testing.
*
* @return String representation of this object's internal state.
*/
protected String toStringVerbose()
{
StringBuilder buf = new StringBuilder(toString());
Iterator<Map.Entry<SessionToken, Throwable>> iter = traces.entrySet().iterator();
while (iter.hasNext())
{
Map.Entry<SessionToken, Throwable> next = iter.next();
SessionToken locker = next.getKey();
Throwable throwable = next.getValue();
buf.append("\nLocker ");
buf.append(locker);
buf.append(" acquisition location:");
printStackTrace(buf, throwable);
}
return buf.toString();
}
/**
* Remove a locking context from the set of contexts which currently
* hold locks on the backing record. If this was the last locking
* context in the set, reset the lock type to <code>LockType.NONE</code>
* and notify any threads which are blocked on this object to resume
* processing.
*
* @param locker
* Unique identifier of locking context to be removed.
*/
protected void removeLocker(SessionToken locker)
{
super.removeLocker(locker);
traces.remove(locker);
}
/**
* Add a locking context to the set of contexts which currently hold
* locks on the backing record. This may be multiple for share locks,
* and must be a single context for exclusive locks (though this
* restriction is not enforced here).
*
* @param locker
* Unique identifier of locking context to be added.
*
* @return <code>true</code> if <code>locker</code> was added; <code>false</code> if
* <code>locker</code> already was in the set of lockers.
*/
boolean addLocker(SessionToken locker)
{
boolean added = super.addLocker(locker);
Throwable throwable = new Throwable();
throwable.fillInStackTrace();
traces.put(locker, throwable);
return added;
}
}
/**
* Print command line usage instructions to the error console for the
* built-in test harness.
*/
private static void printUsage()
{
String[] staticText =
{
"Use: InMemoryLockManager <numThreads> <iterations> <idPoolSize> <maxWorkTime> <noWait>",
"",
"where:",
"",
" <numThreads> Number of concurrent test threads to launch",
" <iterations> Number of iterations performed by each thread",
" <idPoolSize> Size of pool from which record IDs are randomly chosen",
" <maxWorkTime> Maximum milliseconds to pause to simulate work time",
" <noWait> 'true' (omit quotes) to never block when locking a record",
" 'false' (omit quotes) to allow blocking",
" <reportID> (optional) Limit reporting to this record ID only",
" defaults to reporting on all record IDs"
};
LOG.info(StringHelper.arrayToString(staticText));
}
/**
* Command line test harness for this class. Launches a user-specified
* number of concurrent threads, each of which performs a user-specified
* number of iterations. Within each iteration, a record is locked with
* a randomly selected lock type (may be NONE, which indicates no lock).
* Upon acquiring the lock, the current thread sleeps for up to 3 seconds,
* to simulate work time, after which the lock is released. While the
* table names are fixed, the ID of the record to be locked is selected
* randomly from a pool size specified by the user. Smaller ID pools
* create more locking contention; larger pools create less.
* <p>
* Status messages indicating lock requests, acquisitions, and releases
* are emitted to the console, as are error messages (if there is locking
* contention and the user has specified that threads must not block).
*
* @param args
* The following command line arguments are expected:
* <ol>
* <li><b>numThreads:</b> Number of concurrent test threads to
* launch
* <li><b>iterations:</b> Number of iterations performed by
* each thread
* <li><b>idPoolSize:</b> Size of pool from which record IDs
* are randomly chosen
* <li><b>maxWorkTime:</b> Maximum number of milliseconds to
* pause after lock is acquired to simulate work time
* <li><b>noWait:</b> <code>true</code> to never block when
* locking a record; <code>false</code> to allow blocking
* <li><b>reportID:</b> (optional) Limit reporting to the given
* record ID; if not provided, activity on all IDs is
* reported.
* </ol>
*/
public static void main(String[] args)
{
try
{
int numThreads = Integer.parseInt(args[0]);
final int iterations = Integer.parseInt(args[1]);
final int idPoolSize = Integer.parseInt(args[2]);
final int maxWorkTime = Integer.parseInt(args[3]);
final boolean noWait = (Boolean.valueOf(args[4])).booleanValue();
final Long rptID = (args.length > 5 ? Long.parseLong(args[5]) : null);
if (rptID != null && (rptID < 1 || rptID > idPoolSize))
{
throw new IllegalArgumentException("reportID is outside the specified pool");
}
final LockManager<String> lm = new InMemoryLockManager<>();
lm.setDatabase(new Database("db"));
final long[] longest = new long[1];
final Set<Serializable> upgradeIDs = new HashSet<Serializable>();
Runnable r = new Runnable()
{
public void run()
{
for (int i = 0; i < iterations; i++)
{
String tName = Thread.currentThread().getName();
StringBuilder buf = new StringBuilder();
try
{
// Pseudo-random lock type.
LockType lockType = LockType.NONE;
double val = Math.random();
if (val >= 0.85)
{
lockType = (noWait ? LockType.EXCLUSIVE_NO_WAIT : LockType.EXCLUSIVE);
}
else if (val >= 0.25)
{
lockType = (noWait ? LockType.SHARE_NO_WAIT : LockType.SHARE);
}
// Pseudo-random record ID.
val = Math.random() * idPoolSize + 1;
Long recID = Long.valueOf((long) val);
// Pseudo-random work time millisecond pause.
int workTime = (int) (Math.random() * maxWorkTime + 1);
// Request a lock.
RecordIdentifier<String> ident = new RecordIdentifier<>("table", recID);
if (rptID == null || recID.equals(rptID))
{
buf.append(tName);
buf.append(" --> REQUEST LOCK ");
buf.append(lockType);
buf.append(" on ");
buf.append(ident);
synchronized (longest)
{
System.out.println(buf);
}
}
long start = System.currentTimeMillis();
lm.lock(lockType, ident, true);
long elapsed = System.currentTimeMillis() - start;
// Lock has been acquired.
if (rptID == null || recID.equals(rptID))
{
buf.setLength(0);
buf.append(Thread.currentThread().getName());
buf.append(" --> ACQUIRE LOCK ");
buf.append(lockType);
buf.append(" on ");
buf.append(ident);
buf.append(" (in ");
buf.append(elapsed);
synchronized (longest)
{
if (elapsed >= longest[0])
{
longest[0] = elapsed;
buf.append('*');
}
}
buf.append(" millis)");
synchronized (longest)
{
System.out.println(buf);
}
}
// Simulate work, then downgrade and/or release lock.
while (!LockType.NONE.equals(lockType))
{
// Sleep to simulate work time.
start = System.currentTimeMillis();
try
{
Thread.sleep(workTime);
}
catch (InterruptedException exc)
{
throw new EndConditionException(exc);
}
elapsed = System.currentTimeMillis() - start;
start = System.currentTimeMillis();
// Upgrade, downgrade or release lock.
if (!LockType.NONE.equals(lockType))
{
// Release request (default).
LockType requested = LockType.NONE;
if (lockType.isExclusive())
{
if (Math.random() > 0.5)
{
// Downgrade request.
requested = LockType.SHARE;
}
}
else if (lockType.isShare())
{
if (Math.random() > 0.5)
{
synchronized (longest)
{
if (!upgradeIDs.contains(recID))
{
// Upgrade request.
upgradeIDs.add(recID);
requested = LockType.EXCLUSIVE;
if (rptID == null ||
recID.equals(rptID))
{
buf.setLength(0);
buf.append(tName);
buf.append(" --> REQUEST UPGRADE ");
buf.append(lockType);
buf.append(" on ");
buf.append(ident);
buf.append(" (after holding for ");
buf.append(elapsed);
buf.append(" millis)");
System.out.println(buf);
}
}
}
}
}
lm.lock(requested, ident, true);
if (requested.isExclusive())
{
synchronized (longest)
{
upgradeIDs.remove(recID);
}
}
if (rptID == null || recID.equals(rptID))
{
buf.setLength(0);
buf.append(tName);
if (requested.isExclusive())
{
elapsed = System.currentTimeMillis() - start;
buf.append(" --> UPGRADE LOCK ");
buf.append(lockType);
buf.append(" on ");
buf.append(ident);
buf.append(" (in ");
buf.append(elapsed);
synchronized (longest)
{
if (elapsed >= longest[0])
{
longest[0] = elapsed;
buf.append('*');
}
}
buf.append(" millis)");
}
else
{
if (requested.isShare())
{
buf.append(" --> DOWNGRADE LOCK ");
}
else
{
buf.append(" --> RELEASE LOCK ");
}
buf.append(lockType);
buf.append(" on ");
buf.append(ident);
buf.append(" (after holding for ");
buf.append(elapsed);
buf.append(" millis)");
}
synchronized (longest)
{
System.out.println(buf);
}
}
lockType = requested;
}
}
}
catch (Exception exc)
{
buf.setLength(0);
buf.append(tName);
buf.append(" --> ");
buf.append(exc.getMessage());
synchronized (longest)
{
LOG.severe(buf.toString());
}
}
}
}
};
for (int i = 0; i < numThreads; i++)
{
(new Thread(r, "T" + i)).start();
}
}
catch (Exception exc)
{
printUsage();
LOG.severe("", exc);
}
}
/**
* Implementation of <code>LockAdministrator</code>.
*/
private class Admin
implements LockAdministrator
{
/**
* Gather record lock information for the physical database associated
* with the enclosing lock manager.
*
* @return A list of {@link com.goldencode.p2j.admin.RecordLockInfo
* RecordLockInfo} objects. If no locks currently are being
* held, the list will be empty, but it will not be
* <code>null</code>.
*/
public List<RecordLockInfo> collectRecordLockInfo()
{
List<RecordLockInfo> list = null;
synchronized (lockTable)
{
if (!lockTable.isEmpty())
{
list = new ArrayList<>(lockTable.size());
}
for (LockStatus<T> status : lockTable.values())
{
// only want active record locks, not released locks or table locks
if (LockType.NONE.equals(status.getLockType()) ||
status.getRecordIdentifier() instanceof TableIdentifier)
{
continue;
}
RecordLockInfo info = status.getLockInfo();
list.add(info);
}
}
if (list == null)
{
list = Collections.emptyList();
}
return list;
}
}
}