tapedrive.c

Solaris环境下的数据挖掘算法：birch聚类算法。该算法适用于对大量数据的挖掘。

int TapeDrive::write(void *buf, int recSize)
{
  buf = buf;                            // make compiler happy
  recSize = recSize;

  write_cnt++;
  DOASSERT(0, "Random writes not supported on tapes");
  return 0;
}

int TapeDrive::append(void *buf, int recSize)
{
  write_cnt++;

  if (!atEof) {
    gotoEndOfFile();
    atEof = 1;
    bufferBlock = tstat.mt_blkno;
  }

  if (bufferType != writeBuffer) {
    // can simply discard read buffer
    bufferOffset = 0;
    bufferType = writeBuffer;
  }

#ifdef TAPE_BLOCK_PADDING
  if (recSize > blockSize - 1)
    recSize = blockSize - 1;

  if (bufferOffset + recSize + 1 > blockSize)
    flushBuffer();

  DOASSERT(bufferOffset + recSize + 1 <= blockSize, "Inconsistent data");
  memcpy(buffer + bufferOffset, buf, recSize);
  *(buffer + bufferOffset + recSize) = '\0';
  bufferOffset += recSize + 1;
#else
  int bytes = recSize;
  char *p = (char *)buf;
  while(bytes > 0) {
    int spaceLeft = blockSize - bufferOffset;
    int b = (bytes <= spaceLeft ? bytes : spaceLeft);
    DOASSERT(bufferOffset + b <= blockSize, "Inconsistent data");
    memcpy(buffer + bufferOffset, p, b);
    bufferOffset += b;
    bytes -= b;
    p += b;
    if (bufferOffset >= blockSize)
      flushBuffer();
  }
#endif

  return recSize;
}

void TapeDrive::Recover(struct mtget &otstat, short mt_op,
                        daddr_t mt_count)
{
  mt_count = mt_count;                  // make compiler happy
  otstat = otstat;

  int status = 0;

  switch(mt_op) {
    case MTFSF:
      status = ProcessCmdNR(MTBSF, 1);
      if (status >= 0)
          ProcessCmdNR(MTFSF, 1);
      break;
    case MTBSF:
      status = ProcessCmdNR(MTFSF, 1);
      if (status >= 0)
          ProcessCmdNR(MTBSF, 1);
      break;
    case MTFSR:
      status = ProcessCmdNR(MTBSR, 1);
      if (status >= 0)
          ProcessCmdNR(MTFSR, 1);
      break;
    case MTBSR:
      status = ProcessCmdNR(MTFSR, 1);
      if (status >= 0)
          ProcessCmdNR(MTBSR, 1);
      break;
    case MTREW:
      status = ProcessCmdNR(MTFSF, 1);
      break;
    default:
      cout << "Don't know how to recover from an error with op "
           << mt_op << endl;
      break;
  }

  DOASSERT(status >= 0, "Recovery attempt failed");
}

#ifdef TAPE_THREAD
void *TapeDrive::ProcessCmd(void *arg)
{
  TapeDrive &me = *(TapeDrive *)arg;
  return me.ProcessCmd(me._proc_mt_op, me._proc_mt_count);
}
#endif

void *TapeDrive::ProcessCmd(short mt_op, daddr_t mt_count)
{
  static struct mtget otstat;           // original tape status
  int status = ioctl(fileno(file), MTIOCGET, (char *)&otstat);
  if (status < 0)
    reportErrSys("ioctl4");
  DOASSERT(status >= 0, "Cannot get tape status");

  for(int attempt = 0; attempt < 10; attempt++) {
      if (attempt > 0) {
          cout << "Sleeping 2 seconds..." << endl;
          sleep(2);
          cout << "Recovering..." << endl;
          Recover(otstat, mt_op, mt_count);
          cout << "Retrying..." << endl;
      }
      status = ProcessCmdNR(mt_op, mt_count);
      if (status >= 0)
          break;
      cout << "Tape command " << mt_op << ", count " << mt_count
           << " failed, attempt " << attempt << endl;
  }
  
  return (void *)0;
}

int TapeDrive::ProcessCmdNR(short mt_op, daddr_t mt_count)
{
  static struct mtop cmd;
  cmd.mt_op = mt_op;
  cmd.mt_count = mt_count;

  DOASSERT(mt_op >= 0 && mt_op < _max_mt_op, "Invalid tape command");
  mt_ios[mt_op]++;
  mt_cnt[mt_op] += (mt_count >= 0 ? mt_count : -mt_count);
  
  TAPEDBG(cout << "Tape " << fileno(file) << ", command " << mt_op
          << ", count " << mt_count << " started" << endl);

  startTimer();
  int status = ioctl(fileno(file), MTIOCTOP, (char *)&cmd);
  mt_tim[mt_op] += getTimer();

  if (status < 0)
      reportErrSys("ioctl");

  TAPEDBG(cout << "Tape " << fileno(file) << ", command " << mt_op
          << ", count " << mt_count << " finished, status = " << status
          << endl);

  return status;
}

int TapeDrive::command(short mt_op, daddr_t mt_count)
{
  waitForChildProcess();
  DOASSERT(_child <= 0, "Invalid child process ID");

#ifdef TAPE_THREAD
  _proc_mt_op = mt_op;
  _proc_mt_count = mt_count;
  if (pthread_create(&_child, 0, ProcessCmd, this)) {
      reportErrSys("pthread_create");
      return -1;
  }
#else
  // There seems to be a problem forking the tape command. Do it
  // in same process for now.
  _child = fork();

  if (!_child) {
    (void)ProcessCmd(mt_op, mt_count);
    exit(1);
  }

  if (_child < 0) {
      reportErrSys("fork");
      return -1;
  }
#endif

  return 0;
}

void TapeDrive::getStatus()
{
  waitForChildProcess();

#if defined(__aix) || defined(_AIX)
  tstat.mt_resid  = 0;
  tstat.mt_fileno = 0;
  tstat.mt_blkno  = 0;
#else
  int status = ioctl(fileno(file), MTIOCGET, (char *)&tstat);
  if (status < 0)
    reportErrSys("ioctl2");
  DOASSERT(status >= 0, "Cannot get tape status");
#endif
}

void TapeDrive::fillBuffer()
{
  waitForChildProcess();

  read_ios++;

  TAPEDBG2(cout << "Reading " << blockSize << " bytes to " << (void *)buffer
           << " from fd " << fileno(file) << endl);
  TAPEDBG2(cout << "Bufferblock " << bufferBlock << ", bufferOffset "
           << bufferOffset << endl);

#if 0
  startTimer();
#endif

#ifdef USE_FREAD
  size_t status;
#else
  int status;
#endif

  while (1) {
#ifdef USE_FREAD
      status = fread(buffer, blockSize, 1, file);
      if (!status && ferror(file) && errno == EINTR)
          continue;
#else
      status = ::read(fileno(file), buffer, blockSize);
      if (status < 0 && errno == EINTR)
          continue;
#endif
      break;
  }

#if 0
  read_time += getTimer();
#endif

  bufferBlock++;
  bufferOffset = 0;
  bufferBytes = status;

#ifdef USE_FREAD
  if (!status && feof(file)) {          // end of tape file?
#else
  if (!status) {                        // end of tape file?
#endif
    atEof = 1;
    TAPEDBG(cout << "Backing up past file mark we just passed" << endl);
    int status = command(MTBSF, 1);
    DOASSERT(status >= 0, "Cannot operate tape drive");
    return;
  }

  atEof = 0;

#ifdef USE_FREAD
  if (!status && ferror(file)) {        // read error?
#else
  if (status < 0) {                     // read error?
#endif
    cerr << "Read failed: fd " << fileno(file) << ", buffer "
         << (void *)buffer << ", bytes " << blockSize << endl;
    reportErrSys("read");
    exit(1);
  }
  
#ifdef PARTIAL_BLOCK_ERROR
  if (status < blockSize) {             // partial block read?
    cerr << "Partial block read: " << status << " vs. " << blockSize << endl;
    exit(1);
  }
#endif
}

void TapeDrive::flushBuffer()
{
  waitForChildProcess();

  DOASSERT(bufferType == writeBuffer, "Inconsistent data");
  if (!bufferOffset)
    return;

  write_ios++;
  if (bufferOffset < blockSize)
    memset(buffer + bufferOffset, 0, blockSize - bufferOffset);

#if 0
  startTimer();
#endif

  while (1) {
      size_t status = fwrite(buffer, blockSize, 1, file);
      if (!status && ferror(file) && errno == EINTR)
          continue;
      if (status < 1) {
          reportErrSys("fwrite");
          exit(1);
      }
      break;
  }

#if 0
  write_time += getTimer();
#endif

  bufferBlock++;
  bufferOffset = 0;
}

void TapeDrive::gotoBlock(long block)
{
  TAPEDBG(cout << "Go to block " << block << " of tape "
          << fileno(file) << endl);

  getStatus();

  if (fileNo != tstat.mt_fileno         // oops, we're in another file
      || tstat.mt_blkno > 900000000) {  // unsure about location
    gotoBeginningOfFile();
    getStatus();
  }

  long diff = block - tstat.mt_blkno;   // difference in block numbers
  if (!diff)                            // no movement required?
    return;

  if (!block) {                         // do we just want to go to BOF?
    gotoBeginningOfFile();
    return;
  }

  int status = 0;
  if (diff > 0)
    status = command(MTFSR, diff);      // go forward
  else
    status = command(MTBSR, -diff);     // go backward
  DOASSERT(status >= 0, "Cannot operate tape drive");
}

void TapeDrive::gotoBeginningOfFile()
{
  TAPEDBG(cout << "Going to beginning of file" << endl);

  int status = fseek(file, 0, SEEK_SET);
  if (status < 0)
    reportErrSys("fseek");
  DOASSERT(status >= 0, "Cannot operate tape drive");

  if (!fileNo) {                        // first file? just rewind
    int status = command(MTREW, 1);
    DOASSERT(status >= 0, "Cannot operate tape drive");
#if !defined(__alpha) && !defined(__ultrix)
    tstat.mt_fileno = 0;
    tstat.mt_blkno = 0;
    tstat.mt_resid = 0;
#endif
    return;
  }

  getStatus();
  long diff = fileNo - tstat.mt_fileno;

  if (diff > 0) {                       // go forward?
    int status = command(MTFSF, diff);
    DOASSERT(status >= 0, "Cannot operate tape drive");
  } else {                              // else go backward
    int status = command(MTBSF, -diff + 1);
    DOASSERT(status >= 0, "Cannot operate tape drive");
    status = command(MTFSF, 1);
    DOASSERT(status >= 0, "Cannot operate tape drive");
  }

#if !defined(__alpha) && !defined(__ultrix)
  tstat.mt_fileno = fileNo;
  tstat.mt_blkno = 0;
  tstat.mt_resid = 0;
#endif
}

void TapeDrive::gotoEndOfFile()
{ 
  if (atEof)
    return;

  waitForChildProcess();

  const unsigned long skipSize = 1000000;
  static struct mtop cmd;
  cmd.mt_op = MTFSR;
  cmd.mt_count = skipSize;

  while(1) {
    TAPEDBG(cout << "Tape command " << MTFSR << ", count " << skipSize
	    << ", " << flush);
    int status = ioctl(fileno(file), MTIOCTOP, (char *)&cmd);
    TAPEDBG(cout << "status = " << status << endl);
    if (status < 0) {
      getStatus();
      if (tstat.mt_resid > 0) {
	TAPEDBG(cout << "At end of file, drive status " << tstat.mt_dsreg
		<< ", error status " << tstat.mt_erreg << endl);
	mt_ios[MTFSR] -= tstat.mt_resid;
	return;
      }
    }
    reportErrSys("ioctl3");
    DOASSERT(status >= 0, "Cannot operate tape drive");
  }
}