silo.cpp

#include <cstdio>
#include <cstdlib>
#include <errno.h>
#include <algorithm>
#include <dirent.h>
#include <zlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <sys/types.h>
#include "lib/libsilo.h"
#include "lib/debug.h"
#include "silo.h"
using namespace std;
using namespace silo;

void segment_buffer::deserialize(const char *file)
{
    int fd = open(file, O_RDONLY);
    if (fd == -1) return;

    for (;;)
    {
        md5val hash;
        chunk_file_header header;
        if (read(fd, &hash[0], hash.size()) != hash.size() ||
            read(fd, &header, sizeof(header)) != sizeof(header))
            break;

        chunk chk;
        chk.type = header.type;
        chk.hash = hash;
        chk.rawsize = header.rawsize;
        size += chk.rawsize;

        uint32_t blobsize;
        if (read(fd, &blobsize, sizeof(blobsize)) != sizeof(blobsize)) break;

        chk.blob.resize(blobsize);
        if (read(fd, &chk.blob[0], blobsize) != blobsize) break;

        // to avoid packed field error
        auto refcount = header.refcount;
        chunks.emplace_hint(chunks.end(), hash, make_pair(refcount, move(chk)));
    }

    close(fd);
}

void segment_buffer::serialize(const char *file)
{
    int fd = open(file, O_WRONLY | O_CREAT, 0755);
    if (fd == -1) return;

    for (auto &&elem : chunks)
    {
        auto &&hash = elem.first;
        auto refcount = elem.second.first;
        auto &&chk = elem.second.second;

        chunk_file_header header;
        header.refcount = refcount;
        header.type = chk.type;
        header.rawsize = chk.rawsize;

        safe_write(fd, &hash[0], hash.size());
        safe_write(fd, &header, sizeof(header));

        auto blobsize = static_cast<uint32_t>(chk.blob.size());
        safe_write(fd, &blobsize, sizeof(blobsize));
        safe_write(fd, &chk.blob[0], blobsize);
    }

    close(fd);
}

void block_buffer::deserialize(const char *file)
{
    int fd = open(file, O_RDONLY);
    if (fd == -1) return;

    uint32_t segcount;
    if (read(fd, &segcount, sizeof(segcount)) != sizeof(segcount))
    {
        fprintf(stderr, "blkbuf deserialize fail\n");
        exit(1);
    }

    while (segcount--)
    {
        md5val hash;
        if (read(fd, &hash[0], hash.size()) != hash.size())
        {
            fprintf(stderr, "blkbuf deserialize fail\n");
            exit(1);
        }
    }

    for (;;)
    {
        md5val hash;
        chunk_file_header header;
        if (read(fd, &hash[0], hash.size()) != hash.size() ||
            read(fd, &header, sizeof(header)) != sizeof(header))
            break;

        chunk chk;
        chk.type = header.type;
        chk.hash = hash;
        chk.rawsize = header.rawsize;

        uint32_t blobsize;
        if (read(fd, &blobsize, sizeof(blobsize)) != sizeof(blobsize)) break;

        chk.blob.resize(blobsize);
        if (read(fd, &chk.blob[0], blobsize) != blobsize) break;

        // to avoid packed field error
        auto refcount = header.refcount;
        chunks.emplace_hint(chunks.end(), hash, make_pair(refcount, move(chk)));
    }

    close(fd);
}

void block_buffer::serialize(const char *file)
{
    int fd = open(file, O_WRONLY | O_CREAT, 0755);
    if (fd == -1) return;

    uint32_t segcount = segments.size();
    safe_write(fd, &segcount, sizeof(segcount));

    for (auto &&hash : segments)
    {
        safe_write(fd, &hash[0], hash.size());
    }

    for (auto &&elem : chunks)
    {
        auto &&hash = elem.first;
        auto refcount = elem.second.first;
        auto &&chk = elem.second.second;

        chunk_file_header header;
        header.refcount = refcount;
        header.type = chk.type;
        header.rawsize = chk.rawsize;

        safe_write(fd, &hash[0], hash.size());
        safe_write(fd, &header, sizeof(header));

        auto blobsize = static_cast<uint32_t>(chk.blob.size());
        safe_write(fd, &blobsize, sizeof(blobsize));
        safe_write(fd, &chk.blob[0], blobsize);
    }

    close(fd);
}

silofs::silofs(const char *basedir) : base(basedir)
{
    if (base.back() != '/') base.push_back('/');

    mkdir((base + volume_dir).c_str(), 0755);

    segbuf.deserialize((base + segbuf_file).c_str());
    blkbuf.deserialize((base + blkbuf_file).c_str());

    shtbl.load((base + shtable_file).c_str());

    string blkdir = base;
    size_t p = blkdir.size();
    blkdir.resize(blkdir.size() + 15);
    for (lastblock = 0;; lastblock++)
    {
        sprintf(&blkdir[p], "%d", lastblock);
        struct stat st;
        if (stat(blkdir.c_str(), &st) == -1) break;
        if (!S_ISDIR(st.st_mode))
        {
            fprintf(stderr, "%s is not a directory\n", blkdir.c_str());
            exit(1);
        }

        blocks.emplace_back(blkdir.c_str());
    }

    int pfd = open((base + pending_name).c_str(), O_RDONLY);
    if (pfd == -1) return;

    for (;;)
    {
        uint32_t namelen;
        if (::read(pfd, &namelen, sizeof(namelen)) != sizeof(namelen)) break;

        string name(namelen, char{});
        if (::read(pfd, &name[0], namelen) != namelen)
        {
            fprintf(stderr, "pending file error\n");
            exit(1);
        }

        auto &&pending = pendings[name];
        if (::read(pfd, &pending.header, sizeof(pending.header)) != sizeof(pending.header) ||
            ::read(pfd, &pending.lastseg, sizeof(pending.lastseg)) != sizeof(pending.lastseg) ||
            ::read(pfd, &pending.lastblk, sizeof(pending.lastblk)) != sizeof(pending.lastblk) ||
            ::read(pfd, &pending.left, sizeof(pending.left)) != sizeof(pending.left))
        {
            fprintf(stderr, "pending file error\n");
            exit(1);
        }

        uint32_t chkcnt;
        if (::read(pfd, &chkcnt, sizeof(chkcnt)) != sizeof(chkcnt))
        {
            fprintf(stderr, "pending file error\n");
            exit(1);
        }

        while (chkcnt--)
        {
            chunk_info info;
            if (::read(pfd, &info, sizeof(info)) != sizeof(info))
            {
                fprintf(stderr, "pending file error\n");
                exit(1);
            }

            pending.chunks.push_back(info);
        }
    }

    close(pfd);
}

silofs::~silofs()
{
    segbuf.serialize((base + segbuf_file).c_str());
    blkbuf.serialize((base + blkbuf_file).c_str());

    shtbl.save((base + shtable_file).c_str());
    
    int pfd = open((base + pending_name).c_str(), O_WRONLY | O_CREAT, 0755);
    if (pfd == -1) return;

    for (auto &&elem : pendings)
    {
        auto &&name = elem.first;
        auto &&pending = elem.second;
        uint32_t namelen = name.size();
        safe_write(pfd, &namelen, sizeof(namelen));
        safe_write(pfd, &name[0], namelen);

        safe_write(pfd, &pending.header, sizeof(pending.header));
        safe_write(pfd, &pending.lastseg, sizeof(pending.lastseg));
        safe_write(pfd, &pending.lastblk, sizeof(pending.lastblk));
        safe_write(pfd, &pending.left, sizeof(pending.left));

        uint32_t chkcnt = pending.chunks.size();
        safe_write(pfd, &chkcnt, sizeof(chkcnt));

        for (auto &&info : pending.chunks)
        {
            safe_write(pfd, &info, sizeof(info));
        }
    }

    close(pfd);
}

bool silofs::read(const char *file, vector<char> &ret)
{
    ret.clear();

    auto pit = pendings.find(file);
    if (pit != pendings.end())
    {
        for (auto &&info : pit->second.chunks)
        {
            auto blob = readchunk(info);
            ret.insert(ret.end(), blob.begin(), blob.end());
        }

        return true;
    }
    else
    {
        string meta = base + volume_dir + file;
        int fd = open(meta.c_str(), O_RDONLY);
        if (fd == -1)
        {
            return false;
        }

        file_header header;
        if (::read(fd, &header, sizeof(header)) != sizeof(header))
        {
            fprintf(stderr, "%s reading error\n", file);
            exit(1);
        }

        for (off_t i = 0, e = header.chunks; i < e; i++)
        {
            chunk_info info;
            if (::read(fd, &info, sizeof(info)) != sizeof(info))
            {
                fprintf(stderr, "%s reading error\n", file);
                exit(1);
            }

            auto blob = readchunk(info);
            ret.insert(ret.end(), blob.begin(), blob.end());
        }

        close(fd);

        return true;
    }
}


bool silofs::header(const char *file, file_header &header)
{
    auto pit = pendings.find(file);
    if (pit != pendings.end())
    {
        header = pit->second.header;
        fprintf(stderr, "header function size %jd\n", header.size);
        return true;
    }
    else
    {
        string meta = base + volume_dir + file;
        int fd = open(meta.c_str(), O_RDONLY);
        if (fd == -1) return false;

        file_header ret;
        if (::read(fd, &header, sizeof(header)) != sizeof(header))
        {
            fprintf(stderr, "%s reading error\n", file);
            exit(1);
        }

        fprintf(stderr, "header function size %jd\n", header.size);

        close(fd);
        return true;
    }
}

void silofs::write(const char *file, const void *dat, size_t size, file_header header)
{
    remove(file);

    fprintf(stderr, "trying to write %s of size %jd\n", file, size);

    vector<size_t> chunked;
    do_chunking(dat, size, chunked);

    fprintf(stderr, "chunk completed with number %jd\n", chunked.size());

    pending_file &pending = pendings[file];

    header.size = size;
    header.chunks = chunked.size();

    pending.header = header;
    pending.chunks.resize(chunked.size(), {chunk_info::pending, {}});
    pending.left = chunked.size();

    size_t last = 0;
    const char *dat_begin = static_cast<const char *>(dat);
    for (size_t i = 0; i < chunked.size(); i++)
    {
        size_t next = chunked[i];
        auto chk = chunk::frombuffer(dat_begin + last, next - last);
        last = next;

        pending.chunks[i] = {chunk_info::segbuf, chk.hash};

        auto inserted = segbuf.chunks.emplace(chk.hash, make_pair(1, move(chk)));
        if (inserted.second == false)
        {
            // inserted.first: itr, itr->second : pair<count, chk>
            inserted.first->second.first++;
            continue;
        }

        segbuf.size += chk.rawsize;
        if (segbuf.size < SEGMENT_SIZE) continue;

        md5val repid = segbuf.chunks.begin()->first;
        int blkidx;
        if (shtbl.find(blkidx, repid))
        {
            segtoblkidx(blkidx);
        }
        else
        {
            segtoblkbuf();
        }

        if (blkbuf.segments.size() >= SEGMENTS_PER_BLOCK)
        {
            flushblkbuf();
        }
    }

    flushpending();
}

void silofs::writeheader(const char *file, file_header header)
{
    fprintf(stderr, "writeheader size %jd\n", header.size);
    auto pit = pendings.find(file);
    if (pit != pendings.end())
    {
        fprintf(stderr, "writeheader size %jd\n", header.size);
        pit->second.header.atime = header.atime;
        pit->second.header.mtime = header.mtime;
        pit->second.header.ctime = header.ctime;
    }
    else
    {
        string meta = base + volume_dir + file;
        int fd = open(meta.c_str(), O_RDWR);
        if (fd == -1)
        {
            fprintf(stderr, "%s not found\n", file);
            exit(1);
        }

        file_header newheader;

        if (::read(fd, &newheader, sizeof(newheader)) != sizeof(newheader))
        {
            fprintf(stderr, "%s read error\n", file);
            exit(1);
        }

        fprintf(stderr, "writeheader before %jd\n", newheader.size);

        newheader.atime = header.atime;
        newheader.mtime = header.mtime;
        newheader.ctime = header.ctime;

        lseek(fd, 0, SEEK_SET);
		safe_write(fd, &newheader, sizeof(newheader));

        close(fd);
    }
}

bool silofs::remove(const char *file)
{
    auto pit = pendings.find(file);
    if (pit != pendings.end())
    {
        for (auto &&info : pit->second.chunks)
        {
            releasechunk(info);
        }

        pendings.erase(pit);

        return true;
    }
    else
    {
        string meta = base + volume_dir + file;
        int fd = open(meta.c_str(), O_RDONLY);
        if (fd == -1)
        {
            return false;
        }

        file_header header;
        if (::read(fd, &header, sizeof(header)) != sizeof(header))
        {
            fprintf(stderr, "%s reading error 1 %jd\n", file, sizeof(header));
            exit(1);
        }

        for (off_t i = 0, e = header.chunks; i < e; i++)
        {
            chunk_info info;
            if (::read(fd, &info, sizeof(info)) != sizeof(info))
            {
                fprintf(stderr, "%s reading error 2 %jd\n", file, sizeof(info));
                exit(1);
            }
            releasechunk(info);
        }

        close(fd);
        unlink(meta.c_str());
        return true;
    }
}

vector<char> silofs::readchunk(chunk_info info)
{
    if (info.blk_idx < 0)
    {
        auto &&chunks = info.blk_idx == chunk_info::segbuf ?
            segbuf.chunks : blkbuf.chunks;
        auto cit = chunks.find(info.hash);
        if (cit == chunks.end())
        {
            fprintf(stderr, "blockbuffer not found\n");
            exit(1);
        }

        return cit->second.second.unzip();
    }
    else
    {
        if (info.blk_idx >= blocks.size())
        {
            fprintf(stderr, "block %d not found\n", info.blk_idx);
            exit(1);
        }
        return blocks[info.blk_idx].getchunk(info.hash).unzip();
    }
}

void silofs::releasechunk(chunk_info info)
{
    switch (info.blk_idx)
    {
        case chunk_info::segbuf:
            --segbuf.chunks[info.hash].first;
            break;
        case chunk_info::blkbuf:
            --blkbuf.chunks[info.hash].first;
            break;
        default:
            blocks[info.blk_idx].releasechunk(info.hash);
            break;
    }
}

void silofs::segtoblkidx(int idx)
{
    for (auto &&chk : segbuf.chunks)
    {
        blocks[idx].addchunk(chk.second.second, chk.second.first);
    }

    for (auto &&elem : pendings)
    {
        auto &&file = elem.second;
        while (file.lastseg < file.chunks.size() && 
               file.chunks[file.lastseg].blk_idx == chunk_info::segbuf)
        {
            file.chunks[file.lastseg++].blk_idx = idx;
            --file.left;
        }
    }

    segbuf = {};
}

void silofs::segtoblkbuf()
{
    blkbuf.segments.insert(segbuf.chunks.begin()->first);

    auto segitr = segbuf.chunks.begin();
    auto blkitr = blkbuf.chunks.begin();
    auto segend = segbuf.chunks.end();
    auto blkend = blkbuf.chunks.end();

    while (segitr != segend)
    {
        if (blkitr == blkend || segitr->first < blkitr->first)
        {
            if (segitr->second.first == 0)
            {
                segitr = segbuf.chunks.erase(segitr);
                continue;
            }

            blkbuf.chunks.insert(move(*segitr++));
        }
        else if (segitr->first > blkitr->first)
        {
            blkitr++;
        }
        else
        {
            blkitr->second.first += segitr->second.first;
            blkitr++;
            segitr = segbuf.chunks.erase(segitr);
        }
    }

    for (auto &&elem : pendings)
    {
        auto &&file = elem.second;
        while (file.lastseg < file.chunks.size() && 
               file.chunks[file.lastseg].blk_idx == chunk_info::segbuf)
        {
            file.chunks[file.lastseg++].blk_idx = chunk_info::blkbuf;
        }
    }

    segbuf = {};
}

void silofs::flushblkbuf()
{
    string dir = base + to_string(lastblock);
    if (mkdir(dir.c_str(), 0755) == -1)
    {
        fprintf(stderr, "creating dir %s failed\n", dir.c_str());
        exit(1);
    }
    blocks.emplace_back(dir.c_str());

    for (auto &&repid : blkbuf.segments)
    {
        shtbl.insert(repid, lastblock);
    }

    blocks[lastblock].create_with_chunks(blkbuf.chunks);

    for (auto &&elem : pendings)
    {
        auto &&file = elem.second;
        for (bool flag = true; file.lastblk < file.chunks.size() && flag;)
        {
            switch (file.chunks[file.lastblk].blk_idx)
            {
            case chunk_info::pending:
                flag = false;
                break;
            case chunk_info::blkbuf:
                file.chunks[file.lastblk].blk_idx = lastblock;
                --file.left;
            default:
                file.lastblk++;
                break;
            }
        }
    }

    blkbuf = {};

    lastblock++;
}

void silofs::flushpending()
{
    for (auto itr = pendings.begin(); itr != pendings.end();)
    {
        auto &&pending = itr->second;
        if (pending.left == 0)
        {
            string meta = base + volume_dir + itr->first;

            int fd = open(meta.c_str(), O_WRONLY | O_TRUNC | O_CREAT, 0755);
            if (fd == -1)
            {
                int err = errno;
                fprintf(stderr, "flush: %s open fail with errno %d\n", meta.c_str(), err);
                exit(1);
            }

            fprintf(stderr, "flush header size %jd\n", pending.header.size);

            safe_write(fd, &pending.header, sizeof(pending.header));

            for (auto &&info : pending.chunks)
            {
                safe_write(fd, &info, sizeof(info));
            }

            close(fd);
            itr = pendings.erase(itr);
        }
        else
        {
            ++itr;
        }
    }
}

string silofs::getvolumepath(const char *path)
{
    return base + volume_dir + path;
}