hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMultiVersionConcurrencyControl.java

   1 /**
   2  * Licensed to the Apache Software Foundation (ASF) under one
   3  * or more contributor license agreements.  See the NOTICE file
   4  * distributed with this work for additional information
   5  * regarding copyright ownership.  The ASF licenses this file
   6  * to you under the Apache License, Version 2.0 (the
   7  * "License"); you may not use this file except in compliance
   8  * with the License.  You may obtain a copy of the License at
   9  *
  10  *     http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  */
  18 package org.apache.hadoop.hbase.regionserver;
  19
  20 import java.util.Random;
  21 import java.util.concurrent.atomic.AtomicBoolean;
  22 import java.util.concurrent.atomic.AtomicLong;
  23 import org.apache.hadoop.hbase.HBaseClassTestRule;
  24 import org.apache.hadoop.hbase.testclassification.MediumTests;
  25 import org.apache.hadoop.hbase.testclassification.RegionServerTests;
  26 import org.junit.Assert;
  27 import org.junit.ClassRule;
  28 import org.junit.Test;
  29 import org.junit.experimental.categories.Category;
  30
  31 /**
  32  * This is a hammer test that verifies MultiVersionConcurrencyControl in a
  33  * multiple writer single reader scenario.
  34  */
  35 @Category({RegionServerTests.class, MediumTests.class})
  36 public class TestMultiVersionConcurrencyControl {
  37
  38   @ClassRule
  39   public static final HBaseClassTestRule CLASS_RULE =
  40       HBaseClassTestRule.forClass(TestMultiVersionConcurrencyControl.class);
  41
  42   static class Writer implements Runnable {
  43     final AtomicBoolean finished;
  44     final MultiVersionConcurrencyControl mvcc;
  45     final AtomicBoolean status;
  46
  47     Writer(AtomicBoolean finished, MultiVersionConcurrencyControl mvcc, AtomicBoolean status) {
  48       this.finished = finished;
  49       this.mvcc = mvcc;
  50       this.status = status;
  51     }
  52
  53     private Random rnd = new Random();
  54     public boolean failed = false;
  55
  56     @Override
  57     public void run() {
  58       while (!finished.get()) {
  59         MultiVersionConcurrencyControl.WriteEntry e =
  60             mvcc.begin();
  61         // System.out.println("Begin write: " + e.getWriteNumber());
  62         // 10 usec - 500usec (including 0)
  63         int sleepTime = rnd.nextInt(500);
  64         // 500 * 1000 = 500,000ns = 500 usec
  65         // 1 * 100 = 100ns = 1usec
  66         try {
  67           if (sleepTime > 0) Thread.sleep(0, sleepTime * 1000);
  68         } catch (InterruptedException e1) {
  69         }
  70         try {
  71           mvcc.completeAndWait(e);
  72         } catch (RuntimeException ex) {
  73           // got failure
  74           System.out.println(ex.toString());
  75           ex.printStackTrace();
  76           status.set(false);
  77           return;
  78           // Report failure if possible.
  79         }
  80       }
  81     }
  82   }
  83
  84   @Test
  85   public void testParallelism() throws Exception {
  86     final MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl();
  87
  88     final AtomicBoolean finished = new AtomicBoolean(false);
  89
  90     // fail flag for the reader thread
  91     final AtomicBoolean readerFailed = new AtomicBoolean(false);
  92     final AtomicLong failedAt = new AtomicLong();
  93     Runnable reader = new Runnable() {
  94       @Override
  95       public void run() {
  96         long prev = mvcc.getReadPoint();
  97         while (!finished.get()) {
  98           long newPrev = mvcc.getReadPoint();
  99           if (newPrev < prev) {
 100             // serious problem.
 101             System.out.println("Reader got out of order, prev: " + prev + " next was: " + newPrev);
 102             readerFailed.set(true);
 103             // might as well give up
 104             failedAt.set(newPrev);
 105             return;
 106           }
 107         }
 108       }
 109     };
 110
 111     // writer thread parallelism.
 112     int n = 20;
 113     Thread[] writers = new Thread[n];
 114     AtomicBoolean[] statuses = new AtomicBoolean[n];
 115     Thread readThread = new Thread(reader);
 116
 117     for (int i = 0; i < n; ++i) {
 118       statuses[i] = new AtomicBoolean(true);
 119       writers[i] = new Thread(new Writer(finished, mvcc, statuses[i]));
 120       writers[i].start();
 121     }
 122     readThread.start();
 123
 124     try {
 125       Thread.sleep(10 * 1000);
 126     } catch (InterruptedException ex) {
 127     }
 128
 129     finished.set(true);
 130
 131     readThread.join();
 132     for (int i = 0; i < n; ++i) {
 133       writers[i].join();
 134     }
 135
 136     // check failure.
 137     Assert.assertFalse(readerFailed.get());
 138     for (int i = 0; i < n; ++i) {
 139       Assert.assertTrue(statuses[i].get());
 140     }
 141   }
 142 }