root/hydranode/hncore/partdata.cpp

/*
 *  Copyright (C) 2004-2006 Alo Sarv <madcat_@users.sourceforge.net>
 *
 *  This program is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 2 of the License, or
 *  (at your option) any later version.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program; if not, write to the Free Software
 *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

/**
 * \file partdata.cpp Implementation of PartData and related classes
 */

#include <hncore/pch.h>

#include <hnbase/log.h>
#include <hnbase/lambda_placeholders.h>
#include <hnbase/hash.h>
#include <hnbase/prefs.h>
#include <hnbase/timed_callback.h>

#include <hncore/partdata.h>
#include <hncore/partdata_impl.h>
#include <hncore/metadata.h>
#include <hncore/hasher.h>
#include <hncore/hydranode.h>

#include <boost/lambda/lambda.hpp>
#include <boost/lambda/if.hpp>
#include <boost/lambda/bind.hpp>
#include <boost/filesystem/operations.hpp>
#include <boost/algorithm/string/replace.hpp>

#include <fstream>
#include <fcntl.h>

using namespace boost::lambda;
using namespace boost::multi_index;
using namespace CGComm;

static const uint32_t BUF_SIZE_LIMIT = 512*1024; //!< 512k buffer

namespace Detail {

// UsedRange class
// -------------------------
template<typename IterType>
UsedRange::UsedRange(PartData *parent, IterType it) : Range64((*it).begin(),
(*it).end()), m_parent(parent),
m_chunk(new PosIter(project<ID_Pos>(*m_parent->m_chunks, it))) {
        CHECK_THROW(parent != 0);
        CHECK_THROW(*m_chunk != get<ID_Pos>(*parent->m_chunks).end());

        // SpamReduction(tm)
        if (length() > 1) {
                logTrace(TRACE_PARTDATA,
                        boost::format("%s: Using range %d..%d")
                        % m_parent->m_dest.leaf() % begin() % end()
                );
        }

        get<ID_Pos>(*m_parent->m_chunks).modify(
                *m_chunk, ++bind(&Chunk::m_useCnt, __1)
        );
}

UsedRange::UsedRange(PartData *parent, uint64_t begin, uint64_t end) :
Range64(begin, end), m_parent(parent) {
        m_chunk.reset(new PosIter(get<ID_Pos>(*m_parent->m_chunks).end()));

        if (length() > 1) {
                logTrace(TRACE_PARTDATA,
                        boost::format("%s: Using range %d..%d")
                        % m_parent->m_dest.leaf() % this->begin() % this->end()
                );
        }
}

UsedRange::~UsedRange() {
        if (length() > 1) {
                logTrace(TRACE_PARTDATA,
                        boost::format("%s: Un-using range %d..%d")
                        % m_parent->m_dest.leaf() % begin() % end()
                );
        }

        if (*m_chunk != m_parent->m_chunks->get<ID_Pos>().end()) {
                get<ID_Pos>(*m_parent->m_chunks).modify(
                        *m_chunk, --bind(&Chunk::m_useCnt, __1)
                );
        }
}

LockedRangePtr UsedRange::getLock(uint32_t size) {
        return m_parent->getLock(shared_from_this(), size);
}

bool UsedRange::isComplete() const {
        return m_parent->isComplete(*this);
}

// LockedRange class
// ---------------------------
LockedRange::LockedRange(PartData *parent, Range64 r, UsedRangePtr used)
: Range64(r), m_parent(parent),
m_chunk(new PosIter(get<ID_Pos>(*m_parent->m_chunks).end())), m_used(used) {

        // SpamReduction(tm)
        if (length() > 1) {
                logTrace(TRACE_PARTDATA,
                        boost::format("%s: Locking range %d..%d")
                        % m_parent->m_dest.leaf() % begin() % end()
                );
        }

        m_parent->m_locked.merge(*this);
}

LockedRange::LockedRange(
        PartData *parent, Range64 r, PosIter &it, UsedRangePtr used
) : Range64(r), m_parent(parent), m_chunk(new PosIter(it)), m_used(used) {
        CHECK_THROW(parent != 0);

        if (length() > 1) {
                logTrace(TRACE_PARTDATA,
                        boost::format("%s: Locking range %d..%d")
                        % m_parent->m_dest.leaf() % begin() % end()
                );
        }

        m_parent->m_locked.merge(*this);
}

LockedRange::~LockedRange() {
        if (length() > 1) {
                logTrace(TRACE_PARTDATA,
                        boost::format("%s: UnLocking range %d..%d")
                        % m_parent->m_dest.leaf() % begin() % end()
                );
        }

        m_parent->m_locked.erase(*this);
}

void LockedRange::write(uint64_t begin, const std::string &data) {
        if (begin > end() || begin + data.size() - 1 > end()) {
                throw PartData::LockError("Writing outside lock.");
        }
        if (*m_chunk != get<ID_Pos>(*m_parent->m_chunks).end()) {
                m_parent->m_chunks->get<ID_Pos>().modify(
                        *m_chunk, bind(&Chunk::write, __1, begin, data)
                );
        } else {
                m_parent->doWrite(begin, data);
                m_parent->tryComplete();
        }
}

// Chunk class
// ---------------------
Chunk::Chunk(
        PartData *parent, uint64_t begin, uint64_t end,
        uint32_t size, const HashBase *hash
) : Range64(begin, end), m_parent(parent), m_hash(hash), m_verified(
        m_parent->isVerified(*this)
), m_partial(
        m_parent->m_complete.contains(*this) &&
        !m_parent->m_complete.containsFull(*this)
), m_complete(m_parent->isComplete(*this)), m_avail(), m_useCnt(), m_size(size)
{
        if (isComplete() && !isVerified() && hash) {
                HashWorkPtr c = m_parent->verifyRange(*this, m_hash);
                if (c) {
                        c->getEventTable().addHandler(
                                c, this, &Chunk::onHashEvent
                        );
                }
        }
}

Chunk::Chunk(
        PartData *parent, Range64 range, uint32_t size, const HashBase *hash
) : Range64(range), m_parent(parent), m_hash(hash), m_verified(
        m_parent->isVerified(*this)
), m_partial(
        m_parent->m_complete.contains(*this) &&
        !m_parent->m_complete.containsFull(*this)
), m_complete(m_parent->isComplete(*this)), m_avail(), m_useCnt(), m_size(size)
{
        if (isComplete() && !isVerified() && hash) {
                HashWorkPtr c = m_parent->verifyRange(*this, m_hash);
                if (c) {
                        c->getEventTable().addHandler(
                                c, this, &Chunk::onHashEvent
                        );
                }
        }
}

void Chunk::write(uint64_t begin, const std::string &data) {
        m_parent->doWrite(begin, data);
        if (!m_partial) {
                CMPosIndex &idx = m_parent->m_chunks->get<ID_Pos>();
                CMPosIndex::iterator it = idx.find(*this);
                assert(it != idx.end());
                idx.modify(it, bind(&Chunk::m_partial, __1) = true);
        }
}

void Chunk::updateState() const {
        CMPosIndex &idx = m_parent->m_chunks->get<ID_Pos>();
        CMPosIndex::iterator it = idx.find(*this);
        assert(it != idx.end());
        if (m_parent->m_complete.contains(*this)) {
                if (m_parent->m_complete.containsFull(*this)) {
                        idx.modify(it, bind(&Chunk::m_partial, __1) = false);
                        idx.modify(it, bind(&Chunk::m_complete, __1) = true);
                } else {
                        idx.modify(it, bind(&Chunk::m_partial, __1) = true);
                        idx.modify(it, bind(&Chunk::m_complete, __1) = false);
                }
        } else {
                idx.modify(it, bind(&Chunk::m_partial, __1) = false);
                idx.modify(it, bind(&Chunk::m_complete, __1) = false);
        }
        idx.modify(
                it, bind(&Chunk::m_verified, __1) = m_parent->isVerified(*this)
        );
        if (isComplete() && !isVerified()) {
                const_cast<Chunk*>(this)->verify();
        }
}

void Chunk::verify(bool save) {
        logTrace(TRACE_PARTDATA,
                boost::format("%s: Completed chunk %d..%d")
                % m_parent->m_dest.leaf() % this->begin() % this->end()
        );

        CMPosIndex &idx = m_parent->m_chunks->get<ID_Pos>();
        CMPosIndex::iterator it = idx.find(*this);
        assert(it != idx.end());
        idx.modify(it, bind(&Chunk::m_partial, __1) = false);
        idx.modify(it, bind(&Chunk::m_verified, __1) = false);
        idx.modify(it, bind(&Chunk::m_complete, __1) = true);

        if (m_hash) {
                HashWorkPtr c = m_parent->verifyRange(*this, m_hash, save);
                if (c) {
                        c->getEventTable().addHandler(
                                c, this, &Chunk::onHashEvent
                        );
                }
        } else {
                logWarning(
                        boost::format(
                                "%s: Chunk %s..%s has no hash - "
                                "shouldn't happen..."
                        ) % m_parent->m_dest.leaf() %this->begin() % this->end()
                );
                m_parent->tryComplete();
        }
}

void Chunk::onHashEvent(HashWorkPtr c, HashEvent evt){
        assert(*c->getRef() == *getHash());
        c->getEventTable().delHandlers(c);

        CMPosIndex &idx = m_parent->m_chunks->get<ID_Pos>();
        CMPosIndex::iterator it = idx.find(*this);
        assert(it != idx.end());
        if (evt == HASH_FAILED) {
                m_parent->corruption(*this);
                idx.modify(it, bind(&Chunk::m_verified, __1) = false);
                idx.modify(it, bind(&Chunk::m_complete, __1) = false);
                idx.modify(it, bind(&Chunk::m_partial, __1) = false);
                if (m_parent->m_fullJob) {
                        m_parent->m_fullJob->cancel();
                        m_parent->m_fullJob = HashWorkPtr();
                }
                m_parent->m_partStatus[m_size][begin()/m_size] = false;
                logTrace(TRACE_CHUNKS,
                        boost::format("Chunk #%d is corrupt.")
                        % (begin() / m_size)
                );
                assert(!isComplete());
        } else if (evt == HASH_VERIFIED) {
                idx.modify(it, bind(&Chunk::m_complete, __1) = true);
                idx.modify(it, bind(&Chunk::m_verified, __1) = true);
                idx.modify(it, bind(&Chunk::m_partial, __1) = false);
                m_parent->m_partStatus[m_size][begin()/m_size] = true;
                m_parent->m_verified.merge(*this);
                m_parent->m_complete.merge(*this);
                m_parent->onVerified(m_parent, m_size, begin()/m_size);
                logTrace(TRACE_CHUNKS,
                        boost::format("Verified chunk #%d") % (begin() / m_size)
                );
        } else if (evt == HASH_FATAL_ERROR) {
                logError(
                        boost::format("Fatal error hashing file `%s'")
                        % c->getFileName().native_file_string()
                );
        }
        --m_parent->m_pendingHashes;
        m_parent->tryComplete();
}

/**
 * AllocJob allocates disk space for temp file. Emits event 'true' when
 * allocation succeeds, false otherwise.
 */
class AllocJob : public ThreadWork {
public:
        DECLARE_EVENT_TABLE(AllocJobPtr, bool);
        AllocJob(const boost::filesystem::path &file, uint64_t size);
        virtual bool process();
private:
        boost::filesystem::path m_file;
        uint64_t m_size;
};
IMPLEMENT_EVENT_TABLE(AllocJob, AllocJobPtr, bool);

AllocJob::AllocJob(const boost::filesystem::path &file, uint64_t size)
: m_file(file), m_size(size) {}

// allocates disk space in 10mb increments
bool AllocJob::process() {
        int fd = open(m_file.native_file_string().c_str(), O_RDWR|O_BINARY);
        if (!fd) {
                getEventTable().postEvent(AllocJobPtr(this), false);
        } else {
                ::lseek64(fd, m_size - 1, SEEK_SET);
                ::write(fd, "1", 1);
                ::fsync(fd);
                ::close(fd);
                assert(Utils::getFileSize(m_file) <= m_size);
                bool ret = Utils::getFileSize(m_file) == m_size;
                getEventTable().postEvent(AllocJobPtr(this), ret);
        }
        setComplete();
        return true;
}
} // namespace Detail

using namespace Detail;

// PartData exception classes
// --------------------------
PartData::LockError::LockError(const std::string &msg):std::runtime_error(msg){}
PartData::RangeError::RangeError(const std::string &mg):std::runtime_error(mg){}

// PartData class
// --------------
IMPLEMENT_EVENT_TABLE(PartData, PartData*, int);

PartData::PartData(
        uint64_t size,
        const boost::filesystem::path &loc,
        const boost::filesystem::path &dest
) : Object(0), m_size(size), m_loc(loc), m_dest(dest), m_chunks(new ChunkMap),
m_toFlush(), m_md(), m_pendingHashes(), m_sourceCnt(), m_fullSourceCnt(),
m_paused(), m_stopped(), m_autoPaused() {
        initSignals();

        std::ofstream o(loc.string().c_str(), std::ios::binary);
        o.flush();
        getEventTable().postEvent(this, PD_ADDED);
        cleanupName();
}


PartData::PartData(const boost::filesystem::path &p) try : Object(0),
m_size(), m_chunks(new ChunkMap), m_toFlush(), m_md(), m_pendingHashes(),
m_sourceCnt(), m_fullSourceCnt(), m_paused(), m_stopped(), m_autoPaused() {
        initSignals();

        logTrace(TRACE_PARTDATA,
                boost::format("Loading temp file: %s") % p.string()
        );
        using namespace Utils;
        using boost::algorithm::replace_last;

        std::string tmp(p.string());
        replace_last(tmp, ".dat.bak", ""); // in case of backups
        replace_last(tmp, ".dat", "");

        m_loc = boost::filesystem::path(tmp, boost::filesystem::native);

        std::ifstream ifs(p.string().c_str(), std::ios::binary);
        CHECK_THROW(ifs);
        CHECK_THROW(Utils::getVal<uint8_t>(ifs) == OP_PARTDATA);

        (void)Utils::getVal<uint16_t>(ifs);
        if (Utils::getVal<uint8_t>(ifs) != OP_PD_VER) {
                logWarning("Unknown partdata version.");
        }
        m_size = Utils::getVal<uint64_t>(ifs);

        uint16_t tagc = Utils::getVal<uint16_t>(ifs);

        while (tagc-- && ifs) {
                uint8_t   oc = getVal<uint8_t>(ifs);
                uint16_t len = getVal<uint16_t>(ifs);
                switch (oc) {
                        case OP_PD_DESTINATION:
                                m_dest = getVal<std::string>(ifs).value();
                                break;
                        case OP_PD_COMPLETED:
                                if (Utils::getVal<uint8_t>(ifs)!=OP_RANGELIST) {
                                        logWarning("Invalid tag.");
                                        ifs.seekg(len, std::ios::cur);
                                } else {
                                        (void)Utils::getVal<uint16_t>(ifs);
                                        m_complete = RangeList64(ifs);
                                }
                                break;
                        case OP_PD_VERIFIED:
                                if (Utils::getVal<uint8_t>(ifs)!=OP_RANGELIST) {
                                        logWarning("Invalid tag.");
                                        ifs.seekg(len, std::ios::cur);
                                } else {
                                        (void)Utils::getVal<uint16_t>(ifs);
                                        m_verified = RangeList64(ifs);
                                }
                                break;
                        case OP_PD_STATE: {
                                uint8_t state = Utils::getVal<uint8_t>(ifs);
                                if (state == STATE_STOPPED) {
                                        m_stopped = true;
                                } else if (state == STATE_PAUSED) {
                                        m_paused = true;
                                }
                                break;
                        }
                        default:
                                logWarning("Unhandled tag in PartData.");
                                ifs.seekg(len, std::ios::cur);
                                break;
                }
        }

        if (ifs && Utils::getVal<uint8_t>(ifs) == OP_METADATA) {
                (void)Utils::getVal<uint16_t>(ifs);
                m_md = new MetaData(ifs);
        }

        cleanupName();
//      printCompleted();

} catch (std::runtime_error &) {
        delete m_md;
}
MSVC_ONLY(;)

PartData::PartData(
        const boost::filesystem::path &path, MetaData *md,
        RangeList64 completed, RangeList64 verified
) : Object(0), m_size(md->getSize()), m_loc(path), m_complete(completed),
m_verified(verified), m_chunks(new ChunkMap), m_toFlush(), m_md(md),
m_pendingHashes(), m_sourceCnt(), m_fullSourceCnt(), m_paused(), m_stopped(),
m_autoPaused() {
        initSignals();

        for (uint32_t i = 0; i < md->getHashSetCount(); ++i) {
                HashSetBase *hs = md->getHashSet(i);
                if (hs->getChunkSize() && hs->getChunkCnt()) {
                        addHashSet(hs);
                }
        }

        std::string incDir(
                Prefs::instance().read<std::string>("/Incoming", "")
        );

        setDestination(boost::filesystem::path(incDir) / md->getName());
        save();
        getEventTable().postEvent(this, PD_ADDED);
}

// protected default constructor
PartData::PartData() : Object(0), m_size(), m_chunks(new ChunkMap), m_toFlush(),
m_md(), m_pendingHashes(), m_sourceCnt(), m_fullSourceCnt(), m_paused(),
m_stopped(), m_autoPaused() {
        initSignals();
}

PartData::~PartData() {
        getEventTable().delHandlers(this);
}

// boost::signals breaks when the first time the signal is invoced it's done
// from different module than the one where it was originally created in. while
// the majority of these signals are only invoced from inside partdata, they
// are part of partdata's public api, so anyone can invoke them (sadly), hence
// make sure they get properly initalized here.
//
// since many of the signals duplicate events, we can make our life bit easier
// here, and connect the postevent calls to the signals, so we don't have to do
// both things everywhere (which would quickly become errorprone).
void PartData::initSignals() {
        onPaused.connect(
                boost::bind(
                        &EventTable<PartData*, int>::postEvent,
                        &getEventTable(), _b1, PD_PAUSED
                )
        );
        onStopped.connect(
                boost::bind(
                        &EventTable<PartData*, int>::postEvent,
                        &getEventTable(), _b1, PD_STOPPED
                )
        );
        onResumed.connect(
                boost::bind(
                        &EventTable<PartData*, int>::postEvent,
                        &getEventTable(), _b1, PD_RESUMED
                )
        );
        onCompleted.connect(
                boost::bind(
                        &EventTable<PartData*, int>::postEvent,
                        &getEventTable(), _b1, PD_COMPLETE
                )
        );
        onCanceled.connect(
                boost::bind(
                        &EventTable<PartData*, int>::postEvent,
                        &getEventTable(), _b1, PD_CANCELED
                )
        );
}

void PartData::init() {
        CHECK_RET(m_md);

        // add hashes from metadata
        for (uint32_t i = 0; i < m_md->getHashSetCount(); ++i) {
                HashSetBase *hs = m_md->getHashSet(i);
                if (hs->getChunkSize() && hs->getChunkCnt()) {
                        addHashSet(hs);
                }
        }

        // It's possible we loaded a complete file - if so, re-try completing.
        if (isComplete()) {
                Utils::timedCallback(
                        boost::bind(&PartData::tryComplete, this), 0
                );
        } else {
                uint32_t modDate = Utils::getModDate(m_loc);
                bool needRehash = false;
                if (m_md && modDate != m_md->getModDate()) {
                        // allow 1ms errors (FAT32 causes them)
                        if (modDate + 1 != m_md->getModDate()) {
                                needRehash = true;
                        }
                }
                if (needRehash) {
                        logMsg(
                                boost::format(
                                        "%s: Modification date changed "
                                        "(%d != %d), rehashing completed parts."
                                ) % m_dest.leaf() % modDate % m_md->getModDate()
                        );
                        rehashCompleted();
                }
        }

        getEventTable().postEvent(this, PD_ADDED);
}

void PartData::rehashCompleted() {
        CMPosIndex &idx = m_chunks->get<ID_Pos>();
        for (CMPosIndex::iterator i = idx.begin(); i != idx.end(); ++i) {
                if ((*i).getHash()) {
                        idx.modify(i, bind(&Chunk::verify, __1, false));
                }
        }
}

void PartData::addSourceMask(
        uint64_t chunkSize, const std::vector<bool> &chunks
) {
        typedef CMLenIndex::iterator Iter;

        ++m_sourceCnt;
        if (chunks.empty()) {
                addFullSource(chunkSize);
                return;
        }

        // ed2k has 1 more chunk if size == N * chunkSize
        uint64_t expected = getChunkCount(chunkSize);
        if (m_size % chunkSize == 0 && chunkSize == ED2K_PARTSIZE) {
                ++expected;
        }
        if (chunks.size() != expected) {
                boost::format err(
                        "Invalid number of values in chunkmap "
                        "(expected %d, got %d)"
                );
                err % expected % chunks.size();
                throw std::runtime_error(err.str());
        }

        checkAddChunkMap(chunkSize);
        int i = 0;
        CMLenIndex& pi = m_chunks->get<ID_Length>();
        std::pair<Iter, Iter> ret = pi.equal_range(chunkSize);
        for (Iter j = ret.first; j != ret.second; ++j) {
                pi.modify(j, bind(&Chunk::m_avail, __1) += chunks[i++]);
        }
}

void PartData::addFullSource(uint64_t chunkSize) {
        typedef CMLenIndex::iterator Iter;

        checkAddChunkMap(chunkSize);
        CMLenIndex& pi = m_chunks->get<ID_Length>();
        std::pair<Iter, Iter> ret = pi.equal_range(chunkSize);
        for (Iter i = ret.first; i != ret.second; ++i) {
                pi.modify(i, ++bind(&Chunk::m_avail, __1));
        }
        ++m_fullSourceCnt;
}
void PartData::delSourceMask(
        uint64_t chunkSize, const std::vector<bool> &chunks
) {
        typedef CMLenIndex::iterator Iter;

        CHECK_THROW(m_sourceCnt);
        --m_sourceCnt;
        if (chunks.empty()) {
                delFullSource(chunkSize);
                return;
        }

        // ed2k has 1 more chunk if size == N * chunkSize
        uint64_t expected = getChunkCount(chunkSize);
        if (m_size % chunkSize == 0 && chunkSize == ED2K_PARTSIZE) {
                ++expected;
        }
        if (chunks.size() != expected) {
                boost::format err(
                        "Invalid number of values in chunkmap "
                        "(expected %d, got %d)"
                );
                err % expected % chunks.size();
                throw std::runtime_error(err.str());
        }

        int i = 0;
        CMLenIndex& pi = m_chunks->get<ID_Length>();
        std::pair<Iter, Iter> ret = pi.equal_range(chunkSize);
        for (Iter j = ret.first; j != ret.second; ++j) {
                if(chunks[i]) { // ensures we don't integer underflow
                        CHECK_THROW((*j).m_avail);
                }
                pi.modify(j, bind(&Chunk::m_avail, __1) -= chunks[i++]);
        }
}
void PartData::delFullSource(uint64_t chunkSize) {
        typedef CMLenIndex::iterator Iter;

        CHECK_THROW(m_fullSourceCnt);
        --m_fullSourceCnt;
        CMLenIndex& pi = m_chunks->get<ID_Length>();
        std::pair<Iter, Iter> ret = pi.equal_range(chunkSize);
        for (Iter i = ret.first; i != ret.second; ++i) {
                CHECK_THROW((*i).m_avail);
                pi.modify(i, --bind(&Chunk::m_avail, __1));
        }
}

/**
 * Unary function object for usage with doGetRange() method
 *
 * @param r  Ignored
 * @returns Always true
 */
struct TruePred { bool operator()(const Range64 &) { return true; } };

/**
 * Unary function object for usage with doGetRange() method, in order to check
 * if the generated range is contained within the chunkmap contained within this
 * functor.
 *
 * @param r    Range candidate to be checked
 * @return     True if @param r is contained within m_rl, false otherwise
 */
struct CheckPred {
        CheckPred(const std::vector<bool> &chunks, uint32_t chunkSize)
        : m_chunks(chunks), m_chunkSize(chunkSize) {}

        bool operator()(const Range64 &r) {
                if (r.begin() > m_chunks.size() * m_chunkSize) {
                        return false;
                }
                uint32_t chunk1 = r.begin() / m_chunkSize;
                uint32_t chunk2 = r.end() / m_chunkSize + 1;
                while (chunk1 != chunk2) {
                        if (!m_chunks[chunk1++]) {
                                return false;
                        }
                }
                return true;
        }
        const std::vector<bool> &m_chunks;
        uint64_t m_chunkSize;
};


UsedRangePtr PartData::getRange(uint32_t size) {
        TruePred pred;
        return doGetRange(size, pred);
}

UsedRangePtr PartData::getRange(
        uint32_t size, const std::vector<bool> &chunks
) {
        if (chunks.empty()) {
                TruePred pred;
                return doGetRange(size, pred);
        }
        CheckPred pred(chunks, size);
        return doGetRange(size, pred);
}

template<typename Predicate>
UsedRangePtr PartData::getNextChunk(uint64_t size, Predicate &pred) {
        UsedRangePtr ret;
        if (m_complete.empty()) {
                ret = UsedRangePtr(new UsedRange(this, 0, size));
                if (ret->end() > m_size - 1) {
                        ret->end(m_size - 1);
                }
        } else if (m_complete.front().begin() > 0) {
                uint64_t end = m_complete.front().begin() - 1;
                ret = UsedRangePtr(new UsedRange(this, 0, end));
        } else if (m_complete.front().end() < m_size) {
                uint64_t beg = m_complete.front().end() + 1;
                uint64_t end = beg + size > m_size ? m_size : beg +size;
                ret = UsedRangePtr(new UsedRange(this, beg, end));
        }
        if (!ret->getLock(1)) {
                ret.reset();
        }
        return ret;
}

template<typename Predicate>
UsedRangePtr PartData::doGetRange(uint64_t size, Predicate &pred) {
        typedef CMSelectIndex::iterator SIter;

        if (m_chunks->empty()) {
                return getNextChunk(size, pred);
        }
        CMSelectIndex &idx = m_chunks->template get<ID_Selector>();
        std::pair<SIter, SIter> r = idx.equal_range(
                boost::make_tuple(false, true)
        );
        if (!std::distance(r.first, r.second)) {
                return getNextChunk(size, pred);
        }

        uint32_t hopCnt = 0;
        for (SIter i = r.first; i != r.second; ++i) {
                if (pred(*i) && canLock(*i)) {
                        UsedRangePtr ret(new UsedRange(this, i));
                        assert(ret->getLock(1));
                        logTrace(TRACE_CHUNKS,
                                boost::format(
                                        "Selected chunk #%d (avail=%d, "
                                        "usecnt=%d partial=%s complete="
                                        "%s) in %d hops"
                                ) % ((*i).begin() / (*i).m_size)
                                % (*i).m_avail % (*i).m_useCnt
                                % ((*i).m_partial ? "yes" : "no")
                                % ((*i).m_complete ? "yes" : "no")
                                % hopCnt
                        );
                        return ret;
                }
                ++hopCnt;
        }
        logTrace(TRACE_CHUNKS,
                boost::format("Failure to select chunk, did %d hops") % hopCnt
        );
        return UsedRangePtr();
}

uint64_t PartData::getChunkCount(uint64_t chunkSize) const {
        return m_size / chunkSize + (m_size % chunkSize ? 1 : 0);
}

void PartData::checkAddChunkMap(uint64_t cs) {
        typedef CMLenIndex::iterator LIter;

        logTrace(TRACE_PARTDATA, boost::format("Adding chunkmap (cs=%d)") % cs);

        if (m_partStatus.find(cs) == m_partStatus.end()) {
                m_partStatus[cs] = std::vector<bool>(getChunkCount(cs));
        }
        std::pair<LIter, LIter> ret =m_chunks->get<ID_Length>().equal_range(cs);
        if (ret.first == m_chunks->get<ID_Length>().end()) {
                for (uint64_t i = 0; i < getChunkCount(cs); ++i) {
                        uint64_t beg = i * cs;
                        uint64_t end = (i + 1) * cs - 1;
                        if (end > m_size - 1) {
                                end = m_size - 1;
                        }
                        Chunk c(this, beg, end, cs);
                        m_chunks->insert(c);
                        m_partStatus[cs][i] = isComplete(beg, end);
                }
        }
}

void PartData::addHashSet(const HashSetBase *hs) {
        if (!getSize()) {
                logDebug("Cannot add hashes to files of size 0.");
                return;
        }

        CHECK_THROW(hs->getChunkSize() > 0);
        typedef CMLenIndex::iterator LIter;

        logTrace(TRACE_PARTDATA,
                boost::format("Adding hashset (cs=%d cc=%d)")
                % hs->getChunkSize() % hs->getChunkCnt()
        );

        uint32_t cs = hs->getChunkSize();
        std::pair<LIter, LIter> ret =m_chunks->get<ID_Length>().equal_range(cs);

        if (ret.first == m_chunks->get<ID_Length>().end()) {
                checkAddChunkMap(cs);
                ret = m_chunks->get<ID_Length>().equal_range(cs);
        }

        int cc = hs->getChunkCnt();
        int check = std::distance(ret.first, ret.second);

        // file of size of multiple of chunksize has one more chunkhash in ed2k
        // TODO: is this portable across networks?
        CHECK_THROW(m_size % cs ? check == cc : check == cc - 1);

        uint32_t j = 0;
        bool saved = false;
        for (LIter i = ret.first; i != ret.second; ++i) {
                m_chunks->get<ID_Length>().modify(
                        i, bind(&Chunk::m_hash, __1) = &(*hs)[j++]
                );
                if ((*i).isComplete() && !(*i).isVerified()) {
                        if (!saved) {
                                save();
                                saved = true;
                        }
                        m_chunks->get<ID_Length>().modify(
                                i, bind(&Chunk::verify, __1, false)
                        );
                }
        }
}

bool PartData::canLock(const Range64 &r, uint32_t size) const {
        Range64 cand(r.begin(), r.begin());
        typedef RangeList64::CIter CIter;
        CIter i = m_complete.getContains(cand);
        CIter j = m_locked.getContains(cand);
        CIter k = m_dontDownload.getContains(cand);
        do {
                i = m_complete.getContains(cand);
                j = m_locked.getContains(cand);
                k = m_dontDownload.getContains(cand);
                if (i != m_complete.end()) {
                        cand = Range64((*i).end() + 1, (*i).end() + 1);
                } else if (j != m_locked.end()) {
                        cand = Range64((*j).end() + 1, (*j).end() + 1);
                } else if (k != m_dontDownload.end()) {
                        cand = Range64((*k).end() + 1, (*k).end() + 1);
                } else {
                        break;
                }
        } while (r.contains(cand));
        return r.contains(cand);
}

LockedRangePtr PartData::getLock(UsedRangePtr used, uint32_t size) {
        Range64 cand(used->begin(), used->begin());
        typedef RangeList64::CIter CIter;
        CIter i = m_complete.getContains(cand);
        CIter j = m_locked.getContains(cand);
        CIter k = m_dontDownload.getContains(cand);
        do {
                i = m_complete.getContains(cand);
                j = m_locked.getContains(cand);
                k = m_dontDownload.getContains(cand);
                if (i != m_complete.end()) {
                        cand = Range64((*i).end() + 1, (*i).end() + 1);
                } else if (j != m_locked.end()) {
                        cand = Range64((*j).end() + 1, (*j).end() + 1);
                } else if (k != m_dontDownload.end()) {
                        cand = Range64((*k).end() + 1, (*k).end() + 1);
                } else {
                        break;
                }
        } while (used->contains(cand));
        if (!used->contains(cand)) {
                return LockedRangePtr();
        }
        cand.end(cand.begin() + size - 1);
        if (cand.end() > used->end()) {
                cand.end(used->end());
        }
        i = m_complete.getContains(cand);
        if (i != m_complete.end()) {
                cand.end((*i).begin() - 1);
        }
        i = m_locked.getContains(cand);
        if (i != m_locked.end()) {
                cand.end((*i).begin() - 1);
        }
        k = m_dontDownload.getContains(cand);
        if (k != m_dontDownload.end()) {
                cand.end((*k).begin() - 1);
        }
        CHECK_THROW(cand.length() <= size);
        return LockedRangePtr(
                new LockedRange(this, cand, *used->m_chunk, used)
        );
}

// publically available write method, this allows writing to the file without
// aquiring Used/Locked ranges - useful for simple protocols.
// Basically it just forwards the call to doWrite(), however it does a bunch of
// checks prior to the call, as well as bunch of chunkmap updates after the
// call in order to keep everyone happy.
void PartData::write(uint64_t begin, const std::string &data) {
        logTrace(TRACE_PARTDATA,
                boost::format("Safe-writing at offset %d.") % begin
        );

        CHECK_THROW(!m_locked.contains(begin, begin + data.size() - 1));
        CHECK_THROW(!m_complete.contains(begin, begin + data.size() - 1));

        doWrite(begin, data);
        tryComplete();
}

void PartData::updateChunks(Range64 range) {
        CMPosIndex &pi = m_chunks->get<ID_Pos>();
        CMPosIndex::iterator i = pi.lower_bound(Chunk(this, range, 0));
        if (i != pi.end() && (*i).contains(range)) {
                (*i).updateState();
        }
        CMPosIndex::iterator j = i;
        while (i != pi.begin() && (*--i).contains(range)) {
                (*i).updateState();
        }
        while (j != pi.end() && ++j != pi.end() && (*j).contains(range)) {
                (*j).updateState();
        }
}

void PartData::setComplete(Range64 range) {
        m_complete.merge(range);
        m_dontDownload.erase(range);
        updateChunks(range);
}

void PartData::setCorrupt(Range64 range) {
        m_complete.erase(range);
        m_corrupt.merge(range);
        m_verified.erase(range);
        onCorruption(this, range);
        updateChunks(range);
}

void PartData::setVerified(Range64 range) {
        m_verified.merge(range);
        m_corrupt.erase(range);
        m_complete.merge(range);
        updateChunks(range);
}

// don't try to call doComplete() from here, since this method is called from
// Chunk::write(), prior to the chunk attempting to hash itself (if it has
// hashes). Thus, doComplete() check here would incorrectly succeed, since
// m_pendingHashes hasn't been increased, which would then lead to full rehash
// being done prior to last chunkhash, resulting in crash later on during file
// moving.
void PartData::doWrite(uint64_t begin, const std::string &data) {
        // since we allow (on certain conditions) resizing partdata on the fly,
        // this check ensures that we don't allow writing to a file of size 0
        CHECK_THROW(m_size);
        // don't allow writing to paused/stopped file
        CHECK_THROW(isRunning());

        logTrace(TRACE_PARTDATA,
                boost::format("Writing at offset %d, datasize is %d")
                % begin % data.size()
        );

        CHECK_THROW(!m_complete.contains(begin, begin + data.size() - 1));

        m_buffer[begin] = data;
        setComplete(Range64(begin, begin + data.size() - 1));
        m_toFlush += data.size();
        getEventTable().postEvent(this, PD_DATA_ADDED);
        if (m_toFlush >= BUF_SIZE_LIMIT) {
                save();
        }
        dataAdded(this, begin, data.size());
}