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 import static org.apache.hadoop.hbase.HBaseTestingUtility.fam1;
20 import static org.apache.hadoop.hbase.HBaseTestingUtility.fam2;
21 import static org.junit.Assert.assertEquals;
22 import static org.junit.Assert.assertNull;
23 import static org.junit.Assert.fail;
24
25 import java.io.IOException;
26 import java.util.ArrayList;
27 import java.util.Arrays;
28 import java.util.List;
29 import java.util.Random;
30 import java.util.concurrent.CountDownLatch;
31 import java.util.concurrent.atomic.AtomicInteger;
32 import java.util.concurrent.atomic.AtomicLong;
33
34 import org.apache.commons.logging.Log;
35 import org.apache.commons.logging.LogFactory;
36 import org.apache.hadoop.conf.Configuration;
37 import org.apache.hadoop.fs.FileSystem;
38 import org.apache.hadoop.fs.Path;
39 import org.apache.hadoop.hbase.Cell;
40 import org.apache.hadoop.hbase.CellUtil;
41 import org.apache.hadoop.hbase.HBaseTestingUtility;
42 import org.apache.hadoop.hbase.HColumnDescriptor;
43 import org.apache.hadoop.hbase.HConstants;
44 import org.apache.hadoop.hbase.HRegionInfo;
45 import org.apache.hadoop.hbase.HTableDescriptor;
46 import org.apache.hadoop.hbase.MultithreadedTestUtil;
47 import org.apache.hadoop.hbase.MultithreadedTestUtil.TestContext;
48 import org.apache.hadoop.hbase.MultithreadedTestUtil.TestThread;
49 import org.apache.hadoop.hbase.TableName;
50 import org.apache.hadoop.hbase.client.Append;
51 import org.apache.hadoop.hbase.client.Delete;
52 import org.apache.hadoop.hbase.client.Durability;
53 import org.apache.hadoop.hbase.client.Get;
54 import org.apache.hadoop.hbase.client.Increment;
55 import org.apache.hadoop.hbase.client.IsolationLevel;
56 import org.apache.hadoop.hbase.client.Mutation;
57 import org.apache.hadoop.hbase.client.Put;
58 import org.apache.hadoop.hbase.client.Result;
59 import org.apache.hadoop.hbase.client.RowMutations;
60 import org.apache.hadoop.hbase.client.Scan;
61 import org.apache.hadoop.hbase.filter.BinaryComparator;
62 import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
63 import org.apache.hadoop.hbase.io.HeapSize;
64 import org.apache.hadoop.hbase.testclassification.MediumTests;
65 import org.apache.hadoop.hbase.util.Bytes;
66 import org.apache.hadoop.hbase.wal.WAL;
67 import org.junit.After;
68 import org.junit.Before;
69 import org.junit.Rule;
70 import org.junit.Test;
71 import org.junit.experimental.categories.Category;
72 import org.junit.rules.TestName;
73
74
75
76
77
78 @Category(MediumTests.class)
79 public class TestAtomicOperation {
80 static final Log LOG = LogFactory.getLog(TestAtomicOperation.class);
81 @Rule public TestName name = new TestName();
82
83 HRegion region = null;
84 private HBaseTestingUtility TEST_UTIL = HBaseTestingUtility.createLocalHTU();
85
86
87 static byte[] tableName;
88 static final byte[] qual1 = Bytes.toBytes("qual1");
89 static final byte[] qual2 = Bytes.toBytes("qual2");
90 static final byte[] qual3 = Bytes.toBytes("qual3");
91 static final byte[] value1 = Bytes.toBytes("value1");
92 static final byte[] value2 = Bytes.toBytes("value2");
93 static final byte [] row = Bytes.toBytes("rowA");
94 static final byte [] row2 = Bytes.toBytes("rowB");
95
96 @Before
97 public void setup() {
98 tableName = Bytes.toBytes(name.getMethodName());
99 }
100
101 @After
102 public void teardown() throws IOException {
103 if (region != null) {
104 ((HRegion)region).close();
105 region = null;
106 }
107 }
108
109
110
111
112
113
114
115
116
117
118 @Test
119 public void testAppend() throws IOException {
120 initHRegion(tableName, name.getMethodName(), fam1);
121 String v1 = "Ultimate Answer to the Ultimate Question of Life,"+
122 " The Universe, and Everything";
123 String v2 = " is... 42.";
124 Append a = new Append(row);
125 a.setReturnResults(false);
126 a.add(fam1, qual1, Bytes.toBytes(v1));
127 a.add(fam1, qual2, Bytes.toBytes(v2));
128 assertNull(region.append(a, HConstants.NO_NONCE, HConstants.NO_NONCE));
129 a = new Append(row);
130 a.add(fam1, qual1, Bytes.toBytes(v2));
131 a.add(fam1, qual2, Bytes.toBytes(v1));
132 Result result = region.append(a, HConstants.NO_NONCE, HConstants.NO_NONCE);
133 assertEquals(0, Bytes.compareTo(Bytes.toBytes(v1+v2), result.getValue(fam1, qual1)));
134 assertEquals(0, Bytes.compareTo(Bytes.toBytes(v2+v1), result.getValue(fam1, qual2)));
135 }
136
137
138
139
140 @Test
141 public void testIncrementMultiThreadsFastPath() throws IOException {
142 Configuration conf = TEST_UTIL.getConfiguration();
143 String oldValue = conf.get(HRegion.INCREMENT_FAST_BUT_NARROW_CONSISTENCY_KEY);
144 conf.setBoolean(HRegion.INCREMENT_FAST_BUT_NARROW_CONSISTENCY_KEY, true);
145 try {
146 testIncrementMultiThreads(true);
147 } finally {
148 if (oldValue != null) conf.set(HRegion.INCREMENT_FAST_BUT_NARROW_CONSISTENCY_KEY, oldValue);
149 }
150 }
151
152
153
154
155 @Test
156 public void testIncrementMultiThreadsSlowPath() throws IOException {
157 testIncrementMultiThreads(false);
158 }
159
160 private void testIncrementMultiThreads(final boolean fast) throws IOException {
161 LOG.info("Starting test testIncrementMultiThreads");
162
163 initHRegion(tableName, name.getMethodName(), new int[] {1,3}, fam1, fam2);
164
165
166
167 int numThreads = 100;
168 int incrementsPerThread = 1000;
169 Incrementer[] all = new Incrementer[numThreads];
170 int expectedTotal = 0;
171
172
173 for (int i = 0; i < numThreads; i++) {
174 all[i] = new Incrementer(region, i, i, incrementsPerThread);
175 expectedTotal += (i * incrementsPerThread);
176 }
177
178
179 for (int i = 0; i < numThreads; i++) {
180 all[i].start();
181 }
182
183
184 for (int i = 0; i < numThreads; i++) {
185 try {
186 all[i].join();
187 } catch (InterruptedException e) {
188 }
189 }
190
191 assertICV(row, fam1, qual1, expectedTotal, fast);
192 assertICV(row, fam1, qual2, expectedTotal*2, fast);
193 assertICV(row, fam2, qual3, expectedTotal*3, fast);
194 LOG.info("testIncrementMultiThreads successfully verified that total is " +
195 expectedTotal);
196 }
197
198
199 private void assertICV(byte [] row,
200 byte [] familiy,
201 byte[] qualifier,
202 long amount,
203 boolean fast) throws IOException {
204
205 Get get = new Get(row);
206 if (fast) get.setIsolationLevel(IsolationLevel.READ_UNCOMMITTED);
207 get.addColumn(familiy, qualifier);
208 Result result = region.get(get);
209 assertEquals(1, result.size());
210
211 Cell kv = result.rawCells()[0];
212 long r = Bytes.toLong(CellUtil.cloneValue(kv));
213 assertEquals(amount, r);
214 }
215
216 private void initHRegion (byte [] tableName, String callingMethod,
217 byte[] ... families)
218 throws IOException {
219 initHRegion(tableName, callingMethod, null, families);
220 }
221
222 private void initHRegion (byte [] tableName, String callingMethod, int [] maxVersions,
223 byte[] ... families)
224 throws IOException {
225 HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(tableName));
226 int i=0;
227 for(byte [] family : families) {
228 HColumnDescriptor hcd = new HColumnDescriptor(family);
229 hcd.setMaxVersions(maxVersions != null ? maxVersions[i++] : 1);
230 htd.addFamily(hcd);
231 }
232 HRegionInfo info = new HRegionInfo(htd.getTableName(), null, null, false);
233 region = TEST_UTIL.createLocalHRegion(info, htd);
234 }
235
236
237
238
239
240 public static class Incrementer extends Thread {
241
242 private final HRegion region;
243 private final int numIncrements;
244 private final int amount;
245 private final boolean fast;
246
247
248 public Incrementer(HRegion region, int threadNumber, int amount, int numIncrements) {
249 super("Incrementer." + threadNumber);
250 this.region = region;
251 this.numIncrements = numIncrements;
252 this.amount = amount;
253 this.fast = region.getBaseConf().
254 getBoolean(HRegion.INCREMENT_FAST_BUT_NARROW_CONSISTENCY_KEY, false);
255 setDaemon(true);
256 }
257
258 @Override
259 public void run() {
260 for (int i=0; i<numIncrements; i++) {
261 try {
262 Increment inc = new Increment(row);
263 inc.addColumn(fam1, qual1, amount);
264 inc.addColumn(fam1, qual2, amount*2);
265 inc.addColumn(fam2, qual3, amount*3);
266 inc.setDurability(Durability.ASYNC_WAL);
267 Result result = region.increment(inc);
268 assertEquals(Bytes.toLong(result.getValue(fam1, qual1))*2,
269 Bytes.toLong(result.getValue(fam1, qual2)));
270 long fam1Increment = Bytes.toLong(result.getValue(fam1, qual1))*3;
271 long fam2Increment = Bytes.toLong(result.getValue(fam2, qual3));
272 assertEquals("fam1=" + fam1Increment + ", fam2=" + fam2Increment,
273 fam1Increment, fam2Increment);
274 } catch (IOException e) {
275 e.printStackTrace();
276 }
277 }
278 }
279 }
280
281 @Test
282 public void testAppendMultiThreads() throws IOException {
283 LOG.info("Starting test testAppendMultiThreads");
284
285 initHRegion(tableName, name.getMethodName(), new int[] {1,3}, fam1, fam2);
286
287 int numThreads = 100;
288 int opsPerThread = 100;
289 AtomicOperation[] all = new AtomicOperation[numThreads];
290 final byte[] val = new byte[]{1};
291
292 AtomicInteger failures = new AtomicInteger(0);
293
294 for (int i = 0; i < numThreads; i++) {
295 all[i] = new AtomicOperation(region, opsPerThread, null, failures) {
296 @Override
297 public void run() {
298 for (int i=0; i<numOps; i++) {
299 try {
300 Append a = new Append(row);
301 a.add(fam1, qual1, val);
302 a.add(fam1, qual2, val);
303 a.add(fam2, qual3, val);
304 a.setDurability(Durability.ASYNC_WAL);
305 region.append(a, HConstants.NO_NONCE, HConstants.NO_NONCE);
306
307 Get g = new Get(row);
308 Result result = region.get(g);
309 assertEquals(result.getValue(fam1, qual1).length, result.getValue(fam1, qual2).length);
310 assertEquals(result.getValue(fam1, qual1).length, result.getValue(fam2, qual3).length);
311 } catch (IOException e) {
312 e.printStackTrace();
313 failures.incrementAndGet();
314 fail();
315 }
316 }
317 }
318 };
319 }
320
321
322 for (int i = 0; i < numThreads; i++) {
323 all[i].start();
324 }
325
326
327 for (int i = 0; i < numThreads; i++) {
328 try {
329 all[i].join();
330 } catch (InterruptedException e) {
331 }
332 }
333 assertEquals(0, failures.get());
334 Get g = new Get(row);
335 Result result = region.get(g);
336 assertEquals(result.getValue(fam1, qual1).length, 10000);
337 assertEquals(result.getValue(fam1, qual2).length, 10000);
338 assertEquals(result.getValue(fam2, qual3).length, 10000);
339 }
340
341
342
343 @Test
344 public void testRowMutationMultiThreads() throws IOException {
345
346 LOG.info("Starting test testRowMutationMultiThreads");
347 initHRegion(tableName, name.getMethodName(), fam1);
348
349
350
351 int numThreads = 10;
352 int opsPerThread = 500;
353 AtomicOperation[] all = new AtomicOperation[numThreads];
354
355 AtomicLong timeStamps = new AtomicLong(0);
356 AtomicInteger failures = new AtomicInteger(0);
357
358 for (int i = 0; i < numThreads; i++) {
359 all[i] = new AtomicOperation(region, opsPerThread, timeStamps, failures) {
360 @Override
361 public void run() {
362 boolean op = true;
363 for (int i=0; i<numOps; i++) {
364 try {
365
366 if (i%10==0) {
367 synchronized(region) {
368 LOG.debug("flushing");
369 region.flush(true);
370 if (i%100==0) {
371 region.compact(false);
372 }
373 }
374 }
375 long ts = timeStamps.incrementAndGet();
376 RowMutations rm = new RowMutations(row);
377 if (op) {
378 Put p = new Put(row, ts);
379 p.add(fam1, qual1, value1);
380 p.setDurability(Durability.ASYNC_WAL);
381 rm.add(p);
382 Delete d = new Delete(row);
383 d.deleteColumns(fam1, qual2, ts);
384 d.setDurability(Durability.ASYNC_WAL);
385 rm.add(d);
386 } else {
387 Delete d = new Delete(row);
388 d.deleteColumns(fam1, qual1, ts);
389 d.setDurability(Durability.ASYNC_WAL);
390 rm.add(d);
391 Put p = new Put(row, ts);
392 p.add(fam1, qual2, value2);
393 p.setDurability(Durability.ASYNC_WAL);
394 rm.add(p);
395 }
396 region.mutateRow(rm);
397 op ^= true;
398
399 Get g = new Get(row);
400 Result r = region.get(g);
401 if (r.size() != 1) {
402 LOG.debug(r);
403 failures.incrementAndGet();
404 fail();
405 }
406 } catch (IOException e) {
407 e.printStackTrace();
408 failures.incrementAndGet();
409 fail();
410 }
411 }
412 }
413 };
414 }
415
416
417 for (int i = 0; i < numThreads; i++) {
418 all[i].start();
419 }
420
421
422 for (int i = 0; i < numThreads; i++) {
423 try {
424 all[i].join();
425 } catch (InterruptedException e) {
426 }
427 }
428 assertEquals(0, failures.get());
429 }
430
431
432
433
434
435 @Test
436 public void testMultiRowMutationMultiThreads() throws IOException {
437
438 LOG.info("Starting test testMultiRowMutationMultiThreads");
439 initHRegion(tableName, name.getMethodName(), fam1);
440
441
442
443 int numThreads = 10;
444 int opsPerThread = 500;
445 AtomicOperation[] all = new AtomicOperation[numThreads];
446
447 AtomicLong timeStamps = new AtomicLong(0);
448 AtomicInteger failures = new AtomicInteger(0);
449 final List<byte[]> rowsToLock = Arrays.asList(row, row2);
450
451 for (int i = 0; i < numThreads; i++) {
452 all[i] = new AtomicOperation(region, opsPerThread, timeStamps, failures) {
453 @Override
454 public void run() {
455 boolean op = true;
456 for (int i=0; i<numOps; i++) {
457 try {
458
459 if (i%10==0) {
460 synchronized(region) {
461 LOG.debug("flushing");
462 region.flush(true);
463 if (i%100==0) {
464 region.compact(false);
465 }
466 }
467 }
468 long ts = timeStamps.incrementAndGet();
469 List<Mutation> mrm = new ArrayList<Mutation>();
470 if (op) {
471 Put p = new Put(row2, ts);
472 p.add(fam1, qual1, value1);
473 p.setDurability(Durability.ASYNC_WAL);
474 mrm.add(p);
475 Delete d = new Delete(row);
476 d.deleteColumns(fam1, qual1, ts);
477 d.setDurability(Durability.ASYNC_WAL);
478 mrm.add(d);
479 } else {
480 Delete d = new Delete(row2);
481 d.deleteColumns(fam1, qual1, ts);
482 d.setDurability(Durability.ASYNC_WAL);
483 mrm.add(d);
484 Put p = new Put(row, ts);
485 p.setDurability(Durability.ASYNC_WAL);
486 p.add(fam1, qual1, value2);
487 mrm.add(p);
488 }
489 region.mutateRowsWithLocks(mrm, rowsToLock, HConstants.NO_NONCE, HConstants.NO_NONCE);
490 op ^= true;
491
492 Scan s = new Scan(row);
493 RegionScanner rs = region.getScanner(s);
494 List<Cell> r = new ArrayList<Cell>();
495 while (rs.next(r))
496 ;
497 rs.close();
498 if (r.size() != 1) {
499 LOG.debug(r);
500 failures.incrementAndGet();
501 fail();
502 }
503 } catch (IOException e) {
504 e.printStackTrace();
505 failures.incrementAndGet();
506 fail();
507 }
508 }
509 }
510 };
511 }
512
513
514 for (int i = 0; i < numThreads; i++) {
515 all[i].start();
516 }
517
518
519 for (int i = 0; i < numThreads; i++) {
520 try {
521 all[i].join();
522 } catch (InterruptedException e) {
523 }
524 }
525 assertEquals(0, failures.get());
526 }
527
528 public static class AtomicOperation extends Thread {
529 protected final HRegion region;
530 protected final int numOps;
531 protected final AtomicLong timeStamps;
532 protected final AtomicInteger failures;
533 protected final Random r = new Random();
534
535 public AtomicOperation(HRegion region, int numOps, AtomicLong timeStamps,
536 AtomicInteger failures) {
537 this.region = region;
538 this.numOps = numOps;
539 this.timeStamps = timeStamps;
540 this.failures = failures;
541 }
542 }
543
544 private static CountDownLatch latch = new CountDownLatch(1);
545 private enum TestStep {
546 INIT,
547 PUT_STARTED,
548 PUT_COMPLETED,
549 CHECKANDPUT_STARTED,
550 CHECKANDPUT_COMPLETED
551
552 }
553 private static volatile TestStep testStep = TestStep.INIT;
554 private final String family = "f1";
555
556
557
558
559
560
561
562 @Test
563 public void testPutAndCheckAndPutInParallel() throws Exception {
564
565 final String tableName = "testPutAndCheckAndPut";
566 Configuration conf = TEST_UTIL.getConfiguration();
567 conf.setClass(HConstants.REGION_IMPL, MockHRegion.class, HeapSize.class);
568 final MockHRegion region = (MockHRegion) TEST_UTIL.createLocalHRegion(Bytes.toBytes(tableName),
569 null, null, tableName, conf, false, Durability.SYNC_WAL, null, Bytes.toBytes(family));
570
571 Put[] puts = new Put[1];
572 Put put = new Put(Bytes.toBytes("r1"));
573 put.add(Bytes.toBytes(family), Bytes.toBytes("q1"), Bytes.toBytes("10"));
574 puts[0] = put;
575
576 region.batchMutate(puts, HConstants.NO_NONCE, HConstants.NO_NONCE);
577 MultithreadedTestUtil.TestContext ctx =
578 new MultithreadedTestUtil.TestContext(conf);
579 ctx.addThread(new PutThread(ctx, region));
580 ctx.addThread(new CheckAndPutThread(ctx, region));
581 ctx.startThreads();
582 while (testStep != TestStep.CHECKANDPUT_COMPLETED) {
583 Thread.sleep(100);
584 }
585 ctx.stop();
586 Scan s = new Scan();
587 RegionScanner scanner = region.getScanner(s);
588 List<Cell> results = new ArrayList<Cell>();
589 ScannerContext scannerContext = ScannerContext.newBuilder().setBatchLimit(2).build();
590 scanner.next(results, scannerContext);
591 for (Cell keyValue : results) {
592 assertEquals("50",Bytes.toString(CellUtil.cloneValue(keyValue)));
593 }
594
595 }
596
597 private class PutThread extends TestThread {
598 private HRegion region;
599 PutThread(TestContext ctx, HRegion region) {
600 super(ctx);
601 this.region = region;
602 }
603
604 public void doWork() throws Exception {
605 Put[] puts = new Put[1];
606 Put put = new Put(Bytes.toBytes("r1"));
607 put.add(Bytes.toBytes(family), Bytes.toBytes("q1"), Bytes.toBytes("50"));
608 puts[0] = put;
609 testStep = TestStep.PUT_STARTED;
610 region.batchMutate(puts, HConstants.NO_NONCE, HConstants.NO_NONCE);
611 }
612 }
613
614 private class CheckAndPutThread extends TestThread {
615 private HRegion region;
616 CheckAndPutThread(TestContext ctx, HRegion region) {
617 super(ctx);
618 this.region = region;
619 }
620
621 public void doWork() throws Exception {
622 Put[] puts = new Put[1];
623 Put put = new Put(Bytes.toBytes("r1"));
624 put.add(Bytes.toBytes(family), Bytes.toBytes("q1"), Bytes.toBytes("11"));
625 puts[0] = put;
626 while (testStep != TestStep.PUT_COMPLETED) {
627 Thread.sleep(100);
628 }
629 testStep = TestStep.CHECKANDPUT_STARTED;
630 region.checkAndMutate(Bytes.toBytes("r1"), Bytes.toBytes(family), Bytes.toBytes("q1"),
631 CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("10")), put, true);
632 testStep = TestStep.CHECKANDPUT_COMPLETED;
633 }
634 }
635
636 public static class MockHRegion extends HRegion {
637
638 public MockHRegion(Path tableDir, WAL log, FileSystem fs, Configuration conf,
639 final HRegionInfo regionInfo, final HTableDescriptor htd, RegionServerServices rsServices) {
640 super(tableDir, log, fs, conf, regionInfo, htd, rsServices);
641 }
642
643 @Override
644 public RowLock getRowLockInternal(final byte[] row, boolean waitForLock) throws IOException {
645 if (testStep == TestStep.CHECKANDPUT_STARTED) {
646 latch.countDown();
647 }
648 return new WrappedRowLock(super.getRowLockInternal(row, waitForLock));
649 }
650
651 public class WrappedRowLock extends RowLockImpl {
652
653 private WrappedRowLock(RowLock rowLock) {
654 setContext(((RowLockImpl)rowLock).getContext());
655 }
656
657 @Override
658 public void release() {
659 if (testStep == TestStep.INIT) {
660 super.release();
661 return;
662 }
663
664 if (testStep == TestStep.PUT_STARTED) {
665 try {
666 testStep = TestStep.PUT_COMPLETED;
667 super.release();
668
669
670
671
672
673
674
675
676
677
678 latch.await();
679 Thread.sleep(1000);
680 } catch (InterruptedException e) {
681 Thread.currentThread().interrupt();
682 }
683 }
684 else if (testStep == TestStep.CHECKANDPUT_STARTED) {
685 super.release();
686 }
687 }
688 }
689 }
690 }