Reduce Memory usage, Test server may be paging when more than 16G is allocated (#4200)

* Adding GreenLightning

* fix missing quote

create object for json production

* turn off telemetry for default test

* allow any external domain or ip

* Update to next version and template construction

* Revert "Update to next version and template construction"

This reverts commit da4fbc3b085f6e3b3bb50283d4d6e79efffdd3bb.

* Revert "Revert "Update to next version and template construction""

This reverts commit eb05517a192554caec7c5f690eeaaab171a549b6.

* Revert "Update to next version and template construction"

This reverts commit da4fbc3b085f6e3b3bb50283d4d6e79efffdd3bb.

* fix host ip to 0.0.0.0 to take load

added database read tests

* response large enough for multi db

remove epoll, it was not helping

added required headers

* update list of which tests we have implmented

* removed unused dependency

* remove unrequired 16G memory grab to get past travis check.

* Update to next GL version to fix overload issue past 1.5M rps

Refine template method to remove dead argument

* simplify arg parse

fixed issue with multi db under heavy load

* updat to atomic int

* Removed support for MultiTest, not stable at this time.

* added clean for safety

* remove dead code

* remove old comments

* [ci fw-only Java/greenlightning]

added comment

* [ci fw-only Java/greenlightning]

Added tests for remaining multi, update and fortunes

* [ci fw-only Java/greenlightning]

disable update test, seems to be missing some writes.

* [ci fw-only Java/greenlightning]

re-test of DBUpdate

* [ci fw-only Java/greenlightning]

disabled multi and update while tracking issue

* fixed muti paylod response JSON dups.

* [ci fw-only Java/greenlightning]

narrow building

* [ci skip] update readme

* [ci fw-only Java/greenlightning]

Update to version supporting 0.0.0.0 wildcard host

* [ci fw-only Java/greenlightning]

explicit memory min/max setting instead of default

update to new maven for build

* lower required low end to 8G

* [ci fw-only Java/greenlightning]

dropped min to 6G

* [ci fw-only Java/greenlightning]

memory reduction, was using as much as 20G, paging may have slowed test

reduce pipe memory allocations

combined behaviors to reduce memory

reduce db inflight collection to reduce memory

* Revert "[ci fw-only Java/greenlightning]"

This reverts commit 47c351db1ddeb64e65801f15e3c4398d5eca84ad.

* [ci fw-only Java/greenlightning]

Was using 16-20G which may cause paging and slow results

Reduced Pipe lenghts

Reduced concurrent connections

Combined behaviors to reduce Pipe counts for more memory.

Lowered limit to 16G to know it is enforced.

* removed minimum required memory in docker file

* [ci fw-only Java/greenlightning]  Ready for merge

frameworks/Java/greenlightning/greenlightning.dockerfile

@@ -5,7 +5,7 @@ COPY pom.xml pom.xml
 COPY src src
 RUN mvn clean install -q
 
-FROM openjdk:10-jre-slim
+FROM openjdk:10-jdk-slim
 WORKDIR /greenlightning
 COPY --from=maven /greenlightning/target/greenlightning-test.jar app.jar
-CMD ["java", "-server", "-Xms6g", "-Xmx30g", "-XX:+UseNUMA", "-XX:+UseParallelGC", "-XX:+AggressiveOpts", "-jar", "app.jar"]
+CMD ["java", "-server", "-Xmx16g", "-XX:+UseNUMA", "-XX:+UseParallelGC", "-XX:+AggressiveOpts", "-jar", "app.jar"]

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/DBRest.java

@@ -10,14 +10,15 @@ import com.ociweb.gl.api.HTTPResponseService;
 import com.ociweb.gl.api.PubSubMethodListener;
 import com.ociweb.gl.api.RestMethodListener;
 import com.ociweb.gl.api.TickListener;
-import com.ociweb.json.encode.JSONRenderer;
 import com.ociweb.pronghorn.network.config.HTTPContentTypeDefaults;
 import com.ociweb.pronghorn.pipe.ObjectPipe;
+import com.ociweb.pronghorn.util.AppendableBuilder;
 
 import io.reactiverse.pgclient.PgClient;
 import io.reactiverse.pgclient.PgIterator;
 import io.reactiverse.pgclient.PgPool;
 import io.reactiverse.pgclient.PgPoolOptions;
+import io.reactiverse.pgclient.Row;
 import io.reactiverse.pgclient.Tuple;
 
 public class DBRest implements RestMethodListener, PubSubMethodListener, TickListener {
@@ -25,12 +26,25 @@ public class DBRest implements RestMethodListener, PubSubMethodListener, TickLis
 	private final transient PgPoolOptions options;
 	private transient PgPool pool;
 	private final ThreadLocalRandom localRandom = ThreadLocalRandom.current();
-	private final ObjectPipe<ResultObject> inFlight;
-		
+	
+	private final ObjectPipe<ResultObject> DBRestInFlight;
+	private boolean collectionPendingDBRest = false;
+
+	//this collector is for the multi db test so we can collect all the objects until we have them all for 
+	//the request we are currently sending back
+	private transient final List<ResultObject> collectorDBRest = new ArrayList<ResultObject>();
+	private final HTTPResponseService service;
+
+
 	public DBRest(GreenRuntime runtime, PgPoolOptions options, int pipelineBits, int maxResponseCount, int maxResponseSize) {
-		this.options = options;	
-		this.inFlight = new ObjectPipe<ResultObject>(pipelineBits, ResultObject.class,	ResultObject::new);
+		
+		this.options = options;		
 		this.service = runtime.newCommandChannel().newHTTPResponseService(maxResponseCount, maxResponseSize);
+		
+		this.DBRestInFlight = new ObjectPipe<ResultObject>(pipelineBits, ResultObject.class,	ResultObject::new);
+		this.DBUpdateInFlight = new ObjectPipe<ResultObject>(pipelineBits, ResultObject.class,	ResultObject::new);
+		this.fortuneInFlight = new ObjectPipe<FortunesObject>(pipelineBits, FortunesObject.class,	FortunesObject::new);
+		
 	}		
 	
 	private PgPool pool() {
@@ -54,13 +68,13 @@ public class DBRest implements RestMethodListener, PubSubMethodListener, TickLis
 		}
 		
 	
-		if (inFlight.hasRoomFor(queries)) {
+		if (DBRestInFlight.hasRoomFor(queries)) {
 			
 			
 			int q = queries;
 			while (--q >= 0) {
 				
-					final ResultObject target = inFlight.headObject();
+					final ResultObject target = DBRestInFlight.headObject();
 					
 					//already released but not published yet: TODO: we have a problem here!!!
 					assert(null!=target && -1==target.getStatus()) : "found status "+target.getStatus()+" on query "+q+" of "+queries ; //must block that this has been consumed?? should head/tail rsolve.
@@ -87,7 +101,7 @@ public class DBRest implements RestMethodListener, PubSubMethodListener, TickLis
 							}				
 						});	
 								
-					inFlight.moveHeadForward(); //always move to ensure this can be read.
+					DBRestInFlight.moveHeadForward(); //always move to ensure this can be read.
 			
 			}
 				
@@ -102,7 +116,7 @@ public class DBRest implements RestMethodListener, PubSubMethodListener, TickLis
 	
 	public boolean singleRestRequest(HTTPRequestReader request) { 
 
-		final ResultObject target = inFlight.headObject();
+		final ResultObject target = DBRestInFlight.headObject();
 		if (null!=target && -1==target.getStatus()) {
 			target.setConnectionId(request.getConnectionId());
 			target.setSequenceId(request.getSequenceCode());
@@ -127,7 +141,7 @@ public class DBRest implements RestMethodListener, PubSubMethodListener, TickLis
 				});
 
 			
-			inFlight.moveHeadForward(); //always move to ensure this can be read.
+			DBRestInFlight.moveHeadForward(); //always move to ensure this can be read.
 			return true;
 		} else {
 			return false;//can not pick up new work now			
@@ -139,46 +153,51 @@ public class DBRest implements RestMethodListener, PubSubMethodListener, TickLis
 	////////////////////////////////////
 	////////////////////////////////////
 	
-	private final JSONRenderer<List<ResultObject>> multiTemplate = new JSONRenderer<List<ResultObject>>()
-	    	  .array((o,i) -> i<o.size()?o:null)
-		          .startObject((o, i) -> o.get(i))
-					.integer("id", o -> o.getId() )
-					.integer("randomNumber", o -> o.getResult())
-		          .endObject();
-	
-	private final JSONRenderer<ResultObject> singleTemplate = new JSONRenderer<ResultObject>()
-		   	  .startObject()
-				.integer("id", o -> o.getId() )
-				.integer("randomNumber", o -> o.getResult())
-	          .endObject();
-	
-	private boolean collectionPending = false;
-
-	//this collector is for the multi db test so we can collect all the objects until we have them all for 
-	//the request we are currently sending back
-	private final List<ResultObject> collector = new ArrayList<ResultObject>();
-	private final HTTPResponseService service;
-
-
 	@Override
 	public void tickEvent() { 
-		
-		ResultObject temp = inFlight.tailObject();
-		while (isReady(temp)) {			
-			if (consumeResultObject(temp)) {
-				temp = inFlight.tailObject();
-			} else {
-				break;
+		//for DBRest
+		{
+			ResultObject temp = DBRestInFlight.tailObject();
+			while (isReadyDBRest(temp)) {			
+				if (consumeResultObjectDBRest(temp)) {
+					temp = DBRestInFlight.tailObject();
+				} else {
+					break;
+				}
 			}
-		}	   
+		}
+		
+		//forDBUpdate
+		{
+			ResultObject temp = DBUpdateInFlight.tailObject();
+			while (isReady(temp)) {			
+				if (consumeResultObject(temp)) {
+					temp = DBUpdateInFlight.tailObject();
+				} else {
+					break;
+				}
+			}	   
+		}
+		
+		//for fortune
+		{
+			FortunesObject temp = fortuneInFlight.tailObject();
+			while (isReadyFortune(temp)) {			
+				if (consumeResultObjectFortune(temp)) {
+					temp = fortuneInFlight.tailObject();
+				} else {
+					break;
+				}
+			}		
+		}
 		
 	}
 
-	private boolean isReady(ResultObject temp) {
+	private boolean isReadyDBRest(ResultObject temp) {
 
-		if (collectionPending) {
+		if (collectionPendingDBRest) {
 			//now ready to send, we have all the data	
-			if (!publishMultiResponse(collector.get(0).getConnectionId(), collector.get(0).getSequenceId() )) {				
+			if (!publishMultiDBResponse(collectorDBRest.get(0).getConnectionId(), collectorDBRest.get(0).getSequenceId() )) {				
 				return false;
 			}
 		}
@@ -186,7 +205,7 @@ public class DBRest implements RestMethodListener, PubSubMethodListener, TickLis
 		return null!=temp && temp.getStatus()>=0;
 	}
 
-	private boolean consumeResultObject(final ResultObject t) {
+	private boolean consumeResultObjectDBRest(final ResultObject t) {
 		boolean ok;
 						
 		///////////////////////////////
@@ -194,20 +213,20 @@ public class DBRest implements RestMethodListener, PubSubMethodListener, TickLis
 			ok = service.publishHTTPResponse(t.getConnectionId(), t.getSequenceId(), 200,
 				   HTTPContentTypeDefaults.JSON,
 				   w-> {
-					   singleTemplate.render(w, t);
+					   Templates.singleTemplateDBRest.render(w, t);
 					   t.setStatus(-1);
-					   inFlight.moveTailForward();//only move forward when it is consumed.
-					   inFlight.publishTailPosition();
+					   DBRestInFlight.moveTailForward();//only move forward when it is consumed.
+					   DBRestInFlight.publishTailPosition();
 
 				   });					
 		} else {
 			//collect all the objects
-			assert(isValidToAdd(t, collector));
-			collector.add(t);					
-			inFlight.moveTailForward();
-			if (collector.size() == t.getGroupSize()) {
+			assert(isValidToAdd(t, collectorDBRest));
+			collectorDBRest.add(t);					
+			DBRestInFlight.moveTailForward();
+			if (collectorDBRest.size() == t.getGroupSize()) {
 				//now ready to send, we have all the data						
-				ok =publishMultiResponse(t.getConnectionId(), t.getSequenceId());
+				ok =publishMultiDBResponse(t.getConnectionId(), t.getSequenceId());
 				
 			} else {
 				ok = true;//added to list
@@ -233,25 +252,25 @@ public class DBRest implements RestMethodListener, PubSubMethodListener, TickLis
 		return true;
 	}
 
-	private boolean publishMultiResponse(long conId, long seqCode) {
+	private boolean publishMultiDBResponse(long conId, long seqCode) {
 		final boolean result =  service.publishHTTPResponse(conId, seqCode, 200,
 					    				   HTTPContentTypeDefaults.JSON,
 					    				   w-> {
-					    					   multiTemplate.render(w, collector);
+					    					   Templates.multiTemplateDBRest.render(w, collectorDBRest);
 					    					   
-					    					   int c = collector.size();
-					    					   assert(collector.get(0).getGroupSize()==c);
+					    					   int c = collectorDBRest.size();
+					    					   assert(collectorDBRest.get(0).getGroupSize()==c);
 					    					   while (--c >= 0) {
-					    						   assert(collector.get(0).getGroupSize()==collector.size());
-					    						   assert(collector.get(c).getConnectionId() == conId) : c+" expected conId "+conId+" error: "+showCollection(collector);
-					    						   assert(collector.get(c).getSequenceId() == seqCode) : c+" sequence error: "+showCollection(collector);    						   
-					    						   collector.get(c).setStatus(-1);
+					    						   assert(collectorDBRest.get(0).getGroupSize()==collectorDBRest.size());
+					    						   assert(collectorDBRest.get(c).getConnectionId() == conId) : c+" expected conId "+conId+" error: "+showCollection(collectorDBRest);
+					    						   assert(collectorDBRest.get(c).getSequenceId() == seqCode) : c+" sequence error: "+showCollection(collectorDBRest);    						   
+					    						   collectorDBRest.get(c).setStatus(-1);
 					    						 
 					    					   }
-					    					   collector.clear();					    					   
-					    					   inFlight.publishTailPosition();
+					    					   collectorDBRest.clear();					    					   
+					    					   DBRestInFlight.publishTailPosition();
 					    				   });
-		collectionPending = !result;
+		collectionPendingDBRest = !result;
 		return result;
 	}
 
@@ -269,6 +288,249 @@ public class DBRest implements RestMethodListener, PubSubMethodListener, TickLis
 		return builder.toString();
 	}
 	
+	///////////////////////////////////////////////////////////////////////////
+	/////////////////////////////////////////
+	///DB Update
+	/////////////////////////////////////////
+	///////////////////////////////////////////////////////////////////////////
+
+	private ObjectPipe<ResultObject> DBUpdateInFlight;	
+	private boolean collectionPendingDBUpdate = false;	
+	private final List<ResultObject> collectorDBUpdate = new ArrayList<ResultObject>();
+
+	public boolean updateRestRequest(HTTPRequestReader request) {
+		int queries;
+		if (Struct.UPDATES_ROUTE_INT == request.getRouteAssoc() ) {		
+			queries = Math.min(Math.max(1, (request.structured().readInt(Field.QUERIES))),500);		
+		} else {
+			queries = 1;
+		}
+		long conId = request.getConnectionId();
+		long seqCode = request.getSequenceCode();
+
+		if (DBUpdateInFlight.hasRoomFor(queries)) {		
+				    	
+				int q = queries;
+				while (--q >= 0) {
+				
+						final ResultObject worldObject = DBUpdateInFlight.headObject();
+						assert(null!=worldObject);
+											
+						worldObject.setConnectionId(conId);
+						worldObject.setSequenceId(seqCode);
+						worldObject.setStatus(-2);//out for work	
+						worldObject.setGroupSize(queries);
+						
+						worldObject.setId(randomValue());
+												
+						pool().preparedQuery("SELECT * FROM world WHERE id=$1", Tuple.of(worldObject.getId()), r -> {
+								if (r.succeeded()) {
+																		
+									PgIterator resultSet = r.result().iterator();
+							        Tuple row = resultSet.next();			        
+							        
+							        assert(worldObject.getId()==row.getInteger(0));
+							        
+							        //read the existing random value and store it in the world object
+							        worldObject.setResult(row.getInteger(1));
+							        
+							        ///////////////////////////////////
+							        //set the new random value in this object
+							        worldObject.setResult(randomValue());
+							        							       
+							        
+							        pool().preparedQuery("UPDATE world SET randomnumber=$1 WHERE id=$2", 							        		
+							        			Tuple.of(worldObject.getResult(), worldObject.getId()), ar -> {							        	
+										if (ar.succeeded()) {
+											
+								        	worldObject.setStatus(200);							
+								        	
+										} else {	
+											System.out.println("unable to update");
+											if (ar.cause()!=null) {
+												ar.cause().printStackTrace();
+											}
+											
+											worldObject.setStatus(500);
+										}	
+																													
+							        });
+								} else {	
+									System.out.println("unable to query");
+									if (r.cause()!=null) {
+										r.cause().printStackTrace();
+									}
+									
+									worldObject.setStatus(500);
+								}		
+								
+								
+							});	
+									
+						DBUpdateInFlight.moveHeadForward(); //always move to ensure this can be read.
+				
+				}
+				
+			return true;
+		} else {
+			return false;
+		}
+	}
+
+	private boolean isReady(ResultObject temp) {
+
+		if (collectionPendingDBUpdate) {
+			//now ready to send, we have all the data	
+			if (!publishMultiResponse(collectorDBUpdate.get(0).getConnectionId(), collectorDBUpdate.get(0).getSequenceId() )) {
+				return false;
+			}
+		}
+		
+		return null!=temp && temp.getStatus()>=0;
+	}
+
+	private boolean consumeResultObject(final ResultObject t) {
+		boolean ok;
+		//collect all the objects
+		collectorDBUpdate.add(t);
+		DBUpdateInFlight.moveTailForward();//only move forward when it is consumed.
+		if (collectorDBUpdate.size() == t.getGroupSize()) {
+			//now ready to send, we have all the data						
+			ok =publishMultiResponse(t.getConnectionId(), t.getSequenceId());
+		} else {
+			ok = true;//added to list
+		}				
+		
+		return ok;
+	}
+
+	private boolean publishMultiResponse(long conId, long seqCode) {
+		boolean result =  service.publishHTTPResponse(conId, seqCode, 200,
+					    				   HTTPContentTypeDefaults.JSON,
+					    				   w-> {
+					    					   Templates.multiTemplate.render(w, collectorDBUpdate);
+					    					   int c = collectorDBUpdate.size();
+					    					   while (--c>=0) {
+					    						   assert(collectorDBUpdate.get(c).getConnectionId() == conId);
+					    						   assert(collectorDBUpdate.get(c).getSequenceId() == seqCode);					    						   
+					    						   collectorDBUpdate.get(c).setStatus(-1);
+					    					   }
+					    					   collectorDBUpdate.clear();
+					    					   DBUpdateInFlight.publishTailPosition();
+					    				   });
+		collectionPendingDBUpdate = !result;
+		return result;
+	}
+	
+	//////////////////////////////////////////////////////////////////////////
+	//////////////////////////////////////////
+	//Fortune
+	/////////////////////////////////////////////
+	//////////////////////////////////////////////////////////////////////////
+	
+
+	
+	//SQL results write to these object, these same objects are used by template
+	private transient ObjectPipe<FortunesObject> fortuneInFlight;
+	
+	public boolean restFortuneRequest(HTTPRequestReader request) {
+	
+		final FortunesObject target = fortuneInFlight.headObject(); 
+		if (null!=target) {
+			target.setConnectionId(request.getConnectionId());
+			target.setSequenceId(request.getSequenceCode());
+	
+			target.setStatus(-2);//out for work	
+			target.clear();
+		
+			pool().preparedQuery( "SELECT id, message FROM fortune", r -> {
+				    //NOTE: we want to do as little work here a s possible since
+				    //      we want this thread to get back to work on other calls.
+					if (r.succeeded()) {
+						PgIterator resultSet = r.result().iterator();						
+						while (	resultSet.hasNext() ) {
+					        Row next = resultSet.next();
+							target.addFortune(next.getInteger(0), next.getString(1));						
+						}
+						target.setStatus(200);
+					} else {
+						System.out.println("fail: "+r.cause().getLocalizedMessage());
+						target.setStatus(500);
+					}		
+					
+				});
+			
+			fortuneInFlight.moveHeadForward(); //always move to ensure this can be read.  //TODO: remove and combined with above
+			return true;
+		} else {
+			return false;//can not pick up new work now			
+		}		
+	}
+
+	private boolean isReadyFortune(FortunesObject temp) {
+		return null!=temp && temp.getStatus()>=0;
+	}
+
+	private int htmlFortunePos=0;
+	private final transient AppendableBuilder htmlFortuneBuffer = new AppendableBuilder();
+	
+	private boolean consumeResultObjectFortune(final FortunesObject t) {
+					
+		if (0 == htmlFortuneBuffer.byteLength()) {
+			//capture all the output text
+			t.addFortune(0, "Additional fortune added at request time.");
+			t.sort();
+			Templates.fortuneTemplate.render(htmlFortuneBuffer, t);
+			htmlFortunePos = 0;
+		}
+		
+		
+		int bytesRemaining = htmlFortuneBuffer.byteLength() - htmlFortunePos;
+		int roomForWrite = service.maxVarLength();
+		boolean hasContinuation  = bytesRemaining >roomForWrite;
+		
+		//as long as htmlPos does not match the total bytes of the payload keep 
+		//sending out continuation chunks. We do not know how many rows of fortunes
+		//may be in the database.
+		boolean ok;
+		if (0 == htmlFortunePos) {	
+			
+			ok = service.publishHTTPResponse(t.getConnectionId(), t.getSequenceId(), 200, hasContinuation,
+						   HTTPContentTypeDefaults.HTML, 
+						   w-> {
+							   htmlFortunePos += htmlFortuneBuffer.copyTo(w, htmlFortunePos);								   
+							   assert(hasContinuation == (htmlFortunePos!=htmlFortuneBuffer.byteLength())) : "internal error";
+							   
+						   });
+		} else {		
+			ok =service.publishHTTPResponseContinuation(t.getConnectionId(), t.getSequenceId(), hasContinuation,  
+							w-> {
+								htmlFortunePos += htmlFortuneBuffer.copyTo(w,htmlFortunePos);	
+								assert(hasContinuation == (htmlFortunePos!=htmlFortuneBuffer.byteLength())) : "internal error";
+								
+							});
+		}
+		
+		if (ok) {
+			if (htmlFortunePos == htmlFortuneBuffer.byteLength()) {
+				
+				t.setStatus(-1);
+				fortuneInFlight.moveTailForward();//only move forward when it is consumed.
+				fortuneInFlight.publishTailPosition();
+				t.list().clear();
+				htmlFortuneBuffer.clear();
+				return true;//do consume this since it is now fully sent
+			} else {
+				assert(htmlFortunePos < htmlFortuneBuffer.byteLength()) : "internal error";			
+				return false;//still have more to send later
+			}	
+		} else {
+			return false;
+		}		
+		
+	}
+	
+	
 	
 }
 

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/DBUpdate.java

@@ -1,200 +0,0 @@
-package com.ociweb.gl.benchmark;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.ThreadLocalRandom;
-
-import com.ociweb.gl.api.GreenRuntime;
-import com.ociweb.gl.api.HTTPRequestReader;
-import com.ociweb.gl.api.HTTPResponseService;
-import com.ociweb.gl.api.RestMethodListener;
-import com.ociweb.gl.api.TickListener;
-import com.ociweb.json.encode.JSONRenderer;
-import com.ociweb.pronghorn.network.config.HTTPContentTypeDefaults;
-import com.ociweb.pronghorn.pipe.ObjectPipe;
-
-import io.reactiverse.pgclient.PgClient;
-import io.reactiverse.pgclient.PgIterator;
-import io.reactiverse.pgclient.PgPool;
-import io.reactiverse.pgclient.PgPoolOptions;
-import io.reactiverse.pgclient.Tuple;
-
-public class DBUpdate implements RestMethodListener, TickListener {
-
-
-	private final transient PgPoolOptions options;
-	private transient PgPool pool;
-	
-	private final HTTPResponseService service;
-	private ObjectPipe<ResultObject> inFlight;	
-	private boolean collectionPending = false;	
-	private final List<ResultObject> collector = new ArrayList<ResultObject>();
-	
-	private static final ThreadLocalRandom localRandom = ThreadLocalRandom.current();
-	private static final JSONRenderer<List<ResultObject>> multiTemplate = new JSONRenderer<List<ResultObject>>()
-	    	  .array((o,i) -> i<o.size()?o:null)
-		          .startObject((o, i) ->  o.get(i))  
-					.integer("id", o -> o.getId() )
-					.integer("randomNumber", o -> o.getResult())
-		          .endObject();
-	
-	public DBUpdate(GreenRuntime runtime, PgPoolOptions options, int pipelineBits, int maxResponseCount, int maxResponseSize) {
-		this.options = options;
-		this.options.setMaxSize(6);//bump up the connections since we use the pool nested twice		
-		this.service = runtime.newCommandChannel().newHTTPResponseService(maxResponseCount, maxResponseSize);
-		this.inFlight = new ObjectPipe<ResultObject>(pipelineBits, ResultObject.class,	ResultObject::new);
-	}
-	
-	private PgPool pool() {
-		if (null==pool) {
-			pool = PgClient.pool(options);
-		}
-		return pool;
-	}
-	
-	private int randomValue() {
-		return 1+localRandom.nextInt(10000);
-	}	
-	
-	public boolean updateRestRequest(HTTPRequestReader request) {
-		int queries;
-		if (Struct.UPDATES_ROUTE_INT == request.getRouteAssoc() ) {		
-			queries = Math.min(Math.max(1, (request.structured().readInt(Field.QUERIES))),500);		
-		} else {
-			queries = 1;
-		}
-		long conId = request.getConnectionId();
-		long seqCode = request.getSequenceCode();
-
-		if (inFlight.hasRoomFor(queries)) {		
-				    	
-				int q = queries;
-				while (--q >= 0) {
-				
-						final ResultObject worldObject = inFlight.headObject();
-						assert(null!=worldObject);
-											
-						worldObject.setConnectionId(conId);
-						worldObject.setSequenceId(seqCode);
-						worldObject.setStatus(-2);//out for work	
-						worldObject.setGroupSize(queries);
-						
-						worldObject.setId(randomValue());
-												
-						pool().preparedQuery("SELECT * FROM world WHERE id=$1", Tuple.of(worldObject.getId()), r -> {
-								if (r.succeeded()) {
-																		
-									PgIterator resultSet = r.result().iterator();
-							        Tuple row = resultSet.next();			        
-							        
-							        assert(worldObject.getId()==row.getInteger(0));
-							        
-							        //read the existing random value and store it in the world object
-							        worldObject.setResult(row.getInteger(1));
-							        
-							        ///////////////////////////////////
-							        //set the new random value in this object
-							        worldObject.setResult(randomValue());
-							        							       
-							        
-							        pool().preparedQuery("UPDATE world SET randomnumber=$1 WHERE id=$2", 							        		
-							        			Tuple.of(worldObject.getResult(), worldObject.getId()), ar -> {							        	
-										if (ar.succeeded()) {
-											
-								        	worldObject.setStatus(200);							
-								        	
-										} else {	
-											System.out.println("unable to update");
-											if (ar.cause()!=null) {
-												ar.cause().printStackTrace();
-											}
-											
-											worldObject.setStatus(500);
-										}	
-																													
-							        });
-								} else {	
-									System.out.println("unable to query");
-									if (r.cause()!=null) {
-										r.cause().printStackTrace();
-									}
-									
-									worldObject.setStatus(500);
-								}		
-								
-								
-							});	
-									
-						inFlight.moveHeadForward(); //always move to ensure this can be read.
-				
-				}
-				
-			return true;
-		} else {
-			return false;
-		}
-	}
-	
-	
-	@Override
-	public void tickEvent() { 
-		
-		ResultObject temp = inFlight.tailObject();
-		while (isReady(temp)) {			
-			if (consumeResultObject(temp)) {
-				temp = inFlight.tailObject();
-			} else {
-				break;
-			}
-		}	   
-		
-	}
-
-	private boolean isReady(ResultObject temp) {
-
-		if (collectionPending) {
-			//now ready to send, we have all the data	
-			if (!publishMultiResponse(collector.get(0).getConnectionId(), collector.get(0).getSequenceId() )) {
-				return false;
-			}
-		}
-		
-		return null!=temp && temp.getStatus()>=0;
-	}
-
-	private boolean consumeResultObject(final ResultObject t) {
-		boolean ok;
-		//collect all the objects
-		collector.add(t);
-		inFlight.moveTailForward();//only move forward when it is consumed.
-		if (collector.size() == t.getGroupSize()) {
-			//now ready to send, we have all the data						
-			ok =publishMultiResponse(t.getConnectionId(), t.getSequenceId());
-		} else {
-			ok = true;//added to list
-		}				
-		
-		return ok;
-	}
-
-	private boolean publishMultiResponse(long conId, long seqCode) {
-		boolean result =  service.publishHTTPResponse(conId, seqCode, 200,
-					    				   HTTPContentTypeDefaults.JSON,
-					    				   w-> {
-					    					   multiTemplate.render(w, collector);
-					    					   int c = collector.size();
-					    					   while (--c>=0) {
-					    						   assert(collector.get(c).getConnectionId() == conId);
-					    						   assert(collector.get(c).getSequenceId() == seqCode);					    						   
-					    						   collector.get(c).setStatus(-1);
-					    					   }
-					    					   collector.clear();
-					    					   inFlight.publishTailPosition();
-					    				   });
-		collectionPending = !result;
-		return result;
-	}
-	
-	
-	
-}

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/FortuneRest.java

@@ -1,183 +0,0 @@
-package com.ociweb.gl.benchmark;
-
-import com.ociweb.gl.api.GreenRuntime;
-import com.ociweb.gl.api.HTTPRequestReader;
-import com.ociweb.gl.api.HTTPResponseService;
-import com.ociweb.gl.api.RestMethodListener;
-import com.ociweb.gl.api.TickListener;
-import com.ociweb.pronghorn.network.config.HTTPContentTypeDefaults;
-import com.ociweb.pronghorn.pipe.ObjectPipe;
-import com.ociweb.pronghorn.util.AppendableBuilder;
-import com.ociweb.pronghorn.util.Appendables;
-import com.ociweb.pronghorn.util.template.StringTemplateBuilder;
-import com.ociweb.pronghorn.util.template.StringTemplateRenderer;
-
-import io.reactiverse.pgclient.PgClient;
-import io.reactiverse.pgclient.PgIterator;
-import io.reactiverse.pgclient.PgPool;
-import io.reactiverse.pgclient.PgPoolOptions;
-import io.reactiverse.pgclient.Row;
-
-public class FortuneRest implements RestMethodListener, TickListener {
-
-	private static final byte[] ROW_FINISH = "</td></tr>\n".getBytes();
-	private static final byte[] ROW_MIDDLE = "</td><td>".getBytes();
-	private static final byte[] ROW_START = "<tr><td>".getBytes();
-	private final HTTPResponseService service; 
-	
-	private final transient PgPoolOptions options;
-	private transient PgPool pool;
-			
-	//SQL results write to these object, these same objects are used by template
-	private transient ObjectPipe<FortunesObject> inFlight;
-	
-	private static final transient StringTemplateRenderer<FortunesObject> template =		
-			new StringTemplateBuilder<FortunesObject>()
-				   .add("<!DOCTYPE html> <html> <head><title>Fortunes</title></head> <body> <table> <tr><th>id</th><th>message</th></tr>\n")
-			       .add((t,s,i)-> {
-						if (i<s.list().size()) {													
-							t.write(ROW_START);
-							Appendables.appendValue(t, s.list().get(i).getId());
-							t.write(ROW_MIDDLE);							
-							Appendables.appendHTMLEntityEscaped(t, s.list().get(i).getFortune());							
-							t.write(ROW_FINISH);
-							return true;
-						} else {
-							return false;
-						}
-			         })		
-			       .add("</table></body></html>")
-			       .finish();
-
-	
-	public FortuneRest(GreenRuntime runtime, PgPoolOptions options, int pipelineBits, int responseCount, int maxResponseSize) {;
-	    
-		this.options = options;	
-		this.service = runtime.newCommandChannel().newHTTPResponseService(responseCount, maxResponseSize);
-		this.inFlight =  new ObjectPipe<FortunesObject>(pipelineBits, FortunesObject.class,	FortunesObject::new);
-		
-	}
-	
-	
-	private PgPool pool() {
-		if (null==pool) {
-			pool = PgClient.pool(options);
-		}
-		return pool;
-	}
-
-	public boolean restRequest(HTTPRequestReader request) {
-	
-		final FortunesObject target = inFlight.headObject(); 
-		if (null!=target) {
-			target.setConnectionId(request.getConnectionId());
-			target.setSequenceId(request.getSequenceCode());
-	
-			target.setStatus(-2);//out for work	
-			target.clear();
-		
-			pool().preparedQuery( "SELECT id, message FROM fortune", r -> {
-				    //NOTE: we want to do as little work here a s possible since
-				    //      we want this thread to get back to work on other calls.
-					if (r.succeeded()) {
-						PgIterator resultSet = r.result().iterator();						
-						while (	resultSet.hasNext() ) {
-					        Row next = resultSet.next();
-							target.addFortune(next.getInteger(0), next.getString(1));						
-						}
-						target.setStatus(200);
-					} else {
-						System.out.println("fail: "+r.cause().getLocalizedMessage());
-						target.setStatus(500);
-					}		
-					
-				});
-			
-			inFlight.moveHeadForward(); //always move to ensure this can be read.  //TODO: remove and combined with above
-			return true;
-		} else {
-			return false;//can not pick up new work now			
-		}		
-	}
-	
-
-	
-	@Override
-	public void tickEvent() { //TODO: remove tickEvent here and replace with  pub sub to take next...
-		
-		FortunesObject temp = inFlight.tailObject();
-		while (isReady(temp)) {			
-			if (consumeResultObject(temp)) {
-				temp = inFlight.tailObject();
-			} else {
-				break;
-			}
-		}		
-	}
-	
-	private boolean isReady(FortunesObject temp) {
-		return null!=temp && temp.getStatus()>=0;
-	}
-
-	private int htmlPos=0;
-	private final transient AppendableBuilder htmlBuffer = new AppendableBuilder();
-	
-	
-	private boolean consumeResultObject(final FortunesObject t) {
-					
-		if (0 == htmlBuffer.byteLength()) {
-			//capture all the output text
-			t.addFortune(0, "Additional fortune added at request time.");
-			t.sort();
-			template.render(htmlBuffer, t);
-			htmlPos = 0;
-		}
-		
-		
-		int bytesRemaining = htmlBuffer.byteLength() - htmlPos;
-		int roomForWrite = service.maxVarLength();
-		boolean hasContinuation  = bytesRemaining >roomForWrite;
-		
-		//as long as htmlPos does not match the total bytes of the payload keep 
-		//sending out continuation chunks. We do not know how many rows of fortunes
-		//may be in the database.
-		boolean ok;
-		if (0 == htmlPos) {	
-			
-			ok = service.publishHTTPResponse(t.getConnectionId(), t.getSequenceId(), 200, hasContinuation,
-						   HTTPContentTypeDefaults.HTML, 
-						   w-> {
-							   htmlPos += htmlBuffer.copyTo(w, htmlPos);								   
-							   assert(hasContinuation == (htmlPos!=htmlBuffer.byteLength())) : "internal error";
-							   
-						   });
-		} else {		
-			ok =service.publishHTTPResponseContinuation(t.getConnectionId(), t.getSequenceId(), hasContinuation,  
-							w-> {
-								htmlPos += htmlBuffer.copyTo(w,htmlPos);	
-								assert(hasContinuation == (htmlPos!=htmlBuffer.byteLength())) : "internal error";
-								
-							});
-		}
-		
-		if (ok) {
-			if (htmlPos == htmlBuffer.byteLength()) {
-				
-				t.setStatus(-1);
-				inFlight.moveTailForward();//only move forward when it is consumed.
-				inFlight.publishTailPosition();
-				t.list().clear();
-				htmlBuffer.clear();
-				return true;//do consume this since it is now fully sent
-			} else {
-				assert(htmlPos < htmlBuffer.byteLength()) : "internal error";			
-				return false;//still have more to send later
-			}	
-		} else {
-			return false;
-		}		
-		
-	}
-	
-
-}

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/FrameworkTest.java

@@ -32,11 +32,14 @@ public class FrameworkTest implements GreenApp {
     private int queueLengthOfPendingRequests;
     private int telemetryPort;//for monitoring
     private int minMemoryOfInputPipes;
-    private int maxResponseSize;
-	private	int maxResponseCount = 1<<8;
+    private int dbCallMaxResponseSize;
+	private	final int dbCallMaxResponseCount;
     private int pipelineBits;
 	
-    private final PgPoolOptions options;
+	private final int jsonMaxResponseCount;
+	private final int jsonMaxResponseSize;
+	
+    private PgPoolOptions options;
     
 	public static int connectionsPerTrack =   2;
 	public static int connectionPort =        5432;
@@ -53,9 +56,9 @@ public class FrameworkTest implements GreenApp {
     	//this server works best with  -XX:+UseNUMA    	
     	this(System.getProperty("host","0.0.0.0"), 
     		 8080,    //default port for test 
-    		 10,      //default concurrency, 10 to support 280 channels on 14 core boxes
-    		 8*1024, //default max rest requests allowed to queue in wait
-    		 1<<21,   //default network buffer per input socket connection
+    		 5,       //default concurrency, 5 to support 140 channels on 14 core boxes
+    		 2*1024,  //default max rest requests allowed to queue in wait
+    		 1<<20,   //default network buffer per input socket connection
     		 Integer.parseInt(System.getProperty("telemetry.port", "-1")),
     		 "tfb-database", // jdbc:postgresql://tfb-database:5432/hello_world
     		 "hello_world",
@@ -80,8 +83,13 @@ public class FrameworkTest implements GreenApp {
     	this.queueLengthOfPendingRequests = queueLengthOfPendingRequests;
     	this.minMemoryOfInputPipes = minMemoryOfInputPipes;
     	this.telemetryPort = telemetryPort;
-    	this.maxResponseSize = 20_000; //for 500 mult db call in JSON format
-    	this.pipelineBits = 19;
+    	this.pipelineBits = 17;//max concurrent in flight database requests 1<<pipelineBits
+
+    	this.dbCallMaxResponseCount = 1<<6;
+    	this.jsonMaxResponseCount = 1<<13;
+    	
+    	this.dbCallMaxResponseSize = 20_000; //for 500 mult db call in JSON format
+    	this.jsonMaxResponseSize = 1<<9;
     	
     	if (!"127.0.0.1".equals(System.getProperty("host",null))) { 
     		    		
@@ -99,26 +107,29 @@ public class FrameworkTest implements GreenApp {
 	    	}    	
     	}
     	
-	    	options = new PgPoolOptions()
-	    			.setPort(connectionPort)
-	    			.setPipeliningLimit(1<<pipelineBits)
-	    			.setTcpFastOpen(true)
-	    			.setHost(connectionHost)
-	    			.setDatabase(connectionDB)
-	    			.setUser(connectionUser)
-	    			.setPassword(connectionPassword)
-	    			.setCachePreparedStatements(true)
-	    			.setMaxSize(connectionsPerTrack);	    	
 	    		
     	try {
-	    	///early check to know if we have a database or not,
+    		options = new PgPoolOptions()
+    				.setPort(connectionPort)
+    				.setPipeliningLimit(1<<pipelineBits)
+    				.setTcpFastOpen(true)
+    				.setHost(connectionHost)
+    				.setDatabase(connectionDB)
+    				.setUser(connectionUser)
+    				.setPassword(connectionPassword)
+    				.setCachePreparedStatements(true)
+    				.setMaxSize(connectionsPerTrack);	    	
+
+    		///early check to know if we have a database or not,
 	    	///this helps testing to know which tests should be run on different boxes.
 	    	PgClient.pool(options).getConnection(a->{
 	    		foundDB.set(a.succeeded());
-	    		a.result().close();
+	    		if (null!=a.result()) {
+	    			a.result().close();
+	    		}
 	    	});
     	} catch (Throwable t) {
-    		t.printStackTrace();
+    		//t.printStackTrace();
     		System.out.println("No database in use");
     	}
     	
@@ -135,7 +146,7 @@ public class FrameworkTest implements GreenApp {
     			 .setConcurrentChannelsPerEncryptUnit(concurrentWritesPerChannel)
     			 .setMaxQueueIn(queueLengthOfPendingRequests)
     			 .setMinimumInputPipeMemory(minMemoryOfInputPipes)
-    			 .setMaxResponseSize(maxResponseSize) //big enough for large mult db response
+    			 .setMaxResponseSize(dbCallMaxResponseSize) //big enough for large mult db response
     	         .useInsecureServer(); //turn off TLS
 
 		framework.defineRoute()
@@ -183,30 +194,24 @@ public class FrameworkTest implements GreenApp {
 	
     }
 
+
 	public void parallelBehavior(GreenRuntime runtime) {
 
-		SimpleRest restTest = new SimpleRest(runtime);
-		
+
+		SimpleRest restTest = new SimpleRest(runtime, jsonMaxResponseCount, jsonMaxResponseSize);		
 		runtime.registerListener("Simple", restTest)
 		       .includeRoutes(Struct.PLAINTEXT_ROUTE, restTest::plainRestRequest)
 		       .includeRoutes(Struct.JSON_ROUTE, restTest::jsonRestRequest);
 		 
 
-		DBRest dbRestInstance = new DBRest(runtime, options, pipelineBits, maxResponseCount, maxResponseSize);
-		runtime.registerListener("DBRest", dbRestInstance)
+		DBRest dbRestInstance = new DBRest(runtime, options, pipelineBits, dbCallMaxResponseCount, dbCallMaxResponseSize);
+		runtime.registerListener("DBReadWrite", dbRestInstance)
 				.includeRoutes(Struct.DB_SINGLE_ROUTE, dbRestInstance::singleRestRequest)
 				.includeRoutes(Struct.DB_MULTI_ROUTE_TEXT, dbRestInstance::multiRestRequest)		
-		        .includeRoutes(Struct.DB_MULTI_ROUTE_INT, dbRestInstance::multiRestRequest);
-
-		
-		DBUpdate dbUpdateInstance = new DBUpdate(runtime, options, pipelineBits, maxResponseCount, maxResponseSize);
-		runtime.registerListener("DBUpdate", dbUpdateInstance)
-		        .includeRoutes(Struct.UPDATES_ROUTE_TEXT, dbUpdateInstance::updateRestRequest)
-		        .includeRoutes(Struct.UPDATES_ROUTE_INT,  dbUpdateInstance::updateRestRequest);
-		
-		FortuneRest fortuneInstance = new FortuneRest(runtime, options, pipelineBits, maxResponseCount, maxResponseSize);
-		runtime.registerListener("Fortune", fortuneInstance)
-		        .includeRoutes(Struct.FORTUNES_ROUTE, fortuneInstance::restRequest);	
+		        .includeRoutes(Struct.DB_MULTI_ROUTE_INT, dbRestInstance::multiRestRequest)
+				.includeRoutes(Struct.UPDATES_ROUTE_TEXT, dbRestInstance::updateRestRequest)
+				.includeRoutes(Struct.UPDATES_ROUTE_INT,  dbRestInstance::updateRestRequest)
+		        .includeRoutes(Struct.FORTUNES_ROUTE, dbRestInstance::restFortuneRequest);	
 		
 	}
 	 

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/GreenLightning.java

@@ -5,7 +5,7 @@ import com.ociweb.gl.api.GreenRuntime;
 public class GreenLightning {
 
 	public static void main(String[] args) {
-		
+
 		GreenRuntime.run(new FrameworkTest(),args);
 	
 	}

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/SimpleRest.java

@@ -4,20 +4,16 @@ import com.ociweb.gl.api.GreenRuntime;
 import com.ociweb.gl.api.HTTPRequestReader;
 import com.ociweb.gl.api.HTTPResponseService;
 import com.ociweb.gl.api.RestMethodListener;
-import com.ociweb.gl.api.Writable;
 import com.ociweb.json.encode.JSONRenderer;
 import com.ociweb.pronghorn.network.config.HTTPContentTypeDefaults;
-import com.ociweb.pronghorn.pipe.ChannelWriter;
 
 public class SimpleRest implements RestMethodListener {
 
-	private static final int QUEUE_LENGTH = 1<<15;
-	private static final int MAX_MESSAGE_SIZE = 1<<9;
 
 	private final HTTPResponseService responseService;
 	
-	public SimpleRest(GreenRuntime runtime) {
-		responseService = runtime.newCommandChannel().newHTTPResponseService(QUEUE_LENGTH, MAX_MESSAGE_SIZE);		
+	public SimpleRest(GreenRuntime runtime, int maxResponseCount, int maxResponseSize) {
+		responseService = runtime.newCommandChannel().newHTTPResponseService(maxResponseCount, maxResponseSize);		
 	}
 	
 	public boolean jsonRestRequest(HTTPRequestReader request) {

frameworks/Java/greenlightning/src/main/java/com/ociweb/gl/benchmark/Templates.java

@@ -0,0 +1,62 @@
+package com.ociweb.gl.benchmark;
+
+import java.util.List;
+
+import com.ociweb.json.encode.JSONRenderer;
+import com.ociweb.pronghorn.util.Appendables;
+import com.ociweb.pronghorn.util.template.StringTemplateBuilder;
+import com.ociweb.pronghorn.util.template.StringTemplateRenderer;
+
+public class Templates {
+	
+	static final byte[] ROW_FINISH = "</td></tr>\n".getBytes();
+	static final byte[] ROW_MIDDLE = "</td><td>".getBytes();
+	static final byte[] ROW_START = "<tr><td>".getBytes();
+
+	static final transient StringTemplateRenderer<FortunesObject> fortuneTemplate =		
+	new StringTemplateBuilder<FortunesObject>()
+		   .add("<!DOCTYPE html> <html> <head><title>Fortunes</title></head> <body> <table> <tr><th>id</th><th>message</th></tr>\n")
+	       .add((t,s,i)-> {
+				if (i<s.list().size()) {													
+					t.write(Templates.ROW_START);
+					Appendables.appendValue(t, s.list().get(i).getId());
+					t.write(Templates.ROW_MIDDLE);							
+					Appendables.appendHTMLEntityEscaped(t, s.list().get(i).getFortune());							
+					t.write(Templates.ROW_FINISH);
+					return true;
+				} else {
+					return false;
+				}
+	         })		
+	       .add("</table></body></html>")
+	       .finish();
+	
+	////////////////////////////////////
+	////////////////////////////////////
+
+	static final JSONRenderer<List<ResultObject>> multiTemplate = new JSONRenderer<List<ResultObject>>()
+	  .array((o,i) -> i<o.size()?o:null)
+	      .startObject((o, i) ->  o.get(i))  
+			.integer("id", o -> o.getId() )
+			.integer("randomNumber", o -> o.getResult())
+	      .endObject();
+	
+	////////////////////////////////////
+	////////////////////////////////////
+	
+	static final JSONRenderer<List<ResultObject>> multiTemplateDBRest = new JSONRenderer<List<ResultObject>>()
+	    	  .array((o,i) -> i<o.size()?o:null)
+		          .startObject((o, i) -> o.get(i))
+					.integer("id", o -> o.getId() )
+					.integer("randomNumber", o -> o.getResult())
+		          .endObject();
+	////////////////////////////////////
+	////////////////////////////////////
+	
+	static final JSONRenderer<ResultObject> singleTemplateDBRest = new JSONRenderer<ResultObject>()
+		   	  .startObject()
+				.integer("id", o -> o.getId() )
+				.integer("randomNumber", o -> o.getResult())
+	          .endObject();
+
+}