Add safe blocks with Rust-like thread-safety guarantees

[ivanivanyuk1993]

Recently I've read some articles about Rust from my news feed and tried to rewrite some simple sharded concurrent queue and guava-like loading observable cache from C# to Rust

Rust compiler found a race in my guava-like loading observable cache, which I missed and tests didn't notice it too, since chance of race was low, and also I noticed that I didn't need to think as much about thread safety in Rust, so I want to have same ease of writing thread-safe code in C#, hence I propose safe blocks with Rust-like lifetimes and ownership or something similar to have compile-time thread-safety guarantees in C#, since Rust proved that it is possible to achieve

[svick]

Rust is a language designed around its ownership semantics, which means it's used for more than just thread-safety (namely, memory management) and that ownership is everywhere and people who use Rust are okay with that.

How exactly do you suggest to add such a system to C#? Consider that:

1. It has to interoperate well with all the existing code, which offers no guarantees.
2. It has to justify its existence, since it's only used for thread safety, and so it shouldn't be too complicated.

[ivanivanyuk1993]

@svick I think it will be ok if I add examples of working code I have in C#, it's equivalent in Rust, and a code I would like to see in C# with safe feature

Passing tests now code in C#

using System;
using System.Collections.Generic;
using System.Threading;
using CheapSpinLockerNS;

namespace ShardedQueueNS
{
    /// <summary>
    ///     <para>
    ///         ShardedQueue is needed for <see cref="EventLoopGroupSchedulerNS.EventLoopGroupScheduler"/> and
    ///         <see cref="RunAtomicProviderNS.RunAtomicProvider"/> as a highly specialized for their use case
    ///         alternative to slow performing <see cref="System.Collections.Concurrent.ConcurrentQueue{T}" />
    ///     </para>
    ///     <para>
    ///         ShardedQueue is a specialized concurrent queue, which uses spin locking and fights lock contention with
    ///         sharding
    ///     </para>
    ///     <para>
    ///         Note that while it may seem that FIFO order is guaranteed, it is not, because Array-based Queue was chosen
    ///         for shards instead of LinkedList-based
    ///         => There is a possible situation, when multiple Producers triggered long resize of very large shards,
    ///         all but last, then passed enough time for resize to finish, then 1 Producer triggers long resize of
    ///         last shard, and all other threads start to consume or produce, and eventually start spinning on
    ///         last shard, without guarantee which will acquire spin lock first, so we can't even guarantee that there
    ///         will always be item to consume in shard
    ///     </para>
    ///     <para>
    ///         Notice that this queue doesn't track length, since it's increment/decrement position may change
    ///         depending on use case, as well as logic when it goes from 1 to 0 or reverse, or even negative
    ///         (to optimize some hot paths, like in <see cref="EventLoopGroupSchedulerNS.EventLoopGroupScheduler"/>,
    ///         since it is cheaper to restore count to correct state than to enforce that it can not go negative)
    ///     </para>
    /// </summary>
    public class ShardedQueue<T>
    {
        private readonly int[] _isLockedList;
        private readonly int _moduloNumber;
        private readonly Queue<T>[] _queueList;

        private int _headIndex;
        private int _tailIndex;

        public ShardedQueue(int maxConcurrentThreadCount)
        {
            // queueCount should be greater than maxConcurrentThreadCount and a power of 2 (to make modulo more performant)
            var queueCount = (int) Math.Pow(2, Math.Ceiling(Math.Log2(maxConcurrentThreadCount)));
            // Computing modulo number, knowing that x % 2^n == x & (2^n - 1)
            _moduloNumber = queueCount - 1;

            _isLockedList = new int[queueCount];
            _queueList = new Queue<T>[queueCount];
            for (var i = 0; i < queueCount; i++) _queueList[i] = new Queue<T>();
        }

        /// <remarks>
        ///     Note that it will spin until item is added => it can deadlock if DequeueOrSpin is called without guarantee that
        ///     Enqueue will be called
        /// </remarks>
        /// <returns></returns>
        public T DequeueOrSpin()
        {
            T removedItem;
            var queueIndex = Interlocked.Increment(ref _headIndex) & _moduloNumber;

            ref var isLocked = ref _isLockedList[queueIndex];
            var queue = _queueList[queueIndex];
            // Note that since we already incremented _headIndex, it is important to complete operation, if we tried
            // to implement TryRemoveFirst, we would need to somehow balance our _headIndex, complexifying logic and
            // introducing overhead, and even if we need it, we can just check length before calling this method
            while (true)
                if (CheapSpinLocker.TryLock(ref isLocked))
                {
                    // If we acquired lock after Enqueue, we can break spinning, otherwise we should Unlock and restart
                    // spinning, giving Enqueue chance to lock
                    if (queue.TryDequeue(out removedItem))
                    {
                        CheapSpinLocker.Unlock(ref isLocked);
                        break;
                    }

                    CheapSpinLocker.Unlock(ref isLocked);
                }

            return removedItem;
        }

        public void Enqueue(T item)
        {
            var queueIndex = Interlocked.Increment(ref _tailIndex) & _moduloNumber;

            ref var isLocked = ref _isLockedList[queueIndex];
            CheapSpinLocker.AcquireLock(ref isLocked);
            _queueList[queueIndex].Enqueue(item);
            CheapSpinLocker.Unlock(ref isLocked);
        }
    }
}

Passing tests now code in Rust, equivalent to previous sample

use std::collections::VecDeque;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Mutex;

pub struct ShardedQueue<T> {
    /// todo make readonly
    modulo_number: usize,
    /// todo make readonly
    mutexed_queue_shard_list: Vec<Mutex<VecDeque<T>>>,

    head_index: AtomicUsize,
    tail_index: AtomicUsize,
}

impl<T> ShardedQueue<T> {
    pub fn new(max_concurrent_thread_count: usize) -> Self {
        // queue_count should be greater than max_concurrent_thread_count and a power of 2 (to make modulo more performant)
        let queue_count = (2 as usize)
            .pow(
                (max_concurrent_thread_count as f64)
                    .log2()
                    .ceil() as u32
            );
        let mut mutexed_queue_shard_list = Vec::<Mutex<VecDeque<T>>>::with_capacity(queue_count);
        for _i in 0..queue_count {
            mutexed_queue_shard_list.push(
                Mutex::new(
                    VecDeque::<T>::new()
                )
            );
        }
        Self {
            // Computing modulo number, knowing that x % 2^n == x & (2^n - 1)
            modulo_number: queue_count - 1,
            mutexed_queue_shard_list,
            head_index: Default::default(),
            tail_index: Default::default(),
        }
    }

    pub fn pop_front_or_spin(&self) -> T {
        let queue_index = self.head_index.fetch_add(1, Ordering::Relaxed) & self.modulo_number;
        let mutexed_queue_shard = unsafe { self.mutexed_queue_shard_list.get_unchecked(queue_index) };
        loop {
            let mut try_lock_result = mutexed_queue_shard.try_lock();
            if let Ok(ref mut queue_shard) = try_lock_result {
                let item_option = queue_shard.pop_front();
                match item_option {
                    Some(item) => {
                        return item;
                    }
                    _ => {}
                }
            }
        }
    }

    pub fn push_back(&self, item: T) {
        let queue_index = self.tail_index.fetch_add(1, Ordering::Relaxed) & self.modulo_number;
        let mutexed_queue_shard = unsafe { self.mutexed_queue_shard_list.get_unchecked(queue_index) };
        loop {
            let mut try_lock_result = mutexed_queue_shard.try_lock();
            if let Ok(ref mut queue_shard) = try_lock_result {
                queue_shard.push_back(item);
                break;
            }
        }
    }
}

Usage

Let's assume that we wrote same as Rust implementation of safe SafeShardedQueue<T> in C#, using only classes with safe modifier
When we want to use safe classes from regular code, like var shardedQueue = new SafeShardedQueue<T>();, compiler errors and asks us to get one of

• MutationPreventingReference with it's scope's lifetime(which can not be passed outside of scope), var shardedQueue = new MutationPreventingReference(new SafeShardedQueue<T>());
• MutationPreventingGCReference, whose lifetime will end when GC-d reference is collected var shardedQueue = new MutationPreventingGCReference(new SafeShardedQueue<T>());
• MutationAllowingReference with it's scope's lifetime(which can not be passed outside of scope), var shardedQueue = new MutationAllowingReference(new SafeShardedQueue<T>());
• MutationAllowingMutexedGCReference, whose lifetime will end when GC-d reference is collected, and whose value can not be borrowed outside of some TryLock block var shardedQueue = new MutationAllowingMutexedGCReference(new SafeShardedQueue<T>());

I think those references and constraints should be enough to analyze lifetimes and borrows

To use unsafe classes from safe classes, we can use same strategy Rust uses - unsafe blocks, but it should add strong reference to dotnet's GC graph and mark it unreachable when lifetime is over

I think proposed by me solutions would allow to achieve compile-time checked thread-safe code without sacrificing dotnet's mature libraries and legacy codebases