<commit_before>package de.fernunihagen.dna.jkn.scalephant.storage.sstable;

import java.io.File;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import de.fernunihagen.dna.jkn.scalephant.ScalephantConfiguration;
import de.fernunihagen.dna.jkn.scalephant.ScalephantService;
import de.fernunihagen.dna.jkn.scalephant.storage.Memtable;
import de.fernunihagen.dna.jkn.scalephant.storage.Storage;
import de.fernunihagen.dna.jkn.scalephant.storage.StorageManagerException;
import de.fernunihagen.dna.jkn.scalephant.storage.entity.BoundingBox;
import de.fernunihagen.dna.jkn.scalephant.storage.entity.DeletedTuple;
import de.fernunihagen.dna.jkn.scalephant.storage.entity.SSTableName;
import de.fernunihagen.dna.jkn.scalephant.storage.entity.Tuple;
import de.fernunihagen.dna.jkn.scalephant.storage.sstable.compact.SSTableCompactorThread;
import de.fernunihagen.dna.jkn.scalephant.storage.sstable.reader.SSTableFacade;
import de.fernunihagen.dna.jkn.scalephant.storage.sstable.reader.SSTableKeyIndexReader;
import de.fernunihagen.dna.jkn.scalephant.storage.sstable.reader.SSTableReader;
import de.fernunihagen.dna.jkn.scalephant.util.State;

public class SSTableManager implements ScalephantService, Storage {
	
	/**
	 * The name of the table
	 */
	protected final SSTableName name;
	
	/**
	 * The Storage configuration
	 */
	protected final ScalephantConfiguration storageConfiguration;
	
	/**
	 * The number of the table
	 */
	protected AtomicInteger tableNumber;
	
	/**
	 * The reader for existing SSTables
	 */
	protected final List<SSTableFacade> sstableFacades;

	/**
	 * The unflushed memtables
	 */
	protected List<Memtable> unflushedMemtables;

	/**
	 * Ready flag for flush thread
	 */
	protected volatile boolean ready;

	/**
	 * The Flush thread instance
	 */
	protected SSTableFlushThread ssTableFlushThread;
	
	/**
	 * The corresponding storage manager state
	 */
	protected State storageState;
	
	/**
	 * The running threads
	 */
	protected final Set<Thread> runningThreads;
	
	/**
	 * The timeout for a thread join (10 seconds)
	 */
	protected long THREAD_WAIT_TIMEOUT = TimeUnit.SECONDS.toMillis(10);
	
	/**
	 * The logger
	 */
	private final static Logger logger = LoggerFactory.getLogger(SSTableManager.class);

	public SSTableManager(final State storageState, final SSTableName name, final ScalephantConfiguration storageConfiguration) {
		super();

		this.storageConfiguration = storageConfiguration;
		this.storageState = storageState;
		this.name = name;
		this.tableNumber = new AtomicInteger();
		this.tableNumber.set(0);
		this.ready = false;
		
		this.unflushedMemtables = new CopyOnWriteArrayList<Memtable>();
		this.sstableFacades = new CopyOnWriteArrayList<SSTableFacade>();
		this.runningThreads = new HashSet<Thread>();
	}

	/**
	 * Init the instance
	 * 
	 */
	@Override
	public void init() {
		
		if(ready == true) {
			logger.warn("SSTable manager is active and init() is called");
			return;
		}
		
		logger.info("Init a new instance for the table: " + getName());
		unflushedMemtables.clear();
		sstableFacades.clear();
		createSSTableDirIfNeeded();
		
		try {
			scanForExistingTables();
		} catch (StorageManagerException e) {
			// Unable to init the instance
			return;
		}
		
		this.tableNumber = new AtomicInteger();
		this.tableNumber.set(getLastSequencenumberFromReader());

		// Set to ready before the threads are started
		ready = true;

		if(storageConfiguration.isStorageRunMemtableFlushThread()) {
			ssTableFlushThread = new SSTableFlushThread(this);
			final Thread flushThread = new Thread(ssTableFlushThread);
			flushThread.setName("Memtable flush thread for: " + getName());
			flushThread.start();
			runningThreads.add(flushThread);
		} else {
			logger.info("NOT starting the memtable flush thread.");
		}
		
		if(storageConfiguration.isStorageRunCompactThread()) {
			final Thread compactThread = new Thread(new SSTableCompactorThread(this));
			compactThread.setName("Compact thread for: " + getName());
			compactThread.start();
			runningThreads.add(compactThread);
		} else {
			logger.info("NOT starting the sstable compact thread.");
		}
	}

	/**
	 * Shutdown the instance
	 * 
	 */
	@Override
	public void shutdown() {
		logger.info("Shuting down the instance for table: " + getName());
		
		// Set ready to false. The threads will shutdown after completing
		// the running tasks
		ready = false;<commit_msg>Cleanup the init method<commit_after>package de.fernunihagen.dna.jkn.scalephant.storage.sstable;

import java.io.File;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import de.fernunihagen.dna.jkn.scalephant.ScalephantConfiguration;
import de.fernunihagen.dna.jkn.scalephant.ScalephantService;
import de.fernunihagen.dna.jkn.scalephant.storage.Memtable;
import de.fernunihagen.dna.jkn.scalephant.storage.Storage;
import de.fernunihagen.dna.jkn.scalephant.storage.StorageManagerException;
import de.fernunihagen.dna.jkn.scalephant.storage.entity.BoundingBox;
import de.fernunihagen.dna.jkn.scalephant.storage.entity.DeletedTuple;
import de.fernunihagen.dna.jkn.scalephant.storage.entity.SSTableName;
import de.fernunihagen.dna.jkn.scalephant.storage.entity.Tuple;
import de.fernunihagen.dna.jkn.scalephant.storage.sstable.compact.SSTableCompactorThread;
import de.fernunihagen.dna.jkn.scalephant.storage.sstable.reader.SSTableFacade;
import de.fernunihagen.dna.jkn.scalephant.storage.sstable.reader.SSTableKeyIndexReader;
import de.fernunihagen.dna.jkn.scalephant.storage.sstable.reader.SSTableReader;
import de.fernunihagen.dna.jkn.scalephant.util.State;

public class SSTableManager implements ScalephantService, Storage {
	
	/**
	 * The name of the table
	 */
	protected final SSTableName name;
	
	/**
	 * The Storage configuration
	 */
	protected final ScalephantConfiguration storageConfiguration;
	
	/**
	 * The number of the table
	 */
	protected AtomicInteger tableNumber;
	
	/**
	 * The reader for existing SSTables
	 */
	protected final List<SSTableFacade> sstableFacades;

	/**
	 * The unflushed memtables
	 */
	protected List<Memtable> unflushedMemtables;

	/**
	 * Ready flag for flush thread
	 */
	protected volatile boolean ready;

	/**
	 * The Flush thread instance
	 */
	protected SSTableFlushThread ssTableFlushThread;
	
	/**
	 * The corresponding storage manager state
	 */
	protected State storageState;
	
	/**
	 * The running threads
	 */
	protected final Set<Thread> runningThreads;
	
	/**
	 * The timeout for a thread join (10 seconds)
	 */
	protected long THREAD_WAIT_TIMEOUT = TimeUnit.SECONDS.toMillis(10);
	
	/**
	 * The logger
	 */
	private final static Logger logger = LoggerFactory.getLogger(SSTableManager.class);

	public SSTableManager(final State storageState, final SSTableName name, final ScalephantConfiguration storageConfiguration) {
		super();

		this.storageConfiguration = storageConfiguration;
		this.storageState = storageState;
		this.name = name;
		this.tableNumber = new AtomicInteger();
		this.ready = false;
		
		this.unflushedMemtables = new CopyOnWriteArrayList<Memtable>();
		this.sstableFacades = new CopyOnWriteArrayList<SSTableFacade>();
		this.runningThreads = new HashSet<Thread>();
	}

	/**
	 * Init the instance
	 * 
	 */
	@Override
	public void init() {
		
		if(ready == true) {
			logger.warn("SSTable manager is active and init() is called");
			return;
		}
		
		logger.info("Init a new instance for the table: " + getName());
		
		unflushedMemtables.clear();
		sstableFacades.clear();
		runningThreads.clear();
		createSSTableDirIfNeeded();
		
		try {
			scanForExistingTables();
		} catch (StorageManagerException e) {
			logger.error("Unable to init the instance: " + getName(), e);
			return;
		}
		
		tableNumber.set(getLastSequencenumberFromReader());

		// Set to ready before the threads are started
		ready = true;

		startMemtableFlushThread();
		startCompactThread();
	}

	/**
	 * Start the compact thread if needed
	 */
	protected void startCompactThread() {
		if(storageConfiguration.isStorageRunCompactThread()) {
			final Thread compactThread = new Thread(new SSTableCompactorThread(this));
			compactThread.setName("Compact thread for: " + getName());
			compactThread.start();
			runningThreads.add(compactThread);
		} else {
			logger.info("NOT starting the sstable compact thread.");
		}
	}

	/**
	 * Start the memtable flush thread if needed
	 */
	protected void startMemtableFlushThread() {
		if(storageConfiguration.isStorageRunMemtableFlushThread()) {
			ssTableFlushThread = new SSTableFlushThread(this);
			final Thread flushThread = new Thread(ssTableFlushThread);
			flushThread.setName("Memtable flush thread for: " + getName());
			flushThread.start();
			runningThreads.add(flushThread);
		} else {
			logger.info("NOT starting the memtable flush thread.");
		}
	}

	/**
	 * Shutdown the instance
	 */
	@Override
	public void shutdown() {
		logger.info("Shuting down the instance for table: " + getName());
		
		// Set ready to false. The threads will shutdown after completing
		// the running tasks
		ready = false;
{'fraction_non_alphanumeric': 0.10563306409315011, 'fraction_numerical': 0.0013636840448966747, 'mean_word_length': 4.498269896193771, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '4643797', 'n_tokens_mistral': 3271, 'n_tokens_neox': 2922, 'n_words': 634}