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();
                }
            }
        }
    }
private void synchronizeBucketsWithHDD ( ) { int synchronizedBuckets = 0 ; for ( int i = 0 ; i < numberOfBuckets ; i ++ ) { Bucket < Data > oldBucket = bucketContainer . getBucket ( i ) ; if ( oldBucket . elementsInBucket == 0 ) { synchronizedBuckets ++ ; continue ; } if ( DynamicMemoryAllocater . INSTANCES [ gp . instanceID ] . getFreeMemory ( ) == 0 ) { log . info ( "No memory free, theoretically I must force synchronization" ) ; } long elapsedTime = System . currentTimeMillis ( ) - oldBucket . getCreationTime ( ) ; if ( oldBucket . elementsInBucket >= gp . MIN_ELEMENT_IN_BUCKET_BEFORE_SYNC || DynamicMemoryAllocater . INSTANCES [ gp . instanceID ] . getFreeMemory ( ) == 0 || elapsedTime > maxBucketStorageTime || shutDownInitiated || forceInitiated ) { if ( ! startNewThread ( i ) ) { sleep ( ) ; } } } if ( shutDownInitiated ) { log . info ( "{} of {} buckets were synchronized." , synchronizedBuckets , bucketContainer . getNumberOfBuckets ( ) ) ; } 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 ( ) ; } } } }
This method manages the synchronizing-process between {@link BucketContainer} and HDD. It is repeatedly called by
<code>run()</code>, till shutdown is initiated.