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