path: root/searchlib/src/vespa/searchlib/transactionlog/domainpart.cpp

// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.

#include "domainpart.h"
#include <vespa/vespalib/util/crc.h>
#include <vespa/vespalib/util/stringfmt.h>
#include <vespa/vespalib/data/fileheader.h>
#include <vespa/searchlib/common/fileheadercontext.h>
#include <vespa/fastlib/io/bufferedfile.h>
#include <xxhash.h>

#include <vespa/log/log.h>
LOG_SETUP(".transactionlog.domainpart");

using vespalib::make_string;
using vespalib::FileHeader;
using vespalib::string;
using vespalib::getLastErrorString;
using vespalib::IllegalHeaderException;
using vespalib::LockGuard;
using vespalib::nbostream;
using vespalib::nbostream_longlivedbuf;
using vespalib::alloc::Alloc;
using search::common::FileHeaderContext;
using std::runtime_error;

namespace search::transactionlog {

namespace {

void
handleSync(FastOS_FileInterface &file) __attribute__ ((noinline));

string
handleWriteError(const char *text,
                 FastOS_FileInterface &file,
                 int64_t lastKnownGoodPos,
                 const Packet::Entry &entry,
                 int bufLen) __attribute__ ((noinline));

bool
handleReadError(const char *text,
                FastOS_FileInterface &file,
                ssize_t len,
                ssize_t rlen,
                int64_t lastKnownGoodPos,
                bool allowTruncate) __attribute__ ((noinline));

bool
addPacket(Packet &packet,
          const Packet::Entry &e) __attribute__ ((noinline));

bool
tailOfFileIsZero(FastOS_FileInterface &file, int64_t lastKnownGoodPos) __attribute__ ((noinline));

bool
addPacket(Packet &packet, const Packet::Entry &e)
{
    LOG(spam, "Adding serial #%" PRIu64 ", of type %d and size %zd into packet of size %zu and %zu bytes",
              e.serial(), e.type(), e.data().size(), packet.size(), packet.sizeBytes());
    return ! packet.add(e);
}

void
handleSync(FastOS_FileInterface &file)
{
    if ( file.IsOpened() && ! file.Sync() ) {
        int osError = errno;
        throw runtime_error(make_string("Failed to synchronize file '%s' of size %" PRId64 " due to '%s'. "
                                        "Does not know how to handle this so throwing an exception.",
                                        file.GetFileName(), file.GetSize(), FastOS_File::getErrorString(osError).c_str()));
    }
}

string
handleWriteError(const char *text,
                 FastOS_FileInterface &file,
                 int64_t lastKnownGoodPos,
                 const Packet::Entry &entry,
                 int bufLen)
{
    string last(FastOS_File::getLastErrorString());
    string e(make_string("%s. File '%s' at position %" PRId64 " for entry %" PRIu64 " of length %u. "
                         "OS says '%s'. Rewind to last known good position %" PRId64 ".",
                         text, file.GetFileName(), file.GetPosition(), entry.serial(), bufLen,
                         last.c_str(), lastKnownGoodPos));
    LOG(error, "%s",  e.c_str());
    if ( ! file.SetPosition(lastKnownGoodPos) ) {
        last = FastOS_File::getLastErrorString();
        throw runtime_error(make_string("Failed setting position %" PRId64 " of file '%s' of size %" PRId64 ": OS says '%s'",
                                        lastKnownGoodPos, file.GetFileName(), file.GetSize(), last.c_str()));
    }
    handleSync(file);
    return e;
}

string
getError(FastOS_FileInterface & f)
{
    return make_string("File '%s' of size %" PRId64 " has last error of '%s'.",
                       f.GetFileName(), f.GetSize(), FastOS_File::getLastErrorString().c_str());
}

bool
tailOfFileIsZero(FastOS_FileInterface &file, int64_t lastKnownGoodPos)
{
    ssize_t rest(file.GetSize() - lastKnownGoodPos);
    if (rest < 0 || rest > 0x100000) {
        return false;
    }
    std::vector<char> buf(rest, 0);
    file.ReadBuf(&buf[0], buf.size(), lastKnownGoodPos);
    for (char c : buf) {
        if (c != 0) {
            return false;
        }
    }
    return true;
}

bool
handleReadError(const char *text,
                FastOS_FileInterface &file,
                ssize_t len,
                ssize_t rlen,
                int64_t lastKnownGoodPos,
                bool allowTruncate)
{
    bool retval(true);
    if (rlen != -1) {
        string e;
        if (len == rlen) {
            e = make_string("Error in data read of size %zd bytes at pos %" PRId64 " trying to read %s. ",
                             len, file.GetPosition() - rlen, text);
        } else {
            e = make_string("Short Read. Got only %zd of %zd bytes at pos %" PRId64 " trying to read %s. ",
                             rlen, len, file.GetPosition() - rlen, text);
        }
        e += getError(file);
        if (!allowTruncate) {
            LOG(error, "%s", e.c_str());
            throw runtime_error(e);
        }
        // Short read. Log error, Truncate, continue.
        e += make_string(" Truncate to %" PRId64 " and continue", lastKnownGoodPos);
        LOG(error, "%s", e.c_str());
        FastOS_File truncateFile(file.GetFileName());
        file.Close();
        if ( truncateFile.OpenWriteOnlyExisting()) {
            if (truncateFile.SetSize(lastKnownGoodPos)) {
                if (truncateFile.Close()) {
                    if (file.OpenReadOnly()) {
                        if (file.SetPosition(lastKnownGoodPos)) {
                            retval = false;
                        } else {
                            throw runtime_error(make_string("Failed setting position %" PRId64 ". %s", lastKnownGoodPos, getError(file).c_str()));
                        }
                    } else {
                        throw runtime_error(make_string("Failed reopening file after truncate: %s", getError(file).c_str()));
                    }
                } else {
                    throw runtime_error(make_string("Failed closing truncated file: %s", getError(truncateFile).c_str()));
                }
            } else {
                throw runtime_error(make_string("Failed truncating to %" PRId64 ": %s", lastKnownGoodPos, getError(truncateFile).c_str()));
            }
        } else {
            throw runtime_error(make_string("Failed opening for truncating: %s", getError(file).c_str()));
        }
    } else {
        // Some kind of IO error throw exception.
        string errString = FastOS_File::getLastErrorString();
        throw runtime_error(make_string("IO error when reading %zd bytes at pos %" PRId64 "trying to read %s."
                                        " Last known good position is %" PRId64 ": %s",
                                        len, file.GetPosition(), text, lastKnownGoodPos, getError(file).c_str()));
    }
    return retval;
}

}

int64_t
DomainPart::buildPacketMapping(bool allowTruncate)
{
    Fast_BufferedFile transLog;
    transLog.EnableDirectIO();
    if ( ! transLog.OpenReadOnly(_transLog->GetFileName())) {
        throw runtime_error(make_string("Failed opening '%s' for buffered readinf with direct io.", transLog.GetFileName()));
    }
    int64_t fSize(transLog.GetSize());
    int64_t currPos(0);
    try {
        FileHeader header;
        _headerLen = header.readFile(transLog);
        transLog.SetPosition(_headerLen);
        currPos = _headerLen;
    } catch (const IllegalHeaderException &e) {
        transLog.SetPosition(0);
        try {
            FileHeader::FileReader fr(transLog);
            uint32_t header2Len = FileHeader::readSize(fr);
            if (header2Len <= fSize)
                e.throwSelf(); // header not truncated
        } catch (const IllegalHeaderException &e2) {
        }
        if (fSize > 0) {
            // Truncate file (dropping header) if cannot even read
            // header length, or if header has been truncated.
            handleReadError("file header", transLog, 0, FileHeader::getMinSize(), 0, allowTruncate);
        }
    }
    while ((currPos < fSize)) {
        Packet packet;
        SerialNum firstSerial(0);
        SerialNum lastSerial(0);
        int64_t firstPos(currPos);
        bool full(false);
        Alloc buf;
        for(size_t i(0); !full && (currPos < fSize); i++) {
            Packet::Entry e;
            if (read(transLog, e, buf, allowTruncate)) {
                if (e.valid()) {
                    if (i == 0) {
                        firstSerial = e.serial();
                        if (currPos == _headerLen) {
                            _range.from(firstSerial);
                        }
                    }
                    try {
                        full = addPacket(packet, e);
                        if ( ! full ) {
                            lastSerial = e.serial();
                            currPos = transLog.GetPosition();
                            _sz++;
                        } else {
                            transLog.SetPosition(currPos);
                        }
                    } catch (const std::exception & ex) {
                        throw runtime_error(make_string("%s : Failed creating packet for list %s(%" PRIu64 ") at pos(%" PRIu64 ", %" PRIu64 ")",
                                                    ex.what(), transLog.GetFileName(), fSize, currPos, transLog.GetPosition()));
                    }
                } else {
                    throw runtime_error(make_string("Invalid entry reading file %s(%" PRIu64 ") at pos(%" PRIu64 ", %" PRIu64 ")",
                                                transLog.GetFileName(), fSize, currPos, transLog.GetPosition()));
                }
            } else {
                if (transLog.GetSize() != fSize) {
                    fSize = transLog.GetSize();
                } else {
                    throw runtime_error(make_string("Failed reading file %s(%" PRIu64 ") at pos(%" PRIu64 ", %" PRIu64 ")",
                                                transLog.GetFileName(), fSize, currPos, transLog.GetPosition()));
                }
            }
        }
        packet.close();
        if (!packet.empty()) {
            _packets[firstSerial] = packet;
            _range.to(lastSerial);
            {
                LockGuard guard(_lock);
                _skipList.push_back(SkipInfo(firstSerial, firstPos));
            }
        }
    }
    transLog.Close();
    return currPos;
}

DomainPart::DomainPart(const string & name, const string & baseDir, SerialNum s, Crc defaultCrc,
                       const FileHeaderContext &fileHeaderContext, bool allowTruncate) :
    _defaultCrc(defaultCrc),
    _lock(),
    _fileLock(),
    _range(s),
    _sz(0),
    _byteSize(0),
    _packets(),
    _fileName(make_string("%s/%s-%016" PRIu64, baseDir.c_str(), name.c_str(), s)),
    _transLog(std::make_unique<FastOS_File>(_fileName.c_str())),
    _skipList(),
    _headerLen(0),
    _writeLock(),
    _writtenSerial(0),
    _syncedSerial(0)
{
    if (_transLog->OpenReadOnly()) {
        int64_t currPos = buildPacketMapping(allowTruncate);
        if ( ! _transLog->Close() ) {
            throw runtime_error(make_string("Failed closing file '%s' after reading.", _transLog->GetFileName()));
        }
        if ( ! _transLog->OpenWriteOnlyExisting() ) {
            string e(make_string("Failed opening existing file '%s' for writing: %s", _transLog->GetFileName(), getLastErrorString().c_str()));
            LOG(error, "%s", e.c_str());
            throw runtime_error(e);
        }
        if (currPos == 0) {
            // Previous header was truncated.  Write new one.
            writeHeader(fileHeaderContext);
            currPos = _headerLen;
        }
        _byteSize = currPos;
    } else {
        if ( ! _transLog->OpenWriteOnly()) {
            string e(make_string("Failed opening new file '%s' for writing: '%s'", _transLog->GetFileName(), getLastErrorString().c_str()));

            LOG(error, "%s", e.c_str());
            throw runtime_error(e);
        }
        writeHeader(fileHeaderContext);
        _byteSize = _headerLen;
    }
    if ( ! _transLog->SetPosition(_transLog->GetSize()) ) {
        throw runtime_error(make_string("Failed moving write pointer to the end of the file %s(%" PRIu64 ").",
                                        _transLog->GetFileName(), _transLog->GetSize()));
    }
    handleSync(*_transLog);
    _writtenSerial = _range.to();
    _syncedSerial = _writtenSerial;
}

DomainPart::~DomainPart()
{
    close();
}

void
DomainPart::writeHeader(const FileHeaderContext &fileHeaderContext)
{
    typedef vespalib::GenericHeader::Tag Tag;
    FileHeader header;
    assert(_transLog->IsOpened());
    assert(_transLog->IsWriteMode());
    assert(_transLog->GetPosition() == 0);
    fileHeaderContext.addTags(header, _transLog->GetFileName());
    header.putTag(Tag("desc", "Transaction log domain part file"));
    _headerLen = header.writeFile(*_transLog);
}

bool
DomainPart::close()
{
    bool retval(false);
    {
        LockGuard guard(_fileLock);
        /*
         * Sync old domainpart before starting writing new, to avoid
         * hole.  XXX: Feed latency spike due to lack of delayed open
         * for new domainpart.
         */
        handleSync(*_transLog);
        _transLog->dropFromCache();
        retval = _transLog->Close();
        LockGuard wguard(_writeLock);
        _syncedSerial = _writtenSerial;
    }
    if ( ! retval ) {
        throw runtime_error(make_string("Failed closing file '%s' of size %" PRId64 ".",
                                        _transLog->GetFileName(), _transLog->GetSize()));
    }
    {
        LockGuard guard(_lock);
        _packets.clear();
    }
    return retval;
}

bool
DomainPart::isClosed() const {
    return ! _transLog->IsOpened();
}

bool
DomainPart::openAndFind(FastOS_FileInterface &file, const SerialNum &from)
{
    bool retval(file.OpenReadOnly(_transLog->GetFileName()));
    if (retval) {
        int64_t pos(_headerLen);
        LockGuard guard(_lock);
        for(SkipList::const_iterator it(_skipList.begin()), mt(_skipList.end());
            (it < mt) && (it->id() <= from);
            it++)
        {
            pos = it->filePos();
        }
        retval = file.SetPosition(pos);
    }
    return retval;
}

bool
DomainPart::erase(SerialNum to)
{
    bool retval(true);
    if (to > _range.to()) {
        close();
        _transLog->Delete();
    } else {
        _range.from(std::max(to, _range.from()));
    }
    return retval;
}

void
DomainPart::commit(SerialNum firstSerial, const Packet &packet)
{
    int64_t firstPos(_transLog->GetPosition());
    nbostream_longlivedbuf h(packet.getHandle().data(), packet.getHandle().size());
    if (_range.from() == 0) {
        _range.from(firstSerial);
    }
    for (size_t i(0); h.size() > 0; i++) {
        //LOG(spam,
        //"Pos(%d) Len(%d), Lim(%d), Remaining(%d)",
        //h.getPos(), h.getLength(), h.getLimit(), h.getRemaining());
        Packet::Entry entry;
        entry.deserialize(h);
        if (_range.to() < entry.serial()) {
            write(*_transLog, entry);
            _sz++;
            _range.to(entry.serial());
        } else {
            throw runtime_error(make_string("Incomming serial number(%" PRIu64 ") must be bigger than the last one (%" PRIu64 ").",
                                            entry.serial(), _range.to()));
        }
    }

    bool merged(false);
    LockGuard guard(_lock);
    if ( ! _packets.empty() ) {
        Packet & lastPacket = _packets.rbegin()->second;
        if (lastPacket.sizeBytes() < 0xf000) {
            if ( ! (merged = lastPacket.merge(packet)) ) {
                LOG(error, "Failed merging packet [%" PRIu64 ", %" PRIu64 "] with [%" PRIu64 ", %" PRIu64 "]",
                           lastPacket.range().from(), lastPacket.range().to(),
                           packet.range().from(), packet.range().to());
            }
        }
    }
    if (! merged ) {
        _packets[firstSerial] = packet;
        _skipList.push_back(SkipInfo(firstSerial, firstPos));
    }
}

void DomainPart::sync()
{
    SerialNum syncSerial(0);
    {
        LockGuard guard(_writeLock);
        syncSerial = _writtenSerial;
    }
    LockGuard guard(_fileLock);
    handleSync(*_transLog);
    LockGuard wguard(_writeLock);
    if (_syncedSerial < syncSerial) {
        _syncedSerial = syncSerial;
    }
}

bool
DomainPart::visit(SerialNumRange &r, Packet &packet)
{
    bool retval(false);
    LockGuard guard(_lock);
    LOG(debug, "Visit r(%" PRIu64 ", %" PRIu64 "] Checking %" PRIu64 " packets",
               r.from(), r.to(), uint64_t(_packets.size()));
    if ( ! isClosed() ) {
        PacketList::const_iterator start(_packets.lower_bound(r.from() + 1));
        PacketList::const_iterator end(_packets.upper_bound(r.to()));
        if (start != _packets.end()) {
            if ( ! start->second.range().contains(r.from() + 1) &&
                (start != _packets.begin())) {
                PacketList::const_iterator prev(start);
                prev--;
                if (prev->second.range().contains(r.from() + 1)) {
                    start--;
                }
            }
        } else {
            if (!_packets.empty())
                start--;
        }
        if ( start != _packets.end() && start->first <= r.to()) {
            PacketList::const_iterator next(start);
            next++;
            if ((r.from() < start->first) &&
                ((next != end) || ((next != _packets.end()) && ((r.to() + 1) == next->first))))
            {
                packet = start->second;
                LOG(debug, "Visit whole packet[%" PRIu64 ", %" PRIu64 "]", packet.range().from(), packet.range().to());
                if (next != _packets.end()) {
                    r.from(next->first - 1);
                    retval = true;
                } else {
                    /// This is the very last package. Can safely finish.
                }
            } else {
                const nbostream & tmp = start->second.getHandle();
                nbostream_longlivedbuf h(tmp.data(), tmp.size());
                LOG(debug, "Visit partial[%" PRIu64 ", %" PRIu64 "] (%zd, %zd, %zd)",
                           start->second.range().from(), start->second.range().to(), h.rp(), h.size(), h.capacity());
                Packet newPacket(h.size());
                for (; (h.size() > 0) && (r.from() < r.to()); ) {
                    Packet::Entry e;
                    e.deserialize(h);
                    if (r.from() < e.serial()) {
                        if (e.serial() <= r.to()) {
                            LOG(spam, "Adding serial #%" PRIu64 ", of type %d and size %zd into packet of size %zu and %zu bytes",
                                      e.serial(), e.type(), e.data().size(), newPacket.size(), newPacket.sizeBytes());
                            if (newPacket.add(e)) {
                                r.from(e.serial());
                            } else {
                                throw runtime_error("Could not add entry to packet. Here is some mumbo jumbo. Fix.");
                            }
                        } else {
                            // Force breakout on visiting empty interval.
                            r.from(r.to());
                        }
                    }
                }
                newPacket.close();
                packet = newPacket;
                retval = next != _packets.end();
            }
        } else {
            packet.close();
        }
    } else {
        /// File has been closed must continue from file.
        retval = true;
    }
    return retval;
}


bool
DomainPart::visit(FastOS_FileInterface &file, SerialNumRange &r, Packet &packet)
{
    bool retval(true);
    if ( ! file.IsOpened() ) {
        retval = openAndFind(file, r.from() + 1);
    }
    if (retval) {
        Packet newPacket;
        Alloc buf;
        for (bool full(false);!full && retval && (r.from() < r.to());) {
            Packet::Entry e;
            int64_t fPos = file.GetPosition();
            retval = read(file, e, buf, false);
            if (retval &&
                e.valid() &&
                (r.from() < e.serial()) &&
                (e.serial() <= r.to())) {
                try {
                    full = addPacket(newPacket, e);
                } catch (const std::exception & ex) {
                    throw runtime_error(make_string("%s : Failed creating packet for visit %s(%" PRIu64 ") at pos(%" PRIu64 ", %" PRIu64 ")",
                                                    ex.what(), file.GetFileName(), file.GetSize(), fPos, file.GetPosition()));
                }
                if ( !full ) {
                    r.from(e.serial());
                } else {
                    if ( ! file.SetPosition(fPos) ) {
                        throw runtime_error(make_string("Failed setting read position for file '%s' of size %" PRId64 " from %" PRId64 " to %" PRId64 ".",
                                                        file.GetFileName(), file.GetSize(), file.GetPosition(), fPos));
                    }
                }
            }
        }
        newPacket.close();
        packet = newPacket;
    }

    return retval;
}

void
DomainPart::write(FastOS_FileInterface &file, const Packet::Entry &entry)
{
    int64_t lastKnownGoodPos(file.GetPosition());
    int32_t crc(0);
    uint32_t len(entry.serializedSize() + sizeof(crc));
    nbostream os;
    os << static_cast<uint8_t>(_defaultCrc);
    os << len;
    size_t start(os.size());
    entry.serialize(os);
    size_t end(os.size());
    crc = calcCrc(_defaultCrc, os.data() + start, end - start);
    os << crc;
    size_t osSize = os.size();
    assert(osSize == len + sizeof(len) + sizeof(uint8_t));

    LockGuard guard(_writeLock);
    if ( ! file.CheckedWrite(os.data(), osSize) ) {
        throw runtime_error(handleWriteError("Failed writing the entry.", file, lastKnownGoodPos, entry, end - start));
    }
    _writtenSerial = entry.serial();
    _byteSize.store(lastKnownGoodPos + osSize, std::memory_order_release);
}

bool
DomainPart::read(FastOS_FileInterface &file,
                 Packet::Entry &entry,
                 Alloc & buf,
                 bool allowTruncate)
{
    bool retval(true);
    char tmp[5];
    int64_t lastKnownGoodPos(file.GetPosition());
    size_t rlen = file.Read(tmp, sizeof(tmp));
    nbostream his(tmp, sizeof(tmp));
    uint8_t version(-1);
    uint32_t len(0);
    his >> version >> len;
    if ((retval = (rlen == sizeof(tmp)))) {
        if ( ! (retval = (version == ccitt_crc32) || version == xxh64)) {
            string msg(make_string("Version mismatch. Expected 'ccitt_crc32=1' or 'xxh64=2',"
                                             " got %d from '%s' at position %" PRId64,
                                             version, file.GetFileName(), lastKnownGoodPos));
            if ((version == 0) && (len == 0) && tailOfFileIsZero(file, lastKnownGoodPos)) {
                LOG(warning, "%s", msg.c_str());
                return handleReadError("packet version", file, sizeof(tmp), rlen, lastKnownGoodPos, allowTruncate);
            } else {
                throw runtime_error(msg);
            }
        }
        if (len > buf.size()) {
            Alloc::alloc(len).swap(buf);
        }
        rlen = file.Read(buf.get(), len);
        retval = rlen == len;
        if (!retval) {
            retval = handleReadError("packet blob", file, len, rlen, lastKnownGoodPos, allowTruncate);
        } else {
            nbostream_longlivedbuf is(buf.get(), len);
            entry.deserialize(is);
            int32_t crc(0);
            is >> crc;
            int32_t crcVerify(calcCrc(static_cast<Crc>(version), buf.get(), len - sizeof(crc)));
            if (crc != crcVerify) {
                throw runtime_error(make_string("Got bad crc for packet from '%s' (len pos=%" PRId64 ", len=%d) : crcVerify = %d, expected %d",
                                                file.GetFileName(), file.GetPosition() - len - sizeof(len),
                                                static_cast<int>(len), static_cast<int>(crcVerify), static_cast<int>(crc)));
            }
        }
    } else {
        if (rlen == 0) {
           // Eof
        } else {
           retval = handleReadError("packet length", file, sizeof(len), rlen, lastKnownGoodPos, allowTruncate);
        }
    }
    return retval;
}

int32_t DomainPart::calcCrc(Crc version, const void * buf, size_t sz)
{
    if (version == xxh64) {
        return static_cast<int32_t>(XXH64(buf, sz, 0ll));
    } else if (version == ccitt_crc32) {
        vespalib::crc_32_type calculator;
        calculator.process_bytes(buf, sz);
        return calculator.checksum();
    } else {
        LOG_ABORT("should not be reached");
    }
}

}