Thread: [Bigdata-commit] SF.net SVN: bigdata:[7007] branches/READ_CACHE (Page 2)

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

Revision: 7007
          http://bigdata.svn.sourceforge.net/bigdata/?rev=7007&view=rev
Author:   thompsonbry
Date:     2013-03-18 15:03:28 +0000 (Mon, 18 Mar 2013)
Log Message:
-----------
Added awaitHAReady() methods on AbstractJournal - this is not yet  integrated into HAJournalServer. This is exposed through HAGlue.awaitHAReady().

Added status?HA to obtain an HTTP status message that is HA specific.

These additions are not yet tested.

Modified Paths:
--------------
    branches/READ_CACHE/bigdata/src/java/com/bigdata/ha/HAGlue.java
    branches/READ_CACHE/bigdata/src/java/com/bigdata/ha/HAGlueDelegate.java
    branches/READ_CACHE/bigdata/src/java/com/bigdata/journal/AbstractJournal.java
    branches/READ_CACHE/bigdata-sails/src/java/com/bigdata/rdf/sail/webapp/HAStatusServletUtil.java
    branches/READ_CACHE/bigdata-sails/src/java/com/bigdata/rdf/sail/webapp/StatusServlet.java

Modified: branches/READ_CACHE/bigdata/src/java/com/bigdata/ha/HAGlue.java
===================================================================

--- branches/READ_CACHE/bigdata/src/java/com/bigdata/ha/HAGlue.java	2013-03-18 13:40:26 UTC (rev 7006)
+++ branches/READ_CACHE/bigdata/src/java/com/bigdata/ha/HAGlue.java	2013-03-18 15:03:28 UTC (rev 7007)
@@ -29,6 +29,7 @@
 import java.security.DigestException;
 import java.security.NoSuchAlgorithmException;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 import com.bigdata.ha.msg.IHADigestRequest;
@@ -41,6 +42,8 @@
 import com.bigdata.journal.AbstractJournal;
 import com.bigdata.journal.ITransactionService;
 import com.bigdata.journal.Journal;
+import com.bigdata.quorum.AsynchronousQuorumCloseException;
+import com.bigdata.quorum.QuorumException;
 import com.bigdata.service.IService;
 
 /**
@@ -89,6 +92,29 @@
      */
     public Future<Void> enterErrorState() throws IOException;
     
+    /**
+     * Await the service being ready to partitipate in an HA quorum. The
+     * preconditions include:
+     * <ol>
+     * <li>receiving notice of the quorum token via
+     * {@link #setQuorumToken(long)}</li>
+     * <li>The service is joined with the met quorum for that token</li>
+     * <li>If the service is a follower and it's local root blocks were at
+     * <code>commitCounter:=0</code>, then the root blocks from the leader have
+     * been installed on the follower.</li>
+     * <ol>
+     * 
+     * @param timeout
+     *            The timeout to await this condition.
+     * @param units
+     *            The units for that timeout.
+     *            
+     * @return the quorum token for which the service became HA ready.
+     */
+    public long awaitHAReady(long timeout, TimeUnit unit) throws IOException,
+            InterruptedException, TimeoutException, QuorumException,
+            AsynchronousQuorumCloseException;
+
     /*
      * Synchronization.
      * 

Modified: branches/READ_CACHE/bigdata/src/java/com/bigdata/ha/HAGlueDelegate.java
===================================================================
--- branches/READ_CACHE/bigdata/src/java/com/bigdata/ha/HAGlueDelegate.java	2013-03-18 13:40:26 UTC (rev 7006)
+++ branches/READ_CACHE/bigdata/src/java/com/bigdata/ha/HAGlueDelegate.java	2013-03-18 15:03:28 UTC (rev 7007)
@@ -30,10 +30,9 @@
 import java.security.NoSuchAlgorithmException;
 import java.util.UUID;
 import java.util.concurrent.Future;
-import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
-import com.bigdata.concurrent.FutureTaskMon;
 import com.bigdata.ha.msg.IHA2PhaseAbortMessage;
 import com.bigdata.ha.msg.IHA2PhaseCommitMessage;
 import com.bigdata.ha.msg.IHA2PhasePrepareMessage;
@@ -56,6 +55,8 @@
 import com.bigdata.ha.msg.IHAWriteSetStateRequest;
 import com.bigdata.ha.msg.IHAWriteSetStateResponse;
 import com.bigdata.journal.ValidationError;
+import com.bigdata.quorum.AsynchronousQuorumCloseException;
+import com.bigdata.quorum.QuorumException;
 
 /**
  * Delegation pattern.
@@ -247,4 +248,11 @@
         return delegate.getHAWriteSetState(req);
     }
 
+    @Override
+    public long awaitHAReady(long timeout, TimeUnit unit) throws IOException,
+            InterruptedException, QuorumException,
+            AsynchronousQuorumCloseException, TimeoutException {
+        return delegate.awaitHAReady(timeout, unit);
+    }
+
 }

Modified: branches/READ_CACHE/bigdata/src/java/com/bigdata/journal/AbstractJournal.java
===================================================================
--- branches/READ_CACHE/bigdata/src/java/com/bigdata/journal/AbstractJournal.java	2013-03-18 13:40:26 UTC (rev 7006)
+++ branches/READ_CACHE/bigdata/src/java/com/bigdata/journal/AbstractJournal.java	2013-03-18 15:03:28 UTC (rev 7007)
@@ -58,6 +58,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
@@ -130,8 +131,10 @@
 import com.bigdata.journal.jini.ha.HAJournal;
 import com.bigdata.mdi.IResourceMetadata;
 import com.bigdata.mdi.JournalMetadata;
+import com.bigdata.quorum.AsynchronousQuorumCloseException;
 import com.bigdata.quorum.Quorum;
 import com.bigdata.quorum.QuorumActor;
+import com.bigdata.quorum.QuorumException;
 import com.bigdata.quorum.QuorumMember;
 import com.bigdata.rawstore.IAllocationContext;
 import com.bigdata.rawstore.IAllocationManagerStore;
@@ -4930,6 +4933,11 @@
                  * we have to wait until we observe that to cast a new vote.
                  */
                 
+                haReadyToken = Quorum.NO_QUORUM;
+
+                // signal HAReady condition.
+                haReadyCondition.signalAll();
+                
             } else if (didMeet) {
 
                 quorumToken = newValue;
@@ -4938,46 +4946,61 @@
                 final QuorumService<HAGlue> localService = quorum.getClient();
 
                 boolean installedRBs = false;
-                
-                if (_rootBlock.getCommitCounter() == 0L
-                        && localService.isFollower(quorumToken)) {
 
-                    /*
-                     * Take the root blocks from the quorum leader and use them.
-                     */
-                    
-                    // Remote interface for the quorum leader.
-                    final HAGlue leader = localService.getLeader(quorumToken);
+                if (localService.isFollower(quorumToken)) {
 
-                    log.info("Fetching root block from leader.");
-                    final IRootBlockView leaderRB;
-                    try {
-                        leaderRB = leader.getRootBlock(
-                                new HARootBlockRequest(null/* storeUUID */))
-                                .getRootBlock();
-                    } catch (IOException e) {
-                        throw new RuntimeException(e);
-                    }
-
-                    if (leaderRB.getCommitCounter() == 0L) {
-
+                    if (_rootBlock.getCommitCounter() == 0L) {
+                        
                         /*
-                         * Installs the root blocks and does a local abort.
-                         * 
-                         * Note: This code path is only taken when both the
-                         * leader and the follower are at commitCounter==0L.
-                         * This prevents us from accidentally laying down on a
-                         * follower the root blocks corresponding to a leader
-                         * that already has committed write sets.
+                         * Take the root blocks from the quorum leader and use
+                         * them.
                          */
-                        localService.installRootBlocks(
-                                leaderRB.asRootBlock(true/* rootBlock0 */),
-                                leaderRB.asRootBlock(false/* rootBlock0 */));
 
-                        installedRBs = true;
+                        // Remote interface for the quorum leader.
+                        final HAGlue leader = localService
+                                .getLeader(quorumToken);
 
+                        log.info("Fetching root block from leader.");
+                        final IRootBlockView leaderRB;
+                        try {
+                            leaderRB = leader
+                                    .getRootBlock(
+                                            new HARootBlockRequest(null/* storeUUID */))
+                                    .getRootBlock();
+                        } catch (IOException e) {
+                            throw new RuntimeException(e);
+                        }
+
+                        if (leaderRB.getCommitCounter() == 0L) {
+
+                            /*
+                             * Installs the root blocks and does a local abort.
+                             * 
+                             * Note: This code path is only taken when both the
+                             * leader and the follower are at commitCounter==0L.
+                             * This prevents us from accidentally laying down on
+                             * a follower the root blocks corresponding to a
+                             * leader that already has committed write sets.
+                             */
+                            localService
+                                    .installRootBlocks(
+                                            leaderRB.asRootBlock(true/* rootBlock0 */),
+                                            leaderRB.asRootBlock(false/* rootBlock0 */));
+
+                            installedRBs = true;
+
+                        }
+
                     }
 
+                    // ready as follower.
+                    haReadyToken = newValue;
+
+                } else if (localService.isLeader(quorumToken)) {
+
+                    // ready as leader.
+                    haReadyToken = newValue;
+
                 }
 
                 if (!installedRBs) {
@@ -4995,6 +5018,9 @@
 
                 }
                 
+                // signal HAReady condition.
+                haReadyCondition.signalAll();
+                
             } else {
 
                 throw new AssertionError();
@@ -5008,8 +5034,98 @@
         }
 
     }
+    private final Condition haReadyCondition = _fieldReadWriteLock.writeLock().newCondition();
+    private volatile long haReadyToken = Quorum.NO_QUORUM;
+    
+    /**
+     * Await the service being ready to partitipate in an HA quorum. The
+     * preconditions include:
+     * <ol>
+     * <li>receiving notice of the quorum token via
+     * {@link #setQuorumToken(long)}</li>
+     * <li>The service is joined with the met quorum for that token</li>
+     * <li>If the service is a follower and it's local root blocks were at
+     * <code>commitCounter:=0</code>, then the root blocks from the leader have
+     * been installed on the follower.</li>
+     * <ol>
+     * 
+     * @return the quorum token for which the service became HA ready.
+     */
+    final public long awaitHAReady() throws InterruptedException,
+            AsynchronousQuorumCloseException, QuorumException {
+        final WriteLock lock = _fieldReadWriteLock.writeLock();
+        lock.lock();
+        try {
+            long t = Quorum.NO_QUORUM;
+            while (((t = haReadyToken) != Quorum.NO_QUORUM)
+                    && getQuorum().getClient() != null) {
+                haReadyCondition.await();
+            }
+            final QuorumService<?> client = getQuorum().getClient();
+            if (client == null)
+                throw new AsynchronousQuorumCloseException();
+           if (!client.isJoinedMember(t)) {
+                throw new QuorumException();
+            }
+            return t;
+        } finally {
+            lock.unlock();
+        }
+    }
 
     /**
+     * Await the service being ready to partitipate in an HA quorum. The
+     * preconditions include:
+     * <ol>
+     * <li>receiving notice of the quorum token via
+     * {@link #setQuorumToken(long)}</li>
+     * <li>The service is joined with the met quorum for that token</li>
+     * <li>If the service is a follower and it's local root blocks were at
+     * <code>commitCounter:=0</code>, then the root blocks from the leader have
+     * been installed on the follower.</li>
+     * <ol>
+     * 
+     * @param timeout
+     *            The timeout to await this condition.
+     * @param units
+     *            The units for that timeout.
+     *            
+     * @return the quorum token for which the service became HA ready.
+     */
+    final public long awaitHAReady(final long timeout, final TimeUnit units)
+            throws InterruptedException, TimeoutException,
+            AsynchronousQuorumCloseException {
+        final WriteLock lock = _fieldReadWriteLock.writeLock();
+        final long begin = System.nanoTime();
+        final long nanos = units.toNanos(timeout);
+        long remaining = nanos;
+        if (!lock.tryLock(remaining, TimeUnit.NANOSECONDS))
+            throw new TimeoutException();
+        try {
+            // remaining = nanos - (now - begin) [aka elapsed]
+            remaining = nanos - (System.nanoTime() - begin);
+            long t = Quorum.NO_QUORUM;
+            while (((t = haReadyToken) != Quorum.NO_QUORUM)
+                    && getQuorum().getClient() != null && remaining > 0) {
+                if (!haReadyCondition.await(remaining, TimeUnit.NANOSECONDS))
+                    throw new TimeoutException();
+                remaining = nanos - (System.nanoTime() - begin);
+            }
+            final QuorumService<?> client = getQuorum().getClient();
+            if (client == null)
+                throw new AsynchronousQuorumCloseException();
+            if (remaining <= 0)
+                throw new TimeoutException();
+            if (!client.isJoinedMember(t)) {
+                throw new QuorumException();
+            }
+            return t;
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    /**
      * Install identical root blocks on the journal. This is used for a few
      * different conditions in HA.
      * <ol>
@@ -5352,6 +5468,15 @@
         }
 
         @Override
+        public long awaitHAReady(final long timeout, final TimeUnit units)
+                throws AsynchronousQuorumCloseException, InterruptedException,
+                TimeoutException {
+
+            return AbstractJournal.this.awaitHAReady(timeout, units);
+
+        }
+
+        @Override
         public IHADigestResponse computeDigest(final IHADigestRequest req)
                 throws IOException, NoSuchAlgorithmException, DigestException {
 

Modified: branches/READ_CACHE/bigdata-sails/src/java/com/bigdata/rdf/sail/webapp/HAStatusServletUtil.java
===================================================================
--- branches/READ_CACHE/bigdata-sails/src/java/com/bigdata/rdf/sail/webapp/HAStatusServletUtil.java	2013-03-18 13:40:26 UTC (rev 7006)
+++ branches/READ_CACHE/bigdata-sails/src/java/com/bigdata/rdf/sail/webapp/HAStatusServletUtil.java	2013-03-18 15:03:28 UTC (rev 7007)
@@ -27,6 +27,8 @@
 import java.io.IOException;
 import java.io.PrintWriter;
 import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
@@ -39,6 +41,8 @@
 import com.bigdata.journal.IIndexManager;
 import com.bigdata.journal.jini.ha.HAJournal;
 import com.bigdata.journal.jini.ha.SnapshotManager;
+import com.bigdata.quorum.AsynchronousQuorumCloseException;
+import com.bigdata.quorum.Quorum;
 import com.bigdata.quorum.zk.ZKQuorumImpl;
 import com.bigdata.zookeeper.DumpZookeeper;
 
@@ -378,5 +382,71 @@
         return -1;
 
     }
-    
+
+    /**
+     * Special reporting request for HA status.
+     * 
+     * @param req
+     * @param resp
+     * @throws TimeoutException
+     * @throws InterruptedException
+     * @throws AsynchronousQuorumCloseException
+     * @throws IOException
+     */
+    public void doHAStatus(HttpServletRequest req, HttpServletResponse resp)
+            throws IOException {
+
+        if (!(indexManager instanceof HAJournal))
+            return;
+
+        final HAJournal journal = (HAJournal) indexManager;
+
+        final ZKQuorumImpl<HAGlue, QuorumService<HAGlue>> quorum = (ZKQuorumImpl<HAGlue, QuorumService<HAGlue>>) journal
+                .getQuorum();
+
+//        // The current token.
+//        final long quorumToken = quorum.token();
+//
+//        // The last valid token.
+//        final long lastValidToken = quorum.lastValidToken();
+//
+//        final int njoined = quorum.getJoined().length;
+
+        final QuorumService<HAGlue> quorumService = quorum.getClient();
+
+        // check, but do not wait.
+        long haReadyToken = Quorum.NO_QUORUM;
+        try {
+            haReadyToken = journal.awaitHAReady(0/* timeout */,
+                    TimeUnit.NANOSECONDS/* units */);
+        } catch (TimeoutException ex) {
+            // ignore.
+        } catch (AsynchronousQuorumCloseException e) {
+            // ignore.
+        } catch (InterruptedException e) {
+            // propagate the interrupt.
+            Thread.currentThread().interrupt();
+        }
+        
+        final String content;
+        if (haReadyToken != Quorum.NO_QUORUM) {
+            if (quorumService.isLeader(haReadyToken)) {
+                content = "Leader";
+            } else if (quorumService.isFollower(haReadyToken)) {
+                content = "Follower";
+            } else {
+                content = "NotReady";
+            }
+        } else {
+            content = "NotReady";
+        }
+        // TODO Alternatively "max-age=1" for max-age in seconds.
+        resp.addHeader("Cache-Control", "no-cache");
+        BigdataRDFServlet.buildResponse(resp, BigdataRDFServlet.HTTP_OK,
+                BigdataRDFServlet.MIME_TEXT_PLAIN, content);
+
+        return;
+
+    }
+
 }

Modified: branches/READ_CACHE/bigdata-sails/src/java/com/bigdata/rdf/sail/webapp/StatusServlet.java
===================================================================
--- branches/READ_CACHE/bigdata-sails/src/java/com/bigdata/rdf/sail/webapp/StatusServlet.java	2013-03-18 13:40:26 UTC (rev 7006)
+++ branches/READ_CACHE/bigdata-sails/src/java/com/bigdata/rdf/sail/webapp/StatusServlet.java	2013-03-18 15:03:28 UTC (rev 7007)
@@ -154,6 +154,13 @@
     static final String SNAPSHOT = "snapshot";
 
     /**
+     * Special HA status request designed for clients that poll to determine the
+     * status of an HAJournalServer. This option is exclusive of other
+     * parameters.
+     */
+    static final String HA = "HA";
+
+    /**
      * Handles CANCEL requests (terminate a running query).
      */
     @Override
@@ -286,6 +293,15 @@
     protected void doGet(final HttpServletRequest req,
             final HttpServletResponse resp) throws IOException {
 
+        if (req.getParameter(HA) != null
+                && getIndexManager() instanceof AbstractJournal
+                && ((AbstractJournal) getIndexManager()).isHighlyAvailable()) {
+
+            new HAStatusServletUtil(getIndexManager()).doHAStatus(req, resp);
+
+            return;
+        }
+        
         // IRunningQuery objects currently running on the query controller.
         final boolean showQueries = req.getParameter(SHOW_QUERIES) != null;
 

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





Thread: [Bigdata-commit] SF.net SVN: bigdata:[7007] branches/READ_CACHE (Page 2)

Fast, scalable, robust graph database platform

bigdata-commit