public class DRUMSParameterSet<Data extends AbstractKVStorable> {
    /** My private Logger. */
    private static Logger logger = LoggerFactory.getLogger(DRUMSParameterSet.class);

    private static final String PROTOTYPE_FILE = "prototype.dat";
    private static final String PROPERTY_FILE = "drums.properties";

    /**
     * A global count of all instances of {@link DRUMSParameterSet}. This variable is needed if more than one DRUMS will
     * run in one JVM.
     */
    protected static AtomicInteger INSTANCE_COUNT = new AtomicInteger(0);

    /**
     * a identification number of this parameter set. This variable is needed if more than one DRUMS is instantiated in
     * one JVM.
     */
    public final int instanceID;
    /** File extension of the database files that store the {@link AbstractKVStorable}. */
    public final String linkDataFileExtension = ".db";
    /** The size of the key of the implementation of {@link AbstractKVStorable}. */
    private final int keySize;
    /** The number bytes needed by an instance of {@link AbstractKVStorable}. */
    private final int elementSize;
    /** The prototype of the Data of this DRUMS. */
    private Data prototype;
    /** The name of the underlying parameter file. */
    public String parameterfile;

    // ############################ Parameters which can be manipulated from extern #######################
    /** The database directory. All records are written to files in this directory. */
    public String DATABASE_DIRECTORY;
    /** The amount of bytes all buckets are allowed to use. */
    public long BUCKET_MEMORY;
    /**
     * The maximal size of a bucket in bytes. If this size is exceeded, no more elements for this bucket are accepted,
     * until it is synchronized to disk.
     */
    public long MAX_MEMORY_PER_BUCKET;
    /** the size of one chunk in memory */
    public int MEMORY_CHUNK;
    /** the number of retries if a file is locked by another process */
    public int HEADER_FILE_LOCK_RETRY = 100;
    /** The number of bytes, which are read and written at once during synchronization */
    public long SYNC_CHUNK_SIZE;
    /** The size of one chunk in an {@link HeaderIndexFile} */
    public long FILE_CHUNK_SIZE;
    /** The number of threads used for synchronizing. */
    public int NUMBER_OF_SYNCHRONIZER_THREADS = 1;
    /** The minimal number of elements which must be in one bucket, before this bucket is allowed to be synchronized. */
    public int MIN_ELEMENT_IN_BUCKET_BEFORE_SYNC = 1;
    /** the initial size by which the file is enlarged, when no more records will fit into the file. */
    public int INITIAL_INCREMENT_SIZE;
    /** the initial size of a {@link HeaderIndexFile}. */
    public int INITIAL_FILE_SIZE;
    /** The maximal time in milliseconds a bucket is held in memory without synchronization attempt. */
    public long MAX_BUCKET_STORAGE_TIME;

    /**
     * Initialize global parameters by those from the given parameter-file.
     * 
     * @param paramFile
     *            The name of the parameter-file
     * @param prototype
     *            a prototype of the Data of this DRUMS
     * @throws IOException
     */
    public DRUMSParameterSet(String paramFile, Data prototype) throws IOException {
        this.parameterfile = paramFile;
        this.keySize = prototype.getKey().length;
        this.elementSize = prototype.getSize();
        this.prototype = prototype;
        this.instanceID = DRUMSParameterSet.INSTANCE_COUNT.getAndIncrement();
        initParameters();
    }

    /**
     * The standard constructor. Loads all parameters from drums.properties
     * 
     * @param prototype
     *            a prototype of the Data of this DRUMS
     * @throws IOException
     */
    public DRUMSParameterSet(Data prototype) throws IOException {
        this("drums.properties", prototype);
    }

    /**
     * This constructor tries to load an already saved ParameterFile from an existing {@link DRUMS}-table.
     * 
     * @param folder
     *            the directory, where all files can be found
     * @throws DRUMSException
     *             if the given folder is no folder
     * @throws IOException
     */
    public DRUMSParameterSet(File folder) throws DRUMSException, IOException {
        if (!folder.isDirectory()) {
            throw new DRUMSException("The given FilePointer (" + folder
                    + ") is no folder. Please specify a folder, which contains a DRUMS-table.");
        }
        DATABASE_DIRECTORY = folder.getAbsolutePath();

        // load prototype
        try {
            readPrototype();
        } catch (ClassNotFoundException e) {
            throw new IOException("Could not load prototype from " + DATABASE_DIRECTORY + ".", e);
        }
        this.keySize = prototype.getKey().length;
        this.elementSize = prototype.getSize();

        // load parameter file
        this.parameterfile = getPropertiesFilename();
        this.initParameters();

        this.instanceID = DRUMSParameterSet.INSTANCE_COUNT.getAndIncrement();
    }

    /**
     * Returns a clone of the prototype. If you need to access the prototype several times, you should create a pointer
     * on your own instance.
     * 
     * @return clone of the internal prototype
     */
    @SuppressWarnings("unchecked")
    public Data getPrototype() {
        return ((Data) prototype.clone());
    }

    /**
     * Initialises all Parameters.
     * 
     * @throws FileNotFoundException
     */
    private void initParameters() throws FileNotFoundException {
        InputStream fileStream;
        if (new File(parameterfile).exists()) {
            logger.info("Try reading properties directly from {}.", parameterfile);
            fileStream = new FileInputStream(new File(parameterfile));
        } else {
            logger.info("Try reading properties from Resources");
            fileStream = this.getClass().getClassLoader().getResourceAsStream(parameterfile);
        }

        Properties props = new Properties();
        try {
            props.load(fileStream);
        } catch (IOException ex) {
            throw new RuntimeException(ex);
        }
        DATABASE_DIRECTORY = props.getProperty("DATABASE_DIRECTORY");
        BUCKET_MEMORY = parseSize(props.getProperty("BUCKET_MEMORY", "1G"));
        MEMORY_CHUNK = (int) parseSize(props.getProperty("MEMORY_CHUNK", "10K"));
        MAX_MEMORY_PER_BUCKET = parseSize(props.getProperty("MAX_MEMORY_PER_BUCKET", "100M"));
        SYNC_CHUNK_SIZE = parseSize(props.getProperty("SYNC_CHUNK_SIZE", "2M"));
        FILE_CHUNK_SIZE = parseSize(props.getProperty("FILE_CHUNK_SIZE", "32K"));
        // determine exact index size
        FILE_CHUNK_SIZE = FILE_CHUNK_SIZE - FILE_CHUNK_SIZE % prototype.getSize();
        NUMBER_OF_SYNCHRONIZER_THREADS = Integer.valueOf(props.getProperty("NUMBER_OF_SYNCHRONIZER_THREADS", "1"));
        MAX_BUCKET_STORAGE_TIME = Long.valueOf(props.getProperty("MAX_BUCKET_STORAGE_TIME", "84000000"));
        MIN_ELEMENT_IN_BUCKET_BEFORE_SYNC = Integer
                .valueOf(props.getProperty("MIN_ELEMENT_IN_BUCKET_BEFORE_SYNC", "1"));
        HEADER_FILE_LOCK_RETRY = Integer.valueOf(props.getProperty("HEADER_FILE_LOCK_RETRY", "100"));

        INITIAL_FILE_SIZE = (int) parseSize(props.getProperty("INITIAL_FILE_SIZE", "16M"));
        INITIAL_INCREMENT_SIZE = (int) parseSize(props.getProperty("INITIAL_INCREMENT_SIZE", "16M"));
        configToLogInfo();
    }

    /**
     * Outputs the configuration to the Logger.
     */
    public void configToLogInfo() {
        logger.info("----- MEMEORY USAGE -----");
        logger.info("BUCKET_MEMORY = {}", BUCKET_MEMORY);
        logger.info("MEMORY_CHUNK = {}", MEMORY_CHUNK);
        logger.info("MAX_MEMORY_PER_BUCKET = {}", MAX_MEMORY_PER_BUCKET);
        logger.info("CHUNKSIZE = {}", SYNC_CHUNK_SIZE);

        logger.info("----- HeaderIndexFile -----");
        logger.info("INITIAL_FILE_SIZE = {}", INITIAL_FILE_SIZE);
        logger.info("INITIAL_INCREMENT_SIZE = {}", INITIAL_INCREMENT_SIZE);
        logger.info("CHUNK_SIZE = {}", FILE_CHUNK_SIZE);
    }

    private static Pattern p_mem = Pattern.compile("(\\d+)(K|M|G|)");

    /**
     * This methods parses the given String, which should represent a size, to a long. 'K' is interpreted as 1024, 'M'
     * as 1024^2 and 'G' as 1024^3.
     * 
     * @param s
     *            the String to parse
     * @return size in byte. -1 if the String was not parsable.
     */
    public static long parseSize(String s) {
        Matcher m = p_mem.matcher(s);
        if (m.find()) {
            long i = Long.parseLong(m.group(1));
            char multiplier = 'B';
            if (m.group(2) != null && m.group(2).length() == 1) {
                multiplier = m.group(2).charAt(0);
            }
            if (multiplier == 'B') {
                return (long) i;
            }
            if (multiplier == 'K') {
                return (long) i * 1024l;
            }
            if (multiplier == 'M') {
                return (long) i * 1024l * 1024l;
            }
            if (multiplier == 'G') {
                return (long) i * 1024l * 1024l * 1024l;
            }
        }
        return -1;
    }

    /**
     * @return the properties as {@link Properties}-object
     */
    public Properties getProperties() {
        Properties props = new Properties();
        props.setProperty("DATABASE_DIRECTORY", DATABASE_DIRECTORY + "");
        props.setProperty("BUCKET_MEMORY", BUCKET_MEMORY + "");
        props.setProperty("MEMORY_CHUNK", MEMORY_CHUNK + "");
        props.setProperty("MAX_MEMORY_PER_BUCKET", MAX_MEMORY_PER_BUCKET + "");
        props.setProperty("SYNC_CHUNK_SIZE", SYNC_CHUNK_SIZE + "");
        props.setProperty("FILE_CHUNK_SIZE", FILE_CHUNK_SIZE + "");
        props.setProperty("NUMBER_OF_SYNCHRONIZER_THREADS", NUMBER_OF_SYNCHRONIZER_THREADS + "");
        props.setProperty("MAX_BUCKET_STORAGE_TIME", MAX_BUCKET_STORAGE_TIME + "");
        props.setProperty("MIN_ELEMENT_IN_BUCKET_BEFORE_SYNC", MIN_ELEMENT_IN_BUCKET_BEFORE_SYNC + "");
        props.setProperty("HEADER_FILE_LOCK_RETRY", HEADER_FILE_LOCK_RETRY + "");
        props.setProperty("INITIAL_FILE_SIZE", INITIAL_FILE_SIZE + "");
        props.setProperty("INITIAL_INCREMENT_SIZE", INITIAL_INCREMENT_SIZE + "");
        return props;
    }

    /**
     * @return the size of an element in bytes
     */
    public int getElementSize() {
        return elementSize;
    }

    /**
     * @return the size of the key of an element in bytes
     */
    public int getKeySize() {
        return keySize;
    }

    /**
     * This method stores this {@link DRUMSParameterSet}. Two file are generated. The first file contains the prototype.
     * The second file contains all properties.
     * 
     * @throws IOException
     *             if it was not possible to store all information
     */
    public void store() throws IOException {
        this.storePrototype();
        this.storeProperties();
    }

    private String getPrototypeFilename() {
        return DATABASE_DIRECTORY + "/" + PROTOTYPE_FILE;
    }

    private String getPropertiesFilename() {
        return DATABASE_DIRECTORY + "/" + PROPERTY_FILE;
    }

    private final void readPrototype() throws IOException, ClassNotFoundException {
        FileInputStream fis = new FileInputStream(new File(getPrototypeFilename()));
        ObjectInputStream ois = new ObjectInputStream(fis);
        @SuppressWarnings("unchecked")
        Data tmp = (Data) ois.readObject();
        this.prototype = tmp;
        ois.close();
        fis.close();
    }

    private void storePrototype() throws IOException {
        FileOutputStream fos = new FileOutputStream(new File(getPrototypeFilename()));
        ObjectOutputStream oos = new ObjectOutputStream(fos);
        oos.writeObject(getPrototype());
        oos.close();
        fos.close();
    }

    private void storeProperties() throws IOException {
        FileOutputStream fos = new FileOutputStream(new File(getPropertiesFilename()));
        getProperties().store(fos, "Parameters stored automatically after creating DRUMS.");
        fos.close();
    }
}
 public class Bucket<Data extends AbstractKVStorable> {

    /**
     * contains all memory chunks and its elements to store. We use byte-arrays to save memory, because mostly the whole
     * object-structure takes double memory for small objects.
     */
    private byte[][] memory;

    private int lastChunkIndex = -1;
    private int position_in_chunk = 0;
    private long memorySizeInBytes = 0;

    /** the id of the bucket. Should be known by its {@link BucketContainer} and its {@link AbstractHashFunction} */
    private final int bucketId;

    /** the number of elements in this bucket. For faster access, should be public */
    public int elementsInBucket;

    /** prototype of type Data (extending {@link AbstractKVStorable}) for instantiating correct arrays */
    private Data prototype;

    /** Stores the creation time of this bucket. */
    private long creationTime;

    /** A pointer to the {@link DRUMSParameterSet} used by this {@link DRUMS} */
    private DRUMSParameterSet<Data> gp;

    /**
     * Constructor. Needs to know the id of the {@link Bucket} and the maximum size of the {@link Bucket}.
     * 
     * @param bucketId
     *            the identification number of this bucket. Used in {@link BucketContainer} and other classes
     * @param gp
     *            a pointer to the {@link DRUMSParameterSet}
     */
    public Bucket(final int bucketId, DRUMSParameterSet<Data> gp) {
        this.bucketId = bucketId;
        this.memory = new byte[0][];
        this.elementsInBucket = 0;
        this.prototype = gp.getPrototype();
        this.creationTime = System.currentTimeMillis();
        this.gp = gp;
    }

    /**
     * returns a new empty Bucket with the same properties of this bucket
     * 
     * @return a new {@link Bucket} with the same {@link #bucketId}
     * 
     * @throws IOException
     * @throws FileLockException
     */
    public Bucket<Data> getEmptyBucketWithSameProperties() throws FileLockException, IOException {
        Bucket<Data> newBucket = new Bucket<Data>(this.bucketId, gp);
        return newBucket;
    }

    /**
     * @return the id of the bucket. The id is value between 0 and numberOfBuckets (see {@link BucketContainer})
     */
    public int getBucketId() {
        return bucketId;
    }

    /**
     * Adds one {@link AbstractKVStorable}-object. This method have to be synchronized, because it is possible to access
     * the backend in the same moment with the function {@link #getBackend()}.
     * 
     * @param toAdd
     *            the Data to add
     * @return true, if adding the element to the buffer was successful
     */
    public synchronized boolean add(AbstractKVStorable toAdd) {
        if (memorySizeInBytes >= gp.MAX_MEMORY_PER_BUCKET) {
            return false;
        }
        // no memory is available
        try {
            if (lastChunkIndex == -1 || position_in_chunk == memory[lastChunkIndex].length) {
                boolean allocatingSuccessful = enlargeMemory();

                // No more memory was left so we don't add the element and return false.
                if (!allocatingSuccessful) {
                    return false;
                }
            }
        } catch (InterruptedException e) {
            // TODO: error log
            e.printStackTrace();
        }

        byte[] b = toAdd.toByteBuffer().array();
        for (int i = 0; i < b.length; i++, position_in_chunk++) {
            memory[lastChunkIndex][position_in_chunk] = b[i];
        }
        elementsInBucket++;
        return true;
    }

    /**
     * @param element
     *            the element to look for
     * 
     * @return true, if this bucket, contains the given element
     */
    public boolean contains(Data element) {
        boolean contains = false;
        byte[] dst = new byte[gp.getElementSize()];
        for (int m = 0; m < memory.length; m++) {
            ByteBuffer bb = ByteBuffer.wrap(memory[m]);
            if (m == memory.length - 1) {
                bb.limit(position_in_chunk);
            }
            while (bb.remaining() > 0) {
                bb.get(dst);
                if (KeyUtils.compareKey(element.toByteBuffer().array(), dst) > 0) {
                    contains = true;
                    break;
                }
            }
            if (contains) {
                break;
            }
        }
        return contains;
    }

    /**
     * Enlarges the memory by one chunk. If this was successful <code>true</code> will be returned, otherwise
     * <code>false</code>.
     * 
     * @throws InterruptedException
     */
    private boolean enlargeMemory() throws InterruptedException {
        int bytesAllocated = DynamicMemoryAllocater.INSTANCES[gp.instanceID].allocateNextChunk();
        if (bytesAllocated > 0) {

            byte[] mem = new byte[bytesAllocated];
            byte[][] new_mem = new byte[memory.length + 1][];
            for (int i = 0; i < memory.length; i++) {
                new_mem[i] = memory[i];
            }
            lastChunkIndex++;
            new_mem[lastChunkIndex] = mem;
            memory = new_mem;
            position_in_chunk = 0;
            return true;
        } else {
            return false;
        }
    }

    private void sort() {

    }

    /**
     * @return a ByteBuffer containing the requested element. The returned ByteBuffer is in read-only mode.
     * 
     * @param index
     *            the index of the element in the bucket
     * 
     */
    public ByteBuffer getElementAt(int index) {
        if (index >= elementsInBucket) {
            return null;
        } else {
            int chunk = index * gp.getElementSize() / memory[0].length;
            int offset = index * gp.getElementSize() % memory[0].length;
            return ByteBuffer.wrap(memory[chunk], offset, gp.getElementSize()).asReadOnlyBuffer();
        }
    }

    /**
     * Returns the in <code>backend</code> stored {@link AbstractKVStorable}s. First it rebuilds all objects from their
     * byte-arrays and then sorts them.
     * 
     * @return {@link AbstractKVStorable}[] all {@link AbstractKVStorable}s ascending sorted
     */
    // TODO: handle backend internally as long byte-array
    @SuppressWarnings("unchecked")
    public synchronized Data[] getBackend() {
        sort();
        AbstractKVStorable[] data = new AbstractKVStorable[elementsInBucket];
        byte[] dst = new byte[gp.getElementSize()];
        int i = 0;
        for (int m = 0; m < memory.length; m++) {
            ByteBuffer bb = ByteBuffer.wrap(memory[m]);
            if (m == memory.length - 1) {
                bb.limit(position_in_chunk);
            }
            while (bb.remaining() > 0) {
                bb.get(dst);
                data[i++] = prototype.fromByteBuffer(ByteBuffer.wrap(dst));
            }
        }
        Arrays.sort(data, new AbstractKVStorableComparator());
        return (Data[]) data;
    }

    /** @return the number of elements, actually stored */
    public int size() {
        return elementsInBucket;
    }

    /**
     * This method frees the allocated memory.
     * 
     * @return the number of bytes which are available now
     */
    public int freeMemory() {
        int size = 0;
        for (int m = 0; m < memory.length; m++) {
            size += memory[m].length;
            memory[m] = null;
        }
        memory = null;
        return size;
    }

    /** @return the creation time of this bucket. */
    public long getCreationTime() {
        return creationTime;
    }
} public class BucketContainer<Data extends AbstractKVStorable> {
    private static final Logger log = LoggerFactory.getLogger(BucketContainer.class);
    /**
     * Array containing all {@link Bucket}s. The index of the bucket in this array should be the bucketId of the related
     * {@link Bucket}.
     */
    protected final Bucket<Data>[] buckets;

    /** the HashFunction to determine the bucket-id */
    protected final AbstractHashFunction hashFunction;

    protected boolean shutDownInitiated = false;

    /**
     * @param buckets
     *            the buckets, where to store the {@link AbstractKVStorable}
     * @param hashFunction
     *            the {@link AbstractHashFunction} is used to map from the key of a {@link AbstractKVStorable}-object to
     *            the relevant {@link Bucket}
     */
    public BucketContainer(final Bucket<Data>[] buckets, final AbstractHashFunction hashFunction) {
        this.buckets = buckets;
        this.hashFunction = hashFunction;
    }

    /**
     * Add the given records to the {@link Bucket}s, if possible. If all {@link Bucket}s are full the method is
     * blocking.
     * 
     * @param toAdd
     *            the data to add
     * @throws BucketContainerException
     * @throws InterruptedException
     */
    public void addToCache(Data... toAdd) throws BucketContainerException, InterruptedException {
        if (shutDownInitiated) {
            throw new BucketContainerException("Shutdown was already initiated. Could not add the given elements.");
        }
        int throwBucketException = -1;
        for (AbstractKVStorable date : toAdd) {
            int indexOfCache = hashFunction.getBucketId(date.getKey());
            // safety first, check if the bucket exists. If not, try to move on. Throw exception at the end
            if (indexOfCache < buckets.length) {
                Bucket<Data> bucket = buckets[indexOfCache];
                // Blocking process, try to add element
                while (!bucket.add(date)) {
                    Thread.sleep(1000);
                }
            } else {
                throwBucketException = indexOfCache;
            }
        }

        if (throwBucketException != -1) {
            throw new BucketContainerException("Could not insert at least one LinkData-object. One missing bucket was "
                    + throwBucketException);
        }
    }

    /** @return the number of buckets */
    public int getNumberOfBuckets() {
        return hashFunction.getNumberOfBuckets();
    }

    /**
     * @param bucketId
     * @return the bucket with the given bucketId
     */
    public Bucket<Data> getBucket(int bucketId) {
        return buckets[bucketId];
    }

    /**
     * sets a new bucket. The bucketId is read from the given {@link Bucket}-object
     * 
     * @param bucket
     *            the {@link Bucket}-object to add
     * @throws BucketContainerException
     * @throws InterruptedException
     */
    public void setBucket(Bucket<Data> bucket) throws BucketContainerException, InterruptedException {
        buckets[bucket.getBucketId()] = bucket;
    }

    /**
     * @param element
     * @return Checks, if an record with the key of the given element is already in memory.
     */
    public boolean contains(Data element) {
        return buckets[hashFunction.getBucketId(element)].contains(element);
    }

    /**
     * returns the {@link AbstractHashFunction} for mapping keys to {@link Bucket}s.
     * 
     * @return {@link AbstractHashFunction}
     */
    public AbstractHashFunction getHashFunction() {
        return this.hashFunction;
    }

    /** This method initializes the shutdown of this BucketContainer. No records are accepted anymore. */
    public void shutdown() {
        this.shutDownInitiated = true;
        log.info("Shutting down the bucket container.");
    }
}
 public class BucketContainerException extends Exception {
    private static final long serialVersionUID = -7034481737223723755L;

    /**
     * Instantiates a new {@link BucketContainerException} with the given message.
     * 
     * @param message
     */
    public BucketContainerException(String message) {
        super(message);
    }
} public class DynamicMemoryAllocater<Data extends AbstractKVStorable> {
    private static Logger logger = LoggerFactory.getLogger(DynamicMemoryAllocater.class);

    /**
     * An array, containing all instances of {@link DynamicMemoryAllocater}. Each DRUMS-table has its own.
     */
    @SuppressWarnings("rawtypes")
    public static DynamicMemoryAllocater[] INSTANCES = new DynamicMemoryAllocater[0];

    /** The number of already allocated / used bytes */
    private AtomicLong used_bytes;

    /** The number of the maximal allowed bytes, distributed by this class */
    private final long max_allowed_bytes;

    /** The size of one memory chunk. The size is a multiple of the buffersize of a prototype */
    private final int mem_chunksize;

    /**
     * Instantiates a new {@link DynamicMemoryAllocater}. This method is private, cause we want to this class as
     * Singelton.
     * 
     * @param gp
     *            a pointer to the GlobalParameters used by this DRUMS
     */
    private DynamicMemoryAllocater(DRUMSParameterSet<Data> gp) {
        this.used_bytes = new AtomicLong(0);
        this.mem_chunksize = gp.MEMORY_CHUNK - gp.MEMORY_CHUNK % gp.getElementSize();
        this.max_allowed_bytes = gp.BUCKET_MEMORY - gp.BUCKET_MEMORY % this.mem_chunksize;
    }

    /**
     * Instantiates the {@link DynamicMemoryAllocater}, only if there is not already an instance
     * 
     * @param gp
     *            a pointer to the GlobalParameters used by this DRUMS
     */
    public static <Data extends AbstractKVStorable> void instantiate(DRUMSParameterSet<Data> gp) {
        if (INSTANCES.length <= gp.instanceID) {
            INSTANCES = Arrays.copyOf(INSTANCES, gp.instanceID + 1);
            INSTANCES[gp.instanceID] = new DynamicMemoryAllocater<Data>(gp);
        }
    }

    /**
     * This method tries to mark memory as allocated. It allocates as much bytes as defined in <code>MEMORY_CHUNK</code>
     * and returns the number of bytes marked as allocated. This method doesn't really allocate memory.
     * 
     * @return the size of the allocated chunk
     */
    public synchronized int allocateNextChunk() {
        if ((used_bytes.longValue() + (long) mem_chunksize) > max_allowed_bytes) {
            logger.info("No memory left");
            return 0;
        }
        used_bytes.set(used_bytes.longValue() + (long) mem_chunksize);
        return mem_chunksize;
    }

    /**
     * Marks the given amount of memory as free to use.
     * 
     * @param size
     *            the size of memory (in bytes) to free
     */
    public void freeMemory(long size) {
        used_bytes.set(used_bytes.longValue() - size);
    }

    /** @return the number of used bytes, allocated by this {@link DynamicMemoryAllocater} */
    public long getUsedMemory() {
        return used_bytes.get();
    }

    /** @return the maximal allowed bytes to be used by this {@link DynamicMemoryAllocater} */
    public long getMaxMemory() {
        return max_allowed_bytes;
    }

    /** @return the number of bytes not allocated by the {@link DynamicMemoryAllocater} */
    public long getFreeMemory() {
        return max_allowed_bytes - used_bytes.longValue();
    }
} public abstract class AbstractHashFunction implements Serializable {
    private static final long serialVersionUID = -3139493910715670755L;
    /** the number of buckets */
    protected int buckets;

    /** @return the number of buckets, addressed by this hash-function */
    public int getNumberOfBuckets() {
        return buckets;
    }

    /**
     * @param key
     *            the key to map to a bucket-id
     * @return the bucket-id belonging to the given key
     */
    public abstract int getBucketId(byte[] key);

    /**
     * @param key
     *            the element containing the key to map a bucket-id
     * @return the bucket-id belonging to the given {@link AbstractKVStorable}
     */
    public abstract int getBucketId(AbstractKVStorable key);

    /**
     * @param bucketId
     * @return the filename of the bucket for the given bucket-id.
     */
    public abstract String getFilename(int bucketId);

    /**
     * @param dbFilename
     * @return the bucket-id of the bucket belonging to the given filename.
     */
    public abstract int getBucketId(String dbFilename);

    /**
     * Writes this hashfunction to the given {@link OutputStream}
     * 
     * @param os
     * @throws IOException
     */
    public abstract void store(OutputStream os) throws IOException;

    /**
     * Loads the hashfunction from the given {@link InputStream}
     * 
     * @param in
     * @throws IOException
     */
    public abstract void load(InputStream in) throws IOException;
} public class FileLockException extends Exception {
    private static final long serialVersionUID = 1L;

    /**
     * This method constructs a new {@link FileLockException}
     * 
     * @param message
     *            the message to throw
     */
    public FileLockException(String message) {
        super(message);
    }
} public class HeaderIndexFile<Data extends AbstractKVStorable> extends AbstractHeaderFile implements IWritableIndexFile {

    /** the size of the index in bytes */
    protected static final long MAX_INDEX_SIZE_IN_BYTES = 512 * 1024; // 512 kb

    /** the byte offset, where the index starts */
    protected static final int INDEX_OFFSET = HEADER_SIZE; // index starts after the header

    /** the time to wait for retry the access, if the file was locked */
    protected static final int RETRY_CONNECT_WAITTIME = 250;

    /** if true, the file enlarges automatically if <code>size</code> is reached */
    protected static final boolean AUTO_ENLARGE = true;

    /** the size of a stored element in bytes */
    protected int elementSize;// part of the header

    protected int keySize;

    /** shows if the file was closed correctly */
    private byte closedSoftly = 0; // PART OF HEADER (1 bytes)

    /** in bytes */
    protected int chunkSize;

    /** a constant size, by which the file will be resized in byte */
    protected int incrementSize;

    /** a mappedByteBuffer to the index-region. So changes can be made directly */
    protected ByteBuffer indexBuffer;

    /** the number of elements, which can be stored in the index */
    protected int indexSize;

    /** the size of the index in bytes */
    protected long indexSizeInBytes;

    /** A pseudo-index (not each element is indexed, but the chunks where they belong to */
    protected IndexForHeaderIndexFile index;

    /** A pointer to the GlobalParameters used by this DRUMS */
    protected DRUMSParameterSet<Data> gp;

    /**
     * This constructor instantiates a new {@link HeaderIndexFile} with the given <code>fileName</code> in the given
     * {@link AbstractHeaderFile.AccessMode}.
     * 
     * @param fileName
     *            the filename of the underlying OSfile.
     * @param mode
     *            the mode the file should be accessed. READ_ONLY or READ_WRITE
     * @param max_retries_connect
     *            the number of retries to open a channel, if the file is locked
     * @param gp
     *            the {@link DRUMSParameterSet}.
     * @throws FileLockException
     *             if the <code>max_retries_connect</code> is exceeded
     * @throws IOException
     *             if another error with the file-access occured
     */
    public HeaderIndexFile(String fileName, AccessMode mode, int max_retries_connect, DRUMSParameterSet<Data> gp)
            throws FileLockException, IOException {
        this.gp = gp;
        this.incrementSize = gp.INITIAL_INCREMENT_SIZE;
        this.elementSize = gp.getElementSize();
        this.keySize = gp.getKeySize();
        this.osFile = new File(fileName);
        this.mode = mode;
        this.max_retries_connect = max_retries_connect;
        this.init();

        if (closedSoftly == 0) {
            logger.warn("File {} was not closed correctly and might be corrupted", osFile.getName());
        }
    }

    /**
     * This constructor instantiates a new {@link HeaderIndexFile} with the given <code>fileName</code> in the given
     * {@link AbstractHeaderFile.AccessMode}. This is a weak constructor and therefore you can only have read-access.
     * 
     * @param fileName
     *            the filename of the underlying OSfile.
     * @param max_retries_connect
     *            the number of retries to open a channel, if the file is locked
     * @param gp
     *            the {@link DRUMSParameterSet}.
     * @throws FileLockException
     *             if the <code>max_retries_connect</code> is exceeded
     * @throws IOException
     *             if another error with the fileaccess occured
     */
    public HeaderIndexFile(String fileName, int max_retries_connect, DRUMSParameterSet<Data> gp)
            throws FileLockException, IOException {
        this.osFile = new File(fileName);
        this.max_retries_connect = max_retries_connect;
        this.gp = gp;
        this.init();
        if (closedSoftly == 0) {
            logger.warn("File {} was not closed correctly and might be corrupted", osFile.getName());
        }
    }

    protected void init() throws FileLockException, IOException {
        this.contentStart = HEADER_SIZE + MAX_INDEX_SIZE_IN_BYTES;
        if (!osFile.exists()) {
            logger.debug("File {} not found. Initialise new File.", osFile.getAbsolutePath());
            this.createFile();
        } else {
            logger.debug("File {} exists. Try to open it.", osFile.getAbsolutePath());
            openChannel();
        }
        this.contentEnd = size;

        if (chunkSize % elementSize != 0) {
            logger.warn(
                    "The readChunkSize ({}) is not a multiple of elementsize ({}). This might lead to a wrong index",
                    chunkSize, elementSize);
        }
    }

    /**
     * Create a {@link MappedByteBuffer} mapping the given region. The size of the HEADER will be respected
     * automatically. <br>
     * <br>
     * WARNING: This method is for professional use only.
     * 
     * @param offset
     * @param length
     * @return a {@link MappedByteBuffer} pointing to the part of the file specified by offset and length.
     * @throws IOException
     */
    public MappedByteBuffer map(long offset, int length) throws IOException {
        offset += contentStart;
        if (mode == AccessMode.READ_ONLY) {
            return channel.map(FileChannel.MapMode.READ_ONLY, offset, length);
        } else {
            return channel.map(FileChannel.MapMode.READ_WRITE, offset, length);
        }
    }

    /**
     * writes the bytes from the given ByteBuffer to the file beginning at offset. The size of the HEADER will be
     * respected automatically.
     * 
     * @param offset
     * @param sourceBuffer
     * @throws IOException
     */
    @Override
    public void write(long offset, ByteBuffer sourceBuffer) throws IOException {
        offset += contentStart;
        sourceBuffer.position(0);

        long newFilePosition = offset + sourceBuffer.limit();
        if (newFilePosition > contentEnd) {
            logger.debug("Filesize exceeded (contendEnd: {},size: {})", contentEnd, size);
            if (AUTO_ENLARGE) {
                // if the incrementSize is not large enough
                // if (offset + sourceBuffer.limit() > contentEnd + incrementSize) {
                // incrementSize = (int) (offset + sourceBuffer.limit() - contentEnd + incrementSize);
                // }
                this.enlargeFile(newFilePosition - contentEnd);
            } else {
                throw new IOException("Filesize exceeded (" + size
                        + ") and automatic enlargement is disabled. You have to enlarge file manually.");
            }
        }

        if (offset + sourceBuffer.limit() > filledUpTo) {
            filledUpTo = offset + sourceBuffer.limit();
        }
        channel.write(sourceBuffer, offset);
        writeHeader();
    }

    @Override
    public void write(long offset, byte[] sourceBuffer) throws IOException {
        write(offset, ByteBuffer.wrap(sourceBuffer));
    }

    @Override
    public long append(byte[] sourceBuffer) throws IOException {
        return append(ByteBuffer.wrap(sourceBuffer));
    }

    @Override
    public long append(ByteBuffer sourceBuffer) throws IOException {
        long positionInFile = filledUpTo - contentStart;
        write(positionInFile, sourceBuffer);
        return positionInFile;
    }

    /**
     * Reads x bytes from the file to the given ByteBuffer (position was set to zero), where x is the minimum of the
     * capacity of the buffer and the remaining written bytes in the file. The size of the HEADER will be respected
     * automatically. The position is set to the number of read bytes.
     * 
     * @param offset
     *            the file offset, where to start reading
     * @param destBuffer
     *            the buffer to fill
     * @throws IOException
     */
    @Override
    public int read(long offset, ByteBuffer destBuffer) throws IOException {
        offset += contentStart;
        destBuffer.clear();
        int length = destBuffer.capacity();
        if (offset > filledUpTo) {
            String errorMessage = "Tried to read data beginning at " + offset + " in File (max=" + filledUpTo + ") "
                    + osFile.getName();
            logger.debug(errorMessage);
            throw new IOException(errorMessage);
        } else if (offset + length > filledUpTo) {
            logger.debug("Tried to read data over {} in File " + osFile.getName() + ". Limit will be set to {}",
                    filledUpTo, destBuffer.limit());
            destBuffer.limit((int) (filledUpTo - offset));
        }
        offset = checkRegions(offset, destBuffer.limit());
        int readBytes = channel.read(destBuffer, offset);
        return readBytes;
    }

    /**
     * 
     * Reads x bytes from the file to the given ByteBuffer (position was set to zero), where x is the minimum of the
     * capacity of the buffer and the remaining written bytes in the file. The size of the HEADER will be respected
     * automatically. The position is set to the number of read bytes.
     * 
     * @see #read(long, ByteBuffer)
     * @param offset
     *            the file offset, where to start reading
     * @param destBuffer
     *            the buffer to fill
     * @throws IOException
     */
    @Override
    public int read(long offset, byte[] destBuffer) throws IOException {
        ByteBuffer buffer = ByteBuffer.wrap(destBuffer);
        return read(offset, buffer);
    }

    /** this function calculates all relevant informations for the index */
    protected void calcIndexInformations() {
        indexSize = (int) Math.floor(MAX_INDEX_SIZE_IN_BYTES / keySize);
        indexSizeInBytes = indexSize * keySize;
    }

    /**
     * opens the {@link RandomAccessFile} and the corresponding {@link FileChannel}. Optionally reads the header. and
     * the index
     * 
     * @param should
     *            the header be read
     * @param should
     *            the index be read
     * @throws IOException
     */
    protected void openChannel(boolean readHeader, boolean readIndex) throws FileLockException, IOException {
        super.openChannel(readHeader);
        calcIndexInformations();
        if (mode == AccessMode.READ_ONLY) {
            indexBuffer = channel.map(FileChannel.MapMode.READ_ONLY, INDEX_OFFSET, indexSizeInBytes);
        } else {
            indexBuffer = channel.map(FileChannel.MapMode.READ_WRITE, INDEX_OFFSET, indexSizeInBytes);
        }

        if (readIndex) {
            readIndex();
        }
    }

    /**
     * opens the {@link RandomAccessFile} and the corresponding {@link FileChannel}. Optionally reads the header. Reads
     * the index. This may overwrite previous set parameters.
     * 
     * @param should
     *            the header be read
     * @throws IOException
     */
    protected void openChannel(boolean readHeader) throws FileLockException, IOException {
        openChannel(false, true);
    }

    /**
     * opens the {@link RandomAccessFile} and the corresponding {@link FileChannel}. Reads the header and the index.
     * This may overwrite previous set parameters.
     * 
     * @throws IOException
     */
    public void openChannel() throws FileLockException, IOException {
        openChannel(true, true);
    }

    protected void createFile() throws FileLockException, IOException {
        size = gp.INITIAL_FILE_SIZE;
        filledUpTo = contentStart;
        chunkSize = (int) gp.FILE_CHUNK_SIZE;
        openChannel(false, false);
        setSoftlyClosed(true);
        // have to reset the informations, because in #openchannel the empty header was read
        accessFile.setLength(gp.INITIAL_FILE_SIZE);
        writeHeader();
        readIndex(); // index should be empty
    }

    public long getFreeSpace() {
        return size - filledUpTo;
    }

    /** reads and instantiates the index from the <code>indexBuffer</code> */
    public void readIndex() {
        indexBuffer.rewind();
        index = new IndexForHeaderIndexFile(indexSize, keySize, chunkSize, indexBuffer);
    }

    protected void readHeader() {
        headerBuffer.rewind();
        size = headerBuffer.getLong();
        filledUpTo = headerBuffer.getLong();
        closedSoftly = headerBuffer.get();
        chunkSize = headerBuffer.getInt();
        elementSize = headerBuffer.getInt();
        keySize = headerBuffer.getInt();
    }

    protected void writeHeader() {
        headerBuffer.rewind();
        headerBuffer.putLong(size);
        headerBuffer.putLong(filledUpTo);
        headerBuffer.put(closedSoftly);
        headerBuffer.putInt(chunkSize);
        headerBuffer.putInt(elementSize);
        headerBuffer.putInt(keySize);
    }

    /**
     * Rounds up the given size to the next full chunk, and enlarges the file by these amounts of bytes.
     * 
     * @param toEnlarge
     * @throws IOException
     */
    @Override
    public void enlargeFile(long toEnlarge) throws IOException {
        size += (long) Math.ceil((double) toEnlarge / incrementSize) * incrementSize;
        logger.debug("Enlarge filesize of {} to {}", osFile, size);
        contentEnd = size;
        accessFile.setLength(size);

        // TODO check buffer size
        // if there are more chunks than the index can contain
        if ((size - contentStart) / chunkSize > indexSize) {
            logger.error("File-Enlargement not possible. The index becomes too large.");
            throw new IOException("File-Enlargement not possible. The index becomes too large.");
        }
        calcIndexInformations();
    }

    /**
     * @return the instantiated index {@link IndexForHeaderIndexFile}
     */
    public IndexForHeaderIndexFile getIndex() {
        return index;
    }

    /**
     * This method checks, if the keys of all inserted elements are incrementing continuously.
     * 
     * @return true, if the file fulfills this constraint.
     * @throws IOException
     */
    public boolean isConsistent() throws IOException {
        byte[] b = new byte[elementSize];
        long offset = 0;
        byte[] oldKey = null;
        int i = 0;
        while (offset < this.getFilledUpFromContentStart()) {
            this.read(offset, ByteBuffer.wrap(b));
            byte[] key = Arrays.copyOfRange(b, 0, keySize);
            if (oldKey != null && KeyUtils.compareKey(key, oldKey) != 1) {
                logger.error("File is not consistent at record {}. {} not larger than {}", new Object[] { i,
                        KeyUtils.toStringUnsignedInt(key), KeyUtils.toStringUnsignedInt(oldKey) });
                return false;
            }
            oldKey = Arrays.copyOf(key, keySize);
            offset += elementSize;
            i++;
        }
        return true;
    }

    /**
     * This method checks, if the data is consistent with the index.
     * 
     * @return true, if the file fulfills this constraint.
     * @throws IOException
     */
    public boolean isConsitentWithIndex() throws IOException {
        byte[] b = new byte[keySize];
        long offset = 0;
        int maxChunk = getChunkIndex(getFilledUpFromContentStart());
        boolean isConsistent = true;
        for (int i = 1; i <= maxChunk; i++) {
            offset = i * getChunkSize() - elementSize;
            read(offset, ByteBuffer.wrap(b));
            if (KeyUtils.compareKey(getIndex().maxKeyPerChunk[i - 1], b) != 0) {
                logger.error("Index is not consistent to data. Expected {}, but found {}.",
                        Arrays.toString(getIndex().maxKeyPerChunk[i - 1]), Arrays.toString(b));
                isConsistent = false;
            }
        }
        return isConsistent;
    }

    /**
     * This method repairs the index. It runs over all chunks and writes the largest element into the
     * {@link IndexForHeaderIndexFile}. This method should be called, if the file was not closed softly (
     * {@link #isSoftlyClosed()}).
     * 
     * @throws IOException
     */
    public void repairIndex() throws IOException {
        byte[] currentLargestKey = new byte[keySize];
        int maxChunk = getChunkIndex(getFilledUpFromContentStart());
        for (int currentChunkId = 0; currentChunkId <= maxChunk; currentChunkId++) {
            long currentOffset = (currentChunkId + 1) * getChunkSize() - elementSize;
            if (currentOffset < getFilledUpFromContentStart()) {
                read(currentOffset, currentLargestKey);
            } else {
                read(getFilledUpFromContentStart() - elementSize, currentLargestKey);
            }
            getIndex().setLargestKey(currentChunkId, currentLargestKey);
        }
    }

    /** Sets all file-pointers to their initial values. The file itself is not shrinked. */
    public void clear() {
        this.filledUpTo = 0;
        this.contentStart = HEADER_SIZE + MAX_INDEX_SIZE_IN_BYTES;
        this.filledUpTo = contentStart;
    }

    /**
     * calculates the index of the belonging readChunk in the files content
     * 
     * @param offset
     *            the byteOffset in the content
     * @return the determined index
     */
    public int getChunkIndex(long offset) {
        return (int) offset / this.chunkSize;
    }

    /** @return the size of one element. We assume that all elements are equal sized */
    public int getElementSize() {
        return elementSize;
    }

    /** @return the size in bytes of a chunk */
    public int getChunkSize() {
        return chunkSize;
    }

    /** @return true, if the file was closed softly */
    public boolean isSoftlyClosed() {
        return closedSoftly == 1;
    }

    protected void setSoftlyClosed(boolean value) {
        if (headerBuffer == null) {
            return;
        }
        closedSoftly = (byte) (value ? 1 : 0);
        this.writeHeader();
    }

    public void close() {
        if (this.index != null) {
            index.indexBuffer = null;
            index = null;
        }
        if (indexBuffer != null && indexBuffer instanceof MappedByteBuffer) {
            ((MappedByteBuffer) indexBuffer).force();
            indexBuffer = null;
        } else {
            // TODO: write ByteBuffer to file
        }
        super.close();
    }
} public class IndexForHeaderIndexFile {

    /** the interal ByteArrayComparator */
    private ByteArrayComparator comparator = new ByteArrayComparator();

    /** chunkId -> largest key in this chunk */
    protected byte maxKeyPerChunk[][];

    /** the size of one chunk */
    protected int chunkSize;

    /** the number of chunks, represented by this index */
    protected int numberOfChunks;

    /** a {@link MappedByteBuffer}, maps to the corresponding byte-region in memory */
    protected ByteBuffer indexBuffer;

    /** reflects the largest initialized chunkIndex */
    private int filledUpTo;

    private int keySize;

    /**
     * This constructor instantiates a new {@link IndexForHeaderIndexFile}. You need to give the number of the indexed
     * chunks, the chunksize and the {@link MappedByteBuffer}, where the index should be written to.
     * 
     * @param numberOfChunks
     *            the expected or maximal number of chunks
     * @param keySize
     * @param chunkSize
     *            the size of one chunk
     * @param indexBuffer
     *            the buffer were to write the indexinformations
     */
    public IndexForHeaderIndexFile(final int numberOfChunks, final int keySize, final int chunkSize,
            final ByteBuffer indexBuffer) {
        this.keySize = keySize;
        this.numberOfChunks = numberOfChunks;
        this.chunkSize = chunkSize;
        this.indexBuffer = indexBuffer;
        this.maxKeyPerChunk = new byte[numberOfChunks][]; // init lookuparray
        this.filledUpTo = 0; // init the actual indexed chunks
        this.fillIndexFromByteBuffer(); // fill eventual in indexBuffer contained informations
    }

    /**
     * loads the index from the {@link MappedByteBuffer}
     */
    protected void fillIndexFromByteBuffer() {
        indexBuffer.rewind();
        for (int i = 0; i < numberOfChunks; i++) {
            maxKeyPerChunk[i] = new byte[keySize];
            indexBuffer.get(maxKeyPerChunk[i]);
            if (KeyUtils.isNull(maxKeyPerChunk[i]) && i > 0) {
                filledUpTo = i - 1;
                break;
            }
        }
    }

    /**
     * returns the position of chunk with the given index in the file.
     * 
     * @param chunkIndex
     * @return long, the byte-offset of the chunk
     */
    public long getStartOffsetOfChunk(int chunkIndex) {
        return chunkIndex * chunkSize;
    }

    /**
     * returns the position in the file, where the chunk, where the key belongs to starts.
     * 
     * @param key
     * @return long, the byte-offset of the chunk
     */
    public long getStartOffsetOfChunkByKey(byte[] key) {
        return getStartOffsetOfChunk(getChunkId(key));
    }

    /**
     * returns the index of the chunk, where the element could be found. If the element can't be in one of the
     * chunks the method will return -1.
     * 
     * @param key
     *            the key to look for
     * @return the id of the chunk, where the key could be found. If there is no putative chunk, then -1 will returned.
     */
    public int getChunkId(byte[] key) {
        int idx = Arrays.binarySearch(maxKeyPerChunk, 0, filledUpTo + 1, key, comparator);
        idx = idx < 0 ? -idx - 1 : idx;
        if (idx > filledUpTo) {
            return -1;
        } else {
            return idx;
        }
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < filledUpTo; i++) {
            sb.append(i + ": " + KeyUtils.toStringUnsignedInt(maxKeyPerChunk[i])).append("\n");
        }
        return sb.toString();
    }

    /**
     * Checks, if this index is consistent. All inserted keys have to be inserted incrementally.
     * 
     * @return boolean
     */
    public boolean isConsistent() {
        for (int i = 1; i < filledUpTo; i++) {
            int comp1 = KeyUtils.compareKey(maxKeyPerChunk[i], maxKeyPerChunk[i - 1]);
            if (comp1 <= 0) {
                return false;
            }
        }
        return true;
    }

    /**
     * Sets a new largest key in the chunk with the given index.
     * 
     * @param chunkIdx
     * @param largestKeyInChunk
     */
    public void setLargestKey(int chunkIdx, byte[] largestKeyInChunk) {
        maxKeyPerChunk[chunkIdx] = largestKeyInChunk;
        filledUpTo = Math.max(filledUpTo, chunkIdx);
        indexBuffer.position(chunkIdx * keySize);
        indexBuffer.put(largestKeyInChunk);
    }
} public abstract class AbstractKVStorable implements Serializable, Cloneable {
    private static final long serialVersionUID = -3973787626582319301L;

    /**
     * key of the object. This key is very important for storing the objects in ascending order. It have to be coded in
     * the first bytes in the byte representation. The key is public for fast access.
     */
    protected byte[] key;

    /**
     * value of the object. Storing all values in one byte-array reduces the memory amount of a materialized object
     */
    protected byte[] value;

    /**
     * Sets the value of this object. The given array is not copied, only a pointer will be set
     *
     * @param value
     */
    public void setValue(byte[] value) {
        this.value = value;
    }

    /**
     * @return a pointer to the underlying array, which contains all value-parts
     */
    public byte[] getValue() {
        return value;
    }

    /**
     * Sets the key of this object. Be careful by overwriting the old key. The given array is not copied, only a pointer
     * will be set.
     *
     * @param key
     */
    public void setKey(byte[] key) {
        this.key = key;
    }

    /**
     * @return a pointer to the underlying array, which contains all key-parts
     */
    public byte[] getKey() {
        return key;
    }

    @Override
    public abstract AbstractKVStorable clone();

    /**
     * Returns the number of bytes needed to store key and value.
     *
     * @return key.length + value.length
     */
    public int getSize() {
        return key.length + value.length;
    }

    /**
     * Converts the object to a {@link ByteBuffer}
     *
     * @return the object as {@link ByteBuffer}
     */
    public abstract ByteBuffer toByteBuffer();

    /**
     * Initializes the object from the given {@link ByteBuffer}
     *
     * @param bb
     */
    public abstract void initFromByteBuffer(ByteBuffer bb);

    /**
     * Generates a new Object from the given {@link ByteBuffer}.
     *
     * @param bb
     * @return a new instances of this Class
     */
    public abstract <Data extends AbstractKVStorable> Data fromByteBuffer(ByteBuffer bb);

    /**
     * Merges the given {@link AbstractKVStorable} with this one by your implementation.
     *
     * @param element
     * @return a pointer to an {@link AbstractKVStorable}.
     */
    public abstract <Data extends AbstractKVStorable> Data merge(Data element);

    /**
     * Updates the given {@link AbstractKVStorable} with values from this one.
     *
     * @param element
     */
    public abstract <Data extends AbstractKVStorable> void update(Data element);

    /**
     * This method returns false by default. This method must be overwritten in the concrete class,
     *
     * @return true, if this element is marked as deleted
     */
    public boolean isMarkedAsDeleted() {
        return false;
    }

    @Override
    public boolean equals(Object obj) {
        if (getClass() != obj.getClass())
            return false;
        AbstractKVStorable other = (AbstractKVStorable) obj;
        if (!Arrays.equals(key, other.key))
            return false;
        if (!Arrays.equals(value, other.value))
            return false;
        return true;
    }

    /**
     * This method merges all {@link AbstractKVStorable}s in the given array with same keys. The array must been sorted.
     *
     * @param toAdd
     *            this array might contain duplicate entries concerning the key, which must be merged.
     *
     * @return a new array containing unique {@link AbstractKVStorable}s. The array is sorted.
     */
    @SuppressWarnings("unchecked")
    public static <Data extends AbstractKVStorable> Data[] merge(Data[] toAdd) {
        if (toAdd.length == 1) {
            return toAdd;
        }
        // estimate the number of unique elements and check the precondition, that the array must been sorted
        int count = 1;
        for (int i = 0; i < toAdd.length - 1; i++) {
            int compare = KeyUtils.compareKey(toAdd[i].key, toAdd[i + 1].key);
            if (compare > 0) {
                throw new RuntimeException("The given array is not sorted.");
            } else if(compare < 0) {
                count++;
            } else { // if compare == 0
                // do nothing, elements are equal
            }
        }

        AbstractKVStorable[] realToAdd = new AbstractKVStorable[count];
        count = 0;
        // merge Elements in toAdd
        AbstractKVStorable first = toAdd[0];
        for (int k = 1; k < toAdd.length; k++) {
            while (k < toAdd.length && KeyUtils.compareKey(toAdd[k].key, first.key) == 0) {
                first = first.merge(toAdd[k]);
                k++;
            }
            realToAdd[count++] = first;
            if (k < toAdd.length) {
                first = toAdd[k];
            }
        }

        if (count < realToAdd.length && KeyUtils.compareKey(realToAdd[realToAdd.length - 2].key, first.key) != 0) {
            realToAdd[count++] = first;
        }
        return (Data[]) realToAdd;
    }
}
 public class GeneralStorable extends AbstractKVStorable {
    static Logger logger = LoggerFactory.getLogger(GeneralStorable.class);
    private static final long serialVersionUID = 3853444781559739538L;

    /** A pointer to the underlying structure. All cloned elements should point to the same structure. */
    protected GeneralStructure structure;

    /**
     * Basic constructor. Should only be used, when the structure of the
     * 
     * @param s
     *            the structure of this {@link GeneralStorable}.
     */
    public GeneralStorable(GeneralStructure s) {
        structure = s;
        if (structure.keySize == 0) {
            logger.error("The size of the key is 0");
        }

        if (structure.valueSize == 0) {
            logger.warn("The size of the value is 0");
        }

        this.key = new byte[structure.keySize];
        this.value = new byte[structure.valueSize];
        structure.INSTANCE_EXISITS = true;
    }

    /** Constructor for cloning */
    protected GeneralStorable(int keySize, int valueSize, GeneralStructure s) {
        this.structure = s;
        this.key = new byte[keySize];
        this.value = new byte[valueSize];
        structure.INSTANCE_EXISITS = true;
    }

    @Override
    public void initFromByteBuffer(ByteBuffer bb) {
        if (bb.remaining() < key.length + value.length) {
            bb.rewind();
        }
        bb.get(key).get(value);
    }

    @Override
    public GeneralStorable fromByteBuffer(ByteBuffer bb) {
        GeneralStorable object = new GeneralStorable(this.key.length, this.value.length, structure);
        object.initFromByteBuffer(bb);
        return object;
    }

    @Override
    public ByteBuffer toByteBuffer() {
        return ByteBuffer.allocate(key.length + value.length).put(key).put(value);
    }

    @Override
    public GeneralStorable clone() {
        return this.fromByteBuffer(toByteBuffer());
    }

    @Override
    public <Data extends AbstractKVStorable> Data merge(Data element) {
        return element;
    }

    @Override
    public void update(AbstractKVStorable element) {
        element.initFromByteBuffer(this.toByteBuffer());
    }

    /**
     * Sets the value belonging to the given field.
     * 
     * @param field
     *            the name of the field
     * @param value
     *            the value to set
     * @throws IOException
     */
    public void setValue(String field, byte[] value) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        if (!structure.valueHash2Index.containsKey(hash)) {
            throw new IOException("The field " + field + " is unknown.");
        }
        setValue(structure.valueHash2Index.get(hash), value);
    }

    /**
     * Sets the value belonging to the given field.
     * 
     * @param index
     *            the index of the requested field
     * @param value
     *            the value to set
     * @throws IOException
     */
    public void setValue(int index, byte[] value) throws IOException {
        if (index >= structure.valueSizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        int length = structure.valueSizes.get(index);
        if (value.length != length) {
            throw new IOException("The length of the given value is not equal to the expected one. (" + value.length
                    + "!=" + length + ")");
        }
        Bytes.putBytes(this.value, structure.valueByteOffsets.get(index), value, 0, value.length);
    }

    /**
     * Sets the value belonging to the given field.
     * 
     * @param field
     *            the name of the field
     * @param value
     * @throws IOException
     */
    public void setValue(String field, int value) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        setValue(structure.valueHash2Index.get(hash), value);
    }

    /**
     * Sets the value belonging to the given field.
     * 
     * @param index
     *            the index of the requested field
     * @param value
     * @throws IOException
     */
    public void setValue(int index, int value) throws IOException {
        if (index >= structure.valueSizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        Bytes.putInt(this.value, structure.valueByteOffsets.get(index), value);
    }

    /**
     * Sets the value belonging to the given field.
     * 
     * @param field
     *            the name of the field
     * @param value
     * @throws IOException
     */
    public void setValue(String field, float value) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        setValue(structure.valueHash2Index.get(hash), value);
    }

    /**
     * Sets the value belonging to the given field.
     * 
     * @param index
     *            the index of the requested field
     * @param value
     * @throws IOException
     */
    public void setValue(int index, float value) throws IOException {
        if (index >= structure.valueSizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        Bytes.putFloat(this.value, structure.valueByteOffsets.get(index), value);
    }

    /**
     * Sets the value belonging to the given field.
     * 
     * @param field
     *            the name of the field
     * @param value
     * @throws IOException
     */
    public void setValue(String field, long value) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        setValue(structure.valueHash2Index.get(hash), value);
    }

    /**
     * Sets the value belonging to the given field.
     * 
     * @param index
     *            the index of the requested field
     * @param value
     * @throws IOException
     */
    public void setValue(int index, long value) throws IOException {
        if (index >= structure.valueSizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        Bytes.putLong(this.value, structure.valueByteOffsets.get(index), value);
    }

    /**
     * Sets the value belonging to the given field.
     * 
     * @param field
     *            the name of the field
     * @param value
     * @throws IOException
     */
    public void setValue(String field, double value) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        setValue(structure.valueHash2Index.get(hash), value);
    }

    /**
     * Sets the value belonging to the given field.
     * 
     * @param index
     *            the index of the requested field
     * @param value
     * @throws IOException
     */
    public void setValue(int index, double value) throws IOException {
        if (index >= structure.valueSizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        Bytes.putDouble(this.value, structure.valueByteOffsets.get(index), value);
    }

    /**
     * Sets the value belonging to the given field.
     * 
     * @param field
     *            the name of the field
     * @param value
     * @throws IOException
     */
    public void setValue(String field, char value) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        setValue(structure.valueHash2Index.get(hash), value);
    }

    /**
     * Sets the value belonging to the given field.
     * 
     * @param index
     *            the index of the requested field
     * @param value
     * @throws IOException
     */
    public void setValue(int index, char value) throws IOException {
        if (index >= structure.valueSizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        Bytes.putChar(this.value, structure.valueByteOffsets.get(index), value);
    }

    /**
     * Sets the value belonging to the given field.
     * 
     * @param field
     *            the name of the field
     * @param value
     * @throws IOException
     */
    public void setValue(String field, byte value) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        setValue(structure.valueHash2Index.get(hash), value);
    }

    /**
     * Sets the value belonging to the given field.
     * 
     * @param index
     *            the index of the requested field
     * @param value
     * @throws IOException
     */
    public void setValue(int index, byte value) throws IOException {
        if (index >= structure.valueSizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        this.value[structure.valueByteOffsets.get(index)] = value;
    }

    /**
     * Returns the value belonging to the given field as read-only ByteBuffer
     * 
     * @param field
     *            the name of the field
     * @return the requested value
     * @throws IOException
     */
    public ByteBuffer getValue(String field) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        return getValue(structure.valueHash2Index.get(hash));
    }

    /**
     * Returns the value belonging to the given field as read-only ByteBuffer
     * 
     * @param index
     *            the index of the requested field
     * @return the requested value
     * @throws IOException
     */
    public ByteBuffer getValue(int index) throws IOException {
        if (index >= structure.valueSizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        int length = structure.valueSizes.get(index);
        return ByteBuffer.wrap(value, structure.valueByteOffsets.get(index), length).asReadOnlyBuffer();
    }

    /**
     * Returns the value belonging to the given field as Int
     * 
     * @param index
     *            the index of the requested field
     * @return the requested value
     * @throws IOException
     */
    public int getValueAsInt(int index) throws IOException {
        if (index >= structure.valueSizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        return Bytes.toInt(value, structure.valueByteOffsets.get(index));
    }

    /**
     * Returns the value belonging to the given field as int
     * 
     * @param field
     *            the name of the field
     * @return the requested value
     * @throws IOException
     */
    public int getValueAsInt(String field) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        return getValueAsInt(structure.valueHash2Index.get(hash));
    }

    /**
     * Returns the value belonging to the given field as int
     * 
     * @param field
     *            the name of the field
     * @return the requested value
     * @throws IOException
     */
    public float getValueAsFloat(String field) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        return getValueAsFloat(structure.valueHash2Index.get(hash));
    }

    /**
     * Returns the value belonging to the given field as Int
     * 
     * @param index
     *            the index of the requested field
     * @return the requested value
     * @throws IOException
     */
    public float getValueAsFloat(int index) throws IOException {
        if (index >= structure.valueSizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        return Bytes.toFloat(value, structure.valueByteOffsets.get(index));
    }

    /**
     * Returns the value belonging to the given field as long
     * 
     * @param index
     *            the index of the requested field
     * @return the requested value
     * @throws IOException
     */
    public double getValueAsDouble(int index) throws IOException {
        if (index >= structure.valueSizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        return Bytes.toDouble(value, structure.valueByteOffsets.get(index));
    }

    /**
     * Returns the value belonging to the given field as long
     * 
     * @param field
     *            the name of the field
     * @return the requested value
     * @throws IOException
     */
    public double getValueAsDouble(String field) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        return getValueAsDouble(structure.valueHash2Index.get(hash));
    }

    /**
     * Returns the value belonging to the given field as long
     * 
     * @param index
     *            the index of the requested field
     * @return the requested value
     * @throws IOException
     */
    public long getValueAsLong(int index) throws IOException {
        if (index >= structure.valueSizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        return Bytes.toLong(value, structure.valueByteOffsets.get(index));
    }

    /**
     * Returns the value belonging to the given field as long
     * 
     * @param field
     *            the name of the field
     * @return the requested value
     * @throws IOException
     */
    public long getValueAsLong(String field) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        return getValueAsLong(structure.valueHash2Index.get(hash));
    }

    /**
     * Returns the value belonging to the given field as char
     * 
     * @param index
     *            the index of the requested field
     * @return the requested value
     * @throws IOException
     */
    public char getValueAsChar(int index) throws IOException {
        if (index >= structure.valueSizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        return Bytes.toChar(value, structure.valueByteOffsets.get(index));
    }

    /**
     * Returns the value belonging to the given field as char
     * 
     * @param field
     *            the name of the field
     * @return the requested value
     * @throws IOException
     */
    public char getValueAsChar(String field) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        return getValueAsChar(structure.valueHash2Index.get(hash));
    }

    /**
     * Returns the value belonging to the given field as byte
     * 
     * @param field
     *            the name of the field
     * @return the requested value
     * @throws IOException
     */
    public byte getValueAsByte(String field) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        return getValueAsByte(structure.valueHash2Index.get(hash));
    }

    /**
     * Returns the value belonging to the given field as byte
     * 
     * @param index
     *            the index of the requested field
     * @return the requested value
     * @throws IOException
     */
    public byte getValueAsByte(int index) throws IOException {
        if (index >= structure.valueSizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        return value[index];
    }

    /**
     * Sets the key belonging to the given field.
     * 
     * @param field
     *            the name of the field
     * @param key
     *            the key to set
     * @throws IOException
     */
    public void setKey(String field, byte[] key) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        setKey(structure.keyHash2Index.get(hash), key);
    }

    /**
     * Sets the key belonging to the given field.
     * 
     * @param index
     *            the index of the requested field
     * @param key
     *            the key to set
     * @throws IOException
     */
    public void setKey(int index, byte[] key) throws IOException {
        if (index >= structure.keySizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        int length = structure.keySizes.get(index);
        if (key.length != length) {
            throw new IOException("The length of the given key is not equal to the expected one. (" + key.length
                    + "!=" + length + ")");
        }
        Bytes.putBytes(this.key, structure.keyByteOffsets.get(index), key, 0, key.length);
    }

    /**
     * Sets the key belonging to the given field.
     * 
     * @param field
     *            the name of the field
     * @param key
     * @throws IOException
     */
    public void setKey(String field, int key) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        setKey(structure.keyHash2Index.get(hash), key);
    }

    /**
     * Sets the key belonging to the given field.
     * 
     * @param index
     *            the index of the requested field
     * @param key
     * @throws IOException
     */
    public void setKey(int index, int key) throws IOException {
        if (index >= structure.keySizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        int length = structure.keySizes.get(index);
        if (4 != length) {
            throw new IOException("The length of the given key is not equal to the expected one. (" + 4 + "!="
                    + length + ")");
        }
        Bytes.putInt(this.key, structure.keyByteOffsets.get(index), key);
    }

    /**
     * Sets the key belonging to the given field.
     * 
     * @param index
     *            the index of the requested field
     * @param key
     * @throws IOException
     */
    public void setKey(int index, float key) throws IOException {
        if (index >= structure.keySizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        int length = structure.keySizes.get(index);
        if (4 != length) {
            throw new IOException("The length of the given key is not equal to the expected one. (" + 4 + "!="
                    + length + ")");
        }
        Bytes.putFloat(this.key, structure.keyByteOffsets.get(index), key);
    }

    /**
     * Sets the key belonging to the given field.
     * 
     * @param field
     *            the name of the field
     * @param key
     * @throws IOException
     */
    public void setKey(String field, float key) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        setKey(structure.keyHash2Index.get(hash), key);
    }

    /**
     * Sets the key belonging to the given field.
     * 
     * @param field
     *            the name of the field
     * @param key
     * @throws IOException
     */
    public void setKey(String field, double key) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        setKey(structure.keyHash2Index.get(hash), key);
    }

    /**
     * Sets the key belonging to the given field.
     * 
     * @param index
     *            the index of the requested field
     * @param key
     * @throws IOException
     */
    public void setKey(int index, double key) throws IOException {
        if (index >= structure.keySizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        int length = structure.keySizes.get(index);
        if (8 != length) {
            throw new IOException("The length of the given key is not equal to the expected one. (" + 8 + "!="
                    + length + ")");
        }
        Bytes.putDouble(this.key, structure.keyByteOffsets.get(index), key);
    }

    /**
     * Sets the key belonging to the given field.
     * 
     * @param field
     *            the name of the field
     * @param key
     * @throws IOException
     */
    public void setKey(String field, long key) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        setKey(structure.keyHash2Index.get(hash), key);
    }

    /**
     * Sets the key belonging to the given field.
     * 
     * @param index
     *            the index of the requested field
     * @param key
     * @throws IOException
     */
    public void setKey(int index, long key) throws IOException {
        if (index >= structure.keySizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        int length = structure.keySizes.get(index);
        if (8 != length) {
            throw new IOException("The length of the given key is not equal to the expected one. (" + 8 + "!="
                    + length + ")");
        }
        Bytes.putLong(this.key, structure.keyByteOffsets.get(index), key);
    }

    /**
     * Sets the key belonging to the given field.
     * 
     * @param field
     *            the name of the field
     * @param key
     * @throws IOException
     */
    public void setKey(String field, char key) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        setKey(structure.keyHash2Index.get(hash), key);
    }

    /**
     * Sets the key belonging to the given field.
     * 
     * @param index
     *            the index of the requested field
     * @param key
     * @throws IOException
     */
    public void setKey(int index, char key) throws IOException {
        if (index >= structure.keySizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        int length = structure.keySizes.get(index);
        if (2 != length) {
            throw new IOException("The length of the given key is not equal to the expected one. (" + 2 + "!="
                    + length + ")");
        }
        Bytes.putChar(this.key, structure.keyByteOffsets.get(index), key);
    }

    /**
     * Sets the key belonging to the given field.
     * 
     * @param field
     *            the name of the field
     * @param key
     * @throws IOException
     */
    public void setKey(String field, byte key) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        setKey(structure.keyHash2Index.get(hash), key);
    }

    /**
     * Sets the key belonging to the given field.
     * 
     * @param index
     *            the index of the requested field
     * @param key
     * @throws IOException
     */
    public void setKey(int index, byte key) throws IOException {
        if (index >= structure.keySizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        int length = structure.keySizes.get(index);
        if (1 != length) {
            throw new IOException("The length of the given key is not equal to the expected one. (" + 1 + "!="
                    + length + ")");
        }
        this.key[structure.keyByteOffsets.get(index)] = key;
    }

    /**
     * Returns the key belonging to the given field as read-only ByteBuffer
     * 
     * @param field
     *            the name of the field
     * @return the requested value
     * @throws IOException
     */
    public ByteBuffer getKey(String field) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        return getKey(structure.keyHash2Index.get(hash));
    }

    /**
     * Returns the key belonging to the given field as read-only ByteBuffer
     * 
     * @param index
     *            the index of the requested field
     * @return the requested value
     * @throws IOException
     */
    public ByteBuffer getKey(int index) throws IOException {
        if (index >= structure.keySizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        int length = structure.keySizes.get(index);
        return ByteBuffer.wrap(key, structure.keyByteOffsets.get(index), length).asReadOnlyBuffer();
    }

    /**
     * Returns the key belonging to the given field as Int
     * 
     * @param index
     *            the index of the requested field
     * @return the requested value
     * @throws IOException
     */
    public int getKeyAsInt(int index) throws IOException {
        if (index >= structure.keySizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        return Bytes.toInt(key, structure.keyByteOffsets.get(index));
    }

    /**
     * Returns the key belonging to the given field as int
     * 
     * @param field
     *            the name of the field
     * @return the requested value
     * @throws IOException
     */
    public int getKeyAsInt(String field) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        return getKeyAsInt(structure.keyHash2Index.get(hash));
    }

    /**
     * Returns the key belonging to the given field as Int
     * 
     * @param index
     *            the index of the requested field
     * @return the requested value
     * @throws IOException
     */
    public float getKeyAsFloat(int index) throws IOException {
        if (index >= structure.keySizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }

        return Bytes.toFloat(key, structure.keyByteOffsets.get(index));
    }

    /**
     * Returns the key belonging to the given field as int
     * 
     * @param field
     *            the name of the field
     * @return the requested value
     * @throws IOException
     */
    public float getKeyAsFloat(String field) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        return getKeyAsFloat(structure.keyHash2Index.get(hash));
    }

    /**
     * Returns the key belonging to the given field as long
     * 
     * @param index
     *            the index of the requested field
     * @return the requested value
     * @throws IOException
     */
    public long getKeyAsLong(int index) throws IOException {
        if (index >= structure.keySizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        return Bytes.toLong(key, structure.keyByteOffsets.get(index));
    }

    /**
     * Returns the key belonging to the given field as long
     * 
     * @param field
     *            the name of the field
     * @return the requested value
     * @throws IOException
     */
    public long getKeyAsLong(String field) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        return getKeyAsLong(structure.keyHash2Index.get(hash));
    }

    /**
     * Returns the key belonging to the given field as long
     * 
     * @param index
     *            the index of the requested field
     * @return the requested value
     * @throws IOException
     */
    public double getKeyAsDouble(int index) throws IOException {
        if (index >= structure.keySizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }

        return Bytes.toDouble(key, structure.keyByteOffsets.get(index));
    }

    /**
     * Returns the key belonging to the given field as long
     * 
     * @param field
     *            the name of the field
     * @return the requested value
     * @throws IOException
     */
    public double getKeyAsDouble(String field) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        return getKeyAsDouble(structure.keyHash2Index.get(hash));
    }

    /**
     * Returns the key belonging to the given field as char
     * 
     * @param index
     *            the index of the requested field
     * @return the requested value
     * @throws IOException
     */
    public char getKeyAsChar(int index) throws IOException {
        if (index >= structure.keySizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }

        return Bytes.toChar(key, structure.keyByteOffsets.get(index));
    }

    /**
     * Returns the key belonging to the given field as char
     * 
     * @param field
     *            the name of the field
     * @return the requested value
     * @throws IOException
     */
    public char getKeyAsChar(String field) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        return getKeyAsChar(structure.keyHash2Index.get(hash));
    }

    /**
     * Returns the key belonging to the given field as byte
     * 
     * @param index
     *            the index of the requested field
     * @return the requested value
     * @throws IOException
     */
    public byte getKeyAsByte(int index) throws IOException {
        if (index >= structure.keySizes.size()) {
            throw new IOException("Index " + index + " is out of range.");
        }
        return key[structure.keyByteOffsets.get(index)];
    }

    /**
     * Returns the key belonging to the given field as byte
     * 
     * @param field
     *            the name of the field
     * @return the requested value
     * @throws IOException
     */
    public byte getKeyAsByte(String field) throws IOException {
        int hash = Arrays.hashCode(field.getBytes());
        return getKeyAsByte(structure.keyHash2Index.get(hash));
    }

    public String toString() {
        StringBuilder sb = new StringBuilder();
        for (String s : structure.keyPartNames) {
            int hash = Arrays.hashCode(s.getBytes());
            sb.append("structure.keyPart: ").append(s).append("\t");
            sb.append(structure.keySizes.get(structure.keyHash2Index.get(hash))).append("\t");
            sb.append(structure.keyByteOffsets.get(structure.keyHash2Index.get(hash))).append("\n");
        }
        for (String s : structure.valuePartNames) {
            int hash = Arrays.hashCode(s.getBytes());
            sb.append("structure.valuePart: ").append(s).append("\t");
            sb.append(structure.valueSizes.get(structure.valueHash2Index.get(hash))).append("\t");
            sb.append(structure.valueByteOffsets.get(structure.valueHash2Index.get(hash))).append("\n");
        }

        return sb.toString();
    }
} public class SyncManager<Data extends AbstractKVStorable> extends Thread {
    private static final Logger log = LoggerFactory.getLogger(SyncManager.class);

    /** at least this fill level of a bucket must been reached, before synchronizing for this bucket is started */
    public static final double MINFILL_BEFORE_SYNC = 0.3;

    /** Contains the buckets. */
    public final BucketContainer<Data> bucketContainer;

    /** the path to the files, where to store {@link AbstractKVStorable}s */
    private final String pathToDbFiles;

    /** The factory to use by the {@link SyncThread}. */
    private ISynchronizerFactory<Data> synchronizerFactory;

    /** true, if shutdown is initiated. So all buckets will written to HDD. */
    private boolean shutDownInitiated;

    /**
     * true, if force is initiated. So all buckets will written to HDD. In different to shutDown DRUMS is not blocking
     * inserts.
     */
    private boolean forceInitiated;

    /** A set of bucketIds, which are actual in process */
    private Set<Bucket<Data>> actualProcessingBuckets;
    // private Set<Integer> actualProcessingBucketIds;

    /** The {@link ThreadPoolExecutor} handling all {@link SyncThread}s */
    private ThreadPoolExecutor bufferThreads;

    /**
     * the number of allowed buckets waiting for synchronizing. This number should be always larger, than the real
     * number of threads
     */
    private int allowedBucketsInBuffer;

    /** the number of {@link Bucket}s in {@link BucketContainer} */
    /* needed for fast access */
    private int numberOfBuckets;

    /**
     * max time that a bucket can linger in the bucket container before it will be synchronized to disk. The initial
     * storage time will be infinite.
     */
    private long maxBucketStorageTime;

    /**
     * The number of elements that were inserted by Synchronizers. Have to be threadsafe. Because several
     * {@link Synchronizer}s may update this variable.
     */
    private AtomicLong numberOfElementsInserted;

    /**
     * The number of elements that were updated by Synchronizers. Because several {@link Synchronizer}s may update this
     * variable.
     */
    private AtomicLong numberOfElementsUpdated;

    /** A Pointer to the GlobalParameters used by the DRUMS containing this SyncManager */
    DRUMSParameterSet<Data> gp;

    /**
     * Instantiates a new {@link SyncManager} with an already instantiated {@link BucketContainer}. The defined number
     * of Threads in {@link DRUMSParameterSet} will show the number of allowed {@link SyncThread}s used to synchronize
     * the {@link AbstractKVStorable}s from the {@link Bucket}s in {@link BucketContainer} to the HDD.
     * 
     * @param bucketContainer
     * @param synchronizerFactory
     * @param gp
     */
    public SyncManager(
            BucketContainer<Data> bucketContainer,
            ISynchronizerFactory<Data> synchronizerFactory,
            DRUMSParameterSet<Data> gp) {
        this.gp = gp;
        this.bucketContainer = bucketContainer;
        // HashSet<Integer> tmpSet = new HashSet<Integer>();
        HashSet<Bucket<Data>> tmpSet = new HashSet<Bucket<Data>>();
        // this.actualProcessingBucketIds = Collections.synchronizedSet(tmpSet);
        this.actualProcessingBuckets = Collections.synchronizedSet(tmpSet);
        this.pathToDbFiles = gp.DATABASE_DIRECTORY;
        this.setMaxBucketStorageTime(gp.MAX_BUCKET_STORAGE_TIME);
        ;
        this.synchronizerFactory = synchronizerFactory;
        this.allowedBucketsInBuffer = gp.NUMBER_OF_SYNCHRONIZER_THREADS * 4;
        this.numberOfBuckets = bucketContainer.getNumberOfBuckets();

        // instantiating queue and ThreadPoolExecutor for parallel synchronizing
        BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(allowedBucketsInBuffer);
        bufferThreads = new ThreadPoolExecutor(
                gp.NUMBER_OF_SYNCHRONIZER_THREADS,
                gp.NUMBER_OF_SYNCHRONIZER_THREADS,
                Integer.MAX_VALUE,
                TimeUnit.DAYS, queue);
        this.setName("SyncManager:FileStorage");

        numberOfElementsInserted = new AtomicLong();
        numberOfElementsUpdated = new AtomicLong();
    }

    public void run() {
        do {
            synchronizeBucketsWithHDD();

            // We wait for 1 milli second. This is not much but reduces the CPU load a lot.
            try {
                Thread.sleep(1);
            } catch (InterruptedException ex) {
                log.info("Sync manager was interrupted.");
                break;
            }
        } while (!shutDownInitiated || !bucketsEmpty());

        // BucketContainer is not receiving new elements, but there can still be waiting elements
        try {
            do {
                Thread.sleep(1);
                synchronizeBucketsWithHDD();
            } while (!bucketsEmpty());
        } catch (Exception ex) {
            log.error("One exception occurred while synchronizing buckets.", ex);
        }

        bufferThreads.shutdown();
        try {
            bufferThreads.awaitTermination(Integer.MAX_VALUE, TimeUnit.HOURS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * checks, if the buckets are empty. Returns true only if all buckets are empty.
     * 
     * @return true only if all buckets are empty
     */
    private boolean bucketsEmpty() {
        for (int i = 0; i < numberOfBuckets; i++) {
            if (bucketContainer.getBucket(i).elementsInBucket > 0) {
                return false;
            }
        }
        return true;
    }

    /** waits a moment for continuing synchronizing */
    private void sleep() {
        try {
            Thread.sleep(250);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * This method manages the synchronizing-process between {@link BucketContainer} and HDD. It is repeatedly called by
     * <code>run()</code>, till shutdown is initiated.
     */
    private void synchronizeBucketsWithHDD() {
        int synchronizedBuckets = 0;
        // run over all buckets
        for (int i = 0; i < numberOfBuckets; i++) {
            Bucket<Data> oldBucket = bucketContainer.getBucket(i);

            // if the bucket is empty, then do nothing
            if (oldBucket.elementsInBucket == 0) {
                synchronizedBuckets++;
                continue;
            }

            /* ***** TESTING PURPOSE ****** */
            if (DynamicMemoryAllocater.INSTANCES[gp.instanceID].getFreeMemory() == 0) {
                log.info("No memory free, theoretically I must force synchronization");
            }
            /* **************************** */

            long elapsedTime = System.currentTimeMillis() - oldBucket.getCreationTime();
            // if the bucket is full, or the bucket is longer then max bucket storage time within the BucketContainer,
            // or the shutdown was initiated, then try to synchronize the buckets
            // At this point we prevent starvation of one bucket if it not filled for a long period of time.
            if (oldBucket.elementsInBucket >= gp.MIN_ELEMENT_IN_BUCKET_BEFORE_SYNC ||
                    // TODO What do we do if more than one DynamicMemoryAllocator exist???
                    DynamicMemoryAllocater.INSTANCES[gp.instanceID].getFreeMemory() == 0 ||
                    // DynamicMemoryAllocater.INSTANCE.getFreeMemory() == 0 ||
                    elapsedTime > maxBucketStorageTime ||
                    shutDownInitiated ||
                    forceInitiated) {
                if (!startNewThread(i)) {
                    sleep();
                }
            }
        }

        if (shutDownInitiated) {
            log.info("{} of {} buckets were synchronized.", synchronizedBuckets, bucketContainer.getNumberOfBuckets());
        }

        // if the queue is not full try to fill it
        if (bufferThreads.getQueue().size() < bufferThreads.getMaximumPoolSize()) {
            int bucketId = getLargestBucketId();
            if (bucketId != -1) {
                Bucket<Data> pointer = bucketContainer.getBucket(bucketId);
                boolean threadStarted = false;
                if (DynamicMemoryAllocater.INSTANCES[gp.instanceID].getFreeMemory() == 0 ||
                        pointer.elementsInBucket >= gp.MIN_ELEMENT_IN_BUCKET_BEFORE_SYNC ||
                        forceInitiated) {
                    threadStarted = startNewThread(bucketId);
                }

                if (!threadStarted) {
                    sleep();
                }
            }
        }
    }

    /**
     * Starts a new {@link SyncThread} with the given <code>bucketId</code>. Returns true if the {@link SyncThread} was
     * successful started, false otherwise.<br>
     * <br>
     * The {@link SyncThread} is not started, if the bucket is empty, already in process or does not exist.
     * 
     * @param bucketId
     */
    private boolean startNewThread(int bucketId) {
        // bucket is in process or doesnt exist
        if (isBucketProcessed(bucketId) || bucketId == -1) {
            log.debug(
                    "Can't add BufferThread. Bucket {} does not exist or a bucket with the same id is already in processing.",
                    bucketId);
            return false;
        }
        Bucket<Data> oldBucket = bucketContainer.getBucket(bucketId);
        // bucket is empty
        if (oldBucket.elementsInBucket == 0) {
            log.debug("Can't add BufferThread. Bucket {} is empty", bucketId);
            return false;
        }

        // BlockingQueue is full
        if (bufferThreads.getQueue().size() == allowedBucketsInBuffer) {
            log.debug("Can't add BufferThread. Too many threads in queue. {}", actualProcessingBuckets.toString());
            return false;
        }

        actualProcessingBuckets.add(oldBucket);
        // actualProcessingBucketIds.add(bucketId);
        // the old Bucket will be replaced by this new bucket
        try {
            Bucket<Data> newBucket = oldBucket.getEmptyBucketWithSameProperties();
            bucketContainer.setBucket(newBucket);
        } catch (Exception e) {
            e.printStackTrace();
        }

        SyncThread<Data> bufferThread = new SyncThread<Data>(this, oldBucket, actualProcessingBuckets,
                synchronizerFactory, gp);
        bufferThreads.execute(bufferThread);
        return true;
    }

    /** Gets a bucket id and returns <code>true</code> if the bucket is currently processed, otherwise <code>false</code> */
    private boolean isBucketProcessed(int bucketId) {
        synchronized (actualProcessingBuckets) {
            for (Bucket<Data> oneBucket : this.actualProcessingBuckets) {
                if (oneBucket.getBucketId() == bucketId) {
                    return true;
                }
            }
        }
        return false;
    }

    /**
     * Returns the id of the largest bucket. (the bucket with most elements)
     * 
     * @return int
     */
    private int getLargestBucketId() {
        int max = 0;
        int rememberId = -1;
        for (int i = 0; i < numberOfBuckets; i++) {
            Bucket<Data> oldBucket = bucketContainer.getBucket(i);
            if (oldBucket.elementsInBucket > max) {
                max = oldBucket.elementsInBucket;
                rememberId = oldBucket.getBucketId();
            }
        }
        return rememberId;
    }

    /**
     * overwrites the SynchronizerFactory. Be very careful
     * 
     * @param factory
     */
    public void setSynchronizer(ISynchronizerFactory<Data> factory) {
        this.synchronizerFactory = factory;
    }

    /**
     * Stops forcing the DRUMS to synchronize all Buckets
     */
    public void stopForceMode() {
        this.bucketContainer.shutdown();
        log.info("Forcing stopped.");
        forceInitiated = false;
    }

    /**
     * Initiates the a synchronizing of all buckets. All actual buckets with elements will be processed. Be careful. If
     * someone refills the buckets, the thread won't shut down.
     */
    public void startForceMode() {
        this.bucketContainer.shutdown();
        log.info("Forcing DRUMS to synchronize its buckets");
        forceInitiated = true;
    }

    /**
     * Initiates the shutdown. All actual buckets with elements will be processed. Be careful. If someone refills the
     * buckets, the thread won't shut down.
     */
    public void shutdown() {
        this.bucketContainer.shutdown();
        log.info("Shuting down the sync manager");
        shutDownInitiated = true;
    }

    /** @return the directory of the database files. */
    public String getPathToDbFiles() {
        return pathToDbFiles;
    }

    /**
     * @return the current maximal time that a bucket can linger within the BucketContainer.
     */
    public long getMaxBucketStorageTime() {
        return maxBucketStorageTime;
    }

    /**
     * Sets the maximal time that a bucket can linger within the BucketContainer without synchronization to hard drive.
     * 
     * @param maxBucketStorageTime
     */
    public void setMaxBucketStorageTime(long maxBucketStorageTime) {
        this.maxBucketStorageTime = maxBucketStorageTime;
    }

    /** @return a set of buckets which are currently in a synchronization process. */
    public Set<Bucket<Data>> getActualProcessingBuckets() {
        return this.actualProcessingBuckets;
    }

    /** @return the {@link BucketContainer}. */
    public BucketContainer<Data> getBucketContainer() {
        return bucketContainer;
    }

    // #########################################################################################
    // #########################################################################################
    // #########################################################################################
    /* Monitoring methods */

    /** @return the number of buffer threads that are currently active. */
    public int getNumberOfActiveThreads() {
        return bufferThreads.getActiveCount();
    }

    /** @return the maximal number of buffer threads that are allowed. */
    public int getNumberOfMaximalThreads() {
        return bufferThreads.getMaximumPoolSize();
    }

    /**
     * Sums the number of inserted elements.
     * 
     * @param toSumUp
     */
    public void sumUpInserted(long toSumUp) {
        numberOfElementsInserted.addAndGet(toSumUp);
    }

    /**
     * Sums the number of updated elements.
     * 
     * @param toSumUp
     */
    public void sumUpUpdated(long toSumUp) {
        numberOfElementsUpdated.addAndGet(toSumUp);
    }

    /** @return the overall number of inserted elements to the file storage. */
    public long getNumberOfElementsInserted() {
        return numberOfElementsInserted.get();
    }

    /** @return the overall number of updated elements of the file storage. */
    public long getNumberOfElementsUpdated() {
        return numberOfElementsUpdated.get();
    }

} public interface ISynchronizerFactory<Data extends AbstractKVStorable> {
    /**
     * Creates a {@link Synchronizer}, which synchronizes the given {@link Bucket}.
     * 
     * @param databaseFilename
     *            the name of the database file which the synchronizer uses
     * @param String
     *            typeOfStorable the classname of the {@link AbstractKVStorable} to be handled by a Synchronizer
     * @param gp
     *            a pointer to the GlobalParameters
     * @return a new {@link Synchronizer}-instance
     * @throws IOException
     *             if some FileHandling fail
     */
    Synchronizer<Data> createSynchronizer(String databaseFilename, DRUMSParameterSet<Data> gp) throws IOException;

} public class SynchronizerFactory<Data extends AbstractKVStorable> implements ISynchronizerFactory<Data> {
    @Override
    public Synchronizer<Data> createSynchronizer(String databaseFilename, DRUMSParameterSet<Data> gp) throws IOException {
        Synchronizer<Data> synchronizer = null;
        synchronizer = new Synchronizer<Data>(databaseFilename, gp);
        return synchronizer;
    }
} public class UpdateOnlySynchronizer<Data extends AbstractKVStorable> {
    /** */
    private static Logger log = LoggerFactory.getLogger(UpdateOnlySynchronizer.class);

    /** The file instance from which we receive a {@link FileChannel}. */
    protected String dataFilename;

    /** a pointer to the file, where data-objects are stored */
    protected HeaderIndexFile<Data> dataFile;

    /** the header of the bucket, something like an index */
    private IndexForHeaderIndexFile header;

    /** a prototype element of type Data */
    private Data prototype;

    /** the ByteBuffer to work on. Is used for reading and writing, could be replaced by a MappedByteBuffer */
    private ByteBuffer workingBuffer;

    /** A pointer to the global Parameters */
    private DRUMSParameterSet<Data> gp;

    /**
     * This method constructs a {@link Synchronizer}. The name of the file were to write the elements to have to be
     * given.
     * 
     * @param dataFilename
     * @param gp
     * 
     */
    public UpdateOnlySynchronizer(final String dataFilename, final DRUMSParameterSet<Data> gp) {
        this.gp = gp;
        this.dataFilename = dataFilename;
        this.prototype = gp.getPrototype();
        try {
            /* Another thread can have access to this file in parallel. So we must wait to get exclusive access. */
            dataFile = new HeaderIndexFile<Data>(dataFilename, AccessMode.READ_WRITE, Integer.MAX_VALUE, gp);
            header = dataFile.getIndex(); // Pointer to the Index
        } catch (FileLockException ex) {
            log.error("Could not aquire the log for the file to update: {}", dataFilename, ex);
            throw new IllegalStateException(ex);
        } catch (IOException ex) {
            log.error("Error occurred while initialising UpdateOnlySynchronizer: {}", dataFilename, ex);
            throw new IllegalStateException(ex);
        }

        this.workingBuffer = ByteBuffer.allocate((int) dataFile.getChunkSize());
    }

    /**
     * This method handles the given {@link AbstractKVStorable}s from a {@link Bucket}. It provides updates and inserts
     * of those objects knowing all already stored {@link AbstractKVStorable}s in the <code>dataFile</code>.
     * 
     * @param toUpdate
     * @throws IOException
     */
    public void upsert(Data[] toUpdate) throws IOException {
        try {
            int actualChunkIdx = 0, lastChunkIdx = -1;
            long actualChunkOffset = 0, oldChunkOffset = -1;
            int indexInChunk = 0;
            // special case
            for (int i = 0; i < toUpdate.length; i++) {
                // get actual chunkIndex
                actualChunkIdx = header.getChunkId(toUpdate[i].getKey());
                actualChunkOffset = header.getStartOffsetOfChunk(actualChunkIdx);

                if (actualChunkOffset > dataFile.getFilledUpFromContentStart()) {
                    log.warn("Element with key {} was not found. Chunk {} does not exist.", actualChunkIdx,
                            toUpdate[i].getKey());
                    continue;
                }

                // if it is the same chunk as in the last step, use the old readbuffer
                if (actualChunkIdx != lastChunkIdx) {
                    // if we have read a chunk
                    if (oldChunkOffset > -1) {
                        dataFile.write(oldChunkOffset, workingBuffer);
                        indexInChunk = 0;
                    }
                    // read a new part to the readBuffer
                    dataFile.read(actualChunkOffset, workingBuffer);
                }

                int oldIndexInChunk = indexInChunk;
                indexInChunk = updateElementInReadBuffer(toUpdate[i], indexInChunk);
                if (indexInChunk == -1) {
                    log.warn(
                            "Element with key {} was not found and therefore not updated. File: {}, Chunk: {}, Index for searching in buffer: {}, Index after update: {}",
                            new Object[] { toUpdate[i].getKey(), this.dataFilename, actualChunkIdx, oldIndexInChunk,
                                    indexInChunk });
                    indexInChunk = 0;
                }

                lastChunkIdx = actualChunkIdx; // remember last chunk
                oldChunkOffset = actualChunkOffset; // remember last offset of the last chunk
            }
            // if we have read a chunk
            if (oldChunkOffset > -1) {
                dataFile.write(oldChunkOffset, workingBuffer);
            }

        } finally {
            // close the file
            dataFile.close();
        }
    }

    /** traverses the readBuffer */
    private int updateElementInReadBuffer(Data data, int indexInChunk) {
        workingBuffer.position(indexInChunk);
        int minElement = indexInChunk / gp.getElementSize();
        int numberOfEntries = workingBuffer.limit() / gp.getElementSize();
        byte[] actualKey = data.getKey();
        // binary search
        int maxElement = numberOfEntries - 1;
        int midElement;
        int compare;
        byte[] tmpKey = new byte[actualKey.length];
        while (minElement <= maxElement) {
            midElement = minElement + (maxElement - minElement) / 2;
            indexInChunk = midElement * gp.getElementSize();
            workingBuffer.position(indexInChunk);
            workingBuffer.get(tmpKey);
            compare = KeyUtils.compareKey(actualKey, tmpKey);
            if (compare == 0) {
                // first read the old element
                workingBuffer.position(indexInChunk);
                byte[] b = new byte[gp.getElementSize()];
                workingBuffer.get(b);
                @SuppressWarnings("unchecked")
                Data toUpdate = (Data)prototype.fromByteBuffer(ByteBuffer.wrap(b));
                // update the old element and write it
                toUpdate.update(data);
                workingBuffer.position(indexInChunk);
                workingBuffer.put(data.toByteBuffer());
                return indexInChunk;
            } else if (compare < 0) {
                maxElement = midElement - 1;
            } else {
                minElement = midElement + 1;
            }
        }
        return -1;
    }
}
 public class AbstractKVStorableComparator implements Comparator<AbstractKVStorable> {
    @Override
    public int compare(AbstractKVStorable o1, AbstractKVStorable o2) {
        return KeyUtils.compareKey(o1.getKey(), o2.getKey());
    }
}
 public class ByteArrayComparator implements Comparator<byte[]> {
    @Override
    public int compare(byte[] o1, byte[] o2) {
        return KeyUtils.compareKey(o1, o2);
    }
} public class KeyUtils {
    /**
     * transforms the given array of longs to an array of byte-arrays
     * 
     * @param l
     *            the longs to transform
     * @return the to l corresponding array of byte-arrays
     */
    public static byte[][] toByteArray(long[] l) {
        byte[][] b = new byte[l.length][];
        for (int i = 0; i < b.length; i++) {
            b[i] = Bytes.toBytes(l[i]);
        }
        return b;
    }

    /**
     * Checks if the elements of the given key up the given length are 0, or the whole array is null.
     * 
     * @param key
     * @param length
     * @return true, if the key is null or all elements in the array are 0.
     */
    public static boolean isNull(byte[] key, int length) {
        if (key == null) {
            return true;
        }
        for (int i = 0; i < Math.min(key.length, length); i++) {
            if (key[i] != 0) {
                return false;
            }
        }
        return true;
    }

    /**
     * Checks if the elements of the given key are 0, or the whole array is null.
     * 
     * @param key
     * @param length
     * @return true, if the key is null or all elements in the array are 0.
     */
    public static boolean isNull(byte[] key) {
        if (key == null) {
            return true;
        }
        return isNull(key, key.length);
    }

    /**
     * Compares the two byte-arrays on the basis of unsigned bytes. The array will be compared by each element up to the
     * length of the smaller array. If all elements are equal and the array are not equal sized, the larger array is
     * seen as larger.
     * 
     * @param key1
     * @param key2
     * 
     * @return <0 if key1 < key2<br>
     *         0 if key1 == key2<br>
     *         >0 if key1 > key2
     */
    public static int compareKey(byte[] key1, byte[] key2) {
        return compareKey(key1, key2, Math.min(key1.length, key2.length));
    }

    /**
     * Compares the two byte-arrays on the basis of unsigned bytes. The array will be compared by each element up to the
     * length of the smaller array or length. If all elements are equal and the array are not equal sized, the larger
     * array is seen as larger.<br>
     * The parameter length shows, up to which position the arrays are compared
     * 
     * @param key1
     * @param key2
     * @param length
     * 
     * @return <0 if key1 < key2<br>
     *         0 if key1 == key2<br>
     *         >0 if key1 > key2
     */
    // TODO If the given keys have different length the function will not give the right result.
    public static int compareKey(byte[] key1, byte[] key2, int length) {
        return Bytes.compareTo(key1, 0, length, key2, 0, length);
    }

    /**
     * Generates a {@link RangeHashFunction}.
     * 
     * @param min
     *            the minimal value to expect
     * @param max
     *            the maximal value to expect
     * @param buckets
     *            the number of buckets
     * @param suffix
     *            the suffix for all files
     * @param prefix
     *            a prefix for all files
     * @return String representation of the the RangeHashFunction
     * @throws Exception
     */
    public static String generateHashFunction(byte[] min, byte[] max, int buckets, String suffix, String prefix)
            throws Exception {
        String[] Sbuckets = new String[buckets];
        for (int i = 0; i < buckets; i++) {
            Sbuckets[i] = i + "";
        }
        return generateRangeHashFunction(min, max, Sbuckets, suffix, prefix);
    }

    /**
     * Generates a {@link RangeHashFunction}.
     * 
     * @param min
     *            the minimal value to expect
     * @param max
     *            the maximal value to expect
     * @param buckets
     *            an array with the names of the buckets
     * @param suffix
     *            the suffix for all files
     * @param prefix
     *            a prefix for all files
     * @return String representation of the the RangeHashFunction
     * @throws Exception
     */
    public static String generateRangeHashFunction(byte[] min, byte[] max, String[] buckets, String suffix,
            String prefix) throws Exception {
        if (compareKey(min, max) > 0) {
            throw new Exception("The given min is not larger than the max. Buckets could not be determined");
        }
        byte[][] ranges = getMaxValsPerRange(min, max, buckets.length);

        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < min.length; i++) {
            sb.append("b").append("\t");
        }
        sb.append("filename").append("\n");

        for (int i = 0; i < buckets.length; ++i) {
            byte[] val = ranges[i];
            for (int j = 0; j < val.length; j++) {
                int k = val[j] & 0xff;
                sb.append(k + "\t");
            }
            sb.append(prefix + buckets[i] + suffix + "\n");
        }
        return sb.toString();
    }

    /**
     * 
     * @param min
     *            the minimal value to expect
     * @param max
     *            the maximal value to expect
     * @param num
     *            the number of ranges
     * @return an array containing the maximum value per range.
     */
    public static byte[][] getMaxValsPerRange(byte[] min, byte[] max, int num) {
        return Arrays.copyOfRange(Bytes.split(min, max, false, num - 1), 1, num + 1);
    }

    /**
     * Generates a string representation of the given key
     * 
     * @param key
     * @return the String representation of the given key
     */
    public static String toStringUnsignedInt(byte[] key) {
        StringBuilder result = new StringBuilder();
        for (int i = 0; i < key.length; i++) {
            result.append(key[i] & 0xFF);
            result.append(' ');
        }
        return result.toString();
    }
}

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.carrotsearch.hppc.IntObjectOpenHashMap;
import com.carrotsearch.hppc.cursors.IntObjectCursor;
import com.unister.semweb.drums.DRUMSParameterSet;
import com.unister.semweb.drums.bucket.Bucket;
import com.unister.semweb.drums.bucket.BucketContainer;
import com.unister.semweb.drums.bucket.BucketContainerException;
import com.unister.semweb.drums.bucket.DynamicMemoryAllocater;
import com.unister.semweb.drums.bucket.hashfunction.AbstractHashFunction;
import com.unister.semweb.drums.file.FileLockException;
import com.unister.semweb.drums.file.HeaderIndexFile;
import com.unister.semweb.drums.file.IndexForHeaderIndexFile;
import com.unister.semweb.drums.storable.AbstractKVStorable;
import com.unister.semweb.drums.storable.GeneralStorable;
import com.unister.semweb.drums.sync.SyncManager;
import com.unister.semweb.drums.sync.synchronizer.ISynchronizerFactory;
import com.unister.semweb.drums.sync.synchronizer.SynchronizerFactory;
import com.unister.semweb.drums.sync.synchronizer.UpdateOnlySynchronizer;
import com.unister.semweb.drums.util.AbstractKVStorableComparator;
import com.unister.semweb.drums.util.ByteArrayComparator;
import com.unister.semweb.drums.util.KeyUtils;

/* Copyright (C) 2012-2013 Unister GmbH
 * 
 * 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 2 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, write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */
package com.unister.semweb.drums.api;




/**
 * An instance of this class provides access to a DRUMS-table. The instance allows managing the storage of
 * {@link AbstractKVStorable}s. Use the static methods in {@link DRUMSInstantiator} to get an instance of {@link DRUMS}.<br>
 * <br>
 * Use the method {@link #insertOrMerge(AbstractKVStorable...)} to insert or merge records.<br>
 * To update records use the method {@link #update(AbstractKVStorable...)}.<br>
 * Single selects can be performed by the method {@link #select(byte[])}.<br>
 * To perform range selects, a {@link DRUMSReader} should be instantiated using the method {@link #getReader()}.<br>
 * The whole table is scanned best, using an {@link DRUMSIterator} obtained by {@link #getIterator()}.
 * 
 * @author Martin Nettling, Nils Thieme
 * 
 * @param <Data>
 *            an implementation of {@link AbstractKVStorable}, e.g. {@link GeneralStorable}
 */
public class DRUMS<Data extends AbstractKVStorable> {
    private static final Logger logger = LoggerFactory.getLogger(DRUMS.class);

    /** the accessmode for DRUMS */
    public enum AccessMode {
        READ_ONLY, READ_WRITE;
    }

    /** the hashfunction, decides where to search for element, or where to store it */
    private AbstractHashFunction hashFunction;

    /** an array, containing all used buckets */
    private Bucket<Data>[] buckets;

    /** the container for all buckets */
    private BucketContainer<Data> bucketContainer;

    /** the Synchronizer factory, is needed to decide how to insert/update elements */
    private ISynchronizerFactory<Data> synchronizerFactory;

    /** the buffer manages the different synchronize-processes */
    private SyncManager<Data> syncManager;

    /** a prototype of the elements to store */
    private Data prototype;

    /** A pointer to the {@link DRUMSParameterSet} used by this {@link DRUMS} */
    protected DRUMSParameterSet<Data> gp;

    protected DRUMSReader<Data> reader_instance;

    /**
     * This constructor should only be called by factory methods from this package.
     * 
     * @param hashFunction
     *            a consistent hash-function
     * @param accessMode
     *            can be read or read/write
     * @param gp
     *            contains all needed settings
     * @throws Exception
     */
    protected DRUMS(AbstractHashFunction hashFunction, AccessMode accessMode, DRUMSParameterSet<Data> gp)
            throws IOException {
        this.prototype = gp.getPrototype();
        this.hashFunction = hashFunction;
        this.gp = gp;
        DynamicMemoryAllocater.instantiate(gp);
        gp.MIN_ELEMENT_IN_BUCKET_BEFORE_SYNC = (int) ((gp.BUCKET_MEMORY - gp.BUCKET_MEMORY % gp.MEMORY_CHUNK)
                / hashFunction.getNumberOfBuckets() / prototype.getSize() / 2);
        logger.info("Setted MIN_ELEMENT_IN_BUCKET_BEFORE_SYNC to {}", gp.MIN_ELEMENT_IN_BUCKET_BEFORE_SYNC);
        if (accessMode == AccessMode.READ_WRITE) {
            @SuppressWarnings("unchecked")
            Bucket<Data>[] tmp = new Bucket[hashFunction.getNumberOfBuckets()];
            buckets = tmp;
            for (int i = 0; i < hashFunction.getNumberOfBuckets(); i++) {
                buckets[i] = new Bucket<Data>(i, gp);
                String tmpFileName = gp.DATABASE_DIRECTORY + "/" + hashFunction.getFilename(i);
                if (!new File(tmpFileName).exists()) {

                    HeaderIndexFile<Data> tmpFile;
