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.concurrent.ThreadLocalRandom;
  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     public boolean failed = false;
  54
  55     @Override
  56     public void run() {
  57       while (!finished.get()) {
  58         MultiVersionConcurrencyControl.WriteEntry e =
  59             mvcc.begin();
  60         // System.out.println("Begin write: " + e.getWriteNumber());
  61         // 10 usec - 500usec (including 0)
  62         int sleepTime = ThreadLocalRandom.current().nextInt(500);
  63         // 500 * 1000 = 500,000ns = 500 usec
  64         // 1 * 100 = 100ns = 1usec
  65         try {
  66           if (sleepTime > 0) Thread.sleep(0, sleepTime * 1000);
  67         } catch (InterruptedException e1) {
  68         }
  69         try {
  70           mvcc.completeAndWait(e);
  71         } catch (RuntimeException ex) {
  72           // got failure
  73           System.out.println(ex.toString());
  74           ex.printStackTrace();
  75           status.set(false);
  76           return;
  77           // Report failure if possible.
  78         }
  79       }
  80     }
  81   }
  82
  83   @Test
  84   public void testParallelism() throws Exception {
  85     final MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl();
  86
  87     final AtomicBoolean finished = new AtomicBoolean(false);
  88
  89     // fail flag for the reader thread
  90     final AtomicBoolean readerFailed = new AtomicBoolean(false);
  91     final AtomicLong failedAt = new AtomicLong();
  92     Runnable reader = new Runnable() {
  93       @Override
  94       public void run() {
  95         long prev = mvcc.getReadPoint();
  96         while (!finished.get()) {
  97           long newPrev = mvcc.getReadPoint();
  98           if (newPrev < prev) {
  99             // serious problem.
 100             System.out.println("Reader got out of order, prev: " + prev + " next was: " + newPrev);
 101             readerFailed.set(true);
 102             // might as well give up
 103             failedAt.set(newPrev);
 104             return;
 105           }
 106         }
 107       }
 108     };
 109
 110     // writer thread parallelism.
 111     int n = 20;
 112     Thread[] writers = new Thread[n];
 113     AtomicBoolean[] statuses = new AtomicBoolean[n];
 114     Thread readThread = new Thread(reader);
 115
 116     for (int i = 0; i < n; ++i) {
 117       statuses[i] = new AtomicBoolean(true);
 118       writers[i] = new Thread(new Writer(finished, mvcc, statuses[i]));
 119       writers[i].start();
 120     }
 121     readThread.start();
 122
 123     try {
 124       Thread.sleep(10 * 1000);
 125     } catch (InterruptedException ex) {
 126     }
 127
 128     finished.set(true);
 129
 130     readThread.join();
 131     for (int i = 0; i < n; ++i) {
 132       writers[i].join();
 133     }
 134
 135     // check failure.
 136     Assert.assertFalse(readerFailed.get());
 137     for (int i = 0; i < n; ++i) {
 138       Assert.assertTrue(statuses[i].get());
 139     }
 140   }
 141 }