Właściwość List <T> bezpieczna wątkowo

Chcę implementacji List<T>jako właściwości, której można bez wątpienia bezpiecznie używać.

Coś takiego:

private List<T> _list;

private List<T> MyT
{
    get { // return a copy of _list; }
    set { _list = value; }
}

Wygląda na to, że nadal muszę zwrócić kopię (sklonowaną) kolekcji, więc jeśli gdzieś iterujemy kolekcję i w tym samym czasie kolekcja jest ustawiona, nie jest zgłaszany żaden wyjątek.

Jak zaimplementować właściwość kolekcji bezpieczną dla wątków?

Jeśli celujesz w .Net 4, w przestrzeni nazw System.Collections.Concurrent jest kilka opcji

Możesz użyć ConcurrentBag<T>w tym przypadku zamiastList<T>

Mimo że uzyskał najwięcej głosów, zwykle nie można traktować System.Collections.Concurrent.ConcurrentBag<T>jako bezpiecznego zamiennika tego, System.Collections.Generic.List<T>co jest (zwrócił na to już Radek Stromský), którego nie zamówiono.

Ale istnieje klasa o nazwie, System.Collections.Generic.SynchronizedCollection<T>która istnieje już od czasu części .NET 3.0 frameworka, ale jest tak dobrze ukryta w miejscu, w którym nie można się tego spodziewać, że jest mało znana i prawdopodobnie nigdy się o nią nie natknąłeś (przynajmniej Nigdy nie zrobiłem).

SynchronizedCollection<T>jest kompilowany do zestawu System.ServiceModel.dll (który jest częścią profilu klienta, ale nie jest częścią przenośnej biblioteki klas).

Mam nadzieję, że to pomoże.

Myślę, że stworzenie przykładowej klasy ThreadSafeList byłoby łatwe:

public class ThreadSafeList<T> : IList<T>
{
    protected List<T> _interalList = new List<T>();

    // Other Elements of IList implementation

    public IEnumerator<T> GetEnumerator()
    {
        return Clone().GetEnumerator();
    }

    System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator()
    {
        return Clone().GetEnumerator();
    }

    protected static object _lock = new object();

    public List<T> Clone()
    {
        List<T> newList = new List<T>();

        lock (_lock)
        {
            _interalList.ForEach(x => newList.Add(x));
        }

        return newList;
    }
}

Po prostu klonujesz listę przed zażądaniem modułu wyliczającego, a zatem każde wyliczenie działa na kopii, której nie można modyfikować podczas działania.

Nawet akceptowana odpowiedź to ConcurrentBag, nie sądzę, że jest to realne zastąpienie listy we wszystkich przypadkach, ponieważ komentarz Radka do odpowiedzi mówi: „ConcurrentBag jest kolekcją nieuporządkowaną, więc w przeciwieństwie do List nie gwarantuje zamówienia. Nie można również uzyskać dostępu do pozycji według indeksu ”.

Więc jeśli używasz .NET 4.0 lub nowszego, obejściem może być użycie ConcurrentDictionary z liczbą całkowitą TKey jako indeksem tablicy i TValue jako wartością tablicy. Jest to zalecany sposób zastępowania listy w kursie kolekcji współbieżnych języka C # Pluralsight . ConcurrentDictionary rozwiązuje oba wspomniane powyżej problemy: dostęp do indeksu i porządkowanie (nie możemy polegać na porządkowaniu, ponieważ jest to tablica mieszająca pod maską, ale obecna implementacja .NET oszczędza kolejność dodawania elementów).

ArrayListKlasa C # ma Synchronizedmetodę.

var threadSafeArrayList = ArrayList.Synchronized(new ArrayList());

Zwraca to bezpieczne opakowanie dla każdego wystąpienia IList. Wszystkie operacje muszą być wykonywane przez owijkę, aby zapewnić bezpieczeństwo nici.

Jeśli spojrzysz na kod źródłowy listy T ( https://referencesource.microsoft.com/#mscorlib/system/collections/generic/list.cs,c66df6f36c131877 ), zauważysz, że jest tam klasa (co oczywiście jest wewnętrzny - dlaczego Microsoft, dlaczego?!?!) o nazwie SynchronizedList of T. Kopiuję wklejając kod tutaj:

   [Serializable()]
    internal class SynchronizedList : IList<T> {
        private List<T> _list;
        private Object _root;

        internal SynchronizedList(List<T> list) {
            _list = list;
            _root = ((System.Collections.ICollection)list).SyncRoot;
        }

        public int Count {
            get {
                lock (_root) { 
                    return _list.Count; 
                }
            }
        }

        public bool IsReadOnly {
            get {
                return ((ICollection<T>)_list).IsReadOnly;
            }
        }

        public void Add(T item) {
            lock (_root) { 
                _list.Add(item); 
            }
        }

        public void Clear() {
            lock (_root) { 
                _list.Clear(); 
            }
        }

        public bool Contains(T item) {
            lock (_root) { 
                return _list.Contains(item);
            }
        }

        public void CopyTo(T[] array, int arrayIndex) {
            lock (_root) { 
                _list.CopyTo(array, arrayIndex);
            }
        }

        public bool Remove(T item) {
            lock (_root) { 
                return _list.Remove(item);
            }
        }

        System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator() {
            lock (_root) { 
                return _list.GetEnumerator();
            }
        }

        IEnumerator<T> IEnumerable<T>.GetEnumerator() {
            lock (_root) { 
                return ((IEnumerable<T>)_list).GetEnumerator();
            }
        }

        public T this[int index] {
            get {
                lock(_root) {
                    return _list[index];
                }
            }
            set {
                lock(_root) {
                    _list[index] = value;
                }
            }
        }

        public int IndexOf(T item) {
            lock (_root) {
                return _list.IndexOf(item);
            }
        }

        public void Insert(int index, T item) {
            lock (_root) {
                _list.Insert(index, item);
            }
        }

        public void RemoveAt(int index) {
            lock (_root) {
                _list.RemoveAt(index);
            }
        }
    }

Osobiście uważam, że wiedzieli, że można stworzyć lepszą implementację przy użyciu SemaphoreSlim , ale nie do końca do tego doszli.

Możesz także użyć bardziej prymitywnego

Monitor.Enter(lock);
Monitor.Exit(lock);

która używa blokady (zobacz ten post C # Blokowanie obiektu, który jest ponownie przypisywany w bloku blokady ).

Jeśli spodziewasz się wyjątków w kodzie, nie jest to bezpieczne, ale pozwala zrobić coś takiego:

using System;
using System.Collections.Generic;
using System.Threading;
using System.Linq;

public class Something
{
    private readonly object _lock;
    private readonly List<string> _contents;

    public Something()
    {
        _lock = new object();

        _contents = new List<string>();
    }

    public Modifier StartModifying()
    {
        return new Modifier(this);
    }

    public class Modifier : IDisposable
    {
        private readonly Something _thing;

        public Modifier(Something thing)
        {
            _thing = thing;

            Monitor.Enter(Lock);
        }

        public void OneOfLotsOfDifferentOperations(string input)
        {
            DoSomethingWith(input);
        }

        private void DoSomethingWith(string input)
        {
            Contents.Add(input);
        }

        private List<string> Contents
        {
            get { return _thing._contents; }
        }

        private object Lock
        {
            get { return _thing._lock; }
        }

        public void Dispose()
        {
            Monitor.Exit(Lock);
        }
    }
}

public class Caller
{
    public void Use(Something thing)
    {
        using (var modifier = thing.StartModifying())
        {
            modifier.OneOfLotsOfDifferentOperations("A");
            modifier.OneOfLotsOfDifferentOperations("B");

            modifier.OneOfLotsOfDifferentOperations("A");
            modifier.OneOfLotsOfDifferentOperations("A");
            modifier.OneOfLotsOfDifferentOperations("A");
        }
    }
}

Jedną z fajnych rzeczy jest to, że otrzymasz blokadę na czas trwania serii operacji (zamiast blokowania w każdej operacji). Co oznacza, że ​​dane wyjściowe powinny wychodzić w odpowiednich fragmentach (moje użycie tego polegało na wyświetlaniu na ekranie niektórych danych wyjściowych z zewnętrznego procesu)

Bardzo podoba mi się prostota + przejrzystość ThreadSafeList +, która odgrywa ważną rolę w zatrzymywaniu awarii

W .NET Core (w dowolnej wersji) można użyć ImmutableList , który ma wszystkie funkcje List<T>.

Wierzę, _list.ToList()że zrobię ci kopię. Możesz również zapytać o to, jeśli potrzebujesz, na przykład:

_list.Select("query here").ToList(); 

W każdym razie msdn twierdzi, że to rzeczywiście kopia, a nie tylko odniesienie. Aha, i tak, będziesz musiał zablokować ustaloną metodę, jak wskazali inni.

Wygląda na to, że wiele osób, które to znalazły, chce mieć kolekcję o dynamicznych rozmiarach, bezpieczną dla wątków i indeksowaną. Najbliższa i najłatwiejsza rzecz, jaką znam, byłaby.

System.Collections.Concurrent.ConcurrentDictionary<int, YourDataType>

Wymagałoby to upewnienia się, że klucz jest odpowiednio obciążony, jeśli chcesz normalnego zachowania podczas indeksowania. Jeśli zachowasz ostrożność, .count może wystarczyć jako klucz dla każdej dodawanej nowej pary klucz-wartość.

Sugerowałbym każdemu, kto ma do czynienia ze List<T>scenariuszami wielowątkowymi, aby przyjrzał się niezmiennym kolekcjom, w szczególności ImmutableArray .

Uważam, że jest to bardzo przydatne, gdy masz:

1. Stosunkowo niewiele pozycji na liście
2. Niewiele operacji odczytu / zapisu
3. DUŻO jednoczesnego dostępu (tj. Wiele wątków, które uzyskują dostęp do listy w trybie czytania)

Może być również przydatny, gdy musisz zaimplementować zachowanie podobne do transakcji (tj. Cofnąć operację wstawiania / aktualizowania / usuwania w przypadku niepowodzenia)

Oto klasa, o którą prosiłeś:

namespace AI.Collections {
    using System;
    using System.Collections;
    using System.Collections.Generic;
    using System.Linq;
    using System.Runtime.Serialization;
    using System.Threading.Tasks;
    using System.Threading.Tasks.Dataflow;

    /// <summary>
    ///     Just a simple thread safe collection.
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <value>Version 1.5</value>
    /// <remarks>TODO replace locks with AsyncLocks</remarks>
    [DataContract( IsReference = true )]
    public class ThreadSafeList<T> : IList<T> {
        /// <summary>
        ///     TODO replace the locks with a ReaderWriterLockSlim
        /// </summary>
        [DataMember]
        private readonly List<T> _items = new List<T>();

        public ThreadSafeList( IEnumerable<T> items = null ) { this.Add( items ); }

        public long LongCount {
            get {
                lock ( this._items ) {
                    return this._items.LongCount();
                }
            }
        }

        public IEnumerator<T> GetEnumerator() { return this.Clone().GetEnumerator(); }

        IEnumerator IEnumerable.GetEnumerator() { return this.GetEnumerator(); }

        public void Add( T item ) {
            if ( Equals( default( T ), item ) ) {
                return;
            }
            lock ( this._items ) {
                this._items.Add( item );
            }
        }

        public Boolean TryAdd( T item ) {
            try {
                if ( Equals( default( T ), item ) ) {
                    return false;
                }
                lock ( this._items ) {
                    this._items.Add( item );
                    return true;
                }
            }
            catch ( NullReferenceException ) { }
            catch ( ObjectDisposedException ) { }
            catch ( ArgumentNullException ) { }
            catch ( ArgumentOutOfRangeException ) { }
            catch ( ArgumentException ) { }
            return false;
        }

        public void Clear() {
            lock ( this._items ) {
                this._items.Clear();
            }
        }

        public bool Contains( T item ) {
            lock ( this._items ) {
                return this._items.Contains( item );
            }
        }

        public void CopyTo( T[] array, int arrayIndex ) {
            lock ( this._items ) {
                this._items.CopyTo( array, arrayIndex );
            }
        }

        public bool Remove( T item ) {
            lock ( this._items ) {
                return this._items.Remove( item );
            }
        }

        public int Count {
            get {
                lock ( this._items ) {
                    return this._items.Count;
                }
            }
        }

        public bool IsReadOnly { get { return false; } }

        public int IndexOf( T item ) {
            lock ( this._items ) {
                return this._items.IndexOf( item );
            }
        }

        public void Insert( int index, T item ) {
            lock ( this._items ) {
                this._items.Insert( index, item );
            }
        }

        public void RemoveAt( int index ) {
            lock ( this._items ) {
                this._items.RemoveAt( index );
            }
        }

        public T this[ int index ] {
            get {
                lock ( this._items ) {
                    return this._items[ index ];
                }
            }
            set {
                lock ( this._items ) {
                    this._items[ index ] = value;
                }
            }
        }

        /// <summary>
        ///     Add in an enumerable of items.
        /// </summary>
        /// <param name="collection"></param>
        /// <param name="asParallel"></param>
        public void Add( IEnumerable<T> collection, Boolean asParallel = true ) {
            if ( collection == null ) {
                return;
            }
            lock ( this._items ) {
                this._items.AddRange( asParallel
                                              ? collection.AsParallel().Where( arg => !Equals( default( T ), arg ) )
                                              : collection.Where( arg => !Equals( default( T ), arg ) ) );
            }
        }

        public Task AddAsync( T item ) {
            return Task.Factory.StartNew( () => { this.TryAdd( item ); } );
        }

        /// <summary>
        ///     Add in an enumerable of items.
        /// </summary>
        /// <param name="collection"></param>
        public Task AddAsync( IEnumerable<T> collection ) {
            if ( collection == null ) {
                throw new ArgumentNullException( "collection" );
            }

            var produce = new TransformBlock<T, T>( item => item, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = Environment.ProcessorCount } );

            var consume = new ActionBlock<T>( action: async obj => await this.AddAsync( obj ), dataflowBlockOptions: new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = Environment.ProcessorCount } );
            produce.LinkTo( consume );

            return Task.Factory.StartNew( async () => {
                collection.AsParallel().ForAll( item => produce.SendAsync( item ) );
                produce.Complete();
                await consume.Completion;
            } );
        }

        /// <summary>
        ///     Returns a new copy of all items in the <see cref="List{T}" />.
        /// </summary>
        /// <returns></returns>
        public List<T> Clone( Boolean asParallel = true ) {
            lock ( this._items ) {
                return asParallel
                               ? new List<T>( this._items.AsParallel() )
                               : new List<T>( this._items );
            }
        }

        /// <summary>
        ///     Perform the <paramref name="action" /> on each item in the list.
        /// </summary>
        /// <param name="action">
        ///     <paramref name="action" /> to perform on each item.
        /// </param>
        /// <param name="performActionOnClones">
        ///     If true, the <paramref name="action" /> will be performed on a <see cref="Clone" /> of the items.
        /// </param>
        /// <param name="asParallel">
        ///     Use the <see cref="ParallelQuery{TSource}" /> method.
        /// </param>
        /// <param name="inParallel">
        ///     Use the
        ///     <see
        ///         cref="Parallel.ForEach{TSource}(System.Collections.Generic.IEnumerable{TSource},System.Action{TSource})" />
        ///     method.
        /// </param>
        public void ForEach( Action<T> action, Boolean performActionOnClones = true, Boolean asParallel = true, Boolean inParallel = false ) {
            if ( action == null ) {
                throw new ArgumentNullException( "action" );
            }
            var wrapper = new Action<T>( obj => {
                try {
                    action( obj );
                }
                catch ( ArgumentNullException ) {
                    //if a null gets into the list then swallow an ArgumentNullException so we can continue adding
                }
            } );
            if ( performActionOnClones ) {
                var clones = this.Clone( asParallel: asParallel );
                if ( asParallel ) {
                    clones.AsParallel().ForAll( wrapper );
                }
                else if ( inParallel ) {
                    Parallel.ForEach( clones, wrapper );
                }
                else {
                    clones.ForEach( wrapper );
                }
            }
            else {
                lock ( this._items ) {
                    if ( asParallel ) {
                        this._items.AsParallel().ForAll( wrapper );
                    }
                    else if ( inParallel ) {
                        Parallel.ForEach( this._items, wrapper );
                    }
                    else {
                        this._items.ForEach( wrapper );
                    }
                }
            }
        }

        /// <summary>
        ///     Perform the <paramref name="action" /> on each item in the list.
        /// </summary>
        /// <param name="action">
        ///     <paramref name="action" /> to perform on each item.
        /// </param>
        /// <param name="performActionOnClones">
        ///     If true, the <paramref name="action" /> will be performed on a <see cref="Clone" /> of the items.
        /// </param>
        /// <param name="asParallel">
        ///     Use the <see cref="ParallelQuery{TSource}" /> method.
        /// </param>
        /// <param name="inParallel">
        ///     Use the
        ///     <see
        ///         cref="Parallel.ForEach{TSource}(System.Collections.Generic.IEnumerable{TSource},System.Action{TSource})" />
        ///     method.
        /// </param>
        public void ForAll( Action<T> action, Boolean performActionOnClones = true, Boolean asParallel = true, Boolean inParallel = false ) {
            if ( action == null ) {
                throw new ArgumentNullException( "action" );
            }
            var wrapper = new Action<T>( obj => {
                try {
                    action( obj );
                }
                catch ( ArgumentNullException ) {
                    //if a null gets into the list then swallow an ArgumentNullException so we can continue adding
                }
            } );
            if ( performActionOnClones ) {
                var clones = this.Clone( asParallel: asParallel );
                if ( asParallel ) {
                    clones.AsParallel().ForAll( wrapper );
                }
                else if ( inParallel ) {
                    Parallel.ForEach( clones, wrapper );
                }
                else {
                    clones.ForEach( wrapper );
                }
            }
            else {
                lock ( this._items ) {
                    if ( asParallel ) {
                        this._items.AsParallel().ForAll( wrapper );
                    }
                    else if ( inParallel ) {
                        Parallel.ForEach( this._items, wrapper );
                    }
                    else {
                        this._items.ForEach( wrapper );
                    }
                }
            }
        }
    }
}

Zasadniczo, jeśli chcesz bezpiecznie wyliczać, musisz użyć blokady.

Więcej informacji można znaleźć w witrynie MSDN. http://msdn.microsoft.com/en-us/library/6sh2ey19.aspx

Oto część MSDN, która może Cię zainteresować:

Publiczne statyczne elementy członkowskie tego typu (udostępnione w języku Visual Basic) są bezpieczne wątkowo. Nie ma gwarancji, że żadne elementy członkowskie instancji będą bezpieczne dla wątków.

Lista może obsługiwać wielu czytelników jednocześnie, o ile kolekcja nie jest modyfikowana. Wyliczanie w kolekcji nie jest wewnętrznie procedurą bezpieczną dla wątków. W rzadkich przypadkach, gdy wyliczenie rywalizuje z co najmniej jednym dostępem do zapisu, jedynym sposobem zapewnienia bezpieczeństwa wątków jest zablokowanie kolekcji podczas całego wyliczania. Aby umożliwić dostęp do kolekcji przez wiele wątków w celu odczytu i zapisu, należy zaimplementować własną synchronizację.

Oto klasa listy bezpiecznych wątków bez blokady

 public class ConcurrentList   
    {
        private long _i = 1;
        private ConcurrentDictionary<long, T> dict = new ConcurrentDictionary<long, T>();  
        public int Count()
        {
            return dict.Count;
        }
         public List<T> ToList()
         {
            return dict.Values.ToList();
         }

        public T this[int i]
        {
            get
            {
                long ii = dict.Keys.ToArray()[i];
                return dict[ii];
            }
        }
        public void Remove(T item)
        {
            T ov;
            var dicItem = dict.Where(c => c.Value.Equals(item)).FirstOrDefault();
            if (dicItem.Key > 0)
            {
                dict.TryRemove(dicItem.Key, out ov);
            }
            this.CheckReset();
        }
        public void RemoveAt(int i)
        {
            long v = dict.Keys.ToArray()[i];
            T ov;
            dict.TryRemove(v, out ov);
            this.CheckReset();
        }
        public void Add(T item)
        {
            dict.TryAdd(_i, item);
            _i++;
        }
        public IEnumerable<T> Where(Func<T, bool> p)
        {
            return dict.Values.Where(p);
        }
        public T FirstOrDefault(Func<T, bool> p)
        {
            return dict.Values.Where(p).FirstOrDefault();
        }
        public bool Any(Func<T, bool> p)
        {
            return dict.Values.Where(p).Count() > 0 ? true : false;
        }
        public void Clear()
        {
            dict.Clear();
        }
        private void CheckReset()
        {
            if (dict.Count == 0)
            {
                this.Reset();
            }
        }
        private void Reset()
        {
            _i = 1;
        }
    }

Użyj do tego lockinstrukcji. ( Przeczytaj tutaj, aby uzyskać więcej informacji. )

private List<T> _list;

private List<T> MyT
{
    get { return _list; }
    set
    {
        //Lock so only one thread can change the value at any given time.
        lock (_list)
        {
            _list = value;
        }
    }
}

FYI, prawdopodobnie nie jest to dokładnie to, o co prosisz - prawdopodobnie chcesz zablokować dalej w swoim kodzie, ale nie mogę tego założyć. Przyjrzyj się locksłowu kluczowemu i dostosuj jego użycie do swojej konkretnej sytuacji.

Jeśli trzeba, można lockw obu geti setbloku przy użyciu _listzmiennej, która stałaby się więc odczytu / zapisu nie mogą występować jednocześnie.