OAK-7259 - Improve SegmentNodeStoreStats to include number of commits…

… per thread and threads currently waiting on the semaphore

git-svn-id: https://svn.apache.org/repos/asf/jackrabbit/oak/trunk@1825984 13f79535-47bb-0310-9956-ffa450edef68

oak-examples/standalone/pom.xml

@@ -100,6 +100,11 @@
       <artifactId>metrics-core</artifactId>
       <optional>true</optional>
     </dependency>
+    <dependency>
+	  <groupId>com.googlecode.concurrentlinkedhashmap</groupId>
+   	  <artifactId>concurrentlinkedhashmap-lru</artifactId>
+   	  <optional>true</optional>
+	</dependency>
     <dependency>
       <groupId>org.mongodb</groupId>
       <artifactId>mongo-java-driver</artifactId>

oak-examples/webapp/pom.xml

@@ -201,6 +201,11 @@
       <artifactId>metrics-core</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+	  <groupId>com.googlecode.concurrentlinkedhashmap</groupId>
+   	  <artifactId>concurrentlinkedhashmap-lru</artifactId>
+   	  <scope>test</scope>
+	</dependency>
   </dependencies>
 
   <build>

oak-it/pom.xml

@@ -212,5 +212,10 @@
             <artifactId>metrics-core</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+	  		<groupId>com.googlecode.concurrentlinkedhashmap</groupId>
+	    	<artifactId>concurrentlinkedhashmap-lru</artifactId>
+	    	<scope>test</scope>
+      	</dependency>
     </dependencies>
 </project>

oak-jcr/pom.xml

@@ -458,5 +458,10 @@
       <artifactId>metrics-core</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+	  <groupId>com.googlecode.concurrentlinkedhashmap</groupId>
+	  <artifactId>concurrentlinkedhashmap-lru</artifactId>
+	  <scope>test</scope>
+   	</dependency>
   </dependencies>
 </project>

oak-lucene/pom.xml

@@ -387,6 +387,11 @@
       <artifactId>metrics-core</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+ 	  <groupId>com.googlecode.concurrentlinkedhashmap</groupId>
+	  <artifactId>concurrentlinkedhashmap-lru</artifactId>
+	  <scope>test</scope>
+   	</dependency>
     <dependency>
       <groupId>org.apache.commons</groupId>
       <artifactId>commons-exec</artifactId>

oak-parent/pom.xml

@@ -537,6 +537,11 @@
         <artifactId>guava</artifactId>
         <version>${guava.version}</version>
       </dependency>
+      <dependency>
+	  	<groupId>com.googlecode.concurrentlinkedhashmap</groupId>
+	    <artifactId>concurrentlinkedhashmap-lru</artifactId>
+	    <version>1.4.2</version>
+      </dependency>
       <dependency>
         <groupId>org.easymock</groupId>
         <artifactId>easymock</artifactId>

oak-pojosr/pom.xml

@@ -352,5 +352,10 @@
       <artifactId>metrics-core</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+ 	  <groupId>com.googlecode.concurrentlinkedhashmap</groupId>
+      <artifactId>concurrentlinkedhashmap-lru</artifactId>
+      <scope>test</scope>
+   	</dependency>
   </dependencies>
 </project>

oak-run-commons/pom.xml

@@ -113,6 +113,10 @@
             <groupId>io.dropwizard.metrics</groupId>
             <artifactId>metrics-core</artifactId>
         </dependency>
+        <dependency>
+ 	 	 	<groupId>com.googlecode.concurrentlinkedhashmap</groupId>
+	 		 <artifactId>concurrentlinkedhashmap-lru</artifactId>
+   		</dependency>
 
         <!-- Test dependencies -->
         <dependency>

oak-segment-tar/pom.xml

@@ -236,6 +236,11 @@
             <artifactId>jsr305</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>com.googlecode.concurrentlinkedhashmap</groupId>
+            <artifactId>concurrentlinkedhashmap-lru</artifactId>
+            <scope>provided</scope>
+	    </dependency>
         <dependency>
             <groupId>org.osgi</groupId>
             <artifactId>org.osgi.core</artifactId>

oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/CommitsTracker.java

@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.jackrabbit.oak.segment;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.stream.Stream;
+
+import com.googlecode.concurrentlinkedhashmap.ConcurrentLinkedHashMap;
+
+/**
+ * A simple tracker for the source of commits (writes) in
+ * {@link SegmentNodeStore}. It provides two basic functionalities:
+ * <ul>
+ * <li>exposes the number of commits executed per thread</li>
+ * <li>exposes the threads (and possibly their details - i.e., stack traces)
+ * currently waiting on the commit semaphore
+ * </ul>
+ * 
+ * This class delegates thread-safety to its underlying state variables. 
+ */
+class CommitsTracker {
+    private static final int DEFAULT_COMMITS_COUNT_MAP_SIZE = 20;
+    private static final boolean DEFAULT_COLLECT_STACK_TRACES = true;
+
+    private volatile boolean collectStackTraces;
+
+    private final ConcurrentMap<String, String> queuedWritersMap;
+    private final ConcurrentMap<String, Long> commitsCountMap;
+
+    CommitsTracker() {
+        this(DEFAULT_COMMITS_COUNT_MAP_SIZE, DEFAULT_COLLECT_STACK_TRACES);
+    }
+
+    CommitsTracker(int commitsCountMapMaxSize, boolean collectStackTraces) {
+        this.collectStackTraces = collectStackTraces;
+        this.commitsCountMap = new ConcurrentLinkedHashMap.Builder<String, Long>()
+                .maximumWeightedCapacity(commitsCountMapMaxSize).build();
+        this.queuedWritersMap = new ConcurrentHashMap<>();
+    }
+
+    public void trackQueuedCommitOf(Thread t) {
+        String writerDetails = "N/A";
+        if (collectStackTraces) {
+            StringBuilder threadDetails = new StringBuilder();
+            Stream.of(t.getStackTrace()).forEach(threadDetails::append);
+            writerDetails = threadDetails.toString();
+        }
+
+        queuedWritersMap.put(t.getName(), writerDetails);
+    }
+
+    public void trackDequedCommitOf(Thread t) {
+        queuedWritersMap.remove(t.getName());
+    }
+
+    public void trackExecutedCommitOf(Thread t) {
+        commitsCountMap.compute(t.getName(), (w, v) -> v == null ? 1 : v + 1);
+    }
+
+    public void setCollectStackTraces(boolean flag) {
+        this.collectStackTraces = flag;
+    }
+
+    public Map<String, String> getQueuedWritersMap() {
+        return new HashMap<>(queuedWritersMap);
+    }
+
+    public Map<String, Long> getCommitsCountMap() {
+        return new HashMap<>(commitsCountMap);
+    }
+
+}

oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentNodeStoreMonitor.java

@@ -26,56 +26,41 @@ public interface SegmentNodeStoreMonitor {
 
     SegmentNodeStoreMonitor DEFAULT = new SegmentNodeStoreMonitor() {
         @Override
-        public void onCommit() {
+        public void onCommit(Thread t, long time) {
 
         }
 
         @Override
-        public void onCommitQueued() {
+        public void onCommitQueued(Thread t) {
 
         }
 
         @Override
-        public void onCommitDequeued() {
-
-        }
-
-        @Override
-        public void committedAfter(long time) {
-
-        }
-
-        @Override
-        public void dequeuedAfter(long time) {
+        public void onCommitDequeued(Thread t, long time) {
 
         }
     };
 
     /**
-     * Notifies the monitor when a new commit was persisted right away
+     * Notifies the monitor when a new commit was persisted.
+     * @param t the thread which initiated the write
+     * @param time the time spent for persisting the commit
      */
-    void onCommit();
+    void onCommit(Thread t, long time);
 
     /**
      * Notifies the monitor when a new commit couldn't be persisted, but was
-     * queued for later retry
+     * queued for later retry.
+     * 
+     * @param t the thread which initiated the write
      */
-    void onCommitQueued();
+    void onCommitQueued(Thread t);
 
     /**
      * Notifies the monitor when a queued commit was dequeued for processing.
+     * @param t the thread which initiated the write
+     * @param time the time spent in the queue
      */
-    void onCommitDequeued();
+    void onCommitDequeued(Thread t, long time);
 
-    /**
-     * Notifies the monitor time spent (excluding queuing time) for a commit.
-     * @param time the time spent
-     */
-    void committedAfter(long time);
-
-    /**
-     * Notifies the monitor time spent in the queue for a commit, before being processed.
-     * @param time the time spent
-     */
-    void dequeuedAfter(long time);
-}
+}