[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
Re: [xmlblaster] Missing messages
Amendment on my fix:
TopicHandler: moved destroyTimer check point from "private ArrayList
timeout()" to "public final void timeout(Object userData)" in order to
enable the immediate destruction when destroyDelay = 0.
My approach still leaves some gap in order to avoid dead lock:
After a publishing thread unset the destroyTimer and just began to
do actual publishing, another publishing thread just finished its job
and could reset the destroyTimer.
In my testing, the chance didn't occur, but it's possible
theoretically. So my approach only reduced the chance but not solved the
problem completely.
Eugene Wu wrote:
Hi, there:
Figured out one cause for the issue. It could happen when a pulishing
thread and a timeout thread work on a same TopicHandler instance.
For example, after thread A (publishing) retrieved a topicHandler from
topicHandlerMap in line 1695 of RoquestBroker:
Object obj = topicHandlerMap.get(msgUnit.getKeyOid());
the topicHandler's publish method would be invoked. As a synchronized
lock on the topicHandler can not be applyed on the whole pubish()
method (dead-lock concerns), another thread B (timeout) could get a
chance to invoke timeout(Object userData) method of the topicHandler
before thread A sets the topicHandler's state to ALIVE. Therefore, the
topicHandler could be publishing a message and, at the same time, also
could be destroyed, which caused unpredictable results.
In my fix (attached), both publishing thread and timeout thread have a
check point, in which three things are done to let a slightly ahead
thread be able to prevent a slightly behind competing thread from
running:
1) Get the lock on the topicHandler (synchronized lock)
2) Check if the topicHandler is working with a competing thread. If
yes, stop the current thread for timeout or stop using the handler for
publishing. Otherwise, goto step 3.
3) Setup a check mark to prevent its competing thread running. The
mark must be cleared by the framework in a proper time later on.
Here are more detailed scenarios:
a) Publishing thread ahead
After thread A got the topicHandler, its state will be checked. If
it is dead (and should be destroyed now because of a synchronized
lock in timeout method), a new TopicHandler will be created. If the
retrieved topicHandle is not dead, its destroyTimer will be removed.
Later on, thread B will be stoped at the point where it check for the
destroyTimer's null value.
The destroyTimer will be reset by the framework when the
topicHandler is turned to UNREFERENCED.
b) Timeout thread ahead
After thread B got the topicHandler, it will check for the
existence of destroyTimer. If not existed, stop the timeout event.
Otherwise, the thread will continue. Because of a synchronized lock,
it will set the topicHandler as DEAD eventually, which will prevent
thread A from using the obsolete handler.
A dead topicHandle will be garbage collected. (need to confirm this!)
Regards,
Eugene
Index: org/xmlBlaster/engine/TopicHandler.java
===================================================================
--- org/xmlBlaster/engine/TopicHandler.java (revision 15178)
+++ org/xmlBlaster/engine/TopicHandler.java (working copy)
at at -1946,6 +1946,14 at at
return notifyList;
}
+ void unsetDestroyTimer() {
+ if (this.timerKey != null) {
+ log.finer("Unset destroyTimer for TopicHandler: "+this.getUniqueKey());
+ this.destroyTimer.removeTimeoutListener(this.timerKey);
+ this.timerKey = null;
+ }
+ }
+
/**
* Merge the message DOM tree into the big xmlBlaster DOM tree
*/
at at -2196,8 +2204,16 at at
* This timeout occurs after a configured delay (destroyDelay) in UNREFERENCED state
*/
public final void timeout(Object userData) {
- if (log.isLoggable(Level.FINER)) log.finer("Timeout after destroy delay occurred - destroying topic now ...");
- ArrayList notifyList = timeout();
+ if (log.isLoggable(Level.FINER)) log.finer("Timeout after destroy delay occurred - destroying topic (oid="+getUniqueKey()+") now ...");
+ ArrayList notifyList = null;
+ synchronized (this) {
+ if (timerKey==null) {
+ log.finer("Ignored timeout event for the timer was unset by a publishing thread");
+ return;
+ }
+ notifyList = timeout();
+ }
+
if (notifyList != null) notifySubscribersAboutErase(notifyList); // must be outside the synchronize
}
Index: org/xmlBlaster/engine/RequestBroker.java
===================================================================
--- org/xmlBlaster/engine/RequestBroker.java (revision 15178)
+++ org/xmlBlaster/engine/RequestBroker.java (working copy)
at at -1686,6 +1686,7 at at
// Handle local message
// Find or create the topic
+ boolean newlycreated = false;
TopicHandler topicHandler = null;
synchronized(this.topicHandlerMap) {
if (!msgKeyData.getOid().equals(msgUnit.getKeyOid())) {
at at -1695,11 +1696,25 at at
Object obj = topicHandlerMap.get(msgUnit.getKeyOid());
if (obj == null) {
topicHandler = new TopicHandler(this, sessionInfo, msgUnit.getKeyOid()); // adds itself to topicHandlerMap
+ newlycreated = true;
}
else {
topicHandler = (TopicHandler)obj;
}
}
+
+ if (!newlycreated && !topicHandler.isAlive()) {
+ log.finer("The topic is not ALIVE, need to check if it is dying or not");
+ synchronized(topicHandler) { // wiating if the topicHandler is dying.
+ if (topicHandler.isDead()) {
+ log.finer("The topic is DEAD, need to create a new one");
+ topicHandler = new TopicHandler(this, sessionInfo, msgUnit.getKeyOid());
+ }
+ else {
+ topicHandler.unsetDestroyTimer(); // to provent the topicHandler from being timed out
+ }
+ }
+ }
// Process the message
publishReturnQos = topicHandler.publish(sessionInfo, msgUnit, publishQos);