replacementHeap.h

Go to the documentation of this file.

/****************************************************************************
 *
 *  MODULE:     iostream
 *
 
 *  COPYRIGHT (C) 2007 Laura Toma
 *
 *
 
 *  Iostream is a library that implements streams, external memory
 *  sorting on streams, and an external memory priority queue on
 *  streams. These are the fundamental components used in external
 *  memory algorithms.
 
 * Credits: The library was developed by Laura Toma.  The kernel of
 * class STREAM is based on the similar class existent in the GPL TPIE
 * project developed at Duke University. The sorting and priority
 * queue have been developed by Laura Toma based on communications
 * with Rajiv Wickremesinghe. The library was developed as part of
 * porting Terraflow to GRASS in 2001.  PEARL upgrades in 2003 by
 * Rajiv Wickremesinghe as part of the Terracost project.
 
 *
 *  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.  *
 *  **************************************************************************/
 
#ifndef REPLACEMENT_QUEUE_H
#define REPLACEMENT_QUEUE_H
 
#include <assert.h>
 
#include "queue.h"
 
#define RHEAP_DEBUG if (0)
 
/*****************************************************************/
/* encapsulation of the element and the run it comes from;
 */
template <class T>
class HeapElement {
public:
    T value;
    AMI_STREAM<T> *run;
 
    HeapElement() : run(NULL) {};
 
    friend ostream &operator<<(ostream &s, const HeapElement &p)
    {
        return s << "[" << p.value << "]";
    };
};
 
/*****************************************************************/
/*
This is a heap of HeapElements, i.e. elements which come from streams;
when an element is consumed, the heap knows how to replace it with the
next element from the same stream.
 
Compare is a class that has a member function called "compare" which
is used to compare two elements of type T
*/
template <class T, class Compare>
class ReplacementHeap {
private:
    HeapElement<T> *mergeHeap; // the heap;
    size_t arity;              // max size
    size_t size; // represents actual size, i.e. the nb of (non-empty)
                 // runs; they are stored contigously in the first
                 //<size> positions of mergeHeap; once a run becomes
                 // empty, it is deleted and size is decremented.  size
                 // stores the next position where a HeapElement can be
                 // added.
 
protected:
    void heapify(size_t i);
 
    void buildheap();
 
    /* for each run in the heap, read an element from the run into the heap */
    void init();
 
    /* add a run; make sure total nb of runs does not exceed heap arity */
    void addRun(AMI_STREAM<T> *run);
 
    /* delete the i-th run (and the element); that is, swap the ith run
       with the last one, and decrement size; just like in a heap, but
       no heapify.  Note: this function messes up the heap order. If the
       user wants to maintain heap property should call heapify
       specifically.
    */
    void deleteRun(size_t i);
 
public:
    // allocate array mergeHeap and the runs in runList
    ReplacementHeap(size_t arity, queue<char *> *runList);
 
    // delete array mergeHeap
    ~ReplacementHeap();
 
    // is heap empty?
    int empty() const { return (size == 0); }
 
    // delete mergeHeap[0].value, replace it with the next element from
    // the same stream, and re-heapify
    T extract_min();
 
    ostream &print(ostream &s) const
    {
        char *runname;
        off_t runlen;
        s << "Replacementheap " << this << ": " << size << " runs";
        for (size_t i = 0; i < size; i++) {
            s << endl << "  <-  i=" << i << ": " << mergeHeap[i].run;
            assert(mergeHeap[i].run);
            mergeHeap[i].run->name(&runname);
            runlen = mergeHeap[i].run->stream_len();
            s << ", " << runname << ", len=" << runlen;
            delete runname; // this should be safe
        }
        s << endl;
        return s;
    }
};
 
/*****************************************************************/
template <class T, class Compare>
ReplacementHeap<T, Compare>::ReplacementHeap(size_t g_arity,
                                             queue<char *> *runList)
{
    char *name = NULL;
 
    assert(runList && g_arity > 0);
 
    RHEAP_DEBUG cerr << "ReplacementHeap arity=" << g_arity << "\n";
 
    arity = g_arity;
    size = 0; // no run yet
 
    mergeHeap = new HeapElement<T>[arity];
    for (unsigned int i = 0; i < arity; i++) {
        // pop a stream from the list  and add it to heap
        runList->dequeue(&name);
        AMI_STREAM<T> *str = new AMI_STREAM<T>(name);
        assert(str);
        delete name; // str makes its own copy
        addRun(str);
    }
    init();
}
 
/*****************************************************************/
template <class T, class Compare>
ReplacementHeap<T, Compare>::~ReplacementHeap()
{
 
    if (!empty()) {
        cerr << "warning: ~ReplacementHeap: heap not empty!\n";
    }
    // delete the runs first
    for (size_t i = 0; i < size; i++) {
        if (mergeHeap[i].run)
            delete mergeHeap[i].run;
    }
    delete[] mergeHeap;
}
 
/*****************************************************************/
/* add a run; make sure total nb of runs does not exceed heap arity
 */
template <class T, class Compare>
void ReplacementHeap<T, Compare>::addRun(AMI_STREAM<T> *r)
{
 
    assert(r);
 
    if (size == arity) {
        cerr << "ReplacementHeap::addRun size =" << size << ",arity=" << arity
             << " full, cannot add another run.\n";
        assert(0);
        exit(1);
    }
    assert(size < arity);
 
    mergeHeap[size].run = r;
    size++;
 
    RHEAP_DEBUG
    {
        char *strname;
        r->name(&strname);
        cerr << "ReplacementHeap::addRun added run " << strname
             << " (rheap size=" << size << ")" << endl;
        delete strname;
        cerr.flush();
    }
}
 
/*****************************************************************/
/* delete the i-th run (and the value); that is, swap ith element with
   the last one, and decrement size; just like in a heap, but no
   heapify.  Note: this function messes up the heap order. If the user
   wants to maintain heap property should call heapify specifically.
 */
template <class T, class Compare>
void ReplacementHeap<T, Compare>::deleteRun(size_t i)
{
 
    assert(i < size && mergeHeap[i].run);
 
    RHEAP_DEBUG
    {
        cerr << "ReplacementHeap::deleteRun  deleting run " << i << ", "
             << mergeHeap[i].run << endl;
        print(cerr);
    }
 
    // delete it
    delete mergeHeap[i].run;
    // and replace it with
    if (size > 1) {
        mergeHeap[i].value = mergeHeap[size - 1].value;
        mergeHeap[i].run = mergeHeap[size - 1].run;
    }
    size--;
}
 
/*****************************************************************/
/* for each run in the heap, read an element from the run into the
   heap; if ith run is empty, delete it
*/
template <class T, class Compare>
void ReplacementHeap<T, Compare>::init()
{
    AMI_err err;
    T *elt;
    size_t i;
 
    RHEAP_DEBUG cerr << "ReplacementHeap::init ";
 
    i = 0;
    while (i < size) {
 
        assert(mergeHeap[i].run);
 
        // Rewind run i
        err = mergeHeap[i].run->seek(0);
        if (err != AMI_ERROR_NO_ERROR) {
            cerr << "ReplacementHeap::Init(): cannot seek run " << i << "\n";
            assert(0);
            exit(1);
        }
        // read first item from run i
        err = mergeHeap[i].run->read_item(&elt);
        if (err != AMI_ERROR_NO_ERROR) {
            if (err == AMI_ERROR_END_OF_STREAM) {
                deleteRun(i);
                // need to iterate one more time with same i;
            }
            else {
                cerr << "ReplacementHeap::Init(): cannot read run " << i
                     << "\n";
                assert(0);
                exit(1);
            }
        }
        else {
            // copy.... can this be avoided? xxx
            mergeHeap[i].value = *elt;
 
            i++;
        }
    }
    buildheap();
}
 
// Helper functions for navigating through a binary heap.
/* for simplicity the heap structure is slightly modified as:
   0
   |
   1
   /\
  2  3
 /\  /\
4 5 6  7
 
*/
 
// The children of an element of the heap.
static inline size_t rheap_lchild(size_t index)
{
    return 2 * index;
}
 
static inline size_t rheap_rchild(size_t index)
{
    return 2 * index + 1;
}
 
// The parent of an element.
static inline size_t rheap_parent(size_t index)
{
    return index >> 1;
}
// this functions are duplicated in pqheap.H...XXX
 
/*****************************************************************/
template <class T, class Compare>
void ReplacementHeap<T, Compare>::heapify(size_t i)
{
    size_t min_index = i;
    size_t lc = rheap_lchild(i);
    size_t rc = rheap_rchild(i);
 
    Compare cmpobj;
    assert(i < size);
    if ((lc < size) && (cmpobj.compare(mergeHeap[lc].value,
                                       mergeHeap[min_index].value) == -1)) {
        min_index = lc;
    }
    if ((rc < size) && (cmpobj.compare(mergeHeap[rc].value,
                                       mergeHeap[min_index].value) == -1)) {
        min_index = rc;
    }
 
    if (min_index != i) {
        HeapElement<T> tmp = mergeHeap[min_index];
 
        mergeHeap[min_index] = mergeHeap[i];
        mergeHeap[i] = tmp;
 
        heapify(min_index);
    }
 
    return;
}
 
/*****************************************************************/
template <class T, class Compare>
void ReplacementHeap<T, Compare>::buildheap()
{
 
    if (size > 1) {
        for (int i = rheap_parent(size - 1); i >= 0; i--) {
            heapify(i);
        }
    }
    RHEAP_DEBUG cerr << "Buildheap done\n";
    return;
}
 
/*****************************************************************/
template <class T, class Compare>
T ReplacementHeap<T, Compare>::extract_min()
{
    T *elt, min;
    AMI_err err;
 
    assert(!empty()); // user's job to check first if it's empty
    min = mergeHeap[0].value;
 
    // read a new element from the same run
    assert(mergeHeap[0].run);
    err = mergeHeap[0].run->read_item(&elt);
    if (err != AMI_ERROR_NO_ERROR) {
        // if run is empty, delete it
        if (err == AMI_ERROR_END_OF_STREAM) {
            RHEAP_DEBUG cerr << "rheap extract_min: run " << mergeHeap[0].run
                             << " empty. deleting\n ";
            deleteRun(0);
        }
        else {
            cerr << "ReplacementHeap::extract_min: cannot read\n";
            assert(0);
            exit(1);
        }
    }
    else {
        // copy...can this be avoided?
        mergeHeap[0].value = *elt;
    }
 
    // restore heap
    if (size > 0)
        heapify(0);
 
    return min;
}
 
#endif