fixed perf tests

perf/perf-sensei-conf/plugins.xml

@@ -5,5 +5,11 @@
        xsi:schemaLocation="http://www.springframework.org/schema/beans 
                            http://www.springframework.org/schema/beans/spring-beans.xsd">
 
-  <bean id="demo" class="com.sensei.search.nodes.impl.DemoZoieSystemFactory$DemoZoieFactoryFactory" />
+    <bean id="perf-filter" class="com.sensei.perf.indexing.PerfJsonFilter">
+       <constructor-arg value="10000000" />
+    </bean>
+
+	<bean id="perf" class="com.sensei.perf.indexing.PerfStreamDataProviderBuilder">
+	   <constructor-arg ref="perf-filter" />
+	</bean>
 </beans>

perf/perf-sensei-conf/schema.xml

@@ -20,7 +20,7 @@
     lucene - use store in lucene's index
   -->
 
-  <table uid="id" src-data-store="none" src-data-field="src_data" compress-src-data="true" delete-field="" skip-field="">
+  <table uid="uid" src-data-store="none" src-data-field="src_data" compress-src-data="true" delete-field="" skip-field="">
 	<column name="color" type="string" />
 	<column name="category" type="string" />
 	<column name="city" type="string" />

perf/perf-sensei-conf/sensei.properties

@@ -14,7 +14,7 @@ sensei.cluster.url=localhost:2181
 sensei.cluster.timeout=30000
 
 # sensei indexing parameters
-sensei.index.directory = index/cardata
+sensei.index.directory = index/perf/cardata
 
 sensei.index.batchSize = 10000
 sensei.index.batchDelay = 300000
@@ -25,8 +25,9 @@ sensei.index.freshness = 10000
 # index manager parameters
 
 sensei.index.manager.default.maxpartition.id = 1
-sensei.index.manager.default.type = file
-sensei.index.manager.default.file.path = data/cars.json
+sensei.index.manager.default.type = perf
+sensei.index.manager.default.perf.file = data/cars.json
+sensei.index.manager.default.perf.maxIter = 1000
 
 # plugins: from plugins.xml
 

perf/pom.xml

@@ -24,9 +24,10 @@
 				</configuration>
 			</plugin>
 		</plugins>
+
+        <directory>${basedir}/perf-sensei-conf/ext</directory>
 	</build>
 
-
 	<properties>
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 	</properties>

perf/src/main/java/com/sensei/perf/indexing/PerfJsonFilter.java

@@ -0,0 +1,25 @@
+package com.sensei.perf.indexing;
+
+import org.json.JSONObject;
+
+import com.sensei.indexing.api.DataSourceFilter;
+
+public class PerfJsonFilter extends DataSourceFilter<String> {
+
+	private final int _maxCount;
+	private int _count;
+
+	public PerfJsonFilter(int maxCount){
+		_maxCount = maxCount;
+		_count = 0;
+	}
+
+	@Override
+	protected JSONObject doFilter(String data) throws Exception {
+		JSONObject obj = new JSONObject(data);
+		long uid = _count % _maxCount;
+		obj.put("uid",uid);
+		_count++;
+		return obj;
+	}
+}

perf/src/main/java/com/senseidb/perf/indexing/PerfStreamDataProvider.java → perf/src/main/java/com/sensei/perf/indexing/PerfStreamDataProvider.java

@@ -1,4 +1,4 @@
-package com.senseidb.perf.indexing;
+package com.sensei.perf.indexing;
 
 import java.io.File;
 import java.io.IOException;
@@ -12,7 +12,8 @@
 
 import com.sensei.indexing.api.DataSourceFilter;
 import com.sensei.indexing.api.DataSourceFilterable;
-import com.senseidb.perf.indexing.PerfVersion.PerfVersionComparator;
+import com.sensei.perf.indexing.PerfVersion.PerfVersionComparator;
+
 
 public class PerfStreamDataProvider extends StreamDataProvider<JSONObject> implements DataSourceFilterable<String> {
 
@@ -89,8 +90,11 @@ public void setFilter(DataSourceFilter<String> filter) {
 	@Override
 	public void start() {
 		super.start();
+
 		try{
 		  _offset = _startingOffset;
+
+		  logger.info("data file: "+_file.getAbsolutePath()+", offset: "+_offset);
 		  _rad = new RandomAccessFile(_file,"r");
 		  _rad.seek(_offset.version);
 		}

perf/src/main/java/com/sensei/perf/indexing/PerfStreamDataProviderBuilder.java

@@ -0,0 +1,40 @@
+package com.sensei.perf.indexing;
+
+import java.io.File;
+import java.util.Comparator;
+
+import org.apache.commons.configuration.Configuration;
+import org.json.JSONObject;
+import org.springframework.context.ApplicationContext;
+
+import proj.zoie.impl.indexing.StreamDataProvider;
+
+import com.sensei.indexing.api.DataProviderFactoryRegistry.DataProviderBuilder;
+import com.sensei.indexing.api.DataSourceFilter;
+
+public class PerfStreamDataProviderBuilder extends DataProviderBuilder<JSONObject> {
+
+	private final PerfJsonFilter _perfFilter;
+	public PerfStreamDataProviderBuilder(PerfJsonFilter perfFilter){
+		_perfFilter = perfFilter;
+	}
+	@Override
+	public String getName() {
+		return "perf";
+	}
+
+	@Override
+	public StreamDataProvider<JSONObject> buildDataProvider(Configuration conf,
+			DataSourceFilter<JSONObject> dataFilter,
+			Comparator<String> versionComparator, String oldSinceKey,
+			ApplicationContext plugin) throws Exception {
+		String file = conf.getString("file");
+		int maxIter = conf.getInt("maxIter");
+
+		PerfVersion version = PerfVersion.parse(oldSinceKey);
+		PerfStreamDataProvider provider = new PerfStreamDataProvider(new File(file), version, maxIter);
+		provider.setFilter(_perfFilter);
+		return provider;
+	}
+
+}

perf/src/main/java/com/senseidb/perf/indexing/PerfVersion.java → perf/src/main/java/com/sensei/perf/indexing/PerfVersion.java

@@ -1,9 +1,13 @@
-package com.senseidb.perf.indexing;
+package com.sensei.perf.indexing;
 
 import java.util.Comparator;
 
+import org.apache.log4j.Logger;
+
 
 public class PerfVersion implements Comparable<PerfVersion>{
+	private static Logger log = Logger.getLogger(PerfVersion.class);
+
 	public int iter;
 	public long version;
 
@@ -36,10 +40,14 @@ public String toString(){
 	}
 
 	public static final PerfVersion parse(String s){
-		if (s==null || s.length()==0){
+		if (s==null || s.trim().length()==0){
 			return new PerfVersion(0,0L); 
 		}
+
 		String[] parts = s.split(",");
+		if (parts==null || parts.length!=2){
+			return new PerfVersion(0,0L);
+		}
 		int iter = Integer.parseInt(parts[0]);
 		long ver = Long.parseLong(parts[1]);
 

perf/src/main/java/com/senseidb/perf/indexing/client/JMXMonitor.java → perf/src/main/java/com/sensei/perf/indexing/client/JMXMonitor.java

@@ -1,4 +1,4 @@
-package com.senseidb.perf.indexing.client;
+package com.sensei.perf.indexing.client;
 
 import java.io.IOException;
 

perf/src/main/java/com/senseidb/perf/indexing/client/PollingDataCollector.java → perf/src/main/java/com/sensei/perf/indexing/client/PollingDataCollector.java

@@ -1,4 +1,4 @@
-package com.senseidb.perf.indexing.client;
+package com.sensei.perf.indexing.client;
 
 
 public abstract class PollingDataCollector implements Runnable {

perf/src/main/java/com/senseidb/perf/indexing/client/SenseiPerfClient.java → perf/src/main/java/com/sensei/perf/indexing/client/SenseiPerfClient.java

@@ -1,4 +1,4 @@
-package com.senseidb.perf.indexing.client;
+package com.sensei.perf.indexing.client;
 
 import java.io.BufferedReader;
 import java.io.File;
@@ -51,6 +51,7 @@ public class SenseiPerfClient {
 
 	private static ExecutorService _threadPool;
 
+	private static volatile int _numDocs  = 0;
 
 	public static SenseiService buildSvc(String url,String type) throws Exception{
 		if ("rest".equals(type)){
@@ -163,6 +164,12 @@ public static void main(String[] args) throws Exception{
 	    	futureList.add(future);
 	    }
 
+	    SearchReporter searchReportThread = new SearchReporter(1000);
+	    IndexReporter indexReportThread = new IndexReporter(1000);
+
+	    searchReportThread.start();
+	    indexReportThread.start();
+
 	    for(Future<?> future : futureList){
 	    	future.get();
 	    }
@@ -174,18 +181,29 @@ public static void main(String[] args) throws Exception{
 
 		System.out.println("test completed, cleaning up...");
 
+
+	    searchReportThread.terminate();
+	    indexReportThread.terminate();
+
+	    searchReportThread.join();
+	    indexReportThread.join();
+
 		svc.shutdown();
 
 		searchReport.close();
 		indexingReport.close();
 		systemReport.close();
 	}
 
-	private static class SearchReporter implements Runnable{
+	static class SearchReporter extends Thread{
 
 		private volatile boolean  stop;
+		private final long _tickInterval;
 		public SearchReporter(long tickInterval){
+			super("search-reporter");
+			super.setDaemon(true);
 			stop = false;
+			_tickInterval = tickInterval;
 		}
 
 		public void terminate(){
@@ -195,7 +213,55 @@ public void terminate(){
 
 		public void run(){
 			while(!stop){
-
+				StringBuilder builder = new StringBuilder();
+				builder.append(System.currentTimeMillis()).append(",");
+				builder.append(searchTimer.count()).append(",");
+				builder.append(searchTimer.min()).append(",");
+				builder.append(searchTimer.max()).append(",");
+				builder.append(searchTimer.mean()).append(",");
+				double[] percents = searchTimer.percentiles(90,95,99);
+				builder.append(percents[0]).append(",");
+				builder.append(percents[1]).append(",");
+				builder.append(percents[2]);
+				searchReport.println(builder.toString());
+				searchReport.flush();
+				try {
+					Thread.sleep(_tickInterval);
+				} catch (InterruptedException e) {
+					e.printStackTrace();
+				}
+			}
+		}
+	}
+
+private static class IndexReporter extends Thread{
+
+		private volatile boolean  stop;
+		private final long _tickInterval;
+		public IndexReporter(long tickInterval){
+			super("indexing-reporter");
+			super.setDaemon(true);
+			stop = false;
+			_tickInterval = tickInterval;
+		}
+
+		public void terminate(){
+			stop = true;
+			Thread.currentThread().interrupt();
+		}
+
+		public void run(){
+			while(!stop){
+				StringBuilder builder = new StringBuilder();
+				builder.append(System.currentTimeMillis()).append(",");
+				builder.append(_numDocs);
+				indexingReport.println(builder.toString());
+				indexingReport.flush();
+				try {
+					Thread.sleep(_tickInterval);
+				} catch (InterruptedException e) {
+					e.printStackTrace();
+				}
 			}
 		}
 	}
@@ -210,6 +276,10 @@ public SearchThread(String url,String type,String[] queries,long waitInterval,in
 		  super(waitInterval,timeToRun);
 		  try{
 		    _svc = SenseiPerfClient.buildSvc(url, type);
+
+		    // do a search 
+		    SenseiResult res = _svc.doQuery(new SenseiRequest());
+		    _numDocs = res.getTotalDocs();
 		  }
 		  catch(Exception e){
 			_svc = null;
@@ -235,13 +305,15 @@ public void doWork() throws Exception{
 			   DataConfiguration params = new DataConfiguration(mapConf);
 			   final SenseiRequest req = DefaultSenseiJSONServlet.convertSenseiRequest(params);
 			   try{
-				   searchTimer.time(new Callable<SenseiResult>(){
+				   SenseiResult res = searchTimer.time(new Callable<SenseiResult>(){
 
 					@Override
 					public SenseiResult call() throws Exception {
 						return  _svc.doQuery(req);
 					}
-				   });				   
+				   });	
+
+				   _numDocs = res.getTotalDocs();
 			   }
 			   catch(Exception e){
 			     e.printStackTrace();

src/main/java/com/sensei/search/nodes/impl/DefaultJsonQueryBuilderFactory.java

@@ -6,7 +6,6 @@
 import org.apache.lucene.search.Filter;
 import org.apache.lucene.search.MatchAllDocsQuery;
 import org.apache.lucene.search.Query;
-import org.apache.lucene.util.Version;
 import org.json.JSONObject;
 
 import com.sensei.search.nodes.SenseiQueryBuilder;
@@ -34,7 +33,9 @@ public Filter buildFilter() throws ParseException {
 			@Override
 			public Query buildQuery() throws ParseException {
 				if (queryString!=null && queryString.length()>0){
-					return _qparser.parse(queryString);
+					synchronized(_qparser){
+					  return _qparser.parse(queryString);
+					}
 				}
 				else{
 					return new MatchAllDocsQuery();

src/main/java/com/sensei/search/nodes/impl/SimpleQueryBuilder.java

@@ -35,7 +35,9 @@ protected void doBuild(SenseiQuery query, QueryParser parser) throws Exception
       }
 
       if (qString.length()>0){
-        _query = parser.parse(qString);
+    	  synchronized(parser){
+            _query = parser.parse(qString);
+    	  }
       }
     }
   }