<commit_before>package org.cache2k.impl;

/*
 * #%L
 * cache2k core package
 * %%
 * Copyright (C) 2000 - 2016 headissue GmbH, Munich
 * %%
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as
 * published by the Free Software Foundation, either version 3 of the 
 * License, or (at your option) any later version.
 * 
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public 
 * License along with this program.  If not, see
 * <http://www.gnu.org/licenses/gpl-3.0.html>.
 * #L%
 */

import org.cache2k.Cache;
import org.cache2k.CacheConfig;
import org.cache2k.ClosableIterator;
import org.cache2k.StorageConfiguration;
import org.cache2k.impl.threading.Futures;
import org.cache2k.impl.threading.LimitedPooledExecutor;
import org.cache2k.impl.timer.TimerListener;
import org.cache2k.impl.timer.TimerService;
import org.cache2k.spi.SingleProviderResolver;
import org.cache2k.storage.CacheStorage;
import org.cache2k.storage.CacheStorageContext;
import org.cache2k.storage.CacheStorageProvider;
import org.cache2k.storage.FlushableStorage;
import org.cache2k.storage.MarshallerFactory;
import org.cache2k.storage.Marshallers;
import org.cache2k.storage.TransientStorageClass;
import org.cache2k.storage.PurgeableStorage;
import org.cache2k.storage.StorageEntry;
import org.cache2k.impl.util.Log;
import org.cache2k.impl.util.TunableConstants;
import org.cache2k.impl.util.TunableFactory;

import java.io.NotSerializableException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Passes cache operation to the storage layer. Implements common
 * services for the storage layer. This class heavily interacts
 * with the base cache and contains mostly everything special
 * needed if a storage is defined. This means the design is not
 * perfectly layered, in some cases the
 * e.g. the get operation does interacts with the
 * underlying storage, wheres iterate
 *
 * @author <NAME>; created: 2014-05-08
 */
@SuppressWarnings({"unchecked", "SynchronizeOnNonFinalField"})
public class PassingStorageAdapter extends StorageAdapter {

  private Tunable tunable = TunableFactory.get(Tunable.class);
  private BaseCache cache;
  CacheStorage storage;
  boolean passivation = false;
  boolean storageIsTransient = false;
  long errorCount = 0;
  Set<Object> deletedKeys = null;
  StorageContext context;
  StorageConfiguration config;
  ExecutorService executor;

  long flushIntervalMillis = 0;
  Object flushLock = new Object();
  TimerService.CancelHandle flushTimerHandle;
  Future<Void> lastExecutingFlush = new Futures.FinishedFuture<Void>();

  Object purgeRunningLock = new Object();

  Log log;
  StorageAdapter.Parent parent;

  public PassingStorageAdapter(BaseCache _cache, CacheConfig _cacheConfig,
                               StorageConfiguration _storageConfig) {
    cache = _cache;
    parent = _cache;
    context = new StorageContext(_cache);
    context.keyType = _cacheConfig.getKeyType().getType();
    context.valueType = _cacheConfig.getValueType().getType();
    config = _storageConfig;
    if (tunable.useManagerThreadPool) {
      executor = new LimitedPooledExecutor(cache.manager.getThreadPool());
    } else {
      executor = Executors.newCachedThreadPool();
    }
    log = Log.getLog(Cache.class.getName() + ".storage/" + cache.getCompleteName());
    context.log = log;
  }

  /**
   * By default log lifecycle operations as info.
   */
  protected void logLifecycleOperation(String s) {
    log.info(s);
  }

  public void open() {
    try {
      CacheStorageProvider<?> pr = (CacheStorageProvider)
        SingleProviderResolver.getInstance().resolve(config.getImplementation());
      storage = pr.create(context, config);
      if (storage instanceof TransientStorageClass) {
        storageIsTransient = true;
      }
      flushIntervalMillis = config.getFlushIntervalMillis();
      if (!(storage instanceof FlushableStorage)) {
        flushIntervalMillis = -1;
      }
      if (config.isPassivation() || storageIsTransient) {
        deletedKeys = new HashSet<Object>();
        passivation = true;
      }
      logLifecycleOperation("opened, state: " + storage);
    } catch (Exception ex) {
      if (config.isReliable()) {
        disableAndThrow("error initializing, disabled", ex);
      } else {
        disable("error initializing, disabled", ex);
      }
    }
  }

  /**
   * Store entry on cache put. Entry must be locked, since we use the
   * entry directly for handing it over to the storage, it is not
   * allowed to change. The expiry time in the entry does not have
   * a valid value yet, so that is why it is transferred separately.
   */
  public void put(Entry e, long _nextRefreshTime) {
    if (passivation) {
      synchronized (deletedKeys) {
        deletedKeys.remove(e.getKey());
      }
      return;
    }
    StorageEntryForPut se =
        new StorageEntryForPut(e.getKey(), e.getValue(), e.getCreatedOrUpdated(), _nextRefreshTime);
    doPut(se);
  }

  private void doPut(StorageEntry e) {
    try {
      storage.put(e);
      checkStartFlushTimer();<commit_msg>fix exception passing<commit_after>package org.cache2k.impl;

/*
 * #%L
 * cache2k core package
 * %%
 * Copyright (C) 2000 - 2016 headissue GmbH, Munich
 * %%
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as
 * published by the Free Software Foundation, either version 3 of the 
 * License, or (at your option) any later version.
 * 
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public 
 * License along with this program.  If not, see
 * <http://www.gnu.org/licenses/gpl-3.0.html>.
 * #L%
 */

import org.cache2k.Cache;
import org.cache2k.CacheConfig;
import org.cache2k.ClosableIterator;
import org.cache2k.StorageConfiguration;
import org.cache2k.impl.threading.Futures;
import org.cache2k.impl.threading.LimitedPooledExecutor;
import org.cache2k.impl.timer.TimerListener;
import org.cache2k.impl.timer.TimerService;
import org.cache2k.spi.SingleProviderResolver;
import org.cache2k.storage.CacheStorage;
import org.cache2k.storage.CacheStorageContext;
import org.cache2k.storage.CacheStorageProvider;
import org.cache2k.storage.FlushableStorage;
import org.cache2k.storage.MarshallerFactory;
import org.cache2k.storage.Marshallers;
import org.cache2k.storage.TransientStorageClass;
import org.cache2k.storage.PurgeableStorage;
import org.cache2k.storage.StorageEntry;
import org.cache2k.impl.util.Log;
import org.cache2k.impl.util.TunableConstants;
import org.cache2k.impl.util.TunableFactory;

import java.io.NotSerializableException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Passes cache operation to the storage layer. Implements common
 * services for the storage layer. This class heavily interacts
 * with the base cache and contains mostly everything special
 * needed if a storage is defined. This means the design is not
 * perfectly layered, in some cases the
 * e.g. the get operation does interacts with the
 * underlying storage, wheres iterate
 *
 * @author <NAME>; created: 2014-05-08
 */
@SuppressWarnings({"unchecked", "SynchronizeOnNonFinalField"})
public class PassingStorageAdapter extends StorageAdapter {

  private Tunable tunable = TunableFactory.get(Tunable.class);
  private BaseCache cache;
  CacheStorage storage;
  boolean passivation = false;
  boolean storageIsTransient = false;
  long errorCount = 0;
  Set<Object> deletedKeys = null;
  StorageContext context;
  StorageConfiguration config;
  ExecutorService executor;

  long flushIntervalMillis = 0;
  Object flushLock = new Object();
  TimerService.CancelHandle flushTimerHandle;
  Future<Void> lastExecutingFlush = new Futures.FinishedFuture<Void>();

  Object purgeRunningLock = new Object();

  Log log;
  StorageAdapter.Parent parent;

  public PassingStorageAdapter(BaseCache _cache, CacheConfig _cacheConfig,
                               StorageConfiguration _storageConfig) {
    cache = _cache;
    parent = _cache;
    context = new StorageContext(_cache);
    context.keyType = _cacheConfig.getKeyType().getType();
    context.valueType = _cacheConfig.getValueType().getType();
    config = _storageConfig;
    if (tunable.useManagerThreadPool) {
      executor = new LimitedPooledExecutor(cache.manager.getThreadPool());
    } else {
      executor = Executors.newCachedThreadPool();
    }
    log = Log.getLog(Cache.class.getName() + ".storage/" + cache.getCompleteName());
    context.log = log;
  }

  /**
   * By default log lifecycle operations as info.
   */
  protected void logLifecycleOperation(String s) {
    log.info(s);
  }

  public void open() {
    try {
      CacheStorageProvider<?> pr = (CacheStorageProvider)
        SingleProviderResolver.getInstance().resolve(config.getImplementation());
      storage = pr.create(context, config);
      if (storage instanceof TransientStorageClass) {
        storageIsTransient = true;
      }
      flushIntervalMillis = config.getFlushIntervalMillis();
      if (!(storage instanceof FlushableStorage)) {
        flushIntervalMillis = -1;
      }
      if (config.isPassivation() || storageIsTransient) {
        deletedKeys = new HashSet<Object>();
        passivation = true;
      }
      logLifecycleOperation("opened, state: " + storage);
    } catch (Exception ex) {
      if (config.isReliable()) {
        disableAndThrow("error initializing, disabled", ex);
      } else {
        disable("error initializing, disabled", ex);
      }
    }
  }

  /**
   * Store entry on cache put. Entry must be locked, since we use the
   * entry directly for handing it over to the storage, it is not
   * allowed to change. The expiry time in the entry does not have
   * a valid value yet, so that is why it is transferred separately.
   */
  public void put(Entry e, long _nextRefreshTime) {
    if (passivation) {
      synchronized (deletedKeys) {
        deletedKeys.remove(e.getKey());
      }
      return;
    }
    StorageEntryForPut se =
        new StorageEntryForPut(e.getKey(), e.getValueOrException(), e.getCreatedOrUpdated(), _nextRefreshTime);
    doPut(se);
  }

  private void doPut(StorageEntry e) {
    try {
      storage.put(e);
      checkStartFlushTimer();
{'fraction_non_alphanumeric': 0.0936078723228314, 'fraction_numerical': 0.007442322004465393, 'mean_word_length': 3.9913330581923234, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5711234', 'n_tokens_mistral': 3469, 'n_tokens_neox': 3406, 'n_words': 963}