---

buckets.cpp

/****************************************************
 *  Bachelor Project:                               *
 *  External Memory Minimum Spanning Trees          *
 *  by Dominik Schultes <mail@dominik-schultes.de>  *
 *                                                  *
 *                buckets.cpp                       *
 *                                                  *
 *  April 23, 2003                                  *
 ****************************************************/

#include "buckets.h"
#include "../data_structures/duplicatesRemover.h"
#include "../utils/randomizer.h"


template <NodeCount nodesPerBucket>
void Buckets<nodesPerBucket>::initBuckets()
{    
    // create the external buckets
    DS_VERBOSE1( std::cout << std::endl << "create external buckets" << std::endl;
                 Percent percent(_extBuckets.size()) );
                 
    _firstExtBucket = new InternalMemoryBucket(_noOfNodesInIntMem,0);
    for (BucketID i=0; i<_extBuckets.size(); i++) {
        DS_VERBOSE1( percent.printStatus(i) );
        _extBuckets[i] = new EdgesOfSeveralNodes( _prefetchPool, _writePool, 0 );
    }

      
    // distribute the edges to the external buckets
    DS_RANDOMIZE( Randomizer<RandomizerFeistel> randomizer(_graph.noOfNodes()) );
    
    DS_VERBOSE1( std::cout << std::endl
                 << "distribute edges to external buckets" << std::endl;
                 percent.reinit(_graph.size(), true) );
        
    for (; !_graph.empty(); _graph.pop_back()) {
        DS_DONT_RANDOMIZE( RelabeledEdge edge(_graph.back()) );
        DS_RANDOMIZE( RelabeledEdge edge = randomizer.randomize(_graph.back()) );

        DS_VERBOSE1( percent.printStatus(_graph.size()) );
        
        DS_VERBOSE2( std::cout << edge; )

        
        // The source has to be greater than the target.
        if (edge.source() < edge.target()) edge.swap();

        DS_VERBOSE2( std::cout << "  swap  " << edge
                     << " bucketID: " << bucketID(edge.source())
                     << "  relabeled " );

        // add the current edge to the appropriate external bucket
        addToExternalBucket(edge);
        

        DS_VERBOSE2(
            if (bucketID(edge.source()) == -1)
                std::cout << _firstExtBucket->back() << std::endl;
            else
                std::cout << _extBuckets[bucketID(edge.source())]->top() << std::endl; )
    }

    // delete the input graph
    delete &_graph;

    DS_LOGGING1( logging.events().push_back(
                     LoggingEvent("edges distributed to external buckets")) );
}


template <NodeCount nodesPerBucket>
void Buckets<nodesPerBucket>::reduceNodes()
{
    // create/initialize the internal buckets
    DS_REMOVE_DUPLICATES(
        DuplicatesRemover< Buckets<nodesPerBucket> > *duplRem =
            new DuplicatesRemover< Buckets<nodesPerBucket> >(this);
    )
    
    PoolEdgesOfOneNode *pool = new PoolEdgesOfOneNode();
    for (NodeCount i=0; i<nodesPerBucket; i++) {
        _intBuckets[i].setPool(pool);
    }

    // The identifier of the first node in the current external bucket will be used
    // to compute the internal bucket index of a node.
    _firstNodeIDofCurrentBucket = _noOfNodesInIntMem
        + ( _extBuckets.size() * nodesPerBucket );
    
    // Process all buckets from the last to the first but one.
    for (BucketID currentExtBucketID=_extBuckets.size()-1; !_extBuckets.empty();
         currentExtBucketID--) {

        _firstNodeIDofCurrentBucket -= nodesPerBucket;
        
        EdgesOfSeveralNodes & currentExtBucket = *_extBuckets[currentExtBucketID];

        // Set the size of the prefetch buffer that is used by the current external bucket.
        currentExtBucket.set_prefetch_aggr( DS_EXT_PREFETCH_POOL );
        
        DS_LOGGING2( logging.buckets().push_back(
                        LoggingBucket(currentExtBucketID, nodesPerBucket,
                                      currentExtBucket.size())) );
        
        // Read current external bucket and distribute the edges to the internal buckets.
        DS_VERBOSE1( std::cout << std::endl
                     << "process bucket " << currentExtBucketID << std::endl
                     << "  read    ";
                     Percent percent(currentExtBucket.size(), true, 20) );
                
        for (; ! currentExtBucket.empty(); currentExtBucket.pop() ) {
            DS_VERBOSE1( percent.printStatus(currentExtBucket.size()) );
            _intBuckets[currentExtBucket.top().source() - _firstNodeIDofCurrentBucket]
                .push(currentExtBucket.top());
        }

        
        DS_LOGGING2( logging.buckets().back().setBlocksUsed(pool->noOfBlocksUsed());
                     logging.buckets().back().setBlocksFree(pool->noOfBlocksFree()) );
        
        // Relabel the edges. All nodes (in the current bucket)
        // are processed from the last to the first.
        DS_VERBOSE1( std::cout << std::endl << "  relabel ";
                     percent.reinit(nodesPerBucket, true, 20) );
        
        for (NodeID currentNodeIndex = nodesPerBucket-1;
             currentNodeIndex < nodesPerBucket; // NodeID is unsigned
             currentNodeIndex--) {
            
            EdgesOfOneNode &currentIntBucket = _intBuckets[currentNodeIndex];
            
            DS_LOGGING2( logging.buckets().back().
                             addEdgesProcessed(currentIntBucket.size()) );
            DS_VERBOSE1( percent.printStatus(currentNodeIndex) );
            DS_VERBOSE2( std::cout << std::endl
                         << "** currentNodeIndex = " << currentNodeIndex );

            if ( currentIntBucket.empty() ) continue;
            
            // Determine the edge with minimum weight incident to the current node
            // and add it to the resulting minimum spanning tree.
            NodeID newSource = currentIntBucket.determineMinEdge(_result);
            
            // Relabel all edges of the current node.
            for(; ! currentIntBucket.empty(); currentIntBucket.pop() ) {

                // Relabel the current edge.
                // Self loops can be ignored.
                if ( currentIntBucket.top().target() != newSource ) { 

                    DS_DONT_REMOVE_DUPLICATES(
                        add( RelabeledEdge(currentIntBucket.top(), newSource) )
                        );
                    
                    DS_REMOVE_DUPLICATES(
                        duplRem->insert( currentIntBucket.top(), newSource )
                        );
                    
                }
            }

            DS_REMOVE_DUPLICATES( duplRem->clear(newSource) );
            
        }
        
        DS_LOGGING2(
            system("ps -C mst -o rss,%mem --no-headers | head -n 1 >> memoryLog.tmp") );

        // Delete the current bucket which is now empty.
        delete _extBuckets[currentExtBucketID];
        _extBuckets.pop_back();

        // The released memory of the current external bucket
        // can be used by the internal buckets.
        pool->increaseReserveMemory( DS_EXT_BLOCK_SIZE * DS_EXT_PAGE_SIZE );
 
    }

    DS_LOGGING2( logging.buckets().push_back(
                    LoggingBucket(-1, _firstExtBucket->noOfNodes(),
                                  _firstExtBucket->noOfEdges())) );

    // Delete the common pool of the internal buckets.
    delete pool;
    DS_REMOVE_DUPLICATES( delete duplRem );
}


template <NodeCount nodesPerBucket>
Buckets<nodesPerBucket>::BucketID
Buckets<nodesPerBucket>::noOfExtBuckets(NodeCount noOfNodes, NodeCount noOfNodesInIntMem) {
    if (noOfNodes <= noOfNodesInIntMem) return 0;
    
    NodeCount noOfNodesInExtMem = noOfNodes - noOfNodesInIntMem;
    BucketID noOfBuckets = noOfNodesInExtMem / nodesPerBucket;
    if (noOfNodesInExtMem % nodesPerBucket != 0) noOfBuckets++; // round up
    noOfBuckets++; // first external bucket
    
    return noOfBuckets;
}

---