Merge pull request #67 from jentfoo/statChanges

Client: Add the ability to disable stats

Author: jentfoo

gradle.properties

@@ -1,3 +1,3 @@
 group = org.threadly
 version = 4.10-SNAPSHOT
-threadlyVersion = 5.34
+threadlyVersion = 5.37

src/main/java/org/threadly/litesockets/Client.java

@@ -14,7 +14,6 @@
 import org.threadly.litesockets.buffers.ReuseableMergedByteBuffers;
 import org.threadly.litesockets.utils.IOUtils;
 import org.threadly.litesockets.utils.SimpleByteStats;
-import org.threadly.util.ArgumentVerifier;
 import org.threadly.util.Clock;
 import org.threadly.util.ExceptionUtils;
 
@@ -40,9 +39,9 @@ public abstract class Client implements Closeable {
   protected final SocketExecuterCommonBase se;
   protected final long startTime = Clock.lastKnownForwardProgressingMillis();
   protected final Object readerLock = new Object();
-  protected final ClientByteStats stats = new ClientByteStats();
   protected final AtomicBoolean closed = new AtomicBoolean(false);
   protected final ConcurrentLinkedQueue<ClientCloseListener> closerListener = new ConcurrentLinkedQueue<>();
+  protected volatile ClientByteStats stats;
   protected volatile Runnable readerCaller = null;
   protected volatile boolean useNativeBuffers = false;
   protected volatile boolean keepReadBuffer = true;
@@ -51,12 +50,17 @@ public abstract class Client implements Closeable {
   protected volatile int newReadBufferSize = IOUtils.DEFAULT_CLIENT_READ_BUFFER_SIZE;
   private ByteBuffer readByteBuffer = IOUtils.EMPTY_BYTEBUFFER;
 
-  public Client(final SocketExecuterCommonBase se) {
+  public Client(final SocketExecuterCommonBase se, final boolean statsEnabled) {
+    setStatsEnabled(statsEnabled);
+    
     this.se = se;
     this.clientExecutor = se.getExecutorFor(this);
   }
 
-  protected Client(final SocketExecuterCommonBase se, final SubmitterExecutor lclientExecutor) {
+  protected Client(final SocketExecuterCommonBase se, final SubmitterExecutor lclientExecutor, 
+                   final boolean statsEnabled) {
+    setStatsEnabled(statsEnabled);
+    
     this.se = se;
     this.clientExecutor = lclientExecutor;
   }
@@ -120,14 +124,12 @@ protected <T> SettableListenableFuture<T> makeClientSettableListenableFuture() {
    */
   public abstract SocketAddress getRemoteSocketAddress();
 
-
   /**
    * 
    * @return the local {@link SocketAddress} this client is using.
    */
   public abstract SocketAddress getLocalSocketAddress();
 
-
   /**
    * Returns true if this client has data pending in its write buffers.  False if there is no data pending write.
    *
@@ -232,11 +234,11 @@ public void close() {
 
   /*Implemented functions*/
 
-  protected void addReadStats(final int size) {
+  protected void recordReadStats(final int size) {
     stats.addRead(size);
   }
 
-  protected void addWriteStats(final int size) {
+  protected void recordWriteStats(final int size) {
     stats.addWrite(size);
   }
 
@@ -311,8 +313,8 @@ protected void addReadBuffer(final ByteBuffer bb) {
     if (! bb.hasRemaining()) {
       return;
     }
-    addReadStats(bb.remaining());
-    se.addReadAmount(bb.remaining());
+    recordReadStats(bb.remaining());
+    se.recordReadStats(bb.remaining());
     int start;
     // synchronize to ensure readBuffers are not modified by non-client thread getRead call
     synchronized (readerLock) {
@@ -441,24 +443,49 @@ protected boolean setClose() {
   public SimpleByteStats getStats() {
     return stats;
   }
+  
+  /**
+   * Can be invoked with {@code false} to disable stat collection on client.  This can reduce 
+   * minor memory and performance overheads.
+   * 
+   * @param enabled {@code false} to disable stat collection, or {@code true} to enable / reset stats
+   */
+  public void setStatsEnabled(boolean enabled) {
+    if (enabled) {
+      ClientByteStats stats = this.stats;
+      if (stats == ClientByteStats.NO_OP_STATS || stats == null) {
+        this.stats = new ClientByteStats();
+      } else {
+        stats.resetStats();
+      }
+    } else {
+      stats = ClientByteStats.NO_OP_STATS;
+    }
+  }
 
   /**
    * Implementation of the SimpleByteStats.
    */
   private static class ClientByteStats extends SimpleByteStats {
-    public ClientByteStats() {
-      super();
-    }
+    public static final ClientByteStats NO_OP_STATS = new ClientByteStats() {
+      @Override
+      protected void addWrite(final int size) {
+        // ignored
+      }
+
+      @Override
+      protected void addRead(final int size) {
+        // ignored
+      }
+    };
 
     @Override
     protected void addWrite(final int size) {
-      ArgumentVerifier.assertNotNegative(size, "size");
       super.addWrite(size);
     }
 
     @Override
     protected void addRead(final int size) {
-      ArgumentVerifier.assertNotNegative(size, "size");
       super.addRead(size);
     }
   }
@@ -491,7 +518,6 @@ public interface Reader {
     public void onRead(Client client);
   }
 
-
   /**
    * Used to notify when a Client is closed.
    * 

src/main/java/org/threadly/litesockets/NoThreadSocketExecuter.java

@@ -40,7 +40,15 @@ public class NoThreadSocketExecuter extends SocketExecuterCommonBase {
    * Constructs a NoThreadSocketExecuter.  {@link #start()} must still be called before using it.
    */
   public NoThreadSocketExecuter() {
-    super(new NoThreadScheduler());
+    this(new NoThreadScheduler()); 
+  }
+  
+  /**
+   * Constructs a NoThreadSocketExecuter.  {@link #start()} must still be called before using it.
+   */
+  public NoThreadSocketExecuter(NoThreadScheduler scheduler) {
+    super(scheduler);
+
     localNoThreadScheduler = (NoThreadScheduler)schedulerPool; 
   }
 
@@ -88,8 +96,6 @@ public void setUDPServerOperations(final UDPServer udpServer, final boolean enab
   protected void startupService() {
     commonSelector = openSelector();
     this.acceptSelector = commonSelector;
-    this.readSelector = commonSelector;
-    this.writeSelector = commonSelector;
   }
 
   @Override
@@ -219,7 +225,7 @@ public void select(final int delay) {
                     if(server != null) {
                       if(server instanceof UDPServer) {
                         UDPServer us = (UDPServer) server;
-                        stats.addWrite(us.doWrite());
+                        recordWriteStats(us.doWrite());
                         setUDPServerOperations(us, true);
                       }
                     }

src/main/java/org/threadly/litesockets/SingleThreadSocketExecuter.java

@@ -1,35 +1,69 @@
 package org.threadly.litesockets;
 
-import org.threadly.concurrent.SingleThreadScheduler;
+import org.threadly.concurrent.ConfigurableThreadFactory;
+import org.threadly.concurrent.NoThreadScheduler;
+import org.threadly.util.ExceptionUtils;
 
 /**
- * This is a SingleThreaded implementation of a SocketExecuter. 
- * 
- * 
- *
+ * This is a SingleThreaded implementation of a SocketExecuter.  If single threaded performance is 
+ * desired it should be slightly less overhead than a {@link ThreadedSocketExecuter}.
  */
 public class SingleThreadSocketExecuter extends NoThreadSocketExecuter {
   private static final int SELECT_TIME_MS = 10000;
+  private static final ConfigurableThreadFactory THREAD_FACTORY = 
+      new ConfigurableThreadFactory("SingleThreadSocketExecuter-", false, true, 
+                                    Thread.currentThread().getPriority(), null, null);
 
-  private final SingleThreadScheduler sts = new SingleThreadScheduler();
-  
+  private Thread runningThread = null;
+
+  /**
+   * Constructs a SingleThreadSocketExecuter.  {@link #start()} must still be called before using it.
+   */
   public SingleThreadSocketExecuter() {
     super();
   }
+
+  /**
+   * Constructs a SingleThreadSocketExecuter.  {@link #start()} must still be called before using it.
+   * <p>
+   * This accepts a {@link NoThreadScheduler} so that stats can be collected if desired.
+   */
+  public SingleThreadSocketExecuter(NoThreadScheduler scheduler) {
+    super(scheduler);
+  }
 
   @Override
   protected void startupService() {
     super.startupService();
-    sts.execute(()->{
+    runningThread = THREAD_FACTORY.newThread(()->{
       while(isRunning()) {
-        super.select(SELECT_TIME_MS);
+        try {
+          super.select(SELECT_TIME_MS);
+        } catch (Throwable t) {
+          ExceptionUtils.handleException(t);
+        }
       }
     });
+    runningThread.start();
   }
 
+  @Override
+  protected void shutdownService() {
+    super.shutdownService();
+    
+    try {
+      runningThread.join();
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
   public void select() {
   }
 
+  @Override
   public void select(int delay) {
   }
 }

src/main/java/org/threadly/litesockets/SocketExecuter.java

@@ -21,6 +21,30 @@
  *
  */
 public interface SocketExecuter extends Service {
+  /**
+   * By default per-connection stats will be enabled.  Stats can always be disabled on a per-Client 
+   * basis, but this allows the default state of the stats to be enabled / disabled.
+   * 
+   * @param enabled {@code false} if stats should not be collected per-connection by default
+   */
+  public void setPerConnectionStatsEnabled(boolean enabled);
+  
+  
+  /**
+   * Check the total number of bytes pending to be sent by clients.  This is data which has been 
+   * provided to the client to write, but is waiting on the network and kernel to accept the write.
+   * 
+   * @return The total number of bytes pending to write by clients
+   */
+  public long getTotalPendingWriteBytes();
+  
+  /**
+   * Check the total amount of pending reads across all associated clients.  Bytes here indicate 
+   * that the client has been notified a read is available but has not consumed it.
+   * 
+   * @return The total number of bytes pending to read by clients
+   */
+  public long getTotalPendingReadBytes();
 
   /**
    * This will create a UDPServer on the specified {@link SocketExecuter}.

src/main/java/org/threadly/litesockets/SocketExecuterCommonBase.java

@@ -23,45 +23,58 @@
  *  This is a common base class for the Threaded and NoThread SocketExecuters. 
  */
 abstract class SocketExecuterCommonBase extends AbstractService implements SocketExecuter {
-  private final Logger log = Logger.getLogger(this.getClass().toString());
-  protected final SubmitterScheduler acceptScheduler;
-  protected final SubmitterScheduler readScheduler;
-  protected final SubmitterScheduler writeScheduler;
+  protected final Logger log = Logger.getLogger(this.getClass().toString());
   protected final SubmitterScheduler schedulerPool;
+  protected final SubmitterScheduler acceptScheduler;
   protected final ConcurrentHashMap<SocketChannel, Client> clients = new ConcurrentHashMap<>();
   protected final ConcurrentHashMap<SelectableChannel, Server> servers = new ConcurrentHashMap<>();
   protected final SocketExecuterByteStats stats = new SocketExecuterByteStats();
   protected final WatchdogCache dogCache;
-  protected volatile boolean verboseLogging = false;
-  protected Selector readSelector;
-  protected Selector writeSelector;
+  protected volatile boolean perConnectionStatsEnabled = true;
   protected Selector acceptSelector;
 
   SocketExecuterCommonBase(final SubmitterScheduler scheduler) {
-    this(scheduler,scheduler,scheduler,scheduler);
+    this(scheduler, scheduler);
   }
 
-  SocketExecuterCommonBase(final SubmitterScheduler acceptScheduler, 
-      final SubmitterScheduler readScheduler, 
-      final SubmitterScheduler writeScheduler, 
-      final SubmitterScheduler ssi) {
+  SocketExecuterCommonBase(final SubmitterScheduler acceptScheduler, final SubmitterScheduler ssi) {
     log.setParent(Logger.getGlobal());
     ArgumentVerifier.assertNotNull(ssi, "ThreadScheduler");    
     ArgumentVerifier.assertNotNull(acceptScheduler, "Accept Scheduler");
-    ArgumentVerifier.assertNotNull(readScheduler, "Read Scheduler");
-    ArgumentVerifier.assertNotNull(writeScheduler, "Write Scheduler");
+    
     schedulerPool = ssi;
     dogCache = new WatchdogCache(ssi, true);
     this.acceptScheduler = acceptScheduler;
-    this.readScheduler = readScheduler;
-    this.writeScheduler = writeScheduler;
   }
 
-  protected void addReadAmount(int size) {
+  @Override
+  public long getTotalPendingWriteBytes() {
+    long result = 0;
+    for (Client c : clients.values()) {
+      result += c.getWriteBufferSize();
+    }
+    return result;
+  }
+  
+  @Override
+  public long getTotalPendingReadBytes() {
+    long result = 0;
+    for (Client c : clients.values()) {
+      result += c.getReadBufferSize();
+    }
+    return result;
+  }
+
+  @Override
+  public void setPerConnectionStatsEnabled(boolean enabled) {
+    perConnectionStatsEnabled = enabled;
+  }
+
+  protected void recordReadStats(int size) {
     stats.addRead(size);
   }
 
-  protected void addWriteAmount(int size) {
+  protected void recordWriteStats(int size) {
     stats.addWrite(size);
   }
 
@@ -74,7 +87,7 @@ protected void checkRunning() {
   @Override
   public TCPClient createTCPClient(final String host, final int port) throws IOException {
     checkRunning();
-    TCPClient tc = new TCPClient(this, host, port);
+    TCPClient tc = new TCPClient(this, host, port, perConnectionStatsEnabled);
     clients.put(((Client)tc).getChannel(), tc);
     return tc;
   }
@@ -83,7 +96,7 @@ public TCPClient createTCPClient(final String host, final int port) throws IOExc
   @Override
   public TCPClient createTCPClient(final SocketChannel sc) throws IOException {
     checkRunning();
-    final TCPClient tc = new TCPClient(this, sc);
+    final TCPClient tc = new TCPClient(this, sc, perConnectionStatsEnabled);
     clients.put(((Client)tc).getChannel(), tc);
     this.setClientOperations(tc);
     return tc;
@@ -173,10 +186,6 @@ public SimpleByteStats getStats() {
     return stats;
   }
 
-  protected SocketExecuterByteStats writeableStats() {
-    return stats;
-  }
-
   @Override
   public void watchFuture(final ListenableFuture<?> lf, final long delay) {
     dogCache.watch(lf, delay);