class OffHeapCache extends AbstractSearchableCache<K, V> {

  protected void cleanBuffers(final float bufferCleanerThreshold) {
    // Buffers can be fully dirty and we may not find any pointer to resolve that.
    // This case is so unlikely that we did not consider.
    final Set<Entry<K, Pointer>> pointers = pointerMap.entrySet();
    final Map<OffHeapByteBuffer, Float> buffers = new HashMap<>();
    final Set<Integer> buffersToBeCleaned = new HashSet<Integer>();
    // For all pointers we try to understand if there is a need for redistribution.
    for (final Entry<K, Pointer> pointerEntry : pointers) {
      Float ratio = buffers.get(pointerEntry.getValue().getOffHeapByteBuffer());
      if (ratio == null) {
        // calculate the ratio of the dirty
        ratio = getDirtyRatio(pointerEntry.getValue());
        buffers.put(pointerEntry.getValue().getOffHeapByteBuffer(), ratio);
      }
      if (ratio - bufferCleanerThreshold > DELTA) {
        buffersToBeCleaned.add(pointerEntry.getValue().getOffHeapByteBuffer().getIndex());
        writeLock(pointerEntry.getKey());
        try {
          final byte[] payload = bufferStore.retrieve(pointerEntry.getValue());
          final Pointer newPointer = bufferStore
              .store(payload, pointerEntry.getValue().getExpiry());
          pointerMap.put(pointerEntry.getKey(), newPointer);
        } finally {
          writeUnlock(pointerEntry.getKey());
        }
      }
    }
    for (final int bufferIndex : buffersToBeCleaned) {
      bufferStore.free(bufferIndex);
    }
  }

  public  OffHeapCache(final String name, final CacheLoader<K, V> cacheLoader,
      final EvictionListener<K, V> evictionListener,
      final IndexHandler<K, V> indexHandler, final OffHeapByteBufferStore byteBufferStore,
      final Serializer<V> serializer,
      final long bufferCleanerPeriod, final float bufferCleanerThreshold,
      final int concurrencyLevel,
      final long evictionPeriod);

}