improve all stream analysis performance

Author: kimchy

.idea/runConfigurations/Bootstrap__No_Plugins_.xml

@@ -26,7 +26,6 @@
 import org.elasticsearch.node.Node;
 
 import java.io.IOException;
-import java.util.Random;
 import java.util.concurrent.TimeUnit;
 
 import static org.elasticsearch.client.Requests.*;
@@ -42,8 +41,6 @@
 public class SingleThreadIndexingStress {
 
     public static void main(String[] args) throws Exception {
-        Random random = new Random();
-
         Settings settings = settingsBuilder()
                 .put("cluster.routing.schedule", 200, TimeUnit.MILLISECONDS)
                 .put("index.engine.robin.refreshInterval", "-1")
@@ -52,8 +49,10 @@ public static void main(String[] args) throws Exception {
                 .put(SETTING_NUMBER_OF_REPLICAS, 1)
                 .build();
 
-        Node node1 = nodeBuilder().settings(settingsBuilder().put(settings).put("name", "server1")).node();
-        Node node2 = nodeBuilder().settings(settingsBuilder().put(settings).put("name", "server2")).node();
+        Node[] nodes = new Node[1];
+        for (int i = 0; i < nodes.length; i++) {
+            nodes[i] = nodeBuilder().settings(settingsBuilder().put(settings).put("name", "node" + i)).node();
+        }
 
         Node client = nodeBuilder().settings(settingsBuilder().put(settings).put("name", "client")).client(true).node();
 
@@ -82,11 +81,26 @@ public static void main(String[] args) throws Exception {
 
         client.close();
 
-        node1.close();
-        node2.close();
+        for (Node node : nodes) {
+            node.close();
+        }
     }
 
     private static XContentBuilder source(String id, String nameValue) throws IOException {
-        return jsonBuilder().startObject().field("id", id).field("name", nameValue).endObject();
+        long time = System.currentTimeMillis();
+        return jsonBuilder().startObject()
+                .field("id", id)
+                .field("numeric1", time)
+                .field("numeric2", time)
+                .field("numeric3", time)
+                .field("numeric4", time)
+                .field("numeric5", time)
+                .field("numeric6", time)
+                .field("numeric7", time)
+                .field("numeric8", time)
+                .field("numeric9", time)
+                .field("numeric10", time)
+                .field("name", nameValue)
+                .endObject();
     }
 }

modules/benchmark/micro/src/main/java/org/elasticsearch/benchmark/stress/SingleThreadIndexingStress.java

@@ -94,10 +94,7 @@ private void ensureOpen() throws IOException {
      */
     @Override public int read(char cbuf[], int off, int len) throws IOException {
         ensureOpen();
-        if ((off < 0) || (off > cbuf.length) || (len < 0) ||
-                ((off + len) > cbuf.length) || ((off + len) < 0)) {
-            throw new IndexOutOfBoundsException();
-        } else if (len == 0) {
+        if (len == 0) {
             return 0;
         }
         if (next >= length)

modules/elasticsearch/src/main/java/org/elasticsearch/common/io/FastStringReader.java

@@ -21,7 +21,6 @@
 
 import org.elasticsearch.ElasticSearchIllegalStateException;
 import org.elasticsearch.common.collect.Lists;
-import org.elasticsearch.common.io.CharSequenceReader;
 import org.elasticsearch.common.io.FastCharArrayWriter;
 import org.elasticsearch.common.io.FastStringReader;
 
@@ -40,10 +39,10 @@ public class AllEntries extends Reader {
 
     public static class Entry {
         private final String name;
-        private final CharSequenceReader reader;
+        private final FastStringReader reader;
         private final float boost;
 
-        public Entry(String name, CharSequenceReader reader, float boost) {
+        public Entry(String name, FastStringReader reader, float boost) {
             this.name = name;
             this.reader = reader;
             this.boost = boost;
@@ -57,7 +56,7 @@ public float boost() {
             return this.boost;
         }
 
-        public CharSequenceReader reader() {
+        public FastStringReader reader() {
             return this.reader;
         }
     }
@@ -70,7 +69,12 @@ public CharSequenceReader reader() {
 
     private boolean itsSeparatorTime = false;
 
+    private boolean customBoost = false;
+
     public void addText(String name, String text, float boost) {
+        if (boost != 1.0f) {
+            customBoost = true;
+        }
         Entry entry = new Entry(name, new FastStringReader(text), boost);
         entries.add(entry);
     }
@@ -129,29 +133,52 @@ public Entry current() {
         if (current == null) {
             return -1;
         }
-        int result = current.reader().read(cbuf, off, len);
-        if (result == -1) {
-            if (itsSeparatorTime) {
-                itsSeparatorTime = false;
-                cbuf[off] = ' ';
-                return 1;
+        if (customBoost) {
+            int result = current.reader().read(cbuf, off, len);
+            if (result == -1) {
+                if (itsSeparatorTime) {
+                    itsSeparatorTime = false;
+                    cbuf[off] = ' ';
+                    return 1;
+                }
+                itsSeparatorTime = true;
+                // close(); No need to close, we work on in mem readers
+                if (it.hasNext()) {
+                    current = it.next();
+                } else {
+                    current = null;
+                }
+                return read(cbuf, off, len);
             }
-            itsSeparatorTime = true;
-            advance();
-            return read(cbuf, off, len);
+            return result;
+        } else {
+            int read = 0;
+            while (len > 0) {
+                int result = current.reader().read(cbuf, off, len);
+                if (result == -1) {
+                    if (it.hasNext()) {
+                        current = it.next();
+                    } else {
+                        current = null;
+                        return read;
+                    }
+                    cbuf[off++] = ' ';
+                    read++;
+                    len--;
+                } else {
+                    read += result;
+                    off += result;
+                    len -= result;
+                }
+            }
+            return read;
         }
-        return result;
     }
 
     @Override public void close() {
         if (current != null) {
-            try {
-                current.reader().close();
-            } catch (IOException e) {
-                // can't happen...
-            } finally {
-                current = null;
-            }
+            current.reader().close();
+            current = null;
         }
     }
 
@@ -160,16 +187,6 @@ public Entry current() {
         return (current != null) && current.reader().ready();
     }
 
-    /**
-     * Closes the current reader and opens the next one, if any.
-     */
-    private void advance() {
-        close();
-        if (it.hasNext()) {
-            current = it.next();
-        }
-    }
-
     @Override public String toString() {
         StringBuilder sb = new StringBuilder();
         for (Entry entry : entries) {

modules/elasticsearch/src/main/java/org/elasticsearch/common/lucene/all/AllEntries.java

@@ -31,6 +31,8 @@
 import org.elasticsearch.common.lucene.Lucene;
 import org.testng.annotations.Test;
 
+import java.io.IOException;
+
 import static org.hamcrest.MatcherAssert.*;
 import static org.hamcrest.Matchers.*;
 
@@ -40,6 +42,31 @@
 @Test
 public class SimpleAllTests {
 
+    @Test public void testAllEntriesRead() throws Exception {
+        AllEntries allEntries = new AllEntries();
+        allEntries.addText("field1", "something", 1.0f);
+        allEntries.addText("field2", "else", 1.0f);
+
+        for (int i = 1; i < 30; i++) {
+            allEntries.reset();
+            char[] data = new char[i];
+            String value = slurpToString(allEntries, data);
+            assertThat("failed for " + i, value, equalTo("something else"));
+        }
+    }
+
+    private String slurpToString(AllEntries allEntries, char[] data) throws IOException {
+        StringBuilder sb = new StringBuilder();
+        while (true) {
+            int read = allEntries.read(data, 0, data.length);
+            if (read == -1) {
+                break;
+            }
+            sb.append(data, 0, read);
+        }
+        return sb.toString();
+    }
+
     @Test public void testSimpleAllNoBoost() throws Exception {
         Directory dir = new RAMDirectory();
         IndexWriter indexWriter = new IndexWriter(dir, Lucene.STANDARD_ANALYZER, true, IndexWriter.MaxFieldLength.UNLIMITED);