bigdata-commit Mailing List for Blazegraph (powered by bigdata) (Page 64)

SourceForge Headquarters 225 Broadway Suite 1600 San Diego, CA 92101 +1 (858) 422-6466

Revision: 7151
          http://bigdata.svn.sourceforge.net/bigdata/?rev=7151&view=rev
Author:   thompsonbry
Date:     2013-05-20 17:32:22 +0000 (Mon, 20 May 2013)
Log Message:
-----------
AbstractJournal: Added logic to make AbstractJournal.commit() spin for up to 5000ms if required to obtain a commitTime that moves forward with respect to the then current lastCommitTime.  This addresses a potential source of problems with clock skew in HA and could permit us to allow more skew to be present among the nodes of an HA cluster (though I have not yet relaxed that constraint in Journal.assertBefore()).

AbstractJournal: added getRootBlockWithLock() variant.

HAJournalServer.awaitJournalToken(): modified to use getRootBlockWithLock() to ensure atomicity of the state changes that awaitJournalToken() is monitoring.

Journal: More cleanup on the consensus release time protocol.  Changes improve reporting of the root cause of an exception in the release time consensus protocol and shift reporting of clock synchronization problems entirely to the leader (based on the timestamps obtain on the leader and the followers).

@see https://sourceforge.net/apps/trac/bigdata/ticket/530 (Journal HA)

Modified Paths:
--------------
    branches/READ_CACHE/bigdata/src/java/com/bigdata/journal/AbstractJournal.java
    branches/READ_CACHE/bigdata/src/java/com/bigdata/journal/Journal.java
    branches/READ_CACHE/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HAJournalServer.java

Modified: branches/READ_CACHE/bigdata/src/java/com/bigdata/journal/AbstractJournal.java
===================================================================

--- branches/READ_CACHE/bigdata/src/java/com/bigdata/journal/AbstractJournal.java	2013-05-20 17:13:24 UTC (rev 7150)
+++ branches/READ_CACHE/bigdata/src/java/com/bigdata/journal/AbstractJournal.java	2013-05-20 17:32:22 UTC (rev 7151)
@@ -168,6 +168,7 @@
 import com.bigdata.service.IBigdataFederation;
 import com.bigdata.stream.Stream;
 import com.bigdata.util.ChecksumUtility;
+import com.bigdata.util.ClocksNotSynchronizedException;
 import com.bigdata.util.NT;
 
 /**
@@ -2452,7 +2453,44 @@
 //		}
 
 	}
+	
+    /**
+     * Variant of {@link #getRootBlockView()} that takes the internal lock in
+     * order to provide an appropriate synchronization barrier when installing
+     * new root blocks onto an empty journal in HA.
+     * 
+     * @see #installRootBlocks(IRootBlockView, IRootBlockView)
+     */
+    final public IRootBlockView getRootBlockViewWithLock() {
 
+        final ReadLock lock = _fieldReadWriteLock.readLock();
+
+        lock.lock();
+
+        try {
+
+            if (_rootBlock == null) {
+
+                /*
+                 * This can happen before the journal file has been created.
+                 * Once it has been created the root block will always be
+                 * non-null when viewed while holding the lock.
+                 */
+
+                throw new IllegalStateException();
+
+            }
+
+            return _rootBlock;
+
+        } finally {
+
+            lock.unlock();
+
+        }
+
+    }
+
     final public long getLastCommitTime() {
 
 //		final ReadLock lock = _fieldReadWriteLock.readLock();
@@ -2805,17 +2843,75 @@
 
     }
 
+    /**
+     * Get timestamp that will be assigned to this commit point.
+     * <P>
+     * Note: This will spin until commit time advances over
+     * <code>lastCommitTime</code>, but not for more than N milliseconds. This
+     * will allow us to ensure that time moves forward when the leader fails
+     * over to another node with modest clock skew. If there is a large clock
+     * skew, the operator intervention will be required.
+     * <p>
+     * Note: This also makes sense for a non-HA deployment since we still want
+     * time to move forward at each commit point.
+     * 
+     * TODO This also makes sense when the Journal is opened since we often
+     * issue queries against historical commit points on the journal based on
+     * the clock. [Unit test for this in standalone and HA modes?]
+     */
+    private long nextCommitTimestamp() {
+        final IRootBlockView rootBlock = _rootBlock;
+        final long lastCommitTime = rootBlock.getLastCommitTime();
+        if (lastCommitTime < 0)
+            throw new RuntimeException(
+                    "Last commit time is invalid in rootBlock: " + rootBlock);
+        final long commitTime;
+        {
+            final ILocalTransactionManager transactionManager = getLocalTransactionManager();
+
+            boolean warned = false;
+            while (true) {
+                final long t = transactionManager.nextTimestamp();
+                if (t > lastCommitTime) {
+                    /*
+                     * We have a distinct timestamp. Time is moving forward.
+                     */
+                    commitTime = t;
+                    break;
+                }
+                /*
+                 * Time is going backwards.  Figure out by how much.
+                 * 
+                 * Note: delta is in ms.
+                 */
+                final long delta = Math.abs(t - lastCommitTime);
+                if (delta > 5000/* ms */)
+                    throw new ClocksNotSynchronizedException("Clocks off by "
+                            + delta + " ms: lastCommitTime=" + lastCommitTime
+                            + ", but localTimestamp=" + t);
+                if (!warned) {
+                    log.warn("Clocks off by " + delta + " ms: lastCommitTime="
+                            + lastCommitTime + ", but localTimestamp=" + t);
+                    warned = true;
+                }
+                try {
+                    // Wait for the delta to expire.
+                    Thread.sleep(delta/* ms */);
+                } catch (InterruptedException ex) {
+                    // Propagate interrupt.
+                    Thread.currentThread().interrupt();
+                }
+            }
+        }
+        return commitTime;
+    }
+    
     @Override
 	public long commit() {
 
-		final ILocalTransactionManager transactionManager = getLocalTransactionManager();
+		// The timestamp to be assigned to this commit point.
+		final long commitTime = nextCommitTimestamp();
 
-		/*
-		 * Get timestamp that will be assigned to this commit (RMI if the
-		 * journal is part of a distributed federation).
-		 */
-		final long commitTime = transactionManager.nextTimestamp();
-
         // do the commit.
         final IRootBlockView lastRootBlock = _rootBlock;
         final long commitTime2;
@@ -2845,7 +2941,7 @@
 		 * e.g., due to a service outage.
 		 */
 
-		transactionManager.notifyCommit(commitTime);
+        getLocalTransactionManager().notifyCommit(commitTime);
 
 		return commitTime;
 

Modified: branches/READ_CACHE/bigdata/src/java/com/bigdata/journal/Journal.java
===================================================================
--- branches/READ_CACHE/bigdata/src/java/com/bigdata/journal/Journal.java	2013-05-20 17:13:24 UTC (rev 7150)
+++ branches/READ_CACHE/bigdata/src/java/com/bigdata/journal/Journal.java	2013-05-20 17:32:22 UTC (rev 7151)
@@ -27,6 +27,7 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.Collection;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
@@ -359,6 +360,13 @@
          * This is the earliest visible commit point on the leader.
          */
         final private IHANotifyReleaseTimeRequest leadersValue;
+
+        /**
+         * Exception is set by {@link #run()} if there is a problem when the
+         * barrier breaks. The exception is then thrown out to the thread on
+         * the leader that is running commitNow(), forcing the commit to fail.
+         */
+        volatile Throwable cause = null;
         
         /**
          * The message from each of those followers providing their local
@@ -462,12 +470,20 @@
         @Override
         public void run() {
 
+            try {
+            
             if (log.isInfoEnabled())
                 log.info("leader: " + leadersValue);
             
             // This is the timestamp from the BarrierState ctor.
             final long timeLeader = leadersValue.getTimestamp();
             
+            // This is the timestamp for right now.
+            final long timeNow = newConsensusProtocolTimestamp();
+            
+            // The local clock must be moving forward.
+            assertBefore(timeLeader, timeNow);
+
             // Start with the leader's value (from ctor).
             minimumResponse = leadersValue;
 
@@ -485,10 +501,17 @@
 
                 /*
                  * Verify that the timestamp from the ctor is BEFORE the
-                 * timestamp assigned by the follower for its response.
+                 * timestamp assigned by the follower in the GatherTask.
                  */
                 assertBefore(timeLeader, response.getTimestamp());
 
+                /*
+                 * Verify that the timestamp from the GatherTask on the follower
+                 * is before the timestamp obtained at the top of this run()
+                 * method.
+                 */
+                assertBefore(response.getTimestamp(), timeNow);
+
             }
 
             // Restate the consensus as an appropriate message object.
@@ -499,6 +522,13 @@
             if (log.isInfoEnabled())
                 log.info("consensus: " + consensus);
 
+            } catch(Throwable t) {
+                
+                // Set the cause.
+                cause = t;
+                
+            }
+            
         }
 
         /**
@@ -553,26 +583,46 @@
 //            final Future<Void>[] remoteFutures = new Future[joinedServiceIds.length];
 //            final boolean[] remoteDone = new boolean[joinedServiceIds.length];
             
+            final List<Future<Void>> futures = new LinkedList<Future<Void>>();
+
             try {
 
                 final IHAGatherReleaseTimeRequest msg = new HAGatherReleaseTimeRequest(
                         token, timestampOnLeader);
-
+                
                 // Do not send message to self (leader is at index 0).
                 for (int i = 1; i < joinedServiceIds.length; i++) {
 
                     final UUID serviceId = joinedServiceIds[i];
 
                     /*
-                     * Runnable which will execute this message on the remote
-                     * service.
+                     * Message each follower.
+                     * 
+                     * Note: The invoked RMI method submits the GatherTask that
+                     * executes on the follower and returns. It does not block
+                     * waiting for the outcome of the task on the follower.
+                     * Instead, we wait until the barrier breaks. A thread will
+                     * monitor the quorum state and break the barrier if the
+                     * quorum breaks or if a joined service leaves during the
+                     * consensus protocol.
+                     * 
+                     * Note: This uses multiple threads to issue the requests in
+                     * parallel against the followers in order to minimize the
+                     * latency of the protocol.
                      */
-                    
-                    // Resolve joined service.
-                    final HATXSGlue service = getService(serviceId);
-                    
-                    // Message joined service (can throw NPE if service is gone).
-                    service.gatherMinimumVisibleCommitTime(msg);
+                    // Note: throws RejectedExecutionException if shutdown.
+                    futures.add(getExecutorService().submit(
+                            new Callable<Void>() {
+                                public Void call() throws Exception {
+                                    // Resolve joined service.
+                                    final HATXSGlue service = getService(serviceId);
+                                    // Message remote service. 
+                                    // Note: NPE if [service] is gone.
+                                    service.gatherMinimumVisibleCommitTime(msg);
+                                    // Done.
+                                    return null;
+                                }
+                            }));
 
 //                    // add to list of futures we will check.
 //                    remoteFutures[i] = rf;
@@ -741,21 +791,17 @@
 //                }
 
             } finally {
-//                /*
-//                 * Regardless of outcome or errors above, ensure that all remote
-//                 * futures are cancelled.
-//                 */
-//                for (int i = 0; i < remoteFutures.length; i++) {
-//                    final Future<Void> rf = remoteFutures[i];
-//                    if (!remoteDone[i]) {
-//                        // Cancel the request on the remote service (RMI).
-//                        try {
-//                            rf.cancel(true/* mayInterruptIfRunning */);
-//                        } catch (Throwable t) {
-//                            // ignored.
-//                        }
-//                    }
-//                }
+                /*
+                 * Cancel local futures for RMI messages to followers.
+                 * 
+                 * Note: Regardless of outcome or errors above, ensure that the
+                 * futures used to initiate the GatherTask on the followers are
+                 * cancelled. These are local Futures that do RMIs. The RMIs
+                 * should not block when the execute on the follower.
+                 */
+                for (Future<Void> f : futures) {
+                    f.cancel(true/* mayInterruptIfRunning */);
+                }
                 if (!barrier.isBroken()) {
                     /*
                      * If there were any followers that did not message the
@@ -773,10 +819,8 @@
                      * "vote yes". Thus, a single node failure during the
                      * release time consensus protocol will cause the commit to
                      * fail. [Actually, we could use getNumberWaiting(). If it
-                     * is a bare majority, then we could force the barrier to
-                     * meet break (either with reset or with running an await()
-                     * in other threads) and take the barrier break action
-                     * ourselves. E.g., in the thread that calls
+                     * is a bare majority, then we could take the barrier break
+                     * action ourselves. E.g., in the thread that calls
                      * barrier.reset()].
                      */
                     barrier.reset();
@@ -933,11 +977,14 @@
      * Assert that t1 LT t2.
      * 
      * @param t1
+     *            A timestamp from one service.
      * @param t2
+     *            A timestamp from another service.
      * 
      * @throws ClocksNotSynchronizedException
      */
-    private void assertBefore(final long t1, final long t2) {
+    private void assertBefore(final long t1, final long t2)
+            throws ClocksNotSynchronizedException {
 
         if (t1 < t2)
             return;
@@ -945,8 +992,20 @@
         throw new ClocksNotSynchronizedException();
 
     }
-    
+
     /**
+     * Note: This deliberately uses the (non-remote) method
+     * {@link BasicHA#nextTimestamp()}. This is done so we can write a unit test
+     * of the {@link GatherTask} that imposes clock skew by overridding the next
+     * value to be returned by that method.
+     */
+    private long newConsensusProtocolTimestamp() {
+
+        return ((BasicHA) getQuorum().getClient().getService()).nextTimestamp();
+
+    }
+
+    /**
      * {@inheritDoc}
      * <p>
      * Extends the {@link JournalTransactionService} to provide protection for
@@ -1112,6 +1171,17 @@
 
                 }
 
+                if (barrierState.cause != null) {
+
+                    /*
+                     * If an exception was recorded, re-throw it in the thread
+                     * that invoked commitNow().
+                     */
+
+                    throw new RuntimeException(barrierState.cause);
+
+                }
+
                 /**
                  * Update the release time on the leader.
                  * 
@@ -1286,7 +1356,8 @@
             final long commitTime = commitRecord == null ? 0
                     : commitRecord.getTimestamp();
 
-            final long now = getLocalTransactionManager().nextTimestamp();
+//            final long now = getLocalTransactionManager().nextTimestamp();
+            final long now = newConsensusProtocolTimestamp();
 
             final IHANotifyReleaseTimeRequest req = new HANotifyReleaseTimeRequest(
                     serviceId, commitTime, commitCounter, now);
@@ -1407,14 +1478,8 @@
 
                     /*
                      * This timestamp is used to help detect clock skew.
-                     * 
-                     * Note: This deliberately uses the (non-remote)
-                     * nextTimestamp() method on BasicHA. This is being done so
-                     * we can write a unit test of the GatherTask that imposes
-                     * clock skew by overridding the next value to be returned
-                     * by that method.
                      */
-                    now = ((BasicHA)quorumService.getService()).nextTimestamp();
+                    now = newConsensusProtocolTimestamp();
 
                     /*
                      * If the token is invalid, making it impossible for us to
@@ -1428,10 +1493,16 @@
                      * our response. If we hit an assertion, we will still
                      * respond in the finally {} block below.
                      */
-                    
-                    /* Verify event on leader occurs before event on follower.
+
+                    /*
+                     * Note: This assert has been moved to the leader when it
+                     * analyzes the messages from the followers. This allows us
+                     * to report out the nature of the exception on the leader
+                     * and thence back to the client.
                      */
-                    assertBefore(req.getTimestampOnLeader(), now);
+//                    /* Verify event on leader occurs before event on follower.
+//                     */
+//                    assertBefore(req.getTimestampOnLeader(), now);
 
                     if (!quorumService.isFollower(token))
                         throw new QuorumException();
@@ -1601,17 +1672,29 @@
                 // Make a note of the message from this follower.
                 barrierState.responses.put(followerId, req);
                 
+            } catch(RuntimeException e) {
+                
+                /*
+                 * Note: The try {} block can throw RuntimeException but not
+                 * Exception. If anything is thrown, then reset the barrier and
+                 * rethrow the exception.
+                 */
+
+                haLog.error(e, e);
+                
+                // Reset the barrier (barrier will break).
+                barrierState.barrier.reset();
+                
+                // Rethrow the exception.
+                throw new RuntimeException(e);
+                
             } finally {
                 
                 /*
                  * Block until barrier breaks.
                  * 
-                 * Note: We want to await() on the barrier even if there is an
-                 * error in the try{} block. This is necessary to decrement the
-                 * barrier count down to zero.
-                 * 
-                 * TODO If there is an error, we could reset() the barrier
-                 * instead.
+                 * Note: The barrier will break immediately if it was reset in
+                 * the catch{} block above.
                  */
 
                 // follower blocks on Thread on the leader here.
@@ -1619,6 +1702,23 @@
                 
             }
 
+            /*
+             * Check for an error in the consensus protocol.
+             */
+            final Throwable t = barrierState.cause;
+
+            if (t != null) {
+
+                /*
+                 * Log error.
+                 */
+                haLog.error(t, t);
+
+                // rethrow cause.
+                throw new RuntimeException(t);
+
+            }
+
             // Return the consensus.
             final IHANotifyReleaseTimeResponse resp = barrierState.consensus;
 
@@ -1630,7 +1730,7 @@
             }
 
             return resp;
-  
+
         }
 
         /**

Modified: branches/READ_CACHE/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HAJournalServer.java
===================================================================
--- branches/READ_CACHE/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HAJournalServer.java	2013-05-20 17:13:24 UTC (rev 7150)
+++ branches/READ_CACHE/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HAJournalServer.java	2013-05-20 17:32:22 UTC (rev 7151)
@@ -1566,6 +1566,10 @@
                  * yet.
                  */
 //                server.haGlueService.bounceZookeeperConnection();
+                /*
+                 * Note: Try moving to doRejectedCommit() so this will be
+                 * synchronous.
+                 */
                 logLock.lock();
                 try {
                     if (journal.getHALogNexus().isHALogOpen()) {
@@ -3525,7 +3529,8 @@
                     continue;
                 }
                 if (isFollower(token)) {// if (awaitRootBlocks) { 
-                    final IRootBlockView rbSelf = journal.getRootBlockView();
+                    // Check root block, using lock for synchronization barrier.
+                    final IRootBlockView rbSelf = journal.getRootBlockViewWithLock();
                     if (rbSelf.getCommitCounter() == 0L) {
                         /*
                          * Only wait if this is an empty Journal.

This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.





2010	Jan	Feb	Mar	Apr	May	Jun	Jul (139)	Aug (94)	Sep (232)	Oct (143)	Nov (138)	Dec (55)
2011	Jan (127)	Feb (90)	Mar (101)	Apr (74)	May (148)	Jun (241)	Jul (169)	Aug (121)	Sep (157)	Oct (199)	Nov (281)	Dec (75)
2012	Jan (107)	Feb (122)	Mar (184)	Apr (73)	May (14)	Jun (49)	Jul (26)	Aug (103)	Sep (133)	Oct (61)	Nov (51)	Dec (55)
2013	Jan (59)	Feb (72)	Mar (99)	Apr (62)	May (92)	Jun (19)	Jul (31)	Aug (138)	Sep (47)	Oct (83)	Nov (95)	Dec (111)
2014	Jan (125)	Feb (60)	Mar (119)	Apr (136)	May (270)	Jun (83)	Jul (88)	Aug (30)	Sep (47)	Oct (27)	Nov (23)	Dec
2015	Jan	Feb	Mar	Apr	May	Jun	Jul	Aug	Sep (3)	Oct	Nov	Dec
2016	Jan	Feb	Mar (4)	Apr (1)	May	Jun	Jul	Aug	Sep	Oct	Nov	Dec

bigdata-commit Mailing List for Blazegraph (powered by bigdata) (Page 64)

Fast, scalable, robust graph database platform

bigdata-commit — Commit traffic.