Re: [xmlblaster] missing volatile messages

[Eugene Wu <ewu at navahonetworks.com>]

Marcel,

I fixed a multithread racing issue which caused message losing. The 
patch is attached with the mail.

It was concerned with org.xmlBlaster.engine.TopicHandler.publish() and 
entryDestoyed(). Let's assume thread A just called toAlive() in line 
537, which set topic state to ALIVE. Before thread A pushs a msgUnit 
into callback queue(s) through 
TopicHandler.invokeCallbackAndHandleFailure() in line 645 --> 
invokeCallback() in line 1337, another thread B just finished its 
publish in the same topic and might triggered a message distroying, 
which, in turn, might switch topic  state  to  UNREFERENCED. When thread 
A reached TopicHandler.invokeCallback() in line 1321, it would find the 
topic was in invalid state and droped the message.

I used a publishCounter to record the number of running threads in the 
publish method. When entryDetryed() intents to change topic state to 
UNREFERENCED, it will first check the publishCounter. If it is greater 
than one, don't change topic state.

I just dealed with one scenario in one possible way without looking at 
the whole picture.

Eugene
--- xmlBlaster_1.1.1_orig/src/java/org/xmlBlaster/engine/TopicHandler.java	2006-01-03 19:01:41.000000000 +0000
+++ xmlBlaster_1.1.1_my/src/java/org/xmlBlaster/engine/TopicHandler.java	2006-04-12 15:16:53.000000000 +0000
 at  at  -139,6 +139,8  at  at 
 
    private boolean isRegisteredInBigXmlDom = false;
 
+   private int publishCounter = 0; //count the threads running in publish method
+   
    /**
     * This topic is destroyed after given timeout
     * The timer is activated on state change to UNREFERENCED
 at  at  -566,7 +568,12  at  at 
             }
 
             msgUnitWrapper = new MsgUnitWrapper(glob, msgUnit, this.msgUnitCache, initialCounter, 0, -1);
-
+            
+            publishCounter++;
+            if (!isAlive()) {
+                toAlive();
+            }
+            
             // Forcing RAM entry temporary (reset in finally below) to avoid performance critical harddisk IO during initialization, every callback/subject/history queue put()/take() is changing the reference counter of MsgUnitWrapper. For persistent messages this needs to be written to harddisk
             // If the server crashed during this RAM operation it is not critical as the publisher didn't get an ACK yet
             synchronized(this.msgUnitWrapperUnderConstruction) {
 at  at  -649,6 +656,7  at  at 
       finally {
          if (msgUnitWrapper != null) {
             synchronized(this) {
+               publishCounter--;
                synchronized(msgUnitWrapper) {
                   synchronized(this.msgUnitCache) {
                      try {
 at  at  -941,9 +949,12  at  at 
                if (isSoftErased()) {
                   notifyList = toDead(this.creatorSessionName, false);
                }
-               else {
+               else if (publishCounter==0) {
                   notifyList = toUnreferenced(false);
                }
+               else {
+            	   if (log.TRACE) log.trace(ME, "ignored the attempt to set topic unreferenced as other thread in publish");
+               }
             }
             catch (XmlBlasterException e) {
                log.error(ME, "Internal problem with entryDestroyed: " + e.getMessage() + ": " + toXml());