Lowering memory usage from 14g to 7g in attempt to improve performance (#4209)

* 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

* [ci fw-only Java/greenlightning]

Lowering memory usage to 7G from 14G

Clean up design to be easier to read

frameworks/Java/greenlightning/greenlightning.dockerfile

@@ -1,11 +1,11 @@
-FROM maven:3.5.4-jdk-10-slim as maven
+FROM maven:3.5.4-jdk-8-slim as maven
 
 WORKDIR /greenlightning    
 COPY pom.xml pom.xml
 COPY src src
 RUN mvn clean install -q
 
-FROM openjdk:10-jdk-slim
+FROM openjdk:11-jdk-slim
 WORKDIR /greenlightning
 COPY --from=maven /greenlightning/target/greenlightning-test.jar app.jar
-CMD ["java", "-server", "-Xmx16g", "-XX:+UseNUMA", "-XX:+UseParallelGC", "-XX:+AggressiveOpts", "-jar", "app.jar"]
+CMD ["java", "-server", "-Xmx14g", "-XX:+UseNUMA", "-XX:+UseParallelGC", "-XX:+AggressiveOpts", "-jar", "app.jar"]

frameworks/Java/greenlightning/pom.xml

@@ -14,7 +14,7 @@
 		<dependency>
 			<groupId>com.ociweb</groupId>
 			<artifactId>greenlightning</artifactId>
-			<version>1.0.13</version>
+			<version>1.0.14</version>
 		</dependency>
 		<dependency>
 			<groupId>org.slf4j</groupId>

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

@@ -1,535 +1,60 @@
 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.PubSubMethodListener;
 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 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 {
 
-	private final transient PgPoolOptions options;
-	private transient PgPool pool;
-	private final ThreadLocalRandom localRandom = ThreadLocalRandom.current();
+	private final ProcessUpdate processUpdate;
+	private final ProcessFortune processFortune;
+	private final ProcessQuery processQuery;
+	private final PoolManager pm;
 	
-	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.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);
+		pm = new PoolManager(options);
 		
-	}		
-	
-	private PgPool pool() {
-		if (null==pool) {
-			pool = PgClient.pool(options);
-		}
-		return pool;
-	}
-	
-	private int randomValue() {
-		return 1+localRandom.nextInt(10000);
-	}		
-
-	public boolean multiRestRequest(HTTPRequestReader request) { 
+		HTTPResponseService service = runtime.newCommandChannel().newHTTPResponseService(maxResponseCount, maxResponseSize);
 
-		final int queries;
-		if (Struct.DB_MULTI_ROUTE_INT == request.getRouteAssoc() ) {		
-			queries = Math.min(Math.max(1, (request.structured().readInt(Field.QUERIES))),500);		
-		} else {
-			queries = 1;
-		}
+		processUpdate = new ProcessUpdate(pipelineBits, service, pm);
+		processFortune = new ProcessFortune(pipelineBits, service, pm);
+		processQuery = new ProcessQuery(pipelineBits, service, pm);
 		
-	
-		if (DBRestInFlight.hasRoomFor(queries)) {
-			
-			
-			int q = queries;
-			while (--q >= 0) {
-				
-					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.
-									
-					target.setConnectionId(request.getConnectionId());
-					target.setSequenceId(request.getSequenceCode());
-					assert(target.getStatus()==-1);//waiting for work
-					target.setStatus(-2);//out for work	
-					target.setGroupSize(queries);
-				
-					pool().preparedQuery("SELECT * FROM world WHERE id=$1", Tuple.of(randomValue()), r -> {
-							if (r.succeeded()) {
-								
-								PgIterator resultSet = r.result().iterator();
-						        Tuple row = resultSet.next();			        
-						        
-						        target.setId(row.getInteger(0));
-						        target.setResult(row.getInteger(1));					
-								target.setStatus(200);
-								
-							} else {
-								System.out.println("fail: "+r.cause().getLocalizedMessage());
-								target.setStatus(500); 
-							}				
-						});	
-								
-					DBRestInFlight.moveHeadForward(); //always move to ensure this can be read.
-			
-			}
-				
-			return true;
-		} else {
-			return false;
-		}	
 	}
-
-	
-
-	
-	public boolean singleRestRequest(HTTPRequestReader request) { 
-
-		final ResultObject target = DBRestInFlight.headObject();
-		if (null!=target && -1==target.getStatus()) {
-			target.setConnectionId(request.getConnectionId());
-			target.setSequenceId(request.getSequenceCode());
-			assert(target.getStatus()==-1);//waiting for work
-			target.setStatus(-2);//out for work	
-			target.setGroupSize(0);//do not put in a list so mark as 0.
-		
-			pool().preparedQuery("SELECT * FROM world WHERE id=$1", Tuple.of(randomValue()), r -> {
-					if (r.succeeded()) {
-						
-						PgIterator resultSet = r.result().iterator();
-				        Tuple row = resultSet.next();			        
-				        
-				        target.setId(row.getInteger(0));
-				        target.setResult(row.getInteger(1));					
-						target.setStatus(200);
-						
-					} else {
-						System.out.println("fail: "+r.cause().getLocalizedMessage());
-						target.setStatus(500);
-					}				
-				});
-
-			
-			DBRestInFlight.moveHeadForward(); //always move to ensure this can be read.
-			return true;
-		} else {
-			return false;//can not pick up new work now			
-		}
-	}
-
-
-	
-	////////////////////////////////////
-	////////////////////////////////////
 	
 	@Override
 	public void tickEvent() { 
-		//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;
-				}
-			}	   
-		}
+		processUpdate.tickEvent();
+		processFortune.tickEvent();
+		processQuery.tickEvent();
+		//removes DB pool if it is not longer in use
+		pm.clean();
 		
-		//for fortune
-		{
-			FortunesObject temp = fortuneInFlight.tailObject();
-			while (isReadyFortune(temp)) {			
-				if (consumeResultObjectFortune(temp)) {
-					temp = fortuneInFlight.tailObject();
-				} else {
-					break;
-				}
-			}		
-		}
 		
 	}
-
-	private boolean isReadyDBRest(ResultObject temp) {
-
-		if (collectionPendingDBRest) {
-			//now ready to send, we have all the data	
-			if (!publishMultiDBResponse(collectorDBRest.get(0).getConnectionId(), collectorDBRest.get(0).getSequenceId() )) {				
-				return false;
-			}
-		}
-		
-		return null!=temp && temp.getStatus()>=0;
-	}
-
-	private boolean consumeResultObjectDBRest(final ResultObject t) {
-		boolean ok;
-						
-		///////////////////////////////
-		if (0 == t.getGroupSize()) {	
-			ok = service.publishHTTPResponse(t.getConnectionId(), t.getSequenceId(), 200,
-				   HTTPContentTypeDefaults.JSON,
-				   w-> {
-					   Templates.singleTemplateDBRest.render(w, t);
-					   t.setStatus(-1);
-					   DBRestInFlight.moveTailForward();//only move forward when it is consumed.
-					   DBRestInFlight.publishTailPosition();
-
-				   });					
-		} else {
-			//collect all the objects
-			assert(isValidToAdd(t, collectorDBRest));
-			collectorDBRest.add(t);					
-			DBRestInFlight.moveTailForward();
-			if (collectorDBRest.size() == t.getGroupSize()) {
-				//now ready to send, we have all the data						
-				ok =publishMultiDBResponse(t.getConnectionId(), t.getSequenceId());
-				
-			} else {
-				ok = true;//added to list
-			}	
-			//moved forward so we can read next but write logic will still be blocked by state not -1			 
-			
-		}
-		return ok;
-	}
-
-	private boolean isValidToAdd(ResultObject t, List<ResultObject> collector) {
-		if (collector.isEmpty()) {
-			return true;
-		}
-		if (collector.get(0).getSequenceId() != t.getSequenceId()) {
-			
-			System.out.println("show collection: "+showCollection(collector));
-			System.out.println("new result adding con "+t.getConnectionId()+" seq "+t.getSequenceId());
-			
-		};
-		
-		
-		return true;
-	}
-
-	private boolean publishMultiDBResponse(long conId, long seqCode) {
-		final boolean result =  service.publishHTTPResponse(conId, seqCode, 200,
-					    				   HTTPContentTypeDefaults.JSON,
-					    				   w-> {
-					    					   Templates.multiTemplateDBRest.render(w, collectorDBRest);
-					    					   
-					    					   int c = collectorDBRest.size();
-					    					   assert(collectorDBRest.get(0).getGroupSize()==c);
-					    					   while (--c >= 0) {
-					    						   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);
-					    						 
-					    					   }
-					    					   collectorDBRest.clear();					    					   
-					    					   DBRestInFlight.publishTailPosition();
-					    				   });
-		collectionPendingDBRest = !result;
-		return result;
-	}
-
-	private String showCollection(List<ResultObject> collector) {
-		
-		StringBuilder builder = new StringBuilder();
-		builder.append("\n");
-		int i = 0;
-		for(ResultObject ro: collector) {
-			builder.append(++i+" Con:"+ro.getConnectionId()).append(" Id:").append(ro.getId()).append(" Seq:").append(ro.getSequenceId());
-			builder.append("\n");
-		}
-		
-		
-		return builder.toString();
+	
+	public boolean restFortuneRequest(HTTPRequestReader request) {
+		return processFortune.restFortuneRequest(request);
 	}
 	
-	///////////////////////////////////////////////////////////////////////////
-	/////////////////////////////////////////
-	///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;
+		return processUpdate.updateRestRequest(request);
 	}
 	
-	//////////////////////////////////////////////////////////////////////////
-	//////////////////////////////////////////
-	//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			
-		}		
+	public boolean multiRestRequest(HTTPRequestReader request) {
+		return processQuery.multiRestRequest(request);
 	}
 
-	private boolean isReadyFortune(FortunesObject temp) {
-		return null!=temp && temp.getStatus()>=0;
+	public boolean singleRestRequest(HTTPRequestReader request) {
+		return processQuery.singleRestRequest(request);
 	}
-
-	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/FortunesObject.java

@@ -11,8 +11,8 @@ public class FortunesObject {
 	private long connectionId; 
 	private long sequenceId;
 	private int status;
-	private List<FortuneObject> list = new ArrayList<FortuneObject>();
-	private List<FortuneObject> recycle = new ArrayList<FortuneObject>();
+	private List<FortuneObject> list = new ArrayList<FortuneObject>(32); //non zero default
+	private List<FortuneObject> recycle = new ArrayList<FortuneObject>(32);
 
 	public long getConnectionId() {
 		return connectionId;

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

@@ -56,7 +56,7 @@ 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 
-    		 5,       //default concurrency, 5 to support 140 channels on 14 core boxes
+    		 10,      //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")),
@@ -85,8 +85,8 @@ public class FrameworkTest implements GreenApp {
     	this.telemetryPort = telemetryPort;
     	this.pipelineBits = 17;//max concurrent in flight database requests 1<<pipelineBits
 
-    	this.dbCallMaxResponseCount = 1<<6;
-    	this.jsonMaxResponseCount = 1<<13;
+    	this.dbCallMaxResponseCount = 1<<4;
+    	this.jsonMaxResponseCount = 1<<11;
     	
     	this.dbCallMaxResponseSize = 20_000; //for 500 mult db call in JSON format
     	this.jsonMaxResponseSize = 1<<9;
@@ -116,6 +116,7 @@ public class FrameworkTest implements GreenApp {
     				.setHost(connectionHost)
     				.setDatabase(connectionDB)
     				.setUser(connectionUser)
+    				.setIdleTimeout(20)
     				.setPassword(connectionPassword)
     				.setCachePreparedStatements(true)
     				.setMaxSize(connectionsPerTrack);	    	
@@ -142,10 +143,13 @@ public class FrameworkTest implements GreenApp {
 		//for 14 cores this is expected to use less than 16G
 		framework.useHTTP1xServer(bindPort, this::parallelBehavior) //standard auto-scale
     			 .setHost(host)
+    			 .setMaxConnectionBits(13) //8K max client connections.
     			 .setConcurrentChannelsPerDecryptUnit(concurrentWritesPerChannel)
     			 .setConcurrentChannelsPerEncryptUnit(concurrentWritesPerChannel)
     			 .setMaxQueueIn(queueLengthOfPendingRequests)
+    			 
     			 .setMinimumInputPipeMemory(minMemoryOfInputPipes)
+    			 .setMaxQueueOut(32)
     			 .setMaxResponseSize(dbCallMaxResponseSize) //big enough for large mult db response
     	         .useInsecureServer(); //turn off TLS
 

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

@@ -5,7 +5,10 @@ import com.ociweb.gl.api.GreenRuntime;
 public class GreenLightning {
 
 	public static void main(String[] args) {
-
+		
+		//PipeConfig.showConfigsCreatedLargerThan = 1<<23;
+		
+		//System.setProperty("pronghorn.processors", "28"); //TODO: could also lower memory usage by shrinking the stack space...
 		GreenRuntime.run(new FrameworkTest(),args);
 	
 	}

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

@@ -0,0 +1,38 @@
+package com.ociweb.gl.benchmark;
+
+import io.reactiverse.pgclient.PgClient;
+import io.reactiverse.pgclient.PgPool;
+import io.reactiverse.pgclient.PgPoolOptions;
+
+public class PoolManager {
+
+	private final transient PgPoolOptions options;
+	private transient PgPool pool;
+	private transient long lastUsed;
+	
+	public PoolManager(PgPoolOptions options) {
+		this.options = options;		
+	}
+		
+	public PgPool pool() {
+		if (null==pool) {
+			pool = PgClient.pool(options);	//TODO: how to clear this when not in use?		
+		}
+		lastUsed = System.nanoTime();
+		return pool;
+	}
+	
+	public void clean() {
+		//close pool if it has not been used for a while
+		//this gives back some memory and thread resources
+		if (null!=pool) {
+						
+			long duration = System.nanoTime()-lastUsed;
+			if (duration > 60_000_000_000L) {//60 seconds
+				pool.close();
+				pool = null;
+			}
+		}
+	}
+	
+}

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

@@ -0,0 +1,146 @@
+package com.ociweb.gl.benchmark;
+
+import com.ociweb.gl.api.HTTPRequestReader;
+import com.ociweb.gl.api.HTTPResponseService;
+import com.ociweb.pronghorn.network.config.HTTPContentTypeDefaults;
+import com.ociweb.pronghorn.pipe.ObjectPipe;
+import com.ociweb.pronghorn.util.AppendableBuilder;
+
+import io.reactiverse.pgclient.PgIterator;
+import io.reactiverse.pgclient.Row;
+
+public class ProcessFortune {
+
+	private transient ObjectPipe<FortunesObject> fortuneInFlight;
+	private int htmlFortunePos=0;
+	private final transient AppendableBuilder htmlFortuneBuffer = new AppendableBuilder();
+	private final PoolManager pm;
+	private final HTTPResponseService service;
+	
+	public ProcessFortune(int pipelineBits, HTTPResponseService service, PoolManager pm) {
+
+		this.fortuneInFlight = new ObjectPipe<FortunesObject>(pipelineBits, FortunesObject.class,	FortunesObject::new);
+		this.pm = pm;
+		this.service = service;
+	}
+	
+	
+	
+	public void tickEvent() { 
+		
+		//for fortune
+		{
+			FortunesObject temp = fortuneInFlight.tailObject();
+			while (isReadyFortune(temp)) {			
+				if (consumeResultObjectFortune(temp)) {
+					temp = fortuneInFlight.tailObject();
+				} else {
+					break;
+				}
+			}		
+		}
+		
+	}
+	
+
+	
+	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();
+		
+			pm.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 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/ProcessQuery.java

@@ -0,0 +1,241 @@
+package com.ociweb.gl.benchmark;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+
+import com.ociweb.gl.api.HTTPRequestReader;
+import com.ociweb.gl.api.HTTPResponseService;
+import com.ociweb.pronghorn.network.config.HTTPContentTypeDefaults;
+import com.ociweb.pronghorn.pipe.ObjectPipe;
+
+import io.reactiverse.pgclient.PgIterator;
+import io.reactiverse.pgclient.Tuple;
+
+public class ProcessQuery {
+	
+	private final ObjectPipe<ResultObject> DBRestInFlight;
+	private boolean collectionPendingDBRest = false;
+	private transient final List<ResultObject> collectorDBRest = new ArrayList<ResultObject>();
+	private final HTTPResponseService service;
+	private transient final PoolManager pm;
+	private final ThreadLocalRandom localRandom = ThreadLocalRandom.current();
+
+	public ProcessQuery(int pipelineBits, HTTPResponseService service, PoolManager pm) {
+		
+		
+		this.DBRestInFlight = new ObjectPipe<ResultObject>(pipelineBits, ResultObject.class,	ResultObject::new);
+		this.service = service;
+		this.pm = pm;
+		
+		
+	}
+	
+	public void tickEvent() { 
+		//for DBRest
+		{
+			ResultObject temp = DBRestInFlight.tailObject();
+			while (isReadyDBRest(temp)) {			
+				if (consumeResultObjectDBRest(temp)) {
+					temp = DBRestInFlight.tailObject();
+				} else {
+					break;
+				}
+			}
+		}
+		
+		
+	}
+	
+	private int randomValue() {
+		return 1+localRandom.nextInt(10000);
+	}		
+
+	public boolean multiRestRequest(HTTPRequestReader request) { 
+
+		final int queries;
+		if (Struct.DB_MULTI_ROUTE_INT == request.getRouteAssoc() ) {		
+			queries = Math.min(Math.max(1, (request.structured().readInt(Field.QUERIES))),500);		
+		} else {
+			queries = 1;
+		}
+		
+	
+		if (DBRestInFlight.hasRoomFor(queries)) {
+			
+			
+			int q = queries;
+			while (--q >= 0) {
+				
+					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.
+									
+					target.setConnectionId(request.getConnectionId());
+					target.setSequenceId(request.getSequenceCode());
+					assert(target.getStatus()==-1);//waiting for work
+					target.setStatus(-2);//out for work	
+					target.setGroupSize(queries);
+				
+					pm.pool().preparedQuery("SELECT * FROM world WHERE id=$1", Tuple.of(randomValue()), r -> {
+							if (r.succeeded()) {
+								
+								PgIterator resultSet = r.result().iterator();
+						        Tuple row = resultSet.next();			        
+						        
+						        target.setId(row.getInteger(0));
+						        target.setResult(row.getInteger(1));					
+								target.setStatus(200);
+								
+							} else {
+								System.out.println("fail: "+r.cause().getLocalizedMessage());
+								target.setStatus(500); 
+							}				
+						});	
+								
+					DBRestInFlight.moveHeadForward(); //always move to ensure this can be read.
+			
+			}
+				
+			return true;
+		} else {
+			return false;
+		}	
+	}
+
+	
+
+	
+	public boolean singleRestRequest(HTTPRequestReader request) { 
+
+		final ResultObject target = DBRestInFlight.headObject();
+		if (null!=target && -1==target.getStatus()) {
+			target.setConnectionId(request.getConnectionId());
+			target.setSequenceId(request.getSequenceCode());
+			assert(target.getStatus()==-1);//waiting for work
+			target.setStatus(-2);//out for work	
+			target.setGroupSize(0);//do not put in a list so mark as 0.
+		
+			pm.pool().preparedQuery("SELECT * FROM world WHERE id=$1", Tuple.of(randomValue()), r -> {
+					if (r.succeeded()) {
+						
+						PgIterator resultSet = r.result().iterator();
+				        Tuple row = resultSet.next();			        
+				        
+				        target.setId(row.getInteger(0));
+				        target.setResult(row.getInteger(1));					
+						target.setStatus(200);
+						
+					} else {
+						System.out.println("fail: "+r.cause().getLocalizedMessage());
+						target.setStatus(500);
+					}				
+				});
+
+			
+			DBRestInFlight.moveHeadForward(); //always move to ensure this can be read.
+			return true;
+		} else {
+			return false;//can not pick up new work now			
+		}
+	}
+	
+
+	private boolean isReadyDBRest(ResultObject temp) {
+
+		if (collectionPendingDBRest) {
+			//now ready to send, we have all the data	
+			if (!publishMultiDBResponse(collectorDBRest.get(0).getConnectionId(), collectorDBRest.get(0).getSequenceId() )) {				
+				return false;
+			}
+		}
+		
+		return null!=temp && temp.getStatus()>=0;
+	}
+
+	private boolean consumeResultObjectDBRest(final ResultObject t) {
+		boolean ok;
+						
+		///////////////////////////////
+		if (0 == t.getGroupSize()) {	
+			ok = service.publishHTTPResponse(t.getConnectionId(), t.getSequenceId(), 200,
+				   HTTPContentTypeDefaults.JSON,
+				   w-> {
+					   Templates.singleTemplateDBRest.render(w, t);
+					   t.setStatus(-1);
+					   DBRestInFlight.moveTailForward();//only move forward when it is consumed.
+					   DBRestInFlight.publishTailPosition();
+
+				   });					
+		} else {
+			//collect all the objects
+			assert(isValidToAdd(t, collectorDBRest));
+			collectorDBRest.add(t);					
+			DBRestInFlight.moveTailForward();
+			if (collectorDBRest.size() == t.getGroupSize()) {
+				//now ready to send, we have all the data						
+				ok =publishMultiDBResponse(t.getConnectionId(), t.getSequenceId());
+				
+			} else {
+				ok = true;//added to list
+			}	
+			//moved forward so we can read next but write logic will still be blocked by state not -1			 
+			
+		}
+		return ok;
+	}
+
+	private boolean isValidToAdd(ResultObject t, List<ResultObject> collector) {
+		if (collector.isEmpty()) {
+			return true;
+		}
+		if (collector.get(0).getSequenceId() != t.getSequenceId()) {
+			
+			System.out.println("show collection: "+showCollection(collector));
+			System.out.println("new result adding con "+t.getConnectionId()+" seq "+t.getSequenceId());
+			
+		};
+		
+		
+		return true;
+	}
+
+	private boolean publishMultiDBResponse(long conId, long seqCode) {
+		final boolean result =  service.publishHTTPResponse(conId, seqCode, 200,
+					    				   HTTPContentTypeDefaults.JSON,
+					    				   w-> {
+					    					   Templates.multiTemplateDBRest.render(w, collectorDBRest);
+					    					   
+					    					   int c = collectorDBRest.size();
+					    					   assert(collectorDBRest.get(0).getGroupSize()==c);
+					    					   while (--c >= 0) {
+					    						   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);
+					    						 
+					    					   }
+					    					   collectorDBRest.clear();					    					   
+					    					   DBRestInFlight.publishTailPosition();
+					    				   });
+		collectionPendingDBRest = !result;
+		return result;
+	}
+
+	private String showCollection(List<ResultObject> collector) {
+		
+		StringBuilder builder = new StringBuilder();
+		builder.append("\n");
+		int i = 0;
+		for(ResultObject ro: collector) {
+			builder.append(++i+" Con:"+ro.getConnectionId()).append(" Id:").append(ro.getId()).append(" Seq:").append(ro.getSequenceId());
+			builder.append("\n");
+		}
+		
+		
+		return builder.toString();
+	}
+	
+	
+}

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

@@ -0,0 +1,179 @@
+package com.ociweb.gl.benchmark;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+
+import com.ociweb.gl.api.HTTPRequestReader;
+import com.ociweb.gl.api.HTTPResponseService;
+import com.ociweb.pronghorn.network.config.HTTPContentTypeDefaults;
+import com.ociweb.pronghorn.pipe.ObjectPipe;
+
+import io.reactiverse.pgclient.PgIterator;
+import io.reactiverse.pgclient.Tuple;
+
+public class ProcessUpdate {
+	
+	private ObjectPipe<ResultObject> DBUpdateInFlight;	
+	private boolean collectionPendingDBUpdate = false;	
+	private final List<ResultObject> collectorDBUpdate = new ArrayList<ResultObject>();
+	private final ThreadLocalRandom localRandom = ThreadLocalRandom.current();
+	private final HTTPResponseService service;
+	private final PoolManager pm;
+	
+	public ProcessUpdate(int pipelineBits, HTTPResponseService service, PoolManager pm) {
+		this.DBUpdateInFlight = new ObjectPipe<ResultObject>(pipelineBits, ResultObject.class,	ResultObject::new);
+		this.service = service;
+		this.pm = pm;
+	}
+	
+	
+	public void tickEvent() { 
+
+		{
+			ResultObject temp = DBUpdateInFlight.tailObject();
+			while (isReadyDBUpdate(temp)) {			
+				if (consumeResultObjectDBUpdate(temp)) {
+					temp = DBUpdateInFlight.tailObject();
+				} else {
+					break;
+				}
+			}	   
+		}
+			
+	}
+	
+
+	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 (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());
+												
+						pm.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());
+							        							       
+							        
+							        pm.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 isReadyDBUpdate(ResultObject temp) {
+
+		if (collectionPendingDBUpdate) {
+			//now ready to send, we have all the data	
+			if (!publishMultiResponseDBUpdate(collectorDBUpdate.get(0).getConnectionId(), collectorDBUpdate.get(0).getSequenceId() )) {
+				return false;
+			}
+		}
+		
+		return null!=temp && temp.getStatus()>=0;
+	}
+
+	private boolean consumeResultObjectDBUpdate(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 =publishMultiResponseDBUpdate(t.getConnectionId(), t.getSequenceId());
+		} else {
+			ok = true;//added to list
+		}				
+		
+		return ok;
+	}
+
+	private boolean publishMultiResponseDBUpdate(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;
+	}
+	
+	
+	
+	
+	
+}

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

@@ -10,6 +10,7 @@ import com.ociweb.pronghorn.network.config.HTTPContentTypeDefaults;
 public class SimpleRest implements RestMethodListener {
 
 
+	private final byte[] messageBytes = "message".getBytes();
 	private final HTTPResponseService responseService;
 	
 	public SimpleRest(GreenRuntime runtime, int maxResponseCount, int maxResponseSize) {
@@ -25,7 +26,7 @@ public class SimpleRest implements RestMethodListener {
 			//      be created once and held as a member.
 			JSONRenderer<HTTPRequestReader> renderJSON = new JSONRenderer<HTTPRequestReader>()
 					.startObject()
-					.string("message", (o,t) -> t.write(FrameworkTest.payload) )
+					.string(messageBytes, (o,t) -> t.write(FrameworkTest.payload) )
 					.endObject();
 					
 			return responseService.publishHTTPResponse(request,