hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestParallelPut.java

   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.apache.hadoop.hbase.HBaseTestingUtility.fam1;
  21 import static org.junit.Assert.assertEquals;
  22 import static org.junit.Assert.assertTrue;
  23 import java.io.IOException;
  24 import java.util.Random;
  25 import org.apache.hadoop.hbase.Cell;
  26 import org.apache.hadoop.hbase.CellUtil;
  27 import org.apache.hadoop.hbase.HBaseClassTestRule;
  28 import org.apache.hadoop.hbase.HBaseTestingUtility;
  29 import org.apache.hadoop.hbase.HConstants;
  30 import org.apache.hadoop.hbase.HConstants.OperationStatusCode;
  31 import org.apache.hadoop.hbase.HRegionInfo;
  32 import org.apache.hadoop.hbase.TableName;
  33 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
  34 import org.apache.hadoop.hbase.client.Get;
  35 import org.apache.hadoop.hbase.client.Put;
  36 import org.apache.hadoop.hbase.client.Result;
  37 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
  38 import org.apache.hadoop.hbase.testclassification.MediumTests;
  39 import org.apache.hadoop.hbase.testclassification.RegionServerTests;
  40 import org.apache.hadoop.hbase.util.Bytes;
  41 import org.apache.hadoop.hbase.util.EnvironmentEdgeManagerTestHelper;
  42 import org.junit.After;
  43 import org.junit.Before;
  44 import org.junit.BeforeClass;
  45 import org.junit.ClassRule;
  46 import org.junit.Rule;
  47 import org.junit.Test;
  48 import org.junit.experimental.categories.Category;
  49 import org.junit.rules.TestName;
  50 import org.slf4j.Logger;
  51 import org.slf4j.LoggerFactory;
  52
  53 /**
  54  * Testing of multiPut in parallel.
  55  *
  56  */
  57 @Category({RegionServerTests.class, MediumTests.class})
  58 public class TestParallelPut {
  59
  60   @ClassRule
  61   public static final HBaseClassTestRule CLASS_RULE =
  62       HBaseClassTestRule.forClass(TestParallelPut.class);
  63
  64   private static final Logger LOG = LoggerFactory.getLogger(TestParallelPut.class);
  65   @Rule public TestName name = new TestName();
  66
  67   private HRegion region = null;
  68   private static HBaseTestingUtility HBTU = new HBaseTestingUtility();
  69   private static final int THREADS100 = 100;
  70
  71   // Test names
  72   static byte[] tableName;
  73   static final byte[] qual1 = Bytes.toBytes("qual1");
  74   static final byte[] qual2 = Bytes.toBytes("qual2");
  75   static final byte[] qual3 = Bytes.toBytes("qual3");
  76   static final byte[] value1 = Bytes.toBytes("value1");
  77   static final byte[] value2 = Bytes.toBytes("value2");
  78   static final byte [] row = Bytes.toBytes("rowA");
  79   static final byte [] row2 = Bytes.toBytes("rowB");
  80
  81   @BeforeClass
  82   public static void beforeClass() {
  83     // Make sure enough handlers.
  84     HBTU.getConfiguration().setInt(HConstants.REGION_SERVER_HANDLER_COUNT, THREADS100);
  85   }
  86
  87
  88   /**
  89    * @see org.apache.hadoop.hbase.HBaseTestCase#setUp()
  90    */
  91   @Before
  92   public void setUp() throws Exception {
  93     tableName = Bytes.toBytes(name.getMethodName());
  94   }
  95
  96   @After
  97   public void tearDown() throws Exception {
  98     EnvironmentEdgeManagerTestHelper.reset();
  99     if (region != null) {
 100       region.close(true);
 101     }
 102   }
 103
 104   public String getName() {
 105     return name.getMethodName();
 106   }
 107
 108   //////////////////////////////////////////////////////////////////////////////
 109   // New tests that don't spin up a mini cluster but rather just test the
 110   // individual code pieces in the HRegion.
 111   //////////////////////////////////////////////////////////////////////////////
 112
 113   /**
 114    * Test one put command.
 115    */
 116   @Test
 117   public void testPut() throws IOException {
 118     LOG.info("Starting testPut");
 119     this.region = initHRegion(tableName, getName(), fam1);
 120
 121     long value = 1L;
 122
 123     Put put = new Put(row);
 124     put.addColumn(fam1, qual1, Bytes.toBytes(value));
 125     region.put(put);
 126
 127     assertGet(this.region, row, fam1, qual1, Bytes.toBytes(value));
 128   }
 129
 130   /**
 131    * Test multi-threaded Puts.
 132    */
 133   @Test
 134   public void testParallelPuts() throws IOException {
 135
 136     LOG.info("Starting testParallelPuts");
 137
 138     this.region = initHRegion(tableName, getName(), fam1);
 139     int numOps = 1000; // these many operations per thread
 140
 141     // create 100 threads, each will do its own puts
 142     Putter[] all = new Putter[THREADS100];
 143
 144     // create all threads
 145     for (int i = 0; i < THREADS100; i++) {
 146       all[i] = new Putter(region, i, numOps);
 147     }
 148
 149     // run all threads
 150     for (int i = 0; i < THREADS100; i++) {
 151       all[i].start();
 152     }
 153
 154     // wait for all threads to finish
 155     for (int i = 0; i < THREADS100; i++) {
 156       try {
 157         all[i].join();
 158       } catch (InterruptedException e) {
 159         LOG.warn("testParallelPuts encountered InterruptedException." +
 160                  " Ignoring....", e);
 161       }
 162     }
 163     LOG.info("testParallelPuts successfully verified " +
 164              (numOps * THREADS100) + " put operations.");
 165   }
 166
 167
 168   private static void assertGet(final HRegion region, byte [] row, byte [] familiy,
 169       byte[] qualifier, byte[] value) throws IOException {
 170     // run a get and see if the value matches
 171     Get get = new Get(row);
 172     get.addColumn(familiy, qualifier);
 173     Result result = region.get(get);
 174     assertEquals(1, result.size());
 175
 176     Cell kv = result.rawCells()[0];
 177     byte[] r = CellUtil.cloneValue(kv);
 178     assertTrue(Bytes.compareTo(r, value) == 0);
 179   }
 180
 181   private HRegion initHRegion(byte [] tableName, String callingMethod, byte[] ... families)
 182       throws IOException {
 183     TableDescriptorBuilder.ModifyableTableDescriptor tableDescriptor =
 184       new TableDescriptorBuilder.ModifyableTableDescriptor(TableName.valueOf(tableName));
 185     for(byte [] family : families) {
 186       tableDescriptor.setColumnFamily(
 187         new ColumnFamilyDescriptorBuilder.ModifyableColumnFamilyDescriptor(family));
 188     }
 189     HRegionInfo info = new HRegionInfo(tableDescriptor.getTableName(), null, null, false);
 190     return HBTU.createLocalHRegion(info, tableDescriptor);
 191   }
 192
 193   /**
 194    * A thread that makes a few put calls
 195    */
 196   public static class Putter extends Thread {
 197
 198     private final HRegion region;
 199     private final int threadNumber;
 200     private final int numOps;
 201     private final Random rand = new Random();
 202     byte [] rowkey = null;
 203
 204     public Putter(HRegion region, int threadNumber, int numOps) {
 205       this.region = region;
 206       this.threadNumber = threadNumber;
 207       this.numOps = numOps;
 208       this.rowkey = Bytes.toBytes((long)threadNumber); // unique rowid per thread
 209       setDaemon(true);
 210     }
 211
 212     @Override
 213     public void run() {
 214       byte[] value = new byte[100];
 215       Put[]  in = new Put[1];
 216
 217       // iterate for the specified number of operations
 218       for (int i=0; i<numOps; i++) {
 219         // generate random bytes
 220         rand.nextBytes(value);
 221
 222         // put the randombytes and verify that we can read it. This is one
 223         // way of ensuring that rwcc manipulation in HRegion.put() is fine.
 224         Put put = new Put(rowkey);
 225         put.addColumn(fam1, qual1, value);
 226         in[0] = put;
 227         try {
 228           OperationStatus[] ret = region.batchMutate(in);
 229           assertEquals(1, ret.length);
 230           assertEquals(OperationStatusCode.SUCCESS, ret[0].getOperationStatusCode());
 231           assertGet(this.region, rowkey, fam1, qual1, value);
 232         } catch (IOException e) {
 233           assertTrue("Thread id " + threadNumber + " operation " + i + " failed.",
 234                      false);
 235         }
 236       }
 237     }
 238   }
 239 }