View Javadoc

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 static org.junit.Assert.assertEquals;
21  
22  import java.io.IOException;
23  import java.util.ArrayList;
24  import java.util.Arrays;
25  import java.util.Collection;
26  import java.util.List;
27  import java.util.concurrent.ThreadLocalRandom;
28  
29  import org.apache.commons.logging.Log;
30  import org.apache.commons.logging.LogFactory;
31  import org.apache.hadoop.conf.Configuration;
32  import org.apache.hadoop.fs.FileSystem;
33  import org.apache.hadoop.hbase.CategoryBasedTimeout;
34  import org.apache.hadoop.hbase.Cell;
35  import org.apache.hadoop.hbase.HBaseTestingUtility;
36  import org.apache.hadoop.hbase.HConstants;
37  import org.apache.hadoop.hbase.client.Durability;
38  import org.apache.hadoop.hbase.client.Increment;
39  import org.apache.hadoop.hbase.client.Scan;
40  import org.apache.hadoop.hbase.client.TestIncrementsFromClientSide;
41  import org.apache.hadoop.hbase.regionserver.wal.FSHLog;
42  import org.apache.hadoop.hbase.testclassification.MediumTests;
43  import org.apache.hadoop.hbase.util.Bytes;
44  import org.apache.hadoop.hbase.wal.WAL;
45  import org.junit.After;
46  import org.junit.Before;
47  import org.junit.Rule;
48  import org.junit.Test;
49  import org.junit.experimental.categories.Category;
50  import org.junit.rules.TestName;
51  import org.junit.rules.TestRule;
52  import org.junit.runner.RunWith;
53  import org.junit.runners.Parameterized;
54  import org.junit.runners.Parameterized.Parameters;
55  
56  
57  /**
58   * Increments with some concurrency against a region to ensure we get the right answer.
59   * Test is parameterized to run the fast and slow path increments; if fast,
60   * HRegion.INCREMENT_FAST_BUT_NARROW_CONSISTENCY_KEY is true.
61   * 
62   * <p>There is similar test up in TestAtomicOperation. It does a test where it has 100 threads
63   * doing increments across two column families all on one row and the increments are connected to
64   * prove atomicity on row.
65   */
66  @Category(MediumTests.class)
67  @RunWith(Parameterized.class)
68  public class TestRegionIncrement {
69    private static final Log LOG = LogFactory.getLog(TestRegionIncrement.class);
70    @Rule public TestName name = new TestName();
71    @Rule public final TestRule timeout =
72        CategoryBasedTimeout.builder().withTimeout(this.getClass()).
73          withLookingForStuckThread(true).build();
74    private static HBaseTestingUtility TEST_UTIL;
75    private final static byte [] INCREMENT_BYTES = Bytes.toBytes("increment");
76    private static final int THREAD_COUNT = 10;
77    private static final int INCREMENT_COUNT = 10000;
78  
79    @Parameters(name = "fast={0}")
80    public static Collection<Object []> data() {
81      return Arrays.asList(new Object[] {Boolean.FALSE}, new Object [] {Boolean.TRUE});
82    }
83  
84    private final boolean fast;
85  
86    public TestRegionIncrement(final boolean fast) {
87      this.fast = fast;
88    }
89  
90    @Before
91    public void setUp() throws Exception {
92      TEST_UTIL = HBaseTestingUtility.createLocalHTU();
93      if (this.fast) {
94        TEST_UTIL.getConfiguration().
95          setBoolean(HRegion.INCREMENT_FAST_BUT_NARROW_CONSISTENCY_KEY, this.fast);
96      }
97    }
98  
99    @After
100   public void tearDown() throws Exception {
101     TEST_UTIL.cleanupTestDir();
102   }
103 
104   private HRegion getRegion(final Configuration conf, final String tableName) throws IOException {
105     WAL wal = new FSHLog(FileSystem.get(conf), TEST_UTIL.getDataTestDir(),
106       TEST_UTIL.getDataTestDir().toString(), conf);
107     return (HRegion)TEST_UTIL.createLocalHRegion(Bytes.toBytes(tableName),
108       HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_BYTE_ARRAY, tableName, conf,
109       false, Durability.SKIP_WAL, wal, INCREMENT_BYTES);
110   }
111 
112   private void closeRegion(final HRegion region) throws IOException {
113     region.close();
114     region.getWAL().close();
115   }
116 
117   /**
118    * Increments a single cell a bunch of times.
119    */
120   private static class SingleCellIncrementer extends Thread {
121     private final int count;
122     private final HRegion region;
123     private final Increment increment;
124 
125     SingleCellIncrementer(final int i, final int count, final HRegion region,
126         final Increment increment) {
127       super("" + i);
128       setDaemon(true);
129       this.count = count;
130       this.region = region;
131       this.increment = increment;
132     }
133 
134     @Override
135     public void run() {
136       for (int i = 0; i < this.count; i++) {
137         try {
138           this.region.increment(this.increment);
139           // LOG.info(getName() + " " + i);
140         } catch (IOException e) {
141           throw new RuntimeException(e);
142         }
143       }
144     }
145   }
146 
147   /**
148    * Increments a random row's Cell <code>count</code> times.
149    */
150   private static class CrossRowCellIncrementer extends Thread {
151     private final int count;
152     private final HRegion region;
153     private final Increment [] increments;
154 
155     CrossRowCellIncrementer(final int i, final int count, final HRegion region, final int range) {
156       super("" + i);
157       setDaemon(true);
158       this.count = count;
159       this.region = region;
160       this.increments = new Increment[range];
161       for (int ii = 0; ii < range; ii++) {
162         this.increments[ii] = new Increment(Bytes.toBytes(i));
163         this.increments[ii].addColumn(INCREMENT_BYTES, INCREMENT_BYTES, 1);
164       }
165     }
166 
167     @Override
168     public void run() {
169       for (int i = 0; i < this.count; i++) {
170         try {
171           int index = ThreadLocalRandom.current().nextInt(0, this.increments.length);
172           this.region.increment(this.increments[index]);
173           // LOG.info(getName() + " " + index);
174         } catch (IOException e) {
175           throw new RuntimeException(e);
176         }
177       }
178     }
179   }
180 
181   /**
182    * Have each thread update its own Cell. Avoid contention with another thread.
183    * @throws IOException
184    * @throws InterruptedException
185    */
186   @Test
187   public void testUnContendedSingleCellIncrement()
188   throws IOException, InterruptedException {
189     final HRegion region = getRegion(TEST_UTIL.getConfiguration(),
190         TestIncrementsFromClientSide.filterStringSoTableNameSafe(this.name.getMethodName()));
191     long startTime = System.currentTimeMillis();
192     try {
193       SingleCellIncrementer [] threads = new SingleCellIncrementer[THREAD_COUNT];
194       for (int i = 0; i < threads.length; i++) {
195         byte [] rowBytes = Bytes.toBytes(i);
196         Increment increment = new Increment(rowBytes);
197         increment.addColumn(INCREMENT_BYTES, INCREMENT_BYTES, 1);
198         threads[i] = new SingleCellIncrementer(i, INCREMENT_COUNT, region, increment);
199       }
200       for (int i = 0; i < threads.length; i++) {
201         threads[i].start();
202       }
203       for (int i = 0; i < threads.length; i++) {
204         threads[i].join();
205       }
206       RegionScanner regionScanner = region.getScanner(new Scan());
207       List<Cell> cells = new ArrayList<Cell>(THREAD_COUNT);
208       while(regionScanner.next(cells)) continue;
209       assertEquals(THREAD_COUNT, cells.size());
210       long total = 0;
211       for (Cell cell: cells) total += Bytes.toLong(cell.getValue());
212       assertEquals(INCREMENT_COUNT * THREAD_COUNT, total);
213     } finally {
214       closeRegion(region);
215       LOG.info(this.name.getMethodName() + " " + (System.currentTimeMillis() - startTime) + "ms");
216     }
217   }
218 
219   /**
220    * Have each thread update its own Cell. Avoid contention with another thread.
221    * This is 
222    * @throws IOException
223    * @throws InterruptedException
224    */
225   @Test
226   public void testContendedAcrossCellsIncrement()
227   throws IOException, InterruptedException {
228     final HRegion region = getRegion(TEST_UTIL.getConfiguration(),
229         TestIncrementsFromClientSide.filterStringSoTableNameSafe(this.name.getMethodName()));
230     long startTime = System.currentTimeMillis();
231     try {
232       CrossRowCellIncrementer [] threads = new CrossRowCellIncrementer[THREAD_COUNT];
233       for (int i = 0; i < threads.length; i++) {
234         threads[i] = new CrossRowCellIncrementer(i, INCREMENT_COUNT, region, THREAD_COUNT);
235       }
236       for (int i = 0; i < threads.length; i++) {
237         threads[i].start();
238       }
239       for (int i = 0; i < threads.length; i++) {
240         threads[i].join();
241       }
242       RegionScanner regionScanner = region.getScanner(new Scan());
243       List<Cell> cells = new ArrayList<Cell>(100);
244       while(regionScanner.next(cells)) continue;
245       assertEquals(THREAD_COUNT, cells.size());
246       long total = 0;
247       for (Cell cell: cells) total += Bytes.toLong(cell.getValue());
248       assertEquals(INCREMENT_COUNT * THREAD_COUNT, total);
249     } finally {
250       closeRegion(region);
251       LOG.info(this.name.getMethodName() + " " + (System.currentTimeMillis() - startTime) + "ms");
252     }
253   }
254 }