ChunkManagerImpl.cc

Go to the documentation of this file.

// -*- lsst-c++ -*-

/* 
 * LSST Data Management System
 * Copyright 2008, 2009, 2010 LSST Corporation.
 * 
 * This product includes software developed by the
 * LSST Project (http://www.lsst.org/).
 *
 * 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 3 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 LSST License Statement and 
 * the GNU General Public License along with this program.  If not, 
 * see <http://www.lsstcorp.org/LegalNotices/>.
 */
 

#ifndef LSST_AP_CHUNK_MANAGER_IMPL_CC
#define LSST_AP_CHUNK_MANAGER_IMPL_CC

#include <iostream>

#include "boost/format.hpp"

#include "lsst/pex/exceptions.h"

#include "Chunk.cc"
#include "ChunkManagerImpl.h"
#include "SpatialUtil.h"


namespace lsst { namespace ap { namespace detail {

// -- HashedSet ----------------

inline int hash(boost::uint32_t key) {
    key = (~key) + (key << 15); // key = (key << 15) - key - 1;
    key = key ^ (key >> 12);
    key = key + (key << 2);
    key = key ^ (key >> 4);
    key = key * 2057;           // key = (key + (key << 3)) + (key << 11);
    key = key ^ (key >> 16);
    return static_cast<int>(key);
}

inline int hash(int key) {
    return hash(static_cast<boost::uint32_t>(key));
}


template <typename EntryT, int NumEntries>
HashedSet<EntryT, NumEntries>::HashedSet() :
    _free(0), _size(0)
{
    for (int i = 0; i < 2*NumEntries; ++i) {
        _hashTable[i] = -1;
    }
    // initialize linked list of free entries (embedded in the entries themselves)
    for (int i = 0; i < NumEntries - 1; ++i) {
        _entries[i].setId(-1);
        _entries[i].setNextInChain(i + 1);
    }
    _entries[NumEntries - 1].setNextInChain(-1);
}


template <typename EntryT, int NumEntries>
EntryT const * HashedSet<EntryT, NumEntries>::doFind(int const id) const {
    int i = _hashTable[hash(id) & (2*NumEntries - 1)];
    while (i >= 0) {
        if (id == _entries[i].getId()) {
            return &_entries[i];
        }
        i = _entries[i].getNextInChain();
    }
    return 0;
}


template <typename EntryT, int NumEntries>
EntryT * HashedSet<EntryT, NumEntries>::insert(int const id) {

    // check that there is space for another entry
    if (_free < 0) {
        return 0;
    }

    int const bucket = hash(id) & (2*NumEntries - 1);

    int i    = _hashTable[bucket];
    int last = -1;

    while (i >= 0) {
        if (id == _entries[i].getId()) {
            return 0; // already have an entry with the given id
        }
        last = i;
        i    = _entries[i].getNextInChain();
    }

    // take an entry off the free list
    int const c = _free;
    _free = _entries[c].getNextInChain();

    if (last < 0) {
        _hashTable[bucket] = c;
    } else {
        // hash collision - chain to the end of the bucket
        _entries[last].setNextInChain(c);
    }

    // basic entry initialization, then return
    new (&_entries[c]) EntryT();
    _entries[c].setId(id);
    _entries[c].setNextInChain(-1);
    ++_size;
    return &_entries[c];
}


template <typename EntryT, int NumEntries>
std::pair<EntryT *, bool> HashedSet<EntryT, NumEntries>::findOrInsert(int const id) {

    int const bucket = hash(id) & (2*NumEntries - 1);

    int i    = _hashTable[bucket];
    int last = -1;

    while (i >= 0) {
        if (id == _entries[i].getId()) {
            return std::pair<EntryT *, bool>(&_entries[i], false); // found an entry with the given id
        }
        last = i;
        i    = _entries[i].getNextInChain();
    }

    // check that there is a free chunk, if so use it
    int const c = _free;
    if (c < 0) {
        return std::pair<EntryT *, bool>(0, true);
    }
    _free = _entries[c].getNextInChain();

    if (last < 0) {
        _hashTable[bucket] = c;
    } else {
        // hash collision - chain to the end of the bucket
        _entries[last].setNextInChain(c);
    }

    // basic chunk initialization, then return
    new (&_entries[c]) EntryT();
    _entries[c].setId(id);
    _entries[c].setNextInChain(-1);
    ++_size;
    return std::pair<EntryT *, bool>(&_entries[c], true);
}


template <typename EntryT, int NumEntries>
bool HashedSet<EntryT, NumEntries>::erase(int const id) {

    int const bucket = hash(id) & (2*NumEntries - 1);

    int i    = _hashTable[bucket];
    int last = -1;

    while (i >= 0) {
        if (id == _entries[i].getId()) {
            // found entry to erase, unlink it from the chain for its bucket
            if (last < 0) {
                _hashTable[bucket] = _entries[i].getNextInChain();
            } else {
                _entries[last].setNextInChain(_entries[i].getNextInChain());
            }
            // append the entry to the free list
            _entries[i].setId(-1);
            _entries[i].setNextInChain(_free);
            _free = i;
            --_size;
            return true;
        }
        last = i;
        i    = _entries[i].getNextInChain();
    }
    return false;
}


// -- BlockAllocator ----------------

template <typename MutexT, typename DataT, typename TraitsT>
BlockAllocator<MutexT, DataT, TraitsT>::BlockAllocator(
    unsigned char const * const reference,
    std::size_t const offset
) :
    _mutex(),
    _allocator(),
    _offset(static_cast<std::size_t>((reference + offset) - reinterpret_cast<unsigned char * >(this)))
{
    _allocator.reset();
}


template <typename MutexT, typename DataT, typename TraitsT>
std::size_t BlockAllocator<MutexT, DataT, TraitsT>::allocate() {
    int i[1];
    ScopedLock<MutexT> lock(_mutex);
    if (!_allocator.set(i, 1)) {
        throw LSST_EXCEPT(lsst::pex::exceptions::MemoryError, "no free blocks remain");
    }
    return _offset + i[0]*BLOCK_SIZE;
}


template <typename MutexT, typename DataT, typename TraitsT>
void BlockAllocator<MutexT, DataT, TraitsT>::allocate(
    std::size_t * const blockOffsets,
    int const n
) {
    if (n < 0 || n > TraitsT::MAX_BLOCKS_PER_CHUNK) {
        throw LSST_EXCEPT(lsst::pex::exceptions::RangeError,
            "invalid number of memory blocks in allocation request");
    }
    int i[TraitsT::MAX_BLOCKS_PER_CHUNK];

    ScopedLock<MutexT> lock(_mutex);
    if (!_allocator.set(i, n)) {
        throw LSST_EXCEPT(lsst::pex::exceptions::MemoryError,
            "number of free blocks too small to satisfy allocation request");
    }
    for (int j = 0; j < n; ++j) {
        blockOffsets[j] = _offset + i[j]*BLOCK_SIZE;
    }
}


template <typename MutexT, typename DataT, typename TraitsT>
void BlockAllocator<MutexT, DataT, TraitsT>::free(
    std::size_t const * const blockOffsets,
    int const n
) {
    assert(n >= 0 && n <= TraitsT::MAX_BLOCKS_PER_CHUNK &&
        "invalid number of memory blocks in free request");

    // translate block offsets to block indexes
    int i[TraitsT::MAX_BLOCKS_PER_CHUNK];
    for (int j = 0; j < n; ++j) {
        std::size_t off = blockOffsets[j] - _offset;
        assert(off < TraitsT::NUM_BLOCKS*BLOCK_SIZE &&
               "block was not allocated by this allocator");
        assert(off % BLOCK_SIZE == 0 && "invalid block address");
        i[j] = static_cast<int>(off/BLOCK_SIZE);
    }

    // clear bit corresponding to each block to free
    ScopedLock<MutexT> lock(_mutex);
    _allocator.reset(i, n);
}


// -- VisitTracker ----------------

bool VisitTracker::isValid(int const visitId) const {
    Visit const * v = this->find(visitId);
    if (v == 0) {
        return false;
    }
    return !v->failed();
}

void VisitTracker::print(std::ostream & os) const {
    std::vector<int> v;
    v.reserve(size());
    Visit const * const e = end();
    for (Visit const * beg = begin(); beg != e; ++beg) {
        int const id = beg->getId();
        if (id >= 0) {
            v.push_back(id);
        }
    }
    if (v.empty()) {
        os << "    No visits being tracked";
    } else {
        std::sort(v.begin(), v.end());
        boost::format fmt("    visit %1% %|32t|: %2%\n");
        for (std::vector<int>::const_iterator i = v.begin(); i != v.end(); ++i) {
            Visit const * v = find(*i);
            os << fmt % v->getId() % (v->failed() ? "failed" : "in-flight");
        }
    }
    os << std::endl;
}

void VisitTracker::print(int const visitId, std::ostream & os) const {
    Visit const * v = find(visitId);
    boost::format fmt("    visit %1% %|32t|: %2%\n");
    fmt % visitId;
    if (v == 0) {
        os << fmt % "not being tracked";
    } else {
        os << fmt % (v->failed() ? "failed" : "in-flight");
    }
    os << std::endl;
}


// -- SubManager ----------------

template <typename MutexT, typename DataT, typename TraitsT>
void SubManager<MutexT, DataT, TraitsT>::createOrRegisterInterest(
    std::vector<Chunk> & toRead,
    std::vector<Chunk> & toWaitFor,
    int const visitId,
    std::vector<int> const & chunkIds
) {
    std::vector<int>::const_iterator const end = chunkIds.end();
    for (std::vector<int>::const_iterator i = chunkIds.begin(); i != end; ++i) {
        std::pair<Descriptor *, bool> p(_chunks.findOrInsert(*i));
        if (p.second) {
            // new chunk descriptor was allocated
            if (p.first == 0) {
                throw LSST_EXCEPT(lsst::pex::exceptions::MemoryError,
                                  "too many chunks in flight");
            }
            p.first->_visitId = visitId;
            p.first->_usable  = false;
            toRead.push_back(Chunk(p.first, &_allocator));
        } else {
            // existing chunk descriptor was found
            assert(p.first != 0);
            p.first->_interestedParties.enqueue(visitId);
            toWaitFor.push_back(Chunk(p.first, &_allocator));
        }
    }
}


template <typename MutexT, typename DataT, typename TraitsT>
bool SubManager<MutexT, DataT, TraitsT>::checkForOwnership(
    std::vector<Chunk> & toRead,
    std::vector<Chunk> & toWaitFor,
    int const visitId
) {
    std::size_t size = toWaitFor.size();
    std::size_t i = 0;

    while (i < size) {
        Chunk & c = toWaitFor[i];
        if (c.getVisitId() != visitId) {
            ++i;
        } else {
            if (!c.isUsable()) {
                c.clear();
                toRead.push_back(c);
            }
            // deletes element i in O(1) time (but changes element ordering)
            --size;
            if (i < size) {
                toWaitFor[i] = toWaitFor[size];
            }
            toWaitFor.pop_back();
        }
    }
    return toWaitFor.empty();
}


template <typename MutexT, typename DataT, typename TraitsT>
void SubManager<MutexT, DataT, TraitsT>::getChunks(
    std::vector<Chunk> & chunks,
    std::vector<int> const & chunkIds
) {
    std::vector<int>::const_iterator const end = chunkIds.end();
    for (std::vector<int>::const_iterator i = chunkIds.begin(); i != end; ++i) {
        Descriptor * d = _chunks.find(*i);
        if (d != 0) {
            chunks.push_back(Chunk(d, &_allocator));
        }
    }
}


template <typename MutexT, typename DataT, typename TraitsT>
bool SubManager<MutexT, DataT, TraitsT>::relinquishOwnership(
    int const visitId,
    bool const rollback,
    VisitTracker const & tracker
) {
    bool change = false;
    Descriptor * const end = _chunks.end();
    for (Descriptor * i = _chunks.begin(); i != end; ++i) {
        if (i->getId() != -1 && i->_visitId == visitId) {
            bool foundSuccessor = false;
            while (!i->_interestedParties.empty()) {
                int const nextVisitId = static_cast<int>(i->_interestedParties.dequeue());
                if (tracker.isValid(nextVisitId)) {
                    i->_visitId    = nextVisitId;
                    change         = true;
                    foundSuccessor = true;
                    break;
                }
            }
            if (foundSuccessor) {
                Chunk c(i, &_allocator);
                if (rollback) {
                    c.rollback();
                } else {
                    c.commit();
                }
            } else {
                // deallocate chunk
                _allocator.free(i->_blocks, i->_numBlocks);
                _chunks.erase(i->getId());
            }
        }
    }
    return change;
}


namespace {

    template <typename T> struct PtrLessThan {
        bool operator()(T const * t1, T const * t2) { return *t1 < *t2; }
    };

    template <typename DescriptorT>
    bool mergePrint(DescriptorT const * d1, DescriptorT const * d2) {
        if (d1->_visitId != d1->_visitId || d1->_usable != d2->_usable) {
            return false;
        }
        if (d2->_interestedParties.empty() != d2->_interestedParties.empty()) {
            return false;
        }
        return ZoneStripeChunkDecomposition::chunkToStripe(d1->_chunkId) ==
               ZoneStripeChunkDecomposition::chunkToStripe(d2->_chunkId);
    }

    template <typename DescriptorT>
    void printChunks(std::ostream & os, std::vector<DescriptorT const *> const & v) {

        boost::format oneFmt ("        chunk  %1%     in stripe %2% %|32t|: %3%%4%\n");
        boost::format manyFmt("        chunks %1%-%2% in stripe %3% %|32t|: %4%%5%\n");

        int start = 0;
        DescriptorT const * c = v[0];
        int const sz = static_cast<int>(v.size());
        for (int i = 1; i <= sz; ++i) {
            if (i < sz && mergePrint(c, v[i])) {
                continue;
            }
            if (i - start > 1) {
                manyFmt % ZoneStripeChunkDecomposition::chunkToSequence(c->_chunkId);
                manyFmt % ZoneStripeChunkDecomposition::chunkToSequence(v[i - 1]->_chunkId);
                manyFmt % ZoneStripeChunkDecomposition::chunkToStripe(c->_chunkId);
                manyFmt % (c->_usable ? "  usable" : "unusable");
                manyFmt % (c->_interestedParties.empty() ? "" : ", interesting");
                os << manyFmt;
            } else {
                oneFmt % ZoneStripeChunkDecomposition::chunkToSequence(c->_chunkId);
                oneFmt % ZoneStripeChunkDecomposition::chunkToStripe(c->_chunkId);
                oneFmt % (c->_usable ? "  usable" : "unusable");
                oneFmt % (c->_interestedParties.empty() ? "" : ", interesting");
                os << oneFmt;
            }
            if (i < sz) {
                if (c->_visitId != v[i]->_visitId) {
                    os << "    Owned by visit " << v[i]->_visitId << ":\n";
                }
                start = i;
                c = v[i];
            }
        }
    }

} // end of anonymous namespace


template <typename MutexT, typename DataT, typename TraitsT>
void SubManager<MutexT, DataT, TraitsT>::print(std::ostream & os) const {
    Descriptor const * const end = _chunks.end();
    std::vector<Descriptor const *> v;
    v.reserve(_chunks.size());
    for (Descriptor const * beg = _chunks.begin(); beg != end; ++beg) {
        if (beg->_chunkId != -1) {
            v.push_back(beg);
        }
    }
    std::sort(v.begin(), v.end(), PtrLessThan<Descriptor>());
    os << "    Chunks with an owner";
    if (v.empty()) {
        os << ": None";
    } else {
        os << ":\n";
        os << "    Owned by visit " << v[0]->_visitId << ":\n";
        printChunks(os, v);
    }
    os << std::endl;
}


template <typename MutexT, typename DataT, typename TraitsT>
void SubManager<MutexT, DataT, TraitsT>::print(int const chunkId, std::ostream & os) const {
    Descriptor const * c = _chunks.find(chunkId);
    os << "    [" << c->_chunkId << "] chunk " <<
          ZoneStripeChunkDecomposition::chunkToSequence(chunkId) << " in stripe " <<
          ZoneStripeChunkDecomposition::chunkToStripe(chunkId);
    if (c == 0) {
        os << ": not being tracked\n";
    } else {
        os << ":\n        " << (c->_usable ? "usable" : "unusable") << "\n        ";
        if (c->_interestedParties.empty()) {
            os << "un";
        }
        os << "interesting\n        ";
        int sz = c->_size;
        os << sz << " entries in " << c->_nextBlock << " blocks (" <<
              c->_numBlocks << " allocated)\n        ";
        if (sz <= c->_delta) {
            sz = 0;
        } else {
            sz -= c->_delta;
        }
        os << sz << " entries in delta\n";
    }
    os << std::endl;
}


template <typename MutexT, typename DataT, typename TraitsT>
void SubManager<MutexT, DataT, TraitsT>::printVisit(int const visitId, std::ostream & os) const {
    Descriptor const * const end = _chunks.end();
    std::vector<Descriptor const *> v;
    v.reserve(_chunks.size());
    for (Descriptor const * beg = _chunks.begin(); beg != end; ++beg) {
        if (beg->_chunkId != -1 && beg->_visitId == visitId) {
            v.push_back(beg);
        }
    }
    std::sort(v.begin(), v.end(), PtrLessThan<Descriptor>());
    os << "    Chunks belonging to visit " << visitId;
    if (v.empty()) {
        os << ": None";
    } else {
        os << ":\n";
        printChunks(os, v);
    }
    os << std::endl;
}


// -- ChunkManagerImpl ----------------

template <typename MutexT, typename DataT, typename TraitsT>
ChunkManagerImpl<MutexT, DataT, TraitsT>::ChunkManagerImpl() :
    _data(reinterpret_cast<unsigned char *>(this), blocks())
{}


template <typename MutexT, typename DataT, typename TraitsT>
bool ChunkManagerImpl<MutexT, DataT, TraitsT>::isVisitInFlight(int const visitId) {
    ScopedLock<MutexT> lock(_mutex);
    return _visits.isValid(visitId);
}


template <typename MutexT, typename DataT, typename TraitsT>
void ChunkManagerImpl<MutexT, DataT, TraitsT>::failVisit(int const visitId) {
    ScopedLock<MutexT> lock(_mutex);
    Visit * v = _visits.find(visitId);
    if (v != 0) {
        v->setFailed();
    }
}


template <typename MutexT, typename DataT, typename TraitsT>
void ChunkManagerImpl<MutexT, DataT, TraitsT>::registerVisit(int const visitId) {
    ScopedLock<MutexT> lock(_mutex);
    if (_visits.find(visitId) != 0) {
        throw LSST_EXCEPT(lsst::pex::exceptions::InvalidParameterError,
            (boost::format("Cannot start processing visit %1%: visit is already in flight") % visitId).str());
    }
    if (_visits.space() == 0) {
        throw LSST_EXCEPT(lsst::pex::exceptions::LengthError,
            (boost::format("Cannot register visit %1%: too many visits in-flight") % visitId).str());
    }
    Visit * v = _visits.insert(visitId);
    assert(v != 0);
}


template <typename MutexT, typename DataT, typename TraitsT>
void ChunkManagerImpl<MutexT, DataT, TraitsT>::startVisit(
    std::vector<Chunk> & toRead,
    std::vector<Chunk> & toWaitFor,
    int const visitId,
    std::vector<int> const & chunkIds
) {
    toRead.clear();
    toWaitFor.clear();

    // ensure external resources necessary for success are available
    toRead.reserve(chunkIds.size());
    toWaitFor.reserve(chunkIds.size());

    ScopedLock<MutexT> lock(_mutex);
    // ensure internal resources necessary for success are available
    if (_data.space() < static_cast<int>(chunkIds.size())) {
        throw LSST_EXCEPT(lsst::pex::exceptions::LengthError,
                          "requested additional chunks exceed chunk manager capacity");
    }
    if (!_visits.isValid(visitId)) {
        throw LSST_EXCEPT(lsst::pex::exceptions::InvalidParameterError,
            (boost::format("Cannot start processing for visit %1%: visit is not in-flight") % visitId).str());
    }
    // having pre-allocated/checked that there is space for everything,
    // manager state can be modified without throwing
    _data.createOrRegisterInterest(toRead, toWaitFor, visitId, chunkIds);
}


template <typename MutexT, typename DataT, typename TraitsT>
void ChunkManagerImpl<MutexT, DataT, TraitsT>::waitForOwnership(
    std::vector<Chunk> & toRead,
    std::vector<Chunk> & toWaitFor,
    int const visitId,
    TimeSpec const & deadline
) {
    toRead.clear();
    toRead.reserve(toWaitFor.size());

    ScopedLock<MutexT> lock(_mutex);
    while (true) {
        if (_data.checkForOwnership(toRead, toWaitFor, visitId)) {
            break; // all chunks belong to the visit - ok to proceed
        }
        // wait for ownership
        if (!_ownerCondition.wait(lock, deadline)) {
            // TODO: this is a short-term DC3a hack, necessary because there is no way for
            // the pipeline framework to communicate exceptions arising outside of the implementation
            // of AP (e.g. in an IOStage) back to the pipeline itself. This results in visits that
            // fail, but never relinquish ownership of their chunks. For now, take the draconian measure
            // of rolling back all visits except the current one. This needs to be re-thought for DC3b!
            rollbackAllExcept(visitId);
            throw LSST_EXCEPT(lsst::pex::exceptions::TimeoutError,
                (boost::format("Deadline for visit %1% expired") % visitId).str());
        }
    }
}


template <typename MutexT, typename DataT, typename TraitsT>
void ChunkManagerImpl<MutexT, DataT, TraitsT>::getChunks(
    std::vector<Chunk> & chunks,
    std::vector<int> const & chunkIds
) {
    ScopedLock<MutexT> lock(_mutex);
    _data.getChunks(chunks, chunkIds);
}


template <typename MutexT, typename DataT, typename TraitsT>
bool ChunkManagerImpl<MutexT, DataT, TraitsT>::endVisit(
    int const visitId,
    bool const rollback
) {
    ScopedLock<MutexT> lock(_mutex);
    bool roll = rollback || !_visits.isValid(visitId);
    if (!_visits.erase(visitId)) {
        return false;
    }
    // relinquish chunk ownership: if any chunks change hands, notify all threads
    // waiting on chunk ownership to check whether they can proceed
    if (_data.relinquishOwnership(visitId, roll, _visits)) {
        _ownerCondition.notifyAll();
    }
    return !roll;
}


template <typename MutexT, typename DataT, typename TraitsT>
void ChunkManagerImpl<MutexT, DataT, TraitsT>::rollbackAllExcept(int const visitId) {
    for (Visit const * v = _visits.begin(); v != _visits.end(); ++v) {
        int id = v->getId();
        if (id >= 0 && id != visitId) {
            _visits.erase(id);
            _data.relinquishOwnership(id, true, _visits);
        }
    }
}


template <typename MutexT, typename DataT, typename TraitsT>
void ChunkManagerImpl<MutexT, DataT, TraitsT>::printVisits(std::ostream & os) const {
    ScopedLock<MutexT> lock(_mutex);
    _visits.print(os);
}


template <typename MutexT, typename DataT, typename TraitsT>
void ChunkManagerImpl<MutexT, DataT, TraitsT>::printChunks(std::ostream & os) const {
    ScopedLock<MutexT> lock(_mutex);
    _data.print(os);
}


template <typename MutexT, typename DataT, typename TraitsT>
void ChunkManagerImpl<MutexT, DataT, TraitsT>::printVisit(int const visitId, std::ostream  & os) const {
    ScopedLock<MutexT> lock(_mutex);
    _visits.print(visitId, os);
    _data.printVisit(visitId, os);
}


template <typename MutexT, typename DataT, typename TraitsT>
void ChunkManagerImpl<MutexT, DataT, TraitsT>::printChunk(int const chunkId, std::ostream  & os) const {
    ScopedLock<MutexT> lock(_mutex);
    _data.print(chunkId, os);
}


}}} // end of namespace lsst::ap::detail

#endif // LSST_AP_CHUNK_MANAGER_IMPL_CC