empq_adaptive_impl.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 __EMPQ_ADAPTIVE_IMPL_H
#define __EMPQ_ADAPTIVE_IMPL_H
 
#include <stdio.h>
#include <assert.h>
 
#include "ami_config.h"
#include "ami_stream.h"
#include "mm.h"
#include "mm_utils.h"
#include "empq_adaptive.h"
 
#include "ami_sort.h"
 
// EMPQAD_DEBUG defined in "empqAdaptive.H"
 
//------------------------------------------------------------
// allocate an internal pqueue of size precisely twice
// the size of the pqueue within the em_pqueue;
//
// This constructor uses a user defined amount of memory
 
template <class T, class Key>
EMPQueueAdaptive<T, Key>::EMPQueueAdaptive(size_t inMem)
{
    regim = INMEM;
    EMPQAD_DEBUG cout << "EMPQUEUEADAPTIVE: starting in-memory pqueue" << endl;
 
    //------------------------------------------------------------
    // set the size precisely as in empq constructor since we cannot
    // really call the em__pqueue constructor, because we don't want
    // the space allocated; we just want the sizes;
    // AMI_err ae;
    EMPQAD_DEBUG cout << "EMPQUEUEADAPTIVE: desired memory: "
                      << ((float)inMem / (1 << 20)) << "MB" << endl;
 
    initPQ(inMem);
}
 
//------------------------------------------------------------
// This more resembles the original constructor which is greedy
template <class T, class Key>
EMPQueueAdaptive<T, Key>::EMPQueueAdaptive()
{
    regim = INMEM;
    EMPQAD_DEBUG cout << "EMPQUEUEADAPTIVE: starting in-memory pqueue" << endl;
 
    //------------------------------------------------------------
    // set the size precisely as in empq constructor since we cannot
    // really call the em__pqueue constructor, because we don't want
    // the space allocated; we just want the sizes;
    size_t mm_avail = getAvailableMemory();
    EMPQAD_DEBUG cout << "EMPQUEUEADAPTIVE: available memory: "
                      << ((float)mm_avail / (1 << 20)) << "MB" << endl;
 
    initPQ(mm_avail);
}
 
//------------------------------------------------------------
// This method initialized the PQ based on the memory passed
// into it
template <class T, class Key>
void EMPQueueAdaptive<T, Key>::initPQ(size_t initMem)
{
    AMI_err ae;
    EMPQAD_DEBUG cout << "EMPQUEUEADAPTIVE: desired memory: "
                      << ((float)initMem / (1 << 20)) << "MB" << endl;
 
    /* same calculations as empq constructor in order to estimate
       overhead memory; this is because we want to allocate a pqueue of
       size exactly double the size of the pqueue inside the empq;
       switching from this pqueue to empq when the memory fills up will
       be simple */
    //------------------------------------------------------------
    // AMI_STREAM memory usage
    size_t sz_stream;
    AMI_STREAM<T> dummy;
    if ((ae = dummy.main_memory_usage(&sz_stream, MM_STREAM_USAGE_MAXIMUM)) !=
        AMI_ERROR_NO_ERROR) {
        cerr << "EMPQueueAdaptive constr: failing to get stream_usage\n";
        exit(1);
    }
 
    // account for temporary memory usage
    unsigned short max_nbuf = 2;
    unsigned int buf_arity = initMem / (2 * sz_stream);
    if (buf_arity > MAX_STREAMS_OPEN)
        buf_arity = MAX_STREAMS_OPEN;
    unsigned long mm_overhead = buf_arity * sizeof(merge_key<Key>) +
                                max_nbuf * sizeof(em_buffer<T, Key>) +
                                2 * sz_stream + max_nbuf * sz_stream;
    mm_overhead *= 8; // overestimate..this should be fixed with
    // a precise accounting of the extra memory required
 
    EMPQAD_DEBUG cout << "sz_stream: " << sz_stream
                      << " buf_arity: " << buf_arity
                      << " mm_overhead: " << mm_overhead
                      << " mm_avail: " << initMem << ".\n";
 
    EMPQAD_DEBUG cout << "EMPQUEUEADAPTIVE: memory overhead set to "
                      << ((float)mm_overhead / (1 << 20)) << "MB" << endl;
    if (mm_overhead > initMem) {
        cerr << "overhead bigger than available memory (" << initMem << "); "
             << "increase -m and try again\n";
        exit(1);
    }
    initMem -= mm_overhead;
    //------------------------------------------------------------
 
    long pqsize = initMem / sizeof(T);
    EMPQAD_DEBUG cout << "EMPQUEUEADAPTIVE: pqsize set to " << pqsize << endl;
 
    // initialize in memory pqueue and set em to NULL
    im = new MinMaxHeap<T>(pqsize);
    assert(im);
    em = NULL;
}
 
template <class T, class Key>
EMPQueueAdaptive<T, Key>::~EMPQueueAdaptive()
{
    switch (regim) {
    case INMEM:
        delete im;
        break;
    case EXTMEM:
        delete em;
        break;
    case EXTMEM_DEBUG:
        delete dim;
        delete em;
        break;
    }
}
 
// return the maximum nb of elts that can fit
template <class T, class Key>
long EMPQueueAdaptive<T, Key>::maxlen() const
{
    long m = -1;
    switch (regim) {
    case INMEM:
        assert(im);
        m = im->get_maxsize();
        break;
    case EXTMEM:
        assert(em);
        m = em->maxlen();
        break;
    case EXTMEM_DEBUG:
        m = em->maxlen();
        break;
    }
    return m;
}
 
// return true if empty
template <class T, class Key>
bool EMPQueueAdaptive<T, Key>::is_empty() const
{
    bool v = false;
    switch (regim) {
    case INMEM:
        assert(im);
        v = im->empty();
        break;
    case EXTMEM:
        assert(em);
        v = em->is_empty();
        break;
    case EXTMEM_DEBUG:
        assert(dim->empty() == em->is_empty());
        v = em->is_empty();
        break;
    }
    return v;
}
 
// return true if full
template <class T, class Key>
bool EMPQueueAdaptive<T, Key>::is_full() const
{
    cerr << "EMPQueueAdaptive::is_full(): sorry not implemented\n";
    assert(0);
    exit(1);
}
 
// return the element with minimum priority in the structure
template <class T, class Key>
bool EMPQueueAdaptive<T, Key>::min(T &elt)
{
    bool v = false, v1;
    T tmp;
    switch (regim) {
    case INMEM:
        assert(im);
        v = im->min(elt);
        break;
    case EXTMEM:
        assert(em);
        v = em->min(elt);
        break;
    case EXTMEM_DEBUG:
        v1 = dim->min(tmp);
        v = em->min(elt);
        // dim->verify();
        if (!(tmp == elt)) {
            cerr << "------------------------------" << endl;
            cerr << dim << endl;
            cerr << "------------------------------" << endl;
            em->print();
            cerr << "------------------------------" << endl;
            cerr << "tmp=" << tmp << endl;
            cerr << "elt=" << elt << endl;
            cerr << "------------------------------" << endl;
            dim->destructiveVerify();
        }
        assert(v == v1);
        assert(tmp == elt);
        break;
    }
    return v;
}
 
/* switch over to using an external priority queue */
template <class T, class Key>
void EMPQueueAdaptive<T, Key>::clear()
{
    switch (regim) {
    case INMEM:
        im->clear();
        break;
    case EXTMEM:
        em->clear();
        break;
    case EXTMEM_DEBUG:
        dim->clear();
        break;
    }
}
 
template <class T, class Key>
void EMPQueueAdaptive<T, Key>::verify()
{
    switch (regim) {
    case INMEM:
        im->verify();
        break;
    case EXTMEM:
        break;
    case EXTMEM_DEBUG:
        dim->verify();
        break;
    }
}
 
// extract all elts with min key, add them and return their sum
template <class T, class Key>
bool EMPQueueAdaptive<T, Key>::extract_all_min(T &elt)
{
    bool v = false, v1;
    T tmp;
    switch (regim) {
    case INMEM:
        assert(im);
        v = im->extract_all_min(elt);
        break;
    case EXTMEM:
        assert(em);
        v = em->extract_all_min(elt);
        break;
    case EXTMEM_DEBUG:
        v1 = dim->extract_all_min(tmp);
        v = em->extract_all_min(elt);
        assert(dim->size() == em->size());
        assert(v == v1);
        assert(tmp == elt);
        break;
    }
    return v;
}
 
// return the nb of elements in the structure
template <class T, class Key>
long EMPQueueAdaptive<T, Key>::size() const
{
    long v = 0, v1;
    switch (regim) {
    case INMEM:
        assert(im);
        v = im->size();
        break;
    case EXTMEM:
        assert(em);
        v = em->size();
        break;
    case EXTMEM_DEBUG:
        v1 = dim->size();
        v = em->size();
        assert(v == v1);
        break;
    }
    return v;
}
 
// ----------------------------------------------------------------------
template <class T, class Key>
bool EMPQueueAdaptive<T, Key>::extract_min(T &elt)
{
    bool v = false, v1;
    T tmp;
    switch (regim) {
    case INMEM:
        assert(im);
        v = im->extract_min(elt);
        break;
    case EXTMEM:
        assert(em);
        v = em->extract_min(elt);
        break;
    case EXTMEM_DEBUG:
        v1 = dim->extract_min(tmp);
        v = em->extract_min(elt);
        assert(v == v1);
        assert(tmp == elt);
        assert(dim->size() == em->size());
        break;
    }
    return v;
}
 
//------------------------------------------------------------
/* insert an element; if regim == INMEM, try insert it in im, and if
   it is full, extract_max pqsize/2 elements of im into a stream,
   switch to EXTMEM and insert the stream into em; if regim is
   EXTMEM, insert in em; */
template <class T, class Key>
bool EMPQueueAdaptive<T, Key>::insert(const T &elt)
{
    bool v = false;
    switch (regim) {
    case INMEM:
        if (!im->full()) {
            im->insert(elt);
            v = true;
        }
        else {
            makeExternal();
            v = em->insert(elt); // insert the element
        }
        break;
    case EXTMEM:
        v = em->insert(elt);
        break;
    case EXTMEM_DEBUG:
        dim->insert(elt);
        v = em->insert(elt);
        assert(dim->size() == em->size());
        break;
    }
    return v;
}
 
template <class T, class Key>
void EMPQueueAdaptive<T, Key>::makeExternalDebug()
{
    assert(size() == 0);
    switch (regim) {
    case INMEM:
        makeExternal();
        break;
    case EXTMEM:
        break;
    case EXTMEM_DEBUG:
        assert(0);
        break;
    }
    dim = new UnboundedMinMaxHeap<T>();
    regim = EXTMEM_DEBUG;
}
 
template <class T>
class baseCmpType {
public:
    static int compare(const T &x, const T &y)
    {
        return (x < y ? -1 : (x > y ? 1 : 0));
    }
};
 
/* switch over to using an external priority queue */
template <class T, class Key>
void EMPQueueAdaptive<T, Key>::makeExternal()
{
    AMI_err ae;
#ifndef NDEBUG
    long sizeCheck;
    sizeCheck = size();
#endif
 
    assert(regim == INMEM);
    regim = EXTMEM;
 
    EMPQAD_DEBUG cout << endl
                      << "EMPQUEUEADAPTIVE: memory full: "
                      << "switching to external-memory pqueue " << endl;
 
    // create an AMI_stream and write in it biggest half elts of im;
    AMI_STREAM<T> *amis0 = new AMI_STREAM<T>();
    AMI_STREAM<T> *amis1 = NULL;
    assert(amis0);
    unsigned long pqsize = im->size();
    // assert(im->size() == im->get_maxsize());
    T x;
    for (unsigned long i = 0; i < pqsize / 2; i++) {
        int z = im->extract_max(x);
        assert(z);
        ae = amis0->write_item(x);
        assert(ae == AMI_ERROR_NO_ERROR);
    }
    assert(amis0->stream_len() == pqsize / 2);
    EMPQAD_DEBUG
    {
        cout << "written " << pqsize / 2 << " elts to stream\n";
        cout.flush();
    }
 
    assert(im->size() == pqsize / 2 + (pqsize % 2));
 
    EMPQAD_DEBUG LOG_avail_memo();
 
    // sort the stream
    baseCmpType<T> fun;
    AMI_sort(amis0, &amis1, &fun); // XXX laura: replaced this to use a cmp obj
    assert(amis1);
    delete amis0;
    EMPQAD_DEBUG
    {
        cout << "sorted the stream\n";
        cout.flush();
    }
 
    EMPQAD_DEBUG LOG_avail_memo();
 
    // set im to NULL and initialize em from im and amis1
    em = new em_pqueue<T, Key>(im, amis1);
    im = NULL;
    assert(em);
    EMPQAD_DEBUG
    {
        cout << "empq initialized from im\n";
        cout.flush();
    }
    EMPQAD_DEBUG
    {
        em->print_size();
    }
 
    EMPQAD_DEBUG LOG_avail_memo();
#ifndef NDEBUG
    assert(sizeCheck == size());
#endif
}
 
#endif