some more tweaks to configuration and such

core/core-channelmanager2/src/main/java/org/webpieces/nio/impl/cm/basic/BasChannelService.java

@@ -31,6 +31,8 @@ class BasChannelService implements ChannelManager {
 	private BackpressureConfig config;
 
 	BasChannelService(String threadName, JdkSelect apis, BufferPool pool, BackpressureConfig config) {
+		if(config == null)
+			throw new IllegalArgumentException("config must be supplied");
 		this.pool = pool;
 		this.config = config;
 		processor = new KeyProcessor(apis, pool);

http/http-backpressure-tests/src/test/java/org/webpieces/throughput/AsyncConfig.java

@@ -0,0 +1,53 @@
+package org.webpieces.throughput;
+
+import org.webpieces.nio.api.BackpressureConfig;
+
+public class AsyncConfig {
+
+	private Integer clientThreadCount;
+	private Integer serverThreadCount;
+	private boolean isHttps;
+	private int clientMaxConcurrentRequests;
+	private BackpressureConfig backpressureConfig;
+
+	public boolean isHttps() {
+		return isHttps;
+	}
+
+	public void setHttps(boolean isHttps) {
+		this.isHttps = isHttps;
+	}
+
+	public int getClientMaxConcurrentRequests() {
+		return clientMaxConcurrentRequests;
+	}
+
+	public void setHttp2ClientMaxConcurrentRequests(int clientMaxConcurrentRequests) {
+		this.clientMaxConcurrentRequests = clientMaxConcurrentRequests;
+	}
+
+	public void setBackPressureConfig(BackpressureConfig backpressureConfig) {
+		this.backpressureConfig = backpressureConfig;
+	}
+
+	public BackpressureConfig getBackpressureConfig() {
+		return backpressureConfig;
+	}
+
+	public Integer getClientThreadCount() {
+		return clientThreadCount;
+	}
+
+	public void setClientThreadCount(Integer clientThreadCount) {
+		this.clientThreadCount = clientThreadCount;
+	}
+
+	public Integer getServerThreadCount() {
+		return serverThreadCount;
+	}
+
+	public void setServerThreadCount(Integer serverThreadCount) {
+		this.serverThreadCount = serverThreadCount;
+	}
+
+}

http/http-backpressure-tests/src/test/java/org/webpieces/throughput/SingleSocketThroughput.java

@@ -1,11 +1,29 @@
 package org.webpieces.throughput;
 
 import org.webpieces.frontend2.api.Protocol;
+import org.webpieces.nio.api.BackpressureConfig;
 
 public class SingleSocketThroughput {
 
 	public static void main(String[] args) throws InterruptedException {
-		ThroughputEngine example = new ThroughputEngine();
+		//All of these are variables that can impact performance so we surface them here to play with
+		BackpressureConfig backpressureConfig = new BackpressureConfig();
+		//num unacked bytes (client did not ack yet) before we backpressure
+		backpressureConfig.setMaxBytes(8_192*8); 
+		//once backpressure is on, instead of flapping, we wait for client to catch up quite a bit(may increase performance in certain use cases)
+		//or at the very least create more fairness if one client is giving more requests than another
+		backpressureConfig.setStartReadingThreshold(8_192*2); 
+
+		AsyncConfig config = new AsyncConfig();
+		config.setClientThreadCount(null); //turns off mulithreaded but useless for this test!!! since there is only one socket
+		config.setServerThreadCount(null); //turns off mulithreaded but useless for this test!!! since there is only one socket
+		config.setHttps(false);
+		config.setBackPressureConfig(backpressureConfig);
+
+		//this setting only applies to http2 server/client pair...
+		config.setHttp2ClientMaxConcurrentRequests(200);
+
+		ThroughputEngine example = new ThroughputEngine(config);
 
 		//BIG NOTE: It is not fair to set multiThreaded=true BECAUSE in that case you would need
 		//many many sockets because the threadpool causes context switching BUT keeps the one socket

http/http-backpressure-tests/src/test/java/org/webpieces/throughput/ThroughputEngine.java

@@ -16,13 +16,18 @@
 
 public class ThroughputEngine {
 
+	private AsyncConfig config;
+
+	public ThroughputEngine(AsyncConfig config) {
+		this.config = config;
+	}
+
 	protected void start(Mode clientConfig, Mode svrConfig, Protocol protocol) throws InterruptedException {
-		boolean multiThreaded = false;
 		CompletableFuture<InetSocketAddress> future;
 		if(svrConfig == Mode.ASYNCHRONOUS) {
 			//The asynchronous server supports BOTH protocols and automatically ends up doing
 			//the protocol of the client...
-			ServerAsync svr = new ServerAsync(multiThreaded);
+			ServerAsync svr = new ServerAsync(config);
 			future = svr.start();
 		} else if(protocol == Protocol.HTTP11){
 			ServerHttp1_1Sync svr = new ServerHttp1_1Sync();
@@ -32,19 +37,19 @@ protected void start(Mode clientConfig, Mode svrConfig, Protocol protocol) throw
 			future = svr.start();
 		}
 
-		future.thenApply(addr -> runClient(addr, multiThreaded, clientConfig, protocol));
+		future.thenApply(addr -> runClient(addr, config, clientConfig, protocol));
 
 		synchronized(this) {
 			this.wait(); //wait forever
 		}
 	}
 
-	private Void runClient(InetSocketAddress svrAddress, boolean multiThreaded, Mode clientConfig, Protocol protocol) {
+	private Void runClient(InetSocketAddress svrAddress, AsyncConfig config, Mode clientConfig, Protocol protocol) {
 		Clients creator;
 		if(protocol == Protocol.HTTP11)
-			creator = new Http11Clients(multiThreaded);
+			creator = new Http11Clients(config);
 		else
-			creator = new Http2Clients(multiThreaded);
+			creator = new Http2Clients(config);
 
 		if(clientConfig == Mode.ASYNCHRONOUS) {
 			runAsyncClient(svrAddress, protocol, creator);

http/http-backpressure-tests/src/test/java/org/webpieces/throughput/client/Http11Clients.java

@@ -1,34 +1,32 @@
 package org.webpieces.throughput.client;
 
-import java.net.InetSocketAddress;
-
 import org.webpieces.data.api.BufferCreationPool;
 import org.webpieces.http2client.api.Http2Client;
 import org.webpieces.httpclientx.api.Http2to1_1ClientFactory;
-import org.webpieces.nio.api.BackpressureConfig;
 import org.webpieces.nio.api.ChannelManager;
 import org.webpieces.nio.api.ChannelManagerFactory;
+import org.webpieces.throughput.AsyncConfig;
 
 public class Http11Clients implements Clients {
 
-	private boolean multiThreaded;
+	private AsyncConfig config;
 
-	public Http11Clients(boolean multiThreaded) {
-		this.multiThreaded = multiThreaded;
+	public Http11Clients(AsyncConfig config) {
+		this.config = config;
 	}
 
 	@Override
 	public Http2Client createClient() {
-		if(!multiThreaded) {
-			BufferCreationPool pool = new BufferCreationPool();
-			ChannelManagerFactory factory = ChannelManagerFactory.createFactory();
-			ChannelManager chanMgr = factory.createSingleThreadedChanMgr("clientCmLoop", pool, new BackpressureConfig());
+		if(config.getClientThreadCount() != null)
+			return Http2to1_1ClientFactory.createHttpClient(config.getClientThreadCount(), config.getBackpressureConfig());
 
-			Http2Client client = Http2to1_1ClientFactory.createHttpClient(chanMgr, pool);
-			return client;
-		}
+		//single threaded version...
+		BufferCreationPool pool = new BufferCreationPool();
+		ChannelManagerFactory factory = ChannelManagerFactory.createFactory();
+		ChannelManager chanMgr = factory.createSingleThreadedChanMgr("clientCmLoop", pool, config.getBackpressureConfig());
 
-		return Http2to1_1ClientFactory.createHttpClient(20);
+		Http2Client client = Http2to1_1ClientFactory.createHttpClient(chanMgr, pool);
+		return client;
 	}
 
 	@Override

http/http-backpressure-tests/src/test/java/org/webpieces/throughput/client/Http2Clients.java

@@ -2,35 +2,46 @@
 
 import org.webpieces.data.api.BufferCreationPool;
 import org.webpieces.http2client.api.Http2Client;
+import org.webpieces.http2client.api.Http2ClientConfig;
 import org.webpieces.http2client.api.Http2ClientFactory;
-import org.webpieces.nio.api.BackpressureConfig;
 import org.webpieces.nio.api.ChannelManager;
 import org.webpieces.nio.api.ChannelManagerFactory;
+import org.webpieces.throughput.AsyncConfig;
 
 import com.webpieces.http2engine.api.client.Http2Config;
 
 public class Http2Clients implements Clients {
 
-	private boolean multiThreaded;
+	private AsyncConfig config;
+	private Http2Config http2Config;
 
-	public Http2Clients(boolean multiThreaded) {
-		this.multiThreaded = multiThreaded;
+	public Http2Clients(AsyncConfig config) {
+		this.config = config;
+		http2Config = new Http2Config();
+		http2Config.setInitialRemoteMaxConcurrent(config.getClientMaxConcurrentRequests());
 	}
 
 	@Override
 	public Http2Client createClient() {
-		if(multiThreaded)
-			return Http2ClientFactory.createHttpClient(20);
+		if(config.getClientThreadCount() != null)
+			return createMultiThreadedClient();
 
+		//single threaded version...
 		BufferCreationPool pool = new BufferCreationPool();
 		ChannelManagerFactory factory = ChannelManagerFactory.createFactory();
-		ChannelManager chanMgr = factory.createSingleThreadedChanMgr("clientCmLoop", pool, new BackpressureConfig());
-
-		Http2Config http2Config = new Http2Config();
-		http2Config.setInitialRemoteMaxConcurrent(Integer.MAX_VALUE);
+		ChannelManager chanMgr = factory.createSingleThreadedChanMgr("clientCmLoop", pool, config.getBackpressureConfig());
 		return Http2ClientFactory.createHttpClient(http2Config, chanMgr, pool);
 	}
 
+	private Http2Client createMultiThreadedClient() {
+		Http2ClientConfig clientConfig = new Http2ClientConfig();
+		clientConfig.setBackpressureConfig(config.getBackpressureConfig());
+		clientConfig.setHttp2Config(http2Config);
+		clientConfig.setNumThreads(config.getClientThreadCount());
+
+		return Http2ClientFactory.createHttpClient(clientConfig);
+	}
+
 	@Override
 	public SynchronousClient createSyncClient() {
 		return new Http2SynchronousClient();

http/http-backpressure-tests/src/test/java/org/webpieces/throughput/client/ResponseCounterListener.java

@@ -1,10 +1,7 @@
 package org.webpieces.throughput.client;
 
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicLong;
 
-import org.webpieces.util.logging.Logger;
-import org.webpieces.util.logging.LoggerFactory;
 import org.webpieces.util.time.MsgRateRecorder;
 
 import com.webpieces.hpack.api.dto.Http2Response;

http/http-backpressure-tests/src/test/java/org/webpieces/throughput/server/ServerAsync.java

@@ -8,56 +8,71 @@
 import org.webpieces.asyncserver.api.AsyncServerManager;
 import org.webpieces.asyncserver.api.AsyncServerMgrFactory;
 import org.webpieces.data.api.BufferCreationPool;
-import org.webpieces.frontend2.api.FrontendConfig;
+import org.webpieces.frontend2.api.FrontendMgrConfig;
 import org.webpieces.frontend2.api.HttpFrontendFactory;
 import org.webpieces.frontend2.api.HttpFrontendManager;
 import org.webpieces.frontend2.api.HttpServer;
-import org.webpieces.nio.api.BackpressureConfig;
+import org.webpieces.frontend2.api.HttpSvrConfig;
 import org.webpieces.nio.api.ChannelManager;
 import org.webpieces.nio.api.ChannelManagerFactory;
+import org.webpieces.throughput.AsyncConfig;
 import org.webpieces.util.logging.Logger;
 import org.webpieces.util.logging.LoggerFactory;
 
 import com.webpieces.http2engine.api.client.Http2Config;
 
 public class ServerAsync {
 	private static final Logger log = LoggerFactory.getLogger(ServerAsync.class);
+	private AsyncConfig config;
+	private Http2Config http2Config;
 
-	private boolean multiThreaded;
-
-	public ServerAsync(boolean multiThreaded) {
-		this.multiThreaded = multiThreaded;
+	public ServerAsync(AsyncConfig config) {
+		this.config = config;
+		http2Config = new Http2Config();
+		long max = config.getClientMaxConcurrentRequests();
+		http2Config.getLocalSettings().setMaxConcurrentStreams(max);
 	}
 
 	public CompletableFuture<InetSocketAddress> start() {
     	log.error("running ASYNC HTTP1.1 AND HTTP2 SERVER");
 
-		HttpServer server = createFrontend(multiThreaded);
+		HttpServer server = createFrontend();
 		CompletableFuture<Void> future = server.start();
 		return future.thenApply(v -> server.getUnderlyingChannel().getLocalAddress());
 	}
 
-	private HttpServer createFrontend(boolean multiThreaded) {
-		if(multiThreaded)
+	private HttpServer createFrontend() {
+		if(config.getServerThreadCount() != null) {
 			return createFrontendMultiThreaded();
+		}
 
+		log.info("Creating single threaded server");
 		BufferCreationPool pool = new BufferCreationPool();
 
 		ChannelManagerFactory factory = ChannelManagerFactory.createFactory();
-		ChannelManager chanMgr = factory.createSingleThreadedChanMgr("svrCmLoop", pool, new BackpressureConfig());
+		ChannelManager chanMgr = factory.createSingleThreadedChanMgr("svrCmLoop", pool, config.getBackpressureConfig());
 
 		AsyncServerManager svrMgr = AsyncServerMgrFactory.createAsyncServer(chanMgr);
 
-		Http2Config config = new Http2Config();
-		config.setInitialRemoteMaxConcurrent(Integer.MAX_VALUE);
-		HttpFrontendManager mgr = HttpFrontendFactory.createFrontEnd(svrMgr, pool, new BackpressureConfig(), config);
-		return mgr.createHttpServer(new FrontendConfig("asyncsvr"), new EchoListener());
+
+		HttpFrontendManager mgr = HttpFrontendFactory.createFrontEnd(svrMgr, pool, http2Config);
+		return mgr.createHttpServer(new HttpSvrConfig("asyncsvr"), new EchoListener());
 	}
 
 	private HttpServer createFrontendMultiThreaded() {
 		ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor();
-		HttpFrontendManager mgr = HttpFrontendFactory.createFrontEnd("deansvr", 20, timer, new BufferCreationPool(), new BackpressureConfig());
 
-		return mgr.createHttpServer(new FrontendConfig("asyncsvr"), new EchoListener());
+
+		FrontendMgrConfig frontendConfig = new FrontendMgrConfig();
+		frontendConfig.setHttp2Config(http2Config);
+		frontendConfig.setBackpressureConfig(config.getBackpressureConfig());
+		frontendConfig.setThreadPoolSize(config.getServerThreadCount());
+
+		log.info("Creating multithreaded server. threads="+frontendConfig.getThreadPoolSize());
+
+		HttpFrontendManager mgr = HttpFrontendFactory.createFrontEnd(
+				"deansvr", timer, new BufferCreationPool(), frontendConfig);
+
+		return mgr.createHttpServer(new HttpSvrConfig("asyncsvr"), new EchoListener());
 	}
 }

http/http-backpressure-tests/src/test/java/org/webpieces/throughput/server/ServerHttp1_1Sync.java

@@ -59,8 +59,6 @@ public ServerRunnable(ServerSocket server) {
 		}
 
 		public void runImpl() throws IOException {
-	        @SuppressWarnings("resource")
-
 	        Socket socket = server.accept();
 	        InputStream input = socket.getInputStream();
 	        OutputStream output = socket.getOutputStream();

http/http-frontend2/src/main/java/org/webpieces/frontend2/api/FrontendMgrConfig.java

@@ -0,0 +1,32 @@
+package org.webpieces.frontend2.api;
+
+import org.webpieces.nio.api.BackpressureConfig;
+
+import com.webpieces.http2engine.api.client.Http2Config;
+
+public class FrontendMgrConfig {
+	private int threadPoolSize = 20;
+	private BackpressureConfig backpressureConfig = new BackpressureConfig();
+	private Http2Config http2Config = new Http2Config();
+
+	public int getThreadPoolSize() {
+		return threadPoolSize;
+	}
+	public void setThreadPoolSize(int threadPoolSize) {
+		this.threadPoolSize = threadPoolSize;
+	}
+	public BackpressureConfig getBackpressureConfig() {
+		return backpressureConfig;
+	}
+	public void setBackpressureConfig(BackpressureConfig config) {
+		this.backpressureConfig = config;
+	}
+	public Http2Config getHttp2Config() {
+		return http2Config;
+	}
+	public void setHttp2Config(Http2Config http2Config) {
+		this.http2Config = http2Config;
+	}
+
+
+}