activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dej...@apache.org
Subject svn commit: r882511 - in /activemq/branches/activemq-5.3: activemq-core/src/main/java/org/apache/activemq/broker/ activemq-core/src/main/java/org/apache/activemq/store/amq/ activemq-core/src/main/java/org/apache/activemq/store/journal/ activemq-core/sr...
Date Fri, 20 Nov 2009 11:36:47 GMT
Author: dejanb
Date: Fri Nov 20 11:36:45 2009
New Revision: 882511

URL: http://svn.apache.org/viewvc?rev=882511&view=rev
Log:
merging https://issues.apache.org/activemq/browse/AMQ-2042 - 834922,835373,835412,835833,835888,880792,881221,882144

Added:
    activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/util/DefaultIOExceptionHandler.java
      - copied, changed from r835888, activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/DefaultIOExceptionHandler.java
    activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/util/IOExceptionHandler.java
      - copied unchanged from r835888, activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/IOExceptionHandler.java
Modified:
    activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/broker/BrokerService.java
    activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQPersistenceAdapter.java
    activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/journal/JournalPersistenceAdapter.java
    activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaPersistenceAdapter.java
    activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaTransactionStore.java
    activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBPersistenceAdapter.java
    activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
    activemq/branches/activemq-5.3/kahadb/src/main/java/org/apache/kahadb/journal/DataFileAppender.java
    activemq/branches/activemq-5.3/kahadb/src/main/java/org/apache/kahadb/page/PageFile.java

Modified: activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/broker/BrokerService.java
URL: http://svn.apache.org/viewvc/activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/broker/BrokerService.java?rev=882511&r1=882510&r2=882511&view=diff
==============================================================================
--- activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/broker/BrokerService.java
(original)
+++ activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/broker/BrokerService.java
Fri Nov 20 11:36:45 2009
@@ -84,6 +84,8 @@
 import org.apache.activemq.transport.TransportServer;
 import org.apache.activemq.transport.vm.VMTransportFactory;
 import org.apache.activemq.usage.SystemUsage;
+import org.apache.activemq.util.DefaultIOExceptionHandler;
+import org.apache.activemq.util.IOExceptionHandler;
 import org.apache.activemq.util.IOExceptionSupport;
 import org.apache.activemq.util.IOHelper;
 import org.apache.activemq.util.JMXSupport;
@@ -178,7 +180,9 @@
     private int systemExitOnShutdownExitCode;
     private SslContext sslContext;
     private boolean forceStart = false;
-    static {
+    private IOExceptionHandler ioExceptionHandler;
+
+	static {
         String localHostName = "localhost";
         try {
             localHostName = java.net.InetAddress.getLocalHost().getHostName();
@@ -481,6 +485,9 @@
                 }
             }
             brokerId = broker.getBrokerId();
+            if (ioExceptionHandler == null) {
+            	setIoExceptionHandler(new DefaultIOExceptionHandler());
+            }
             LOG.info("ActiveMQ JMS Message Broker (" + getBrokerName() + ", " + brokerId
+ ") started");
             getBroker().brokerServiceStarted();
             startedLatch.countDown();
@@ -2008,6 +2015,14 @@
             }
         }
     }
+    
+    public void handleIOException(IOException exception) {
+        if (ioExceptionHandler != null) {
+            ioExceptionHandler.handle(exception);
+         } else {
+            LOG.info("Ignoring IO exception, " + exception, exception);
+         }
+    }
 
     /**
      * Starts all destiantions in persistence store. This includes all inactive
@@ -2111,5 +2126,10 @@
         this.passiveSlave = passiveSlave;
     }
     
+    public void setIoExceptionHandler(IOExceptionHandler ioExceptionHandler) {
+        ioExceptionHandler.setBrokerService(this);
+        this.ioExceptionHandler = ioExceptionHandler;
+    }
+    
    
 }
\ No newline at end of file

Modified: activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQPersistenceAdapter.java
URL: http://svn.apache.org/viewvc/activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQPersistenceAdapter.java?rev=882511&r1=882510&r2=882511&view=diff
==============================================================================
--- activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQPersistenceAdapter.java
(original)
+++ activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQPersistenceAdapter.java
Fri Nov 20 11:36:45 2009
@@ -695,7 +695,13 @@
     }
     
     public Location writeCommand(DataStructure command, boolean syncHint,boolean forceSync)
throws IOException {
-        return asyncDataManager.write(wireFormat.marshal(command), (forceSync||(syncHint
&& syncOnWrite)));
+    	try {
+    		return asyncDataManager.write(wireFormat.marshal(command), (forceSync||(syncHint &&
syncOnWrite)));
+    	} catch (IOException ioe) {
+    		LOG.error("Failed to write command: " + command + ". Reason: " + ioe, ioe);
+        	brokerService.handleIOException(ioe);
+        	throw ioe;
+        }
     }
 
     private Location writeTraceMessage(String message, boolean sync) throws IOException {

Modified: activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/journal/JournalPersistenceAdapter.java
URL: http://svn.apache.org/viewvc/activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/journal/JournalPersistenceAdapter.java?rev=882511&r1=882510&r2=882511&view=diff
==============================================================================
--- activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/journal/JournalPersistenceAdapter.java
(original)
+++ activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/journal/JournalPersistenceAdapter.java
Fri Nov 20 11:36:45 2009
@@ -623,7 +623,7 @@
         	    return journal.write(toPacket(wireFormat.marshal(command)), sync);
             } catch (IOException ioe) {
         	    LOG.error("Cannot write to the journal", ioe);
-        	    stopBroker();
+        	    brokerService.handleIOException(ioe);
         	    throw ioe;
             }
         }
@@ -725,17 +725,5 @@
             ((BrokerServiceAware)pa).setBrokerService(brokerService);
         }
     }
-    
-    protected void stopBroker() {
-        new Thread() {
-           public void run() {
-        	   try {
-    	            brokerService.stop();
-    	        } catch (Exception e) {
-    	            LOG.warn("Failure occured while stopping broker");
-    	        }    			
-    		}
-    	}.start();
-    }
 
 }

Modified: activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaPersistenceAdapter.java
URL: http://svn.apache.org/viewvc/activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaPersistenceAdapter.java?rev=882511&r1=882510&r2=882511&view=diff
==============================================================================
--- activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaPersistenceAdapter.java
(original)
+++ activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaPersistenceAdapter.java
Fri Nov 20 11:36:45 2009
@@ -24,6 +24,8 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.BrokerServiceAware;
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ActiveMQQueue;
@@ -54,7 +56,7 @@
  * @org.apache.xbean.XBean
  * @version $Revision: 1.4 $
  */
-public class KahaPersistenceAdapter implements PersistenceAdapter {
+public class KahaPersistenceAdapter implements PersistenceAdapter, BrokerServiceAware {
 
     private static final int STORE_LOCKED_WAIT_DELAY = 10 * 1000;
     private static final Log LOG = LogFactory.getLog(KahaPersistenceAdapter.class);
@@ -73,6 +75,7 @@
     private boolean initialized;
     private final AtomicLong storeSize;
     private boolean persistentIndex = true;
+    private BrokerService brokerService;
 
     
     public KahaPersistenceAdapter(AtomicLong size) {
@@ -175,6 +178,7 @@
                     container.setValueMarshaller(new TransactionMarshaller(wireFormat));
                     container.load();
                     transactionStore = new KahaTransactionStore(this, container);
+                    transactionStore.setBrokerService(brokerService);
                     break;
                 } catch (StoreLockedExcpetion e) {
                     LOG.info("Store is locked... waiting " + (STORE_LOCKED_WAIT_DELAY / 1000)
@@ -361,6 +365,10 @@
             wireFormat.setTightEncodingEnabled(true);
         }
     }
+
+	public void setBrokerService(BrokerService brokerService) {
+		this.brokerService = brokerService;
+	}
   
 
 }

Modified: activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaTransactionStore.java
URL: http://svn.apache.org/viewvc/activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaTransactionStore.java?rev=882511&r1=882510&r2=882511&view=diff
==============================================================================
--- activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaTransactionStore.java
(original)
+++ activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaTransactionStore.java
Fri Nov 20 11:36:45 2009
@@ -24,17 +24,23 @@
 
 import javax.transaction.xa.XAException;
 
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.BrokerServiceAware;
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.command.Message;
 import org.apache.activemq.command.MessageAck;
 import org.apache.activemq.command.TransactionId;
 import org.apache.activemq.command.XATransactionId;
+import org.apache.activemq.kaha.RuntimeStoreException;
 import org.apache.activemq.store.MessageStore;
 import org.apache.activemq.store.ProxyMessageStore;
 import org.apache.activemq.store.ProxyTopicMessageStore;
 import org.apache.activemq.store.TopicMessageStore;
 import org.apache.activemq.store.TransactionRecoveryListener;
 import org.apache.activemq.store.TransactionStore;
+import org.apache.activemq.store.journal.JournalPersistenceAdapter;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 
 /**
  * Provides a TransactionStore implementation that can create transaction aware
@@ -42,10 +48,14 @@
  * 
  * @version $Revision: 1.4 $
  */
-public class KahaTransactionStore implements TransactionStore {
+public class KahaTransactionStore implements TransactionStore, BrokerServiceAware {	
+    private static final Log LOG = LogFactory.getLog(KahaTransactionStore.class);
+	
     private Map transactions = new ConcurrentHashMap();
     private Map prepared;
     private KahaPersistenceAdapter adaptor;
+    
+    private BrokerService brokerService;
 
     KahaTransactionStore(KahaPersistenceAdapter adaptor, Map preparedMap) {
         this.adaptor = adaptor;
@@ -130,12 +140,19 @@
      * @throws IOException
      */
     void addMessage(final MessageStore destination, final Message message) throws IOException
{
-        if (message.isInTransaction()) {
-            KahaTransaction tx = getOrCreateTx(message.getTransactionId());
-            tx.add((KahaMessageStore)destination, message);
-        } else {
-            destination.addMessage(null, message);
-        }
+    	try {
+    		if (message.isInTransaction()) {
+    			KahaTransaction tx = getOrCreateTx(message.getTransactionId());
+    			tx.add((KahaMessageStore)destination, message);
+    		} else {
+    			destination.addMessage(null, message);
+    		}
+    	} catch (RuntimeStoreException rse) {
+            if (rse.getCause() instanceof IOException) {
+                brokerService.handleIOException((IOException)rse.getCause());
+            }
+            throw rse;
+    	}
     }
 
     /**
@@ -143,12 +160,19 @@
      * @throws IOException
      */
     final void removeMessage(final MessageStore destination, final MessageAck ack) throws
IOException {
-        if (ack.isInTransaction()) {
-            KahaTransaction tx = getOrCreateTx(ack.getTransactionId());
-            tx.add((KahaMessageStore)destination, ack);
-        } else {
-            destination.removeMessage(null, ack);
-        }
+    	try {
+    		if (ack.isInTransaction()) {
+    			KahaTransaction tx = getOrCreateTx(ack.getTransactionId());
+    			tx.add((KahaMessageStore)destination, ack);
+    		} else {
+    			destination.removeMessage(null, ack);
+    		}
+    	} catch (RuntimeStoreException rse) {
+            if (rse.getCause() instanceof IOException) {
+                brokerService.handleIOException((IOException)rse.getCause());
+            }
+            throw rse;
+    	}
     }
 
     protected synchronized KahaTransaction getTx(TransactionId key) {
@@ -181,4 +205,8 @@
     protected MessageStore getStoreById(Object id) {
         return adaptor.retrieveMessageStore(id);
     }
+
+	public void setBrokerService(BrokerService brokerService) {
+		this.brokerService = brokerService;
+	}
 }

Modified: activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBPersistenceAdapter.java
URL: http://svn.apache.org/viewvc/activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBPersistenceAdapter.java?rev=882511&r1=882510&r2=882511&view=diff
==============================================================================
--- activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBPersistenceAdapter.java
(original)
+++ activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBPersistenceAdapter.java
Fri Nov 20 11:36:45 2009
@@ -17,6 +17,8 @@
 package org.apache.activemq.store.kahadb;
 
 import org.apache.activeio.journal.Journal;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.BrokerServiceAware;
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ActiveMQQueue;
@@ -37,7 +39,7 @@
  * @org.apache.xbean.XBean element="kahaDB"
  * @version $Revision: 1.17 $
  */
-public class KahaDBPersistenceAdapter implements PersistenceAdapter {
+public class KahaDBPersistenceAdapter implements PersistenceAdapter, BrokerServiceAware {
     private KahaDBStore letter = new KahaDBStore();
     
 
@@ -364,4 +366,8 @@
     public void setCheckForCorruptJournalFiles(boolean checkForCorruptJournalFiles) {
         letter.setCheckForCorruptJournalFiles(checkForCorruptJournalFiles);
     }
+
+	public void setBrokerService(BrokerService brokerService) {
+		letter.setBrokerService(brokerService);
+	}
 }

Modified: activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
URL: http://svn.apache.org/viewvc/activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java?rev=882511&r1=882510&r2=882511&view=diff
==============================================================================
--- activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
(original)
+++ activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java
Fri Nov 20 11:36:45 2009
@@ -36,6 +36,8 @@
 import java.util.Map.Entry;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.BrokerServiceAware;
 import org.apache.activemq.command.ConnectionId;
 import org.apache.activemq.command.LocalTransactionId;
 import org.apache.activemq.command.SubscriptionInfo;
@@ -75,8 +77,11 @@
 import org.apache.kahadb.util.SequenceSet;
 import org.apache.kahadb.util.StringMarshaller;
 import org.apache.kahadb.util.VariableMarshaller;
+import org.springframework.core.enums.LetterCodedLabeledEnum;
 
-public class MessageDatabase {
+public class MessageDatabase implements BrokerServiceAware {
+	
+	private BrokerService brokerService;
 
     public static final String PROPERTY_LOG_SLOW_ACCESS_TIME = "org.apache.activemq.store.kahadb.LOG_SLOW_ACCESS_TIME";
     public static final int LOG_SLOW_ACCESS_TIME = Integer.parseInt(System.getProperty(PROPERTY_LOG_SLOW_ACCESS_TIME,
"500"));
@@ -227,6 +232,41 @@
         }
 	}
 	
+	private void startCheckpoint() {
+        checkpointThread = new Thread("ActiveMQ Journal Checkpoint Worker") {
+            public void run() {
+                try {
+                    long lastCleanup = System.currentTimeMillis();
+                    long lastCheckpoint = System.currentTimeMillis();
+                    // Sleep for a short time so we can periodically check 
+                    // to see if we need to exit this thread.
+                    long sleepTime = Math.min(checkpointInterval, 500);
+                    while (opened.get()) {
+                        
+                        Thread.sleep(sleepTime);
+                        long now = System.currentTimeMillis();
+                        if( now - lastCleanup >= cleanupInterval ) {
+                            checkpointCleanup(true);
+                            lastCleanup = now;
+                            lastCheckpoint = now;
+                        } else if( now - lastCheckpoint >= checkpointInterval ) {
+                            checkpointCleanup(false);
+                            lastCheckpoint = now;
+                        }
+                    }
+                } catch (InterruptedException e) {
+                    // Looks like someone really wants us to exit this thread...
+                } catch (IOException ioe) {
+                    LOG.error("Checkpoint failed", ioe);
+                    brokerService.handleIOException(ioe);
+                }
+            }
+                    
+        };
+        checkpointThread.setDaemon(true);
+        checkpointThread.start();
+	}
+	
 	/**
 	 * @throws IOException
 	 */
@@ -236,34 +276,7 @@
             
 	        loadPageFile();
 	        
-	        checkpointThread = new Thread("ActiveMQ Journal Checkpoint Worker") {
-	            public void run() {
-	                try {
-	                    long lastCleanup = System.currentTimeMillis();
-	                    long lastCheckpoint = System.currentTimeMillis();
-	                    
-	                    // Sleep for a short time so we can periodically check 
-	                    // to see if we need to exit this thread.
-	                    long sleepTime = Math.min(checkpointInterval, 500);
-	                    while (opened.get()) {
-	                        Thread.sleep(sleepTime);
-	                        long now = System.currentTimeMillis();
-	                        if( now - lastCleanup >= cleanupInterval ) {
-	                            checkpointCleanup(true);
-	                            lastCleanup = now;
-	                            lastCheckpoint = now;
-	                        } else if( now - lastCheckpoint >= checkpointInterval ) {
-	                            checkpointCleanup(false);
-	                            lastCheckpoint = now;
-	                        }
-	                    }
-	                } catch (InterruptedException e) {
-	                    // Looks like someone really wants us to exit this thread...
-	                }
-	            }
-	        };
-	        checkpointThread.setDaemon(true);
-	        checkpointThread.start();
+	        startCheckpoint();
             recover();
 		}
 	}
@@ -575,26 +588,22 @@
         return journal.getNextLocation(null);
 	}
 
-    protected void checkpointCleanup(final boolean cleanup) {
-        try {
-        	long start = System.currentTimeMillis();
-            synchronized (indexMutex) {
-            	if( !opened.get() ) {
-            		return;
-            	}
-                pageFile.tx().execute(new Transaction.Closure<IOException>() {
-                    public void execute(Transaction tx) throws IOException {
-                        checkpointUpdate(tx, cleanup);
-                    }
-                });
-            }
-        	long end = System.currentTimeMillis();
-        	if( LOG_SLOW_ACCESS_TIME>0 && end-start > LOG_SLOW_ACCESS_TIME) {
-        		LOG.info("Slow KahaDB access: cleanup took "+(end-start));
+    protected void checkpointCleanup(final boolean cleanup) throws IOException {
+    	long start = System.currentTimeMillis();
+        synchronized (indexMutex) {
+        	if( !opened.get() ) {
+        		return;
         	}
-        } catch (IOException e) {
-        	e.printStackTrace();
+            pageFile.tx().execute(new Transaction.Closure<IOException>() {
+                public void execute(Transaction tx) throws IOException {
+                    checkpointUpdate(tx, cleanup);
+                }
+            });
         }
+    	long end = System.currentTimeMillis();
+    	if( LOG_SLOW_ACCESS_TIME>0 && end-start > LOG_SLOW_ACCESS_TIME) {
+    		LOG.info("Slow KahaDB access: cleanup took "+(end-start));
+    	}
     }
 
     
@@ -617,32 +626,40 @@
     }
 
     /**
-     * All updated are are funneled through this method. The updates a converted
+     * All updated are are funneled through this method. The updates are converted
      * to a JournalMessage which is logged to the journal and then the data from
      * the JournalMessage is used to update the index just like it would be done
-     * durring a recovery process.
+     * during a recovery process.
      */
     public Location store(JournalCommand data, boolean sync) throws IOException {
-
-    	
-        int size = data.serializedSizeFramed();
-        DataByteArrayOutputStream os = new DataByteArrayOutputStream(size + 1);
-        os.writeByte(data.type().getNumber());
-        data.writeFramed(os);
-
-        long start = System.currentTimeMillis();
-        Location location = journal.write(os.toByteSequence(), sync);
-        long start2 = System.currentTimeMillis();
-        process(data, location);
-    	long end = System.currentTimeMillis();
-    	if( LOG_SLOW_ACCESS_TIME>0 && end-start > LOG_SLOW_ACCESS_TIME) {
-    		LOG.info("Slow KahaDB access: Journal append took: "+(start2-start)+" ms, Index update
took "+(end-start2)+" ms");
+    	try {
+            int size = data.serializedSizeFramed();
+            DataByteArrayOutputStream os = new DataByteArrayOutputStream(size + 1);
+            os.writeByte(data.type().getNumber());
+            data.writeFramed(os);
+    
+            long start = System.currentTimeMillis();
+            Location location = journal.write(os.toByteSequence(), sync);
+            long start2 = System.currentTimeMillis();
+            process(data, location);
+        	long end = System.currentTimeMillis();
+        	if( LOG_SLOW_ACCESS_TIME>0 && end-start > LOG_SLOW_ACCESS_TIME) {
+        		LOG.info("Slow KahaDB access: Journal append took: "+(start2-start)+" ms, Index
update took "+(end-start2)+" ms");
+        	}
+    
+            synchronized (indexMutex) {
+            	metadata.lastUpdate = location;
+            }
+            if (!checkpointThread.isAlive()) {
+                LOG.info("KahaDB: Recovering checkpoint thread after exception");
+                startCheckpoint();
+            }
+            return location;
+    	} catch (IOException ioe) {
+            LOG.error("KahaDB failed to store to Journal", ioe);
+            brokerService.handleIOException(ioe);
+    	    throw ioe;
     	}
-
-        synchronized (indexMutex) {
-        	metadata.lastUpdate = location;
-        }
-        return location;
     }
 
     /**
@@ -1530,4 +1547,8 @@
     public void setChecksumJournalFiles(boolean checksumJournalFiles) {
         this.checksumJournalFiles = checksumJournalFiles;
     }
+
+	public void setBrokerService(BrokerService brokerService) {
+		this.brokerService = brokerService;
+	}
 }

Copied: activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/util/DefaultIOExceptionHandler.java
(from r835888, activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/DefaultIOExceptionHandler.java)
URL: http://svn.apache.org/viewvc/activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/util/DefaultIOExceptionHandler.java?p2=activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/util/DefaultIOExceptionHandler.java&p1=activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/DefaultIOExceptionHandler.java&r1=835888&r2=882511&rev=882511&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/DefaultIOExceptionHandler.java
(original)
+++ activemq/branches/activemq-5.3/activemq-core/src/main/java/org/apache/activemq/util/DefaultIOExceptionHandler.java
Fri Nov 20 11:36:45 2009
@@ -28,12 +28,25 @@
             .getLog(DefaultIOExceptionHandler.class);
     private BrokerService broker;
     private boolean ignoreAllErrors = false;
+    private boolean ignoreNoSpaceErrors = true;
+    private String noSpaceMessage = "space";
 
     public void handle(IOException exception) {
         if (ignoreAllErrors) {
             LOG.info("Ignoring IO exception, " + exception, exception);
             return;
         }
+        
+        if (ignoreNoSpaceErrors) {
+            Throwable cause = exception;
+            while (cause != null && cause instanceof IOException) {
+                if (cause.getMessage().contains(noSpaceMessage)) {
+                    LOG.info("Ignoring no space left exception, " + exception, exception);
+                    return;
+                }
+                cause = cause.getCause();
+            }
+        }
 
         LOG.info("Stopping the broker due to IO exception, " + exception, exception);
         new Thread() {
@@ -59,4 +72,20 @@
         this.ignoreAllErrors = ignoreAllErrors;
     }
 
+    public boolean isIgnoreNoSpaceErrors() {
+        return ignoreNoSpaceErrors;
+    }
+
+    public void setIgnoreNoSpaceErrors(boolean ignoreNoSpaceErrors) {
+        this.ignoreNoSpaceErrors = ignoreNoSpaceErrors;
+    }
+
+    public String getNoSpaceMessage() {
+        return noSpaceMessage;
+    }
+
+    public void setNoSpaceMessage(String noSpaceMessage) {
+        this.noSpaceMessage = noSpaceMessage;
+    }
+
 }

Modified: activemq/branches/activemq-5.3/kahadb/src/main/java/org/apache/kahadb/journal/DataFileAppender.java
URL: http://svn.apache.org/viewvc/activemq/branches/activemq-5.3/kahadb/src/main/java/org/apache/kahadb/journal/DataFileAppender.java?rev=882511&r1=882510&r2=882511&view=diff
==============================================================================
--- activemq/branches/activemq-5.3/kahadb/src/main/java/org/apache/kahadb/journal/DataFileAppender.java
(original)
+++ activemq/branches/activemq-5.3/kahadb/src/main/java/org/apache/kahadb/journal/DataFileAppender.java
Fri Nov 20 11:36:45 2009
@@ -21,6 +21,7 @@
 import java.io.RandomAccessFile;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.zip.Adler32;
 import java.util.zip.Checksum;
 
@@ -85,6 +86,7 @@
         public final CountDownLatch latch = new CountDownLatch(1);
 		private final int offset;
         public int size = Journal.BATCH_CONTROL_RECORD_SIZE;
+        public AtomicReference<IOException> exception = new AtomicReference<IOException>();
 
         public WriteBatch(DataFile dataFile, int offset, WriteCommand write) throws IOException
{
             this.dataFile = dataFile;
@@ -158,7 +160,7 @@
      * @throws
      */
     public Location storeItem(ByteSequence data, byte type, boolean sync) throws IOException
{
-
+    	
         // Write the packet our internal buffer.
         int size = data.getLength() + Journal.RECORD_HEAD_SPACE;
 
@@ -184,6 +186,10 @@
             } catch (InterruptedException e) {
                 throw new InterruptedIOException();
             }
+            IOException exception = batch.exception.get(); 
+            if (exception != null) {
+            	throw exception;
+            }
         }	
 
         return location;
@@ -213,10 +219,7 @@
             if (shutdown) {
                 throw new IOException("Async Writter Thread Shutdown");
             }
-            if (firstAsyncException != null) {
-                throw firstAsyncException;
-            }
-
+            
             if (!running) {
                 running = true;
                 thread = new Thread() {
@@ -228,6 +231,11 @@
                 thread.setDaemon(true);
                 thread.setName("ActiveMQ Data File Writer");
                 thread.start();
+                firstAsyncException = null;
+            }
+            
+            if (firstAsyncException != null) {
+                throw firstAsyncException;
             }
 
             while ( true ) {
@@ -298,6 +306,7 @@
     protected void processQueue() {
         DataFile dataFile = null;
         RandomAccessFile file = null;
+        WriteBatch wb = null;
         try {
 
             DataByteArrayOutputStream buff = new DataByteArrayOutputStream(maxWriteBatchSize);
@@ -321,7 +330,7 @@
                     enqueueMutex.notify();
                 }
 
-                WriteBatch wb = (WriteBatch)o;
+                wb = (WriteBatch)o;
                 if (dataFile != wb.dataFile) {
                     if (file != null) {
                         file.setLength(dataFile.getLength());
@@ -405,6 +414,14 @@
         } catch (IOException e) {
             synchronized (enqueueMutex) {
                 firstAsyncException = e;
+                if (wb != null) {
+                    wb.latch.countDown();
+                    wb.exception.set(e);
+                }
+                if (nextWriteBatch != null) {
+            	    nextWriteBatch.latch.countDown();
+            	    nextWriteBatch.exception.set(e);
+                }
             }
         } catch (InterruptedException e) {
         } finally {
@@ -415,6 +432,7 @@
             } catch (Throwable ignore) {
             }
             shutdownDone.countDown();
+            running = false;
         }
     }
 

Modified: activemq/branches/activemq-5.3/kahadb/src/main/java/org/apache/kahadb/page/PageFile.java
URL: http://svn.apache.org/viewvc/activemq/branches/activemq-5.3/kahadb/src/main/java/org/apache/kahadb/page/PageFile.java?rev=882511&r1=882510&r2=882511&view=diff
==============================================================================
--- activemq/branches/activemq-5.3/kahadb/src/main/java/org/apache/kahadb/page/PageFile.java
(original)
+++ activemq/branches/activemq-5.3/kahadb/src/main/java/org/apache/kahadb/page/PageFile.java
Fri Nov 20 11:36:45 2009
@@ -44,6 +44,7 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kahadb.util.DataByteArrayOutputStream;
+import org.apache.kahadb.util.IOExceptionSupport;
 import org.apache.kahadb.util.IOHelper;
 import org.apache.kahadb.util.IntrospectionSupport;
 import org.apache.kahadb.util.LRUCache;
@@ -165,8 +166,8 @@
         }
         
         void begin() {
-            diskBound = current;
-            current = null;
+           diskBound = current;
+           current = null;
         }
         
         /**
@@ -176,6 +177,10 @@
             diskBound=null;
             return current == null;
         }
+        
+        boolean isDone() {
+            return diskBound == null && current == null;
+        }
 
     }
     
@@ -937,12 +942,15 @@
             // If there is not enough to write, wait for a notification...
 
             batch = new ArrayList<PageWrite>(writes.size());
-            // build a write batch from the current write cache. 
+            // build a write batch from the current write cache.
             for (PageWrite write : writes.values()) {
                 batch.add(write);
                 // Move the current write to the diskBound write, this lets folks update
the 
                 // page again without blocking for this write.
                 write.begin();
+                if (write.diskBound == null) {
+                    batch.remove(write);
+                }
             }
 
             // Grab on to the existing checkpoint latch cause once we do this write we can

@@ -951,71 +959,82 @@
             this.checkpointLatch=null;
         }
         
- 
-       if (enableRecoveryFile) {
-           
-           // Using Adler-32 instead of CRC-32 because it's much faster and it's 
-           // weakness for short messages with few hundred bytes is not a factor in this
case since we know 
-           // our write batches are going to much larger.
-           Checksum checksum = new Adler32();
-           for (PageWrite w : batch) {
-               checksum.update(w.diskBound, 0, pageSize);
-           }
-           
-           // Can we shrink the recovery buffer??
-           if( recoveryPageCount > recoveryFileMaxPageCount ) {
-               int t = Math.max(recoveryFileMinPageCount, batch.size());
-               recoveryFile.setLength(recoveryFileSizeForPages(t));
-           }
-           
-            // Record the page writes in the recovery buffer.
-            recoveryFile.seek(0);
-            // Store the next tx id...
-            recoveryFile.writeLong(nextTxid.get());
-            // Store the checksum for thw write batch so that on recovery we know if we have
a consistent 
-            // write batch on disk.
-            recoveryFile.writeLong(checksum.getValue());
-            // Write the # of pages that will follow
-            recoveryFile.writeInt(batch.size());
-            
-            
-            // Write the pages.
-            recoveryFile.seek(RECOVERY_FILE_HEADER_SIZE);
+       try {
+            if (enableRecoveryFile) {
+
+                // Using Adler-32 instead of CRC-32 because it's much faster and
+                // it's
+                // weakness for short messages with few hundred bytes is not a
+                // factor in this case since we know
+                // our write batches are going to much larger.
+                Checksum checksum = new Adler32();
+                for (PageWrite w : batch) {
+                    try {
+                        checksum.update(w.diskBound, 0, pageSize);
+                    } catch (Throwable t) {
+                        throw IOExceptionSupport.create(
+                                "Cannot create recovery file. Reason: " + t, t);
+                    }
+                }
+
+                // Can we shrink the recovery buffer??
+                if (recoveryPageCount > recoveryFileMaxPageCount) {
+                    int t = Math.max(recoveryFileMinPageCount, batch.size());
+                    recoveryFile.setLength(recoveryFileSizeForPages(t));
+                }
+
+                // Record the page writes in the recovery buffer.
+                recoveryFile.seek(0);
+                // Store the next tx id...
+                recoveryFile.writeLong(nextTxid.get());
+                // Store the checksum for thw write batch so that on recovery we
+                // know if we have a consistent
+                // write batch on disk.
+                recoveryFile.writeLong(checksum.getValue());
+                // Write the # of pages that will follow
+                recoveryFile.writeInt(batch.size());
+
+                // Write the pages.
+                recoveryFile.seek(RECOVERY_FILE_HEADER_SIZE);
+
+                for (PageWrite w : batch) {
+                    recoveryFile.writeLong(w.page.getPageId());
+                    recoveryFile.write(w.diskBound, 0, pageSize);
+                }
+
+                if (enableDiskSyncs) {
+                    // Sync to make sure recovery buffer writes land on disk..
+                    recoveryFile.getFD().sync();
+                }
+
+                recoveryPageCount = batch.size();
+            }
+
             for (PageWrite w : batch) {
-                recoveryFile.writeLong(w.page.getPageId());
-                recoveryFile.write(w.diskBound, 0, pageSize);
+                writeFile.seek(toOffset(w.page.getPageId()));
+                writeFile.write(w.diskBound, 0, pageSize);
+                w.done();
             }
-            
+
+            // Sync again
             if (enableDiskSyncs) {
-                // Sync to make sure recovery buffer writes land on disk..
-                recoveryFile.getFD().sync();
+                writeFile.getFD().sync();
             }
-            
-            recoveryPageCount = batch.size();
-        }
-       
-        
-        for (PageWrite w : batch) {
-            writeFile.seek(toOffset(w.page.getPageId()));
-            writeFile.write(w.diskBound, 0, pageSize);
-        }
-        
-        // Sync again
-        if( enableDiskSyncs ) {
-            writeFile.getFD().sync();
-        }
-        
-        synchronized( writes ) {
-            for (PageWrite w : batch) {
-                // If there are no more pending writes, then remove it from the write cache.
-                if( w.done() ) {
-                    writes.remove(w.page.getPageId());
+
+        } finally {
+            synchronized (writes) {
+                for (PageWrite w : batch) {
+                    // If there are no more pending writes, then remove it from
+                    // the write cache.
+                    if (w.isDone()) {
+                        writes.remove(w.page.getPageId());
+                    }
                 }
             }
-        }
-        
-        if( checkpointLatch!=null ) {
-            checkpointLatch.countDown();
+            
+            if( checkpointLatch!=null ) {
+                checkpointLatch.countDown();
+            }
         }
     }
 



Mime
View raw message