<commit_before>////////////////////////////////////////////////////////////////////////////////
/// @brief datafiles
///
/// @file
///
/// DISCLAIMER
///
/// Copyright 2010-2011 triagens GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
/// you may not use this file except in compliance with the License.
/// You may obtain a copy of the License at
///
///     http://www.apache.org/licenses/LICENSE-2.0
///
/// Unless required by applicable law or agreed to in writing, software
/// distributed under the License is distributed on an "AS IS" BASIS,
/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
/// See the License for the specific language governing permissions and
/// limitations under the License.
///
/// Copyright holder is triAGENS GmbH, Cologne, Germany
///
/// @author Dr. <NAME>
/// @author Copyright 2011, triagens GmbH, Cologne, Germany
////////////////////////////////////////////////////////////////////////////////

#ifdef _WIN32
#include <BasicsC/win-utils.h>
#endif

#include "datafile.h"

#include "BasicsC/hashes.h"
#include "BasicsC/logging.h"
#include "BasicsC/memory-map.h"
#include "BasicsC/strings.h"
#include "BasicsC/files.h"


// #define DEBUG_DATAFILE 1

// -----------------------------------------------------------------------------
// --SECTION--                                                 private functions
// -----------------------------------------------------------------------------

////////////////////////////////////////////////////////////////////////////////
/// @addtogroup VocBase
/// @{
////////////////////////////////////////////////////////////////////////////////

////////////////////////////////////////////////////////////////////////////////
/// @brief return whether the datafile is a physical file (true) or an
/// anonymous mapped region (false)
////////////////////////////////////////////////////////////////////////////////

static bool IsPhysicalDatafile (const TRI_datafile_t* const datafile) {
  return datafile->_filename != NULL;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief return the name of a datafile
////////////////////////////////////////////////////////////////////////////////

static const char* GetNameDatafile (const TRI_datafile_t* const datafile) {
  if (datafile->_filename == NULL) {
    // anonymous regions do not have a filename
    return "anonymous region";
  }

  // return name of the physical file
  return datafile->_filename;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief close a datafile
////////////////////////////////////////////////////////////////////////////////

static void CloseDatafile (TRI_datafile_t* const datafile) {
  assert(datafile->_state != TRI_DF_STATE_CLOSED);

  if (datafile->isPhysical(datafile)) {
    TRI_CLOSE(datafile->_fd);
  }

  datafile->_state = TRI_DF_STATE_CLOSED;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief destroy a datafile
////////////////////////////////////////////////////////////////////////////////

static void DestroyDatafile (TRI_datafile_t* const datafile) {
  if (datafile->_filename != NULL) {
    TRI_FreeString(TRI_CORE_MEM_ZONE, datafile->_filename);
  }
}

////////////////////////////////////////////////////////////////////////////////
/// @brief sync the data of a datafile
////////////////////////////////////////////////////////////////////////////////

static bool SyncDatafile (const TRI_datafile_t* const datafile, 
                          char const* begin, 
                          char const* end) {
  // TODO: remove
  void** mmHandle = NULL;

  if (datafile->_filename == NULL) {
    // anonymous regions do not need to be synced
    return true;
  }

  assert(datafile->_fd > 0);

  return TRI_msync(datafile->_fd, mmHandle, begin, end);
}

////////////////////////////////////////////////////////////////////////////////
/// @brief truncate the datafile to a specific length
////////////////////////////////////////////////////////////////////////////////

static int TruncateDatafile (TRI_datafile_t* const datafile, const off_t length) {
  if (datafile->isPhysical(datafile)) {
    // only physical files can be truncated
    return ftruncate(datafile->_fd, length);
  }

  // for anonymous regions, this is a non-op
  return 0;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief creates a new sparse datafile
///
/// returns the file descriptor or -1 if the file cannot be created
////////////////////////////////////////////////////////////////////////////////

static int CreateSparseFile (char const* filename, 
                             const TRI_voc_size_t maximalSize) {
  off_t offset;
  char zero;
  ssize_t res;
  int fd;

  // open the file
  fd = TRI_CREATE(filename, O_CREAT | O_EXCL | O_RDWR, S_IRUSR | S_IWUSR);

  if (fd < 0) {
    TRI_set_errno(TRI_ERROR_SYS_ERROR);

    LOG_ERROR("cannot create datafile '%s': %s", filename, TRI_last_error());
    return -1;
  }

  // create sparse file
  offset = TRI_LSEEK(fd, maximalSize - 1, SEEK_SET);

  if (offset == (off_t) -1) {
    TRI_set_errno(TRI_ERROR_SYS_ERROR);
    TRI_CLOSE(fd);
    
    // remove empty file
    TRI_UnlinkFile(filename);

    LOG_ERROR("cannot seek in datafile '%s': '%s'", filename, TRI_last_error());
    return -1;
  }

  zero = 0;
  res = TRI_WRITE(fd, &zero, 1);

  if (res < 0) {
    TRI_set_errno(TRI_ERROR_SYS_ERROR);
    TRI_CLOSE(fd);
    
    // remove empty file
    TRI_UnlinkFile(filename);

    LOG_ERROR("cannot create sparse datafile '%s': '%s'", filename, TRI_last_error());
    return -1;
  }

  return fd;
}


////////////////////////////////////////////////////////////////////////////////
/// @brief initialises a datafile
////////////////////////////////////////////////////////////////////////////////

static void InitDatafile (TRI_datafile_t* datafile,
                          char* filename,
                          int fd,
                          void* mmHandle,
                          TRI_voc_size_t maximalSize,
                          TRI_voc_size_t currentSize,
                          TRI_voc_fid_t tick,
                          char* data) {

  // filename is a string for physical datafiles, and NULL for anonymous regions
  // fd is a positive value for physical datafiles, and -1 for anonymous regions
  if (filename == NULL) {
    assert(fd == -1);
  }
  else {
    assert(fd > 0);
  }

  datafile->_state       = TRI_DF_STATE_READ;
  datafile->_fid         = tick;

  datafile->_filename    = filename;
  datafile->_fd          = fd;
  datafile->_mmHandle    = mmHandle;

  datafile->_maximalSize = maximalSize;
  datafile->_currentSize = currentSize;
  datafile->_footerSize  = sizeof(TRI_df_footer_marker_t);

  datafile->_isSealed    = false;
  datafile->_lastError   = TRI_ERROR_NO_ERROR;

  datafile->_full        = false;

  datafile->_data        = data;
  datafile->_next        = data + currentSize;

  datafile->_synced      = data;
  datafile->_written     = NULL;

  // initialise function pointers
  datafile->isPhysical   = &IsPhysicalDatafile;
  datafile->getName      = &GetNameDatafile;
  datafile->close        = &CloseDatafile;
  datafile->destroy      = &DestroyDatafile;
  datafile->sync         = &SyncDatafile;
  datafile->truncate     = &TruncateDatafile;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief truncates a datafile
///
/// Create a truncated datafile, seal it and rename the old.
////////////////////////////////////////////////////////////////////////////////

static int TruncateAndSealDatafile (TRI_datafile_t* datafile, 
                                    TRI_voc_size_t vocSize) {
  char* filename;
  char* oldname;
  char zero;
  int fd;
  int res;
  size_t maximalSize;
  size_t offset;
  void* data;
  void* mmHandle;
  
  // this function must not be called for non-physical datafiles
  assert(datafile->isPhysical(datafile));
    
  // use multiples of page-size
  maximalSize = ((vocSize + sizeof(TRI_df_footer_marker_t) + PageSize - 1) / PageSize) * PageSize;

  // sanity check
  if (sizeof(TRI_df_header_marker_t) + sizeof(TRI_df_footer_marker_t) > maximalSize) {
    LOG_ERROR("cannot create datafile '%s', maximal size '%u' is too small", datafile->getName(datafile), (unsigned int) maximalSize);
    return TRI_set_errno(TRI_ERROR_ARANGO_MAXIMAL_SIZE_TOO_SMALL);
  }

  // open the file
  filename = TRI_Concatenate2String(datafile->_filename, ".new");

  fd = TRI_CREATE(filename, O_CREAT | O_EXCL | O_RDWR, S_IRUSR | S_IWUSR);

  if (fd < 0) {
    LOG_ERROR("cannot create new datafile '%s': '%s'", filename, TRI_last_error());
    return TRI_set_errno(TRI_ERROR_SYS_ERROR);
  }

  // create sparse file
  offset = TRI_LSEEK(fd, maximalSize - 1, SEEK_SET);

  if (offset == (off_t) -1) {
    TRI_set_errno(TRI_ERROR_SYS_ERROR);
    TRI_CLOSE(fd);

    // remove empty file
    TRI_UnlinkFile(filename);

    LOG_ERROR("cannot seek in new datafile '%s': '%s'", filename, TRI_last_error());
    return TRI_ERROR_SYS_ERROR;
  }

  zero = 0;
  res = TRI_WRITE(fd, &zero, 1);

  if (res < 0) {
    TRI_set_errno(TRI_ERROR_SYS_ERROR);
    TRI_CLOSE(fd);
    
    // remove empty file
    TRI_UnlinkFile(filename);

    LOG_ERROR("cannot create sparse datafile '%s': '%s'", filename, TRI_last_error());
    return TRI_ERROR_SYS_ERROR;
  }

  // memory map the data
  res = TRI_MMFile(0, maximalSize, PROT_WRITE | PROT_READ, MAP_SHARED, fd, &mmHandle, 0, &data);
  
  if (res != TRI_ERROR_NO_ERROR) {  
    TRI_set_errno(res);
    TRI_CLOSE(fd);

    // remove empty file
    TRI_UnlinkFile(filename);

    LOG_ERROR("cannot memory map file '%s': '%s'", filename, TRI_last_error());
    return TRI_errno();
  }

  // copy the data
  memcpy(data, datafile->_data, vocSize);

  // patch the datafile structure
  res = TRI_UNMMFile(datafile->_data, datafile->_maximalSize, datafile->_fd, &(datafile->_mmHandle));

  if (res < 0) {
    LOG_ERROR("munmap failed with: %d", res);
    return res;
  }

  // .............................................................................................
  // For windows: Mem mapped files use handles
  // the datafile->_mmHandle handle object has been closed in the underlying 
  // TRI_UNMMFile(...) call above so we do not need to close it for the associated file below
  // .............................................................................................
  
  TRI_CLOSE(datafile->_fd);

  
  datafile->_data = data;
  datafile->_next = (char*)(data) + vocSize;
  datafile->_maximalSize = maximalSize;
  datafile->_fd = fd;
  datafile->_mmHandle = mmHandle;
  
  // rename files
  oldname = TRI_Concatenate2String(datafile->_filename, ".corrupted");

  res = TRI_RenameFile(datafile->_filename, oldname);

  if (res != TRI_ERROR_NO_ERROR) {
    TRI_FreeString(TRI_CORE_MEM_ZONE, filename);
    TRI_FreeString(TRI_CORE_MEM_ZONE, oldname);

    return res;
  }

  res = TRI_RenameFile(filename, datafile->_filename);

  if (res != TRI_ERROR_NO_ERROR) {
    TRI_FreeString(TRI_CORE_MEM_ZONE, filename);
    TRI_FreeString(TRI_CORE_MEM_ZONE, oldname);

    return res;
  }

  TRI_SealDatafile(datafile);
  return TRI_ERROR_NO_ERROR;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief scans a datafile
////////////////////////////////////////////////////////////////////////////////

static TRI_df_scan_t ScanDatafile (TRI_datafile_t const* datafile) {
  TRI_df_scan_t scan;
  TRI_df_scan_entry_t entry;
  TRI_voc_size_t currentSize;
  char* end;
  char* ptr;

  // this function must not be called for non-physical datafiles
  assert(datafile->isPhysical(datafile));

  ptr = datafile->_data;
  end = datafile->_data + datafile->_currentSize;
  currentSize = 0;

  TRI_InitVector(&scan._entries, TRI_CORE_MEM_ZONE, sizeof(TRI_df_scan_entry_t));

  scan._currentSize = datafile->_currentSize;
  scan._maximalSize = datafile->_maximalSize;
  scan._numberMarkers = 0;
  scan._status = 1;

  if (datafile->_currentSize == 0) {
    end = datafile->_data + datafile->_maximalSize;
  }

  while (ptr < end) {
    TRI_df_marker_t* marker = (TRI_df_marker_t*) ptr;
    bool ok;
    size_t size;

    memset(&entry, 0, sizeof(entry));

    entry._position = ptr - datafile->_data;
    entry._size = marker->_size;
    entry._tick = marker->_tick;
    entry._type = marker->_type;
    entry._status = 1;

    if (marker->_size == 0 && marker->_crc == 0 && marker->_type == 0 && marker->_tick == 0) {
      entry._status = 2;

      scan._endPosition = currentSize;

      TRI_PushBackVector(&scan._entries, &entry);
      return scan;
    }

    ++scan._numberMarkers;

    if (marker->_size == 0) {
      entry._status = 3;

      scan._status = 2;
      scan._endPosition = currentSize;

      TRI_PushBackVector(&scan._entries, &entry);
      return scan;
    }

    if (marker->_size < sizeof(TRI_df_marker_t)) {
      entry._status = 4;

      scan._endPosition = currentSize;
      scan._status = 3;

      TRI_PushBackVector(&scan._entries, &entry);
      return scan;
    }

    ok = TRI_CheckCrcMarkerDatafile(marker);

    if (! ok) {
      entry._status = 5;
      scan._status = 4;
    }

    TRI_PushBackVector(&scan._entries, &entry);

    size = TRI_DF_ALIGN_BLOCK(marker->_size);
    currentSize += size;

    if (marker->_type == TRI_DF_MARKER_FOOTER) {
      scan._endPosition = currentSize;
      return scan;
    }

    ptr += size;
  }

  return scan;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief checks a datafile
////////////////////////////////////////////////////////////////////////////////

static bool CheckDatafile (TRI_datafile_t* datafile) {
  TRI_voc_size_t currentSize;
  char* end;
  char* ptr;
  
  // this function must not be called for non-physical datafiles
  assert(datafile->isPhysical(datafile));

  ptr = datafile->_data;
  end = datafile->_data + datafile->_currentSize;
  currentSize = 0;

  if (datafile->_currentSize == 0) {
    LOG_WARNING("current size is 0 in read-only datafile '%s', trying to fix", datafile->getName(datafile));

    end = datafile->_data + datafile->_maximalSize;
  }

  while (ptr < end) {
    TRI_df_marker_t* marker = (TRI_df_marker_t*) ptr;
    bool ok;
    size_t size;

#ifdef DEBUG_DATAFILE
    LOG_TRACE("MARKER: size %lu, tick %lx, crc %lx, type %u",
              (unsigned long) marker->_size,
              (unsigned long) marker->_tick,
              (unsigned long) marker->_crc,
              (unsigned int) marker->_type);
#endif

    if (marker->_size == 0 && marker->_crc == 0 && marker->_type == 0 && marker->_tick == 0) {
      LOG_DEBUG("reached end of datafile '%s' data, current size %lu", 
                datafile->getName(datafile), 
                (unsigned long) currentSize);

      datafile->_currentSize = currentSize;
      datafile->_next = datafile->_data + datafile->_currentSize;

      return true;
    }

    if (marker->_size == 0) {
      LOG_DEBUG("reached end of datafile '%s' data, current size %lu", 
                datafile->getName(datafile), 
                (unsigned long) currentSize);

      datafile->_currentSize = currentSize;
      datafile->_next = datafile->_data + datafile->_currentSize;

      return true;
    }

    if (marker->_size < sizeof(TRI_df_marker_t)) {
      datafile->_lastError = TRI_set_errno(TRI_ERROR_ARANGO_CORRUPTED_DATAFILE);
      datafile->_currentSize = currentSize;
      datafile->_next = datafile->_data + datafile->_currentSize;
      datafile->_state = TRI_DF_STATE_OPEN_ERROR;

      LOG_WARNING("marker in datafile '%s' too small, size %lu, should be at least %lu",
                  datafile->getName(datafile),
                  (unsigned long) marker->_size,
                  (unsigned long) sizeof(TRI_df_marker_t));

      return false;
    }

    ok = TRI_CheckCrcMarkerDatafile(marker);

    if (! ok) {
      datafile->_lastError = TRI_set_errno(TRI_ERROR_ARANGO_CORRUPTED_DATAFILE);
      datafile->_currentSize = currentSize;
      datafile->_next = datafile->_data + datafile->_currentSize;
      datafile->_state = TRI_DF_STATE_OPEN_ERROR;

      LOG_WARNING("crc mismatch found in datafile '%s'", datafile->getName(datafile));

      return false;
    }

    TRI_UpdateTickVocBase(marker->_tick);

    size = TRI_DF_ALIGN_BLOCK(marker->_size);
    currentSize += size;

    if (marker->_type == TRI_DF_MARKER_FOOTER) {
      LOG_DEBUG("found footer, reached end of datafile '%s', current size %lu",
                datafile->getName(datafile),
                (unsigned long) currentSize);

      datafile->_isSealed = true;
      datafile->_currentSize = currentSize;
      datafile->_next = datafile->_data + datafile->_currentSize;

      return true;
    }

    ptr += size;
  }

  return true;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief opens a datafile
////////////////////////////////////////////////////////////////////////////////

static TRI_datafile_t* OpenDatafile (char const* filename, bool ignoreErrors) {
  TRI_datafile_t* datafile;
  TRI_voc_size_t size;
  bool ok;
  void* data;
  char* ptr;
  int fd;
  int res;
  ssize_t len;
  struct stat status;
  TRI_df_header_marker_t header;
  void* mmHandle;
  
  // this function must not be called for non-physical datafiles
  assert(filename != NULL);

  // ..........................................................................  
  // attempt to open a datafile file
  // ..........................................................................  

  fd = TRI_OPEN(filename, O_RDWR);

  if (fd < 0) {
    TRI_set_errno(TRI_ERROR_SYS_ERROR);

    LOG_ERROR("cannot open datafile '%s': '%s'", filename, TRI_last_error());

    return NULL;
  }


  // compute the size of the file
  res = fstat(fd, &status);

  if (res < 0) {
    TRI_set_errno(TRI_ERROR_SYS_ERROR);
    TRI_CLOSE(fd);

    LOG_ERROR("cannot get status of datafile '%s': %s", filename, TRI_last_error());

    return NULL;
  }

  // check that file is not too small
  size = status.st_size;

  if (size < sizeof(TRI_df_header_marker_t) + sizeof(TRI_df_footer_marker_t)) {
    TRI_set_errno(TRI_ERROR_ARANGO_CORRUPTED_DATAFILE);
    TRI_CLOSE(fd);

    LOG_ERROR("datafile '%s' is corrupted, size is only %u", filename, (unsigned int) size);

    return NULL;
  }

  // read header from file
  ptr = (char*) &header;
  len = sizeof(TRI_df_header_marker_t);

  ok = TRI_ReadPointer(fd, ptr, len);

  if (! ok) {
    LOG_ERROR("cannot read datafile header from '%s': %s", filename, TRI_last_error());

    TRI_CLOSE(fd);
    return NULL;
  }

  // check CRC
  ok = TRI_CheckCrcMarkerDatafile(&header.base);

  if (! ok) {
    TRI_set_errno(TRI_ERROR_ARANGO_CORRUPTED_DATAFILE);

    LOG_ERROR("corrupted datafile header read from '%s'", filename);

    if (! ignoreErrors) {
      TRI_CLOSE(fd);
      return NULL;
    }
  }

  // check the datafile version
  if (ok) {
    if (header._version != TRI_DF_VERSION) {
      TRI_set_errno(TRI_ERROR_ARANGO_CORRUPTED_DATAFILE);

      LOG_ERROR("unknown datafile version '%u' in datafile '%s'",
                (unsigned int) header._version,
                filename);

      if (! ignoreErrors) {
        TRI_CLOSE(fd);
        return NULL;
      }
    }
  }

  // check the maximal size
  if (size > header._maximalSize) {
    LOG_WARNING("datafile '%s' has size '%u', but maximal size is '%u'",
                filename,
                (unsigned int) size,
                (unsigned int) header._maximalSize);
  }

  // map datafile into memory
  res = TRI_MMFile(0, size, PROT_READ, MAP_SHARED, fd, &mmHandle, 0, &data);
  
  if (res != TRI_ERROR_NO_ERROR) {
    TRI_set_errno(res);
    TRI_CLOSE(fd);

    LOG_ERROR("cannot memory map datafile '%s': '%d'", filename, res);
    return NULL;
  }

  // create datafile structure
  datafile = TRI_Allocate(TRI_UNKNOWN_MEM_ZONE, sizeof(TRI_datafile_t), false);

  if (datafile == NULL) {
    return NULL;
  }

  InitDatafile(datafile,
               TRI_DuplicateString(filename),<commit_msg>do not show irrelevant warning<commit_after>////////////////////////////////////////////////////////////////////////////////
/// @brief datafiles
///
/// @file
///
/// DISCLAIMER
///
/// Copyright 2010-2011 triagens GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
/// you may not use this file except in compliance with the License.
/// You may obtain a copy of the License at
///
///     http://www.apache.org/licenses/LICENSE-2.0
///
/// Unless required by applicable law or agreed to in writing, software
/// distributed under the License is distributed on an "AS IS" BASIS,
/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
/// See the License for the specific language governing permissions and
/// limitations under the License.
///
/// Copyright holder is triAGENS GmbH, Cologne, Germany
///
/// @author Dr. <NAME>
/// @author Copyright 2011, triagens GmbH, Cologne, Germany
////////////////////////////////////////////////////////////////////////////////

#ifdef _WIN32
#include <BasicsC/win-utils.h>
#endif

#include "datafile.h"

#include "BasicsC/hashes.h"
#include "BasicsC/logging.h"
#include "BasicsC/memory-map.h"
#include "BasicsC/strings.h"
#include "BasicsC/files.h"


// #define DEBUG_DATAFILE 1

// -----------------------------------------------------------------------------
// --SECTION--                                                 private functions
// -----------------------------------------------------------------------------

////////////////////////////////////////////////////////////////////////////////
/// @addtogroup VocBase
/// @{
////////////////////////////////////////////////////////////////////////////////

////////////////////////////////////////////////////////////////////////////////
/// @brief return whether the datafile is a physical file (true) or an
/// anonymous mapped region (false)
////////////////////////////////////////////////////////////////////////////////

static bool IsPhysicalDatafile (const TRI_datafile_t* const datafile) {
  return datafile->_filename != NULL;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief return the name of a datafile
////////////////////////////////////////////////////////////////////////////////

static const char* GetNameDatafile (const TRI_datafile_t* const datafile) {
  if (datafile->_filename == NULL) {
    // anonymous regions do not have a filename
    return "anonymous region";
  }

  // return name of the physical file
  return datafile->_filename;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief close a datafile
////////////////////////////////////////////////////////////////////////////////

static void CloseDatafile (TRI_datafile_t* const datafile) {
  assert(datafile->_state != TRI_DF_STATE_CLOSED);

  if (datafile->isPhysical(datafile)) {
    TRI_CLOSE(datafile->_fd);
  }

  datafile->_state = TRI_DF_STATE_CLOSED;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief destroy a datafile
////////////////////////////////////////////////////////////////////////////////

static void DestroyDatafile (TRI_datafile_t* const datafile) {
  if (datafile->_filename != NULL) {
    TRI_FreeString(TRI_CORE_MEM_ZONE, datafile->_filename);
  }
}

////////////////////////////////////////////////////////////////////////////////
/// @brief sync the data of a datafile
////////////////////////////////////////////////////////////////////////////////

static bool SyncDatafile (const TRI_datafile_t* const datafile, 
                          char const* begin, 
                          char const* end) {
  // TODO: remove
  void** mmHandle = NULL;

  if (datafile->_filename == NULL) {
    // anonymous regions do not need to be synced
    return true;
  }

  assert(datafile->_fd > 0);

  return TRI_msync(datafile->_fd, mmHandle, begin, end);
}

////////////////////////////////////////////////////////////////////////////////
/// @brief truncate the datafile to a specific length
////////////////////////////////////////////////////////////////////////////////

static int TruncateDatafile (TRI_datafile_t* const datafile, const off_t length) {
  if (datafile->isPhysical(datafile)) {
    // only physical files can be truncated
    return ftruncate(datafile->_fd, length);
  }

  // for anonymous regions, this is a non-op
  return 0;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief creates a new sparse datafile
///
/// returns the file descriptor or -1 if the file cannot be created
////////////////////////////////////////////////////////////////////////////////

static int CreateSparseFile (char const* filename, 
                             const TRI_voc_size_t maximalSize) {
  off_t offset;
  char zero;
  ssize_t res;
  int fd;

  // open the file
  fd = TRI_CREATE(filename, O_CREAT | O_EXCL | O_RDWR, S_IRUSR | S_IWUSR);

  if (fd < 0) {
    TRI_set_errno(TRI_ERROR_SYS_ERROR);

    LOG_ERROR("cannot create datafile '%s': %s", filename, TRI_last_error());
    return -1;
  }

  // create sparse file
  offset = TRI_LSEEK(fd, maximalSize - 1, SEEK_SET);

  if (offset == (off_t) -1) {
    TRI_set_errno(TRI_ERROR_SYS_ERROR);
    TRI_CLOSE(fd);
    
    // remove empty file
    TRI_UnlinkFile(filename);

    LOG_ERROR("cannot seek in datafile '%s': '%s'", filename, TRI_last_error());
    return -1;
  }

  zero = 0;
  res = TRI_WRITE(fd, &zero, 1);

  if (res < 0) {
    TRI_set_errno(TRI_ERROR_SYS_ERROR);
    TRI_CLOSE(fd);
    
    // remove empty file
    TRI_UnlinkFile(filename);

    LOG_ERROR("cannot create sparse datafile '%s': '%s'", filename, TRI_last_error());
    return -1;
  }

  return fd;
}


////////////////////////////////////////////////////////////////////////////////
/// @brief initialises a datafile
////////////////////////////////////////////////////////////////////////////////

static void InitDatafile (TRI_datafile_t* datafile,
                          char* filename,
                          int fd,
                          void* mmHandle,
                          TRI_voc_size_t maximalSize,
                          TRI_voc_size_t currentSize,
                          TRI_voc_fid_t tick,
                          char* data) {

  // filename is a string for physical datafiles, and NULL for anonymous regions
  // fd is a positive value for physical datafiles, and -1 for anonymous regions
  if (filename == NULL) {
    assert(fd == -1);
  }
  else {
    assert(fd > 0);
  }

  datafile->_state       = TRI_DF_STATE_READ;
  datafile->_fid         = tick;

  datafile->_filename    = filename;
  datafile->_fd          = fd;
  datafile->_mmHandle    = mmHandle;

  datafile->_maximalSize = maximalSize;
  datafile->_currentSize = currentSize;
  datafile->_footerSize  = sizeof(TRI_df_footer_marker_t);

  datafile->_isSealed    = false;
  datafile->_lastError   = TRI_ERROR_NO_ERROR;

  datafile->_full        = false;

  datafile->_data        = data;
  datafile->_next        = data + currentSize;

  datafile->_synced      = data;
  datafile->_written     = NULL;

  // initialise function pointers
  datafile->isPhysical   = &IsPhysicalDatafile;
  datafile->getName      = &GetNameDatafile;
  datafile->close        = &CloseDatafile;
  datafile->destroy      = &DestroyDatafile;
  datafile->sync         = &SyncDatafile;
  datafile->truncate     = &TruncateDatafile;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief truncates a datafile
///
/// Create a truncated datafile, seal it and rename the old.
////////////////////////////////////////////////////////////////////////////////

static int TruncateAndSealDatafile (TRI_datafile_t* datafile, 
                                    TRI_voc_size_t vocSize) {
  char* filename;
  char* oldname;
  char zero;
  int fd;
  int res;
  size_t maximalSize;
  size_t offset;
  void* data;
  void* mmHandle;
  
  // this function must not be called for non-physical datafiles
  assert(datafile->isPhysical(datafile));
    
  // use multiples of page-size
  maximalSize = ((vocSize + sizeof(TRI_df_footer_marker_t) + PageSize - 1) / PageSize) * PageSize;

  // sanity check
  if (sizeof(TRI_df_header_marker_t) + sizeof(TRI_df_footer_marker_t) > maximalSize) {
    LOG_ERROR("cannot create datafile '%s', maximal size '%u' is too small", datafile->getName(datafile), (unsigned int) maximalSize);
    return TRI_set_errno(TRI_ERROR_ARANGO_MAXIMAL_SIZE_TOO_SMALL);
  }

  // open the file
  filename = TRI_Concatenate2String(datafile->_filename, ".new");

  fd = TRI_CREATE(filename, O_CREAT | O_EXCL | O_RDWR, S_IRUSR | S_IWUSR);

  if (fd < 0) {
    LOG_ERROR("cannot create new datafile '%s': '%s'", filename, TRI_last_error());
    return TRI_set_errno(TRI_ERROR_SYS_ERROR);
  }

  // create sparse file
  offset = TRI_LSEEK(fd, maximalSize - 1, SEEK_SET);

  if (offset == (off_t) -1) {
    TRI_set_errno(TRI_ERROR_SYS_ERROR);
    TRI_CLOSE(fd);

    // remove empty file
    TRI_UnlinkFile(filename);

    LOG_ERROR("cannot seek in new datafile '%s': '%s'", filename, TRI_last_error());
    return TRI_ERROR_SYS_ERROR;
  }

  zero = 0;
  res = TRI_WRITE(fd, &zero, 1);

  if (res < 0) {
    TRI_set_errno(TRI_ERROR_SYS_ERROR);
    TRI_CLOSE(fd);
    
    // remove empty file
    TRI_UnlinkFile(filename);

    LOG_ERROR("cannot create sparse datafile '%s': '%s'", filename, TRI_last_error());
    return TRI_ERROR_SYS_ERROR;
  }

  // memory map the data
  res = TRI_MMFile(0, maximalSize, PROT_WRITE | PROT_READ, MAP_SHARED, fd, &mmHandle, 0, &data);
  
  if (res != TRI_ERROR_NO_ERROR) {  
    TRI_set_errno(res);
    TRI_CLOSE(fd);

    // remove empty file
    TRI_UnlinkFile(filename);

    LOG_ERROR("cannot memory map file '%s': '%s'", filename, TRI_last_error());
    return TRI_errno();
  }

  // copy the data
  memcpy(data, datafile->_data, vocSize);

  // patch the datafile structure
  res = TRI_UNMMFile(datafile->_data, datafile->_maximalSize, datafile->_fd, &(datafile->_mmHandle));

  if (res < 0) {
    LOG_ERROR("munmap failed with: %d", res);
    return res;
  }

  // .............................................................................................
  // For windows: Mem mapped files use handles
  // the datafile->_mmHandle handle object has been closed in the underlying 
  // TRI_UNMMFile(...) call above so we do not need to close it for the associated file below
  // .............................................................................................
  
  TRI_CLOSE(datafile->_fd);

  
  datafile->_data = data;
  datafile->_next = (char*)(data) + vocSize;
  datafile->_maximalSize = maximalSize;
  datafile->_fd = fd;
  datafile->_mmHandle = mmHandle;
  
  // rename files
  oldname = TRI_Concatenate2String(datafile->_filename, ".corrupted");

  res = TRI_RenameFile(datafile->_filename, oldname);

  if (res != TRI_ERROR_NO_ERROR) {
    TRI_FreeString(TRI_CORE_MEM_ZONE, filename);
    TRI_FreeString(TRI_CORE_MEM_ZONE, oldname);

    return res;
  }

  res = TRI_RenameFile(filename, datafile->_filename);

  if (res != TRI_ERROR_NO_ERROR) {
    TRI_FreeString(TRI_CORE_MEM_ZONE, filename);
    TRI_FreeString(TRI_CORE_MEM_ZONE, oldname);

    return res;
  }

  TRI_SealDatafile(datafile);
  return TRI_ERROR_NO_ERROR;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief scans a datafile
////////////////////////////////////////////////////////////////////////////////

static TRI_df_scan_t ScanDatafile (TRI_datafile_t const* datafile) {
  TRI_df_scan_t scan;
  TRI_df_scan_entry_t entry;
  TRI_voc_size_t currentSize;
  char* end;
  char* ptr;

  // this function must not be called for non-physical datafiles
  assert(datafile->isPhysical(datafile));

  ptr = datafile->_data;
  end = datafile->_data + datafile->_currentSize;
  currentSize = 0;

  TRI_InitVector(&scan._entries, TRI_CORE_MEM_ZONE, sizeof(TRI_df_scan_entry_t));

  scan._currentSize = datafile->_currentSize;
  scan._maximalSize = datafile->_maximalSize;
  scan._numberMarkers = 0;
  scan._status = 1;

  if (datafile->_currentSize == 0) {
    end = datafile->_data + datafile->_maximalSize;
  }

  while (ptr < end) {
    TRI_df_marker_t* marker = (TRI_df_marker_t*) ptr;
    bool ok;
    size_t size;

    memset(&entry, 0, sizeof(entry));

    entry._position = ptr - datafile->_data;
    entry._size = marker->_size;
    entry._tick = marker->_tick;
    entry._type = marker->_type;
    entry._status = 1;

    if (marker->_size == 0 && marker->_crc == 0 && marker->_type == 0 && marker->_tick == 0) {
      entry._status = 2;

      scan._endPosition = currentSize;

      TRI_PushBackVector(&scan._entries, &entry);
      return scan;
    }

    ++scan._numberMarkers;

    if (marker->_size == 0) {
      entry._status = 3;

      scan._status = 2;
      scan._endPosition = currentSize;

      TRI_PushBackVector(&scan._entries, &entry);
      return scan;
    }

    if (marker->_size < sizeof(TRI_df_marker_t)) {
      entry._status = 4;

      scan._endPosition = currentSize;
      scan._status = 3;

      TRI_PushBackVector(&scan._entries, &entry);
      return scan;
    }

    ok = TRI_CheckCrcMarkerDatafile(marker);

    if (! ok) {
      entry._status = 5;
      scan._status = 4;
    }

    TRI_PushBackVector(&scan._entries, &entry);

    size = TRI_DF_ALIGN_BLOCK(marker->_size);
    currentSize += size;

    if (marker->_type == TRI_DF_MARKER_FOOTER) {
      scan._endPosition = currentSize;
      return scan;
    }

    ptr += size;
  }

  return scan;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief checks a datafile
////////////////////////////////////////////////////////////////////////////////

static bool CheckDatafile (TRI_datafile_t* datafile) {
  TRI_voc_size_t currentSize;
  char* end;
  char* ptr;
  
  // this function must not be called for non-physical datafiles
  assert(datafile->isPhysical(datafile));

  ptr = datafile->_data;
  end = datafile->_data + datafile->_currentSize;
  currentSize = 0;

  if (datafile->_currentSize == 0) {
    LOG_WARNING("current size is 0 in read-only datafile '%s', trying to fix", datafile->getName(datafile));

    end = datafile->_data + datafile->_maximalSize;
  }

  while (ptr < end) {
    TRI_df_marker_t* marker = (TRI_df_marker_t*) ptr;
    bool ok;
    size_t size;

#ifdef DEBUG_DATAFILE
    LOG_TRACE("MARKER: size %lu, tick %lx, crc %lx, type %u",
              (unsigned long) marker->_size,
              (unsigned long) marker->_tick,
              (unsigned long) marker->_crc,
              (unsigned int) marker->_type);
#endif

    if (marker->_size == 0 && marker->_crc == 0 && marker->_type == 0 && marker->_tick == 0) {
      LOG_DEBUG("reached end of datafile '%s' data, current size %lu", 
                datafile->getName(datafile), 
                (unsigned long) currentSize);

      datafile->_currentSize = currentSize;
      datafile->_next = datafile->_data + datafile->_currentSize;

      return true;
    }

    if (marker->_size == 0) {
      LOG_DEBUG("reached end of datafile '%s' data, current size %lu", 
                datafile->getName(datafile), 
                (unsigned long) currentSize);

      datafile->_currentSize = currentSize;
      datafile->_next = datafile->_data + datafile->_currentSize;

      return true;
    }

    if (marker->_size < sizeof(TRI_df_marker_t)) {
      datafile->_lastError = TRI_set_errno(TRI_ERROR_ARANGO_CORRUPTED_DATAFILE);
      datafile->_currentSize = currentSize;
      datafile->_next = datafile->_data + datafile->_currentSize;
      datafile->_state = TRI_DF_STATE_OPEN_ERROR;

      LOG_WARNING("marker in datafile '%s' too small, size %lu, should be at least %lu",
                  datafile->getName(datafile),
                  (unsigned long) marker->_size,
                  (unsigned long) sizeof(TRI_df_marker_t));

      return false;
    }

    ok = TRI_CheckCrcMarkerDatafile(marker);

    if (! ok) {
      datafile->_lastError = TRI_set_errno(TRI_ERROR_ARANGO_CORRUPTED_DATAFILE);
      datafile->_currentSize = currentSize;
      datafile->_next = datafile->_data + datafile->_currentSize;
      datafile->_state = TRI_DF_STATE_OPEN_ERROR;

      LOG_WARNING("crc mismatch found in datafile '%s'", datafile->getName(datafile));

      return false;
    }

    TRI_UpdateTickVocBase(marker->_tick);

    size = TRI_DF_ALIGN_BLOCK(marker->_size);
    currentSize += size;

    if (marker->_type == TRI_DF_MARKER_FOOTER) {
      LOG_DEBUG("found footer, reached end of datafile '%s', current size %lu",
                datafile->getName(datafile),
                (unsigned long) currentSize);

      datafile->_isSealed = true;
      datafile->_currentSize = currentSize;
      datafile->_next = datafile->_data + datafile->_currentSize;

      return true;
    }

    ptr += size;
  }

  return true;
}

////////////////////////////////////////////////////////////////////////////////
/// @brief opens a datafile
////////////////////////////////////////////////////////////////////////////////

static TRI_datafile_t* OpenDatafile (char const* filename, bool ignoreErrors) {
  TRI_datafile_t* datafile;
  TRI_voc_size_t size;
  bool ok;
  void* data;
  char* ptr;
  int fd;
  int res;
  ssize_t len;
  struct stat status;
  TRI_df_header_marker_t header;
  void* mmHandle;
  
  // this function must not be called for non-physical datafiles
  assert(filename != NULL);

  // ..........................................................................  
  // attempt to open a datafile file
  // ..........................................................................  

  fd = TRI_OPEN(filename, O_RDWR);

  if (fd < 0) {
    TRI_set_errno(TRI_ERROR_SYS_ERROR);

    LOG_ERROR("cannot open datafile '%s': '%s'", filename, TRI_last_error());

    return NULL;
  }


  // compute the size of the file
  res = fstat(fd, &status);

  if (res < 0) {
    TRI_set_errno(TRI_ERROR_SYS_ERROR);
    TRI_CLOSE(fd);

    LOG_ERROR("cannot get status of datafile '%s': %s", filename, TRI_last_error());

    return NULL;
  }

  // check that file is not too small
  size = status.st_size;

  if (size < sizeof(TRI_df_header_marker_t) + sizeof(TRI_df_footer_marker_t)) {
    TRI_set_errno(TRI_ERROR_ARANGO_CORRUPTED_DATAFILE);
    TRI_CLOSE(fd);

    LOG_ERROR("datafile '%s' is corrupted, size is only %u", filename, (unsigned int) size);

    return NULL;
  }

  // read header from file
  ptr = (char*) &header;
  len = sizeof(TRI_df_header_marker_t);

  ok = TRI_ReadPointer(fd, ptr, len);

  if (! ok) {
    LOG_ERROR("cannot read datafile header from '%s': %s", filename, TRI_last_error());

    TRI_CLOSE(fd);
    return NULL;
  }

  // check CRC
  ok = TRI_CheckCrcMarkerDatafile(&header.base);

  if (! ok) {
    TRI_set_errno(TRI_ERROR_ARANGO_CORRUPTED_DATAFILE);

    LOG_ERROR("corrupted datafile header read from '%s'", filename);

    if (! ignoreErrors) {
      TRI_CLOSE(fd);
      return NULL;
    }
  }

  // check the datafile version
  if (ok) {
    if (header._version != TRI_DF_VERSION) {
      TRI_set_errno(TRI_ERROR_ARANGO_CORRUPTED_DATAFILE);

      LOG_ERROR("unknown datafile version '%u' in datafile '%s'",
                (unsigned int) header._version,
                filename);

      if (! ignoreErrors) {
        TRI_CLOSE(fd);
        return NULL;
      }
    }
  }

  // check the maximal size
  if (size > header._maximalSize) {
    LOG_DEBUG("datafile '%s' has size '%u', but maximal size is '%u'",
              filename,
              (unsigned int) size,
              (unsigned int) header._maximalSize);
  }

  // map datafile into memory
  res = TRI_MMFile(0, size, PROT_READ, MAP_SHARED, fd, &mmHandle, 0, &data);
  
  if (res != TRI_ERROR_NO_ERROR) {
    TRI_set_errno(res);
    TRI_CLOSE(fd);

    LOG_ERROR("cannot memory map datafile '%s': '%d'", filename, res);
    return NULL;
  }

  // create datafile structure
  datafile = TRI_Allocate(TRI_UNKNOWN_MEM_ZONE, sizeof(TRI_datafile_t), false);

  if (datafile == NULL) {
    return NULL;
  }

  InitDatafile(datafile,
               TRI_DuplicateString(filename),
{'fraction_non_alphanumeric': 0.2365883136423057, 'fraction_numerical': 0.0037491058875761536, 'mean_word_length': 1.7897887566228583, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 267, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6783789', 'n_tokens_mistral': 12344, 'n_tokens_neox': 11688, 'n_words': 2915}