embuffer.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 __EMBUFFER_H
#define __EMBUFFER_H
 
#include <stdio.h>
#include <assert.h>
#include <stdlib.h>
#include <math.h>
 
#include "ami_config.h" //for SAVE_MEMORY
#include "ami_stream.h"
#include "mm.h"
#include "mm_utils.h"
#include "pqheap.h"
 
#define MY_LOG_DEBUG_ID(x) // inhibit debug printing
// #define MY_LOG_DEBUG_ID(x) LOG_DEBUG_ID(x)
 
/**********************************************************
                  DEBUGGING FLAGS
***********************************************************/
 
// setting this enables checking that the streams/arrays inserted in
// buffers are sorted in increasing order
// #define EMBUF_CHECK_INSERT
 
// enable checking that stream name is the same as the one stored in
// the buffer name[]
// #define EMBUF_CHECK_NAME
 
// enable printing names as they are checked
// #define EMBUF_CHECK_NAME_PRINT
 
// enable printing when streams in  a buffer are shifted left to
// check that names are shifted accordingly
// #define EMBUF_DELETE_STREAM_PRINT
 
// enable printing the name of the stream which is inserted in buff
// #define EMBUF_PRINT_INSERT
 
// enable printing the stream names/sizes in cleanup()
// #define EMBUF_CLEANUP_PRINT
 
// enable printing when get/put_stream is called (for each stream)
// #define EMBUF_PRINT_GETPUT_STREAM
 
// enable printing when get/put_streams is called
// #define EMBUF_PRINT_GETPUT_STREAMS
 
/***********************************************************/
 
/*****************************************************************/
/* encapsulation of the key together with stream_id; used during
   stream merging to save space;
*/
template <class KEY>
class merge_key {
public:
    KEY k;
    unsigned int str_id; // id of the stream where key comes from
 
public:
    merge_key() : str_id(0) {}
 
    merge_key(const KEY &x, const unsigned int sid) : k(x), str_id(sid) {}
 
    ~merge_key() {}
 
    void set(const KEY &x, const unsigned int sid)
    {
        k = x;
        str_id = sid;
    }
    KEY key() const { return k; }
    unsigned int stream_id() const { return str_id; }
    KEY getPriority() const { return k; }
 
    friend ostream &operator<<(ostream &s, const merge_key<KEY> &x)
    {
        return s << "<str_id=" << x.str_id << "> " << x.k << " ";
    }
    friend int operator<(const merge_key &x, const merge_key &y)
    {
        return (x.k < y.k);
    }
    friend int operator<=(const merge_key &x, const merge_key &y)
    {
        return (x.k <= y.k);
    }
    friend int operator>(const merge_key &x, const merge_key &y)
    {
        return (x.k > y.k);
    }
    friend int operator>=(const merge_key &x, const merge_key &y)
    {
        return (x.k >= y.k);
    }
    friend int operator!=(const merge_key &x, const merge_key &y)
    {
        return (x.k != y.k);
    }
    friend int operator==(const merge_key &x, const merge_key &y)
    {
        return (x.k == y.k);
    }
    friend merge_key operator+(const merge_key &x, const merge_key &y UNUSED)
    {
        assert(0);
        return x;
        //  Key sum = x.k + y.k;
        //  merge_key f(sum, x.str_id);
        //  return f;
    }
};
 
/*****************************************************************
 *****************************************************************
 *****************************************************************
 
 external_memory buffer
 
 Each level-i buffer can store up to <arity>^i * <basesize> items,
 where typically <arity> is \theta(m) and <basesize> is \theta(M);
 therefore log_m{n/m} buffers are needed to store N items, one
 buffer for each level 1..log_m{n/m}. All buffers must have same
 values or <arity> and <basesize>.
 
 Functionality:
 
 A level-i on-disk buffer stores <arity>^i * <basesize> items of
 data, organized in <arity> streams of <arity>^{i-1} items each;
 <basesize> is same for all buffers and equal to the size of the
 level 0 buffer (in memory buffer).
 
 Invariant: all the <arity> streams of a level-i buffer are in
 sorted order; in this way sorting the buffer is done by merging the
 <arity> streams in linear time.
 
 Items are inserted in level i-buffer only a whole stream at a time
 (<arity>^{i-1}*<basesize> items). When all the <arity> streams of
 the buffer are full, the buffer is sorted and emptied into a stream
 of a level (i+1)-buffer.
 
 The <arity> streams of a buffer are allocated contigously from left
 to r ight. The unused streams are NULL; The buffer keeps the index of
 the last used(non-NULL) stream. When a buffer becomes full and is
 empty, all its buffers are set to NULL.
 
 *****************************************************************
 *****************************************************************
 ***************************************************************** */
 
/* T is a type with priority of type K and method getPriority() */
template <class T, class Key>
class em_buffer {
private:
    // number of streams in a buffer;
    unsigned int arity;
 
    // level of buffer: between 1 and log_arity{n/arity}; (level-0 buffer
    // has a slightly different behaviour so it is implemented as a
    // different class <im_buffer>)
    unsigned short level;
 
    // level-i buffer contains m streams of data, each of size
    // arity^{i-1}*basesize;
    AMI_STREAM<T> **data;
 
    // the buffers can be depleted to fill the internal pq;
    // keep an array which counts, for each stream, how many elements
    // have been deleted (implicitly from the beginning of stream)
    long *deleted;
 
    // nb of items in each substream; this can be found out by calling
    // stream_len() on the stream, but it is more costly esp in the case
    // when streams are on disk and must be moved in and out just to find
    // stream length; streamsize is set only at stream creation, and the
    // actual size must subtract the number of iteme deleted from the
    // bos
    unsigned long *streamsize;
 
    // index of the next available(empty) stream (out of the total m
    // streams in the buffer);
    unsigned int index;
 
    // nb of items in a stream of level_1 buffer
    unsigned long basesize;
 
public:
    // create a level-i buffer of given basesize;
    em_buffer(const unsigned short i, const unsigned long bs,
              const unsigned int ar);
 
    // copy constructor;
    em_buffer(const em_buffer &buf);
 
    // free the stream array and the streams pointers
    ~em_buffer();
 
    // return the level of the buffer;
    unsigned short get_level() const { return level; }
 
    // return the ith stream (load stream in memory)
    AMI_STREAM<T> *get_stream(unsigned int i);
 
    // return a pointer to the streams of the buffer (loads streams in
    // memory)
    AMI_STREAM<T> **get_streams();
 
    // put the ith stream back to disk
    void put_stream(unsigned int i);
 
    // called in pair with get_streams to put all streams back to disk
    void put_streams();
 
    // return a pointer to the array of deletion count for each stream
    long *get_bos() const { return deleted; }
 
    // return the index of the last stream in buffer which contains data;
    unsigned int laststream() const { return index - 1; }
 
    // return the index of the next available stream in the buffer
    unsigned int nextstream() const { return index; }
 
    // increment the index of the next available stream in the buffer
    void incr_nextstream() { ++index; }
 
    // return nb of (non-empty) streams in buffer
    unsigned int get_nbstreams() const { return index; }
 
    // return arity
    unsigned int get_arity() const { return arity; }
 
    // return total nb of deleted elements in all active streams of the buffer
    long total_deleted() const
    {
        long tot = 0;
        for (unsigned int i = 0; i < index; i++) {
            tot += deleted[i];
        }
        return tot;
    }
 
    // mark as deleted one more element from i'th stream
    void incr_deleted(unsigned int i)
    {
        assert(i < index);
        deleted[i]++;
    }
 
    // return the nominal size of a stream (nb of items):
    // arity^{level-1}*basesize;
    unsigned long get_stream_maxlen() const
    {
        return (unsigned long)pow((double)arity, (double)level - 1) * basesize;
    }
 
    // return the actual size of stream i; i must be the index of a valid
    // stream
    unsigned long get_stream_len(unsigned int i)
    {
        // assert(i>= 0 && i<index);
        return streamsize[i] - deleted[i];
    }
 
    // return the total current size of the buffer; account for the
    // deleted elements;
    unsigned long get_buf_len()
    {
        unsigned long tot = 0;
        for (unsigned int i = 0; i < index; i++) {
            tot += get_stream_len(i);
        }
        return tot;
    }
 
    // return the total maximal capacity of the buffer
    unsigned long get_buf_maxlen() { return arity * get_stream_maxlen(); }
 
    // return true if buffer is empty (all streams are empty)
    bool is_empty() { return ((nextstream() == 0) || (get_buf_len() == 0)); }
 
    // return true if buffer is full(all streams are full)
    bool is_full() const { return (nextstream() == arity); }
 
    // reset
    void reset();
 
    // clean buffer: in case some streams have been emptied by deletion
    // delete them and shift streams left;
    void cleanup();
 
    // create and return a stream which contains all elements of all
    // streams of the buffer in sorted ascending order of their
    // keys(priorities);
    AMI_STREAM<T> *sort();
 
    // insert an array into the buffer; can only insert one
    // level-i-full-stream-len nb of items at a time; assume the length
    // of the array is precisely the streamlen of level-i buffer n =
    // (pow(arity,level-1)*basesize); assume array is sorted; return the
    // number of items actually inserted
    long insert(T *a, long n);
 
    // insert a stream into the buffer; assume the length of the stream
    // is precisely the streamlen of level-i buffer n =
    // (pow(arity,level-1)*basesize); the <nextstream> pointer of buffer
    // is set to point to the argument stream; (in this way no stream
    // copying is done, just one pointer copy). The user should be aware
    // the the argument stream is 'lost' - that is a stream cannot be
    // inserted repeatedly into many buffers because this would lead to
    // several buffers pointing to the same stream.
 
    // stream is assumed sorted; bos = how many elements are deleted
    // from the beginning of stream;
 
    // return the number of items actually inserted
    long insert(AMI_STREAM<T> *str, long bos = 0);
 
    // print range of elements in buffer
    void print_range();
 
    // print all elements in buffer
    void print();
 
    // prints the sizes of the streams in the buffer
    void print_stream_sizes();
 
    // print the elements in the buffer
    friend ostream &operator<<(ostream &s, em_buffer &b)
    {
        s << "BUFFER_" << b.level << ": ";
        if (b.index == 0) {
            s << "[]";
        }
        s << "\n";
        b.get_streams();
        for (unsigned int i = 0; i < b.index; i++) {
            b.print_stream(s, i);
        }
        b.put_streams();
        return s;
    }
 
private:
    // merge the input streams; there are <arity> streams in total;
    // write output in <outstream>; the input streams are assumed sorted
    // in increasing order of their keys;
    AMI_err substream_merge(AMI_STREAM<T> **instreams, unsigned int arity,
                            AMI_STREAM<T> *outstream);
 
    // print to stream the elements in i'th stream
    void print_stream(ostream &s, unsigned int i);
 
#ifdef SAVE_MEMORY
    // array of names of streams;
    char **name;
 
    // return the designated name for stream i
    char *get_stream_name(unsigned int i) const;
 
    // print all stream names in buffer
    void print_stream_names();
 
    // checks that name[i] is the same as stream name; stream i must be in
    // memory (by a previous get_stream call, for instance) in order to
    // find its length
    void check_name(unsigned int i);
#endif
};
 
/************************************************************/
// create a level-i buffer of given basesize;
template <class T, class Key>
em_buffer<T, Key>::em_buffer(const unsigned short i, const unsigned long bs,
                             const unsigned int ar)
    : arity(ar), level(i), basesize(bs)
{
 
    assert(level >= 1);
 
    char str[100];
    snprintf(str, sizeof(str),
             "em_buffer: allocate %d AMI_STREAM*, total %ld\n", arity,
             (long)(arity * sizeof(AMI_STREAM<T> *)));
    MEMORY_LOG(str);
    // allocate STREAM* array
    data = new AMI_STREAM<T> *[arity];
 
    // allocate deleted array
    snprintf(str, sizeof(str), "em_buffer: allocate deleted array: %ld\n",
             (long)(arity * sizeof(long)));
    MEMORY_LOG(str);
    deleted = new long[arity];
 
    // allocate streamsize array
    snprintf(str, sizeof(str), "em_buffer: allocate streamsize array: %ld\n",
             (long)(arity * sizeof(long)));
    MEMORY_LOG(str);
    streamsize = new unsigned long[arity];
 
#ifdef SAVE_MEMORY
    // allocate name array
    snprintf(str, sizeof(str), "em_buffer: allocate name array: %ld\n",
             (long)(arity * sizeof(char *)));
    MEMORY_LOG(str);
    name = new char *[arity];
    assert(name);
#endif
 
    // assert data
    if ((!data) || (!deleted) || (!streamsize)) {
        cerr << "em_buffer: cannot allocate\n";
        exit(1);
    }
 
    // initialize the <arity> streams to NULL, deleted[], streamsize[]
    // and name[]
    for (unsigned int ui = 0; ui < arity; ui++) {
        data[ui] = NULL;
        deleted[ui] = 0;
        streamsize[ui] = 0;
#ifdef SAVE_MEMORY
        name[ui] = NULL;
#endif
    }
    // set index
    index = 0;
 
#ifdef SAVE_MEMORY
    // streams_in_memory = false;
#endif
}
 
/************************************************************/
// copy constructor;
template <class T, class Key>
em_buffer<T, Key>::em_buffer(const em_buffer &buf)
    : level(buf.level), basesize(buf.basesize), index(buf.index),
      arity(buf.arity)
{
 
    assert(0); // should not get called
 
    MEMORY_LOG("em_buffer: copy constr start\n");
    get_streams();
    for (unsigned int i = 0; i < index; i++) {
        assert(data[i]);
        delete data[i]; // delete old stream if existing
        data[i] = NULL;
 
        // call copy constructor; i'm not sure that it actually duplicates
        // the stream and copies the data; should that in the BTE
        // sometimes..
        data[i] = new AMI_STREAM<T>(*buf.data[i]);
        deleted[i] = buf.deleted[i];
        streamsize[i] = buf.streamsize[i];
#ifdef SAVE_MEMORY
        assert(name[i]);
        delete name[i];
        name[i] = NULL;
        name[i] = buf.name[i];
#endif
    }
    put_streams();
    MEMORY_LOG("em_buffer: copy constr end\n");
}
 
/************************************************************/
// free the stream array and the streams pointers
template <class T, class Key>
em_buffer<T, Key>::~em_buffer()
{
 
    assert(data);
    // delete the m streams in the buffer
    get_streams();
    for (unsigned int i = 0; i < index; i++) {
        assert(data[i]);
#ifdef SAVE_MEMORY
        check_name(i);
        delete name[i];
#endif
        delete data[i];
        data[i] = NULL;
    }
 
    delete[] data;
    delete[] deleted;
    delete[] streamsize;
#ifdef SAVE_MEMORY
    delete[] name;
#endif
}
 
#ifdef SAVE_MEMORY
/************************************************************/
// checks that name[i] is the same as stream name; stream i must be in
// memory (by a previous get_stream call, for instance) in order to
// find its length
template <class T, class Key>
void em_buffer<T, Key>::check_name(unsigned int i UNUSED)
{
 
#ifdef EMBUF_CHECK_NAME
    assert(i >= 0 && i < index);
    assert(data[i]);
 
    char *fooname;
    data[i]->name(&fooname); // name() allocates the string
#ifdef EMBUF_CHECK_NAME_PRINT
    cout << "::check_name: checking stream [" << level << "," << i
         << "] name:" << fooname << endl;
    cout.flush();
#endif
    if (strcmp(name[i], fooname) != 0) {
        cerr << "name[" << i << "]=" << name[i] << ", streamname=" << fooname
             << endl;
    }
    assert(strcmp(fooname, name[i]) == 0);
    delete fooname;
#endif
}
#endif
 
/************************************************************/
// if SAVE_MEMORY flag is set, load the stream in memory; return the
// ith stream
template <class T, class Key>
AMI_STREAM<T> *em_buffer<T, Key>::get_stream(unsigned int i)
{
 
    assert(i < index);
 
#ifdef SAVE_MEMORY
    MY_LOG_DEBUG_ID("em_buffer::get_stream");
    MY_LOG_DEBUG_ID(i);
 
    if (data[i] == NULL) {
 
        // stream is on disk, load it in memory
        assert(name[i]);
        MY_LOG_DEBUG_ID("load stream in memory");
        MY_LOG_DEBUG_ID(name[i]);
 
#ifdef EMBUF_PRINT_GETPUT_STREAM
        cout << "get_stream:: name[" << i << "]=" << name[i] << " from disk\n";
        cout.flush();
#endif
 
        // assert that file exists
        FILE *fp;
        if ((fp = fopen(name[i], "rb")) == NULL) {
            cerr << "get_stream: checking that stream " << name[i]
                 << "exists\n";
            perror(name[i]);
            assert(0);
            exit(1);
        }
        fclose(fp);
 
        // create an AMI_STREAM from file
        data[i] = new AMI_STREAM<T>(name[i]);
        assert(data[i]);
    }
    else {
 
        // if data[i] not NULL, stream must be already in memory
        MY_LOG_DEBUG_ID("stream not NULL");
        MY_LOG_DEBUG_ID(data[i]->sprint());
    }
#endif
 
    // NOW STREAM IS IN MEMORY
 
    // some assertion checks
    assert(data[i]);
    assert(data[i]->stream_len() == streamsize[i]);
#ifdef SAVE_MEMORY
    check_name(i);
#endif
 
    return data[i];
}
 
/************************************************************/
// if SAVE_MEMORY flag is set, put the i'th stream back on disk
template <class T, class Key>
void em_buffer<T, Key>::put_stream(unsigned int i)
{
 
    assert(i < index);
 
#ifdef SAVE_MEMORY
    MY_LOG_DEBUG_ID("em_buffer::put_stream");
    MY_LOG_DEBUG_ID(i);
 
    if (data[i] != NULL) {
 
        // stream is in memory, put it on disk
        MY_LOG_DEBUG_ID("stream put to disk");
        MY_LOG_DEBUG_ID(data[i]->sprint());
 
        check_name(i);
#ifdef EMBUF_PRINT_GETPUT_STREAM
        cout << "put_stream:: name[" << i << "]=" << name[i] << " to disk\n";
        cout.flush();
#endif
 
        // make stream persistent and delete it
        data[i]->persist(PERSIST_PERSISTENT);
        delete data[i];
        data[i] = NULL;
    }
    else {
 
        // data[i] is NULL, so stream must be already put on disk
        MY_LOG_DEBUG_ID("stream is NULL");
    }
#endif
}
 
/************************************************************/
// return a pointer to the streams of the buffer
template <class T, class Key>
AMI_STREAM<T> **em_buffer<T, Key>::get_streams()
{
 
#ifdef SAVE_MEMORY
    MY_LOG_DEBUG_ID("em_buffer::get_streams: reading streams from disk");
#ifdef EMBUF_PRINT_GETPUT_STREAMS
    cout << "em_buffer::get_streams (buffer " << level << ")";
    cout << ": index = " << index << "(arity=" << arity << ")\n";
    cout.flush();
#endif
 
    for (unsigned int i = 0; i < index; i++) {
        get_stream(i);
        assert(data[i]);
    }
 
#endif
 
    return data;
}
 
/************************************************************/
// called in pair with load_streams to store streams on disk
// and release the memory
template <class T, class Key>
void em_buffer<T, Key>::put_streams()
{
 
#ifdef SAVE_MEMORY
    MY_LOG_DEBUG_ID("em_buffer::put_streams: writing streams on disk");
#ifdef EMBUF_PRINT_GETPUT_STREAMS
    cout << "em_buffer::put_streams (buffer " << level << ")";
    cout << ": index = " << index << "(arity=" << arity << ")\n";
    cout.flush();
#endif
 
    for (unsigned int i = 0; i < index; i++) {
        put_stream(i);
        assert(data[i] == NULL);
    }
#endif
}
 
#ifdef SAVE_MEMORY
/************************************************************/
// return the name of the ith stream
template <class T, class Key>
char *em_buffer<T, Key>::get_stream_name(unsigned int i) const
{
 
    assert(i >= 0 && i < index);
    assert(name[i]);
    return name[i];
}
#endif
 
#ifdef SAVE_MEMORY
/************************************************************/
template <class T, class Key>
void em_buffer<T, Key>::print_stream_names()
{
    unsigned int i;
    for (i = 0; i < index; i++) {
        assert(name[i]);
        cout << "stream " << i << ": " << name[i] << endl;
    }
    cout.flush();
}
#endif
 
/************************************************************/
// clean buffer in case some streams have been emptied by deletion
template <class T, class Key>
void em_buffer<T, Key>::cleanup()
{
 
    MY_LOG_DEBUG_ID("em_buffer::cleanup()");
#ifdef EMBUF_CLEANUP_PRINT
#ifdef SAVE_MEMORY
    if (index > 0) {
        cout << "before cleanup:\n";
        print_stream_names();
        print_stream_sizes();
        cout.flush();
    }
#endif
#endif
 
    // load all streams in memory
    get_streams();
 
    // count streams of size=0
    unsigned int i, empty = 0;
    for (i = 0; i < index; i++) {
 
        if (get_stream_len(i) == 0) {
            // printing..
#ifdef EMBUF_DELETE_STREAM_PRINT
            cout << "deleting stream [" << level << "," << i << "]:";
#ifdef SAVE_MEMORY
            cout << name[i];
#endif
            cout << endl;
            cout.flush();
#endif
 
#ifdef SAVE_MEMORY
            // stream is empty ==> delete its name
            assert(name[i]);
            delete name[i];
            name[i] = NULL;
#endif
 
            // stream is empty ==> reset data
            assert(data[i]);
            // data[i]->persist(PERSIST_DELETE); //this is done automatically..
            delete data[i];
            data[i] = NULL;
            deleted[i] = 0;
            streamsize[i] = 0;
            empty++;
        }
    }
    // streams are in memory; all streams which are NULL must have been
    // deleted
 
    // shift streams to the left in case holes were introduced
    unsigned int j = 0;
    if (empty) {
#ifdef EMBUF_DELETE_STREAM_PRINT
        cout << "em_buffer::cleanup: shifting streams\n";
        cout.flush();
#endif
        for (i = 0; i < index; i++) {
            // if i'th stream is not empty, shift it left if necessary
            if (data[i]) {
                if (i != j) {
                    // set j'th stream to point to i'th stream
                    // cout << j << " set to " << i << endl; cout.flush();
                    data[j] = data[i];
                    deleted[j] = deleted[i];
                    streamsize[j] = streamsize[i];
                    // set i'th stream to point to NULL
                    data[i] = NULL;
                    deleted[i] = 0;
                    streamsize[i] = 0;
#ifdef SAVE_MEMORY
                    // fix the names
                    /*         already done          assert(name[j]); */
                    /*                   delete name[j]; */
                    name[j] = name[i];
                    name[i] = NULL;
                    check_name(j);
#endif
                }
                else {
                    // cout << i << " left the same" << endl;
                }
                j++;
            } // if data[i] != NULL
        } // for i
 
        // set the index
        assert(index == j + empty);
        index = j;
 
#ifdef EMBUF_DELETE_STREAM_PRINT
        cout << "em_buffer::cleanup: index set to " << index << endl;
        cout.flush();
#endif
    } // if empty
 
    // put streams back to disk
    put_streams();
 
#ifdef EMBUF_CLEANUP_PRINT
#ifdef SAVE_MEMORY
    if (index > 0) {
        cout << "after cleanup:\n";
        print_stream_names();
        print_stream_sizes();
        cout.flush();
    }
#endif
#endif
}
 
/************************************************************/
// delete all streams
template <class T, class Key>
void em_buffer<T, Key>::reset()
{
 
    get_streams();
 
    // make streams not-persistent and delete them
    for (unsigned int i = 0; i < index; i++) {
        assert(data[i]);
        assert(streamsize[i] == data[i]->stream_len());
#ifdef SAVE_MEMORY
        check_name(i);
        assert(name[i]);
        delete name[i];
        name[i] = NULL;
#endif
 
        data[i]->persist(PERSIST_DELETE);
 
        delete data[i];
        data[i] = NULL;
        deleted[i] = 0;
        streamsize[i] = 0;
    }
 
    index = 0;
}
 
/************************************************************/
// create and return a stream which contains all elements of
// all streams of the buffer in sorted ascending order of
// their keys  (priorities);
template <class T, class Key>
AMI_STREAM<T> *em_buffer<T, Key>::sort()
{
 
    // create stream
    MEMORY_LOG("em_buffer::sort: allocate new AMI_STREAM\n");
 
    AMI_STREAM<T> *sorted_stream =
        new AMI_STREAM<T>(); /* will be deleted in insert() */
    assert(sorted_stream);
 
    // merge the streams into sorted stream
    AMI_err aerr;
    // Key dummykey;
    // must modify this to seek after deleted[i] elements!!!!!!!!!!!!!
    //  aerr = MIAMI_single_merge_Key(data, arity, sorted_stream,
    //                                    0, dummykey);
    // could not use AMI_merge so i had to write my own..
 
    get_streams();
 
    aerr = substream_merge(data, arity, sorted_stream);
    assert(aerr == AMI_ERROR_NO_ERROR);
 
    put_streams();
 
    return sorted_stream;
}
 
/************************************************************/
/* merge the input streams; there are <arity> streams in total; write
   output in <outstream>;
 
   the input streams are assumed sorted in increasing order of their
   keys;
 
   assumes the instreams are in memory (no need for get_streams()) */
template <class T, class Key>
AMI_err em_buffer<T, Key>::substream_merge(AMI_STREAM<T> **instreams,
                                           unsigned int arity,
                                           AMI_STREAM<T> *outstream)
{
 
    unsigned int i, j;
 
    // some assertion checks
    assert(instreams);
    assert(outstream);
    for (i = 0; i < arity; i++) {
        assert(instreams[i]);
#ifdef SAVE_MEMORY
        check_name(i);
#endif
    }
 
    std::vector<T *> in_objects(
        arity); // pointers to current leading elements of streams
    AMI_err ami_err;
 
    char str[200];
    snprintf(str, sizeof(str),
             "em_buffer::substream_merge: allocate keys array, total %ldB\n",
             (long)((long)arity * sizeof(merge_key<Key>)));
    MEMORY_LOG(str);
 
    // keys array is initialized with smallest key from each stream (only
    // non-null keys must be included)
    merge_key<Key> *keys;
    // merge_key<Key>* keys = new (merge_key<Key>)[arity];
    typedef merge_key<Key> footype;
    keys = new footype[arity];
    assert(keys);
 
    // count number of non-empty streams
    j = 0;
    // rewind and read the first item from every stream initializing
    // in_objects and keys
    for (i = 0; i < arity; i++) {
        assert(instreams[i]);
        // rewind stream
        if ((ami_err = instreams[i]->seek(deleted[i])) != AMI_ERROR_NO_ERROR) {
            return ami_err;
        }
        // read first item from stream
        if ((ami_err = instreams[i]->read_item(&(in_objects[i]))) !=
            AMI_ERROR_NO_ERROR) {
            if (ami_err == AMI_ERROR_END_OF_STREAM) {
                in_objects[i] = NULL;
            }
            else {
                return ami_err;
            }
        }
        else {
            // include this key in the array of keys
            Key k = in_objects[i]->getPriority();
            keys[j].set(k, i);
            j++;
        }
    }
    unsigned int NonEmptyRuns = j;
 
    // build heap from the array of keys
    pqheap_t1<merge_key<Key>> mergeheap(keys, NonEmptyRuns);
 
    // repeatedly extract_min from heap, write it to output stream and
    // insert next element from same stream
    merge_key<Key> minelt;
    // rewind output buffer
    ami_err = outstream->seek(0);
    assert(ami_err == AMI_ERROR_NO_ERROR);
    while (!mergeheap.empty()) {
        // find min key and id of the stream from whereit comes
        mergeheap.min(minelt);
        i = minelt.stream_id();
        // write min item to output stream
        if ((ami_err = outstream->write_item(*in_objects[i])) !=
            AMI_ERROR_NO_ERROR) {
            return ami_err;
        }
        // read next item from same input stream
        if ((ami_err = instreams[i]->read_item(&(in_objects[i]))) !=
            AMI_ERROR_NO_ERROR) {
            if (ami_err != AMI_ERROR_END_OF_STREAM) {
                return ami_err;
            }
        }
        // extract the min from the heap and insert next key from same stream
        if (ami_err == AMI_ERROR_END_OF_STREAM) {
            mergeheap.delete_min();
        }
        else {
            Key k = in_objects[i]->getPriority();
            merge_key<Key> nextit(k, i);
            mergeheap.delete_min_and_insert(nextit);
        }
    } // while
 
    // delete [] keys;
    //!!! KEYS BELONGS NOW TO MERGEHEAP, AND WILL BE DELETED BY THE
    // DESTRUCTOR OF MERGEHEAP (CALLED AUUTOMATICALLY ON FUNCTION EXIT) IF
    // I DELETE KEYS EXPLICITLY, THEY WILL BE DELETED AGAIN BY DESTRUCTOR,
    // AND EVERYTHING SCREWS UP..
 
    return AMI_ERROR_NO_ERROR;
}
 
/************************************************************/
// insert an array into the buffer; assume array is sorted; return the
// number of items actually inserted; if SAVE_MEMORY FLAG is on, put
// stream on disk and release its memory
template <class T, class Key>
long em_buffer<T, Key>::insert(T *a, long n)
{
 
    assert(a);
 
    if (is_full()) {
        cout << "em_buffer::insert: buffer full\n";
        return 0;
    }
 
    // can only insert one full stream at a time
    // relaxed..
    // assert(n == get_stream_maxlen());
 
    // create the stream
    MEMORY_LOG("em_buffer::insert(from array): allocate AMI_STREAM\n");
    AMI_STREAM<T> *str = new AMI_STREAM<T>();
    assert(str);
 
    // write the array to stream
    AMI_err ae;
    for (long i = 0; i < n; i++) {
        ae = str->write_item(a[i]);
        assert(ae == AMI_ERROR_NO_ERROR);
    }
    assert(n == str->stream_len());
 
    // insert the stream in the buffer
    return insert(str);
}
 
/************************************************************/
/* insert a stream into the buffer; the next free entry in the buffer
   is set to point to the stream; if SAVE_MEMORY flag is on, the
   stream is put to disk;
 
   the <nextstream> pointer of buffer is set to point to the argument
   stream; (in this way no stream copying is done, just one pointer
   copy). The user should be aware the the argument stream is 'lost' -
   that is a stream cannot be inserted repeatedly into many buffers
   because this would lead to several buffers pointing to the same
   stream.
 
   stream is assume stream is sorted; bos = how many elements must be
   skipped (were deleted) from the beginning of stream;
 
   return the number of items actually inserted */
template <class T, class Key>
long em_buffer<T, Key>::insert(AMI_STREAM<T> *str, long bos)
{
 
    assert(str);
 
    if (is_full()) {
        cout << "em_buffer::insert: buffer full\n";
        return 0;
    }
 
    // can only insert one level-i-full-stream at a time;
    // relaxed..can specify bos;
    // not only that, but the length of the stream can be smaller
    // than nominal length, because a stream is normally obtained by
    // merging streams which can be shorter;
    // assert(str->stream_len() == get_stream_len() - bos);
 
#ifdef EMBUF_CHECK_INSERT
    // check that stream is sorted
    cout << "CHECK_INSERT: checking stream is sorted\n";
    AMI_err ae;
    str->seek(0);
    T *crt = NULL, *prev = NULL;
    while (str->read_item(&crt)) {
        assert(ae == AMI_ERROR_NO_ERROR);
        if (prev)
            assert(*prev <= *crt);
    }
#endif
 
    // nextstream must be empty
    assert(str);
    assert(data[nextstream()] == NULL);
    assert(deleted[nextstream()] == 0);
    assert(streamsize[nextstream()] == 0);
#ifdef SAVE_MEMORY
    assert(name[nextstream()] == NULL);
#endif
 
    // set next entry i the buffer to point to this stream
    data[nextstream()] = str;
    deleted[nextstream()] = bos;
    streamsize[nextstream()] = str->stream_len();
#ifdef SAVE_MEMORY
    // set next name entry in buffer to point to this stream's name
    char *s;
    str->name(&s); // name() allocates the string
    name[nextstream()] = s;
 
    // put stream on disk and release its memory
    str->persist(PERSIST_PERSISTENT);
    delete str; // stream should be persistent; just delete it
    data[nextstream()] = NULL;
 
#ifdef EMBUF_PRINT_INSERT
    cout << "insert stream " << s << " at buf [" << level << "," << nextstream()
         << "]" << endl;
#endif
#endif
 
    // increment the index of next available stream in buffer
    incr_nextstream();
 
#ifdef EMBUF_PRINT_INSERT
    print_stream_sizes();
    print_stream_names();
#endif
 
#ifdef SAVE_MEMORY
    MY_LOG_DEBUG_ID("em_buffer::insert(): inserted stream ");
    MY_LOG_DEBUG_ID(name[nextstream() - 1]);
#endif
 
    // return nb of items inserted
    return get_stream_len(nextstream() - 1);
}
 
/************************************************************/
// print the elements of the i'th stream of the buffer to a stream;
// assumes stream is in memory;
template <class T, class Key>
void em_buffer<T, Key>::print_stream(ostream &s, unsigned int i)
{
 
    assert(data[i]);
    assert((i >= 0) && (i < index));
 
    AMI_err ae;
    T *x;
 
    s << "STREAM " << i << ": [";
 
    ae = data[i]->seek(deleted[i]);
    assert(ae == AMI_ERROR_NO_ERROR);
 
    for (long j = 0; j < get_stream_len(i); j++) {
        ae = data[i]->read_item(&x);
        assert(ae == AMI_ERROR_NO_ERROR);
        s << *x << ",";
    }
    s << "]\n";
}
 
/************************************************************/
// print elements range in buffer (read first and last element in each
// substream and find global min and max)
template <class T, class Key>
void em_buffer<T, Key>::print_range()
{
 
    T *min, *max;
    AMI_err ae;
 
    get_streams();
 
    for (unsigned int i = 0; i < index; i++) {
        cout << "[";
        // read min element in substream i
        ae = data[i]->seek(deleted[i]);
        assert(ae == AMI_ERROR_NO_ERROR);
        ae = data[i]->read_item(&min);
        assert(ae == AMI_ERROR_NO_ERROR);
        cout << min->getPriority() << "..";
        // read max element in substream i
        ae = data[i]->seek(streamsize[i] - 1);
        assert(ae == AMI_ERROR_NO_ERROR);
        ae = data[i]->read_item(&max);
        assert(ae == AMI_ERROR_NO_ERROR);
        cout << max->getPriority() << " (sz=" << get_stream_len(i) << ")] ";
    }
    for (unsigned int i = index; i < arity; i++) {
        cout << "[] ";
    }
 
    put_streams();
}
 
/************************************************************/
// print all elements in buffer
template <class T, class Key>
void em_buffer<T, Key>::print()
{
 
    T *x;
    AMI_err ae;
 
    get_streams();
 
    for (unsigned int i = 0; i < index; i++) {
        cout << "    [";
        ae = data[i]->seek(deleted[i]);
        assert(ae == AMI_ERROR_NO_ERROR);
        for (unsigned long j = 0; j < get_stream_len(i); j++) {
            ae = data[i]->read_item(&x);
            assert(ae == AMI_ERROR_NO_ERROR);
            cout << x->getPriority() << ",";
        }
        cout << "]" << endl;
    }
    for (unsigned int i = index; i < arity; i++) {
        cout << "[] ";
    }
 
    put_streams();
}
 
/************************************************************/
// print the sizes of the substreams in the buffer
template <class T, class Key>
void em_buffer<T, Key>::print_stream_sizes()
{
 
    cout << "(streams=" << index << ") sizes=[";
    for (unsigned int i = 0; i < arity; i++) {
        cout << get_stream_len(i) << ",";
    }
    cout << "]" << endl;
    cout.flush();
}
 
#endif