source: mainline/uspace/lib/urcu/rcu.c@ 156b6406

/*
 * Copyright (c) 2012 Adam Hraska
 * 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.
 * - The name of the author may not be used to endorse or promote products
 *   derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``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 AUTHOR 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.
 */

/** @addtogroup liburcu
 * @{
 */
/**
 * @file
 * 
 * User space RCU is based on URCU utilizing signals [1]. This 
 * implementation does not however signal each thread of the process 
 * to issue a memory barrier. Instead, we introduced a syscall that
 * issues memory barriers (via IPIs) on cpus that are running threads
 * of the current process. First, it does not require us to schedule
 * and run every thread of the process. Second, IPIs are less intrusive 
 * than switching contexts and entering user space.
 * 
 * This algorithm is further modified to require a single instead of
 * two reader group changes per grace period. Signal-URCU flips
 * the reader group and waits for readers of the previous group 
 * twice in succession in order to wait for new readers that were
 * delayed and mistakenly associated with the previous reader group. 
 * The modified algorithm ensures that the new reader group is
 * always empty (by explicitly waiting for it to become empty).
 * Only then does it flip the reader group and wait for preexisting
 * readers of the old reader group (invariant of SRCU [2, 3]).
 * 
 * 
 * [1] User-level implementations of read-copy update,
 *     2012, appendix
 *     http://www.rdrop.com/users/paulmck/RCU/urcu-supp-accepted.2011.08.30a.pdf
 * 
 * [2] linux/kernel/srcu.c in Linux 3.5-rc2,
 *     2012
 *     http://tomoyo.sourceforge.jp/cgi-bin/lxr/source/kernel/srcu.c?v=linux-3.5-rc2-ccs-1.8.3
 *
 * [3] [RFC PATCH 5/5 single-thread-version] implement 
 *     per-domain single-thread state machine,
 *     2012, Lai
 *     https://lkml.org/lkml/2012/3/6/586
 */

#include "rcu.h"
#include <fibril_synch.h>
#include <fibril.h>
#include <stdio.h>
#include <compiler/barrier.h>
#include <libarch/barrier.h>
#include <futex.h>
#include <macros.h>
#include <async.h>
#include <adt/list.h>
#include <smp_memory_barrier.h>
#include <assert.h>
#include <time.h>


/** RCU sleeps for RCU_SLEEP_MS before polling an active RCU reader again. */
#define RCU_SLEEP_MS        10

#define RCU_NESTING_SHIFT   1
#define RCU_NESTING_INC     (1 << RCU_NESTING_SHIFT)
#define RCU_GROUP_BIT_MASK  (size_t)(RCU_NESTING_INC - 1)
#define RCU_GROUP_A         (size_t)(0 | RCU_NESTING_INC)
#define RCU_GROUP_B         (size_t)(1 | RCU_NESTING_INC)


/** Fibril local RCU data. */
typedef struct fibril_rcu_data {
        size_t nesting_cnt;
        link_t link;
        bool registered;
} fibril_rcu_data_t;

/** Process global RCU data. */
typedef struct rcu_data {
        size_t cur_gp;
        size_t reader_group;
        futex_t list_futex;
        list_t fibrils_list;
        struct {
                futex_t futex;
                bool locked;
                list_t blocked_fibrils;
                size_t blocked_thread_cnt;
                futex_t futex_blocking_threads;
        } sync_lock;
} rcu_data_t;

typedef struct blocked_fibril {
        fid_t id;
        link_t link;
        bool is_ready;
} blocked_fibril_t;


/** Fibril local RCU data. */
static fibril_local fibril_rcu_data_t fibril_rcu = {
        .nesting_cnt = 0,
        .link = {
                .next = NULL,
                .prev = NULL
        },
        .registered = false
};

/** Process global RCU data. */
static rcu_data_t rcu = {
        .cur_gp = 0,
        .reader_group = RCU_GROUP_A,
        .list_futex = FUTEX_INITIALIZER,
        .fibrils_list = LIST_INITIALIZER(rcu.fibrils_list),
        .sync_lock = {
                .futex = FUTEX_INITIALIZER,
                .locked = false,
                .blocked_fibrils = LIST_INITIALIZER(rcu.sync_lock.blocked_fibrils),
                .blocked_thread_cnt = 0,
                .futex_blocking_threads = FUTEX_INITIALIZE(0),
        },
};


static void wait_for_readers(size_t reader_group, blocking_mode_t blocking_mode);
static void force_mb_in_all_threads(void);
static bool is_preexisting_reader(const fibril_rcu_data_t *fib, size_t group);

static void lock_sync(blocking_mode_t blocking_mode);
static void unlock_sync(void);
static void sync_sleep(blocking_mode_t blocking_mode);

static bool is_in_group(size_t nesting_cnt, size_t group);
static bool is_in_reader_section(size_t nesting_cnt);
static size_t get_other_group(size_t group);


/** Registers a fibril so it may start using RCU read sections.
 * 
 * A fibril must be registered with rcu before it can enter RCU critical
 * sections delineated by rcu_read_lock() and rcu_read_unlock().
 */
void rcu_register_fibril(void)
{
        assert(!fibril_rcu.registered);
        
        futex_down(&rcu.list_futex);
        list_append(&fibril_rcu.link, &rcu.fibrils_list);
        futex_up(&rcu.list_futex);
        
        fibril_rcu.registered = true;
}

/** Deregisters a fibril that had been using RCU read sections.
 * 
 * A fibril must be deregistered before it exits if it had
 * been registered with rcu via rcu_register_fibril().
 */
void rcu_deregister_fibril(void)
{
        assert(fibril_rcu.registered);
        
        /* 
         * Forcefully unlock any reader sections. The fibril is exiting
         * so it is not holding any references to data protected by the
         * rcu section. Therefore, it is safe to unlock. Otherwise, 
         * rcu_synchronize() would wait indefinitely.
         */
        memory_barrier();
        fibril_rcu.nesting_cnt = 0;
        
        futex_down(&rcu.list_futex);
        list_remove(&fibril_rcu.link);
        futex_up(&rcu.list_futex);

        fibril_rcu.registered = false;
}

/** Delimits the start of an RCU reader critical section. 
 * 
 * RCU reader sections may be nested.  
 */
void rcu_read_lock(void)
{
        assert(fibril_rcu.registered);
        
        size_t nesting_cnt = ACCESS_ONCE(fibril_rcu.nesting_cnt);
        
        if (0 == (nesting_cnt >> RCU_NESTING_SHIFT)) {
                ACCESS_ONCE(fibril_rcu.nesting_cnt) = ACCESS_ONCE(rcu.reader_group);
                /* Required by MB_FORCE_L */
                compiler_barrier(); /* CC_BAR_L */
        } else {
                ACCESS_ONCE(fibril_rcu.nesting_cnt) = nesting_cnt + RCU_NESTING_INC;
        }
}

/** Delimits the start of an RCU reader critical section. */
void rcu_read_unlock(void)
{
        assert(fibril_rcu.registered);
        /* assert(rcu_read_locked()); */
        
        /* Required by MB_FORCE_U */
        compiler_barrier(); /* CC_BAR_U */
        /* todo: ACCESS_ONCE(nesting_cnt) ? */
        fibril_rcu.nesting_cnt -= RCU_NESTING_INC;
}

/** Returns true if the current fibril is in an RCU reader section. */
bool rcu_read_locked(void)
{
        return 0 != (fibril_rcu.nesting_cnt >> RCU_NESTING_SHIFT);
}

/** Blocks until all preexisting readers exit their critical sections. */
void _rcu_synchronize(blocking_mode_t blocking_mode)
{
        assert(!rcu_read_locked());
        
        /* Contain load of rcu.cur_gp. */
        memory_barrier();

        /* Approximately the number of the GP in progress. */
        size_t gp_in_progress = ACCESS_ONCE(rcu.cur_gp);
        
        lock_sync(blocking_mode);
        
        /* 
         * Exit early if we were stuck waiting for the mutex for a full grace 
         * period. Started waiting during gp_in_progress (or gp_in_progress + 1
         * if the value propagated to this cpu too late) so wait for the next
         * full GP, gp_in_progress + 1, to finish. Ie don't wait if the GP
         * after that, gp_in_progress + 2, already started.
         */
        /* rcu.cur_gp >= gp_in_progress + 2, but tolerates overflows. */
        if (rcu.cur_gp != gp_in_progress && rcu.cur_gp + 1 != gp_in_progress) {
                unlock_sync();
                return;
        }
        
        ++ACCESS_ONCE(rcu.cur_gp);
        
        /* 
         * Pairs up with MB_FORCE_L (ie CC_BAR_L). Makes changes prior 
         * to rcu_synchronize() visible to new readers. 
         */
        memory_barrier(); /* MB_A */
        
        /* 
         * Pairs up with MB_A. 
         * 
         * If the memory barrier is issued before CC_BAR_L in the target
         * thread, it pairs up with MB_A and the thread sees all changes
         * prior to rcu_synchronize(). Ie any reader sections are new
         * rcu readers.  
         * 
         * If the memory barrier is issued after CC_BAR_L, it pairs up
         * with MB_B and it will make the most recent nesting_cnt visible
         * in this thread. Since the reader may have already accessed
         * memory protected by RCU (it ran instructions passed CC_BAR_L),
         * it is a preexisting reader. Seeing the most recent nesting_cnt 
         * ensures the thread will be identified as a preexisting reader
         * and we will wait for it in wait_for_readers(old_reader_group).
         */
        force_mb_in_all_threads(); /* MB_FORCE_L */
        
        /* 
         * Pairs with MB_FORCE_L (ie CC_BAR_L, CC_BAR_U) and makes the most
         * current fibril.nesting_cnt visible to this cpu.
         */
        read_barrier(); /* MB_B */
        
        size_t new_reader_group = get_other_group(rcu.reader_group);
        wait_for_readers(new_reader_group, blocking_mode);
        
        /* Separates waiting for readers in new_reader_group from group flip. */
        memory_barrier();
        
        /* Flip the group new readers should associate with. */
        size_t old_reader_group = rcu.reader_group;
        rcu.reader_group = new_reader_group;

        /* Flip the group before waiting for preexisting readers in the old group.*/
        memory_barrier();
        
        wait_for_readers(old_reader_group, blocking_mode);
        
        /* MB_FORCE_U  */
        force_mb_in_all_threads(); /* MB_FORCE_U */
        
        unlock_sync();
}

/** Issues a memory barrier in each thread of this process. */
static void force_mb_in_all_threads(void)
{
        /* 
         * Only issue barriers in running threads. The scheduler will 
         * execute additional memory barriers when switching to threads
         * of the process that are currently not running.
         */
        smp_memory_barrier();
}

/** Waits for readers of reader_group to exit their readers sections. */
static void wait_for_readers(size_t reader_group, blocking_mode_t blocking_mode)
{
        futex_down(&rcu.list_futex);
        
        list_t quiescent_fibrils;
        list_initialize(&quiescent_fibrils);
        
        while (!list_empty(&rcu.fibrils_list)) {
                list_foreach_safe(rcu.fibrils_list, fibril_it, next_fibril) {
                        fibril_rcu_data_t *fib = member_to_inst(fibril_it, 
                                fibril_rcu_data_t, link);
                        
                        if (is_preexisting_reader(fib, reader_group)) {
                                futex_up(&rcu.list_futex);
                                sync_sleep(blocking_mode);
                                futex_down(&rcu.list_futex);
                                /* Break to while loop. */
                                break;
                        } else {
                                list_remove(fibril_it);
                                list_append(fibril_it, &quiescent_fibrils);
                        }
                }
        }
        
        list_concat(&rcu.fibrils_list, &quiescent_fibrils);
        futex_up(&rcu.list_futex);
}

static void lock_sync(blocking_mode_t blocking_mode)
{
        futex_down(&rcu.sync_lock.futex);
        if (rcu.sync_lock.locked) {
                if (blocking_mode == BM_BLOCK_FIBRIL) {
                        blocked_fibril_t blocked_fib;
                        blocked_fib.id = fibril_get_id();
                                
                        list_append(&blocked_fib.link, &rcu.sync_lock.blocked_fibrils);
                        
                        do {
                                blocked_fib.is_ready = false;
                                futex_up(&rcu.sync_lock.futex);
                                fibril_switch(FIBRIL_TO_MANAGER);
                                futex_down(&rcu.sync_lock.futex);
                        } while (rcu.sync_lock.locked);
                        
                        list_remove(&blocked_fib.link);
                        rcu.sync_lock.locked = true;
                } else {
                        assert(blocking_mode == BM_BLOCK_THREAD);
                        rcu.sync_lock.blocked_thread_cnt++;
                        futex_up(&rcu.sync_lock.futex);
                        futex_down(&rcu.sync_lock.futex_blocking_threads);
                }
        } else {
                rcu.sync_lock.locked = true;
        }
}

static void unlock_sync(void)
{
        assert(rcu.sync_lock.locked);
        
        /* 
         * Blocked threads have a priority over fibrils when accessing sync().
         * Pass the lock onto a waiting thread.
         */
        if (0 < rcu.sync_lock.blocked_thread_cnt) {
                --rcu.sync_lock.blocked_thread_cnt;
                futex_up(&rcu.sync_lock.futex_blocking_threads);
        } else {
                /* Unlock but wake up any fibrils waiting for the lock. */
                
                if (!list_empty(&rcu.sync_lock.blocked_fibrils)) {
                        blocked_fibril_t *blocked_fib = member_to_inst(
                                list_first(&rcu.sync_lock.blocked_fibrils), blocked_fibril_t, link);
        
                        if (!blocked_fib->is_ready) {
                                blocked_fib->is_ready = true;
                                fibril_add_ready(blocked_fib->id);
                        }
                }
                
                rcu.sync_lock.locked = false;
                futex_up(&rcu.sync_lock.futex);
        }
}

static void sync_sleep(blocking_mode_t blocking_mode)
{
        assert(rcu.sync_lock.locked);
        /* 
         * Release the futex to avoid deadlocks in singlethreaded apps 
         * but keep sync locked. 
         */
        futex_up(&rcu.sync_lock.futex);

        if (blocking_mode == BM_BLOCK_FIBRIL) {
                async_usleep(RCU_SLEEP_MS * 1000);
        } else {
                usleep(RCU_SLEEP_MS * 1000);
        }
                
        futex_down(&rcu.sync_lock.futex);
}


static bool is_preexisting_reader(const fibril_rcu_data_t *fib, size_t group)
{
        size_t nesting_cnt = ACCESS_ONCE(fib->nesting_cnt);
        
        return is_in_group(nesting_cnt, group) && is_in_reader_section(nesting_cnt);
}

static size_t get_other_group(size_t group)
{
        if (group == RCU_GROUP_A) 
                return RCU_GROUP_B;
        else
                return RCU_GROUP_A;
}

static bool is_in_reader_section(size_t nesting_cnt)
{
        return RCU_NESTING_INC <= nesting_cnt;
}

static bool is_in_group(size_t nesting_cnt, size_t group)
{
        return (nesting_cnt & RCU_GROUP_BIT_MASK) == (group & RCU_GROUP_BIT_MASK);
}



/** @}
 */