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.junit.Assert.assertEquals; 021import static org.junit.Assert.assertTrue; 022 023import java.lang.management.ManagementFactory; 024import java.nio.ByteBuffer; 025import java.util.ArrayList; 026import java.util.List; 027import java.util.Random; 028import java.util.concurrent.ThreadLocalRandom; 029import org.apache.hadoop.conf.Configuration; 030import org.apache.hadoop.hbase.ByteBufferKeyValue; 031import org.apache.hadoop.hbase.ExtendedCell; 032import org.apache.hadoop.hbase.HBaseClassTestRule; 033import org.apache.hadoop.hbase.HBaseConfiguration; 034import org.apache.hadoop.hbase.KeyValue; 035import org.apache.hadoop.hbase.testclassification.RegionServerTests; 036import org.apache.hadoop.hbase.testclassification.SmallTests; 037import org.apache.hadoop.hbase.util.Bytes; 038import org.junit.BeforeClass; 039import org.junit.ClassRule; 040import org.junit.Ignore; 041import org.junit.Test; 042import org.junit.experimental.categories.Category; 043 044@Ignore // See HBASE-19742 for issue on reenabling. 045@Category({ RegionServerTests.class, SmallTests.class }) 046public class TestMemstoreLABWithoutPool { 047 048 @ClassRule 049 public static final HBaseClassTestRule CLASS_RULE = 050 HBaseClassTestRule.forClass(TestMemstoreLABWithoutPool.class); 051 052 private final static Configuration conf = new Configuration(); 053 054 private static final byte[] rk = Bytes.toBytes("r1"); 055 private static final byte[] cf = Bytes.toBytes("f"); 056 private static final byte[] q = Bytes.toBytes("q"); 057 058 @BeforeClass 059 public static void setUpBeforeClass() throws Exception { 060 long globalMemStoreLimit = 061 (long) (ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getMax() * 0.8); 062 // disable pool 063 ChunkCreator.initialize(MemStoreLABImpl.CHUNK_SIZE_DEFAULT + Bytes.SIZEOF_LONG, false, 064 globalMemStoreLimit, 0.0f, MemStoreLAB.POOL_INITIAL_SIZE_DEFAULT, null, 065 MemStoreLAB.INDEX_CHUNK_SIZE_PERCENTAGE_DEFAULT); 066 } 067 068 /** 069 * Test a bunch of random allocations 070 */ 071 @Test 072 public void testLABRandomAllocation() { 073 MemStoreLAB mslab = new MemStoreLABImpl(); 074 int expectedOff = 0; 075 ByteBuffer lastBuffer = null; 076 int lastChunkId = -1; 077 // 100K iterations by 0-1K alloc -> 50MB expected 078 // should be reasonable for unit test and also cover wraparound 079 // behavior 080 Random rand = ThreadLocalRandom.current(); 081 for (int i = 0; i < 100000; i++) { 082 int valSize = rand.nextInt(1000); 083 KeyValue kv = new KeyValue(rk, cf, q, new byte[valSize]); 084 int size = kv.getSerializedSize(); 085 ByteBufferKeyValue newKv = (ByteBufferKeyValue) mslab.copyCellInto(kv); 086 if (newKv.getBuffer() != lastBuffer) { 087 // since we add the chunkID at the 0th offset of the chunk and the 088 // chunkid is an int we need to account for those 4 bytes 089 expectedOff = Bytes.SIZEOF_INT; 090 lastBuffer = newKv.getBuffer(); 091 int chunkId = newKv.getBuffer().getInt(0); 092 assertTrue("chunkid should be different", chunkId != lastChunkId); 093 lastChunkId = chunkId; 094 } 095 assertEquals(expectedOff, newKv.getOffset()); 096 assertTrue("Allocation overruns buffer", 097 newKv.getOffset() + size <= newKv.getBuffer().capacity()); 098 expectedOff += size; 099 } 100 } 101 102 /** 103 * Test frequent chunk retirement with chunk pool triggered by lots of threads, making sure 104 * there's no memory leak (HBASE-16195) 105 * @throws Exception if any error occurred 106 */ 107 @Test 108 public void testLABChunkQueueWithMultipleMSLABs() throws Exception { 109 Configuration conf = HBaseConfiguration.create(); 110 MemStoreLABImpl[] mslab = new MemStoreLABImpl[10]; 111 for (int i = 0; i < 10; i++) { 112 mslab[i] = new MemStoreLABImpl(conf); 113 } 114 // launch multiple threads to trigger frequent chunk retirement 115 List<Thread> threads = new ArrayList<>(); 116 // create smaller sized kvs 117 final KeyValue kv = 118 new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("q"), new byte[0]); 119 for (int i = 0; i < 10; i++) { 120 for (int j = 0; j < 10; j++) { 121 threads.add(getChunkQueueTestThread(mslab[i], "testLABChunkQueue-" + j, kv)); 122 } 123 } 124 for (Thread thread : threads) { 125 thread.start(); 126 } 127 // let it run for some time 128 Thread.sleep(3000); 129 for (Thread thread : threads) { 130 thread.interrupt(); 131 } 132 boolean threadsRunning = true; 133 boolean alive = false; 134 while (threadsRunning) { 135 alive = false; 136 for (Thread thread : threads) { 137 if (thread.isAlive()) { 138 alive = true; 139 break; 140 } 141 } 142 if (!alive) { 143 threadsRunning = false; 144 } 145 } 146 // close the mslab 147 for (int i = 0; i < 10; i++) { 148 mslab[i].close(); 149 } 150 // all of the chunkIds would have been returned back 151 assertTrue("All the chunks must have been cleared", 152 ChunkCreator.instance.numberOfMappedChunks() == 0); 153 } 154 155 private Thread getChunkQueueTestThread(final MemStoreLABImpl mslab, String threadName, 156 ExtendedCell cellToCopyInto) { 157 Thread thread = new Thread() { 158 volatile boolean stopped = false; 159 160 @Override 161 public void run() { 162 while (!stopped) { 163 // keep triggering chunk retirement 164 mslab.copyCellInto(cellToCopyInto); 165 } 166 } 167 168 @Override 169 public void interrupt() { 170 this.stopped = true; 171 } 172 }; 173 thread.setName(threadName); 174 thread.setDaemon(true); 175 return thread; 176 } 177}