ami_stream.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 _AMI_STREAM_H
#define _AMI_STREAM_H

#include <grass/config.h>

#include <sys/types.h>
#include <sys/stat.h>
#include <stdio.h>
#include <stdlib.h>
#include <assert.h>
#include <fcntl.h>
#include <errno.h>
#include <unistd.h>

#include <cstring>
#include <iostream>
using std::cout;
using std::cerr;
using std::endl;
using std::ostream;
using std::ofstream;
using std::istream;

extern "C" {
#include <grass/gis.h>
}

#define MAX_STREAMS_OPEN 200

#include "mm.h" // Get the memory manager.

#define DEBUG_DELETE if(0)
#define DEBUG_STREAM_LEN if(0)
#define DEBUG_ASSERT if(0)

// The name of the environment variable which keeps the name of the
// directory where streams are stored
#define STREAM_TMPDIR "STREAM_DIR"

// All streams will be names STREAM_*****
#define BASE_NAME "STREAM"

#define STREAM_BUFFER_SIZE (1<<18)


//
// AMI error codes are returned using the AMI_err type.
//
enum AMI_err {
  AMI_ERROR_NO_ERROR = 0,
  AMI_ERROR_IO_ERROR,
  AMI_ERROR_END_OF_STREAM,
  AMI_ERROR_OUT_OF_RANGE,
  AMI_ERROR_READ_ONLY,
  AMI_ERROR_OS_ERROR,
  AMI_ERROR_MM_ERROR,
  AMI_ERROR_OBJECT_INITIALIZATION,
  AMI_ERROR_PERMISSION_DENIED,
  AMI_ERROR_INSUFFICIENT_MAIN_MEMORY,
  AMI_ERROR_INSUFFICIENT_AVAILABLE_STREAMS,
  AMI_ERROR_ENV_UNDEFINED,
  AMI_ERROR_NO_MAIN_MEMORY_OPERATION,
};

extern const char *ami_str_error[];

//
// AMI stream types passed to constructors
//
enum AMI_stream_type {
    AMI_READ_STREAM = 1,        // Open existing stream for reading
    AMI_WRITE_STREAM,           // Open for writing.  Create if non-existent
    AMI_APPEND_STREAM,          // Open for writing at end.  Create if needed.
    AMI_READ_WRITE_STREAM,      // Open to read and write.
    AMI_APPEND_WRITE_STREAM     // Open for writing at end (write only mode).
};




enum persistence {
    // Delete the stream from the disk when it is destructed.
  PERSIST_DELETE = 0,
  // Do not delete the stream from the disk when it is destructed.
  PERSIST_PERSISTENT,
  // Delete each block of data from the disk as it is read.
  PERSIST_READ_ONCE
};

/* an un-templated version makes for easier debugging */
class UntypedStream {
protected:
  FILE * fp;
  int fildes;   //descriptor of file
  AMI_stream_type  access_mode;
  char path[BUFSIZ];
  persistence per;

  //0 for streams, positive for substreams
  unsigned int substream_level;
  
  // If this stream is actually a substream, these will be set to
  // indicate the portion of the file that is part of this stream.  If
  // the stream is the whole file, they will be set to -1. Both are in
  // T units.
  off_t logical_bos;
  off_t logical_eos;

  //stream buffer passed in the call to setvbuf when file is opened
  char* buf;
  int eof_reached;

 public:
  static unsigned int get_block_length()  {
    return STREAM_BUFFER_SIZE;
    //return getpagesize();
  };

};

template<class T> 
class AMI_STREAM : public UntypedStream {

protected:

  T read_tmp;                                   /* this is ugly... RW */

public:
  // An AMI_stream with default name
  AMI_STREAM();
  
  // An AMI stream based on a specific path name.
  AMI_STREAM(const char *path_name, AMI_stream_type st = AMI_READ_WRITE_STREAM);

  // convenience function with split path_name
  //AMI_STREAM(const char *dir_name, const char *file_name, AMI_stream_type st);
  
  
  // A psuedo-constructor for substreams.
  AMI_err new_substream(AMI_stream_type st, off_t sub_begin, off_t sub_end,
                        AMI_STREAM<T> **sub_stream);

  // Destructor
  ~AMI_STREAM(void);
  
  // Read and write elements.
  AMI_err read_item(T **elt);
  AMI_err write_item(const T &elt);
  AMI_err read_array(T *data, off_t len, off_t *lenp=NULL);
  AMI_err write_array(const T *data, off_t len);
  
  // Return the number of items in the stream.
  off_t stream_len(void);
  
  // Return the path name of this stream.
  AMI_err name(char **stream_name);
  const char* name() const;
  
  // Move to a specific item in the stream.
  AMI_err seek(off_t offset);

  // Query memory usage
  static AMI_err main_memory_usage(size_t *usage,
                            MM_stream_usage usage_type= MM_STREAM_USAGE_OVERHEAD);
  
  void persist(persistence p);
  
  char *sprint();

  // have we hit the end of the stream
  int eof();
};


/**********************************************************************/


/**********************************************************************/
/* creates a random file name, opens the file for reading and writing
   and and returns a file descriptor */
/* int ami_single_temp_name(char *base, char* tmp_path); */
/* fix from Andy Danner */
int ami_single_temp_name(const std::string& base, char* tmp_path); 


/**********************************************************************/
/* given fd=fide descriptor, associates with it a stream aopened in
   access_mode and returns it */
FILE*  open_stream(int fd, AMI_stream_type st);


/**********************************************************************/
/* open the file whose name is pathname in access mode */
FILE* open_stream(char* pathname, AMI_stream_type st);




/********************************************************************/
//  An  AMI stream with default name.
template<class T>
AMI_STREAM<T>::AMI_STREAM() {
  
  access_mode = AMI_READ_WRITE_STREAM;
  int fd = ami_single_temp_name(BASE_NAME, path);
  fildes = fd;
  fp = open_stream(fd, access_mode);
  
  /* a stream is by default buffered with a buffer of size BUFSIZ=1K */
  buf = new char[STREAM_BUFFER_SIZE];
  if (setvbuf(fp, buf, _IOFBF, STREAM_BUFFER_SIZE) != 0) {
    cerr << "ERROR: setvbuf failed (stream " << path << ") with: "
         << strerror(errno) << endl;
    exit(1);
  }
  
  // By default, all streams are deleted at destruction time.
  per = PERSIST_DELETE;

   // Not a substream.
  substream_level  = 0;
  logical_bos = logical_eos = -1;

  // why is this here in the first place?? -RW
  seek(0);

  eof_reached = 0;

  // Register memory usage before returning.
  //size_t usage; 
  //main_memory_usage(&usage,  MM_STREAM_USAGE_CURRENT);
  //MM_manager.register_allocation(usage);
}



/**********************************************************************/
// An AMI stream based on a specific path name.
template<class T>
AMI_STREAM<T>::AMI_STREAM(const char *path_name, AMI_stream_type st) {

  access_mode = st;

  if(path_name == NULL) {
        int fd = ami_single_temp_name(BASE_NAME, path);
        fildes = fd;
        fp = open_stream(fd, access_mode);
  } else {
        strcpy(path, path_name);
        fp = open_stream(path, st);  
        fildes = -1;
  }

  /* a stream is by default buffered with a buffer of size BUFSIZ=1K */
  buf = new char[STREAM_BUFFER_SIZE];
  if (setvbuf(fp, buf, _IOFBF, STREAM_BUFFER_SIZE) != 0) {
    cerr << "ERROR: setvbuf failed (stream " << path << ") with: "
         << strerror(errno) << endl;
    exit(1);
  }

  eof_reached = 0;

  // By default, all streams are deleted at destruction time.
  if(st == AMI_READ_STREAM) {
        per = PERSIST_PERSISTENT;
  } else {
        per = PERSIST_DELETE;
  }

  // Not a substream.
  substream_level  = 0;
  logical_bos = logical_eos = -1;
  
  seek(0);

  // Register memory usage before returning.
  //size_t usage; 
  //main_memory_usage(&usage,  MM_STREAM_USAGE_CURRENT);
  //MM_manager.register_allocation(usage);
}



/**********************************************************************/
 // A psuedo-constructor for substreams.
template<class T>
AMI_err AMI_STREAM<T>::new_substream(AMI_stream_type st,
                                     off_t sub_begin,
                                     off_t sub_end,
                                     AMI_STREAM<T> **sub_stream) {

  //assume this for now
  assert(st == AMI_READ_STREAM);

#ifdef __MINGW32__
  /* MINGW32: reopen file here for stream_len() below */
  //reopen the file 
  AMI_STREAM<T> *substr = new AMI_STREAM<T>(path, st);
#endif

  //check range
  if (substream_level) {
     if( (sub_begin >= (logical_eos - logical_bos)) ||
         (sub_end >= (logical_eos - logical_bos)) ) {
       
       return AMI_ERROR_OUT_OF_RANGE;
     }
  }  else {
    off_t len = stream_len();
    if (sub_begin > len || sub_end > len) {

      return AMI_ERROR_OUT_OF_RANGE;
    }
  }

#ifndef __MINGW32__
  //reopen the file 
  AMI_STREAM<T> *substr = new AMI_STREAM<T>(path, st);
#endif

  // Set up the beginning and end positions.
  if (substream_level) {
    substr->logical_bos = logical_bos + sub_begin;
    substr->logical_eos = logical_bos + sub_end + 1;
  } else {
    substr->logical_bos = sub_begin;
    substr->logical_eos = sub_end + 1;
  }
  
  // Set the current position.
  substr->seek(0);

  substr->eof_reached = 0;

  //set substream level
  substr->substream_level = substream_level + 1;

  substr->per = per;   //set persistence

  //*sub_stream = (AMI_base_stream < T > *)substr;
  *sub_stream = substr;
  return  AMI_ERROR_NO_ERROR;
}



/**********************************************************************/
// Return the number of items in the stream.
template<class T>
off_t AMI_STREAM<T>::stream_len(void) {

  fflush(fp);

#ifdef __MINGW32__
  //stat() fails on MS Windows if the file is open, so try G_ftell() instead.
  //FIXME: not 64bit safe, but WinGrass isn't either right now.
  off_t posn_save, st_size;

  posn_save = G_ftell(fp);
  if(posn_save == -1) {
     perror("ERROR: AMI_STREAM::stream_len(): ftell(fp) failed ");
     perror(path);
     exit(1);
  }

  G_fseek(fp, 0, SEEK_END);
  st_size = G_ftell(fp);
  if(st_size == -1) {
     perror("ERROR: AMI_STREAM::stream_len(): ftell[SEEK_END] failed ");
     perror(path);
     exit(1);
  }

  G_fseek(fp, posn_save, SEEK_SET);

  //debug stream_len:
  DEBUG_STREAM_LEN fprintf(stderr, "%s: length = %lld   sizeof(T)=%d\n",
          path, st_size, sizeof(T));

  return (st_size / sizeof(T));
#else
  struct stat statbuf;
  if (stat(path, &statbuf) == -1) {
    perror("AMI_STREAM::stream_len(): fstat failed ");
    DEBUG_ASSERT assert(0);
    exit(1);
  }

  //debug stream_len:
  DEBUG_STREAM_LEN fprintf(stderr, "%s: length = %lld   sizeof(T)=%lud\n",
          path, (long long int)statbuf.st_size, sizeof(T));

  return (statbuf.st_size / sizeof(T));
#endif
}



/**********************************************************************/
// Return the path name of this stream.
template<class T>
AMI_err AMI_STREAM<T>::name(char **stream_name)  {
  
  *stream_name = new char [strlen(path) + 1];
  strcpy(*stream_name, path);
  
  return AMI_ERROR_NO_ERROR;
}

// Return the path name of this stream.
template<class T>
const char *
AMI_STREAM<T>::name() const {
  return path;
}



/**********************************************************************/
// Move to a specific offset within the (sub)stream.
template<class T>
AMI_err AMI_STREAM<T>::seek(off_t offset) {

  off_t seek_offset;
  
  if (substream_level) {    //substream
    if (offset  > (unsigned) (logical_eos - logical_bos)) {
      //offset out of range
      cerr << "ERROR: AMI_STREAM::seek bos=" << logical_bos << ", eos="
           << logical_eos << ", offset " << offset << " out of range.\n";
      DEBUG_ASSERT assert(0);
      exit(1);
    } else {
      //offset in range 
      seek_offset = (logical_bos + offset) * sizeof(T);
    }


  } else {
    //not a substream
    seek_offset = offset * sizeof(T);
  }

  G_fseek(fp, seek_offset, SEEK_SET);
  
  return AMI_ERROR_NO_ERROR;
}




/**********************************************************************/
// Query memory usage
template<class T>
AMI_err
AMI_STREAM<T>::main_memory_usage(size_t *usage, MM_stream_usage usage_type) {
  
   switch (usage_type) {
   case MM_STREAM_USAGE_OVERHEAD:
     *usage = sizeof (AMI_STREAM<T>);
     break;
   case MM_STREAM_USAGE_BUFFER:
     // *usage = get_block_length();
     *usage = STREAM_BUFFER_SIZE*sizeof(char);
     break;
   case MM_STREAM_USAGE_CURRENT:
   case MM_STREAM_USAGE_MAXIMUM:
     // *usage = sizeof (*this) + get_block_length();
     *usage = sizeof (AMI_STREAM<T>) + STREAM_BUFFER_SIZE*sizeof(char);
     break;
   }
   return AMI_ERROR_NO_ERROR;
}



/**********************************************************************/
template<class T>
AMI_STREAM<T>::~AMI_STREAM(void)  {
  
  DEBUG_DELETE cerr << "~AMI_STREAM: " << path << "(" << this << ")\n";
  assert(fp);
  fclose(fp);
  delete buf;
  
  // Get rid of the file if not persistent and if not substream.
  if ((per != PERSIST_PERSISTENT) && (substream_level == 0)) {
    if (unlink(path) == -1) {
      cerr << "ERROR: AMI_STREAM: failed to unlink " << path << endl;
      perror("cannot unlink: ");
      DEBUG_ASSERT assert(0);
      exit(1);
    }
  }
  // Register memory deallocation before returning.
  //size_t usage; 
  //main_memory_usage(&usage,  MM_STREAM_USAGE_CURRENT);
  //MM_manager.register_deallocation(usage);
 }



/**********************************************************************/
template<class T>
AMI_err AMI_STREAM<T>::read_item(T **elt)  {

  assert(fp);

  //if we go past substream range
  if ((logical_eos >= 0) && G_ftell(fp) >= sizeof(T) * logical_eos) {
    return AMI_ERROR_END_OF_STREAM;
  
  } else {
    if (fread((char *) (&read_tmp), sizeof(T), 1, fp) < 1) {
      if(feof(fp)) {
        eof_reached = 1;
        return AMI_ERROR_END_OF_STREAM;
      } else {
        cerr << "ERROR: file=" << path << ":";
        perror("cannot read!");    
        return AMI_ERROR_IO_ERROR;
      }
    }
    
    *elt = &read_tmp;
    return AMI_ERROR_NO_ERROR; 
  }
}




/**********************************************************************/
template<class T>
AMI_err AMI_STREAM<T>::read_array(T *data, off_t len, off_t *lenp) {
  size_t nobj;
  assert(fp);
  
  //if we go past substream range
  if ((logical_eos >= 0) && G_ftell(fp) >= sizeof(T) * logical_eos) {
        eof_reached = 1;
    return AMI_ERROR_END_OF_STREAM;
    
  } else {
    nobj = fread((void*)data, sizeof(T), len, fp);

    if (nobj < len) {           /* some kind of error */
      if(feof(fp)) {
        if(lenp) *lenp = nobj;
        eof_reached = 1;
        return AMI_ERROR_END_OF_STREAM;
      } else {
        cerr << "ERROR: file=" << path << ":";
        perror("cannot read!");    
        return AMI_ERROR_IO_ERROR;
      }
    }
    if(lenp) *lenp = nobj;
    return AMI_ERROR_NO_ERROR; 
  }
}




/**********************************************************************/
template<class T>
AMI_err AMI_STREAM<T>::write_item(const T &elt) {

  assert(fp);
  //if we go past substream range
  if ((logical_eos >= 0) && G_ftell(fp) >= sizeof(T) * logical_eos) {
    return AMI_ERROR_END_OF_STREAM;
  
  } else {
    if (fwrite((char*)(&elt), sizeof(T), 1,fp) < 1) {
      cerr << "ERROR: AMI_STREAM::write_item failed.\n";
      if (*path)
        perror(path);
      else
        perror("AMI_STREAM::write_item: ");
      DEBUG_ASSERT assert(0);
      exit(1);
    }

    return AMI_ERROR_NO_ERROR;
  }
}


/**********************************************************************/
template<class T>
AMI_err AMI_STREAM<T>::write_array(const T *data, off_t len) {
  size_t nobj;

  assert(fp);
  //if we go past substream range
  if ((logical_eos >= 0) && G_ftell(fp) >= sizeof(T) * logical_eos) {
    return AMI_ERROR_END_OF_STREAM;
    
  } else {
    nobj = fwrite(data, sizeof(T), len, fp);
    if (nobj  < len) {
      cerr << "ERROR: AMI_STREAM::write_array failed.\n";
      if (*path)
        perror(path);
      else
        perror("AMI_STREAM::write_array: ");
      DEBUG_ASSERT assert(0);
      exit(1);
    }
   return AMI_ERROR_NO_ERROR;
  }
}
        

/**********************************************************************/
template<class T>
void AMI_STREAM<T>::persist(persistence p)  {
  per = p;
}



/**********************************************************************/
// sprint()
// Return a string describing the stream
//
// This function gives easy access to the file name, length.
// It is not reentrant, but this should not be too much of a problem 
// if you are careful.
template<class T>
char *AMI_STREAM<T>::sprint()  {
  static char desc[BUFSIZ];
  sprintf(desc, "[AMI_STREAM %s %ld]", path, (long)stream_len());
  return desc;
}

/**********************************************************************/
template<class T>
int AMI_STREAM<T>::eof()  {
  return eof_reached;
}


#endif // _AMI_STREAM_H