Chunk.h

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_H
#define LSST_AP_CHUNK_H

#include <cassert>
#include <stdexcept>
#include <algorithm>

#include "boost/noncopyable.hpp"
#include "boost/static_assert.hpp"
// #include "boost/type_traits/has_trivial_assign.hpp"
// #include "boost/type_traits/has_trivial_copy.hpp"
// #include "boost/type_traits/has_trivial_destructor.hpp"

#include "Common.h"
#include "DataTraits.h"
#include "Fifo.h"


namespace lsst {
namespace ap {


struct BinChunkHeader {

    static boost::uint32_t const MAGIC = 0xdecade14;

    boost::uint32_t _magic;
    int _numRecords;
    int _numDeletes;
    int _recordSize;

    BinChunkHeader() :
        _magic(MAGIC),
        _numRecords(0),
        _numDeletes(0),
        _recordSize(0)
    {}

    bool isValid() const {
        return _magic == MAGIC;
    }
};


template <int MaxBlocksPerChunk>
class ChunkDescriptor : private boost::noncopyable {
public :

    int _chunkId;   
    int _visitId;   
    int _nextChunk; 

    bool _usable;

    int _numBlocks; 
    int _nextBlock; 
    int _index;     
    int _size;      
    int _delta;     
    std::size_t _curBlockOffset; 

    Fifo<MAX_VISITS_IN_FLIGHT> _interestedParties;
    std::size_t _blocks[MaxBlocksPerChunk];


    ChunkDescriptor() { initialize(); }

    void initialize();

    void clear() {
        _nextBlock      = 0;
        _index          = 0;
        _size           = 0;
        _delta          = 0;
        _curBlockOffset = 0;
    }

    bool operator<(ChunkDescriptor const & cd) const {
        return _visitId < cd._visitId || (_visitId == cd._visitId && _chunkId < cd._chunkId);
    }

    // detail::HashedSet requirements
    int getId() const { return _chunkId; }
    int getNextInChain() const { return _nextChunk; }
    void setId(int const id) { _chunkId = id; }
    void setNextInChain(int const id) { _nextChunk = id; }
};


typedef unsigned char ChunkEntryFlag;


template <typename AllocatorT, typename DataT, typename TraitsT = DataTraits<DataT> >
class ChunkRef {

//    BOOST_STATIC_ASSERT(boost::has_trivial_assign<DataT>::value);
//    BOOST_STATIC_ASSERT(boost::has_trivial_copy<DataT>::value);
//    BOOST_STATIC_ASSERT(boost::has_trivial_destructor<DataT>::value);

public  :

    typedef DataT Entry;

    enum EntryFlag {
        IN_DELTA         = 0x01,
        UNCOMMITTED      = 0x02,
        INSERTED         = 0x04,
        DELETED          = 0x08
    };

    static int const ENTRIES_PER_BLOCK_LOG2 = TraitsT::ENTRIES_PER_BLOCK_LOG2;
    static int const MAX_BLOCKS = TraitsT::MAX_BLOCKS_PER_CHUNK;
    static std::size_t const BLOCK_SIZE =
        (sizeof(DataT) + sizeof(ChunkEntryFlag)) << ENTRIES_PER_BLOCK_LOG2;

    typedef ChunkDescriptor<MAX_BLOCKS> Descriptor;

    ChunkRef(Descriptor * desc, AllocatorT * all) : _descriptor(desc), _allocator(all) {}

    ~ChunkRef() {
        _descriptor = 0;
        _allocator  = 0;
    }

    boost::int64_t getId() const {
        return _descriptor->_chunkId;
    }
    int getVisitId() const {
        return _descriptor->_visitId;
    }
    bool isUsable() const {
        return _descriptor->_usable;
    }
    void setUsable() {
        _descriptor->_usable = true;
    }

    DataT const & get(int const i) const {
        assert(i >= 0 && i < _descriptor->_size);
        return *reinterpret_cast<DataT const *>(map(
            _descriptor->_blocks[i >> ENTRIES_PER_BLOCK_LOG2] +
            sizeof(ChunkEntryFlag)*(1 << ENTRIES_PER_BLOCK_LOG2) +
            (i & ((1 << ENTRIES_PER_BLOCK_LOG2) - 1))*sizeof(DataT)
        ));
    }

    DataT const * getBlock(int const b) const {
        assert(b >= 0 && b < _descriptor->_numBlocks);
        return reinterpret_cast<DataT const *>(map(
            _descriptor->_blocks[b] + (sizeof(ChunkEntryFlag) << ENTRIES_PER_BLOCK_LOG2)
        ));
    }

    DataT * getBlock(int const b) {
        assert(b >= 0 && b < _descriptor->_numBlocks);
        return reinterpret_cast<DataT *>(map(
            _descriptor->_blocks[b] + (sizeof(ChunkEntryFlag) << ENTRIES_PER_BLOCK_LOG2)
        ));
    }

    ChunkEntryFlag getFlag(int const i) const {
        assert(i >= 0 && i < _descriptor->_size);
        return *reinterpret_cast<ChunkEntryFlag const *>(map(
            _descriptor->_blocks[i >> ENTRIES_PER_BLOCK_LOG2] +
            (i & ((1u << ENTRIES_PER_BLOCK_LOG2) - 1))*sizeof(ChunkEntryFlag)
        ));
    }

    ChunkEntryFlag const * getFlagBlock(int const b) const {
        assert(b >= 0 && b < _descriptor->_numBlocks);
        return reinterpret_cast<ChunkEntryFlag const *>(map(_descriptor->_blocks[b]));
    }

    ChunkEntryFlag * getFlagBlock(int const b) {
        assert(b >= 0 && b < _descriptor->_numBlocks);
        return reinterpret_cast<ChunkEntryFlag *>(map(_descriptor->_blocks[b]));
    }

    void insert(DataT const & data) {
        insert(data, IN_DELTA | UNCOMMITTED | INSERTED);
    }

    void remove(int const i) {
        assert(i >= 0 && i < _descriptor->_size);
        ChunkEntryFlag * f = reinterpret_cast<ChunkEntryFlag *>(map(
            _descriptor->_blocks[i >> ENTRIES_PER_BLOCK_LOG2] +
            (i & ((1 << ENTRIES_PER_BLOCK_LOG2) - 1))*sizeof(ChunkEntryFlag)
        ));
        if ((*f & DELETED) == 0) { // removing an entry that is already deleted is a no-op
            *f |= DELETED | UNCOMMITTED;
        }
    }

    int size() const {
        return _descriptor->_size;
    }

    int delta() const {
        return _descriptor->_delta;
    }

    int capacity() const {
        return _descriptor->_numBlocks << ENTRIES_PER_BLOCK_LOG2;
    }

    int blocks() const {
        return _descriptor->_nextBlock;
    }

    int entries(int const b) const {
        assert(b >= 0 && b < _descriptor->_nextBlock);
        return (b < _descriptor->_nextBlock - 1) ? (1 << ENTRIES_PER_BLOCK_LOG2) : _descriptor->_index;
    }

    void clear() {
        _descriptor->clear();
    }

    void reserve(int const n);
    bool pack(int const i = 0);
    bool rollback();
    void commit(bool clearDelta = false);

    void read(std::string const & name, bool const compressed);
    void readDelta(std::string const & name, bool const compressed);

    void write(
        std::string const & name,
        bool        const   overwrite,
        bool        const   compressed,
        bool        const   withDelta = true
    ) const;
    void writeDelta(
        std::string const & name,
        bool        const   overwrite,
        bool        const   compressed
    ) const;

private :

    // Ensure number of entries per-block is a power of 2 greater than 512.
    // This avoids alignment issues for chunk entries and guarantees that
    // flags and chunk entries live on different cachelines.
    BOOST_STATIC_ASSERT(TraitsT::ENTRIES_PER_BLOCK_LOG2 >= 9);

    Descriptor * _descriptor;
    AllocatorT * _allocator;

    unsigned char * map(std::size_t const off) {
        return reinterpret_cast<unsigned char *>(_allocator) + off;
    }

    unsigned char const * map(std::size_t const off) const {
        return reinterpret_cast<unsigned char const *>(_allocator) + off;
    }

    void applyDeletes(
        int const * const deletes,
        int const numDeletes,
        int const end
    );

    void insert(DataT const & data, ChunkEntryFlag const flags);

    void setFlags(
        int const b,
        ChunkEntryFlag const flags,
        int const i = 0,
        int const n = (1 << ENTRIES_PER_BLOCK_LOG2)
    );
};


}} // end of namespace lsst::ap

#endif // LSST_AP_CHUNK_H