tapedrive.c

数据挖掘经典的hierarchial clustering algorithm

/*
  This file contributed by the Tape Join project and modified locally.
  See attached copyright notice.
*/

/*
  $Id: tapedrive.C,v 1.15 1996/12/03 20:24:58 jussi Exp $

  $Log: tapedrive.C,v $
  Revision 1.15  1996/12/03 20:24:58  jussi
  Renamed preprocessor flags.

  Revision 1.14  1996/11/23 21:33:35  jussi
  Fixed some bugs when compiled with PROCESS_TASK.

  Revision 1.13  1996/10/02 15:24:02  wenger
  Improved error handling (modified a number of places in the code to use
  the DevError class).

  Revision 1.12  1996/09/26 18:55:42  jussi
  Added support for 64-bit file offsets. Tape commands are now
  executed in a subprocess or thread which increases parallelism
  in the system.

  Revision 1.11  1996/07/18 02:48:24  jussi
  Make this code compile in Ultrix.

  Revision 1.10  1996/07/12 00:55:39  jussi
  Updated copyright information to reflect original source.

  Revision 1.9  1996/06/27 16:04:25  jussi
  Added a cast in memchr() so that the code compiles cleanly in Linux.

  Revision 1.8  1996/04/16 20:56:25  jussi
  Replaced assert() calls with DOASSERT macro.

  Revision 1.7  1996/01/13 03:22:51  jussi
  Removed #include <tar.h> -- this is in tapedrive.h.

  Revision 1.6  1995/12/28 17:50:00  jussi
  Small fixes to remove new compiler warnings.

  Revision 1.5  1995/11/09 22:23:28  jussi
  Added debugging statements.

  Revision 1.4  1995/10/31 17:13:17  jussi
  Added tar archive handling routines and data structures.

  Revision 1.3  1995/09/22 15:46:22  jussi
  Added copyright message.

  Revision 1.2  1995/09/05 20:31:56  jussi
  Added CVS header.
*/

/*
  Copyright 1993-1996 by Jussi Myllymaki

  Permission to use, copy, modify, and distribute this software and its
  documentation for any purpose and without fee is hereby granted,
  provided that the above copyright notice appear in all copies and that
  both that copyright notice and this permission notice appear in
  supporting documentation, and that the name(s) of the copyright holder(s)
  not be used in advertising or publicity pertaining to distribution of
  the software without specific, written prior permission. The copyright
  holder(s) make no representations about the suitability of this
  software for any purpose.  It is provided "as is" without express
  or implied warranty.

  Author: Jussi Myllymaki
*/

//#define TAPE_DEBUG
//#define TAPE_DEBUG2

#include <iostream.h>
#include <unistd.h>
#include <string.h>

#include "tapedrive.h"
#include "DCE.h"
#include "Exit.h"
#include "DevError.h"

// Use fake fileno and blkno to make this file compile in Alpha
// until a real fix is found. The problem is that in Alpha/OSF,
// struct mtget (mtio.h) does not include fields mt_fileno and
// mt_blkno. See warning printed out below.

#if defined(__alpha) || defined(__ultrix)
#define mt_fileno mt_resid * 0
#define mt_blkno  mt_resid * 0
#endif

#define USE_FWRITEBUF
//#define USE_FREAD

char *TapeDrive::_mt_op_name[] = { "WEOF", "FSF", "BSF", "FSR",
                                   "BSR", "REW", "OFFL", "NOP",
                                   "RETEN", "ERASE", "EOM", "NBSF",
                                   "SRSZ", "GRSZ", "LOAD" };

TapeDrive::TapeDrive(char *name, char *mode, int fno, int blockSz) :
        initialized(0), fileNo(fno), blockSize(blockSz),
        haveTarHeader(0), tarFileSize(0), tarFileOffset(0)
{
#if defined(__alpha) || defined(__ultrix)
  cerr << "Warning: In Ultrix and Alpha/OSF/1, file number and block number"
       << endl;
  cerr << "         inquiry does not work. TapeDrive will probably fail."
       << endl;
#endif

  _child = 0;

  if (!(file = fopen(name, mode))) {
    reportErrSys("fopen");
    return;
  }

#ifdef USE_FWRITEBUF
  // fwrite() in flushBuffer() will write 8 kB blocks to tape even
  // if we request larger blocks; we have to force it to use
  // the specified block size
  int fwriteBufSize = blockSize + 8;
  _fwriteBuf = new char [fwriteBufSize];
  DOASSERT(_fwriteBuf, "Out of memory");
  if (setvbuf(file, _fwriteBuf, _IOFBF, fwriteBufSize) != 0) {
      reportErrSys("setvbuf");
      exit(1);
  }
#endif

#if 0
  setbuf(file, 0);
#endif

  atEof = (mode[0] == 'w');

  for(unsigned int i = 0; i < _max_mt_op; i++)
    mt_tim[i] = mt_ios[i] = mt_cnt[i] = 0;
  read_time = read_ios = read_cnt = 0;
  write_time = write_ios = write_cnt = 0;

  // If the caller gave us the 'target' file number, let's use it.
  // If no number was given, use the current file on the tape.
  if (fileNo < 0) {
    getStatus();
    fileNo = tstat.mt_fileno;
  }
  gotoBeginningOfFile();

  buffer = new char [blockSize];
  DOASSERT(buffer, "Out of memory");
  bufferType = readBuffer;
  bufferBlock = 0;
  bufferOffset = 0;
  bufferBytes = 0;

  initialized = 1;
}

TapeDrive::~TapeDrive()
{
  if (!initialized)
    return;

  if (bufferType == writeBuffer)
    flushBuffer();
  delete buffer;

  waitForChildProcess();

  setbuf(file, 0);

  if (fclose(file))
    reportErrSys("fclose");

#ifdef USE_FWRITEBUF
  delete _fwriteBuf;
#endif
}

void TapeDrive::printStats()
{
  cout << "Tape usage statistics:" << endl;
  cout << "  cmd\tcalls\tcount\tavgtime" << endl;
  for(unsigned int i = 0; i < _max_mt_op; i++) {
    if (mt_ios[i] > 0) {
      cout << "  " << _mt_op_name[i]
           << "\t" << mt_ios[i]
           << "\t" << mt_cnt[i]
           << "\t" << mt_tim[i] / (mt_cnt[i] ? mt_cnt[i] : 1) << endl;
    }
  }
  cout << "  read\t" << read_ios << "\t" << read_cnt
       << "\t" << read_time / (read_ios ? read_ios : 1) << endl;
  cout << "  write\t" << write_ios << "\t" << write_cnt
       << "\t" << write_time / (write_ios ? write_ios : 1) << endl;
}

void TapeDrive::readTarHeader()
{
  DOASSERT(!haveTarHeader, "Invalid tar header flag");

  int bytes = read(&tarHeader, sizeof tarHeader);
  DOASSERT(bytes == sizeof tarHeader, "Invalid tar header size");
  tarFileSize = oct2int(tarHeader.dbuf.size);
  tarFileOffset = 0;

  TAPEDBG(cout << "Read tar header: " << tarHeader.dbuf.name << ", size "
          << tarFileSize << endl);

  haveTarHeader = 1;
}

unsigned long int TapeDrive::oct2int(char *buf)
{
  unsigned long int num = 0;
  while(*buf == ' ') buf++;
  while(*buf != ' ')
    num = 8 * num + (*buf++ - '0');
  return num;
}

void TapeDrive::waitForChildProcess()
{
  if (_child > 0) {
    TAPEDBG(cout << "Waiting for tape " << fileno(file)
            << " to become idle" << endl);
#ifdef TAPE_THREAD
    (void)pthread_join(_child, 0);
#else
    while(1) {
        int status;
        pid_t child = wait(&status);
        if (child < 0) {
            if (errno == EINTR)
                continue;
            if (errno != ECHILD) {
                reportErrSys("wait");
                exit(1);
            }
        } else
            break;
    }
#endif
    TAPEDBG(cout << "Tape " << fileno(file) << " has become idle" << endl);
    _child = 0;
  }
}

long long TapeDrive::seek(long long offset)
{
  TAPEDBG(cout << "Seek to offset " << offset << " of tape "
          << fileno(file) << endl);

  DOASSERT(offset >= 0, "Invalid tape offset");

  if (bufferType == writeBuffer) {      // flush out write buffer
    flushBuffer();
    bufferType = readBuffer;
    bufferBlock = 0;
  }

  long long lblock = offset / blockSize;
  long long loff = offset % blockSize;

  long block = lblock;
  long off = loff;

  TAPEDBG(cout << "Seeking to lblock " << lblock << ", block "
          << block << " of tape " << fileno(file) << endl);

  if (block != bufferBlock - 1) {       // not current block?
    gotoBlock(block);                   // goto new block location
    fillBuffer();                       // read it into buffer
  }

  bufferBlock = block + 1;              // on tape, just past current block
  bufferOffset = off;

  if (bufferOffset > bufferBytes) {
    cerr << "Seeking past end of file" << endl;
    exit(1);
  }

  return offset;
}

int TapeDrive::read(void *buf, int recSize, int binary)
{
  TAPEDBG2(cout << "Read request for " << recSize << " "
           << (binary ? "binary" : "ASCII")
           << " bytes to " << (void *)buf << endl);

  if (bufferType != readBuffer) {
    cerr << "Must do a seek before switching from writing to reading" << endl;
    exit(1);
  }

  DOASSERT(bufferOffset >= 0 && bufferOffset <= blockSize,
           "Inconsistent data");
  DOASSERT(bufferBytes >= 0 && bufferBytes <= blockSize, "Inconsistent data");
  DOASSERT(bufferOffset <= bufferBytes, "Inconsistent data");

#ifdef TARFILESIZE
  if (haveTarHeader                     // is file in a tar archive?
      && tarFileOffset >= tarFileSize)  // and at end of file?
    atEof = 1;
#endif

  if (atEof)                            // already at end of tape file?
    return 0;

  if (bufferOffset >= bufferBytes) {    // no more bytes in buffer?
    fillBuffer();                       // get next block from file
    if (!bufferBytes) {                 // end of file?

#ifdef TARFILESIZE
      // for non-tar files, end of tape file is the natural end of
      // user file; for tar files, the file size indicated in the
      // tar header should trigger the atEof statement a few lines
      // up; it is an error if the tape file (tar file) ends before
      // the file inside the tar file
      if (haveTarHeader)
        cerr << "File in tar archive prematurely terminated." << endl;
#endif

      return 0;
    }
  }

  read_cnt++;
  if (read_cnt % 1000 == 0)
    TAPEDBG2(cout << read_cnt << " " << flush);

#ifdef TAPE_BLOCK_PADDING
  char *start = buffer + bufferOffset;        // starting point for this record
  DOASSERT(*start != 0, "Unexpected record"); // must not be an empty record

  if (recSize > bufferBytes - bufferOffset)
    recSize = bufferBytes - bufferOffset;

#ifdef TARFILESIZE
  if (haveTarHeader                     // past EOF of file in tar archive?
      && recSize > tarFileSize - tarFileOffset)
    recSize = tarFileSize - tarFileOffset;
#endif

  if (!binary) {                        // reading an ASCII record?
    char *end = (char *)memchr(start, 0, recSize);
    DOASSERT(end, "End of record not found");
    recSize = end - start;              // do not include record separator
  }
  TAPEDBG2(cout << "Copying " << recSize << " bytes to "
           << (void *)buf << endl);
  memcpy(buf, start, recSize);
  bufferOffset += recSize + 1;          // go past record separator too
  if (!binary                           // in ASCII mode?
      && bufferOffset < bufferBytes     // still data left but...
      && !buffer[bufferOffset]))        // last record in block?
    bufferOffset = bufferBytes;         // must fetch new block next time

#else

  int bytesLeft = recSize;

#ifdef TARFILESIZE
  if (haveTarHeader                     // past EOF of file in tar archive?
      && bytesLeft > tarFileSize - tarFileOffset)
    bytesLeft = tarFileSize - tarFileOffset;
#endif

  recSize = 0;
  char *p = (char *)buf;
  while(bytesLeft > 0) {
    char *start = buffer + bufferOffset;
    int b = bufferBytes - bufferOffset; // bytes left in buffer
    if (bytesLeft < b)                  // caller doesn't want that many?
      b = bytesLeft;
    char *end = 0;
    if (!binary) {                      // reading an ASCII record?
      end = (char *)memchr((void *)start, '\n', b);
      if (end)                          // found newline = end of record?
        b = end - start + 1;
    }
    TAPEDBG2(cout << "Copying " << b << " bytes to " << (void *)p << endl);
    memcpy(p, start, b);
    bufferOffset += b;
    bytesLeft -= b;
    recSize += b;
    p += b;
    if (end)                            // found newline = end of record?
      break;
    if (bufferOffset >= bufferBytes)    // need next block?
      fillBuffer();
    if (!bufferBytes)                   // end of physical file?
      break;
  }

  if (!binary                           // in ASCII mode?
      && bufferOffset < bufferBytes     // still data left but...
      && !buffer[bufferOffset])         // end of logical file (NULL char)?
    bufferOffset = bufferBytes;         // must try to fetch block next time
#endif

#ifdef TARFILESIZE
  if (haveTarHeader) {
     tarFileOffset += recSize;
     DOASSERT(tarFileOffset <= tarFileSize, "Inconsistent data");
  }
#endif

  return recSize;
}