TxWrapper.java
/*
** Module : TxWrapper.java
** Abstract : management of the transaction-related state, where buffers are involved.
**
** Copyright (c) 2022-2025, Golden Code Development Corporation.
**
** -#- -I- --Date-- ----------------------------------------Description---------------------------------------
** 001 CA 20220906 Created initial version - extracted state which is transaction related and it involves
** buffers, to reduce the complexity of BufferManager scope notification.
** CA 20221006 The 'nearestExternal' support needs to be in BufferManager, as it is used for computing
** a scope in ChangeBroker - BufferManager is registered for external block scope.
** SVL 20230113 Improved performance by replacing some "for-each" loops with indexed "for" loops.
** 002 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 003 CA 20230607 TRANSACTION-MODE AUTO must preserve the tx in the root appserver block - this is required
** because in modes other than State-free, requests can be executed on the same agent who had
** TRANSACTION-MODE AUTO, and this must see as active the tx initially set.
** 004 DDF 20230627 Made FORCE_NO_UNDO_TEMP_TABLES non-final, renamed it.
** DDF 20230627 Made forceNoUndoTempTables accessible through a method.
** 005 AL2 20231013 Removed reclaiming of keys for _temp database. Reclaiming happens now inside FWD-H2.
** 006 CA 20231031 Added 'BlockDefinition' parameter to 'scopeStart'.
** 007 DDF 20231020 ErrorConditionException should be thrown when a ValidationException occurs, not create
** a new IllegalStateException.
** 008 RAA 20231220 Stored recordBuffer.getCurrentRecord() call into a variable.
** 009 AL2 20240310 Added AggresiveFlushWrapper. This is a proxy for the TxWrapper which only validates.
** 010 CA 20240314 Improved AggresiveFlushWrapper - commitable registration is done only for blocks which
** have a dirty buffer (directly or one of its inner blocks).
** 011 CA 20240315 Fixed regression in CA/010 - use tx level instead of dirtyBuffers size when cleaning the
** aggressive flusher registration.
** 012 CA 20240423 Performance optimization, resolve the loadedBuffers dictionary only once in scopeFinished.
** 013 DDF 20240508 Added deregisterGlobalSavepointHooks() method.
** 014 CA 20240809 Avoid iterator usage, for performance improvement.
** Populate the dirty and loaded buffers scopes only when data is added to them.
** 015 CA 20240827 Notify that a record has been flushed - for non-temp records, this will allow invalidation
** of the FastFindCache for that DMO.
** 021 OM 20240909 Improved Database API.
** 022 OM 20241128 Multi-tenant runtime support: selected the proper persistence context, eventually
** based on [sharedDb] parameter.
** 023 OM 20250130 A bit of code cleanup.
** 024 LS 20250403 Updated begin(), commit() and rollback() to handle the PersistenceException using DBUtils.
** 025 ES 20250424 Handle StopConditionException through 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;
import java.util.*;
import java.util.logging.*;
import com.goldencode.p2j.persist.dirty.*;
import com.goldencode.p2j.persist.lock.*;
import com.goldencode.p2j.persist.orm.*;
import com.goldencode.p2j.security.ContextLocal;
import com.goldencode.p2j.security.WeightFactor;
import com.goldencode.p2j.util.*;
import com.goldencode.p2j.util.ErrorManager;
import com.goldencode.p2j.util.logging.*;
import com.goldencode.util.MutableInteger;
/**
* An object which manages the "master" {@code Commitable} and {@code Finalizable} events of an
* application transaction. It creates a database transaction to correspond with the application
* transaction, and as it receives events from the {@code TransactionManager}, it delegates related
* responsibilities to various components of the persistence framework.
*/
class TxWrapper
implements Commitable,
Finalizable
{
/** Logger */
private static final CentralLogger LOG = CentralLogger.get(TxWrapper.class.getName());
/** Stores context local instances of this class */
private static final ContextLocal<WorkArea> context = new ContextLocal<WorkArea>()
{
@Override public WeightFactor getWeight() { return WeightFactor.LEVEL_2; }
@Override protected WorkArea initialValue() { return (new WorkArea()); }
};
/** Backing database */
private final Database database;
/** Dirty share context */
private final DirtyShareContext dirtyContext;
/** Savepoint manager (can be {@code null} */
private final SavepointManager savepointManager;
/** Persistence context, for access to database session and transaction */
private Persistence.Context persistenceCtx = null;
/** Roll back the transaction when block finishes, if not previously acted upon */
private boolean autoRollback = false;
/** {@code true} for an initial transaction only until the first commit or rollback */
private boolean initialTx = false;
/** The {@link WorkArea} instance. */
private final WorkArea wa;
/**
* The sole constructor.
*
* @param wa
* The {@link WorkArea} instance.
* @param database
* The {@code Database} where the transaction takes place.
*/
TxWrapper(WorkArea wa, Database database)
{
this.wa = wa;
this.database = database;
// Because the BufferManager manages all record buffers within the current user context, we cannot
// separate the DIRTY-READ from the normal tables at this moment. Since this instance is only used
// for final cleanup, we acquire the context for the DIRTY-READ database so that context will be
// cleaned up correctly. The non-DIRTY-READ information will be ignored.
this.dirtyContext = DirtyShareFactory.getContextInstance(database);
this.savepointManager = (database.isTemporary() && DatabaseManager.isForceNoUndoTempTables())
? null
: new SavepointManager();
// register for master transaction commit/rollback
wa.txHelper.registerTransactionCommit(this, true);
// register for master transaction finish/iterate
wa.txHelper.registerTransactionFinish(this, true);
}
/**
* Get the helper to access the {@link Commitable} and {@link Resettable} state required to maintain
* transaction-related state where buffers are involved.
*
* @param bm
* The {@link BufferManager} instance.
*
* @return The {@link TxWrapperHelper} instance.
*/
public static TxWrapperHelper getHelper(BufferManager bm)
{
WorkArea wa = context.get();
if (wa.bm == null)
{
wa.bm = bm;
}
return new TxWrapperHelper(wa);
}
/**
* Implementation of {@code Finalizable} interface: provides a notification that the block whose scope
* in which the object is registered is about to iterate and attempt another pass. This provides an
* opportunity to preserve state from the previous pass and make preparations for the coming pass.
*/
@Override
public void iterate()
{
if (!isActive())
{
return;
}
if (LOG.isLoggable(Level.FINE))
{
LOG.log(Level.FINE, message(persistenceCtx.getPersistence(), "master transaction iterate"));
}
end();
begin();
}
/**
* Implementation of {@code Finalizable} interface: provides a notification that the block whose scope
* in which the object is registered is about to retry the same iteration (all loop control data is
* unchanged). This provides an opportunity to clear or reset state from the previous pass.
*/
@Override
public void retry()
{
if (!isActive())
{
return;
}
if (LOG.isLoggable(Level.FINE))
{
LOG.log(Level.FINE, message(persistenceCtx.getPersistence(), "master transaction retry"));
}
end();
begin();
}
/**
* Implementation of {@code Finalizable} interface: Provides a notification that the scope in which the
* object is registered is ending and the object's reference may or may not be lost after this method
* is called. This provides a natural and standard mechanism to clean-up any resources that may need
* to be closed explicitly such as operating system files or sockets.
*/
@Override
public void finished()
{
try
{
if (autoRollback)
{
if (isActive() && LOG.isLoggable(Level.FINE))
{
LOG.log(Level.FINE, message(persistenceCtx.getPersistence(), "triggering auto-rollback"));
}
rollback(true);
}
}
finally
{
if (isActive() && LOG.isLoggable(Level.FINE))
{
LOG.log(Level.FINE, message(persistenceCtx.getPersistence(), "master transaction finished"));
}
wa.txWrapperMap.remove(database);
if (isActive())
{
end();
}
else
{
wa.inactiveTxWrappers.remove(database);
}
}
}
/**
* Implementation of {@code Finalizable} interface: provides a notification that the external program
* in which the object is registered is being deleted and the object's reference may be lost after this
* method is called.
*/
@Override
public void deleted()
{
deregisterGlobalSavepointHooks();
}
/**
* Implementation of {@code Commitable} interface: saves any temporary state to the persistent form or
* storage that is associated with this class.
*
* @param transaction
* {@code true} if this is a full transaction and {@code false} if this is only a
* sub-transaction (a nested scope with transaction support).
*/
@Override
public void commit(boolean transaction)
{
if (!isActive())
{
return;
}
if (LOG.isLoggable(Level.FINE))
{
LOG.log(Level.FINE, message(persistenceCtx.getPersistence(), "master transaction commit"));
}
try
{
initialTx = false;
autoRollback = false;
persistenceCtx.commit();
}
catch (PersistenceException exc)
{
if (LOG.isLoggable(Level.SEVERE))
{
LOG.log(Level.SEVERE, message(persistenceCtx.getPersistence(), "commit error"), exc);
}
DBUtils.handleException(database, exc);
// No good way to recover from this. Let it abend.
throw new RuntimeException(exc);
}
}
/**
* Implementation of {@code Commitable} interface: notifies that a rollback has just occurred (all state
* maintained through the {@link Undoable} interface has been already rolled back). This notification
* can be used to implement cleanup of any custom state that needs to be maintained in this case. It can
* also be used to implement a rollback that is completely independent of the {@code Undoable}
* (externally driven) interface.
*
* @param transaction
* {@code true} if this is a full transaction and {@code false} if this is only a
* sub-transaction (a nested scope with transaction support).
*/
@Override
public void rollback(boolean transaction)
{
if (!isActive())
{
return;
}
if (LOG.isLoggable(Level.FINE))
{
LOG.log(Level.FINE, message(persistenceCtx.getPersistence(), "master transaction rollback"));
}
if (database.isTemporary() && savepointManager == null)
{
// we force all temp-tables to be no-undo, so instead of rollback, do a
// commit of the current transaction
commit(transaction);
return;
}
initialTx = false;
autoRollback = false;
try
{
persistenceCtx.rollback(false);
}
catch (PersistenceException exc)
{
if (LOG.isLoggable(Level.SEVERE))
{
LOG.log(Level.SEVERE, message(persistenceCtx.getPersistence(), "rollback error"), exc);
}
DBUtils.handleException(database, exc);
// No good way to recover from this. Let it abend.
throw new RuntimeException(exc);
}
}
/**
* Implementation of {@code Commitable} interface: Notifies that a block is about to be normally exited
* or normally iterated and that any deferred validation processing should be executed.
* <p>
* This particular implementation is a NOP.
*
* @param transaction
* Ignored.
* @param aggressiveFlush
* Ignored.
*/
@Override
public void validate(boolean transaction, boolean aggressiveFlush)
{
// no-op
}
/**
* Registers the commitable and finalizable for savepoint manager, if any.
*
* @param fullTx
* {@code true} if this is a full-transaction.
* @param blockDepth
* zero-based depth of target block, starting from the outermost block. <code>0</code> indicates
* the global block.
*/
void registerSavepointHooks(boolean fullTx, int blockDepth)
{
if (savepointManager != null)
{
savepointManager.registerBlockHooks(wa.txHelper, fullTx, blockDepth);
}
}
/**
* Deregisters the finalizable for savepoint manager from the global block, if any.
*/
void deregisterGlobalSavepointHooks()
{
if (savepointManager != null)
{
savepointManager.deregisterGlobalBlockHooks(wa.txHelper);
}
}
/**
* Activates the transaction. This entails starting a database-level transaction and activating a
* savepoint manager for that transaction.
*/
void activate()
{
if (isActive())
{
return;
}
Persistence persistence = PersistenceFactory.getInstance(database);
persistenceCtx = persistence.getContext(!database.isTenant());
if (LOG.isLoggable(Level.FINE))
{
LOG.log(Level.FINE,
message(persistenceCtx.getPersistence(),
"registered for master transaction commit/finished"));
persistenceCtx.enableTransactionTrace(true);
}
begin();
initialTx = true;
}
/**
* Checks whether this {@code TxWrapper} is active.
*
* @return {@code true} only if {@code TxWrapper} was activated.
*/
boolean isActive()
{
return persistenceCtx != null;
}
/**
* Convert the current object to its string representation to be used during debugging.
*
* @return A short string representation of this object to be used while debugging.
*/
@Override
public String toString()
{
return "TxWrapper{database=" + database + '}';
}
/**
* Starts the actual transaction at persistence level.
*/
private void begin()
{
try
{
if (!persistenceCtx.beginTransaction(savepointManager) && !initialTx)
{
// We are here because we tried to begin a database level transaction and one is
// already open. This situation generally is invalid, except in an edge case
// where a new transaction is opened while in TM's iterate worker. This can happen
// if, for example, a new database is accessed during the execution of a trigger
// fired during iterate processing. The only case we have seen of this so far is
// due to a record being released during RB's iterate processing, which then fires
// a write trigger, which accesses a buffer in a separate database (e.g., for
// audit purposes) which did not previously have a transaction open. This in turn
// invokes this object's c'tor, which registers for master finish notification,
// but also opens a new transaction. We then get here a moment later during master
// finish processing. Commit has not yet been called, since commitable processing
// for that block already has occurred. The solution is to check the initialTx
// flag and simply ignore the fact that a new transaction could not be created if
// that flag is true. The flag will be set to false on the first commit or
// rollback.
Persistence persistence = persistenceCtx.getPersistence(); // same persistence
if (LOG.isLoggable(Level.FINE))
{
Throwable beginTxTrace = persistenceCtx.getBeginTxTrace();
if (beginTxTrace != null)
{
String msg = "Current transaction previously started at location...";
LOG.log(Level.FINE, persistence.message(msg), beginTxTrace);
}
}
throw new IllegalStateException(
message(persistence, "database transaction already active"));
}
autoRollback = true;
}
catch (PersistenceException exc)
{
DBUtils.handleException(database, exc);
ErrorManager.handleStopException(new StopConditionException(exc));
}
}
/**
* Shuts down the transaction at persistence level and performs key management.
*/
private void end()
{
// clear nursery
persistenceCtx.getNursery().transactionEnded();
// clean up dirty shares for this transaction
if (dirtyContext != null)
{
dirtyContext.cleanup();
}
// release or downgrade record locks which may have been acquired during the transaction
RecordLockContext lockCtx = persistenceCtx.getRecordLockContext();
Set<Long> released = lockCtx.transactionEnded();
// reclaim any primary keys for records deleted and committed or created and rolled back
// during the transaction, provided those keys have been fully released (not just
// downgraded) by the end of the transaction; note that this may "leak" reclaiming some
// primary keys downgraded and still held as share locks by open buffers, for buffers
// whose scope extends beyond the end of the transaction
wa.bm.reclaimPendingKeys(persistenceCtx.getPersistence(), released);
// Allow the connection manager to perform its end of transaction cleanup. This must
// occur after all transaction processing, because it may result in tearing down the
// database's session factory.
wa.connMgr.transactionEnded(database);
}
/**
* Compose a message suitable for logging. Reports open buffer scopes and
* context information in addition to a custom message.
*
* @param persistence
* Persistence object which will provide context info.
* @param text
* Custom message text.
*
* @return Composed message text.
*/
private String message(Persistence persistence, String text)
{
return persistence.message("scope " + wa.bm.getOpenBufferScopes() + ": " + text);
}
/**
* Helper instance to access the {@link WorkArea} state without context-local lookups.
*/
static class TxWrapperHelper
{
/** The {@link WorkArea} instance. */
private final WorkArea wa;
/**
* Initialize this helper.
*
* @param wa
* The {@link WorkArea} instance.
*/
public TxWrapperHelper(WorkArea wa)
{
this.wa = wa;
}
/**
* Register this buffer as 'dirty' in the current scope.
* <p>
* This registration doesn't mean that at the time of {@link #commit}, {@link #rollback}, or
* {@link #validate} the buffer's record is still 'dirty' - this is used to limit the set of buffers
* receiving such notifications, to avoid iterating over all {@link BufferManager#openBuffers}.
*
* @param recordBuffer
* The record buffer.
*/
public void registerDirtyBuffer(RecordBuffer recordBuffer)
{
wa.registerDirtyBuffer(recordBuffer);
}
/**
* Notification that a record has been flushed.
*
* @param buffer
* The buffer instance.
* @param dmoId
* The DMO id.
*/
void recordFlushed(RecordBuffer buffer, int dmoId)
{
if (buffer.isTemporary())
{
return;
}
FastFindCache ffCache = buffer.getFastFindCache();
Set<Long> dmoIds = wa.ffCacheChanges.computeIfAbsent(ffCache, (k) -> new HashSet<>());
dmoIds.add(FastFindCache.combine(dmoId, null));
}
/**
* Deregister the given, dynamic record buffer by removing it from the buffer manager's tracking.
*
* @param buffer
* Dynamic record buffer to be removed.
*/
void deregisterDynamicBuffer(RecordBuffer buffer)
{
wa.deregisterDynamicBuffer(buffer);
}
/**
* Executes logic to explicitly begin a transaction.
*
* @param inTx
* Flag indicating if we are in a transaction.
* @param fullTx
* Flag indicating if we are in a full transaction.
* @param blockDepth
* The current block depth (all blocks tracked by the {@code TransactionManager}).
*/
public void beginTx(boolean inTx, boolean fullTx, int blockDepth)
{
wa.beginTx(inTx, fullTx, false, blockDepth);
}
/**
* Executes logic to explicitly end a transaction.
*
* @param inTx
* Flag indicating if we are in a transaction.
* @param fullTx
* Flag indicating if we are in a full transaction.
*/
public void endTx(boolean inTx, boolean fullTx)
{
wa.endTx(inTx, fullTx);
}
/**
* Executes logic required after a full transaction has ended.
*/
public void endTxPost()
{
// if any DBs were used in this tx block and remained connected, disconnect now
wa.endTxPost();
}
/**
* Check whether the current block is within an application-level transaction.
*
* @return {@code true} if we are in a transaction, else {@code false}.
*/
public boolean isTransaction()
{
return wa.isTransaction();
}
/**
* Get the number of block scope transitions made. This is meant to be a semaphore of sorts, to allow
* a caller to determine whether this value has changed since some previous point when it was last
* retrieved. Since it is only meant to detect that the current block has changed over time, we do not
* protect against overflow.
*
* @return The current unique block ID.
*/
public long getScopeTransitions()
{
return wa.getScopeTransitions();
}
/**
* Get the current counter value of application-leve transactions which have been opened.
*
* @return Transaction count.
*/
public long getTransactionCount()
{
return wa.getTransactionCount();
}
/**
* Track the buffer's database. This will increment the {@link WorkArea#activeDatabases} counter when
* a new buffer is activated, and decrement it when it exists its scope.
*
* @param buffer
* The buffer to track.
* @param add
* Flag indicating if the buffer is opening or closing the scope.
*
* @throws IllegalStateException
* If we are removing an active buffer and there is no entry for it in
* {@link WorkArea#activeDatabases}.
*/
public void trackDatabase(RecordBuffer buffer, boolean add)
{
wa.trackDatabase(buffer, add);
}
/**
* Given a database which has one or more buffers with an open scope operating within an application
* level transaction, possibly activate the transaction wrapper associated with that database.
* Activation entails starting a database-level transaction and activating a savepoint manager
* for that transaction.
* <p>
* If the transaction wrapper for the given database already is active, do nothing.
*
* @param database
* Database with at least one buffer with an open scope.
*/
public void maybeActivateTxWrapper(Database database)
{
wa.maybeActivateTxWrapper(database);
}
/**
* Indicate whether the current block is "important" from the standpoint of transitioning into
* or out of its scope. A block is considered "important" in this sense if it is either
* completely outside of an application transaction, or if it is within a transaction AND has transaction
* block properties (i.e., is not a NO_TRANSACTION block).
* <p>
* This notion of importance is used to determine whether the various scoped data structures managed by
* this class need to add or delete a scope when transitioning into or out of the block, respectively,
* and whether we need to perform processing (e.g., copying/moving data between scopes) associated with
* such transitions.
*
* @return {@code true} if the block is considered "important" by the above criteria, else
* {@code false}.
*/
public boolean isImportantBlockTransition()
{
return wa.isImportantBlockTransition();
}
/**
* Register the {@link WorkArea} scopeable as pending, to be added to the next block.
*/
public void registerPendingScopeable()
{
wa.txHelper.registerPendingScopeable(wa);
}
/**
* Notify the buffer manager that a new record was loaded (or the old record was reloaded)
* into the specified record buffer.
*
* @param recordBuffer
* The record buffer in which a record was loaded (or the old record was reloaded).
*/
public void notifyRecordWasLoaded(RecordBuffer recordBuffer)
{
wa.notifyRecordWasLoaded(recordBuffer);
}
/**
* Remove all the buffers from the dirty buffers scope.
*
* @param allBuffers
* The buffers to remove.
*/
public void cleanDirtyBuffers(Map<String, RecordBuffer> allBuffers)
{
for (RecordBuffer buf : allBuffers.values())
{
// remove it from the dirty buffers
wa.dirtyBuffers.removeEntryThroughScope(buf, wa.dirtyBuffers.size());
}
}
}
/**
* Manages transaction-related state where buffers are involved. State and methods are extracted from
* {@link BufferManager}, and these require scope notification and management for each
* {@link #isImportantBlockTransition() important block}.
*/
private static class WorkArea
implements Commitable,
Resettable,
Scopeable
{
/** Database to transaction wrapper instance map */
private final Map<Database, TxWrapper> txWrapperMap = new HashMap<>();
/** Transaction helper */
private final TransactionManager.TransactionHelper txHelper;
/** Helper to use the ProcedureManager without any context local lookups. */
private final ProcedureManager.ProcedureHelper pm = ProcedureManager.getProcedureHelper();
/** The buffer manager instance. */
private BufferManager bm = null;
/** Connection manager */
private final ConnectionManager connMgr;
/** Transaction wrappers not yet activated */
private final Map<Database, TxWrapper> inactiveTxWrappers = new HashMap<>();
/** Transaction finalizable to manage transaction state locally (avoid context-local calls) */
private final Finalizable xactFin = new Finalizable()
{
@Override public void finished() { transactionDepth = -1; invalidateFastFindCache(); }
@Override public void deleted() { invalidateFastFindCache(); }
@Override public void iterate() { invalidateFastFindCache(); }
@Override public void retry() { invalidateFastFindCache(); }
};
/** Context-local access to unique constraint tracker */
private final UniqueTracker.Context uniqueTrackerCtx = UniqueTracker.getContext();
/**
* The real scope depth counter, as {@link #loadedBuffers} and {@link #dirtyBuffers} push scopes only
* when they are used.
*/
private int scopeDepth = 0;
/** Bitset with a flag indicating which scope was not yet loaded, for {@link #loadedBuffers}. */
private final BitSet emptyLoadedScopes = new BitSet();
/** Bitset with a flag indicating which scope was not yet loaded, for {@link #dirtyBuffers}. */
private final BitSet emptyDirtyScopes = new BitSet();
/** Counter of the number of scope transitions (push or pop) */
private long scopeTransitions = 0L;
/** Block depth at which full application transaction began, or -1 if not in a transaction */
private int transactionDepth = -1;
/** Counter of the number of application-level transactions */
private long transactionCount = 0L;
/** Counter of in-use buffers for each database. */
private final Map<Database, MutableInteger> activeDatabases = new LinkedHashMap<>();
/**
* Scoped dictionary which key set contains all buffers into which a new record was loaded (or
* the old record was reloaded) into corresponding scopes. Value set doesn't contain any useful
* information and is used as a temporary data storage during copying from one scope level to
* another.
*/
private final ScopedDictionary<RecordBuffer, Boolean> loadedBuffers = new ScopedDictionary<>();
/**
* The buffers touched via a Create, Update, Copy or Delete operation, in each scope. If the full transaction
* block is not {@link #commit committed} or {@link #rollback rolled back}, then the buffers will be merged
* in the previous scope.
*/
private final TailScopedDictionary<RecordBuffer, Boolean> dirtyBuffers = new TailScopedDictionary<>();
/** The scopes which require an {@link AggresiveFlushWrapper} to be registered as commitable. */
private final BitSet aggressiveFlushScopes = new BitSet();
/** The singleton of {@link AggresiveFlushWrapper} for this instance. */
private final AggresiveFlushWrapper aggressiveFlusher = new AggresiveFlushWrapper(this);
/** {@link FastFindCache} instances which need to have the specified DMOs invalidated. */
private final Map<FastFindCache, Set<Long>> ffCacheChanges = new IdentityHashMap<>();
/**
* Initialize this instance.
*/
public WorkArea()
{
txHelper = TransactionManager.getTransactionHelper();
connMgr = ConnectionManager.get();
loadedBuffers.setIdentityKeys(true);
dirtyBuffers.setIdentityKeys(true);
txHelper.registerResettable(this);
}
/**
* A full transaction or sub-transaction is committing. Notify all dirty buffers in the current scope.
*
* @param transaction
* {@code true} if full transaction; {@code false} if sub-transaction.
*/
@Override
public void commit(boolean transaction)
{
boolean hasDirty = !emptyDirtyScopes.get(scopeDepth);
Map<RecordBuffer, Boolean> buffers = hasDirty ? dirtyBuffers.getDictionaryAtScope(0, false) : null;
if (buffers != null && !buffers.isEmpty())
{
// avoid capturing lambdas
if (transaction)
{
buffers.forEach((buf, b) -> buf.commit(true));
buffers.clear();
}
else
{
buffers.forEach((buf, b) -> buf.commit(false));
}
}
}
/**
* A full transaction or sub-transaction is rolling back. Notify all dirty buffers in the current scope.
*
* @param transaction
* {@code true} if full transaction; {@code false} if sub-transaction.
*/
@Override
public void rollback(boolean transaction)
{
boolean hasDirty = !emptyDirtyScopes.get(scopeDepth);
Map<RecordBuffer, Boolean> buffers = hasDirty ? dirtyBuffers.getDictionaryAtScope(0, false) : null;
if (buffers != null && !buffers.isEmpty())
{
// avoid capturing lambdas
if (transaction)
{
buffers.forEach((buf, b) -> buf.rollback(true));
buffers.clear();
}
else
{
buffers.forEach((buf, b) -> buf.rollback(false));
}
}
}
/**
* A block is exiting or iterating normally within a transaction. Notify all dirty buffers in the current
* scope.
*
* @param transaction
* {@code true} if full transaction; {@code false} if sub-transaction.
* @param aggressiveFlush
* {@code true} if transaction manager is in aggressive subtransaction
* flush mode, indicating that any transient buffers should be validated and
* flushed, regardless of other state.
*/
@Override
public void validate(boolean transaction, boolean aggressiveFlush)
{
boolean hasDirty = !emptyDirtyScopes.get(scopeDepth);
TailMap<RecordBuffer, Boolean> buffers = hasDirty ? dirtyBuffers.getDictionaryAtScope(0, false) : null;
if (buffers != null && !buffers.isEmpty())
{
ArrayList<Map<RecordBuffer, Boolean>> validatedAddedDirtyBuffers = null;
try (AutoCloseable itersafe = buffers.flagIterating())
{
for (RecordBuffer buf : buffers.keySet())
{
buf.validate(transaction, aggressiveFlush);
// The next block is rarely entered, only in edge cases.
// Details:
// The invocation of buf.validate can (for example) result in
// an ABL publish + subscribe (from an OEDB write trigger),
// resulting in more buffers that need to be validated.
// Since we are iterating, the extra buffers are added to
// a standalone 'tail' collection. We validate them immediately,
// and add them to dirtyBuffers after iterating has completed.
Map<RecordBuffer, Boolean> addedDirtyBuffers = buffers.pop();
while (addedDirtyBuffers != null)
{
// Validate all additions; avoid iterator usage
addedDirtyBuffers.forEach((v, b) -> v.validate(transaction, aggressiveFlush));
// Store the validated additions to add them afterwards
// A list is used for clarity and to facilitate possible
// removal logic in the future (which can't happen at the moment).
if (validatedAddedDirtyBuffers == null)
{
validatedAddedDirtyBuffers = new ArrayList<>();
}
validatedAddedDirtyBuffers.add(addedDirtyBuffers);
// Fetch possible new additions of the added validate invocations.
// We take control of this collection, at the TailMap it
// is forgotten.
addedDirtyBuffers = buffers.pop();
}
}
}
catch (ErrorConditionException e)
{
throw e;
}
catch (Exception e)
{
throw new IllegalStateException("Exception at dirtyBuffers.flagIterating block", e);
}
// The next block is rarely entered, only in edge cases, see above.
if (validatedAddedDirtyBuffers != null)
{
for (int i = 0; i < validatedAddedDirtyBuffers.size(); i++)
{
Map<RecordBuffer, Boolean> bufmap = validatedAddedDirtyBuffers.get(i);
// Note: addEntry(false, ...) because:
// We do not need the global scope.
// At the top of this method, scope 0 is used, not scope -1 (global).
bufmap.forEach((k, v) -> dirtyBuffers.addEntry(false, k, v));
}
}
// do not clear the buffers here, only commit/rollback can clear it
}
}
/**
* Register this buffer as 'dirty' in the current scope.
* <p>
* This registration doesn't mean that at the time of {@link #commit}, {@link #rollback}, or
* {@link #validate} the buffer's record is still 'dirty' - this is used to limit the set of buffers
* receiving such notifications, to avoid iterating over all {@link BufferManager#openBuffers}.
*
* @param recordBuffer
* The record buffer.
*/
public void registerDirtyBuffer(RecordBuffer recordBuffer)
{
if (!aggressiveFlushScopes.isEmpty())
{
for (int blockDepth = aggressiveFlushScopes.nextSetBit(0);
blockDepth >= 0;
blockDepth = aggressiveFlushScopes.nextSetBit(blockDepth + 1))
{
txHelper.registerCommitAt(blockDepth, aggressiveFlusher);
}
// registration is done only once for all scopes
aggressiveFlushScopes.clear();
}
checkDirtyScopes();
dirtyBuffers.addEntry(false, recordBuffer, false);
}
/**
* Deregister the given, dynamic record buffer by removing it from the buffer manager's tracking.
*
* @param buffer
* Dynamic record buffer to be removed.
*/
public void deregisterDynamicBuffer(RecordBuffer buffer)
{
int[] loadedBuffersScopes = buffer.getLoadedBuffersScope();
int loadedScopes = loadedBuffers.size();
for (int scope : loadedBuffersScopes)
{
int depth = loadedScopes - scope;
Map<RecordBuffer, Boolean> loaded = loadedBuffers.getDictionaryAtScope(depth, false);
if (loaded != null)
{
loaded.remove(buffer);
}
}
// remove from global scope always
Map<RecordBuffer, Boolean> loaded = loadedBuffers.getDictionaryAtScope(-1, false);
if (loaded != null)
{
loaded.remove(buffer);
}
}
/**
* Notify the buffer manager that a new record was loaded (or the old record was reloaded)
* into the specified record buffer.
*
* @param recordBuffer
* The record buffer in which a record was loaded (or the old record was reloaded).
*/
public void notifyRecordWasLoaded(RecordBuffer recordBuffer)
{
Record currentRecord = recordBuffer.getCurrentRecord();
if (currentRecord != null && !recordBuffer.isDynamic())
{
pm.notifyStaticRecordWasLoaded(recordBuffer);
}
if (currentRecord == null || recordBuffer.isTemporary())
{
return;
}
checkLoadedScopes();
loadedBuffers.addEntry(false, recordBuffer, false);
recordBuffer.addLoadedBuffersScope(loadedBuffers.size());
}
/**
* Get the number of block scope transitions made. This is meant to be a semaphore of sorts, to allow
* a caller to determine whether this value has changed since some previous point when it was last
* retrieved. Since it is only meant to detect that the current block has changed over time, we do not
* protect against overflow.
*
* @return The current unique block ID.
*/
public long getScopeTransitions()
{
return scopeTransitions;
}
/**
* Track the buffer's database. This will increment the {@link #activeDatabases} counter when a new buffer
* is activated, and decrement it when it exists its scope.
*
* @param buffer
* The buffer to track.
* @param add
* Flag indicating if the buffer is opening or closing the scope.
*
* @throws IllegalStateException
* If we are removing an active buffer and there is no entry for it in {@link #activeDatabases}.
*/
public void trackDatabase(RecordBuffer buffer, boolean add)
{
if (add)
{
Database database = buffer.getDatabase();
MutableInteger count = activeDatabases.get(database);
if (count == null)
{
activeDatabases.put(database, count = new MutableInteger(0));
}
count.set(count.get() + 1);
}
else if (buffer.isActive() && buffer.getOpenScopeCount() > 0)
{
Database database = buffer.getDatabase();
MutableInteger count = activeDatabases.get(database);
if (count == null)
{
throw new IllegalStateException("Tracking for " + database + " is unbalanced!");
}
if (count.get() == 1)
{
activeDatabases.remove(database);
}
else
{
count.set(count.get() - 1);
}
}
}
/**
* Given a database which has one or more buffers with an open scope operating within an application
* level transaction, possibly activate the transaction wrapper associated with that database.
* Activation entails starting a database-level transaction and activating a savepoint manager
* for that transaction.
* <p>
* If the transaction wrapper for the given database already is active, do nothing.
*
* @param database
* Database with at least one buffer with an open scope.
*/
public void maybeActivateTxWrapper(Database database)
{
TxWrapper txw = inactiveTxWrappers.remove(database);
if (txw != null)
{
txw.activate();
}
}
/**
* Check whether the current block is within an application-level transaction.
*
* @return {@code true} if we are in a transaction, else {@code false}.
*/
public boolean isTransaction()
{
return transactionDepth >= 0;
}
/**
* Get the current counter value of application-leve transactions which have been opened.
*
* @return Transaction count.
*/
public long getTransactionCount()
{
return transactionCount;
}
/**
* Create an inactive transaction wrapper for the given database. It will be activated the
* first time an open buffer scope is detected or when a new buffer scope is opened, at which
* time a database-level transaction will be opened.
*
* @param database
* Database for which this transaction wrapper is created.
*
* @return Transaction wrapper object.
*/
private TxWrapper createTxWrapper(Database database)
{
TxWrapper txw = new TxWrapper(this, database);
txWrapperMap.put(database, txw);
inactiveTxWrappers.put(database, txw);
return txw;
}
/**
* Executes logic to explicitly begin a transaction.
*
* @param inTx
* Flag indicating if we are in a transaction.
* @param fullTx
* Flag indicating if we are in a full transaction.
* @param aggresiveFlush
* Flag indicating if the block isn't a transaction but it still needs to aggressively flush.
* @param blockDepth
* The current block depth (all blocks tracked by the {@code TransactionManager}).
*/
public void beginTx(boolean inTx, boolean fullTx, boolean aggresiveFlush, int blockDepth)
{
if (inTx)
{
if (fullTx)
{
// create a database xaction wrapper for each connected database, but do not
// activate it yet
txWrapperMap.clear(); // should be empty already
inactiveTxWrappers.clear(); // should be empty already
ArrayList<Database> databases = connMgr.getActiveDatabases();
for (int i = 0; i < databases.size(); i++)
{
Database database = databases.get(i);
createTxWrapper(database);
}
transactionDepth = blockDepth - 1; // match TM transaction level
transactionCount++;
// register for master transaction finish/iterate to clear transaction depth
txHelper.registerTransactionFinish(xactFin, true);
}
// register various services for callbacks if current block has transaction properties
if (txHelper.currentTransactionLevel() != TransactionManager.NO_TRANSACTION)
{
// register savepoint manager for callbacks at full and sub-transaction blocks
for (TxWrapper txw : txWrapperMap.values())
{
txw.registerSavepointHooks(fullTx, blockDepth - 1);
}
// register to get Commitable notifications for open buffers
txHelper.registerCommitAt(blockDepth - 1, this);
// register unique tracker context to get Commitable/Finalizable notifications
uniqueTrackerCtx.register(txHelper, blockDepth - 1);
}
}
else if (aggresiveFlush)
{
aggressiveFlushScopes.set(blockDepth - 1);
}
else
{
aggressiveFlushScopes.clear(blockDepth - 1);
}
scopeTransitions++;
if (fullTx)
{
// activate a pending TxWrapper for any database with a buffer that already is open
// avoid iterator usage
activeDatabases.forEach((db, c) -> maybeActivateTxWrapper(db));
}
}
/**
* Executes logic to explicitly end a transaction.
*
* @param inTx
* Flag indicating if we are in a transaction.
* @param fullTx
* Flag indicating if we are in a full transaction.
*/
public void endTx(boolean inTx, boolean fullTx)
{
scopeTransitions++;
if (!inTx)
{
aggressiveFlushScopes.clear(txHelper.getNestingLevel() - 1);
}
}
/**
* Executes logic required after a full transaction has ended.
*/
public void endTxPost()
{
// if any DBs were used in this tx block and remained connected, disconnect now
connMgr.transactionEnded();
}
/**
* Indicate whether the current block is "important" from the standpoint of transitioning into
* or out of its scope. A block is considered "important" in this sense if it is either
* completely outside of an application transaction, or if it is within a transaction AND has transaction
* block properties (i.e., is not a NO_TRANSACTION block).
* <p>
* This notion of importance is used to determine whether the various scoped data structures managed by
* this class need to add or delete a scope when transitioning into or out of the block, respectively,
* and whether we need to perform processing (e.g., copying/moving data between scopes) associated with
* such transitions.
*
* @return {@code true} if the block is considered "important" by the above criteria, else
* {@code false}.
*/
public boolean isImportantBlockTransition()
{
return txHelper.isGlobalBlock() ||
!txHelper.isTransaction() ||
txHelper.currentTransactionLevel() != TransactionManager.NO_TRANSACTION;
}
/**
* Releases the buffers into which a new record was loaded (or the old
* record was reloaded) into the current or deeper scopes, or only clears
* the list of buffers pending to be released.
*
* @param clearOnly
* If <code>true</code> then clear the list of buffers pending to be
* released, otherwise perform release of these buffers and clear
* the list after that.
*/
@Override
public void resetState(boolean clearOnly)
{
boolean hasLoaded = !emptyLoadedScopes.get(scopeDepth);
Map<RecordBuffer, Boolean> buffers = hasLoaded ? loadedBuffers.getDictionaryAtScope(0, false) : null;
if (buffers == null || buffers.isEmpty())
{
return;
}
if (!clearOnly)
{
// avoid iterators
buffers.forEach((buf, b) -> buf.release(false, false));
}
buffers.clear();
}
/**
* Provides a notification that a new scope is about to be entered.
*
* @param block
* The explicit block definition which required this notification.
*/
@Override
public void scopeStart(BlockDefinition block)
{
scopeDepth++;
// both scopes are empty
emptyLoadedScopes.set(scopeDepth, true);
emptyDirtyScopes.set(scopeDepth, true);
boolean inTx = txHelper.isTransaction();
boolean fullTx = inTx && txHelper.isFullTransaction();
int blockDepth = txHelper.getNestingLevel();
boolean aggresiveFlush = block.forFlushing;
beginTx(inTx, fullTx, aggresiveFlush, blockDepth);
}
/**
* Provides a notification that a scope is about to be exited.
*/
@Override
public void scopeFinished()
{
// here we will be out of sync with the TM on a transaction boundary, because our
// transaction depth is cleared with a transaction finalizable, which is called before this
// scope finished notification, so we rely on the TM.isTransaction() instead of our own
boolean inTx = txHelper.isTransaction();
boolean fullTx = txHelper.isFullTransaction();
endTx(inTx, fullTx);
if (!emptyDirtyScopes.get(scopeDepth))
{
Map<RecordBuffer, Boolean> dirtyBufs = dirtyBuffers.getDictionaryAtScope(0, false);
dirtyBuffers.deleteScope();
// ensure we are populated properly
try
{
scopeDepth--;
checkDirtyScopes();
}
finally
{
scopeDepth++;
}
if (dirtyBufs != null && !dirtyBufs.isEmpty() && dirtyBuffers.size() > 1)
{
// do not merge into the global block, as all changes should have been already committed or rolled
// back. this is required as temporary buffers can be changed without an active transaction present
// on the stack: in this case, they will be pushed to the previous scope until the global block is
// reached, when they will be discarded.
// non-capturing lambda
dirtyBufs.forEach((buffer, b) ->
{
// merge into previous scope, but only buffers with records which were changed, as the record
// associated (at initial dirty registration of the buffer) may have been flushed to the database,
// and the current record may not be 'dirty'
if (buffer.getCurrentRecord() != null)
{
Record record = buffer.getCurrentRecord();
if (buffer.isTouched() ||
record.checkState(DmoState.NEW) ||
record.checkState(DmoState.STALE) ||
record.checkState(DmoState.DELETING) ||
record.checkState(DmoState.DELETED) ||
record.checkState(DmoState.INVALID) ||
record.checkState(DmoState.CHANGED) ||
record.checkState(DmoState.TRACKED))
{
dirtyBuffers.addEntry(false, buffer, false);
}
}
});
}
}
// Copy the set of buffers contained in the current scope of the
// loadedBuffers (which is about to be removed) to the enclosing scope.
// Only buffers which are open at the enclosing or a higher scope
// should be copied.
if (loadedBuffers.size() > 1 && !emptyLoadedScopes.get(scopeDepth))
{
Map<RecordBuffer, Boolean> currentLoadedBuffers = loadedBuffers.getDictionaryAtScope(0, false);
if (!currentLoadedBuffers.isEmpty())
{
Collection<RecordBuffer> currentOpenBuffers = bm.getOpenBuffers();
Map<RecordBuffer, Boolean> loadedBufs = loadedBuffers.getDictionaryAtScope(1, true);
int loadedBufsize = loadedBuffers.size();
for (RecordBuffer buffer : currentLoadedBuffers.keySet())
{
if (buffer.getOpenScopeCount() > 1 || !currentOpenBuffers.contains(buffer))
{
// if this buffer is missing at the current scope of open
// buffers or it presents there but has open scope count > 1,
// that means that it presents at a higher level and it should
// be copied to the enclosing scope
loadedBufs.put(buffer, Boolean.FALSE);
buffer.addLoadedBuffersScope(loadedBufsize - 1);
}
}
}
loadedBuffers.deleteScope();
}
if (fullTx)
{
endTxPost();
}
// don't care about reseting the scope's bit
scopeDepth--;
}
/**
* Provides a notification that an external scope is about to be deleted.
*/
@Override
public void scopeDeleted()
{
// no-op
}
/**
* Get the {@link ScopeId} for the instance.
*
* @return The scope ID, or <code>null</code> for cases when the scope notification is not done via
* {@link BlockDefinition#scopeList}.
*/
@Override
public ScopeId getScopeId()
{
return ScopeId.TX_WRAPPER;
}
/**
* Check the {@link #emptyLoadedScopes} and ensure that the entire {@link #loadedBuffers} scope stack
* is populated.
*/
private void checkLoadedScopes()
{
int depth = scopeDepth;
while (depth >= 0 && emptyLoadedScopes.get(depth))
{
loadedBuffers.addScope(null);
emptyLoadedScopes.set(depth, false);
depth = depth - 1;
}
}
/**
* Check the {@link #emptyDirtyScopes} and ensure that the entire {@link #dirtyBuffers} scope stack
* is populated.
*/
private void checkDirtyScopes()
{
int depth = scopeDepth;
while (depth >= 0 && emptyDirtyScopes.get(depth))
{
dirtyBuffers.addScope(null);
emptyDirtyScopes.set(depth, false);
depth = depth - 1;
}
}
/**
* Invalidate all DMOs with {@link #ffCacheChanges}.
*/
private void invalidateFastFindCache()
{
if (ffCacheChanges.isEmpty())
{
return;
}
ffCacheChanges.forEach(FastFindCache::invalidate);
ffCacheChanges.clear();
}
}
/**
* A proxy for a {@link Commitable} that suppressed the commits and rollbacks. This should
* be used in aggressive flushing contexts that are not registered as transactions. This is usually
* the case with {@code forEach} constructs that should aggressively validate and flush records after
* each iteration, even if not inside a transaction.
*/
static class AggresiveFlushWrapper
implements Commitable
{
/** The proxied transaction wrapper which has its commit and rollback suppressed */
private final Commitable commitable;
/**
* Basic constructor.
*
* @param commitable
* The Commitable to be proxied
*/
public AggresiveFlushWrapper(Commitable commitable)
{
this.commitable = commitable;
}
/**
* {@inheritDoc}
*/
@Override
public void commit(boolean transaction)
{
// no-op
}
/**
* {@inheritDoc}
*/
@Override
public void rollback(boolean transaction)
{
// no-op
}
/**
* {@inheritDoc}
*/
@Override
public void validate(boolean transaction, boolean aggressiveFlush)
throws ErrorConditionException
{
commitable.validate(transaction, aggressiveFlush);
}
}
}