Turn on suggestions
Auto-suggest helps you quickly narrow down your search results by suggesting possible matches as you type.
Showing results for
- VMware Technology Network
- :
- Cloud & SDDC
- :
- vCenter
- :
- vCenter Hyperic Discussions
- :
- hq r5775 - in trunk/src/org/hyperic/hq: escalatio...
VMware Cloud Community
Options
- Subscribe to RSS Feed
- Mark Topic as New
- Mark Topic as Read
- Float this Topic for Current User
- Bookmark
- Subscribe
- Mute
- Printer Friendly Page
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
hq r5775 - in trunk/src/org/hyperic/hq:
escalation/server/session events/ext
2 Replies
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
Hi Tom,
Two comments.
In EscalationManagerEJBImpl.startEscalation(), I see that you left
creator.createEscalatable() after checking for escalation state.
Which occurred to me is the right behavior after our conversation.
If an escalation is in progress, we don't even create new alerts for
the same alert definition, and the original actions definitely should
not execute.
Would it be safer to put the EscalationRuntime.getInstance
().releaseMutex() finally block with the same try as the the
EscalationRuntime.getInstance().acquireMutex()? A developer can come
along and insert some code that throws an exception between the
acquire and the try block, and you never release the mutex.
Charles
On Aug 31, 2007, at 3:19 PM, tkeeney@hyperic.com wrote:
> Author: tkeeney
> Date: 2007-08-31 15:19:56 -0700 (Fri, 31 Aug 2007)
> New Revision: 5775
> URL: http://svn.hyperic.org/?view=rev&root=Hyperic+HQ&revision=5775
>
> Modified:
> trunk/src/org/hyperic/hq/escalation/server/session/
> EscalationManagerEJBImpl.java
> trunk/src/org/hyperic/hq/escalation/server/session/
> EscalationRuntime.java
> trunk/src/org/hyperic/hq/events/ext/AbstractTrigger.java
> Log:
> [HHQ-915] When attempting to start an escalation, need to account
> for other threads having already started the escalation yet haven't
> committed their txn yet. Keeping a cache of these uncommitted txns
> so we know not to start the same escalation that is now in progress.
>
> Modified: trunk/src/org/hyperic/hq/escalation/server/session/
> EscalationManagerEJBImpl.java
> ===================================================================
> --- trunk/src/org/hyperic/hq/escalation/server/session/
> EscalationManagerEJBImpl.java 2007-08-31 21:58:26 UTC (rev 5774)
> +++ trunk/src/org/hyperic/hq/escalation/server/session/
> EscalationManagerEJBImpl.java 2007-08-31 22:19:56 UTC (rev 5775)
> @@ -245,35 +245,116 @@
> }
>
> /**
> - * Start an escalation. If the escalation has already been
> started, then
> - * this method call will be a no-op.
> + * Start an escalation. If the entity performing escalations
> does not have
> + * an assigned escalation or if the escalation has already
> been started,
> + * then this method call will be a no-op.
> *
> - * @param def Definition to start the escalation for
> + * @param def The entity performing escalations.
> * @param creator Object which will create an {@link
> Escalatable} object
> * if invoking this method actually starts an
> escalation.
> + * @return
> + *
> + * no escalation assigned to the entity or the
> escalation
> + * is already in progress.
> *
> * @ejb:interface-method
> */
> - public void startEscalation(PerformsEscalations def,
> - EscalatableCreator creator)
> - {
> - EscalationState curState = _stateDAO.find(def);
> - Escalatable alert;
> -
> + public boolean startEscalation(PerformsEscalations def,
> + EscalatableCreator creator) {
> if (def.getEscalation() == null)
> - return;
> + return false;
>
> - if (curState != null) {
> + boolean started = false;
> +
> + try {
> + // Get the lock right now!!! We have to be sure the
> escalation
> + // was scheduled for post commit before letting anyone
> else through.
> + // Assume we may throw an unchecked exception prior to
> scheduling.
> + // This is possible, especially when creating the
> escalatable.
> + EscalationRuntime.getInstance().acquireMutex();
> +
> + try {
> + if (escalationStateExists(def)) {
> + return started = false;
> + }
> +
> + try {
> + Escalatable alert = creator.createEscalatable();
> + EscalationState curState = new EscalationState
> (alert);
> + _stateDAO.save(curState);
> + _log.debug("Escalation started: state=" +
> curState.getId());
> + EscalationRuntime.getInstance
> ().scheduleEscalation(curState);
> + started = true;
> + } finally {
> + if (!started) {
> + EscalationRuntime.getInstance()
> + .removeFromUncommittedEscalationStateCach
> e(def, false);
> + }
> + }
> +
> + } finally {
> + EscalationRuntime.getInstance().releaseMutex();
> + }
> + } catch (InterruptedException e) {
> + _log.error("Failed to start escalation for " +
> + "alert def id="+def.getId()+
> + "; type="+def.getAlertType().getCode(), e);
> + }
> +
> + return started;
> + }
> +
> + private boolean escalationStateExists(PerformsEscalations def) {
> + // Checks if there is an uncommitted escalation state for
> this def.
> + boolean existsInCache = EscalationRuntime.getInstance()
> + .addToUncommittedEscalationStateCache
> (def);
> +
> + boolean existsInDb = false;
> +
> + try {
> + // Checks if there is a committed escalation state for
> this def.
> + existsInDb = _stateDAO.find(def) != null;
> + } catch (Exception e) {
> + _log.warn("There is more than one escalation in
> progress for " +
> + "alert def id="+def.getId()+
> + "; type="+def.getAlertType().getCode(), e);
> + // HHQ-915: A hibernate exception will occur when
> looking up the
> + // escalation state if more than one exists. This
> shouldn't happen,
> + // but if it does, don't create another escalation.
> + existsInDb = true;
> + }
> +
> + // Possible scenarios when storing an escalation state ->
> + // how to remove the def from the uncommitted cache:
> + // in_cache=false, in_db=false -> schedule to remove on
> commit
> + // in_cache=true, in_db=false -> do nothing,
> + // - will be removed from
> cache post-commit
> + // in_cache=false, in_db=true -> remove immediately
> + // in_cache=true, in_db=true -> (a timing issue),
> + // - will be removed from
> cache post-commit,
> + // but to be safe, remove
> immediately
> + EscalationRuntime runtime = EscalationRuntime.getInstance();
> +
> + if (existsInCache) {
> + if (existsInDb) {
> + runtime.removeFromUncommittedEscalationStateCache
> (def, false);
> + } else {
> + // do nothing
> + }
> + } else {
> + if (existsInDb) {
> + runtime.removeFromUncommittedEscalationStateCache
> (def, false);
> + } else {
> + runtime.removeFromUncommittedEscalationStateCache
> (def, true);
> + }
> + }
> +
> + if (existsInCache || existsInDb) {
> _log.debug("startEscalation called on [" + def + "]
> but it was " +
> - "already running");
> - return;
> + "already running");
> }
>
> - alert = creator.createEscalatable();
> - curState = new EscalationState(alert);
> - _stateDAO.save(curState);
> - _log.debug("Escalation started: state=" + curState.getId());
> - EscalationRuntime.getInstance().scheduleEscalation(curState);
> + return existsInCache || existsInDb;
> }
>
> /**
>
> Modified: trunk/src/org/hyperic/hq/escalation/server/session/
> EscalationRuntime.java
> ===================================================================
> --- trunk/src/org/hyperic/hq/escalation/server/session/
> EscalationRuntime.java 2007-08-31 21:58:26 UTC (rev 5774)
> +++ trunk/src/org/hyperic/hq/escalation/server/session/
> EscalationRuntime.java 2007-08-31 22:19:56 UTC (rev 5775)
> @@ -25,6 +25,7 @@
> package org.hyperic.hq.escalation.server.session;
>
> import java.util.ArrayList;
> +import java.util.Collections;
> import java.util.Date;
> import java.util.HashMap;
> import java.util.HashSet;
> @@ -42,6 +43,7 @@
> import EDU.oswego.cs.dl.util.concurrent.Executor;
> import EDU.oswego.cs.dl.util.concurrent.LinkedQueue;
> import EDU.oswego.cs.dl.util.concurrent.PooledExecutor;
> +import EDU.oswego.cs.dl.util.concurrent.Semaphore;
>
> /**
> * This class manages the runtime execution of escalation chains.
> The
> @@ -71,9 +73,14 @@
>
> private final ThreadLocal _batchUnscheduleTxnListeners = new
> ThreadLocal();
> private final Log _log = LogFactory.getLog
> (EscalationRuntime.class);
> - private final ClockDaemon _schedule = new
> ClockDaemon();
> - private final Map _stateIdsToTasks = new
> HashMap();
> - private final Map _esclEntityIdsToStateIds
> = new HashMap();
> + private final ClockDaemon _schedule = new ClockDaemon();
> + private final Map _stateIdsToTasks = new HashMap();
> + private final Map _esclEntityIdsToStateIds = new HashMap();
> +
> + private final Semaphore _mutex = new Semaphore(1);
> +
> + private final Set _uncomittedEscalatingEntities =
> + Collections.synchronizedSet(new HashSet
> ());
> private final PooledExecutor _executor;
> private final EscalationManagerLocal _esclMan;
>
> @@ -128,7 +135,7 @@
> * Unschedule the execution of an escalation state. The
> unschedule will
> * only occur if the transaction successfully commits.
> */
> - void unscheduleEscalation(EscalationState state) {
> + public void unscheduleEscalation(EscalationState state) {
> final Integer stateId = state.getId();
>
> HQApp.getInstance().addTransactionListener(new
> TransactionListener() {
> @@ -148,7 +155,7 @@
> * entity that performs escalations. The unschedule will only
> occur if the
> * transaction successfully commits.
> */
> - void unscheduleAllEscalationsFor(PerformsEscalations def) {
> + public void unscheduleAllEscalationsFor(PerformsEscalations
> def) {
> BatchUnscheduleEscalationsTransactionListener
> batchTxnListener =
> (BatchUnscheduleEscalationsTransactionListener)
> _batchUnscheduleTxnListeners.get();
> @@ -260,6 +267,62 @@
> }
>
> /**
> + * Acquire the mutex.
> + *
> + * @throws InterruptedException
> + */
> + public void acquireMutex() throws InterruptedException {
> + _mutex.acquire();
> + }
> +
> + /**
> + * Release the mutex.
> + */
> + public void releaseMutex() {
> + _mutex.release();
> + }
> +
> + /**
> + * Add the uncommitted escalation state for this entity
> performing escalations
> + * to the uncommitted escalation state cache. This cache is
> used to track
> + * escalation states that have been scheduled but are not
> visible to other
> + * threads prior to the transaction commit.
> + *
> + * @param def The entity that performs escalations.
> + * @return
> uncommitted escalation state.
> + */
> + public boolean addToUncommittedEscalationStateCache
> (PerformsEscalations def) {
> + return !_uncomittedEscalatingEntities.add(new
> EscalatingEntityIdentifier(def));
> + }
> +
> + /**
> + * Remove the uncommitted escalation state for this entity
> performing
> + * escalations from the uncommitted escalation state cache.
> + *
> + * @param def The entity that performs escalations.
> + * @param postTxnCommit
> commit;
> + *
> + */
> + public void removeFromUncommittedEscalationStateCache(final
> PerformsEscalations def,
> + boolean
> postTxnCommit) {
> + if (postTxnCommit) {
> + HQApp.getInstance().addTransactionListener(new
> TransactionListener() {
> +
> + public void afterCommit(boolean success) {
> + removeFromUncommittedEscalationStateCache(def,
> false);
> + }
> +
> + public void beforeCommit() {
> + }
> +
> + });
> + } else {
> + _uncomittedEscalatingEntities.remove(new
> EscalatingEntityIdentifier(def));
> + }
> +
> + }
> +
> + /**
> * This method introduces an escalation state to the runtime.
> The
> * escalation will be invoked according to the next action
> time of the
> * state.
> @@ -267,12 +330,13 @@
> * If the state had been previously scheduled, it will be
> rescheduled with
> * the new time.
> */
> - void scheduleEscalation(final EscalationState state) {
> + public void scheduleEscalation(final EscalationState state) {
> final long schedTime = state.getNextActionTime();
>
> HQApp.getInstance().addTransactionListener(new
> TransactionListener() {
> public void afterCommit(boolean success) {
> _log.debug("Transaction committed: success=" +
> success);
> +
> if (success) {
> scheduleEscalation_(state, schedTime);
> }
> @@ -316,7 +380,8 @@
> _esclMan.executeState(stateId);
> }
>
> - static EscalationRuntime getInstance() {
> + public static EscalationRuntime getInstance() {
> return INSTANCE;
> }
> +
> }
>
> Modified: trunk/src/org/hyperic/hq/events/ext/AbstractTrigger.java
> ===================================================================
> --- trunk/src/org/hyperic/hq/events/ext/AbstractTrigger.java
> 2007-08-31 21:58:26 UTC (rev 5774)
> +++ trunk/src/org/hyperic/hq/events/ext/AbstractTrigger.java
> 2007-08-31 22:19:56 UTC (rev 5775)
> @@ -185,13 +185,15 @@
> new ClassicEscalatableCreator(alertDef, event);
>
> // Now start escalation
> - if (alertDef.getEscalation() != null) {
> - EscalationManagerEJBImpl.getOne().startEscalation
> (alertDef,
> -
> creator);
> - }
> - else {
> + boolean started = EscalationManagerEJBImpl.getOne()
> + .startEscalation(alertDef,
> creator);
> +
> + // If there is no escalation or it wasn't started,
> then execute
> + // the classic escalations.
> + if (!started) {
> creator.createEscalatable();
> }
> +
> } catch (FinderException e) {
> throw new ActionExecuteException(
> "Alert Definition not found for trigger: " + getId
> ());
>
Two comments.
In EscalationManagerEJBImpl.startEscalation(), I see that you left
creator.createEscalatable() after checking for escalation state.
Which occurred to me is the right behavior after our conversation.
If an escalation is in progress, we don't even create new alerts for
the same alert definition, and the original actions definitely should
not execute.
Would it be safer to put the EscalationRuntime.getInstance
().releaseMutex() finally block with the same try as the the
EscalationRuntime.getInstance().acquireMutex()? A developer can come
along and insert some code that throws an exception between the
acquire and the try block, and you never release the mutex.
Charles
On Aug 31, 2007, at 3:19 PM, tkeeney@hyperic.com wrote:
> Author: tkeeney
> Date: 2007-08-31 15:19:56 -0700 (Fri, 31 Aug 2007)
> New Revision: 5775
> URL: http://svn.hyperic.org/?view=rev&root=Hyperic+HQ&revision=5775
>
> Modified:
> trunk/src/org/hyperic/hq/escalation/server/session/
> EscalationManagerEJBImpl.java
> trunk/src/org/hyperic/hq/escalation/server/session/
> EscalationRuntime.java
> trunk/src/org/hyperic/hq/events/ext/AbstractTrigger.java
> Log:
> [HHQ-915] When attempting to start an escalation, need to account
> for other threads having already started the escalation yet haven't
> committed their txn yet. Keeping a cache of these uncommitted txns
> so we know not to start the same escalation that is now in progress.
>
> Modified: trunk/src/org/hyperic/hq/escalation/server/session/
> EscalationManagerEJBImpl.java
> ===================================================================
> --- trunk/src/org/hyperic/hq/escalation/server/session/
> EscalationManagerEJBImpl.java 2007-08-31 21:58:26 UTC (rev 5774)
> +++ trunk/src/org/hyperic/hq/escalation/server/session/
> EscalationManagerEJBImpl.java 2007-08-31 22:19:56 UTC (rev 5775)
> @@ -245,35 +245,116 @@
> }
>
> /**
> - * Start an escalation. If the escalation has already been
> started, then
> - * this method call will be a no-op.
> + * Start an escalation. If the entity performing escalations
> does not have
> + * an assigned escalation or if the escalation has already
> been started,
> + * then this method call will be a no-op.
> *
> - * @param def Definition to start the escalation for
> + * @param def The entity performing escalations.
> * @param creator Object which will create an {@link
> Escalatable} object
> * if invoking this method actually starts an
> escalation.
> + * @return
true if the escalation is started;> + *
false if not because either there is> + * no escalation assigned to the entity or the
> escalation
> + * is already in progress.
> *
> * @ejb:interface-method
> */
> - public void startEscalation(PerformsEscalations def,
> - EscalatableCreator creator)
> - {
> - EscalationState curState = _stateDAO.find(def);
> - Escalatable alert;
> -
> + public boolean startEscalation(PerformsEscalations def,
> + EscalatableCreator creator) {
> if (def.getEscalation() == null)
> - return;
> + return false;
>
> - if (curState != null) {
> + boolean started = false;
> +
> + try {
> + // Get the lock right now!!! We have to be sure the
> escalation
> + // was scheduled for post commit before letting anyone
> else through.
> + // Assume we may throw an unchecked exception prior to
> scheduling.
> + // This is possible, especially when creating the
> escalatable.
> + EscalationRuntime.getInstance().acquireMutex();
> +
> + try {
> + if (escalationStateExists(def)) {
> + return started = false;
> + }
> +
> + try {
> + Escalatable alert = creator.createEscalatable();
> + EscalationState curState = new EscalationState
> (alert);
> + _stateDAO.save(curState);
> + _log.debug("Escalation started: state=" +
> curState.getId());
> + EscalationRuntime.getInstance
> ().scheduleEscalation(curState);
> + started = true;
> + } finally {
> + if (!started) {
> + EscalationRuntime.getInstance()
> + .removeFromUncommittedEscalationStateCach
> e(def, false);
> + }
> + }
> +
> + } finally {
> + EscalationRuntime.getInstance().releaseMutex();
> + }
> + } catch (InterruptedException e) {
> + _log.error("Failed to start escalation for " +
> + "alert def id="+def.getId()+
> + "; type="+def.getAlertType().getCode(), e);
> + }
> +
> + return started;
> + }
> +
> + private boolean escalationStateExists(PerformsEscalations def) {
> + // Checks if there is an uncommitted escalation state for
> this def.
> + boolean existsInCache = EscalationRuntime.getInstance()
> + .addToUncommittedEscalationStateCache
> (def);
> +
> + boolean existsInDb = false;
> +
> + try {
> + // Checks if there is a committed escalation state for
> this def.
> + existsInDb = _stateDAO.find(def) != null;
> + } catch (Exception e) {
> + _log.warn("There is more than one escalation in
> progress for " +
> + "alert def id="+def.getId()+
> + "; type="+def.getAlertType().getCode(), e);
> + // HHQ-915: A hibernate exception will occur when
> looking up the
> + // escalation state if more than one exists. This
> shouldn't happen,
> + // but if it does, don't create another escalation.
> + existsInDb = true;
> + }
> +
> + // Possible scenarios when storing an escalation state ->
> + // how to remove the def from the uncommitted cache:
> + // in_cache=false, in_db=false -> schedule to remove on
> commit
> + // in_cache=true, in_db=false -> do nothing,
> + // - will be removed from
> cache post-commit
> + // in_cache=false, in_db=true -> remove immediately
> + // in_cache=true, in_db=true -> (a timing issue),
> + // - will be removed from
> cache post-commit,
> + // but to be safe, remove
> immediately
> + EscalationRuntime runtime = EscalationRuntime.getInstance();
> +
> + if (existsInCache) {
> + if (existsInDb) {
> + runtime.removeFromUncommittedEscalationStateCache
> (def, false);
> + } else {
> + // do nothing
> + }
> + } else {
> + if (existsInDb) {
> + runtime.removeFromUncommittedEscalationStateCache
> (def, false);
> + } else {
> + runtime.removeFromUncommittedEscalationStateCache
> (def, true);
> + }
> + }
> +
> + if (existsInCache || existsInDb) {
> _log.debug("startEscalation called on [" + def + "]
> but it was " +
> - "already running");
> - return;
> + "already running");
> }
>
> - alert = creator.createEscalatable();
> - curState = new EscalationState(alert);
> - _stateDAO.save(curState);
> - _log.debug("Escalation started: state=" + curState.getId());
> - EscalationRuntime.getInstance().scheduleEscalation(curState);
> + return existsInCache || existsInDb;
> }
>
> /**
>
> Modified: trunk/src/org/hyperic/hq/escalation/server/session/
> EscalationRuntime.java
> ===================================================================
> --- trunk/src/org/hyperic/hq/escalation/server/session/
> EscalationRuntime.java 2007-08-31 21:58:26 UTC (rev 5774)
> +++ trunk/src/org/hyperic/hq/escalation/server/session/
> EscalationRuntime.java 2007-08-31 22:19:56 UTC (rev 5775)
> @@ -25,6 +25,7 @@
> package org.hyperic.hq.escalation.server.session;
>
> import java.util.ArrayList;
> +import java.util.Collections;
> import java.util.Date;
> import java.util.HashMap;
> import java.util.HashSet;
> @@ -42,6 +43,7 @@
> import EDU.oswego.cs.dl.util.concurrent.Executor;
> import EDU.oswego.cs.dl.util.concurrent.LinkedQueue;
> import EDU.oswego.cs.dl.util.concurrent.PooledExecutor;
> +import EDU.oswego.cs.dl.util.concurrent.Semaphore;
>
> /**
> * This class manages the runtime execution of escalation chains.
> The
> @@ -71,9 +73,14 @@
>
> private final ThreadLocal _batchUnscheduleTxnListeners = new
> ThreadLocal();
> private final Log _log = LogFactory.getLog
> (EscalationRuntime.class);
> - private final ClockDaemon _schedule = new
> ClockDaemon();
> - private final Map _stateIdsToTasks = new
> HashMap();
> - private final Map _esclEntityIdsToStateIds
> = new HashMap();
> + private final ClockDaemon _schedule = new ClockDaemon();
> + private final Map _stateIdsToTasks = new HashMap();
> + private final Map _esclEntityIdsToStateIds = new HashMap();
> +
> + private final Semaphore _mutex = new Semaphore(1);
> +
> + private final Set _uncomittedEscalatingEntities =
> + Collections.synchronizedSet(new HashSet
> ());
> private final PooledExecutor _executor;
> private final EscalationManagerLocal _esclMan;
>
> @@ -128,7 +135,7 @@
> * Unschedule the execution of an escalation state. The
> unschedule will
> * only occur if the transaction successfully commits.
> */
> - void unscheduleEscalation(EscalationState state) {
> + public void unscheduleEscalation(EscalationState state) {
> final Integer stateId = state.getId();
>
> HQApp.getInstance().addTransactionListener(new
> TransactionListener() {
> @@ -148,7 +155,7 @@
> * entity that performs escalations. The unschedule will only
> occur if the
> * transaction successfully commits.
> */
> - void unscheduleAllEscalationsFor(PerformsEscalations def) {
> + public void unscheduleAllEscalationsFor(PerformsEscalations
> def) {
> BatchUnscheduleEscalationsTransactionListener
> batchTxnListener =
> (BatchUnscheduleEscalationsTransactionListener)
> _batchUnscheduleTxnListeners.get();
> @@ -260,6 +267,62 @@
> }
>
> /**
> + * Acquire the mutex.
> + *
> + * @throws InterruptedException
> + */
> + public void acquireMutex() throws InterruptedException {
> + _mutex.acquire();
> + }
> +
> + /**
> + * Release the mutex.
> + */
> + public void releaseMutex() {
> + _mutex.release();
> + }
> +
> + /**
> + * Add the uncommitted escalation state for this entity
> performing escalations
> + * to the uncommitted escalation state cache. This cache is
> used to track
> + * escalation states that have been scheduled but are not
> visible to other
> + * threads prior to the transaction commit.
> + *
> + * @param def The entity that performs escalations.
> + * @return
true if there is already an > uncommitted escalation state.
> + */
> + public boolean addToUncommittedEscalationStateCache
> (PerformsEscalations def) {
> + return !_uncomittedEscalatingEntities.add(new
> EscalatingEntityIdentifier(def));
> + }
> +
> + /**
> + * Remove the uncommitted escalation state for this entity
> performing
> + * escalations from the uncommitted escalation state cache.
> + *
> + * @param def The entity that performs escalations.
> + * @param postTxnCommit
true to remove post txn > commit;
> + *
false to remove immediately.> + */
> + public void removeFromUncommittedEscalationStateCache(final
> PerformsEscalations def,
> + boolean
> postTxnCommit) {
> + if (postTxnCommit) {
> + HQApp.getInstance().addTransactionListener(new
> TransactionListener() {
> +
> + public void afterCommit(boolean success) {
> + removeFromUncommittedEscalationStateCache(def,
> false);
> + }
> +
> + public void beforeCommit() {
> + }
> +
> + });
> + } else {
> + _uncomittedEscalatingEntities.remove(new
> EscalatingEntityIdentifier(def));
> + }
> +
> + }
> +
> + /**
> * This method introduces an escalation state to the runtime.
> The
> * escalation will be invoked according to the next action
> time of the
> * state.
> @@ -267,12 +330,13 @@
> * If the state had been previously scheduled, it will be
> rescheduled with
> * the new time.
> */
> - void scheduleEscalation(final EscalationState state) {
> + public void scheduleEscalation(final EscalationState state) {
> final long schedTime = state.getNextActionTime();
>
> HQApp.getInstance().addTransactionListener(new
> TransactionListener() {
> public void afterCommit(boolean success) {
> _log.debug("Transaction committed: success=" +
> success);
> +
> if (success) {
> scheduleEscalation_(state, schedTime);
> }
> @@ -316,7 +380,8 @@
> _esclMan.executeState(stateId);
> }
>
> - static EscalationRuntime getInstance() {
> + public static EscalationRuntime getInstance() {
> return INSTANCE;
> }
> +
> }
>
> Modified: trunk/src/org/hyperic/hq/events/ext/AbstractTrigger.java
> ===================================================================
> --- trunk/src/org/hyperic/hq/events/ext/AbstractTrigger.java
> 2007-08-31 21:58:26 UTC (rev 5774)
> +++ trunk/src/org/hyperic/hq/events/ext/AbstractTrigger.java
> 2007-08-31 22:19:56 UTC (rev 5775)
> @@ -185,13 +185,15 @@
> new ClassicEscalatableCreator(alertDef, event);
>
> // Now start escalation
> - if (alertDef.getEscalation() != null) {
> - EscalationManagerEJBImpl.getOne().startEscalation
> (alertDef,
> -
> creator);
> - }
> - else {
> + boolean started = EscalationManagerEJBImpl.getOne()
> + .startEscalation(alertDef,
> creator);
> +
> + // If there is no escalation or it wasn't started,
> then execute
> + // the classic escalations.
> + if (!started) {
> creator.createEscalatable();
> }
> +
> } catch (FinderException e) {
> throw new ActionExecuteException(
> "Alert Definition not found for trigger: " + getId
> ());
>
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
I agree with this.
Moreover, it seems like none of this should be exposed outside of the
runtime. We should be asking the runtime to perform more general
tasks, not specific things like removing from caches or locking/
unlocking. Lots of those methods were originally marked as package-
private since they were internal and shouldn't be exposed.
-- Jon
On Sep 4, 2007, at 10:48 AM, Charles Lee wrote:
> Hi Tom,
>
> Two comments.
>
> In EscalationManagerEJBImpl.startEscalation(), I see that you left
> creator.createEscalatable() after checking for escalation state.
> Which occurred to me is the right behavior after our conversation.
> If an escalation is in progress, we don't even create new alerts
> for the same alert definition, and the original actions definitely
> should not execute.
>
> Would it be safer to put the EscalationRuntime.getInstance
> ().releaseMutex() finally block with the same try as the the
> EscalationRuntime.getInstance().acquireMutex()? A developer can
> come along and insert some code that throws an exception between
> the acquire and the try block, and you never release the mutex.
>
> Charles
>
>
>
> On Aug 31, 2007, at 3:19 PM, tkeeney@hyperic.com wrote:
>
>> Author: tkeeney
>> Date: 2007-08-31 15:19:56 -0700 (Fri, 31 Aug 2007)
>> New Revision: 5775
>> URL: http://svn.hyperic.org/?view=rev&root=Hyperic+HQ&revision=5775
>>
>> Modified:
>> trunk/src/org/hyperic/hq/escalation/server/session/
>> EscalationManagerEJBImpl.java
>> trunk/src/org/hyperic/hq/escalation/server/session/
>> EscalationRuntime.java
>> trunk/src/org/hyperic/hq/events/ext/AbstractTrigger.java
>> Log:
>> [HHQ-915] When attempting to start an escalation, need to account
>> for other threads having already started the escalation yet
>> haven't committed their txn yet. Keeping a cache of these
>> uncommitted txns so we know not to start the same escalation that
>> is now in progress.
>>
>> Modified: trunk/src/org/hyperic/hq/escalation/server/session/
>> EscalationManagerEJBImpl.java
>> ===================================================================
>> --- trunk/src/org/hyperic/hq/escalation/server/session/
>> EscalationManagerEJBImpl.java 2007-08-31 21:58:26 UTC (rev 5774)
>> +++ trunk/src/org/hyperic/hq/escalation/server/session/
>> EscalationManagerEJBImpl.java 2007-08-31 22:19:56 UTC (rev 5775)
>> @@ -245,35 +245,116 @@
>> }
>>
>> /**
>> - * Start an escalation. If the escalation has already been
>> started, then
>> - * this method call will be a no-op.
>> + * Start an escalation. If the entity performing escalations
>> does not have
>> + * an assigned escalation or if the escalation has already
>> been started,
>> + * then this method call will be a no-op.
>> *
>> - * @param def Definition to start the escalation for
>> + * @param def The entity performing escalations.
>> * @param creator Object which will create an {@link
>> Escalatable} object
>> * if invoking this method actually starts an
>> escalation.
>> + * @return
>> + *
>> there is
>> + * no escalation assigned to the entity or the
>> escalation
>> + * is already in progress.
>> *
>> * @ejb:interface-method
>> */
>> - public void startEscalation(PerformsEscalations def,
>> - EscalatableCreator creator)
>> - {
>> - EscalationState curState = _stateDAO.find(def);
>> - Escalatable alert;
>> -
>> + public boolean startEscalation(PerformsEscalations def,
>> + EscalatableCreator creator) {
>> if (def.getEscalation() == null)
>> - return;
>> + return false;
>>
>> - if (curState != null) {
>> + boolean started = false;
>> +
>> + try {
>> + // Get the lock right now!!! We have to be sure the
>> escalation
>> + // was scheduled for post commit before letting
>> anyone else through.
>> + // Assume we may throw an unchecked exception prior
>> to scheduling.
>> + // This is possible, especially when creating the
>> escalatable.
>> + EscalationRuntime.getInstance().acquireMutex();
>> +
>> + try {
>> + if (escalationStateExists(def)) {
>> + return started = false;
>> + }
>> +
>> + try {
>> + Escalatable alert = creator.createEscalatable();
>> + EscalationState curState = new EscalationState
>> (alert);
>> + _stateDAO.save(curState);
>> + _log.debug("Escalation started: state=" +
>> curState.getId());
>> + EscalationRuntime.getInstance
>> ().scheduleEscalation(curState);
>> + started = true;
>> + } finally {
>> + if (!started) {
>> + EscalationRuntime.getInstance()
>> + .removeFromUncommittedEscalationStateCac
>> he(def, false);
>> + }
>> + }
>> +
>> + } finally {
>> + EscalationRuntime.getInstance().releaseMutex();
>> + }
>> + } catch (InterruptedException e) {
>> + _log.error("Failed to start escalation for " +
>> + "alert def id="+def.getId()+
>> + "; type="+def.getAlertType().getCode(), e);
>> + }
>> +
>> + return started;
>> + }
>> +
>> + private boolean escalationStateExists(PerformsEscalations def) {
>> + // Checks if there is an uncommitted escalation state for
>> this def.
>> + boolean existsInCache = EscalationRuntime.getInstance()
>> + .addToUncommittedEscalationStateCach
>> e(def);
>> +
>> + boolean existsInDb = false;
>> +
>> + try {
>> + // Checks if there is a committed escalation state
>> for this def.
>> + existsInDb = _stateDAO.find(def) != null;
>> + } catch (Exception e) {
>> + _log.warn("There is more than one escalation in
>> progress for " +
>> + "alert def id="+def.getId()+
>> + "; type="+def.getAlertType().getCode(), e);
>> + // HHQ-915: A hibernate exception will occur when
>> looking up the
>> + // escalation state if more than one exists. This
>> shouldn't happen,
>> + // but if it does, don't create another escalation.
>> + existsInDb = true;
>> + }
>> +
>> + // Possible scenarios when storing an escalation state ->
>> + // how to remove the def from the uncommitted cache:
>> + // in_cache=false, in_db=false -> schedule to remove on
>> commit
>> + // in_cache=true, in_db=false -> do nothing,
>> + // - will be removed from
>> cache post-commit
>> + // in_cache=false, in_db=true -> remove immediately
>> + // in_cache=true, in_db=true -> (a timing issue),
>> + // - will be removed from
>> cache post-commit,
>> + // but to be safe, remove
>> immediately
>> + EscalationRuntime runtime = EscalationRuntime.getInstance();
>> +
>> + if (existsInCache) {
>> + if (existsInDb) {
>> + runtime.removeFromUncommittedEscalationStateCache
>> (def, false);
>> + } else {
>> + // do nothing
>> + }
>> + } else {
>> + if (existsInDb) {
>> + runtime.removeFromUncommittedEscalationStateCache
>> (def, false);
>> + } else {
>> + runtime.removeFromUncommittedEscalationStateCache
>> (def, true);
>> + }
>> + }
>> +
>> + if (existsInCache || existsInDb) {
>> _log.debug("startEscalation called on [" + def + "]
>> but it was " +
>> - "already running");
>> - return;
>> + "already running");
>> }
>>
>> - alert = creator.createEscalatable();
>> - curState = new EscalationState(alert);
>> - _stateDAO.save(curState);
>> - _log.debug("Escalation started: state=" + curState.getId());
>> - EscalationRuntime.getInstance().scheduleEscalation
>> (curState);
>> + return existsInCache || existsInDb;
>> }
>>
>> /**
>>
>> Modified: trunk/src/org/hyperic/hq/escalation/server/session/
>> EscalationRuntime.java
>> ===================================================================
>> --- trunk/src/org/hyperic/hq/escalation/server/session/
>> EscalationRuntime.java 2007-08-31 21:58:26 UTC (rev 5774)
>> +++ trunk/src/org/hyperic/hq/escalation/server/session/
>> EscalationRuntime.java 2007-08-31 22:19:56 UTC (rev 5775)
>> @@ -25,6 +25,7 @@
>> package org.hyperic.hq.escalation.server.session;
>>
>> import java.util.ArrayList;
>> +import java.util.Collections;
>> import java.util.Date;
>> import java.util.HashMap;
>> import java.util.HashSet;
>> @@ -42,6 +43,7 @@
>> import EDU.oswego.cs.dl.util.concurrent.Executor;
>> import EDU.oswego.cs.dl.util.concurrent.LinkedQueue;
>> import EDU.oswego.cs.dl.util.concurrent.PooledExecutor;
>> +import EDU.oswego.cs.dl.util.concurrent.Semaphore;
>>
>> /**
>> * This class manages the runtime execution of escalation
>> chains. The
>> @@ -71,9 +73,14 @@
>>
>> private final ThreadLocal _batchUnscheduleTxnListeners = new
>> ThreadLocal();
>> private final Log _log = LogFactory.getLog
>> (EscalationRuntime.class);
>> - private final ClockDaemon _schedule = new
>> ClockDaemon();
>> - private final Map _stateIdsToTasks = new
>> HashMap();
>> - private final Map
>> _esclEntityIdsToStateIds = new HashMap();
>> + private final ClockDaemon _schedule = new ClockDaemon();
>> + private final Map _stateIdsToTasks = new HashMap();
>> + private final Map _esclEntityIdsToStateIds = new HashMap();
>> +
>> + private final Semaphore _mutex = new Semaphore(1);
>> +
>> + private final Set _uncomittedEscalatingEntities =
>> + Collections.synchronizedSet(new
>> HashSet());
>> private final PooledExecutor _executor;
>> private final EscalationManagerLocal _esclMan;
>>
>> @@ -128,7 +135,7 @@
>> * Unschedule the execution of an escalation state. The
>> unschedule will
>> * only occur if the transaction successfully commits.
>> */
>> - void unscheduleEscalation(EscalationState state) {
>> + public void unscheduleEscalation(EscalationState state) {
>> final Integer stateId = state.getId();
>>
>> HQApp.getInstance().addTransactionListener(new
>> TransactionListener() {
>> @@ -148,7 +155,7 @@
>> * entity that performs escalations. The unschedule will only
>> occur if the
>> * transaction successfully commits.
>> */
>> - void unscheduleAllEscalationsFor(PerformsEscalations def) {
>> + public void unscheduleAllEscalationsFor(PerformsEscalations
>> def) {
>> BatchUnscheduleEscalationsTransactionListener
>> batchTxnListener =
>> (BatchUnscheduleEscalationsTransactionListener)
>> _batchUnscheduleTxnListeners.get();
>> @@ -260,6 +267,62 @@
>> }
>>
>> /**
>> + * Acquire the mutex.
>> + *
>> + * @throws InterruptedException
>> + */
>> + public void acquireMutex() throws InterruptedException {
>> + _mutex.acquire();
>> + }
>> +
>> + /**
>> + * Release the mutex.
>> + */
>> + public void releaseMutex() {
>> + _mutex.release();
>> + }
>> +
>> + /**
>> + * Add the uncommitted escalation state for this entity
>> performing escalations
>> + * to the uncommitted escalation state cache. This cache is
>> used to track
>> + * escalation states that have been scheduled but are not
>> visible to other
>> + * threads prior to the transaction commit.
>> + *
>> + * @param def The entity that performs escalations.
>> + * @return
>> uncommitted escalation state.
>> + */
>> + public boolean addToUncommittedEscalationStateCache
>> (PerformsEscalations def) {
>> + return !_uncomittedEscalatingEntities.add(new
>> EscalatingEntityIdentifier(def));
>> + }
>> +
>> + /**
>> + * Remove the uncommitted escalation state for this entity
>> performing
>> + * escalations from the uncommitted escalation state cache.
>> + *
>> + * @param def The entity that performs escalations.
>> + * @param postTxnCommit
>> commit;
>> + *
>> immediately.
>> + */
>> + public void removeFromUncommittedEscalationStateCache(final
>> PerformsEscalations def,
>> + boolean
>> postTxnCommit) {
>> + if (postTxnCommit) {
>> + HQApp.getInstance().addTransactionListener(new
>> TransactionListener() {
>> +
>> + public void afterCommit(boolean success) {
>> + removeFromUncommittedEscalationStateCache
>> (def, false);
>> + }
>> +
>> + public void beforeCommit() {
>> + }
>> +
>> + });
>> + } else {
>> + _uncomittedEscalatingEntities.remove(new
>> EscalatingEntityIdentifier(def));
>> + }
>> +
>> + }
>> +
>> + /**
>> * This method introduces an escalation state to the
>> runtime. The
>> * escalation will be invoked according to the next action
>> time of the
>> * state.
>> @@ -267,12 +330,13 @@
>> * If the state had been previously scheduled, it will be
>> rescheduled with
>> * the new time.
>> */
>> - void scheduleEscalation(final EscalationState state) {
>> + public void scheduleEscalation(final EscalationState state) {
>> final long schedTime = state.getNextActionTime();
>>
>> HQApp.getInstance().addTransactionListener(new
>> TransactionListener() {
>> public void afterCommit(boolean success) {
>> _log.debug("Transaction committed: success=" +
>> success);
>> +
>> if (success) {
>> scheduleEscalation_(state, schedTime);
>> }
>> @@ -316,7 +380,8 @@
>> _esclMan.executeState(stateId);
>> }
>>
>> - static EscalationRuntime getInstance() {
>> + public static EscalationRuntime getInstance() {
>> return INSTANCE;
>> }
>> +
>> }
>>
>> Modified: trunk/src/org/hyperic/hq/events/ext/AbstractTrigger.java
>> ===================================================================
>> --- trunk/src/org/hyperic/hq/events/ext/AbstractTrigger.java
>> 2007-08-31 21:58:26 UTC (rev 5774)
>> +++ trunk/src/org/hyperic/hq/events/ext/AbstractTrigger.java
>> 2007-08-31 22:19:56 UTC (rev 5775)
>> @@ -185,13 +185,15 @@
>> new ClassicEscalatableCreator(alertDef, event);
>>
>> // Now start escalation
>> - if (alertDef.getEscalation() != null) {
>> - EscalationManagerEJBImpl.getOne().startEscalation
>> (alertDef,
>> -
>> creator);
>> - }
>> - else {
>> + boolean started = EscalationManagerEJBImpl.getOne()
>> + .startEscalation
>> (alertDef, creator);
>> +
>> + // If there is no escalation or it wasn't started,
>> then execute
>> + // the classic escalations.
>> + if (!started) {
>> creator.createEscalatable();
>> }
>> +
>> } catch (FinderException e) {
>> throw new ActionExecuteException(
>> "Alert Definition not found for trigger: " + getId
>> ());
>>
>
>
Moreover, it seems like none of this should be exposed outside of the
runtime. We should be asking the runtime to perform more general
tasks, not specific things like removing from caches or locking/
unlocking. Lots of those methods were originally marked as package-
private since they were internal and shouldn't be exposed.
-- Jon
On Sep 4, 2007, at 10:48 AM, Charles Lee wrote:
> Hi Tom,
>
> Two comments.
>
> In EscalationManagerEJBImpl.startEscalation(), I see that you left
> creator.createEscalatable() after checking for escalation state.
> Which occurred to me is the right behavior after our conversation.
> If an escalation is in progress, we don't even create new alerts
> for the same alert definition, and the original actions definitely
> should not execute.
>
> Would it be safer to put the EscalationRuntime.getInstance
> ().releaseMutex() finally block with the same try as the the
> EscalationRuntime.getInstance().acquireMutex()? A developer can
> come along and insert some code that throws an exception between
> the acquire and the try block, and you never release the mutex.
>
> Charles
>
>
>
> On Aug 31, 2007, at 3:19 PM, tkeeney@hyperic.com wrote:
>
>> Author: tkeeney
>> Date: 2007-08-31 15:19:56 -0700 (Fri, 31 Aug 2007)
>> New Revision: 5775
>> URL: http://svn.hyperic.org/?view=rev&root=Hyperic+HQ&revision=5775
>>
>> Modified:
>> trunk/src/org/hyperic/hq/escalation/server/session/
>> EscalationManagerEJBImpl.java
>> trunk/src/org/hyperic/hq/escalation/server/session/
>> EscalationRuntime.java
>> trunk/src/org/hyperic/hq/events/ext/AbstractTrigger.java
>> Log:
>> [HHQ-915] When attempting to start an escalation, need to account
>> for other threads having already started the escalation yet
>> haven't committed their txn yet. Keeping a cache of these
>> uncommitted txns so we know not to start the same escalation that
>> is now in progress.
>>
>> Modified: trunk/src/org/hyperic/hq/escalation/server/session/
>> EscalationManagerEJBImpl.java
>> ===================================================================
>> --- trunk/src/org/hyperic/hq/escalation/server/session/
>> EscalationManagerEJBImpl.java 2007-08-31 21:58:26 UTC (rev 5774)
>> +++ trunk/src/org/hyperic/hq/escalation/server/session/
>> EscalationManagerEJBImpl.java 2007-08-31 22:19:56 UTC (rev 5775)
>> @@ -245,35 +245,116 @@
>> }
>>
>> /**
>> - * Start an escalation. If the escalation has already been
>> started, then
>> - * this method call will be a no-op.
>> + * Start an escalation. If the entity performing escalations
>> does not have
>> + * an assigned escalation or if the escalation has already
>> been started,
>> + * then this method call will be a no-op.
>> *
>> - * @param def Definition to start the escalation for
>> + * @param def The entity performing escalations.
>> * @param creator Object which will create an {@link
>> Escalatable} object
>> * if invoking this method actually starts an
>> escalation.
>> + * @return
true if the escalation is started;>> + *
false if not because either >> there is
>> + * no escalation assigned to the entity or the
>> escalation
>> + * is already in progress.
>> *
>> * @ejb:interface-method
>> */
>> - public void startEscalation(PerformsEscalations def,
>> - EscalatableCreator creator)
>> - {
>> - EscalationState curState = _stateDAO.find(def);
>> - Escalatable alert;
>> -
>> + public boolean startEscalation(PerformsEscalations def,
>> + EscalatableCreator creator) {
>> if (def.getEscalation() == null)
>> - return;
>> + return false;
>>
>> - if (curState != null) {
>> + boolean started = false;
>> +
>> + try {
>> + // Get the lock right now!!! We have to be sure the
>> escalation
>> + // was scheduled for post commit before letting
>> anyone else through.
>> + // Assume we may throw an unchecked exception prior
>> to scheduling.
>> + // This is possible, especially when creating the
>> escalatable.
>> + EscalationRuntime.getInstance().acquireMutex();
>> +
>> + try {
>> + if (escalationStateExists(def)) {
>> + return started = false;
>> + }
>> +
>> + try {
>> + Escalatable alert = creator.createEscalatable();
>> + EscalationState curState = new EscalationState
>> (alert);
>> + _stateDAO.save(curState);
>> + _log.debug("Escalation started: state=" +
>> curState.getId());
>> + EscalationRuntime.getInstance
>> ().scheduleEscalation(curState);
>> + started = true;
>> + } finally {
>> + if (!started) {
>> + EscalationRuntime.getInstance()
>> + .removeFromUncommittedEscalationStateCac
>> he(def, false);
>> + }
>> + }
>> +
>> + } finally {
>> + EscalationRuntime.getInstance().releaseMutex();
>> + }
>> + } catch (InterruptedException e) {
>> + _log.error("Failed to start escalation for " +
>> + "alert def id="+def.getId()+
>> + "; type="+def.getAlertType().getCode(), e);
>> + }
>> +
>> + return started;
>> + }
>> +
>> + private boolean escalationStateExists(PerformsEscalations def) {
>> + // Checks if there is an uncommitted escalation state for
>> this def.
>> + boolean existsInCache = EscalationRuntime.getInstance()
>> + .addToUncommittedEscalationStateCach
>> e(def);
>> +
>> + boolean existsInDb = false;
>> +
>> + try {
>> + // Checks if there is a committed escalation state
>> for this def.
>> + existsInDb = _stateDAO.find(def) != null;
>> + } catch (Exception e) {
>> + _log.warn("There is more than one escalation in
>> progress for " +
>> + "alert def id="+def.getId()+
>> + "; type="+def.getAlertType().getCode(), e);
>> + // HHQ-915: A hibernate exception will occur when
>> looking up the
>> + // escalation state if more than one exists. This
>> shouldn't happen,
>> + // but if it does, don't create another escalation.
>> + existsInDb = true;
>> + }
>> +
>> + // Possible scenarios when storing an escalation state ->
>> + // how to remove the def from the uncommitted cache:
>> + // in_cache=false, in_db=false -> schedule to remove on
>> commit
>> + // in_cache=true, in_db=false -> do nothing,
>> + // - will be removed from
>> cache post-commit
>> + // in_cache=false, in_db=true -> remove immediately
>> + // in_cache=true, in_db=true -> (a timing issue),
>> + // - will be removed from
>> cache post-commit,
>> + // but to be safe, remove
>> immediately
>> + EscalationRuntime runtime = EscalationRuntime.getInstance();
>> +
>> + if (existsInCache) {
>> + if (existsInDb) {
>> + runtime.removeFromUncommittedEscalationStateCache
>> (def, false);
>> + } else {
>> + // do nothing
>> + }
>> + } else {
>> + if (existsInDb) {
>> + runtime.removeFromUncommittedEscalationStateCache
>> (def, false);
>> + } else {
>> + runtime.removeFromUncommittedEscalationStateCache
>> (def, true);
>> + }
>> + }
>> +
>> + if (existsInCache || existsInDb) {
>> _log.debug("startEscalation called on [" + def + "]
>> but it was " +
>> - "already running");
>> - return;
>> + "already running");
>> }
>>
>> - alert = creator.createEscalatable();
>> - curState = new EscalationState(alert);
>> - _stateDAO.save(curState);
>> - _log.debug("Escalation started: state=" + curState.getId());
>> - EscalationRuntime.getInstance().scheduleEscalation
>> (curState);
>> + return existsInCache || existsInDb;
>> }
>>
>> /**
>>
>> Modified: trunk/src/org/hyperic/hq/escalation/server/session/
>> EscalationRuntime.java
>> ===================================================================
>> --- trunk/src/org/hyperic/hq/escalation/server/session/
>> EscalationRuntime.java 2007-08-31 21:58:26 UTC (rev 5774)
>> +++ trunk/src/org/hyperic/hq/escalation/server/session/
>> EscalationRuntime.java 2007-08-31 22:19:56 UTC (rev 5775)
>> @@ -25,6 +25,7 @@
>> package org.hyperic.hq.escalation.server.session;
>>
>> import java.util.ArrayList;
>> +import java.util.Collections;
>> import java.util.Date;
>> import java.util.HashMap;
>> import java.util.HashSet;
>> @@ -42,6 +43,7 @@
>> import EDU.oswego.cs.dl.util.concurrent.Executor;
>> import EDU.oswego.cs.dl.util.concurrent.LinkedQueue;
>> import EDU.oswego.cs.dl.util.concurrent.PooledExecutor;
>> +import EDU.oswego.cs.dl.util.concurrent.Semaphore;
>>
>> /**
>> * This class manages the runtime execution of escalation
>> chains. The
>> @@ -71,9 +73,14 @@
>>
>> private final ThreadLocal _batchUnscheduleTxnListeners = new
>> ThreadLocal();
>> private final Log _log = LogFactory.getLog
>> (EscalationRuntime.class);
>> - private final ClockDaemon _schedule = new
>> ClockDaemon();
>> - private final Map _stateIdsToTasks = new
>> HashMap();
>> - private final Map
>> _esclEntityIdsToStateIds = new HashMap();
>> + private final ClockDaemon _schedule = new ClockDaemon();
>> + private final Map _stateIdsToTasks = new HashMap();
>> + private final Map _esclEntityIdsToStateIds = new HashMap();
>> +
>> + private final Semaphore _mutex = new Semaphore(1);
>> +
>> + private final Set _uncomittedEscalatingEntities =
>> + Collections.synchronizedSet(new
>> HashSet());
>> private final PooledExecutor _executor;
>> private final EscalationManagerLocal _esclMan;
>>
>> @@ -128,7 +135,7 @@
>> * Unschedule the execution of an escalation state. The
>> unschedule will
>> * only occur if the transaction successfully commits.
>> */
>> - void unscheduleEscalation(EscalationState state) {
>> + public void unscheduleEscalation(EscalationState state) {
>> final Integer stateId = state.getId();
>>
>> HQApp.getInstance().addTransactionListener(new
>> TransactionListener() {
>> @@ -148,7 +155,7 @@
>> * entity that performs escalations. The unschedule will only
>> occur if the
>> * transaction successfully commits.
>> */
>> - void unscheduleAllEscalationsFor(PerformsEscalations def) {
>> + public void unscheduleAllEscalationsFor(PerformsEscalations
>> def) {
>> BatchUnscheduleEscalationsTransactionListener
>> batchTxnListener =
>> (BatchUnscheduleEscalationsTransactionListener)
>> _batchUnscheduleTxnListeners.get();
>> @@ -260,6 +267,62 @@
>> }
>>
>> /**
>> + * Acquire the mutex.
>> + *
>> + * @throws InterruptedException
>> + */
>> + public void acquireMutex() throws InterruptedException {
>> + _mutex.acquire();
>> + }
>> +
>> + /**
>> + * Release the mutex.
>> + */
>> + public void releaseMutex() {
>> + _mutex.release();
>> + }
>> +
>> + /**
>> + * Add the uncommitted escalation state for this entity
>> performing escalations
>> + * to the uncommitted escalation state cache. This cache is
>> used to track
>> + * escalation states that have been scheduled but are not
>> visible to other
>> + * threads prior to the transaction commit.
>> + *
>> + * @param def The entity that performs escalations.
>> + * @return
true if there is already an >> uncommitted escalation state.
>> + */
>> + public boolean addToUncommittedEscalationStateCache
>> (PerformsEscalations def) {
>> + return !_uncomittedEscalatingEntities.add(new
>> EscalatingEntityIdentifier(def));
>> + }
>> +
>> + /**
>> + * Remove the uncommitted escalation state for this entity
>> performing
>> + * escalations from the uncommitted escalation state cache.
>> + *
>> + * @param def The entity that performs escalations.
>> + * @param postTxnCommit
true to remove post txn >> commit;
>> + *
false to remove >> immediately.
>> + */
>> + public void removeFromUncommittedEscalationStateCache(final
>> PerformsEscalations def,
>> + boolean
>> postTxnCommit) {
>> + if (postTxnCommit) {
>> + HQApp.getInstance().addTransactionListener(new
>> TransactionListener() {
>> +
>> + public void afterCommit(boolean success) {
>> + removeFromUncommittedEscalationStateCache
>> (def, false);
>> + }
>> +
>> + public void beforeCommit() {
>> + }
>> +
>> + });
>> + } else {
>> + _uncomittedEscalatingEntities.remove(new
>> EscalatingEntityIdentifier(def));
>> + }
>> +
>> + }
>> +
>> + /**
>> * This method introduces an escalation state to the
>> runtime. The
>> * escalation will be invoked according to the next action
>> time of the
>> * state.
>> @@ -267,12 +330,13 @@
>> * If the state had been previously scheduled, it will be
>> rescheduled with
>> * the new time.
>> */
>> - void scheduleEscalation(final EscalationState state) {
>> + public void scheduleEscalation(final EscalationState state) {
>> final long schedTime = state.getNextActionTime();
>>
>> HQApp.getInstance().addTransactionListener(new
>> TransactionListener() {
>> public void afterCommit(boolean success) {
>> _log.debug("Transaction committed: success=" +
>> success);
>> +
>> if (success) {
>> scheduleEscalation_(state, schedTime);
>> }
>> @@ -316,7 +380,8 @@
>> _esclMan.executeState(stateId);
>> }
>>
>> - static EscalationRuntime getInstance() {
>> + public static EscalationRuntime getInstance() {
>> return INSTANCE;
>> }
>> +
>> }
>>
>> Modified: trunk/src/org/hyperic/hq/events/ext/AbstractTrigger.java
>> ===================================================================
>> --- trunk/src/org/hyperic/hq/events/ext/AbstractTrigger.java
>> 2007-08-31 21:58:26 UTC (rev 5774)
>> +++ trunk/src/org/hyperic/hq/events/ext/AbstractTrigger.java
>> 2007-08-31 22:19:56 UTC (rev 5775)
>> @@ -185,13 +185,15 @@
>> new ClassicEscalatableCreator(alertDef, event);
>>
>> // Now start escalation
>> - if (alertDef.getEscalation() != null) {
>> - EscalationManagerEJBImpl.getOne().startEscalation
>> (alertDef,
>> -
>> creator);
>> - }
>> - else {
>> + boolean started = EscalationManagerEJBImpl.getOne()
>> + .startEscalation
>> (alertDef, creator);
>> +
>> + // If there is no escalation or it wasn't started,
>> then execute
>> + // the classic escalations.
>> + if (!started) {
>> creator.createEscalatable();
>> }
>> +
>> } catch (FinderException e) {
>> throw new ActionExecuteException(
>> "Alert Definition not found for trigger: " + getId
>> ());
>>
>
>