1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.regionserver;
20
21 import java.lang.management.MemoryUsage;
22 import java.util.concurrent.BlockingQueue;
23 import java.util.concurrent.Executors;
24 import java.util.concurrent.LinkedBlockingQueue;
25 import java.util.concurrent.ScheduledExecutorService;
26 import java.util.concurrent.TimeUnit;
27 import java.util.concurrent.atomic.AtomicLong;
28
29 import org.apache.commons.logging.Log;
30 import org.apache.commons.logging.LogFactory;
31 import org.apache.hadoop.hbase.classification.InterfaceAudience;
32 import org.apache.hadoop.conf.Configuration;
33 import org.apache.hadoop.hbase.io.util.HeapMemorySizeUtil;
34 import org.apache.hadoop.hbase.regionserver.HeapMemStoreLAB.Chunk;
35 import org.apache.hadoop.util.StringUtils;
36
37 import com.google.common.annotations.VisibleForTesting;
38 import com.google.common.util.concurrent.ThreadFactoryBuilder;
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54 @SuppressWarnings("javadoc")
55 @InterfaceAudience.Private
56 public class MemStoreChunkPool {
57 private static final Log LOG = LogFactory.getLog(MemStoreChunkPool.class);
58 final static String CHUNK_POOL_MAXSIZE_KEY = "hbase.hregion.memstore.chunkpool.maxsize";
59 final static String CHUNK_POOL_INITIALSIZE_KEY = "hbase.hregion.memstore.chunkpool.initialsize";
60 final static float POOL_MAX_SIZE_DEFAULT = 0.0f;
61 final static float POOL_INITIAL_SIZE_DEFAULT = 0.0f;
62
63
64 private static MemStoreChunkPool GLOBAL_INSTANCE;
65
66 static boolean chunkPoolDisabled = false;
67
68 private final int maxCount;
69
70
71 private final BlockingQueue<Chunk> reclaimedChunks;
72 private final int chunkSize;
73
74
75 private final ScheduledExecutorService scheduleThreadPool;
76
77 private static final int statThreadPeriod = 60 * 5;
78 private AtomicLong createdChunkCount = new AtomicLong();
79 private AtomicLong reusedChunkCount = new AtomicLong();
80 private AtomicLong requestedChunkCount = new AtomicLong();
81
82 MemStoreChunkPool(Configuration conf, int chunkSize, int maxCount,
83 int initialCount) {
84 this.maxCount = maxCount;
85 this.chunkSize = chunkSize;
86 this.reclaimedChunks = new LinkedBlockingQueue<Chunk>();
87 for (int i = 0; i < initialCount; i++) {
88 Chunk chunk = new Chunk(chunkSize);
89 chunk.init();
90 reclaimedChunks.add(chunk);
91 }
92 createdChunkCount.set(initialCount);
93 scheduleThreadPool = Executors.newScheduledThreadPool(1,
94 new ThreadFactoryBuilder().setNameFormat("MemStoreChunkPool Statistics")
95 .setDaemon(true).build());
96 this.scheduleThreadPool.scheduleAtFixedRate(new StatisticsThread(this),
97 statThreadPeriod, statThreadPeriod, TimeUnit.SECONDS);
98 }
99
100
101
102
103
104
105 Chunk getChunk() {
106 requestedChunkCount.incrementAndGet();
107 Chunk chunk = reclaimedChunks.poll();
108 if (chunk == null) {
109 chunk = new Chunk(chunkSize);
110 createdChunkCount.incrementAndGet();
111 } else {
112 chunk.reset();
113 reusedChunkCount.incrementAndGet();
114 }
115 return chunk;
116 }
117
118
119
120
121
122
123 void putbackChunks(BlockingQueue<Chunk> chunks) {
124 int maxNumToPutback = this.maxCount - reclaimedChunks.size();
125 if (maxNumToPutback <= 0) {
126 return;
127 }
128 chunks.drainTo(reclaimedChunks, maxNumToPutback);
129
130 if (chunks.size() > 0) {
131 if (LOG.isTraceEnabled()) {
132 LOG.trace("Left " + chunks.size() + " unreclaimable chunks, removing them from queue");
133 }
134 chunks.clear();
135 }
136 }
137
138
139
140
141
142
143 void putbackChunk(Chunk chunk) {
144 if (reclaimedChunks.size() >= this.maxCount) {
145 return;
146 }
147 reclaimedChunks.add(chunk);
148 }
149
150 int getPoolSize() {
151 return this.reclaimedChunks.size();
152 }
153
154
155
156
157 void clearChunks() {
158 this.reclaimedChunks.clear();
159 }
160
161 private static class StatisticsThread extends Thread {
162 MemStoreChunkPool mcp;
163
164 public StatisticsThread(MemStoreChunkPool mcp) {
165 super("MemStoreChunkPool.StatisticsThread");
166 setDaemon(true);
167 this.mcp = mcp;
168 }
169
170 @Override
171 public void run() {
172 mcp.logStats();
173 }
174 }
175
176 private void logStats() {
177 long total = createdChunkCount.get();
178 long reused = reusedChunkCount.get();
179 long available = reclaimedChunks.size();
180 long requested = requestedChunkCount.get();
181 LOG.info("Stats: chunk in pool=" + available
182 + ", chunk in use=" + (total - available)
183 + ", total chunk=" + total
184 + ", reused chunk=" + reused
185 + ", reuse ratio=" + (requested == 0 ? "0" : StringUtils.formatPercent(
186 (float) reused / (float) requested, 2)));
187 }
188
189
190
191
192
193 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="DC_DOUBLECHECK",
194 justification="Intentional")
195 static MemStoreChunkPool getPool(Configuration conf) {
196 if (GLOBAL_INSTANCE != null) return GLOBAL_INSTANCE;
197
198 synchronized (MemStoreChunkPool.class) {
199 if (chunkPoolDisabled) return null;
200 if (GLOBAL_INSTANCE != null) return GLOBAL_INSTANCE;
201 float poolSizePercentage = conf.getFloat(CHUNK_POOL_MAXSIZE_KEY, POOL_MAX_SIZE_DEFAULT);
202 if (poolSizePercentage <= 0) {
203 chunkPoolDisabled = true;
204 return null;
205 }
206 if (poolSizePercentage > 1.0) {
207 throw new IllegalArgumentException(CHUNK_POOL_MAXSIZE_KEY + " must be between 0.0 and 1.0");
208 }
209 long heapMax = -1L;
210 final MemoryUsage usage = HeapMemorySizeUtil.safeGetHeapMemoryUsage();
211 if (usage != null) {
212 heapMax = usage.getMax();
213 }
214 long globalMemStoreLimit = (long) (heapMax * HeapMemorySizeUtil.getGlobalMemStorePercent(conf,
215 false));
216 int chunkSize = conf.getInt(HeapMemStoreLAB.CHUNK_SIZE_KEY,
217 HeapMemStoreLAB.CHUNK_SIZE_DEFAULT);
218 int maxCount = (int) (globalMemStoreLimit * poolSizePercentage / chunkSize);
219
220 float initialCountPercentage = conf.getFloat(CHUNK_POOL_INITIALSIZE_KEY,
221 POOL_INITIAL_SIZE_DEFAULT);
222 if (initialCountPercentage > 1.0 || initialCountPercentage < 0) {
223 throw new IllegalArgumentException(CHUNK_POOL_INITIALSIZE_KEY
224 + " must be between 0.0 and 1.0");
225 }
226
227 int initialCount = (int) (initialCountPercentage * maxCount);
228 LOG.info("Allocating MemStoreChunkPool with chunk size " + StringUtils.byteDesc(chunkSize)
229 + ", max count " + maxCount + ", initial count " + initialCount);
230 GLOBAL_INSTANCE = new MemStoreChunkPool(conf, chunkSize, maxCount, initialCount);
231 return GLOBAL_INSTANCE;
232 }
233 }
234
235 int getMaxCount() {
236 return this.maxCount;
237 }
238
239 @VisibleForTesting
240 static void clearDisableFlag() {
241 chunkPoolDisabled = false;
242 }
243
244 }