lucene-java-3.4.0/lucene/src/test/org/apache/lucene/index/TestStressNRT.java

   1 package org.apache.lucene.index;
   2
   3 /**
   4  * Licensed to the Apache Software Foundation (ASF) under one or more
   5  * contributor license agreements.  See the NOTICE file distributed with
   6  * this work for additional information regarding copyright ownership.
   7  * The ASF licenses this file to You under the Apache License, Version 2.0
   8  * (the "License"); you may not use this file except in compliance with
   9  * the License.  You may obtain a copy of the License at
  10  *
  11  *     http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  */
  19
  20 import java.util.ArrayList;
  21 import java.util.HashMap;
  22 import java.util.List;
  23 import java.util.Map;
  24 import java.util.Random;
  25 import java.util.concurrent.ConcurrentHashMap;
  26 import java.util.concurrent.atomic.AtomicInteger;
  27 import java.util.concurrent.atomic.AtomicLong;
  28
  29 import org.apache.lucene.analysis.MockAnalyzer;
  30 import org.apache.lucene.document.Document;
  31 import org.apache.lucene.document.Field;
  32 import org.apache.lucene.search.IndexSearcher;
  33 import org.apache.lucene.search.Query;
  34 import org.apache.lucene.search.ScoreDoc;
  35 import org.apache.lucene.search.TermQuery;
  36 import org.apache.lucene.search.TopDocs;
  37 import org.apache.lucene.store.Directory;
  38 import org.apache.lucene.util.LuceneTestCase;
  39 import org.apache.lucene.util._TestUtil;
  40
  41 public class TestStressNRT extends LuceneTestCase {
  42   volatile IndexReader reader;
  43
  44   final ConcurrentHashMap<Integer,Long> model = new ConcurrentHashMap<Integer,Long>();
  45   Map<Integer,Long> committedModel = new HashMap<Integer,Long>();
  46   long snapshotCount;
  47   long committedModelClock;
  48   volatile int lastId;
  49   final String field = "val_l";
  50   Object[] syncArr;
  51
  52   private void initModel(int ndocs) {
  53     snapshotCount = 0;
  54     committedModelClock = 0;
  55     lastId = 0;
  56
  57     syncArr = new Object[ndocs];
  58
  59     for (int i=0; i<ndocs; i++) {
  60       model.put(i, -1L);
  61       syncArr[i] = new Object();
  62     }
  63     committedModel.putAll(model);
  64   }
  65
  66   public void test() throws Exception {
  67     // update variables
  68     final int commitPercent = random.nextInt(20);
  69     final int softCommitPercent = random.nextInt(100); // what percent of the commits are soft
  70     final int deletePercent = random.nextInt(50);
  71     final int deleteByQueryPercent = random.nextInt(25);
  72     final int ndocs = atLeast(50);
  73     final int nWriteThreads = _TestUtil.nextInt(random, 1, TEST_NIGHTLY ? 10 : 5);
  74     final int maxConcurrentCommits = _TestUtil.nextInt(random, 1, TEST_NIGHTLY ? 10 : 5);   // number of committers at a time... needed if we want to avoid commit errors due to exceeding the max
  75
  76     final boolean tombstones = random.nextBoolean();
  77
  78
  79     // query variables
  80     final AtomicLong operations = new AtomicLong(atLeast(50000));  // number of query operations to perform in total
  81
  82     final int nReadThreads = _TestUtil.nextInt(random, 1, TEST_NIGHTLY ? 10 : 5);
  83     initModel(ndocs);
  84
  85     if (VERBOSE) {
  86       System.out.println("\n");
  87       System.out.println("TEST: commitPercent=" + commitPercent);
  88       System.out.println("TEST: softCommitPercent=" + softCommitPercent);
  89       System.out.println("TEST: deletePercent=" + deletePercent);
  90       System.out.println("TEST: deleteByQueryPercent=" + deleteByQueryPercent);
  91       System.out.println("TEST: ndocs=" + ndocs);
  92       System.out.println("TEST: nWriteThreads=" + nWriteThreads);
  93       System.out.println("TEST: nReadThreads=" + nReadThreads);
  94       System.out.println("TEST: maxConcurrentCommits=" + maxConcurrentCommits);
  95       System.out.println("TEST: tombstones=" + tombstones);
  96       System.out.println("TEST: operations=" + operations);
  97       System.out.println("\n");
  98     }
  99
 100     final AtomicInteger numCommitting = new AtomicInteger();
 101
 102     List<Thread> threads = new ArrayList<Thread>();
 103
 104     Directory dir = newDirectory();
 105
 106     final RandomIndexWriter writer = new RandomIndexWriter(random, dir, newIndexWriterConfig(TEST_VERSION_CURRENT, new MockAnalyzer(random)));
 107     writer.setDoRandomOptimizeAssert(false);
 108     writer.w.setInfoStream(VERBOSE ? System.out : null);
 109     writer.commit();
 110     reader = IndexReader.open(dir);
 111
 112     for (int i=0; i<nWriteThreads; i++) {
 113       Thread thread = new Thread("WRITER"+i) {
 114         Random rand = new Random(random.nextInt());
 115
 116         @Override
 117         public void run() {
 118           try {
 119             while (operations.get() > 0) {
 120               int oper = rand.nextInt(100);
 121
 122               if (oper < commitPercent) {
 123                 if (numCommitting.incrementAndGet() <= maxConcurrentCommits) {
 124                   Map<Integer,Long> newCommittedModel;
 125                   long version;
 126                   IndexReader oldReader;
 127
 128                   synchronized(TestStressNRT.this) {
 129                     newCommittedModel = new HashMap<Integer,Long>(model);  // take a snapshot
 130                     version = snapshotCount++;
 131                     oldReader = reader;
 132                     oldReader.incRef();  // increment the reference since we will use this for reopening
 133                   }
 134
 135                   IndexReader newReader;
 136                   if (rand.nextInt(100) < softCommitPercent) {
 137                     // assertU(h.commit("softCommit","true"));
 138                     if (random.nextBoolean()) {
 139                       if (VERBOSE) {
 140                         System.out.println("TEST: " + Thread.currentThread().getName() + ": call writer.getReader");
 141                       }
 142                       newReader = writer.getReader(true);
 143                     } else {
 144                       if (VERBOSE) {
 145                         System.out.println("TEST: " + Thread.currentThread().getName() + ": reopen reader=" + oldReader + " version=" + version);
 146                       }
 147                       newReader = oldReader.reopen(writer.w, true);
 148                     }
 149                   } else {
 150                     // assertU(commit());
 151                     if (VERBOSE) {
 152                       System.out.println("TEST: " + Thread.currentThread().getName() + ": commit+reopen reader=" + oldReader + " version=" + version);
 153                     }
 154                     writer.commit();
 155                     if (VERBOSE) {
 156                       System.out.println("TEST: " + Thread.currentThread().getName() + ": now reopen after commit");
 157                     }
 158                     newReader = oldReader.reopen();
 159                   }
 160
 161                   // Code below assumes newReader comes w/
 162                   // extra ref:
 163                   if (newReader == oldReader) {
 164                     newReader.incRef();
 165                   }
 166
 167                   oldReader.decRef();
 168
 169                   synchronized(TestStressNRT.this) {
 170                     // install the new reader if it's newest (and check the current version since another reader may have already been installed)
 171                     //System.out.println(Thread.currentThread().getName() + ": newVersion=" + newReader.getVersion());
 172                     assert newReader.getRefCount() > 0;
 173                     assert reader.getRefCount() > 0;
 174                     if (newReader.getVersion() > reader.getVersion()) {
 175                       if (VERBOSE) {
 176                         System.out.println("TEST: " + Thread.currentThread().getName() + ": install new reader=" + newReader);
 177                       }
 178                       reader.decRef();
 179                       reader = newReader;
 180
 181                       // Silly: forces fieldInfos to be
 182                       // loaded so we don't hit IOE on later
 183                       // reader.toString
 184                       newReader.toString();
 185
 186                       // install this snapshot only if it's newer than the current one
 187                       if (version >= committedModelClock) {
 188                         if (VERBOSE) {
 189                           System.out.println("TEST: " + Thread.currentThread().getName() + ": install new model version=" + version);
 190                         }
 191                         committedModel = newCommittedModel;
 192                         committedModelClock = version;
 193                       } else {
 194                         if (VERBOSE) {
 195                           System.out.println("TEST: " + Thread.currentThread().getName() + ": skip install new model version=" + version);
 196                         }
 197                       }
 198                     } else {
 199                       // if the same reader, don't decRef.
 200                       if (VERBOSE) {
 201                         System.out.println("TEST: " + Thread.currentThread().getName() + ": skip install new reader=" + newReader);
 202                       }
 203                       newReader.decRef();
 204                     }
 205                   }
 206                 }
 207                 numCommitting.decrementAndGet();
 208               } else {
 209
 210                 int id = rand.nextInt(ndocs);
 211                 Object sync = syncArr[id];
 212
 213                 // set the lastId before we actually change it sometimes to try and
 214                 // uncover more race conditions between writing and reading
 215                 boolean before = random.nextBoolean();
 216                 if (before) {
 217                   lastId = id;
 218                 }
 219
 220                 // We can't concurrently update the same document and retain our invariants of increasing values
 221                 // since we can't guarantee what order the updates will be executed.
 222                 synchronized (sync) {
 223                   Long val = model.get(id);
 224                   long nextVal = Math.abs(val)+1;
 225
 226                   if (oper < commitPercent + deletePercent) {
 227                     // assertU("<delete><id>" + id + "</id></delete>");
 228
 229                     // add tombstone first
 230                     if (tombstones) {
 231                       Document d = new Document();
 232                       d.add(new Field("id","-"+Integer.toString(id), Field.Store.YES, Field.Index.NOT_ANALYZED_NO_NORMS));
 233                       d.add(new Field(field, Long.toString(nextVal), Field.Store.YES, Field.Index.NO));
 234                       writer.updateDocument(new Term("id", "-"+Integer.toString(id)), d);
 235                     }
 236
 237                     if (VERBOSE) {
 238                       System.out.println("TEST: " + Thread.currentThread().getName() + ": term delDocs id:" + id + " nextVal=" + nextVal);
 239                     }
 240                     writer.deleteDocuments(new Term("id",Integer.toString(id)));
 241                     model.put(id, -nextVal);
 242                   } else if (oper < commitPercent + deletePercent + deleteByQueryPercent) {
 243                     //assertU("<delete><query>id:" + id + "</query></delete>");
 244
 245                     // add tombstone first
 246                     if (tombstones) {
 247                       Document d = new Document();
 248                       d.add(new Field("id","-"+Integer.toString(id), Field.Store.YES, Field.Index.NOT_ANALYZED_NO_NORMS));
 249                       d.add(new Field(field, Long.toString(nextVal), Field.Store.YES, Field.Index.NO));
 250                       writer.updateDocument(new Term("id", "-"+Integer.toString(id)), d);
 251                     }
 252
 253                     if (VERBOSE) {
 254                       System.out.println("TEST: " + Thread.currentThread().getName() + ": query delDocs id:" + id + " nextVal=" + nextVal);
 255                     }
 256                     writer.deleteDocuments(new TermQuery(new Term("id", Integer.toString(id))));
 257                     model.put(id, -nextVal);
 258                   } else {
 259                     // assertU(adoc("id",Integer.toString(id), field, Long.toString(nextVal)));
 260                     Document d = new Document();
 261                     d.add(newField("id",Integer.toString(id), Field.Store.YES, Field.Index.NOT_ANALYZED_NO_NORMS));
 262                     d.add(newField(field, Long.toString(nextVal), Field.Store.YES, Field.Index.NO));
 263                     if (VERBOSE) {
 264                       System.out.println("TEST: " + Thread.currentThread().getName() + ": u id:" + id + " val=" + nextVal);
 265                     }
 266                     writer.updateDocument(new Term("id", Integer.toString(id)), d);
 267                     if (tombstones) {
 268                       // remove tombstone after new addition (this should be optional?)
 269                       writer.deleteDocuments(new Term("id","-"+Integer.toString(id)));
 270                     }
 271                     model.put(id, nextVal);
 272                   }
 273                 }
 274
 275                 if (!before) {
 276                   lastId = id;
 277                 }
 278               }
 279             }
 280           } catch (Throwable e) {
 281             System.out.println(Thread.currentThread().getName() + ": FAILED: unexpected exception");
 282             e.printStackTrace(System.out);
 283             throw new RuntimeException(e);
 284           }
 285         }
 286       };
 287
 288       threads.add(thread);
 289     }
 290
 291     for (int i=0; i<nReadThreads; i++) {
 292       Thread thread = new Thread("READER"+i) {
 293         Random rand = new Random(random.nextInt());
 294
 295         @Override
 296         public void run() {
 297           try {
 298             while (operations.decrementAndGet() >= 0) {
 299               // bias toward a recently changed doc
 300               int id = rand.nextInt(100) < 25 ? lastId : rand.nextInt(ndocs);
 301
 302               // when indexing, we update the index, then the model
 303               // so when querying, we should first check the model, and then the index
 304
 305               long val;
 306               IndexReader r;
 307               synchronized(TestStressNRT.this) {
 308                 val = committedModel.get(id);
 309                 r = reader;
 310                 r.incRef();
 311               }
 312
 313               if (VERBOSE) {
 314                 System.out.println("TEST: " + Thread.currentThread().getName() + ": s id=" + id + " val=" + val + " r=" + r.getVersion());
 315               }
 316
 317               //  sreq = req("wt","json", "q","id:"+Integer.toString(id), "omitHeader","true");
 318               IndexSearcher searcher = new IndexSearcher(r);
 319               Query q = new TermQuery(new Term("id",Integer.toString(id)));
 320               TopDocs results = searcher.search(q, 10);
 321
 322               if (results.totalHits == 0 && tombstones) {
 323                 // if we couldn't find the doc, look for its tombstone
 324                 q = new TermQuery(new Term("id","-"+Integer.toString(id)));
 325                 results = searcher.search(q, 1);
 326                 if (results.totalHits == 0) {
 327                   if (val == -1L) {
 328                     // expected... no doc was added yet
 329                     r.decRef();
 330                     continue;
 331                   }
 332                   fail("No documents or tombstones found for id " + id + ", expected at least " + val + " reader=" + r);
 333                 }
 334               }
 335
 336               if (results.totalHits == 0 && !tombstones) {
 337                 // nothing to do - we can't tell anything from a deleted doc without tombstones
 338               } else {
 339                 // we should have found the document, or its tombstone
 340                 if (results.totalHits != 1) {
 341                   System.out.println("FAIL: hits id:" + id + " val=" + val);
 342                   for(ScoreDoc sd : results.scoreDocs) {
 343                     final Document doc = r.document(sd.doc);
 344                     System.out.println("  docID=" + sd.doc + " id:" + doc.get("id") + " foundVal=" + doc.get(field));
 345                   }
 346                   fail("id=" + id + " reader=" + r + " totalHits=" + results.totalHits);
 347                 }
 348                 Document doc = searcher.doc(results.scoreDocs[0].doc);
 349                 long foundVal = Long.parseLong(doc.get(field));
 350                 if (foundVal < Math.abs(val)) {
 351                   fail("foundVal=" + foundVal + " val=" + val + " id=" + id + " reader=" + r);
 352                 }
 353               }
 354
 355               r.decRef();
 356             }
 357           } catch (Throwable e) {
 358             operations.set(-1L);
 359             System.out.println(Thread.currentThread().getName() + ": FAILED: unexpected exception");
 360             e.printStackTrace(System.out);
 361             throw new RuntimeException(e);
 362           }
 363         }
 364       };
 365
 366       threads.add(thread);
 367     }
 368
 369     for (Thread thread : threads) {
 370       thread.start();
 371     }
 372
 373     for (Thread thread : threads) {
 374       thread.join();
 375     }
 376
 377     writer.close();
 378     if (VERBOSE) {
 379       System.out.println("TEST: close reader=" + reader);
 380     }
 381     reader.close();
 382     dir.close();
 383   }
 384 }