mariadb-columnstore-engine/utils/idbdatafile/BufferedFile.cpp

/* Copyright (C) 2014 InfiniDB, Inc.

   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; version 2 of
   the License.

   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.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
   MA 02110-1301, USA. */

#include <stdio.h>
#include <string>
#include <sys/stat.h>
#include <sstream>
#include <iostream>
#include "utility.h"
#include "BufferedFile.h"
#include "IDBLogger.h"
using namespace std;

namespace idbdatafile
{
BufferedFile::BufferedFile(const char* fname, const char* mode, unsigned opts)
 : IDBDataFile(fname), m_fp(0), m_buffer(0)
{
  m_fp = fopen(fname, mode);

  int err = errno;

  bool create = (mode[0]!='r') || (mode[1]!=0);

  #define UNABLE_OPEN_BUFFER_TEXT "Unable to open Buffered file.s"
  #define UNABLE_OPEN_BUFFER_THROW(X) throw std::runtime_error(UNABLE_OPEN_BUFFER_TEXT " " X)

  if (m_fp == NULL)
  {
    switch(err) {
      case EACCES:
        if (create) {
          UNABLE_OPEN_BUFFER_THROW("You do not have permission to create a file at that location, or a file already exists there owned by another user, which you are not allowed to modify.");
        } else {
          UNABLE_OPEN_BUFFER_THROW("You do not have permission to open the buffered file or one of the folders in its path.");
        }
        
        break;

        case EBUSY:
        UNABLE_OPEN_BUFFER_THROW("File creation is not possible because the disk is currently being used by the system.");
        
        break;

        case EDQUOT:
        UNABLE_OPEN_BUFFER_THROW("There is not enough space on the disk to create a new file.");
        
        break;

        case EFAULT:
        UNABLE_OPEN_BUFFER_THROW("This is a bug caused by an invalid fpath parameter in the constructor call of the BufferedFile class. Notify the vendor.");
        
        break;

        case EFBIG:  case EOVERFLOW:
        UNABLE_OPEN_BUFFER_THROW("The file is too large to open. It exceeds 4 gigabytes. Notify the vendor to support opening large files.");
        
        break;

        case EINTR:
        UNABLE_OPEN_BUFFER_THROW("Waiting to open the file on a slow device was interrupted.");
        
        break;

        case EINVAL:
        UNABLE_OPEN_BUFFER_THROW("The path contains a character that is not allowed by the operating system.");
        
        break;

        case EISDIR:
        UNABLE_OPEN_BUFFER_THROW("TThe path refers to an existing folder, so it cannot be opened as a file.");
        
        break;

        case ELOOP:
        UNABLE_OPEN_BUFFER_THROW("The path is cyclic via symbolic links.");
        
        break;

        case EMFILE:
        UNABLE_OPEN_BUFFER_THROW("The operating system does not support such a large number of open files per process.");
        
        break;

        case ENFILE:
        UNABLE_OPEN_BUFFER_THROW("The operating system does not support such a large number of open files.");
        
        break;

        case ENAMETOOLONG:
        UNABLE_OPEN_BUFFER_THROW("The path is too long.");
        
        break;

        case ENODEV:
        UNABLE_OPEN_BUFFER_THROW("The path points to a device that does not exist.");
        
        break;

        case ENOENT:
        if (mode[0] == 'r' && mode[1]==0) {
          UNABLE_OPEN_BUFFER_THROW("The file does not exist.");
        } else {
          UNABLE_OPEN_BUFFER_THROW("One of the folders in the path does not exist.");
        }
        
        break;

        case ENOMEM:
        UNABLE_OPEN_BUFFER_THROW("There is not enough memory to open the file.");
        
        break;

        case ENOSPC:
        UNABLE_OPEN_BUFFER_THROW("There is not enough disk space to create a new file.");
        
        break;

        case ENOTDIR:
        UNABLE_OPEN_BUFFER_THROW("A folder used in the path is not actually a folder, but something else.");
        
        break;

        case ENXIO:
        UNABLE_OPEN_BUFFER_THROW("The path refers to a UNIX domain socket or special device file that does not exist.");
        
        break;

        case EPERM:
        UNABLE_OPEN_BUFFER_THROW("The file cannot be opened because it is sealed.");
        
        break;

        case EROFS:
        UNABLE_OPEN_BUFFER_THROW("This file system is read-only.");
        
        break;

        case ETXTBSY:
        UNABLE_OPEN_BUFFER_THROW("The destination is busy.");
        
        break;

        default:

        throw std::runtime_error(UNABLE_OPEN_BUFFER_TEXT);

        break;
      }

      #undef UNABLE_OPEN_BUFFER_TEXT
      #undef UNABLE_OPEN_BUFFER_THROW
  }
  
  applyOptions(opts);
}

void BufferedFile::applyOptions(unsigned opts)
{
  if (opts & IDBDataFile::USE_VBUF)
  {
    const int DEFAULT_BUFSIZ = 1 * 1024 * 1024;
    m_buffer = new char[DEFAULT_BUFSIZ];
    setvbuf(m_fp, m_buffer, _IOFBF, DEFAULT_BUFSIZ);
  }
  else if (opts & IDBDataFile::USE_NOVBUF)
  {
    setvbuf(m_fp, NULL, _IONBF, 0);
  }
}

BufferedFile::~BufferedFile()
{
  close();
  m_fp = 0;
  delete[] m_buffer;
}

ssize_t BufferedFile::pread(void* ptr, off64_t offset, size_t count)
{
  ssize_t ret = 0;
  int savedErrno;
  ssize_t curpos = tell();

  seek(offset, SEEK_SET);
  ret = read(ptr, count);
  savedErrno = errno;
  seek(curpos, SEEK_SET);

  if (IDBLogger::isEnabled())
    IDBLogger::logRW("pread", m_fname, this, offset, count, ret);

  errno = savedErrno;
  return ret;
}

ssize_t BufferedFile::read(void* ptr, size_t count)
{
  ssize_t ret = 0;
  ssize_t offset = tell();
  int savedErrno = -1;
  size_t progress = 0;
  uint8_t* ptr8 = (uint8_t*)ptr;

  while (progress < count)
  {
    ret = fread(ptr8 + progress, 1, count - progress, m_fp);
    savedErrno = errno;

    if (ret <= 0)
    {
      if (ferror(m_fp))
      {
        errno = savedErrno;
        return -1;
      }
      else if (feof(m_fp))
        return progress;
    }

    progress += ret;
  }

  if (IDBLogger::isEnabled())
    IDBLogger::logRW("read", m_fname, this, offset, count, progress);

  errno = savedErrno;
  return progress;
}

ssize_t BufferedFile::write(const void* ptr, size_t count)
{
  ssize_t ret = 0;
  off64_t offset = tell();
  int savedErrno = 0;
  size_t progress = 0;
  uint8_t* ptr8 = (uint8_t*)ptr;

  while (progress < count)
  {
    ret = fwrite(ptr8 + progress, 1, count - progress, m_fp);
    savedErrno = errno;

    if (ret <= 0 && ferror(m_fp))
    {
      errno = savedErrno;
      return -1;
    }
    else if (ret > 0)
      progress += ret;

    // can fwrite() continually return 0 with no error?
  }

  if (IDBLogger::isEnabled())
    IDBLogger::logRW("write", m_fname, this, offset, count, progress);

  errno = savedErrno;
  return progress;
}

int BufferedFile::seek(off64_t offset, int whence)
{
  int ret = 0;
  int savedErrno;
  ret = fseek(m_fp, offset, whence);
  savedErrno = errno;

  if (IDBLogger::isEnabled())
    IDBLogger::logSeek(m_fname, this, offset, whence, ret);

  errno = savedErrno;
  return ret;
}

int BufferedFile::truncate(off64_t length)
{
  int ret = 0;
  int savedErrno;

  ret = ftruncate(fileno(m_fp), length);
  savedErrno = errno;

  if (IDBLogger::isEnabled())
    IDBLogger::logTruncate(m_fname, this, length, ret);

  errno = savedErrno;
  return ret;
}

off64_t BufferedFile::size()
{
  // going to calculate size 2 ways - first, via seek
  off64_t length = -1;
  off64_t here;

  flockfile(m_fp);

  try
  {
    if ((here = ftell(m_fp)) > -1)
    {
      if (fseek(m_fp, 0, SEEK_END) > -1)
      {
        length = ftell(m_fp);
        fseek(m_fp, here, SEEK_SET);
      }
    }

    funlockfile(m_fp);
  }
  catch (...)
  {
    funlockfile(m_fp);
  }

  return length;
}

off64_t BufferedFile::tell()
{
  return ftell(m_fp);
}

int BufferedFile::flush()
{
  int rc = fflush(m_fp);
  int savedErrno = errno;

  if (rc == 0)
  {
    rc = fsync(fileno(m_fp));
    savedErrno = errno;
  }

  if (IDBLogger::isEnabled())
    IDBLogger::logNoArg(m_fname, this, "flush", rc);

  errno = savedErrno;
  return rc;
}

time_t BufferedFile::mtime()
{
  time_t ret = 0;
  struct stat statbuf;

  if (::fstat(fileno(m_fp), &statbuf) == 0)
    ret = statbuf.st_mtime;
  else
    ret = (time_t)-1;

  return ret;
}

int BufferedFile::close()
{
  int ret = fclose(m_fp);
  int savedErrno = errno;

  if (IDBLogger::isEnabled())
    IDBLogger::logNoArg(m_fname, this, "close", ret);

  errno = savedErrno;
  return ret;
}

/**
     @brief
    The wrapper for fallocate function.
     @see
    This one is used in shared/we_fileop.cpp to skip expensive file preallocation.
*/
int BufferedFile::fallocate(int mode, off64_t offset, off64_t length)
{
  int ret = 0;
  int savedErrno = 0;

  ret = ::fallocate(fileno(m_fp), mode, offset, length);
  savedErrno = errno;

  if (IDBLogger::isEnabled())
  {
    IDBLogger::logNoArg(m_fname, this, "fallocate", errno);
  }

  errno = savedErrno;
  return ret;
}

}  // namespace idbdatafile