001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.regionserver;
019
020import static org.apache.hadoop.hbase.HBaseTestingUtil.fam1;
021import static org.apache.hadoop.hbase.HBaseTestingUtil.fam2;
022import static org.junit.Assert.assertEquals;
023import static org.junit.Assert.assertNotNull;
024import static org.junit.Assert.assertTrue;
025import static org.junit.Assert.fail;
026
027import java.io.IOException;
028import java.util.ArrayList;
029import java.util.Arrays;
030import java.util.List;
031import java.util.Objects;
032import java.util.Random;
033import java.util.concurrent.CountDownLatch;
034import java.util.concurrent.atomic.AtomicInteger;
035import java.util.concurrent.atomic.AtomicLong;
036import org.apache.hadoop.conf.Configuration;
037import org.apache.hadoop.fs.FileSystem;
038import org.apache.hadoop.fs.Path;
039import org.apache.hadoop.hbase.Cell;
040import org.apache.hadoop.hbase.CellUtil;
041import org.apache.hadoop.hbase.CompareOperator;
042import org.apache.hadoop.hbase.HBaseClassTestRule;
043import org.apache.hadoop.hbase.HBaseTestingUtil;
044import org.apache.hadoop.hbase.HConstants;
045import org.apache.hadoop.hbase.MultithreadedTestUtil;
046import org.apache.hadoop.hbase.MultithreadedTestUtil.TestContext;
047import org.apache.hadoop.hbase.MultithreadedTestUtil.TestThread;
048import org.apache.hadoop.hbase.TableName;
049import org.apache.hadoop.hbase.client.Append;
050import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
051import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
052import org.apache.hadoop.hbase.client.Delete;
053import org.apache.hadoop.hbase.client.Durability;
054import org.apache.hadoop.hbase.client.Get;
055import org.apache.hadoop.hbase.client.Increment;
056import org.apache.hadoop.hbase.client.IsolationLevel;
057import org.apache.hadoop.hbase.client.Mutation;
058import org.apache.hadoop.hbase.client.Put;
059import org.apache.hadoop.hbase.client.RegionInfo;
060import org.apache.hadoop.hbase.client.RegionInfoBuilder;
061import org.apache.hadoop.hbase.client.Result;
062import org.apache.hadoop.hbase.client.RowMutations;
063import org.apache.hadoop.hbase.client.Scan;
064import org.apache.hadoop.hbase.client.TableDescriptor;
065import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
066import org.apache.hadoop.hbase.filter.BinaryComparator;
067import org.apache.hadoop.hbase.io.HeapSize;
068import org.apache.hadoop.hbase.io.hfile.BlockCache;
069import org.apache.hadoop.hbase.io.hfile.CacheConfig;
070import org.apache.hadoop.hbase.testclassification.LargeTests;
071import org.apache.hadoop.hbase.testclassification.VerySlowRegionServerTests;
072import org.apache.hadoop.hbase.util.Bytes;
073import org.apache.hadoop.hbase.wal.WAL;
074import org.junit.After;
075import org.junit.Before;
076import org.junit.ClassRule;
077import org.junit.Rule;
078import org.junit.Test;
079import org.junit.experimental.categories.Category;
080import org.junit.rules.TestName;
081import org.slf4j.Logger;
082import org.slf4j.LoggerFactory;
083
084/**
085 * Testing of HRegion.incrementColumnValue, HRegion.increment,
086 * and HRegion.append
087 */
088@Category({VerySlowRegionServerTests.class, LargeTests.class}) // Starts 100 threads
089public class TestAtomicOperation {
090
091  @ClassRule
092  public static final HBaseClassTestRule CLASS_RULE =
093      HBaseClassTestRule.forClass(TestAtomicOperation.class);
094
095  private static final Logger LOG = LoggerFactory.getLogger(TestAtomicOperation.class);
096  @Rule public TestName name = new TestName();
097
098  HRegion region = null;
099  private HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
100
101  // Test names
102  static  byte[] tableName;
103  static final byte[] qual1 = Bytes.toBytes("qual1");
104  static final byte[] qual2 = Bytes.toBytes("qual2");
105  static final byte[] qual3 = Bytes.toBytes("qual3");
106  static final byte[] value1 = Bytes.toBytes("value1");
107  static final byte[] value2 = Bytes.toBytes("value2");
108  static final byte [] row = Bytes.toBytes("rowA");
109  static final byte [] row2 = Bytes.toBytes("rowB");
110
111  @Before
112  public void setup() {
113    tableName = Bytes.toBytes(name.getMethodName());
114  }
115
116  @After
117  public void teardown() throws IOException {
118    if (region != null) {
119      CacheConfig cacheConfig = region.getStores().get(0).getCacheConfig();
120      region.close();
121      WAL wal = region.getWAL();
122      if (wal != null) {
123        wal.close();
124      }
125      cacheConfig.getBlockCache().ifPresent(BlockCache::shutdown);
126      region = null;
127    }
128  }
129
130  //////////////////////////////////////////////////////////////////////////////
131  // New tests that doesn't spin up a mini cluster but rather just test the
132  // individual code pieces in the HRegion.
133  //////////////////////////////////////////////////////////////////////////////
134
135  /**
136   * Test basic append operation.
137   * More tests in
138   * @see org.apache.hadoop.hbase.client.TestFromClientSide#testAppend()
139   */
140  @Test
141  public void testAppend() throws IOException {
142    initHRegion(tableName, name.getMethodName(), fam1);
143    String v1 = "Ultimate Answer to the Ultimate Question of Life,"+
144    " The Universe, and Everything";
145    String v2 = " is... 42.";
146    Append a = new Append(row);
147    a.setReturnResults(false);
148    a.addColumn(fam1, qual1, Bytes.toBytes(v1));
149    a.addColumn(fam1, qual2, Bytes.toBytes(v2));
150    assertTrue(region.append(a, HConstants.NO_NONCE, HConstants.NO_NONCE).isEmpty());
151    a = new Append(row);
152    a.addColumn(fam1, qual1, Bytes.toBytes(v2));
153    a.addColumn(fam1, qual2, Bytes.toBytes(v1));
154    Result result = region.append(a, HConstants.NO_NONCE, HConstants.NO_NONCE);
155    assertEquals(0, Bytes.compareTo(Bytes.toBytes(v1+v2), result.getValue(fam1, qual1)));
156    assertEquals(0, Bytes.compareTo(Bytes.toBytes(v2+v1), result.getValue(fam1, qual2)));
157  }
158
159  @Test
160  public void testAppendWithMultipleFamilies() throws IOException {
161    final byte[] fam3 = Bytes.toBytes("colfamily31");
162    initHRegion(tableName, name.getMethodName(), fam1, fam2, fam3);
163    String v1 = "Appended";
164    String v2 = "Value";
165
166    Append a = new Append(row);
167    a.setReturnResults(false);
168    a.addColumn(fam1, qual1, Bytes.toBytes(v1));
169    a.addColumn(fam2, qual2, Bytes.toBytes(v2));
170    Result result = region.append(a, HConstants.NO_NONCE, HConstants.NO_NONCE);
171    assertTrue("Expected an empty result but result contains " + result.size() + " keys",
172      result.isEmpty());
173
174    a = new Append(row);
175    a.addColumn(fam2, qual2, Bytes.toBytes(v1));
176    a.addColumn(fam1, qual1, Bytes.toBytes(v2));
177    a.addColumn(fam3, qual3, Bytes.toBytes(v2));
178    a.addColumn(fam1, qual2, Bytes.toBytes(v1));
179
180    result = region.append(a, HConstants.NO_NONCE, HConstants.NO_NONCE);
181
182    byte[] actualValue1 = result.getValue(fam1, qual1);
183    byte[] actualValue2 = result.getValue(fam2, qual2);
184    byte[] actualValue3 = result.getValue(fam3, qual3);
185    byte[] actualValue4 = result.getValue(fam1, qual2);
186
187    assertNotNull("Value1 should bot be null", actualValue1);
188    assertNotNull("Value2 should bot be null", actualValue2);
189    assertNotNull("Value3 should bot be null", actualValue3);
190    assertNotNull("Value4 should bot be null", actualValue4);
191    assertEquals(0, Bytes.compareTo(Bytes.toBytes(v1 + v2), actualValue1));
192    assertEquals(0, Bytes.compareTo(Bytes.toBytes(v2 + v1), actualValue2));
193    assertEquals(0, Bytes.compareTo(Bytes.toBytes(v2), actualValue3));
194    assertEquals(0, Bytes.compareTo(Bytes.toBytes(v1), actualValue4));
195  }
196
197  @Test
198  public void testAppendWithNonExistingFamily() throws IOException {
199    initHRegion(tableName, name.getMethodName(), fam1);
200    final String v1 = "Value";
201    final Append a = new Append(row);
202    a.addColumn(fam1, qual1, Bytes.toBytes(v1));
203    a.addColumn(fam2, qual2, Bytes.toBytes(v1));
204    Result result = null;
205    try {
206      result = region.append(a, HConstants.NO_NONCE, HConstants.NO_NONCE);
207      fail("Append operation should fail with NoSuchColumnFamilyException.");
208    } catch (NoSuchColumnFamilyException e) {
209      assertEquals(null, result);
210    } catch (Exception e) {
211      fail("Append operation should fail with NoSuchColumnFamilyException.");
212    }
213  }
214
215  @Test
216  public void testIncrementWithNonExistingFamily() throws IOException {
217    initHRegion(tableName, name.getMethodName(), fam1);
218    final Increment inc = new Increment(row);
219    inc.addColumn(fam1, qual1, 1);
220    inc.addColumn(fam2, qual2, 1);
221    inc.setDurability(Durability.ASYNC_WAL);
222    try {
223      region.increment(inc, HConstants.NO_NONCE, HConstants.NO_NONCE);
224    } catch (NoSuchColumnFamilyException e) {
225      final Get g = new Get(row);
226      final Result result = region.get(g);
227      assertEquals(null, result.getValue(fam1, qual1));
228      assertEquals(null, result.getValue(fam2, qual2));
229    } catch (Exception e) {
230      fail("Increment operation should fail with NoSuchColumnFamilyException.");
231    }
232  }
233
234  /**
235   * Test multi-threaded increments.
236   */
237  @Test
238  public void testIncrementMultiThreads() throws IOException {
239    boolean fast = true;
240    LOG.info("Starting test testIncrementMultiThreads");
241    // run a with mixed column families (1 and 3 versions)
242    initHRegion(tableName, name.getMethodName(), new int[] {1,3}, fam1, fam2);
243
244    // Create 100 threads, each will increment by its own quantity. All 100 threads update the
245    // same row over two column families.
246    int numThreads = 100;
247    int incrementsPerThread = 1000;
248    Incrementer[] all = new Incrementer[numThreads];
249    int expectedTotal = 0;
250    // create all threads
251    for (int i = 0; i < numThreads; i++) {
252      all[i] = new Incrementer(region, i, i, incrementsPerThread);
253      expectedTotal += (i * incrementsPerThread);
254    }
255
256    // run all threads
257    for (int i = 0; i < numThreads; i++) {
258      all[i].start();
259    }
260
261    // wait for all threads to finish
262    for (int i = 0; i < numThreads; i++) {
263      try {
264        all[i].join();
265      } catch (InterruptedException e) {
266        LOG.info("Ignored", e);
267      }
268    }
269    assertICV(row, fam1, qual1, expectedTotal, fast);
270    assertICV(row, fam1, qual2, expectedTotal*2, fast);
271    assertICV(row, fam2, qual3, expectedTotal*3, fast);
272    LOG.info("testIncrementMultiThreads successfully verified that total is " + expectedTotal);
273  }
274
275
276  private void assertICV(byte [] row,
277                         byte [] familiy,
278                         byte[] qualifier,
279                         long amount,
280                         boolean fast) throws IOException {
281    // run a get and see?
282    Get get = new Get(row);
283    if (fast) get.setIsolationLevel(IsolationLevel.READ_UNCOMMITTED);
284    get.addColumn(familiy, qualifier);
285    Result result = region.get(get);
286    assertEquals(1, result.size());
287
288    Cell kv = result.rawCells()[0];
289    long r = Bytes.toLong(CellUtil.cloneValue(kv));
290    assertEquals(amount, r);
291  }
292
293  private void initHRegion (byte [] tableName, String callingMethod,
294      byte[] ... families)
295    throws IOException {
296    initHRegion(tableName, callingMethod, null, families);
297  }
298
299  private void initHRegion(byte[] tableName, String callingMethod, int[] maxVersions,
300    byte[]... families) throws IOException {
301    TableDescriptorBuilder builder =
302      TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName));
303
304    int i = 0;
305    for (byte[] family : families) {
306      ColumnFamilyDescriptor familyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(family)
307        .setMaxVersions(maxVersions != null ? maxVersions[i++] : 1).build();
308      builder.setColumnFamily(familyDescriptor);
309    }
310    TableDescriptor tableDescriptor = builder.build();
311    RegionInfo info = RegionInfoBuilder.newBuilder(tableDescriptor.getTableName()).build();
312    region = TEST_UTIL.createLocalHRegion(info, tableDescriptor);
313  }
314
315  /**
316   * A thread that makes increment calls always on the same row, this.row against two column
317   * families on this row.
318   */
319  public static class Incrementer extends Thread {
320
321    private final Region region;
322    private final int numIncrements;
323    private final int amount;
324
325
326    public Incrementer(Region region, int threadNumber, int amount, int numIncrements) {
327      super("Incrementer." + threadNumber);
328      this.region = region;
329      this.numIncrements = numIncrements;
330      this.amount = amount;
331      setDaemon(true);
332    }
333
334    @Override
335    public void run() {
336      for (int i = 0; i < numIncrements; i++) {
337        try {
338          Increment inc = new Increment(row);
339          inc.addColumn(fam1, qual1, amount);
340          inc.addColumn(fam1, qual2, amount*2);
341          inc.addColumn(fam2, qual3, amount*3);
342          inc.setDurability(Durability.ASYNC_WAL);
343          Result result = region.increment(inc);
344          if (result != null) {
345            assertEquals(Bytes.toLong(result.getValue(fam1, qual1))*2,
346              Bytes.toLong(result.getValue(fam1, qual2)));
347            assertTrue(result.getValue(fam2, qual3) != null);
348            assertEquals(Bytes.toLong(result.getValue(fam1, qual1))*3,
349              Bytes.toLong(result.getValue(fam2, qual3)));
350            assertEquals(Bytes.toLong(result.getValue(fam1, qual1))*2,
351               Bytes.toLong(result.getValue(fam1, qual2)));
352            long fam1Increment = Bytes.toLong(result.getValue(fam1, qual1))*3;
353            long fam2Increment = Bytes.toLong(result.getValue(fam2, qual3));
354            assertEquals("fam1=" + fam1Increment + ", fam2=" + fam2Increment,
355              fam1Increment, fam2Increment);
356          }
357        } catch (IOException e) {
358          e.printStackTrace();
359        }
360      }
361    }
362  }
363
364  @Test
365  public void testAppendMultiThreads() throws IOException {
366    LOG.info("Starting test testAppendMultiThreads");
367    // run a with mixed column families (1 and 3 versions)
368    initHRegion(tableName, name.getMethodName(), new int[] {1,3}, fam1, fam2);
369
370    int numThreads = 100;
371    int opsPerThread = 100;
372    AtomicOperation[] all = new AtomicOperation[numThreads];
373    final byte[] val = new byte[]{1};
374
375    AtomicInteger failures = new AtomicInteger(0);
376    // create all threads
377    for (int i = 0; i < numThreads; i++) {
378      all[i] = new AtomicOperation(region, opsPerThread, null, failures) {
379        @Override
380        public void run() {
381          for (int i=0; i<numOps; i++) {
382            try {
383              Append a = new Append(row);
384              a.addColumn(fam1, qual1, val);
385              a.addColumn(fam1, qual2, val);
386              a.addColumn(fam2, qual3, val);
387              a.setDurability(Durability.ASYNC_WAL);
388              region.append(a, HConstants.NO_NONCE, HConstants.NO_NONCE);
389
390              Get g = new Get(row);
391              Result result = region.get(g);
392              assertEquals(result.getValue(fam1, qual1).length, result.getValue(fam1, qual2).length);
393              assertEquals(result.getValue(fam1, qual1).length, result.getValue(fam2, qual3).length);
394            } catch (IOException e) {
395              e.printStackTrace();
396              failures.incrementAndGet();
397              fail();
398            }
399          }
400        }
401      };
402    }
403
404    // run all threads
405    for (int i = 0; i < numThreads; i++) {
406      all[i].start();
407    }
408
409    // wait for all threads to finish
410    for (int i = 0; i < numThreads; i++) {
411      try {
412        all[i].join();
413      } catch (InterruptedException e) {
414      }
415    }
416    assertEquals(0, failures.get());
417    Get g = new Get(row);
418    Result result = region.get(g);
419    assertEquals(10000, result.getValue(fam1, qual1).length);
420    assertEquals(10000, result.getValue(fam1, qual2).length);
421    assertEquals(10000, result.getValue(fam2, qual3).length);
422  }
423  /**
424   * Test multi-threaded row mutations.
425   */
426  @Test
427  public void testRowMutationMultiThreads() throws IOException {
428    LOG.info("Starting test testRowMutationMultiThreads");
429    initHRegion(tableName, name.getMethodName(), fam1);
430
431    // create 10 threads, each will alternate between adding and
432    // removing a column
433    int numThreads = 10;
434    int opsPerThread = 250;
435    AtomicOperation[] all = new AtomicOperation[numThreads];
436
437    AtomicLong timeStamps = new AtomicLong(0);
438    AtomicInteger failures = new AtomicInteger(0);
439    // create all threads
440    for (int i = 0; i < numThreads; i++) {
441      all[i] = new AtomicOperation(region, opsPerThread, timeStamps, failures) {
442        @Override
443        public void run() {
444          boolean op = true;
445          for (int i=0; i<numOps; i++) {
446            try {
447              // throw in some flushes
448              if (i%10==0) {
449                synchronized(region) {
450                  LOG.debug("flushing");
451                  region.flush(true);
452                  if (i%100==0) {
453                    region.compact(false);
454                  }
455                }
456              }
457              long ts = timeStamps.incrementAndGet();
458              RowMutations rm = new RowMutations(row);
459              if (op) {
460                Put p = new Put(row, ts);
461                p.addColumn(fam1, qual1, value1);
462                p.setDurability(Durability.ASYNC_WAL);
463                rm.add(p);
464                Delete d = new Delete(row);
465                d.addColumns(fam1, qual2, ts);
466                d.setDurability(Durability.ASYNC_WAL);
467                rm.add(d);
468              } else {
469                Delete d = new Delete(row);
470                d.addColumns(fam1, qual1, ts);
471                d.setDurability(Durability.ASYNC_WAL);
472                rm.add(d);
473                Put p = new Put(row, ts);
474                p.addColumn(fam1, qual2, value2);
475                p.setDurability(Durability.ASYNC_WAL);
476                rm.add(p);
477              }
478              region.mutateRow(rm);
479              op ^= true;
480              // check: should always see exactly one column
481              Get g = new Get(row);
482              Result r = region.get(g);
483              if (r.size() != 1) {
484                LOG.debug(Objects.toString(r));
485                failures.incrementAndGet();
486                fail();
487              }
488            } catch (IOException e) {
489              e.printStackTrace();
490              failures.incrementAndGet();
491              fail();
492            }
493          }
494        }
495      };
496    }
497
498    // run all threads
499    for (int i = 0; i < numThreads; i++) {
500      all[i].start();
501    }
502
503    // wait for all threads to finish
504    for (int i = 0; i < numThreads; i++) {
505      try {
506        all[i].join();
507      } catch (InterruptedException e) {
508      }
509    }
510    assertEquals(0, failures.get());
511  }
512
513
514  /**
515   * Test multi-threaded region mutations.
516   */
517  @Test
518  public void testMultiRowMutationMultiThreads() throws IOException {
519
520    LOG.info("Starting test testMultiRowMutationMultiThreads");
521    initHRegion(tableName, name.getMethodName(), fam1);
522
523    // create 10 threads, each will alternate between adding and
524    // removing a column
525    int numThreads = 10;
526    int opsPerThread = 250;
527    AtomicOperation[] all = new AtomicOperation[numThreads];
528
529    AtomicLong timeStamps = new AtomicLong(0);
530    AtomicInteger failures = new AtomicInteger(0);
531    final List<byte[]> rowsToLock = Arrays.asList(row, row2);
532    // create all threads
533    for (int i = 0; i < numThreads; i++) {
534      all[i] = new AtomicOperation(region, opsPerThread, timeStamps, failures) {
535        @Override
536        public void run() {
537          boolean op = true;
538          for (int i=0; i<numOps; i++) {
539            try {
540              // throw in some flushes
541              if (i%10==0) {
542                synchronized(region) {
543                  LOG.debug("flushing");
544                  region.flush(true);
545                  if (i%100==0) {
546                    region.compact(false);
547                  }
548                }
549              }
550              long ts = timeStamps.incrementAndGet();
551              List<Mutation> mrm = new ArrayList<>();
552              if (op) {
553                Put p = new Put(row2, ts);
554                p.addColumn(fam1, qual1, value1);
555                p.setDurability(Durability.ASYNC_WAL);
556                mrm.add(p);
557                Delete d = new Delete(row);
558                d.addColumns(fam1, qual1, ts);
559                d.setDurability(Durability.ASYNC_WAL);
560                mrm.add(d);
561              } else {
562                Delete d = new Delete(row2);
563                d.addColumns(fam1, qual1, ts);
564                d.setDurability(Durability.ASYNC_WAL);
565                mrm.add(d);
566                Put p = new Put(row, ts);
567                p.setDurability(Durability.ASYNC_WAL);
568                p.addColumn(fam1, qual1, value2);
569                mrm.add(p);
570              }
571              region.mutateRowsWithLocks(mrm, rowsToLock, HConstants.NO_NONCE, HConstants.NO_NONCE);
572              op ^= true;
573              // check: should always see exactly one column
574              Scan s = new Scan().withStartRow(row);
575              RegionScanner rs = region.getScanner(s);
576              List<Cell> r = new ArrayList<>();
577              while (rs.next(r))
578                ;
579              rs.close();
580              if (r.size() != 1) {
581                LOG.debug(Objects.toString(r));
582                failures.incrementAndGet();
583                fail();
584              }
585            } catch (IOException e) {
586              e.printStackTrace();
587              failures.incrementAndGet();
588              fail();
589            }
590          }
591        }
592      };
593    }
594
595    // run all threads
596    for (int i = 0; i < numThreads; i++) {
597      all[i].start();
598    }
599
600    // wait for all threads to finish
601    for (int i = 0; i < numThreads; i++) {
602      try {
603        all[i].join();
604      } catch (InterruptedException e) {
605      }
606    }
607    assertEquals(0, failures.get());
608  }
609
610  public static class AtomicOperation extends Thread {
611    protected final HRegion region;
612    protected final int numOps;
613    protected final AtomicLong timeStamps;
614    protected final AtomicInteger failures;
615    protected final Random r = new Random();
616
617    public AtomicOperation(HRegion region, int numOps, AtomicLong timeStamps,
618        AtomicInteger failures) {
619      this.region = region;
620      this.numOps = numOps;
621      this.timeStamps = timeStamps;
622      this.failures = failures;
623    }
624  }
625
626  private static CountDownLatch latch = new CountDownLatch(1);
627  private enum TestStep {
628    INIT,                  // initial put of 10 to set value of the cell
629    PUT_STARTED,           // began doing a put of 50 to cell
630    PUT_COMPLETED,         // put complete (released RowLock, but may not have advanced MVCC).
631    CHECKANDPUT_STARTED,   // began checkAndPut: if 10 -> 11
632    CHECKANDPUT_COMPLETED  // completed checkAndPut
633    // NOTE: at the end of these steps, the value of the cell should be 50, not 11!
634  }
635  private static volatile TestStep testStep = TestStep.INIT;
636  private final String family = "f1";
637
638  /**
639   * Test written as a verifier for HBASE-7051, CheckAndPut should properly read
640   * MVCC.
641   *
642   * Moved into TestAtomicOperation from its original location, TestHBase7051
643   */
644  @Test
645  public void testPutAndCheckAndPutInParallel() throws Exception {
646    Configuration conf = TEST_UTIL.getConfiguration();
647    conf.setClass(HConstants.REGION_IMPL, MockHRegion.class, HeapSize.class);
648    TableDescriptorBuilder tableDescriptorBuilder =
649      TableDescriptorBuilder.newBuilder(TableName.valueOf(name.getMethodName()));
650    ColumnFamilyDescriptor columnFamilyDescriptor =
651      ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family)).build();
652    tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
653    this.region = TEST_UTIL.createLocalHRegion(tableDescriptorBuilder.build(), null, null);
654    Put[] puts = new Put[1];
655    Put put = new Put(Bytes.toBytes("r1"));
656    put.addColumn(Bytes.toBytes(family), Bytes.toBytes("q1"), Bytes.toBytes("10"));
657    puts[0] = put;
658
659    region.batchMutate(puts);
660    MultithreadedTestUtil.TestContext ctx =
661      new MultithreadedTestUtil.TestContext(conf);
662    ctx.addThread(new PutThread(ctx, region));
663    ctx.addThread(new CheckAndPutThread(ctx, region));
664    ctx.startThreads();
665    while (testStep != TestStep.CHECKANDPUT_COMPLETED) {
666      Thread.sleep(100);
667    }
668    ctx.stop();
669    Scan s = new Scan();
670    RegionScanner scanner = region.getScanner(s);
671    List<Cell> results = new ArrayList<>();
672    ScannerContext scannerContext = ScannerContext.newBuilder().setBatchLimit(2).build();
673    scanner.next(results, scannerContext);
674    for (Cell keyValue : results) {
675      assertEquals("50",Bytes.toString(CellUtil.cloneValue(keyValue)));
676    }
677  }
678
679  private class PutThread extends TestThread {
680    private Region region;
681    PutThread(TestContext ctx, Region region) {
682      super(ctx);
683      this.region = region;
684    }
685
686    @Override
687    public void doWork() throws Exception {
688      Put[] puts = new Put[1];
689      Put put = new Put(Bytes.toBytes("r1"));
690      put.addColumn(Bytes.toBytes(family), Bytes.toBytes("q1"), Bytes.toBytes("50"));
691      puts[0] = put;
692      testStep = TestStep.PUT_STARTED;
693      region.batchMutate(puts);
694    }
695  }
696
697  private class CheckAndPutThread extends TestThread {
698    private Region region;
699    CheckAndPutThread(TestContext ctx, Region region) {
700      super(ctx);
701      this.region = region;
702   }
703
704    @Override
705    public void doWork() throws Exception {
706      Put[] puts = new Put[1];
707      Put put = new Put(Bytes.toBytes("r1"));
708      put.addColumn(Bytes.toBytes(family), Bytes.toBytes("q1"), Bytes.toBytes("11"));
709      puts[0] = put;
710      while (testStep != TestStep.PUT_COMPLETED) {
711        Thread.sleep(100);
712      }
713      testStep = TestStep.CHECKANDPUT_STARTED;
714      region.checkAndMutate(Bytes.toBytes("r1"), Bytes.toBytes(family), Bytes.toBytes("q1"),
715        CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("10")), put);
716      testStep = TestStep.CHECKANDPUT_COMPLETED;
717    }
718  }
719
720  public static class MockHRegion extends HRegion {
721
722    public MockHRegion(Path tableDir, WAL log, FileSystem fs, Configuration conf,
723        final RegionInfo regionInfo, final TableDescriptor htd, RegionServerServices rsServices) {
724      super(tableDir, log, fs, conf, regionInfo, htd, rsServices);
725    }
726
727    @Override
728    protected RowLock getRowLockInternal(final byte[] row, boolean readLock,
729        final RowLock prevRowlock) throws IOException {
730      if (testStep == TestStep.CHECKANDPUT_STARTED) {
731        latch.countDown();
732      }
733      return new WrappedRowLock(super.getRowLockInternal(row, readLock, null));
734    }
735
736    public class WrappedRowLock implements RowLock {
737
738      private final RowLock rowLock;
739
740      private WrappedRowLock(RowLock rowLock) {
741        this.rowLock = rowLock;
742      }
743
744
745      @Override
746      public void release() {
747        if (testStep == TestStep.INIT) {
748          this.rowLock.release();
749          return;
750        }
751
752        if (testStep == TestStep.PUT_STARTED) {
753          try {
754            testStep = TestStep.PUT_COMPLETED;
755            this.rowLock.release();
756            // put has been written to the memstore and the row lock has been released, but the
757            // MVCC has not been advanced.  Prior to fixing HBASE-7051, the following order of
758            // operations would cause the non-atomicity to show up:
759            // 1) Put releases row lock (where we are now)
760            // 2) CheckAndPut grabs row lock and reads the value prior to the put (10)
761            //    because the MVCC has not advanced
762            // 3) Put advances MVCC
763            // So, in order to recreate this order, we wait for the checkAndPut to grab the rowLock
764            // (see below), and then wait some more to give the checkAndPut time to read the old
765            // value.
766            latch.await();
767            Thread.sleep(1000);
768          } catch (InterruptedException e) {
769            Thread.currentThread().interrupt();
770          }
771        }
772        else if (testStep == TestStep.CHECKANDPUT_STARTED) {
773          this.rowLock.release();
774        }
775      }
776    }
777  }
778}