log_carray.h

Go to the documentation of this file.

/*
 * (c) Copyright 2014, Hewlett-Packard Development Company, LP
 */
/*
     Shore-MT -- Multi-threaded port of the SHORE storage manager

                       Copyright (c) 2010-2014
      Data Intensive Applications and Systems Labaratory (DIAS)
               Ecole Polytechnique Federale de Lausanne

                         All Rights Reserved.

   Permission to use, copy, modify and distribute this software and
   its documentation is hereby granted, provided that both the
   copyright notice and this permission notice appear in all copies of
   the software, derivative works or modified versions, and any
   portions thereof, and that both notices appear in supporting
   documentation.

   This code 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. THE AUTHORS
   DISCLAIM ANY LIABILITY OF ANY KIND FOR ANY DAMAGES WHATSOEVER
   RESULTING FROM THE USE OF THIS SOFTWARE.
*/
#ifndef __LOG_CARRAY_H
#define __LOG_CARRAY_H

#include <cstdint>
#include <boost/static_assert.hpp>
#include "w_base.h"
#include "w_error.h"
#include "mcs_lock.h"
#include "lsn.h"

const bool CARRAY_RELEASE_DELEGATION = false;

typedef int64_t carray_status_t;

typedef uint32_t carray_slotid_t;

struct CArraySlot {
    mcs_lock::qnode me2;                // +16 -> 16

// Logging information. Also useful as padding for cacheline (64 byte).
    lsn_t lsn;                          // +8 -> 24
    int64_t old_end;                    // +8 -> 32
    int64_t start_pos;                  // +8 -> 40
    int64_t pos;                        // +8 -> 48
    int64_t new_end;                    // +8 -> 56
    int64_t new_base;                   // +8 -> 64

    carray_status_t count;              // +8 -> 72

    mcs_lock::qnode me;                 // +16 -> 88
    mcs_lock::qnode* pred2;             // +8 -> 96
    w_error_codes error;                // +sizeof(w_error_codes)

    char padding[32 - sizeof(w_error_codes)]; // +32-sizeof(w_error_codes) -> 128

    CArraySlot volatile* vthis() {
        return this;
    }

    const CArraySlot volatile* vthis() const {
        return this;
    }
};
// Doesn't compile in old Debian.
//BOOST_STATIC_ASSERT_MSG((sizeof(CArraySlot) % CACHELINE_SIZE) == 0,
//    "size of CArraySlot must be aligned to CACHELINE_SIZE for better performance");

class ConsolidationArray {
public:
    ConsolidationArray(int active_slot_count);

    ~ConsolidationArray();

    enum Constants {
        ALL_SLOT_COUNT = 256,

        DEFAULT_ACTIVE_SLOT_COUNT = 3,

        SLOT_AVAILABLE = 0,
        SLOT_UNUSED = -1,
        SLOT_PENDING = -2,
        SLOT_FINISHED = -4,
    };

    static carray_status_t join_carray_status(carray_status_t current_status, int32_t size) {
        w_assert1(size >= 0);
        w_assert1(current_status >= 0);
        const carray_status_t THREAD_INCREMENT = 1L << 32;
        return current_status + size + THREAD_INCREMENT;
    }

    static int32_t extract_carray_log_size(carray_status_t current_status) {
        w_assert1(current_status >= 0);
        // bug fixed: lower 32-bit is the size
        //const carray_status_t THREAD_MASK = 0xFFFF;
        const carray_status_t THREAD_MASK = 0xFFFFFFFF;
        return current_status & THREAD_MASK;
    }

    CArraySlot* join_slot(int32_t size, carray_status_t& status);

    void join_expose(CArraySlot* slot);

    CArraySlot* grab_delegated_expose(CArraySlot* slot);

    void wait_for_leader(CArraySlot* slot);

    bool wait_for_expose(CArraySlot* slot);

    void replace_active_slot(CArraySlot* slot);

private:
    int _indexof(const CArraySlot* slot) const;

    int32_t _slot_mark;

    const int32_t _active_slot_count;

    CArraySlot _all_slots[ALL_SLOT_COUNT];

    CArraySlot** _active_slots;

    // paddings to make sure mcs_lock are in different cacheline char _padding[CACHELINE_SIZE]; /** @endcond */
    mcs_lock _expose_lock;
};

inline int ConsolidationArray::_indexof(const CArraySlot* info) const {
    return info - _all_slots;
}

#endif // __LOG_CARRAY_H