View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  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,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.regionserver;
20  
21  import static org.apache.hadoop.util.StringUtils.humanReadableInt;
22  
23  import java.io.IOException;
24  import java.lang.Thread.UncaughtExceptionHandler;
25  import java.lang.management.MemoryUsage;
26  import java.util.ArrayList;
27  import java.util.ConcurrentModificationException;
28  import java.util.HashMap;
29  import java.util.HashSet;
30  import java.util.List;
31  import java.util.Map;
32  import java.util.Set;
33  import java.util.SortedMap;
34  import java.util.concurrent.BlockingQueue;
35  import java.util.concurrent.DelayQueue;
36  import java.util.concurrent.Delayed;
37  import java.util.concurrent.ThreadFactory;
38  import java.util.concurrent.TimeUnit;
39  import java.util.concurrent.atomic.AtomicBoolean;
40  import java.util.concurrent.locks.ReentrantReadWriteLock;
41  
42  import org.apache.commons.logging.Log;
43  import org.apache.commons.logging.LogFactory;
44  import org.apache.hadoop.conf.Configuration;
45  import org.apache.hadoop.hbase.DroppedSnapshotException;
46  import org.apache.hadoop.hbase.HConstants;
47  import org.apache.hadoop.hbase.RemoteExceptionHandler;
48  import org.apache.hadoop.hbase.classification.InterfaceAudience;
49  import org.apache.hadoop.hbase.client.RegionReplicaUtil;
50  import org.apache.hadoop.hbase.io.util.HeapMemorySizeUtil;
51  import org.apache.hadoop.hbase.regionserver.Region.FlushResult;
52  import org.apache.hadoop.hbase.util.Bytes;
53  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
54  import org.apache.hadoop.hbase.util.HasThread;
55  import org.apache.hadoop.hbase.util.ServerRegionReplicaUtil;
56  import org.apache.hadoop.hbase.util.Threads;
57  import org.apache.hadoop.util.StringUtils;
58  import org.apache.hadoop.util.StringUtils.TraditionalBinaryPrefix;
59  import org.apache.htrace.Trace;
60  import org.apache.htrace.TraceScope;
61  import org.apache.hadoop.hbase.util.Counter;
62  
63  import com.google.common.base.Preconditions;
64  
65  /**
66   * Thread that flushes cache on request
67   *
68   * NOTE: This class extends Thread rather than Chore because the sleep time
69   * can be interrupted when there is something to do, rather than the Chore
70   * sleep time which is invariant.
71   *
72   * @see FlushRequester
73   */
74  @InterfaceAudience.Private
75  class MemStoreFlusher implements FlushRequester {
76    private static final Log LOG = LogFactory.getLog(MemStoreFlusher.class);
77  
78    private Configuration conf;
79    // These two data members go together.  Any entry in the one must have
80    // a corresponding entry in the other.
81    private final BlockingQueue<FlushQueueEntry> flushQueue =
82      new DelayQueue<FlushQueueEntry>();
83    private final Map<Region, FlushRegionEntry> regionsInQueue =
84      new HashMap<Region, FlushRegionEntry>();
85    private AtomicBoolean wakeupPending = new AtomicBoolean();
86  
87    private final long threadWakeFrequency;
88    private final HRegionServer server;
89    private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
90    private final Object blockSignal = new Object();
91  
92    protected long globalMemStoreLimit;
93    protected float globalMemStoreLimitLowMarkPercent;
94    protected long globalMemStoreLimitLowMark;
95  
96    private long blockingWaitTime;
97    private final Counter updatesBlockedMsHighWater = new Counter();
98  
99    private final FlushHandler[] flushHandlers;
100   private List<FlushRequestListener> flushRequestListeners = new ArrayList<FlushRequestListener>(1);
101 
102   /**
103    * @param conf
104    * @param server
105    */
106   public MemStoreFlusher(final Configuration conf,
107       final HRegionServer server) {
108     super();
109     this.conf = conf;
110     this.server = server;
111     this.threadWakeFrequency =
112       conf.getLong(HConstants.THREAD_WAKE_FREQUENCY, 10 * 1000);
113     long max = -1L;
114     final MemoryUsage usage = HeapMemorySizeUtil.safeGetHeapMemoryUsage();
115     if (usage != null) {
116       max = usage.getMax();
117     }
118     float globalMemStorePercent = HeapMemorySizeUtil.getGlobalMemStorePercent(conf, true);
119     this.globalMemStoreLimit = (long) (max * globalMemStorePercent);
120     this.globalMemStoreLimitLowMarkPercent =
121         HeapMemorySizeUtil.getGlobalMemStoreLowerMark(conf, globalMemStorePercent);
122     this.globalMemStoreLimitLowMark =
123         (long) (this.globalMemStoreLimit * this.globalMemStoreLimitLowMarkPercent);
124 
125     this.blockingWaitTime = conf.getInt("hbase.hstore.blockingWaitTime",
126       90000);
127     int handlerCount = conf.getInt("hbase.hstore.flusher.count", 2);
128     this.flushHandlers = new FlushHandler[handlerCount];
129     LOG.info("globalMemStoreLimit="
130         + TraditionalBinaryPrefix.long2String(this.globalMemStoreLimit, "", 1)
131         + ", globalMemStoreLimitLowMark="
132         + TraditionalBinaryPrefix.long2String(this.globalMemStoreLimitLowMark, "", 1)
133         + ", maxHeap=" + TraditionalBinaryPrefix.long2String(max, "", 1));
134   }
135 
136   public Counter getUpdatesBlockedMsHighWater() {
137     return this.updatesBlockedMsHighWater;
138   }
139 
140   /**
141    * The memstore across all regions has exceeded the low water mark. Pick
142    * one region to flush and flush it synchronously (this is called from the
143    * flush thread)
144    * @return true if successful
145    */
146   private boolean flushOneForGlobalPressure() {
147     SortedMap<Long, Region> regionsBySize = server.getCopyOfOnlineRegionsSortedBySize();
148     Set<Region> excludedRegions = new HashSet<Region>();
149 
150     double secondaryMultiplier
151       = ServerRegionReplicaUtil.getRegionReplicaStoreFileRefreshMultiplier(conf);
152 
153     boolean flushedOne = false;
154     while (!flushedOne) {
155       // Find the biggest region that doesn't have too many storefiles
156       // (might be null!)
157       Region bestFlushableRegion = getBiggestMemstoreRegion(regionsBySize, excludedRegions, true);
158       // Find the biggest region, total, even if it might have too many flushes.
159       Region bestAnyRegion = getBiggestMemstoreRegion(
160           regionsBySize, excludedRegions, false);
161       // Find the biggest region that is a secondary region
162       Region bestRegionReplica = getBiggestMemstoreOfRegionReplica(regionsBySize,
163         excludedRegions);
164 
165       if (bestAnyRegion == null && bestRegionReplica == null) {
166         LOG.error("Above memory mark but there are no flushable regions!");
167         return false;
168       }
169 
170       Region regionToFlush;
171       if (bestFlushableRegion != null &&
172           bestAnyRegion.getMemstoreSize() > 2 * bestFlushableRegion.getMemstoreSize()) {
173         // Even if it's not supposed to be flushed, pick a region if it's more than twice
174         // as big as the best flushable one - otherwise when we're under pressure we make
175         // lots of little flushes and cause lots of compactions, etc, which just makes
176         // life worse!
177         if (LOG.isDebugEnabled()) {
178           LOG.debug("Under global heap pressure: " + "Region "
179               + bestAnyRegion.getRegionInfo().getRegionNameAsString()
180               + " has too many " + "store files, but is "
181               + TraditionalBinaryPrefix.long2String(bestAnyRegion.getMemstoreSize(), "", 1)
182               + " vs best flushable region's "
183               + TraditionalBinaryPrefix.long2String(bestFlushableRegion.getMemstoreSize(), "", 1)
184               + ". Choosing the bigger.");
185         }
186         regionToFlush = bestAnyRegion;
187       } else {
188         if (bestFlushableRegion == null) {
189           regionToFlush = bestAnyRegion;
190         } else {
191           regionToFlush = bestFlushableRegion;
192         }
193       }
194 
195       Preconditions.checkState(
196         (regionToFlush != null && regionToFlush.getMemstoreSize() > 0) ||
197         (bestRegionReplica != null && bestRegionReplica.getMemstoreSize() > 0));
198 
199       if (regionToFlush == null ||
200           (bestRegionReplica != null &&
201            ServerRegionReplicaUtil.isRegionReplicaStoreFileRefreshEnabled(conf) &&
202            (bestRegionReplica.getMemstoreSize()
203                > secondaryMultiplier * regionToFlush.getMemstoreSize()))) {
204         LOG.info("Refreshing storefiles of region " + bestRegionReplica +
205           " due to global heap pressure. memstore size=" + StringUtils.humanReadableInt(
206             server.getRegionServerAccounting().getGlobalMemstoreSize()));
207         flushedOne = refreshStoreFilesAndReclaimMemory(bestRegionReplica);
208         if (!flushedOne) {
209           LOG.info("Excluding secondary region " + bestRegionReplica +
210               " - trying to find a different region to refresh files.");
211           excludedRegions.add(bestRegionReplica);
212         }
213       } else {
214         LOG.info("Flush of region " + regionToFlush + " due to global heap pressure. "
215             + "Total Memstore size="
216             + humanReadableInt(server.getRegionServerAccounting().getGlobalMemstoreSize())
217             + ", Region memstore size="
218             + humanReadableInt(regionToFlush.getMemstoreSize()));
219         flushedOne = flushRegion(regionToFlush, true, false);
220 
221         if (!flushedOne) {
222           LOG.info("Excluding unflushable region " + regionToFlush +
223               " - trying to find a different region to flush.");
224           excludedRegions.add(regionToFlush);
225         }
226       }
227     }
228     return true;
229   }
230 
231   private class FlushHandler extends HasThread {
232 
233     private FlushHandler(String name) {
234       super(name);
235     }
236 
237     @Override
238     public void run() {
239       while (!server.isStopped()) {
240         FlushQueueEntry fqe = null;
241         try {
242           wakeupPending.set(false); // allow someone to wake us up again
243           fqe = flushQueue.poll(threadWakeFrequency, TimeUnit.MILLISECONDS);
244           if (fqe == null || fqe instanceof WakeupFlushThread) {
245             if (isAboveLowWaterMark()) {
246               LOG.debug("Flush thread woke up because memory above low water="
247                   + TraditionalBinaryPrefix.long2String(globalMemStoreLimitLowMark, "", 1));
248               if (!flushOneForGlobalPressure()) {
249                 // Wasn't able to flush any region, but we're above low water mark
250                 // This is unlikely to happen, but might happen when closing the
251                 // entire server - another thread is flushing regions. We'll just
252                 // sleep a little bit to avoid spinning, and then pretend that
253                 // we flushed one, so anyone blocked will check again
254                 Thread.sleep(1000);
255                 wakeUpIfBlocking();
256               }
257               // Enqueue another one of these tokens so we'll wake up again
258               wakeupFlushThread();
259             }
260             continue;
261           }
262           FlushRegionEntry fre = (FlushRegionEntry) fqe;
263           if (!flushRegion(fre)) {
264             break;
265           }
266         } catch (InterruptedException ex) {
267           continue;
268         } catch (ConcurrentModificationException ex) {
269           continue;
270         } catch (Exception ex) {
271           LOG.error("Cache flusher failed for entry " + fqe, ex);
272           if (!server.checkFileSystem()) {
273             break;
274           }
275         }
276       }
277       synchronized (regionsInQueue) {
278         regionsInQueue.clear();
279         flushQueue.clear();
280       }
281 
282       // Signal anyone waiting, so they see the close flag
283       wakeUpIfBlocking();
284       LOG.info(getName() + " exiting");
285     }
286   }
287 
288 
289   private void wakeupFlushThread() {
290     if (wakeupPending.compareAndSet(false, true)) {
291       flushQueue.add(new WakeupFlushThread());
292     }
293   }
294 
295   private Region getBiggestMemstoreRegion(
296       SortedMap<Long, Region> regionsBySize,
297       Set<Region> excludedRegions,
298       boolean checkStoreFileCount) {
299     synchronized (regionsInQueue) {
300       for (Region region : regionsBySize.values()) {
301         if (excludedRegions.contains(region)) {
302           continue;
303         }
304 
305         if (((HRegion)region).writestate.flushing ||
306             !((HRegion)region).writestate.writesEnabled) {
307           continue;
308         }
309 
310         if (checkStoreFileCount && isTooManyStoreFiles(region)) {
311           continue;
312         }
313         return region;
314       }
315     }
316     return null;
317   }
318 
319   private Region getBiggestMemstoreOfRegionReplica(SortedMap<Long, Region> regionsBySize,
320       Set<Region> excludedRegions) {
321     synchronized (regionsInQueue) {
322       for (Region region : regionsBySize.values()) {
323         if (excludedRegions.contains(region)) {
324           continue;
325         }
326 
327         if (RegionReplicaUtil.isDefaultReplica(region.getRegionInfo())) {
328           continue;
329         }
330 
331         return region;
332       }
333     }
334     return null;
335   }
336 
337   private boolean refreshStoreFilesAndReclaimMemory(Region region) {
338     try {
339       return region.refreshStoreFiles();
340     } catch (IOException e) {
341       LOG.warn("Refreshing store files failed with exception", e);
342     }
343     return false;
344   }
345 
346   /**
347    * Return true if global memory usage is above the high watermark
348    */
349   private boolean isAboveHighWaterMark() {
350     return server.getRegionServerAccounting().
351       getGlobalMemstoreSize() >= globalMemStoreLimit;
352   }
353 
354   /**
355    * Return true if we're above the high watermark
356    */
357   private boolean isAboveLowWaterMark() {
358     return server.getRegionServerAccounting().
359       getGlobalMemstoreSize() >= globalMemStoreLimitLowMark;
360   }
361 
362   @Override
363   public boolean requestFlush(Region r, boolean forceFlushAllStores) {
364     synchronized (regionsInQueue) {
365       if (!regionsInQueue.containsKey(r)) {
366         // This entry has no delay so it will be added at the top of the flush
367         // queue.  It'll come out near immediately.
368         FlushRegionEntry fqe = new FlushRegionEntry(r, forceFlushAllStores);
369         this.regionsInQueue.put(r, fqe);
370         this.flushQueue.add(fqe);
371         ((HRegion)r).incrementFlushesQueuedCount();
372         return true;
373       }
374       return false;
375     }
376   }
377 
378   @Override
379   public boolean requestDelayedFlush(Region r, long delay, boolean forceFlushAllStores) {
380     synchronized (regionsInQueue) {
381       if (!regionsInQueue.containsKey(r)) {
382         // This entry has some delay
383         FlushRegionEntry fqe = new FlushRegionEntry(r, forceFlushAllStores);
384         fqe.requeue(delay);
385         this.regionsInQueue.put(r, fqe);
386         this.flushQueue.add(fqe);
387         ((HRegion)r).incrementFlushesQueuedCount();
388         return true;
389       }
390       return false;
391     }
392   }
393 
394   public int getFlushQueueSize() {
395     return flushQueue.size();
396   }
397 
398   /**
399    * Only interrupt once it's done with a run through the work loop.
400    */
401   void interruptIfNecessary() {
402     lock.writeLock().lock();
403     try {
404       for (FlushHandler flushHander : flushHandlers) {
405         if (flushHander != null) flushHander.interrupt();
406       }
407     } finally {
408       lock.writeLock().unlock();
409     }
410   }
411 
412   synchronized void start(UncaughtExceptionHandler eh) {
413     ThreadFactory flusherThreadFactory = Threads.newDaemonThreadFactory(
414         server.getServerName().toShortString() + "-MemStoreFlusher", eh);
415     for (int i = 0; i < flushHandlers.length; i++) {
416       flushHandlers[i] = new FlushHandler("MemStoreFlusher." + i);
417       flusherThreadFactory.newThread(flushHandlers[i]);
418       flushHandlers[i].start();
419     }
420   }
421 
422   boolean isAlive() {
423     for (FlushHandler flushHander : flushHandlers) {
424       if (flushHander != null && flushHander.isAlive()) {
425         return true;
426       }
427     }
428     return false;
429   }
430 
431   void join() {
432     for (FlushHandler flushHander : flushHandlers) {
433       if (flushHander != null) {
434         Threads.shutdown(flushHander.getThread());
435       }
436     }
437   }
438 
439   /**
440    * A flushRegion that checks store file count.  If too many, puts the flush
441    * on delay queue to retry later.
442    * @param fqe
443    * @return true if the region was successfully flushed, false otherwise. If
444    * false, there will be accompanying log messages explaining why the region was
445    * not flushed.
446    */
447   private boolean flushRegion(final FlushRegionEntry fqe) {
448     Region region = fqe.region;
449     if (!region.getRegionInfo().isMetaRegion() &&
450         isTooManyStoreFiles(region)) {
451       if (fqe.isMaximumWait(this.blockingWaitTime)) {
452         LOG.info("Waited " + (EnvironmentEdgeManager.currentTime() - fqe.createTime) +
453           "ms on a compaction to clean up 'too many store files'; waited " +
454           "long enough... proceeding with flush of " +
455           region.getRegionInfo().getRegionNameAsString());
456       } else {
457         // If this is first time we've been put off, then emit a log message.
458         if (fqe.getRequeueCount() <= 0) {
459           // Note: We don't impose blockingStoreFiles constraint on meta regions
460           LOG.warn("Region " + region.getRegionInfo().getRegionNameAsString() + " has too many " +
461             "store files; delaying flush up to " + this.blockingWaitTime + "ms");
462           if (!this.server.compactSplitThread.requestSplit(region)) {
463             try {
464               this.server.compactSplitThread.requestSystemCompaction(
465                   region, Thread.currentThread().getName());
466             } catch (IOException e) {
467               LOG.error("Cache flush failed for region " +
468                   Bytes.toStringBinary(region.getRegionInfo().getRegionName()),
469                 RemoteExceptionHandler.checkIOException(e));
470             }
471           }
472         }
473 
474         // Put back on the queue.  Have it come back out of the queue
475         // after a delay of this.blockingWaitTime / 100 ms.
476         this.flushQueue.add(fqe.requeue(this.blockingWaitTime / 100));
477         // Tell a lie, it's not flushed but it's ok
478         return true;
479       }
480     }
481     return flushRegion(region, false, fqe.isForceFlushAllStores());
482   }
483 
484   /**
485    * Flush a region.
486    * @param region Region to flush.
487    * @param emergencyFlush Set if we are being force flushed. If true the region
488    * needs to be removed from the flush queue. If false, when we were called
489    * from the main flusher run loop and we got the entry to flush by calling
490    * poll on the flush queue (which removed it).
491    * @param forceFlushAllStores whether we want to flush all store.
492    * @return true if the region was successfully flushed, false otherwise. If
493    * false, there will be accompanying log messages explaining why the region was
494    * not flushed.
495    */
496   private boolean flushRegion(final Region region, final boolean emergencyFlush,
497       boolean forceFlushAllStores) {
498     synchronized (this.regionsInQueue) {
499       FlushRegionEntry fqe = this.regionsInQueue.remove(region);
500       // Use the start time of the FlushRegionEntry if available
501       if (fqe != null && emergencyFlush) {
502         // Need to remove from region from delay queue.  When NOT an
503         // emergencyFlush, then item was removed via a flushQueue.poll.
504         flushQueue.remove(fqe);
505       }
506     }
507 
508     lock.readLock().lock();
509     try {
510       notifyFlushRequest(region, emergencyFlush);
511       FlushResult flushResult = region.flush(forceFlushAllStores);
512       boolean shouldCompact = flushResult.isCompactionNeeded();
513       // We just want to check the size
514       boolean shouldSplit = ((HRegion)region).checkSplit() != null;
515       if (shouldSplit) {
516         this.server.compactSplitThread.requestSplit(region);
517       } else if (shouldCompact) {
518         server.compactSplitThread.requestSystemCompaction(
519             region, Thread.currentThread().getName());
520       }
521     } catch (DroppedSnapshotException ex) {
522       // Cache flush can fail in a few places. If it fails in a critical
523       // section, we get a DroppedSnapshotException and a replay of wal
524       // is required. Currently the only way to do this is a restart of
525       // the server. Abort because hdfs is probably bad (HBASE-644 is a case
526       // where hdfs was bad but passed the hdfs check).
527       server.abort("Replay of WAL required. Forcing server shutdown", ex);
528       return false;
529     } catch (IOException ex) {
530       LOG.error("Cache flush failed" + (region != null ? (" for region " +
531           Bytes.toStringBinary(region.getRegionInfo().getRegionName())) : ""),
532         RemoteExceptionHandler.checkIOException(ex));
533       if (!server.checkFileSystem()) {
534         return false;
535       }
536     } finally {
537       lock.readLock().unlock();
538       wakeUpIfBlocking();
539     }
540     return true;
541   }
542 
543   private void notifyFlushRequest(Region region, boolean emergencyFlush) {
544     FlushType type = FlushType.NORMAL;
545     if (emergencyFlush) {
546       type = isAboveHighWaterMark() ? FlushType.ABOVE_HIGHER_MARK : FlushType.ABOVE_LOWER_MARK;
547     }
548     for (FlushRequestListener listener : flushRequestListeners) {
549       listener.flushRequested(type, region);
550     }
551   }
552 
553   private void wakeUpIfBlocking() {
554     synchronized (blockSignal) {
555       blockSignal.notifyAll();
556     }
557   }
558 
559   private boolean isTooManyStoreFiles(Region region) {
560 
561     // When compaction is disabled, the region is flushable
562     if (!region.getTableDesc().isCompactionEnabled()) {
563       return false;
564     }
565 
566     for (Store store : region.getStores()) {
567       if (store.hasTooManyStoreFiles()) {
568         return true;
569       }
570     }
571     return false;
572   }
573 
574   /**
575    * Check if the regionserver's memstore memory usage is greater than the
576    * limit. If so, flush regions with the biggest memstores until we're down
577    * to the lower limit. This method blocks callers until we're down to a safe
578    * amount of memstore consumption.
579    */
580   public void reclaimMemStoreMemory() {
581     TraceScope scope = Trace.startSpan("MemStoreFluser.reclaimMemStoreMemory");
582     if (isAboveHighWaterMark()) {
583       if (Trace.isTracing()) {
584         scope.getSpan().addTimelineAnnotation("Force Flush. We're above high water mark.");
585       }
586       long start = EnvironmentEdgeManager.currentTime();
587       synchronized (this.blockSignal) {
588         boolean blocked = false;
589         long startTime = 0;
590         boolean interrupted = false;
591         try {
592           while (isAboveHighWaterMark() && !server.isStopped()) {
593             if (!blocked) {
594               startTime = EnvironmentEdgeManager.currentTime();
595               LOG.info("Blocking updates on "
596                   + server.toString()
597                   + ": the global memstore size "
598                   + TraditionalBinaryPrefix.long2String(server.getRegionServerAccounting()
599                       .getGlobalMemstoreSize(), "", 1) + " is >= than blocking "
600                   + TraditionalBinaryPrefix.long2String(globalMemStoreLimit, "", 1) + " size");
601             }
602             blocked = true;
603             wakeupFlushThread();
604             try {
605               // we should be able to wait forever, but we've seen a bug where
606               // we miss a notify, so put a 5 second bound on it at least.
607               blockSignal.wait(5 * 1000);
608             } catch (InterruptedException ie) {
609               LOG.warn("Interrupted while waiting");
610               interrupted = true;
611             }
612             long took = EnvironmentEdgeManager.currentTime() - start;
613             LOG.warn("Memstore is above high water mark and block " + took + "ms");
614           }
615         } finally {
616           if (interrupted) {
617             Thread.currentThread().interrupt();
618           }
619         }
620 
621         if(blocked){
622           final long totalTime = EnvironmentEdgeManager.currentTime() - startTime;
623           if(totalTime > 0){
624             this.updatesBlockedMsHighWater.add(totalTime);
625           }
626           LOG.info("Unblocking updates for server " + server.toString());
627         }
628       }
629     } else if (isAboveLowWaterMark()) {
630       wakeupFlushThread();
631     }
632     scope.close();
633   }
634   @Override
635   public String toString() {
636     return "flush_queue="
637         + flushQueue.size();
638   }
639 
640   public String dumpQueue() {
641     StringBuilder queueList = new StringBuilder();
642     queueList.append("Flush Queue Queue dump:\n");
643     queueList.append("  Flush Queue:\n");
644     java.util.Iterator<FlushQueueEntry> it = flushQueue.iterator();
645 
646     while(it.hasNext()){
647       queueList.append("    "+it.next().toString());
648       queueList.append("\n");
649     }
650 
651     return queueList.toString();
652   }
653 
654   /**
655    * Register a MemstoreFlushListener
656    * @param listener
657    */
658   @Override
659   public void registerFlushRequestListener(final FlushRequestListener listener) {
660     this.flushRequestListeners.add(listener);
661   }
662 
663   /**
664    * Unregister the listener from MemstoreFlushListeners
665    * @param listener
666    * @return true when passed listener is unregistered successfully.
667    */
668   @Override
669   public boolean unregisterFlushRequestListener(final FlushRequestListener listener) {
670     return this.flushRequestListeners.remove(listener);
671   }
672 
673   /**
674    * Sets the global memstore limit to a new size.
675    * @param globalMemStoreSize
676    */
677   @Override
678   public void setGlobalMemstoreLimit(long globalMemStoreSize) {
679     this.globalMemStoreLimit = globalMemStoreSize;
680     this.globalMemStoreLimitLowMark =
681         (long) (this.globalMemStoreLimitLowMarkPercent * globalMemStoreSize);
682     reclaimMemStoreMemory();
683   }
684 
685   public long getMemoryLimit() {
686     return this.globalMemStoreLimit;
687   }
688 
689   interface FlushQueueEntry extends Delayed {
690   }
691 
692   /**
693    * Token to insert into the flush queue that ensures that the flusher does not sleep
694    */
695   static class WakeupFlushThread implements FlushQueueEntry {
696     @Override
697     public long getDelay(TimeUnit unit) {
698       return 0;
699     }
700 
701     @Override
702     public int compareTo(Delayed o) {
703       return -1;
704     }
705 
706     @Override
707     public boolean equals(Object obj) {
708       return (this == obj);
709     }
710   }
711 
712   /**
713    * Datastructure used in the flush queue.  Holds region and retry count.
714    * Keeps tabs on how old this object is.  Implements {@link Delayed}.  On
715    * construction, the delay is zero. When added to a delay queue, we'll come
716    * out near immediately.  Call {@link #requeue(long)} passing delay in
717    * milliseconds before readding to delay queue if you want it to stay there
718    * a while.
719    */
720   static class FlushRegionEntry implements FlushQueueEntry {
721     private final Region region;
722 
723     private final long createTime;
724     private long whenToExpire;
725     private int requeueCount = 0;
726 
727     private boolean forceFlushAllStores;
728 
729     FlushRegionEntry(final Region r, boolean forceFlushAllStores) {
730       this.region = r;
731       this.createTime = EnvironmentEdgeManager.currentTime();
732       this.whenToExpire = this.createTime;
733       this.forceFlushAllStores = forceFlushAllStores;
734     }
735 
736     /**
737      * @param maximumWait
738      * @return True if we have been delayed > <code>maximumWait</code> milliseconds.
739      */
740     public boolean isMaximumWait(final long maximumWait) {
741       return (EnvironmentEdgeManager.currentTime() - this.createTime) > maximumWait;
742     }
743 
744     /**
745      * @return Count of times {@link #requeue(long)} was called; i.e this is
746      * number of times we've been requeued.
747      */
748     public int getRequeueCount() {
749       return this.requeueCount;
750     }
751 
752     /**
753      * @return whether we need to flush all stores.
754      */
755     public boolean isForceFlushAllStores() {
756       return forceFlushAllStores;
757     }
758 
759     /**
760      * @param when When to expire, when to come up out of the queue.
761      * Specify in milliseconds.  This method adds EnvironmentEdgeManager.currentTime()
762      * to whatever you pass.
763      * @return This.
764      */
765     public FlushRegionEntry requeue(final long when) {
766       this.whenToExpire = EnvironmentEdgeManager.currentTime() + when;
767       this.requeueCount++;
768       return this;
769     }
770 
771     @Override
772     public long getDelay(TimeUnit unit) {
773       return unit.convert(this.whenToExpire - EnvironmentEdgeManager.currentTime(),
774           TimeUnit.MILLISECONDS);
775     }
776 
777     @Override
778     public int compareTo(Delayed other) {
779       // Delay is compared first. If there is a tie, compare region's hash code
780       int ret = Long.valueOf(getDelay(TimeUnit.MILLISECONDS) -
781         other.getDelay(TimeUnit.MILLISECONDS)).intValue();
782       if (ret != 0) {
783         return ret;
784       }
785       FlushQueueEntry otherEntry = (FlushQueueEntry) other;
786       return hashCode() - otherEntry.hashCode();
787     }
788 
789     @Override
790     public String toString() {
791       return "[flush region "+Bytes.toStringBinary(region.getRegionInfo().getRegionName())+"]";
792     }
793 
794     @Override
795     public int hashCode() {
796       int hash = (int) getDelay(TimeUnit.MILLISECONDS);
797       return hash ^ region.hashCode();
798     }
799 
800    @Override
801     public boolean equals(Object obj) {
802       if (this == obj) {
803         return true;
804       }
805       if (obj == null || getClass() != obj.getClass()) {
806         return false;
807       }
808       Delayed other = (Delayed) obj;
809       return compareTo(other) == 0;
810     }
811   }
812 }