file.cpp

最新版本！fastdb是高效的内存数据库系统

int dbFile::read(void* buf, size_t& readBytes, size_t size)
{  
    char* dst = (char*)buf;
    readBytes = 0;
    while (size != 0) { 
        DWORD count;
        size_t quant = size > MAX_READ_BUFFER_SIZE ? MAX_READ_BUFFER_SIZE : size;
        if (!ReadFile(fh, dst, quant, &count, NULL)) { 
            return GetLastError();
        } else { 
            dst += count;
            size -= count;
            readBytes += count;
            if (count != quant) { 
                break;
            }
        }
    }
    return ok;
}

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 (size_t i = 0, n = pageMapSize; i < n; i++) { 
        if (map[i] != 0) { 
            size_t pos = i << (dbModMapBlockBits + 5);
            unsigned mask = map[i];
            int count = 0;
            do { 
                size_t 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
#ifdef FUZZY_CHECKPOINT
                writer->put(pos, mmapAddr + pos, size);
#else
                if (!write(pos, mmapAddr + pos, size)) { 
                    return GetLastError();
                }
#endif
#endif
                pos += size;
            } while (mask != 0);
            map[i] = 0;
        }
    }
#endif
#if !defined(NO_MMAP) && !defined(DISKLESS_CONFIGURATION) && !defined(REPLICATION_SUPPORT) && !defined(NO_FLUSH_ON_COMMIT)
    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, readonly ? PAGE_READONLY : PAGE_READWRITE, 
                           0, size, W32_STRING(sharedName));
    int status = GetLastError();
    if (mh == NULL) { 
        printf("CreateFileMapping failed: %d\n", status);
        return status;
    }
    mmapAddr = (char*)MapViewOfFile(mh, readonly ? FILE_MAP_READ : 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;
        }
#ifdef FUZZY_CHECKPOINT
        delete writer;
#endif
        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  
#if defined(_WINCE) || defined(UNICODE)
    wchar_t cnvBuf[CNV_BUF_SIZE];
    int len = FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM,
                            NULL,
                            code,
                            0,
                            cnvBuf,
                            CNV_BUF_SIZE-1,
                            NULL);
	cnvBuf[CNV_BUF_SIZE-1] = '\0';
	len = wcstombs(buf, cnvBuf, bufSize);
#else
    int len = FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM,
                            NULL,
                            code,
                            0,
                            buf,
                            bufSize-1,
                            NULL);
#endif
    if (len == 0) { 
        char errcode[64];
        sprintf(errcode, "unknown error code %u", code);
        strncpy(buf, errcode, bufSize-1);
    }
#else
    char errcode[64];
    sprintf(errcode, "unknown error code %u", code);
    strncpy(buf, errcode, bufSize-1);
#endif
    buf[bufSize-1] = '\0';
    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

#ifndef O_DIRECT
#define O_DIRECT 0
#endif


#ifdef PROTECT_DATABASE
void dbFile::protect(size_t pos, size_t size)
{
    int rc = mprotect(mmapAddr + pos, DOALIGN(size, pageSize), PROT_READ);
    assert(rc == 0);
}

void dbFile::unprotect(size_t pos, size_t size)
{
    int rc = mprotect(mmapAddr + pos, DOALIGN(size, pageSize), PROT_READ|PROT_WRITE);
    assert(rc == 0);
}
#endif

int dbFile::open(char const* name, char const*, bool readonly, size_t initSize, bool replicationSupport)
{
    this->readonly = readonly;
    fd = -1;
#if defined(USE_SYSV_SHARED_MEMORY) && defined(DISKLESS_CONFIGURATION)
    if (!shmem.open(name, initSize)) { 
        return errno;
    }
    mmapSize = initSize;           
    mmapAddr = shmem.get_base();
#else
#ifndef NO_MMAP
    int mmap_attr = MAP_SHARED;
#endif
    int status;
#ifdef DISKLESS_CONFIGURATION
#ifndef NO_MMAP
#ifndef MAP_ANONYMOUS 
    fd = ::open("/dev/zero", O_RDWR, 0);
#else
    mmap_attr |= MAP_ANONYMOUS;
#endif // MAP_ANONYMOUS 
#endif // NO_MMAP
    mmapSize = initSize;
#else // DISKLESS_CONFIGURATION
    int open_flags = readonly ? O_RDONLY : O_RDWR/*|O_DSYNC*/|O_CREAT
#ifdef NO_MMAP
                |O_DIRECT
#endif
	;
#ifdef VXWORKS
    // Check if the file exists
    bool bFileExists = true;
    struct stat statBuf;
    if (::stat(name, &statBuf) != OK)
    {
        if (errno == ENOENT || errno == ENOTDIR || errno == ENAMETOOLONG || errno == ENODEV)
            bFileExists = false;
        // if we got some other error - lets assume the file exists...
    }
    if (bFileExists)
        open_flags &= ~O_CREAT;
#endif // VXWORKS
    fd = ::open(name, open_flags, 0666);
    if (fd < 0) { 
     	int orig_errno = errno;
        dbTrace("failed opening file '%s' - fd - %d, errno - %d\n",
            name, fd, orig_errno);
        return orig_errno;
    }
#ifdef UNIX
    set_fd_close_on_exec(fd);
#endif // UNIX

#if defined(__sun)
    directio(fd, DIRECTIO_ON);
#endif
    mmapSize = lseek(fd, 0, SEEK_END); 
    if (!readonly && mmapSize == 0) { 
        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;
    }
#ifdef USE_SYSV_SHARED_MEMORY
    if (!shmem.open(name, mmapSize)) { 
        status = errno;
        if (fd >= 0) { 
            ::close(fd);
        }
        return status;
    }
    mmapAddr = shmem.get_base();
#else
    mmapAddr = (char*)valloc(mmapSize);
    if (mmapAddr == NULL) { 
        status = errno;
        if (fd >= 0) { 
            ::close(fd);
        }
        return status;
    }
#endif
    lseek(fd, 0, SEEK_SET); 
    size_t readBytes = 0;
    if (read(mmapAddr, readBytes, fileSize) != ok || readBytes != fileSize) { 
#ifdef USE_SYSV_SHARED_MEMORY
        shmem.close();
#else
        free(mmapAddr);
#endif