review only: Apexmalhar 2335

benchmark/src/main/java/com/datatorrent/benchmark/monitor/ResourceMonitorService.java

@@ -0,0 +1,96 @@
+/**
+ * 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 com.datatorrent.benchmark.monitor;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ResourceMonitorService
+{
+  public static ResourceMonitorService create(long period)
+  {
+    return new ResourceMonitorService(period);
+  }
+
+  private long period;
+  private ScheduledExecutorService executorService;
+
+  private ResourceMonitorService()
+  {
+    this(1000);
+  }
+
+  private ResourceMonitorService(long period)
+  {
+    this.period = period;
+  }
+
+  public ResourceMonitorService start()
+  {
+    executorService = Executors.newSingleThreadScheduledExecutor();
+    executorService.scheduleAtFixedRate(new LogGarbageCollectionTimeTask(period), 0, period, TimeUnit.MILLISECONDS);
+    executorService.scheduleAtFixedRate(new LogSystemResourceTask(), 0, period, TimeUnit.MILLISECONDS);
+
+    return this;
+  }
+
+  public void shutdownNow()
+  {
+    executorService.shutdownNow();
+  }
+
+  public static class LogGarbageCollectionTimeTask implements Runnable
+  {
+    private static final Logger logger = LoggerFactory.getLogger(LogGarbageCollectionTimeTask.class);
+
+    private long period;
+    private long lastGarbageCollectionTime = 0;
+
+    public LogGarbageCollectionTimeTask(long period)
+    {
+      this.period = period;
+    }
+
+    @Override
+    public void run()
+    {
+      long garbageCollectionTime = SystemResourceMonitor.create().getGarbageCollectionTime();
+      logger.info("Garbage Collection Time: total: {}; This period: {}; ratio: {}%", garbageCollectionTime,
+          garbageCollectionTime - lastGarbageCollectionTime,
+          (garbageCollectionTime - lastGarbageCollectionTime) * 100 / period);
+      lastGarbageCollectionTime = garbageCollectionTime;
+    }
+  }
+
+  public static class LogSystemResourceTask implements Runnable
+  {
+    private static final Logger logger = LoggerFactory.getLogger(LogSystemResourceTask.class);
+    private SystemResourceMonitor allResourceMonitor = SystemResourceMonitor.create(SystemResourceMonitor.allResources);
+
+    @Override
+    public void run()
+    {
+      logger.info("System Resource: {}", allResourceMonitor.getFormattedSystemResourceUsage());
+    }
+  }
+}

benchmark/src/main/java/com/datatorrent/benchmark/monitor/SystemResourceMonitor.java

@@ -0,0 +1,171 @@
+/**
+ * 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 com.datatorrent.benchmark.monitor;
+
+import java.lang.management.GarbageCollectorMXBean;
+import java.lang.management.ManagementFactory;
+import java.lang.management.OperatingSystemMXBean;
+import java.lang.reflect.Method;
+import java.text.NumberFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Throwables;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+public class SystemResourceMonitor
+{
+  public static enum SystemResource
+  {
+    /**
+     * memory
+     */
+    CommittedVirtualMemorySize, FreePhysicalMemorySize, TotalPhysicalMemorySize, TotalSwapSpaceSize, FreeSwapSpaceSize,
+
+    /**
+     * cpu
+     */
+    ProcessCpuTime, SystemCpuLoad, ProcessCpuLoad,
+
+    /**
+     * file
+     */
+    OpenFileDescriptorCount, MaxFileDescriptorCount
+  }
+
+  public static SystemResource[] memoryResources = new SystemResource[] {
+      SystemResource.CommittedVirtualMemorySize,
+      SystemResource.FreePhysicalMemorySize,
+      SystemResource.TotalPhysicalMemorySize,
+      SystemResource.TotalSwapSpaceSize,
+      SystemResource.FreeSwapSpaceSize};
+
+  public static SystemResource[] cpuResources = new SystemResource[] {
+      SystemResource.ProcessCpuTime,
+      SystemResource.SystemCpuLoad,
+      SystemResource.ProcessCpuLoad};
+
+  public static SystemResource[] fileResources = new SystemResource[] {
+      SystemResource.OpenFileDescriptorCount,
+      SystemResource.MaxFileDescriptorCount};
+
+  public static SystemResource[] allResources = SystemResource.values();
+
+  private static Map<Set<SystemResource>, SystemResourceMonitor> resourcesToMonitors = Maps.newHashMap();
+
+  private static SystemResourceMonitor defaultInstance;
+
+  public static SystemResourceMonitor create()
+  {
+    if (defaultInstance == null) {
+      synchronized (SystemResourceMonitor.class) {
+        if (defaultInstance == null) {
+          defaultInstance = new SystemResourceMonitor();
+        }
+      }
+    }
+    return defaultInstance;
+  }
+
+  public static SystemResourceMonitor create(SystemResource[] systemResources)
+  {
+    SystemResourceMonitor monitor = resourcesToMonitors.get(Sets.newHashSet(systemResources));
+    if (monitor == null) {
+      synchronized (SystemResourceMonitor.class) {
+        if (monitor == null) {
+          monitor = new SystemResourceMonitor(systemResources);
+          resourcesToMonitors.put(Sets.newHashSet(systemResources), monitor);
+        }
+      }
+    }
+    return monitor;
+  }
+
+  private SystemResource[] systemResources;
+  private OperatingSystemMXBean operatingSystemMXBean = ManagementFactory.getOperatingSystemMXBean();
+  private Map<SystemResource, Number> systemResourceUsage = Maps.newEnumMap(SystemResource.class);
+  private List<GarbageCollectorMXBean> garbageCollectorMXBeans = ManagementFactory.getGarbageCollectorMXBeans();
+
+  private SystemResourceMonitor()
+  {
+  }
+
+  private SystemResourceMonitor(SystemResource[] systemResources)
+  {
+    if (systemResources == null || systemResources.length == 0) {
+      throw new IllegalArgumentException("Resources should not null or empty");
+    }
+    this.systemResources = systemResources;
+  }
+
+  public Map<SystemResource, Number> getSystemResourceUsage()
+  {
+    systemResourceUsage.clear();
+    for (SystemResource sr : systemResources) {
+      Method getterMethod = getGetterMethod(sr);
+      getterMethod.setAccessible(true);
+      try {
+        systemResourceUsage.put(sr, (Number)getterMethod.invoke(operatingSystemMXBean));
+      } catch (Exception e) {
+        throw Throwables.propagate(e);
+      }
+    }
+    return systemResourceUsage;
+  }
+
+  public String getAppName()
+  {
+    return operatingSystemMXBean.getName();
+  }
+
+  public String getFormattedSystemResourceUsage()
+  {
+    Map<SystemResource, Number> usage = getSystemResourceUsage();
+    StringBuilder sb = new StringBuilder();
+    for (Map.Entry<SystemResource, Number> entry : usage.entrySet()) {
+      formatEntry(entry, sb);
+    }
+    return sb.toString();
+  }
+
+  private void formatEntry(Map.Entry<SystemResource, Number> entry, StringBuilder sb)
+  {
+    sb.append("\n\t").append(entry.getKey()).append(": ").append(NumberFormat.getInstance().format(entry.getValue()));
+  }
+
+  private Method getGetterMethod(SystemResource sr)
+  {
+    try {
+      return operatingSystemMXBean.getClass().getMethod("get" + sr.name(), null);
+    } catch (Exception e) {
+      throw Throwables.propagate(e);
+    }
+  }
+
+  public long getGarbageCollectionTime()
+  {
+    long time = 0;
+    for (GarbageCollectorMXBean collector : garbageCollectorMXBeans) {
+      time += collector.getCollectionTime();
+    }
+    return time;
+  }
+}

benchmark/src/main/java/com/datatorrent/benchmark/state/ManagedStateBenchmarkApp.java

@@ -48,6 +48,7 @@
 @ApplicationAnnotation(name = "ManagedStateBenchmark")
 public class ManagedStateBenchmarkApp implements StreamingApplication
 {
+  public static final int APP_WINDOWS = 20;
   protected static final String PROP_STORE_PATH = "dt.application.ManagedStateBenchmark.storeBasePath";
   protected static final String DEFAULT_BASE_PATH = "ManagedStateBenchmark/Store";
 
@@ -64,13 +65,15 @@ public void populateDAG(DAG dag, Configuration conf)
     dag.setAttribute(gen, OperatorContext.STATS_LISTENERS, Lists.newArrayList((StatsListener)sl));
 
     storeOperator = new StoreOperator();
+    storeOperator.setNumOfWindowPerStatistics(1);
     storeOperator.setStore(createStore(conf));
     storeOperator.setTimeRange(timeRange);
     storeOperator = dag.addOperator("Store", storeOperator);
 
     dag.setAttribute(storeOperator, OperatorContext.STATS_LISTENERS, Lists.newArrayList((StatsListener)sl));
 
     dag.addStream("Events", gen.data, storeOperator.input).setLocality(Locality.CONTAINER_LOCAL);
+    dag.setAttribute(OperatorContext.APPLICATION_WINDOW_COUNT, APP_WINDOWS);
   }
 
   public ManagedTimeUnifiedStateImpl createStore(Configuration conf)
@@ -96,7 +99,7 @@ public static class TestGenerator extends BaseOperator implements InputOperator
     public final transient DefaultOutputPort<KeyValPair<byte[], byte[]>> data = new DefaultOutputPort<KeyValPair<byte[], byte[]>>();
     int emitBatchSize = 1000;
     byte[] val = ByteBuffer.allocate(1000).putLong(1234).array();
-    int rate = 20000;
+    int rate = 20000 * APP_WINDOWS;
     int emitCount = 0;
     private final Random random = new Random();
     private int range = 1000 * 60; // one minute range of hot keys

benchmark/src/main/java/com/datatorrent/benchmark/state/StoreOperator.java

@@ -34,6 +34,7 @@
 import com.datatorrent.api.Context.OperatorContext;
 import com.datatorrent.api.DefaultInputPort;
 import com.datatorrent.api.Operator;
+import com.datatorrent.benchmark.monitor.ResourceMonitorService;
 import com.datatorrent.common.util.BaseOperator;
 import com.datatorrent.lib.util.KeyValPair;
 import com.datatorrent.netlet.util.Slice;
@@ -51,7 +52,8 @@ public enum ExecMode
     DO_NOTHING
   }
 
-  private static final int numOfWindowPerStatistics = 120;
+  private int numOfWindowPerStatistics = 120;
+  private static final int BUCKET_NUM = 20;
 
   //this is the store we are going to use
   private ManagedTimeUnifiedStateImpl store;
@@ -62,9 +64,10 @@ public enum ExecMode
   private int windowCountPerStatistics = 0;
   private long statisticsBeginTime = 0;
 
-  private ExecMode execMode = ExecMode.INSERT;
+  public ExecMode execMode = ExecMode.INSERT;
   private int timeRange = 1000 * 60;
 
+  private transient ResourceMonitorService monitorService;
   public final transient DefaultInputPort<KeyValPair<byte[], byte[]>> input = new DefaultInputPort<KeyValPair<byte[], byte[]>>()
   {
     @Override
@@ -79,11 +82,13 @@ public void setup(OperatorContext context)
   {
     logger.info("The execute mode is: {}", execMode.name());
     store.setup(context);
+    monitorService = ResourceMonitorService.create(60000).start();
   }
 
   @Override
   public void teardown()
   {
+    monitorService.shutdownNow();
   }
 
   @Override
@@ -283,4 +288,8 @@ public void setTimeRange(int timeRange)
     this.timeRange = timeRange;
   }
 
+  public void setNumOfWindowPerStatistics(int numOfWindowPerStatistics)
+  {
+    this.numOfWindowPerStatistics = numOfWindowPerStatistics;
+  }
 }

benchmark/src/test/java/com/datatorrent/benchmark/state/ManagedStateBenchmarkAppTest.java

@@ -72,7 +72,7 @@ public void test(ExecMode exeMode) throws Exception
     DAG dag = lma.getDAG();
 
     super.populateDAG(dag, conf);
-    storeOperator.setExecMode(exeMode);
+    storeOperator.execMode = exeMode;
 
     StreamingApplication app = new StreamingApplication()
     {
@@ -86,7 +86,7 @@ public void populateDAG(DAG dag, Configuration conf)
 
     // Create local cluster
     final LocalMode.Controller lc = lma.getController();
-    lc.run(300000);
+    lc.run(3000000);
 
     lc.shutdown();
   }