file.cpp

FastDb是高效的内存数据库系统

    status = GetLastError();

    if (cmh == NULL)
    {
      CloseHandle(cfh);
      return status;
    }

    diskUpdateCount = (int*)MapViewOfFile(cmh, FILE_MAP_ALL_ACCESS,
                                          0, 0, 0);

    if (diskUpdateCount == NULL)
    {
      status = GetLastError();
      CloseHandle(cmh);
      CloseHandle(cfh);
      return status;
    }

    rootPage = dbMalloc(pageSize);
    int maxCount = 0;

    for (int i = 0; i < nPages; i++)
    {
      int count = diskUpdateCount[i];
      currUpdateCount[i] = count;

      if (count > maxCount)
      {
        maxCount = count;
      }
    }

    updateCounter = maxCount;
    nRecovered = 0;
    recoveredEvent.open(true);
    syncEvent.open();
    startSync();
  }

#endif
  return ok;
}

#if defined(REPLICATION_SUPPORT)
void dbFile::syncToDisk()
{
  syncThread.setPriority(dbThread::THR_PRI_LOW);
  dbCriticalSection cs(syncCS);

  while (doSync)
  {
    int i, j, k;
    int maxUpdated = 0;

    for (i = 0; i < int(mmapSize >> dbModMapBlockBits);)
    {
      int updateCounters[dbMaxSyncSegmentSize];

      for (j=i; j < int(mmapSize >> dbModMapBlockBits) && j-i < dbMaxSyncSegmentSize
           && currUpdateCount[j] > diskUpdateCount[j]; j++)
      {
        updateCounters[j-i] = currUpdateCount[j];
      }

      if (i != j)
      {
        size_t pos = (i << dbModMapBlockBits) & ~(pageSize-1);
        size_t size = (((j-i) << dbModMapBlockBits) + pageSize - 1) & ~(pageSize-1);
#ifdef NO_MMAP

        DWORD written;

        if (SetFilePointer(fh, pos, NULL, FILE_BEGIN) != pos ||
            !WriteFile(fh, mmapAddr + pos, size, &written, NULL)
            || written != (DWORD)size)
        {
          dbTrace("Failed to save page to the disk, position=%ld, size=%ld, error=%d\n",
                  (long)pos, (long)size, GetLastError());
        }

#else
        FlushViewOfFile(mmapAddr + pos, size);

#endif

        for (k = 0; i < j; k++, i++)
        {
          diskUpdateCount[i] = updateCounters[k];
        }

        maxUpdated = i;
      }
      else
      {
        i += 1;
      }

      if (!doSync)
      {
        return;
      }
    }

    if (maxUpdated != 0)
    {
      FlushViewOfFile(diskUpdateCount, maxUpdated*sizeof(int));
    }

    if (closing && maxUpdated == 0)
    {
      return;
    }
    else
    {
      syncEvent.wait(syncCS, dbSyncTimeout);
    }
  }
}

#endif

int dbFile::create(const char* name, bool noBuffering)
{
  fh = CreateFile(name, GENERIC_READ|GENERIC_WRITE, 0, FASTDB_SECURITY_ATTRIBUTES, CREATE_ALWAYS,
                  (noBuffering ? FILE_FLAG_NO_BUFFERING : 0)|FILE_FLAG_SEQUENTIAL_SCAN, NULL);

  if (fh == INVALID_HANDLE_VALUE)
  {
    return GetLastError();
  }

  mh = NULL;
  mmapAddr = NULL;
  sharedName = NULL;
  return ok;
}

int dbFile::read(void* buf, size_t& readBytes, size_t size)
{
  DWORD count;

  if (ReadFile(fh, buf, size, &count, NULL))
  {
    readBytes = count;
    return ok;
  }
  else
  {
    readBytes = 0;
    return GetLastError();
  }
}

int dbFile::write(void const* buf, size_t& writtenBytes, size_t size)
{
  DWORD count;

  if (WriteFile(fh, buf, size, &count, NULL))
  {
    writtenBytes = count;
    return ok;
  }
  else
  {
    writtenBytes = 0;
    return GetLastError();
  }
}


int dbFile::flush(bool physical)
{
#if defined(REPLICATION_SUPPORT)
  dbCriticalSection cs(replCS);

  if (db == NULL)
  {
    physical = true;
  }

  if (!physical)
  {
    updateCounter += 1;
  }

#endif
#if defined(REPLICATION_SUPPORT) || (defined(NO_MMAP) && !defined(DISKLESS_CONFIGURATION))
  int* map = pageMap;

  for (int i = 0, n = pageMapSize; i < n; i++)
  {
    if (map[i] != 0)
    {
      size_t pos = (size_t)i << (dbModMapBlockBits + 5);
      unsigned mask = map[i];
      int count = 0;

      do
      {
        int size = 0;

        while ((mask & 1) == 0)
        {
          pos += dbModMapBlockSize;
          mask >>= 1;
          count += 1;
        }

        while (true)
        {
          do
          {
#ifdef REPLICATION_SUPPORT

            if (!physical)
            {
              currUpdateCount[(pos + size) >> dbModMapBlockBits] = updateCounter;
            }

#endif
            size += dbModMapBlockSize;

            mask >>= 1;

            count += 1;
          }
          while ((mask & 1) != 0);

          if (i+1 < n && count == 32 && size < dbMaxSyncSegmentSize*dbModMapBlockSize
              && (map[i+1] & 1) != 0)
          {
            map[i] = 0;
            mask = map[++i];
            count = 0;
          }
          else
          {
            break;
          }
        }

#if defined(REPLICATION_SUPPORT)
        if (db != NULL)
        {
          if (!physical)
          {
            for (int j = db->nServers; --j >= 0;)
            {
              if (db->con[j].status == dbReplicatedDatabase::ST_STANDBY)
              {
                ReplicationRequest rr;
                rr.op = ReplicationRequest::RR_UPDATE_PAGE;
                rr.nodeId = db->id;
                rr.page.updateCount = updateCounter;
                rr.page.offs = pos;
                rr.size = size;
                db->writeReq(j, rr, mmapAddr + pos, size);
              }
            }
          }

          pos += size;
          continue;
        }

#endif
#ifndef DISKLESS_CONFIGURATION
        DWORD written;

        if (SetFilePointer(fh, pos, NULL, FILE_BEGIN) != pos ||
            !WriteFile(fh, mmapAddr + pos, size, &written, NULL)
            || written != (DWORD)size)
        {
          return GetLastError();
        }

#endif
        pos += size;
      }
      while (mask != 0);

      map[i] = 0;
    }
  }

#endif
#if !defined(NO_MMAP) && !defined(DISKLESS_CONFIGURATION) && !defined(REPLICATION_SUPPORT)
  if (!FlushViewOfFile(mmapAddr, mmapSize))
  {
    return GetLastError();
  }

#endif
  return ok;
}

int dbFile::setSize(size_t size, char const* sharedName, bool initialize)
{
#if defined(REPLICATION_SUPPORT)
  dbCriticalSection cs1(syncCS);
  dbCriticalSection cs2(replCS);
#endif
#ifdef DISKLESS_CONFIGURATION

  assert(false);
#else
#ifdef NO_MMAP

  char* newBuf = (char*)VirtualAlloc(NULL, size, MEM_COMMIT|MEM_RESERVE, PAGE_READWRITE);

  if (newBuf == NULL)
  {
    return GetLastError();
  }

  if (SetFilePointer(fh, size, NULL, FILE_BEGIN) != size || !SetEndOfFile(fh))
  {
    return GetLastError();
  }

  memcpy(newBuf, mmapAddr, mmapSize);
  VirtualFree(mmapAddr, 0, MEM_RELEASE);
  mmapAddr = newBuf;
  mmapSize = size;
#else

  if (!UnmapViewOfFile(mmapAddr) || !CloseHandle(mh))
  {
    return GetLastError();
  }

  mh = CreateFileMapping(fh, FASTDB_SECURITY_ATTRIBUTES, PAGE_READWRITE, 0, size, sharedName);
  int status = GetLastError();

  if (mh == NULL)
  {
    printf("CreateFileMapping failed: %d\n", status);
    return status;
  }

  mmapAddr = (char*)MapViewOfFile(mh, FILE_MAP_ALL_ACCESS, 0, 0, 0);

  if (mmapAddr == NULL)
  {
    return GetLastError();
  }

  if (initialize && status != ERROR_ALREADY_EXISTS)
    //&& osinfo.dwPlatformId != VER_PLATFORM_WIN32_NT)
  {
    // Windows 95 doesn't initialize pages
    memset(mmapAddr+mmapSize, 0, size - mmapSize);
  }

  mmapSize = size;
#endif
#if defined(NO_MMAP) || defined(REPLICATION_SUPPORT)

  int newPageMapSize = (size + dbModMapBlockSize*32 - 1) >> (dbModMapBlockBits + 5);
  int* newPageMap = new int[newPageMapSize];
  memcpy(newPageMap, pageMap, pageMapSize*sizeof(int));
  memset(newPageMap + pageMapSize, 0,
         (newPageMapSize-pageMapSize)*sizeof(int));
  delete[] pageMap;
  pageMapSize = newPageMapSize;
  pageMap = newPageMap;
#endif

#endif

  return ok;
}

int dbFile::close()
{
  delete[] sharedName;
#if defined(REPLICATION_SUPPORT)

  if (db != NULL)
  {
    closing = true;
    stopSync();
    {
      dbCriticalSection cs(replCS);

      if (nRecovered != 0)
      {
        recoveredEvent.wait(replCS);
      }
    }

    syncEvent.close();
    recoveredEvent.close();
    UnmapViewOfFile(diskUpdateCount);
    CloseHandle(cmh);
    CloseHandle(cfh);
  }

  delete[] currUpdateCount;
  currUpdateCount = NULL;
  dbFree(rootPage);
  rootPage = NULL;
#endif

  if (mmapAddr != NULL)
  {
#if defined(NO_MMAP)
    int rc = flush();

    if (rc != ok)
    {
      return rc;
    }

    VirtualFree(mmapAddr, 0, MEM_RELEASE);
    delete[] pageMap;
#else

    if (!UnmapViewOfFile(mmapAddr))
    {
      return GetLastError();
    }

#if defined(REPLICATION_SUPPORT)
    delete[] pageMap;

#endif
#endif

  }

  if (mh != NULL)
  {
    if (!CloseHandle(mh))
    {
      return GetLastError();
    }
  }

  return fh == INVALID_HANDLE_VALUE || CloseHandle(fh) ? ok : GetLastError();
}

char* dbFile::errorText(int code, char* buf, size_t bufSize)
{
#ifndef PHAR_LAP
  int len = FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM,
                          NULL,
                          code,
                          0,
                          buf,
                          bufSize,
                          NULL);

  if (len == 0)
  {
    char errcode[64];
    sprintf(errcode, "unknown error code %u", code);
    strncpy(buf, errcode, bufSize);
  }

#else
  char errcode[64];

  sprintf(errcode, "unknown error code %u", code);

  strncpy(buf, errcode, bufSize);

#endif

  return buf;
}

#else // Unix

#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <sys/stat.h>
#ifndef NO_MMAP
#include <sys/mman.h>
#endif

#ifndef O_SYNC
#define O_SYNC O_FSYNC
#endif

#ifndef O_DSYNC
#define O_DSYNC O_SYNC
#endif


int dbFile::open(char const* name, char const*, bool readonly, size_t initSize, bool replicationSupport)
{
#if defined(USE_SYSV_SHARED_MEMORY) && defined(DISKLESS_CONFIGURATION)

  if (!shmem.open(name, initSize))
  {
    return errno;
  }

  mmapSize = initSize;
  mmapAddr = shmem.get_base();
  fd = -1;
#else
#ifndef NO_MMAP

int mmap_attr = MAP_SHARED;
#endif

int status;
#ifdef DISKLESS_CONFIGURATION
#ifndef MAP_ANONYMOUS

fd = ::open("/dev/zero", O_RDWR, 0);
#else

fd = -1;
mmap_attr |= MAP_ANONYMOUS;
#endif // MAP_ANONYMOUS

mmapSize = initSize;
#else // DISKLESS_CONFIGURATION

fd = ::open(name, readonly ? O_RDONLY : O_RDWR/*|O_DSYNC*/|O_CREAT, 0666);

if (fd < 0)
{
  return errno;
}

mmapSize = lseek(fd, 0, SEEK_END);

if (!readonly && mmapSize < initSize)
{
  mmapSize = initSize;

  if (ftruncate(fd, mmapSize) != ok)
  {
    status = errno;

    if (fd >= 0)
    {
      ::close(fd);
    }

    return status;
  }
}

#endif // DISKLESS_CONFIGURATION
#ifdef NO_MMAP
size_t fileSize = mmapSize;

if (!readonly && mmapSize < initSize)
{
  mmapSize = initSize;