1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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
59
60
61
62
63
64
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
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
140 } catch (IOException e) {
141 throw new RuntimeException(e);
142 }
143 }
144 }
145 }
146
147
148
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
174 } catch (IOException e) {
175 throw new RuntimeException(e);
176 }
177 }
178 }
179 }
180
181
182
183
184
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
221
222
223
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 }