View Javadoc

1   /**
2    * Copyright The Apache Software Foundation
3    *
4    * Licensed to the Apache Software Foundation (ASF) under one or more
5    * contributor license agreements. See the NOTICE file distributed with this
6    * work for additional information regarding copyright ownership. The ASF
7    * licenses this file to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance with the License.
9    * You may obtain a copy of the License at
10   *
11   * http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
15   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
16   * License for the specific language governing permissions and limitations
17   * under the License.
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   * A pool of {@link HeapMemStoreLAB.Chunk} instances.
42   * 
43   * MemStoreChunkPool caches a number of retired chunks for reusing, it could
44   * decrease allocating bytes when writing, thereby optimizing the garbage
45   * collection on JVM.
46   * 
47   * The pool instance is globally unique and could be obtained through
48   * {@link MemStoreChunkPool#getPool(Configuration)}
49   * 
50   * {@link MemStoreChunkPool#getChunk()} is called when MemStoreLAB allocating
51   * bytes, and {@link MemStoreChunkPool#putbackChunks(BlockingQueue)} is called
52   * when MemStore clearing snapshot for flush
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    // Static reference to the MemStoreChunkPool
64    private static MemStoreChunkPool GLOBAL_INSTANCE;
65    /** Boolean whether we have disabled the memstore chunk pool entirely. */
66    static boolean chunkPoolDisabled = false;
67  
68    private final int maxCount;
69  
70    // A queue of reclaimed chunks
71    private final BlockingQueue<Chunk> reclaimedChunks;
72    private final int chunkSize;
73  
74    /** Statistics thread schedule pool */
75    private final ScheduledExecutorService scheduleThreadPool;
76    /** Statistics thread */
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    * Poll a chunk from the pool, reset it if not null, else create a new chunk
102    * to return
103    * @return a chunk
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    * Add the chunks to the pool, when the pool achieves the max size, it will
120    * skip the remaining chunks
121    * @param chunks
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     // clear reference of any non-reclaimable chunks
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    * Add the chunk to the pool, if the pool has achieved the max size, it will
140    * skip it
141    * @param chunk
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    * Only used in testing
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    * @param conf
191    * @return the global MemStoreChunkPool instance
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 }