GlobalEventManager.java
/*
** Module : GlobalEventManager.java
** Abstract : Cross-context clearinghouse for DMO change events
**
** Copyright (c) 2004-2024, Golden Code Development Corporation.
**
** -#- -I- --Date-- --JPRM-- ----------------------------------Description-----------------------------------
** 001 ECF 20080813 @39540 Created initial version. Cross-context
** clearinghouse for DMO change events.
** 002 ECF 20080926 @39964 Fixed ConcurrentModificationException in
** cleanupContext().
** 003 ECF 20090619 @42964 Named PurgeWorker thread for easier debugging.
** 004 SVL 20090715 @43193 Added OOME protection into PurgeWorker.
** 005 IAS 20160331 Fixed updateEntityRefCount method.
** 006 ECF 20160815 Synchronized registration; replaced Apache commons logging with J2SE
** logging.
** 007 ECF 20200901 New ORM implementation.
** 008 TJD 20220504 Upgrade do Java 11 minor changes
** 009 SVL 20230113 Improved performance by replacing some "for-each" loops with indexed "for" loops.
** 010 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 011 GBB 20230825 SecurityManager session methods calls updated.
** 012 TJD 20240123 Java 17 compatibility updates
*/
/*
** 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.dirty;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.logging.*;
import com.goldencode.p2j.persist.*;
import com.goldencode.p2j.persist.Record;
import com.goldencode.p2j.persist.event.*;
import com.goldencode.p2j.security.*;
import com.goldencode.p2j.security.SecurityManager;
import com.goldencode.p2j.util.logging.*;
/**
* A clearinghouse for events which represent changes to records which affect
* any index of their backing tables. Events originate in the {@link
* DirtyShareManager} and are posted here for further handling. Objects which
* wish to process such events {@link #register(String) register} and {@link
* #deregister(long) deregister} via the <code>DirtyShareManager</code>.
* Events also are {@link #getEvents(long, long) collected} via the
* <code>DirtyShareManager</code>. Since all access is managed through that
* interface, this class is package private.
* <p>
* This class is described as a "clearinghouse" rather than a "dispatcher",
* because events are not dispatched via a "push" model. Rather, they are
* collected by an instance of this object (one per physical database), and
* await {@link #getEvents(long, long) requests} by interested parties (i.e.,
* registrants). This is a "pull" model for distributing these events.
* <p>
* A registrant registers interest in a particular type of DMO. Only events
* which affect DMO types for which interest has been registered are collected
* in this object's event queue. All other events are discarded upon receipt.
* The following methods are used by a <code>DirtyShareManager</code> to
* enqueue events:
* <ul>
* <li>{@link #enqueueInsert(String, Record)}
* <li>{@link #enqueueDelete(String, Long)}
* <li>{@link #enqueueUpdate(String, Record, String[])}
* </ul>
* <p>
* When a registrant collects new events, only those events which did not
* originate in that registrant's own context are provided. It is assumed
* that other mechanisms already exist to process "local" events.
* <p>
* The "pull" model of event distribution is used to prevent unnecessary event
* traffic, since in the common case, very few objects are interested in
* receiving such events. At the time of writing, this event distribution
* mechanism is intended primarily to support the need for certain
* <code>AdaptiveQuery</code> instances to invalidate their results in
* response to DMO changes in other sessions. By default, this is limited to
* instances of <code>AdaptiveFind</code>, which are more sensitive to such
* changes than common <code>AdaptiveQuery</code> instances.
* <p>
* Because a pull-based event model cannot be certain when all interested
* parties have processed each applicable event (or whether they ever will),
* some safeguards have been established to prevent both the premature release
* of needed events, and to prevent the unnecessary accumulation of old events
* which are no longer useful.
* <p>
* Each registrant is assigned a unique, sequential (ascending) registrant
* identifier upon registration. Registrants are tracked by these IDs. Each
* event is assigned a unique, sequential (ascending) event identifier, using
* the same sequence. Thus, IDs are unique across both all registrants and
* all events, and they are guaranteed to increase sequentially. When an
* event is enqueued, its ID is guaranteed to be higher than the registrant(s)
* interested in it.
* <p>
* When a registrant requests events, it provides the ID of the last event it
* retrieved. All new events (if any) collected after that ID are retrieved.
* When a registrant is no longer interested in events, it deregisters itself.
* Events older than the oldest registrant are purged at various points.
* <p>
* To prevent an idle registrant from pinning old events in the queue, an idle
* timeout check is applied at various points. If a registrant has been idle
* longer than the permitted timeout value, it is automatically deregistered.
* The next time that registrant requests new events (if ever), an exception
* is thrown, and it is up to the registrant to update its state accordingly.
* <p>
* In case a context is interrupted unexpectedly, all of its registrants are
* automatically deregistered. Even if this were not the case, the idle
* timeout would eventually clean up these resources.
* <p>
* TODO: enable directory-based configuration of idle timeout.
*/
final class GlobalEventManager
{
/** Logger */
private static final CentralLogger log = CentralLogger.get(GlobalEventManager.class.getName());
/** Default time, in milliseconds, for a registrant to be idle before being expired */
private static final long DEFAULT_IDLE_TIME = 600000;
/** Database for which this object manages global DMO change events */
private final Database database;
/** Map of interested party (registrant) IDs to registrants */
private final SortedMap<Long, Registrant> interested = new TreeMap<>();
/** Map of entity names to reference counts of interested parties */
private final ConcurrentMap<String, AtomicInteger> entities = new ConcurrentHashMap<>();
/** Object which tracks registrants by context for cleanup purposes */
private final ContextLocal<Context> context = new ContextLocal<Context>()
{
protected Context initialValue() { return (new Context()); }
protected void cleanup(Context ctx) { cleanupContext(ctx); }
};
/** Time, in milliseconds, for a registrant to be idle before being expired */
private final long idleTimeout = DEFAULT_IDLE_TIME;
/** Event queue */
private List<EventWrapper> eventQueue = null;
/** Variable from which unique registrant and event IDs are assigned */
private volatile long nextID = 0L;
/**
* Constructor
*
* @param database
* Database for which events are tracked by this object.
*/
GlobalEventManager(Database database)
{
this.database = database;
// Create event queue.
resetQueue();
// Launch purge worker.
new PurgeWorker();
}
/**
* Register interest in index changes for a particular DMO type.
*
* @param entity
* DMO entity name.
*
* @return Unique identifier for this registrant.
*
* @see #deregister(long)
*/
long register(String entity)
{
Long id;
synchronized (this)
{
id = Long.valueOf(++nextID);
Registrant reg = new Registrant(entity);
interested.put(id, reg);
updateEntityRefCount(entity, 1);
}
context.get().localIDs.add(id);
return id;
}
/**
* Deregister interest in index changes for a particular DMO type.
* <p>
* Invocation of this method triggers a purge sweep in a separate thread.
*
* @param id
* Unique identifier previously assigned to the registrant to be
* deregistered.
*
* @see #register(String)
*/
synchronized void deregister(long id)
{
deregister(id, true);
}
/**
* Retrieve any events which have been enqueued, which are newer than
* <code>lastID</code>, and which correspond to the DMO entity for which
* the specified registrant previously has registered interest.
* <p>
* As a side effect of this method, expired events may be detected, which
* are then purged from the event queue.
*
* @param registerID
* A unique identifier for an object which has registered interest
* in record-changing events.
* @param lastID
* A unique identifier representing the last event retrieved by
* the caller. If less than 0, it is assumed no events have yet
* been retrieved since the caller registered interest in these
* events.
*
* @return Array of applicable, new events, or <code>null</code> if no
* such events have been collected.
*
* @throws EventRegistrationException
* if <code>registerID</code> is invalid or has expired.
*/
synchronized GlobalChangeEvent[] getEvents(long registerID, long lastID)
throws EventRegistrationException
{
Registrant reg = interested.get(registerID);
if (reg == null)
{
throw new EventRegistrationException(registerID);
}
reg.setLastUse();
int size = eventQueue.size();
if (size == 0)
{
return null;
}
long oldestRegisterID = interested.firstKey();
List<GlobalChangeEvent> events = null;
if (lastID < 0)
{
lastID = registerID;
}
ListIterator<EventWrapper> iter = eventQueue.listIterator(size);
while (iter.hasPrevious())
{
EventWrapper ew = iter.previous();
GlobalChangeEvent event = ew.getEvent();
long eventID = event.getEventID();
if (eventID <= lastID)
{
if (eventID < oldestRegisterID)
{
// Event has expired; purge it.
iter.remove();
}
continue;
}
if (ew.isOfInterest(reg))
{
if (events == null)
{
events = new ArrayList<>();
}
events.add(event);
}
}
if (events == null)
{
return null;
}
return events.toArray(new GlobalChangeEvent[events.size()]);
}
/**
* Enqueue a DMO update event, if any registrant is interested in it.
*
* @param entity
* DMO entity name; identifies DMO type affected by update.
* @param dmo
* The DMO instance in its <i>post-update</i> state.
* @param properties
* Names of all properties which have been updated.
*/
synchronized void enqueueUpdate(String entity, Record dmo, String[] properties)
{
if (isEntityReferenced(entity))
{
GlobalChangeEvent event = new GlobalChangeEvent(++nextID, entity, dmo, properties);
enqueueEvent(event);
}
}
/**
* Enqueue a DMO insert event, if any registrant is interested in it.
*
* @param entity
* DMO entity name; identifies DMO type affected by insert.
* @param dmo
* The new DMO instance.
*/
synchronized void enqueueInsert(String entity, Record dmo)
{
if (isEntityReferenced(entity))
{
GlobalChangeEvent event = new GlobalChangeEvent(++nextID, entity, dmo);
enqueueEvent(event);
}
}
/**
* Enqueue a DMO delete event, if any registrant is interested in it.
*
* @param entity
* DMO entity name; identifies DMO type affected by update.
* @param primaryKey
* Primary key ID of the deleted DMO instance.
*/
synchronized void enqueueDelete(String entity, Long primaryKey)
{
if (isEntityReferenced(entity))
{
GlobalChangeEvent event = new GlobalChangeEvent(++nextID, entity, primaryKey);
enqueueEvent(event);
}
}
/**
* Enqueue a DMO update, insert, or delete event.
* <p>
* Invocation of this method triggers a purge sweep in a separate thread.
*
* @param event
* Event object which encapsulates information about the change.
*/
private void enqueueEvent(GlobalChangeEvent event)
{
EventWrapper ew = new EventWrapper(event);
eventQueue.add(ew);
// Trigger purge sweep.
notify();
}
/**
* Deregister interest in index changes for a particular DMO type.
* <p>
* Invocation of this method may trigger a purge sweep in a separate thread.
*
* @param id
* Unique identifier previously assigned to the registrant to be deregistered.
* @param purge
* <code>true</code> to trigger a purge sweep in a separate thread
* after the deregistration takes place; else <code>false</code>.
*/
private void deregister(long id, boolean purge)
{
Registrant reg = interested.remove(id);
if (reg != null)
{
updateEntityRefCount(reg.getEntity(), -1);
context.get().localIDs.remove(id);
}
if (purge)
{
// Trigger purge sweep.
notify();
}
}
/**
* Modify a reference count for the given DMO entity. When a reference
* count for any entity reaches 0, this indicates that no registrants
* currently are interested in events about this DMO type.
*
* @param entity
* DMO entity name.
* @param delta
* Amount to adjust the reference count. Should be 1 to increment
* or -1 to decrement.
*/
private void updateEntityRefCount(String entity, int delta)
{
AtomicInteger count = entities.get(entity);
if (count == null)
{
count = new AtomicInteger(0);
AtomicInteger prevCount = entities.putIfAbsent(entity, count);
if (prevCount != null)
{
count = prevCount;
}
}
int nv = count.addAndGet(delta);
if (nv == 0)
{
entities.remove(entity, count);
}
}
/**
* Report whether any registrant is interested in events about the given
* DMO type.
*
* @param entity
* DMO entity name.
*
* @return <code>true</code> if any registrant is interested in the given
* DMO type, else <code>false</code>.
*/
private boolean isEntityReferenced(String entity)
{
if (interested.isEmpty())
{
return false;
}
AtomicInteger count = entities.get(entity);
return count != null && count.get() > 0;
}
/**
* Reset the event queue by instantiating a new, empty queue and discarding
* the old one, if any.
*/
private void resetQueue()
{
eventQueue = new LinkedList<>();
}
/**
* Invoked when a context is ending, this method deregisters any
* registrants associated with the local context. If the context is ending
* normally, there should be nothing for this method to do. However, if
* the context is ending abruptly, some registrants may need to be cleaned
* up.
* <p>
* Invocation of this method triggers a purge sweep in a separate thread if
* any "orphaned" registrants were detected.
*
* @param ctx
* Context requiring cleanup at end of life.
*/
private void cleanupContext(Context ctx)
{
Set<Long> localIDs = ctx.localIDs;
if (localIDs.isEmpty())
{
return;
}
synchronized (this)
{
Iterator<Long> iter = localIDs.iterator();
while (iter.hasNext())
{
// Update data structures directly, rather than calling
// deregister(long, boolean), because that method will try to
// update localIDs, causing ConcurrentModificationException.
Long id = iter.next();
Registrant reg = interested.remove(id);
if (reg != null)
{
updateEntityRefCount(reg.getEntity(), -1);
}
iter.remove();
}
// Trigger purge sweep.
notify();
}
}
/**
* Object which tracks registrants associated with the current session,
* primarily for end-of-context cleanup purposes.
*/
private static class Context
{
/** Set of IDs registered to current context */
private final Set<Long> localIDs = new HashSet<>(4);
}
/**
* A party which is interested in tracking index change for a particular
* DMO type. Also tracks the timestamp of the last event request for this
* registrant.
*/
private static class Registrant
{
/** DMO entity name */
private final String entity;
/** Millisecond timestamp of last event collection request */
private long lastUse;
/**
* Constructor.
*
* @param entity
* DMO entity name.
*/
Registrant(String entity)
{
this.entity = entity;
setLastUse();
}
/**
* Get name of DMO entity being tracked by this registrant.
*
* @return DMO entity name.
*/
private String getEntity()
{
return entity;
}
/**
* Get timestamp of the last event request made for this registrant.
*
* @return Millisecond timestamp value.
*/
private long getLastUse()
{
return lastUse;
}
/**
* Set the timestamp of the last event request made for this registrant
* to the current system time (since the epoch) in milliseconds.
*/
private void setLastUse()
{
lastUse = System.currentTimeMillis();
}
}
/**
* Object which wraps a change event to remember the unique identifier of
* the session which originated the event. This allows us to filter events
* by originating session. It is assumed registrants are not interested in
* receiving events from this clearinghouse, which they themselves generated.
*/
private static class EventWrapper
{
/** Security manager which provides session IDs */
private static final SecurityManager security = SecurityManager.getInstance();
/** Unique ID of current context */
private final Integer sessionID;
/** Event representing a DMO change (insert, update, delete) */
private final GlobalChangeEvent event;
/**
* Constructor. Records current session ID at the time of construction.
*
* @param event
* Event representing a DMO change (insert, update, delete).
*/
private EventWrapper(GlobalChangeEvent event)
{
this.sessionID = security.sessionSm.getSessionId();
this.event = event;
}
/**
* Get the event wrapped by this object.
*
* @return Event representing a DMO change (insert, update, delete).
*/
private GlobalChangeEvent getEvent()
{
return event;
}
/**
* Determine whether the event wrapped by this object is of interest to
* the current context. It is considered of interest if it originated
* in a different context, and is associated with the given DMO entity.
*
* @param registration
* Party interested in DMO changes.
*
* @return <code>true</code> if this event should be delivered to the
* current context, else <code>false</code>.
*/
private boolean isOfInterest(Registrant registration)
{
String entity = registration.getEntity();
if (!entity.equals(event.getEntity()))
{
return false;
}
Integer currentSessionID = security.sessionSm.getSessionId();
return (!sessionID.equals(currentSessionID));
}
}
/**
* A worker which purges idle registrants from the registrant map and
* expired events from the event queue in a dedicated, daemon thread. The
* worker runs in an infinite loop, waiting on each pass for a notification
* from the event manager to do its work.
*/
private class PurgeWorker
extends Thread
{
/**
* Constructor which launches this worker's dedicated thread.
*/
private PurgeWorker()
{
super("GlobalEventManager Queue Purge Thread");
setDaemon(true);
start();
}
/**
* Run in an infinite loop, waiting for purge notifications from the
* event manager at the top of each pass. When a notification is
* received, first purge idle registrants, then purge expired events.
*/
public void run()
{
while (true)
{
try
{
GlobalEventManager gem = GlobalEventManager.this;
synchronized (gem)
{
gem.wait();
purgeRegistrants();
purgeEvents();
}
}
catch (InterruptedException exc)
{
if (log.isLoggable(Level.WARNING))
{
log.log(Level.WARNING,
"Persistence event clearing house purge thread for " +
database +
" interrupted; resuming work.",
exc);
}
}
catch (OutOfMemoryError oome)
{
try
{
log.log(Level.SEVERE, "OOME in purge worker", oome);
}
catch (OutOfMemoryError e)
{
// Ignore.
}
}
}
}
/**
* Walk from oldest to most recent registrant, deregistering those that
* have been idle more than the required timeout period.
*/
private void purgeRegistrants()
{
ArrayList<Long> dead = null;
long threshold = System.currentTimeMillis() - idleTimeout;
Iterator<SortedMap.Entry<Long, Registrant>> iter = interested.entrySet().iterator();
while (iter.hasNext())
{
SortedMap.Entry<Long, Registrant> entry = iter.next();
Registrant reg = entry.getValue();
if (reg.getLastUse() < threshold)
{
if (dead == null)
{
dead = new ArrayList<>();
}
dead.add(entry.getKey());
}
else
{
break;
}
}
// We deregister in a separate pass, because calling deregister() in
// the loop above would raise ConcurrentModificationException.
if (dead != null)
{
for (int i = 0; i < dead.size(); i++)
{
Long id = dead.get(i);
deregister(id, false);
}
}
}
/**
* Purge expired events from the event queue, or reset the queue entirely if there are no
* registrants at the time this method is invoked.
*/
private void purgeEvents()
{
int size = eventQueue.size();
if (size == 0)
{
return;
}
if (interested.isEmpty())
{
resetQueue();
return;
}
long oldestRegisterID = interested.firstKey();
ListIterator<EventWrapper> iter = eventQueue.listIterator(size);
while (iter.hasPrevious())
{
EventWrapper ew = iter.previous();
GlobalChangeEvent event = ew.getEvent();
if (event.getEventID() > oldestRegisterID)
{
break;
}
iter.remove();
}
}
}
}