Thread: [Bigdata-commit] SF.net SVN: bigdata:[3549] branches/JOURNAL_HA_BRANCH/bigdata-rdf/src

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

Revision: 3549
          http://bigdata.svn.sourceforge.net/bigdata/?rev=3549&view=rev
Author:   thompsonbry
Date:     2010-09-14 19:14:30 +0000 (Tue, 14 Sep 2010)

Log Message:
-----------
Modified LoadStats and ClosureStats to use CATs in preparation for a concurrent data loader.

Modified Paths:
--------------
    branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/ClosureStats.java
    branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/TruthMaintenance.java
    branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/load/SingleResourceReaderTask.java
    branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/rio/LoadStats.java
    branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/store/DataLoader.java
    branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/test/com/bigdata/rdf/metrics/TaskATest.java
    branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/test/com/bigdata/rdf/metrics/TestMetrics.java

Modified: branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/ClosureStats.java
===================================================================

--- branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/ClosureStats.java	2010-09-14 19:13:44 UTC (rev 3548)
+++ branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/ClosureStats.java	2010-09-14 19:14:30 UTC (rev 3549)
@@ -24,6 +24,8 @@
 */
 package com.bigdata.rdf.inf;
 
+import com.bigdata.counters.CAT;
+
 /**
  * Statistics collected when performing inference.
  * 
@@ -38,13 +40,13 @@
      * change in the #of statements in the database across the closure
      * operation.
      */
-    public long mutationCount;
+    public final CAT mutationCount = new CAT();
 
     /**
      * Time to compute the entailments and store them within the database
      * (milliseconds).
      */
-    public long elapsed;
+    public final CAT elapsed = new CAT();
 
     public ClosureStats() {
         
@@ -55,26 +57,26 @@
      * @param mutationCount
      * @param elapsed
      */
-    public ClosureStats(long mutationCount,long elapsed) {
+    public ClosureStats(final long mutationCount,final long elapsed) {
         
-        this.mutationCount = mutationCount;
+        this.mutationCount.set(mutationCount);
         
-        this.elapsed = elapsed;
+        this.elapsed.set( elapsed);
         
     }
     
-    public synchronized void add(ClosureStats o) {
+    public void add(final ClosureStats o) {
         
-        this.mutationCount += o.mutationCount;
+        this.mutationCount.add( o.mutationCount.get());
         
-        this.elapsed += o.elapsed;
+        this.elapsed.add(o.elapsed.get());
         
     }
     
     public String toString() {
 
-        return getClass().getSimpleName() + "{mutationCount=" + mutationCount
-                + ", elapsed=" + elapsed + "ms}";
+        return getClass().getSimpleName() + "{mutationCount=" + mutationCount.estimate_get()
+                + ", elapsed=" + elapsed.estimate_get() + "ms}";
         
     }
     

Modified: branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/TruthMaintenance.java
===================================================================
--- branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/TruthMaintenance.java	2010-09-14 19:13:44 UTC (rev 3548)
+++ branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/inf/TruthMaintenance.java	2010-09-14 19:14:30 UTC (rev 3549)
@@ -440,7 +440,7 @@
 
         final long elapsed = System.currentTimeMillis() - begin;
         
-        stats.elapsed += elapsed;
+        stats.elapsed.add(elapsed);
 
         if (INFO)
             log.info("Computed closure in " + elapsed + "ms");

Modified: branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/load/SingleResourceReaderTask.java
===================================================================
--- branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/load/SingleResourceReaderTask.java	2010-09-14 19:13:44 UTC (rev 3548)
+++ branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/load/SingleResourceReaderTask.java	2010-09-14 19:14:30 UTC (rev 3549)
@@ -148,7 +148,7 @@
         }
         
         // Note: IFF the task succeeds!
-        toldTriples.addAndGet(loadStats.toldTriples);
+        toldTriples.addAndGet(loadStats.toldTriples.get());
 
     }
 
@@ -194,9 +194,11 @@
 
             final long now = System.currentTimeMillis();
             
-            stats.toldTriples = nstmts;
+            stats.toldTriples.set(nstmts);
 
-            stats.totalTime = stats.loadTime = now - begin;
+            stats.totalTime.set( now - begin );
+
+            stats.loadTime.set( now - begin );
             
             /*
              * This reports the load rate for the file, but this will only

Modified: branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/rio/LoadStats.java
===================================================================
--- branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/rio/LoadStats.java	2010-09-14 19:13:44 UTC (rev 3548)
+++ branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/rio/LoadStats.java	2010-09-14 19:14:30 UTC (rev 3549)
@@ -24,6 +24,7 @@
 */
 package com.bigdata.rdf.rio;
 
+import com.bigdata.counters.CAT;
 import com.bigdata.rdf.inf.ClosureStats;
 
 /**
@@ -34,12 +35,12 @@
  */
 public class LoadStats {
 
-    public long toldTriples;
-    public long loadTime;
-    public long commitTime;
-    public long totalTime;
+    public final CAT toldTriples = new CAT();
+    public final CAT loadTime = new CAT();
+    public final CAT commitTime = new CAT();
+    public final CAT totalTime = new CAT();
     
-    private transient long lastReportTime = 0l;
+    private transient volatile long lastReportTime = 0l;
 
     /**
      * The internal with which this class will log on {@link System#out} in
@@ -55,19 +56,19 @@
 
     public long triplesPerSecond() {
 
-        return ((long) (((double) toldTriples) / ((double) totalTime) * 1000d));
+        return ((long) (((double) toldTriples.estimate_get()) / ((double) totalTime.estimate_get()) * 1000d));
 
     }
 
     public void add(final LoadStats stats) {
 
-        toldTriples += stats.toldTriples;
+        toldTriples.add(stats.toldTriples.get());
 
-        loadTime += stats.loadTime;
+        loadTime.add(stats.loadTime.get());
 
-        commitTime += stats.commitTime;
+        commitTime.add(stats.commitTime.get());
 
-        totalTime += stats.totalTime;
+        totalTime.add(stats.totalTime.get());
 
         if (stats.closureStats != null) {
 
@@ -82,7 +83,7 @@
 
         if (lastReportTime == 0L) {
 
-            if (loadTime >= REPORT_INTERVAL) {
+            if (loadTime.estimate_get() >= REPORT_INTERVAL) {
 
                 System.out.println("loading: " + toString());
 
@@ -111,14 +112,14 @@
 
         return toldTriples
                 + " stmts added in "
-                + ((double) loadTime)
+                + ((double) loadTime.estimate_get())
                 / 1000d
                 + " secs, rate= "
                 + triplesPerSecond()
                 + ", commitLatency="
-                + commitTime
+                + commitTime.estimate_get()
                 + "ms"
-                + (closureStats.elapsed!=0L? "\n"+closureStats.toString() : "");
+                + (closureStats.elapsed.estimate_get()!=0L? "\n"+closureStats.toString() : "");
 
     }
     

Modified: branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/store/DataLoader.java
===================================================================
--- branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/store/DataLoader.java	2010-09-14 19:13:44 UTC (rev 3548)
+++ branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/java/com/bigdata/rdf/store/DataLoader.java	2010-09-14 19:14:30 UTC (rev 3549)
@@ -230,8 +230,8 @@
 
         if (buffer != null) {
 
-            if(log.isInfoEnabled())
-                log.info("Flushing the buffer.");
+            if(log.isDebugEnabled())
+                log.debug("Flushing the buffer.");
             
             buffer.flush();
             
@@ -610,7 +610,7 @@
             
             database.commit();
 
-            totals.commitTime += System.currentTimeMillis() - beginCommit;
+            totals.commitTime.add(System.currentTimeMillis() - beginCommit);
 
             if (log.isInfoEnabled())
                 log.info("commit: latency="+totals.commitTime+"ms");
@@ -838,8 +838,8 @@
 
         if (file.isDirectory()) {
 
-            if (log.isInfoEnabled())
-                log.info("loading directory: " + file);
+            if (log.isDebugEnabled())
+                log.debug("loading directory: " + file);
 
 //            final LoadStats loadStats = new LoadStats();
 
@@ -1007,9 +1007,9 @@
 
             final long nstmts = loader.getStatementsAdded();
 
-            stats.toldTriples = nstmts;
+            stats.toldTriples.set( nstmts );
 
-            stats.loadTime = System.currentTimeMillis() - begin;
+            stats.loadTime.set(System.currentTimeMillis() - begin);
 
             if (closureEnum == ClosureEnum.Incremental
                     || (endOfBatch && closureEnum == ClosureEnum.Batch)) {
@@ -1037,20 +1037,24 @@
 
                 database.commit();
 
-                stats.commitTime = System.currentTimeMillis() - beginCommit;
+                stats.commitTime.set(System.currentTimeMillis() - beginCommit);
 
                 if (log.isInfoEnabled())
                     log.info("commit: latency=" + stats.commitTime + "ms");
 
             }
 
-            stats.totalTime = System.currentTimeMillis() - begin;
+            stats.totalTime.set(System.currentTimeMillis() - begin);
 
+            // aggregate stats
+            totals.add(stats);
+
             if (log.isInfoEnabled()) {
-                log.info(stats.toString());
+				log.info("file:: " + stats + "; totals:: " + totals);
                 if (buffer != null
                         && buffer.getDatabase() instanceof AbstractLocalTripleStore) {
-                    log.info(((AbstractLocalTripleStore) buffer.getDatabase())
+                	if(log.isDebugEnabled())
+                    log.debug(((AbstractLocalTripleStore) buffer.getDatabase())
                             .getLocalBTreeBytesWritten(new StringBuilder())
                             .toString());
                 }
@@ -1060,6 +1064,9 @@
             
         } catch ( Exception ex ) {
 
+        	// aggregate stats even for exceptions.
+            totals.add(stats);
+
             /*
              * Note: discard anything in the buffer in case auto-flush is
              * disabled. This prevents the buffer from retaining data after a
@@ -1096,11 +1103,11 @@
             
             throw ex2;
             
-        } finally {
+//        } finally {
+//            
+//            // aggregate regardless of the outcome.
+//            totals.add(stats);
             
-            // aggregate regardless of the outcome.
-            totals.add(stats);
-            
         }
 
     }
@@ -1436,8 +1443,8 @@
                     || (name.endsWith(".gz") && RDFFormat.forFileName(name
                             .substring(0, name.length() - 3)) != null);
 
-			if (log.isInfoEnabled())
-				log.info("dir=" + dir + ", name=" + name + " : isRDF=" + isRDF);
+			if (log.isDebugEnabled())
+				log.debug("dir=" + dir + ", name=" + name + " : isRDF=" + isRDF);
 
             return isRDF;
 

Modified: branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/test/com/bigdata/rdf/metrics/TaskATest.java
===================================================================
--- branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/test/com/bigdata/rdf/metrics/TaskATest.java	2010-09-14 19:13:44 UTC (rev 3548)
+++ branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/test/com/bigdata/rdf/metrics/TaskATest.java	2010-09-14 19:14:30 UTC (rev 3549)
@@ -475,23 +475,23 @@
             
             
             // Explicit + (Entailments = Axioms + Inferred)
-            final long totalTriples = loadStats[run].toldTriples
-                    + (closureStats!=null?closureStats.mutationCount : 0);
+            final long totalTriples = loadStats[run].toldTriples.get()
+                    + (closureStats!=null?closureStats.mutationCount.get() : 0);
 
             // loadTime + closureTime + commitTime.
-            final long totalTime = loadStats[run].loadTime
-                    + (closureStats != null ? closureStats.elapsed : 0)
-                    + loadStats[run].commitTime;
+            final long totalTime = loadStats[run].loadTime.get()
+                    + (closureStats != null ? closureStats.elapsed.get() : 0)
+                    + loadStats[run].commitTime.get();
             
             System.out.println( all_sources[ run * 3 ]+ ", " +
                     ( errors[ run ] == null
                       ? "Ok"
                             +", "+loadStats[run].toldTriples
-                            +", "+loadStats[run].loadTime/1000
-                            +", "+tps(loadStats[run].toldTriples,loadStats[run].loadTime)
-                            +", "+(closureStats!=null?closureStats.mutationCount:"")
-                            +", "+(closureStats!=null?closureStats.elapsed/1000:"")
-                            +", "+(closureStats!=null?tps(closureStats.mutationCount,closureStats.elapsed):"")
+                            +", "+loadStats[run].loadTime.get()/1000
+                            +", "+tps(loadStats[run].toldTriples.get(),loadStats[run].loadTime.get())
+                            +", "+(closureStats!=null?closureStats.mutationCount.get():"")
+                            +", "+(closureStats!=null?closureStats.elapsed.get()/1000:"")
+                            +", "+(closureStats!=null?tps(closureStats.mutationCount.get(),closureStats.elapsed.get()):"")
                             +", "+loadStats[run].commitTime
                             +", "+tps(totalTriples,totalTime)
 

Modified: branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/test/com/bigdata/rdf/metrics/TestMetrics.java
===================================================================
--- branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/test/com/bigdata/rdf/metrics/TestMetrics.java	2010-09-14 19:13:44 UTC (rev 3548)
+++ branches/JOURNAL_HA_BRANCH/bigdata-rdf/src/test/com/bigdata/rdf/metrics/TestMetrics.java	2010-09-14 19:14:30 UTC (rev 3549)
@@ -1085,7 +1085,7 @@
             /*
              * #of explicit statements loaded.
              */
-            toldTriples = loadStats.toldTriples;
+            toldTriples = loadStats.toldTriples.get();
             
             /*
              * This is the elapsed time for the entire transaction in which the file
@@ -1093,19 +1093,19 @@
              * store, and the time required to perform the transaction commit.
              */
 //            transactionTime = System.currentTimeMillis() - begin;
-            transactionTime = loadStats.totalTime;
+            transactionTime = loadStats.totalTime.get();
 
             /*
              * This is the time required to load the triples exclusive of the
              * startup and commit time for the transaction.
              */
-            loadTime = loadStats.loadTime;
+            loadTime = loadStats.loadTime.get();
 
             /*
              * A pragmatic estimate of the commit time that assumes the transaction
              * start time is zero. 
              */
-            commitTime = loadStats.commitTime;
+            commitTime = loadStats.commitTime.get();
 //            commitTime = transactionTime - loadTime;
 
             /*
@@ -1132,7 +1132,7 @@
 
             statementsAdded = statementCount1 - statementCount0;
 //            inferencesAdded = inferenceCount1 - inferenceCount0;
-            inferencesAdded = loadStats.closureStats.mutationCount;
+            inferencesAdded = loadStats.closureStats.mutationCount.get();
 //            int explicitAdded   = statementsAdded - inferencesAdded;
             proofsAdded     = proofCount1 - proofCount0;
             urisAdded       = uriCount1 - uriCount0;


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:[3549] branches/JOURNAL_HA_BRANCH/bigdata-rdf/src

Fast, scalable, robust graph database platform

bigdata-commit