delaunay_sync.h

Go to the documentation of this file.

/*
 *  Copyright (c) 2000-2022 Inria
 *  All rights reserved.
 *
 *  Redistribution and use in source and binary forms, with or without
 *  modification, are permitted provided that the following conditions are met:
 *
 *  * Redistributions of source code must retain the above copyright notice,
 *  this list of conditions and the following disclaimer.
 *  * Redistributions in binary form must reproduce the above copyright notice,
 *  this list of conditions and the following disclaimer in the documentation
 *  and/or other materials provided with the distribution.
 *  * Neither the name of the ALICE Project-Team nor the names of its
 *  contributors may be used to endorse or promote products derived from this
 *  software without specific prior written permission.
 *
 *  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 *  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 *  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 *  ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
 *  LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 *  CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 *  SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 *  INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 *  CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 *  ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 *  POSSIBILITY OF SUCH DAMAGE.
 *
 *  Contact: Bruno Levy
 *
 *     https://www.inria.fr/fr/bruno-levy
 *
 *     Inria,
 *     Domaine de Voluceau,
 *     78150 Le Chesnay - Rocquencourt
 *     FRANCE
 *
 */
 
#ifndef DELAUNAY_SYNC_H
#define DELAUNAY_SYNC_H
 
#include <geogram/basic/common.h>
#include <geogram/basic/assert.h>
#include <atomic>
 
// In GARGANTUA mode (64-bit indices), we also enable
// more than 127 concurrent threads (16-bits cell status,
// that contain owner thread id).
#ifdef GARGANTUA
#define GEO_CONNECTION_MACHINE
#endif
 
namespace GEO {
 
    class CellStatusArray {
    public:
#ifdef GEO_CONNECTION_MACHINE
        // For machines that can run more than 127 concurrent threads
        typedef uint16_t thread_index_t;
        typedef uint16_t cell_status_t;
        static constexpr cell_status_t FREE_CELL     = 32767;
        static constexpr cell_status_t THREAD_MASK   = 32767;
        static constexpr cell_status_t CONFLICT_MASK = 32768;
#else
        typedef uint8_t thread_index_t;
        typedef uint8_t cell_status_t;
        static constexpr cell_status_t FREE_CELL     = 127;
        static constexpr cell_status_t THREAD_MASK   = 127;
        static constexpr cell_status_t CONFLICT_MASK = 128;
#endif
 
        static constexpr index_t MAX_THREADS = index_t(THREAD_MASK)-1;
 
        CellStatusArray() : cell_status_(nullptr), size_(0), capacity_(0) {
        }
 
        CellStatusArray(index_t size_in) :
            cell_status_(nullptr), size_(0), capacity_(0) {
            resize(size_in,size_in);
        }
 
        ~CellStatusArray() {
            clear();
        }
 
        CellStatusArray(const CellStatusArray& rhs) = delete;
 
        CellStatusArray& operator=(const CellStatusArray& rhs) = delete;
 
        cell_status_t acquire_cell(index_t cell, cell_status_t status) {
            geo_debug_assert(cell < size_);
            cell_status_t expected = FREE_CELL;
            // strong: acquire_cell is not used in a spinlock-like
            // spinning loop (so we do not want to have "false negatives")
            cell_status_[cell].compare_exchange_strong(
                expected,status,
                std::memory_order_acquire,std::memory_order_acquire
            ); // this one could probably be relaxed ----^
            // if compare_exchange was not sucessful, expected contains
            // the current stored value.
            return (expected & THREAD_MASK);
        }
 
        void release_cell(index_t cell) {
            geo_debug_assert(cell < size_);
            cell_status_[cell].store(FREE_CELL, std::memory_order_release);
        }
 
        cell_status_t cell_thread(index_t cell) const {
            geo_debug_assert(cell < size_);
            return (
                cell_status_[cell].load(std::memory_order_relaxed) &
                THREAD_MASK
            );
        }
 
        bool cell_is_marked_as_conflict(index_t cell) const {
            geo_debug_assert(cell < size_);
            return(
                (
                    cell_status_[cell].load(std::memory_order_relaxed) &
                    CONFLICT_MASK
                ) != 0
            );
        }
 
        void mark_cell_as_conflict(index_t cell) {
            // memory_order_relaxed because this function is always called from
            // a thread that previously acquired the cell.
            cell_status_[cell].fetch_or(
                CONFLICT_MASK, std::memory_order_relaxed
            );
        }
 
        void set_cell_status(index_t cell, cell_status_t status) {
            geo_debug_assert(cell < size_);
            cell_status_[cell].store(status, std::memory_order_relaxed);
        }
 
        void resize(index_t size_in, index_t capacity_in) {
            geo_debug_assert(capacity_in >= size_in);
            geo_debug_assert(!Process::is_running_threads());
            if(capacity_in > capacity_) {
                capacity_ = capacity_in;
                std::atomic<cell_status_t>* old_cell_status = cell_status_;
                cell_status_ = new std::atomic<cell_status_t>[capacity_];
                for(index_t i=0; i<capacity_; ++i) {
                    cell_status_t val = (i < size_) ?
                        old_cell_status[i].load(std::memory_order_relaxed) :
                        FREE_CELL;
                    std::atomic_init(&cell_status_[i],val);
                }
                delete[] old_cell_status;
            }
            size_ = size_in;
#ifdef __cpp_lib_atomic_is_always_lock_free
            static_assert(std::atomic<cell_status_t>::is_always_lock_free);
#else
            geo_debug_assert(size_ == 0 || cell_status_[0].is_lock_free());
#endif
        }
 
        void resize(index_t size_in) {
            resize(size_in, size_in);
        }
 
        void reserve(index_t new_capacity) {
            if(new_capacity > capacity_) {
                resize(size_, new_capacity);
            }
        }
 
        void grow() {
            if(size_+1 >= capacity_) {
                resize(size_+1, std::max(capacity_*2,size_+1));
            } else {
                resize(size_+1, capacity_);
            }
        }
 
        index_t size() const {
            return size_;
        }
 
        void clear() {
            geo_debug_assert(!Process::is_running_threads());
#ifdef GEO_DEBUG
            for(index_t i=0; i<size_; ++i) {
                geo_debug_assert(cell_thread(i) == FREE_CELL);
            }
#endif
            delete[] cell_status_;
            cell_status_ = nullptr;
            size_ = 0;
            capacity_ = 0;
        }
 
    private:
        std::atomic<cell_status_t>* cell_status_;
        index_t size_;
        index_t capacity_;
    };
}
 
#endif