Thread: [Bigdata-commit] SF.net SVN: bigdata:[7392] branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src

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

Revision: 7392
          http://bigdata.svn.sourceforge.net/bigdata/?rev=7392&view=rev
Author:   martyncutcher
Date:     2013-09-10 13:55:55 +0000 (Tue, 10 Sep 2013)
Log Message:
-----------
Change DumpLogDigests to provide entries for logs that are not found to simplify comparisons between services with different available log files.
And register protection with the HALogNexus to avoid file removal while digests are being computed.

Modified Paths:
--------------
    branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/DumpLogDigests.java
    branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HALogNexus.java
    branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3DumpLogs.java

Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/DumpLogDigests.java
===================================================================

--- branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/DumpLogDigests.java	2013-09-10 13:50:17 UTC (rev 7391)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/DumpLogDigests.java	2013-09-10 13:55:55 UTC (rev 7392)
@@ -52,6 +52,7 @@
 import com.bigdata.journal.ITransactionService;
 import com.bigdata.journal.jini.ha.HAClient.HAConnection;
 import com.bigdata.journal.jini.ha.HALogIndex.IHALogRecord;
+import com.bigdata.journal.jini.ha.SnapshotIndex.ISnapshotRecord;
 import com.bigdata.quorum.Quorum;
 import com.bigdata.quorum.QuorumClient;
 
@@ -61,8 +62,6 @@
  * Accesses Zookeeper state and then connects to each service member to
  * retrieve log digest information.
  * 
- * The states are pinned by starting a read transaction on any of the services.
- * 
  * This task will return an object with the transaction reference to be released
  * and the range of commit counters for the logs to be checked.
  * 
@@ -135,7 +134,7 @@
 		    final LogDigestParams params = pinner.submit(new PinLogs(), false).get();
 		    
 		    if (log.isInfoEnabled())
-		    	log.info("Pinning startCC: " + params.startCC + ", endCC: " + params.endCC);
+		    	log.info("Pinning startCC: " + params.startCC + ", endCC: " + params.endCC + ", last snapshot: " + params.snapshotCC);
 		    
 		    /**
 		     *  Now access serviceIDs so that we can use discovery to gain HAGlue interface.
@@ -217,18 +216,26 @@
      * LogDigestParams with PinLogs and UnpinLogs tasks ensure that
      * transactions (and logs?) will not be removed while the digests
      * are computed.
-     * TODO: Check if this will really pin the logs?
+     * <p>
+     * In fact creating a transaction does not protect the halog files,
+     * which are now protected from removal during a digest computation
+     * by a protectDigest call on the HALogNexus in GetLogInfo.
+     * <p>
+     * TODO: The read transaction is currently retained but should be removed
+     * if/when it becomes clear that it has no use.
      */
     @SuppressWarnings("serial")
 	static public class LogDigestParams implements Serializable {
     	final public long tx;
     	final public long startCC;
     	final public long endCC;
+    	final public long snapshotCC;
     	
-    	LogDigestParams(final long tx, final long startCC, final long endCC) {
+    	LogDigestParams(final long tx, final long startCC, final long endCC, long sscc) {
     		this.tx = tx;
     		this.startCC = startCC;
     		this.endCC = endCC;
+    		this.snapshotCC = sscc;
     	}
     }
     
@@ -251,9 +258,14 @@
 				startCC = logs.next().getCommitCounter();
 			} else {
 				startCC = endCC;
-			}			
+			}
 			
-			return new LogDigestParams(tx, startCC, endCC);
+			final SnapshotManager ssmgr = ha.getSnapshotManager();
+			final ISnapshotRecord rec = ssmgr.getNewestSnapshot();
+			final long sscc = rec != null ? rec.getCommitCounter() : -1;
+			
+			// return new LogDigestParams(tx, startCC-3, endCC+3); // try asking for more logs than available
+			 return new LogDigestParams(tx, startCC, endCC, sscc);
 		}
     	
     }
@@ -283,7 +295,7 @@
      * The GetLogInfo callable is submitted to each service, retrieving a
      * List of HALogInfo data elements for each commit counter log
      * requested.
-     * 
+     * <p>
      * It is parameterized for start and end commit counters for the logs and
      * the number of serviceThreads to split the digest computations across.
      */
@@ -310,6 +322,8 @@
 			HAJournal ha = (HAJournal) this.getIndexManager();
 						
 			final HALogNexus nexus = ha.getHALogNexus();
+			nexus.protectDigest(startCC);
+			try {
 			long openCC = nexus.getCommitCounter();
 			log.warn("Open Commit Counter: " + openCC + ", startCC: " + startCC + ", endCC: " + endCC);
 			
@@ -340,11 +354,15 @@
 	
 				            r.computeDigest(digest);
 				            
-							infos.add(new HALogInfo(file.getName(), r.isLive(), digest.digest()));
+							infos.add(new HALogInfo(cur, r.isLive(), digest.digest()));
 						} catch (FileNotFoundException fnf) {
-							// half expected
+							// permitted
+							infos.add(new HALogInfo(cur, false, null /*digest*/));
 						} catch (Throwable t) {
 							log.warn("Unexpected error", t);
+							
+							// FIXME: what to do here?
+							infos.add(new HALogInfo(cur, false, "ERROR".getBytes()));
 						}
 						
 						return null;
@@ -362,26 +380,40 @@
 	        es.shutdown();
 			
 			return new ArrayList<HALogInfo>(infos);
+			} finally {
+				nexus.releaseProtectDigest();
+			}
 		}
     	
     }
     
     @SuppressWarnings("serial")
+    /**
+     * If the log does not exist then a null digest is defined.
+     */
 	static public class HALogInfo implements Serializable, Comparable<HALogInfo> {
-    	final public String logName;
+    	final public long commitCounter;
     	final public boolean isOpen;
     	final public byte[] digest;
     	
-    	HALogInfo(final String logName, final boolean isOpen, final byte[] bs) {
-    		this.logName = logName;
+    	HALogInfo(final long commitCounter, final boolean isOpen, final byte[] bs) {
+    		this.commitCounter = commitCounter;
     		this.isOpen = isOpen;
     		this.digest = bs;
     	}
 
 		@Override
 		public int compareTo(HALogInfo other) {
-			return logName.compareTo(other.logName);
+			
+			// should not be same commit counter!
+			assert commitCounter != other.commitCounter;
+			
+			return commitCounter < other.commitCounter ? -1 : 1;
 		}
+		
+		public boolean exists() {
+			return digest != null;
+		}
     }
     
     /**
@@ -406,8 +438,8 @@
     		StringBuilder sb = new StringBuilder();
     		sb.append("Service: " + service + "\n");
     		for (HALogInfo li : logInfos) {
-    			sb.append(li.logName);
-    			sb.append(" " + BytesUtil.toHexString(li.digest));
+    			sb.append("CC[" + li.commitCounter + "]");
+    			sb.append(" " + (li.digest == null ? "NOT FOUND" : BytesUtil.toHexString(li.digest)));
     			sb.append(" " + (li.isOpen ? "open" : "closed"));
     			sb.append("\n");
     		}
@@ -436,15 +468,18 @@
     }
     
     /**
-     * Two required arguments:
-     * <HA configuration file>
-     * <ServiceRoot>
-     * 
+     * There are two required arguments:
+     * <li>HA configuration file</li>
+     * <li>ServiceRoot</li>
+     * <p>
      * A third optional argument 
      * "summary"
-     * which, if present, only generates output for HALogInfo that is different for teh joined services.
-     * 
-     * The DumpLogDigests dump method returns an Iterator over the halog files.
+     * which, if present, only generates output for HALogInfo that is different for the joined services.
+     * <p>
+     * The DumpLogDigests dump method returns an Iterator over the halog files returning ServiceLog
+     * objects.
+     * <p>
+     * The command line invocation simply writes a string representation of the returned data to standard out.
      */
     static public void main(final String[] args) throws ConfigurationException, IOException, InterruptedException, ExecutionException {
     	if (args.length < 2 || args.length > 3) {
@@ -583,6 +618,8 @@
 			ret.add(haglue);
 		}
 		
+		client.disconnect(true/*immediate shutdown*/);
+		
 		return ret;
 
 	}

Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HALogNexus.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HALogNexus.java	2013-09-10 13:50:17 UTC (rev 7391)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HALogNexus.java	2013-09-10 13:55:55 UTC (rev 7392)
@@ -32,6 +32,7 @@
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.Iterator;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -114,6 +115,12 @@
      *      IHAWriteMessage, ByteBuffer)
      */
     volatile IHAWriteMessage lastLiveHAWriteMessage = null;
+    
+    /*
+     * Set to protect log files against deletion while a digest is
+     * computed.  This is checked by deleteHALogs.
+     */
+    private  final AtomicLong digestLog = new AtomicLong(-1L);
 
     /**
      * Filter visits all HALog files <strong>except</strong> the current HALog
@@ -809,6 +816,21 @@
     }
     
     /**
+     * Protects logs from removal while a digest is being computed
+     * @param earliestDigest
+     */
+    void protectDigest(final long earliestDigest) {
+    	digestLog.set(earliestDigest);
+    }
+    
+    /**
+     * Releases current protection against log removal
+     */
+    void releaseProtectDigest() {
+    	digestLog.set(-1L);
+    }
+    
+    /**
      * Delete HALogs that are no longer required.
      * 
      * @param earliestRetainedSnapshotCommitCounter
@@ -832,7 +854,13 @@
 
             final long closingCommitCounter = r.getCommitCounter();
             
-            final boolean deleteFile = closingCommitCounter < earliestRetainedSnapshotCommitCounter;
+            final boolean deleteFile;
+            if (closingCommitCounter < earliestRetainedSnapshotCommitCounter) {
+            	// now check if protected by the digestLog field (set to -1 if not active)
+        		deleteFile = digestLog.get() == -1 || closingCommitCounter < digestLog.get();
+        	} else {
+        		deleteFile = false;
+        	}
 
             if (!deleteFile) {
 

Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3DumpLogs.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3DumpLogs.java	2013-09-10 13:50:17 UTC (rev 7391)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3DumpLogs.java	2013-09-10 13:55:55 UTC (rev 7392)
@@ -25,6 +25,7 @@
 
 import java.io.PrintWriter;
 import java.io.StringWriter;
+import java.util.Calendar;
 import java.util.Iterator;
 import java.util.UUID;
 
@@ -32,7 +33,58 @@
 
 public class TestHA3DumpLogs extends AbstractHA3JournalServerTestCase {
 
-	public TestHA3DumpLogs() {
+    @Override
+    protected String[] getOverrides() {
+
+        /*
+         * We need to set the time at which the DefaultSnapshotPolicy runs to
+         * some point in the Future in order to avoid test failures due to
+         * violated assumptions when the policy runs up self-triggering (based
+         * on the specified run time) during a CI run.
+         */
+        final String neverRun = getNeverRunSnapshotTime();
+        
+        return new String[]{
+                "com.bigdata.journal.jini.ha.HAJournalServer.restorePolicy=new com.bigdata.journal.jini.ha.DefaultRestorePolicy(0L,1,0)",
+                "com.bigdata.journal.jini.ha.HAJournalServer.snapshotPolicy=new com.bigdata.journal.jini.ha.DefaultSnapshotPolicy("+neverRun+",0)",
+                "com.bigdata.journal.jini.ha.HAJournalServer.onlineDisasterRecovery=true"
+        };
+        
+    }
+
+    /**
+     * We need to set the time at which the {@link DefaultSnapshotPolicy} runs
+     * to some point in the future in order to avoid test failures due to
+     * violated assumptions when the policy runs up self-triggering (based on
+     * the specified run time) during a CI run.
+     * <p>
+     * We do this by adding one hour to [now] and then converting it into the
+     * 'hhmm' format as an integer.
+     * 
+     * @return The "never run" time as hhmm.
+     */
+    static protected String getNeverRunSnapshotTime() {
+        
+        // Right now.
+        final Calendar c = Calendar.getInstance();
+        
+        // Plus an hour.
+        c.add(Calendar.HOUR_OF_DAY, 1);
+        
+        // Get the hour.
+        final int hh = c.get(Calendar.HOUR_OF_DAY);
+        
+        // And the minutes.
+        final int mm = c.get(Calendar.MINUTE);
+        
+        // Format as hhmm.
+        final String neverRun = "" + hh + (mm < 10 ? "0" : "") + mm;
+
+        return neverRun;
+        
+    }
+
+    public TestHA3DumpLogs() {
 		
 	}
 	

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:[7392] branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src

Fast, scalable, robust graph database platform

bigdata-commit