Concurrent Array (Ordered) Class with Indexing, and without Blocking<\/p>\n\n\n\n
Updated: Aug-2,2021<\/p>\n\n\n\n
RandomBigInt rng = new RandomBigInt(64);\n ulong[] sla1 = Enumerable.Range(0, 1000000).AsParallel().WithDegreeOfParallelism(10).Select(i => (ulong)rng.Next()).ToArray();\n CcArray<ulong> ccl = new CcArray<ulong>(1000000);\n\nExample 1: \n Parallel.ForEach(sla1, new ParallelOptions { MaxDegreeOfParallelism = 10 }, (v, p, i) =>\n {\n ccl.Add(v, (int)i);\n });\n\nExample 2:\n Parallel.For(0,sla1.Length,i=>\n {\n ccl.Add(sla1[i], (int)i);\n });\n\nExample 3:\n sla1.AsEnumerable().Select( (v,i)=> new {value=v,index=i}).AsParallel().WithDegreeOfParallelism(10).ForAll(x=>\n {\n ccl[x.index] = x.value;\n });\n\n Var asa = new ulong[1000000];\n for (var i = 0; i < 1000000; ++i)\n asa[i] = ccl[i];\n var exc = sla1.Except(asa).ToArray();<\/pre>\n\n\n\nusing System;\nusing System.Collections;\nusing System.Collections.Generic;\nusing System.Diagnostics;\nusing System.Threading;\n[DebuggerDisplay(\"Count = {\" + nameof(Count) + \"}\")]\npublic class CcArray<T> : IEnumerable<T>\n{\n private readonly int _size;\n private volatile int _activeThreadPosition;\n private volatile int[] _activeThreads;\n private volatile IntSet18<T>[] _array;\n public volatile int NumberOfActiveThreads;\n public CcArray(int size)\n {\n ThreadPool.GetMaxThreads(out var nW, out var nI);\n _array = new IntSet18<T>[nW];\n _size = size;\n NumberOfActiveThreads = 0;\n _activeThreadPosition = 0;\n _activeThreads = new int[Environment.ProcessorCount];\n _activeThreads.Fill(-1);\n }\n public T this[int index]\n {\n get\n {\n var (idx, trd) = FindIndex(index);\n return idx != -1 ? _array[trd].Slots[idx].Value : default;\n }\n set\n {\n var (idx, trd) = FindIndex(index);\n if (idx != -1)\n _array[trd].Slots[idx].Value = value;\n else Add(value, index);\n }\n }\n public int Count\n {\n get\n {\n if (_activeThreads == null)\n throw new Exception(\"Initialization has not completed. Note: The constructor cannot be void.\");\n var totalCount = 0;\n for (var i = 0; i < _activeThreads.Length; ++i)\n if (_activeThreads[i] != -1)\n if (_array[_activeThreads[i]].Allocated)\n totalCount += _array[_activeThreads[i]].Count;\n return totalCount;\n }\n }\n public IEnumerator<T> GetEnumerator()\n {\n return GetEnum();\n }\n IEnumerator IEnumerable.GetEnumerator()\n {\n return GetEnum();\n }\n public void Add(T item, int index, bool updateValue = false)\n {\n if (_activeThreads == null)\n throw new Exception(\"Initialization has not completed. Note: The constructor cannot be void.\");\n var id = ProcessThread();\n var (idx, trd) = FindIndex(index);\n if (idx == -1)\n _array[id].Add(index, item);\n else if (updateValue)\n _array[trd].Slots[idx].Value = item;\n }\n public (int idx, int trd) FindIndex(int index)\n {\n if (_activeThreads == null)\n throw new Exception(\"Initialization has not completed. Note: The constructor cannot be void.\");\n for (var i = 0; i < _activeThreads.Length; ++i)\n if (_activeThreads[i] != -1)\n if (_array[_activeThreads[i]].Allocated)\n {\n var idx = _array[_activeThreads[i]].FindKey(index);\n if (idx != -1)\n return (idx, _activeThreads[i]);\n }\n return (-1, -1);\n }\n private int ProcessThread()\n {\n var id = Thread.CurrentThread.ManagedThreadId;\n if (!_array[id].Allocated)\n {\n Interlocked.Increment(ref NumberOfActiveThreads);\n _array[id] = new IntSet18<T>(_size \/ NumberOfActiveThreads);\n if (_activeThreadPosition >= _activeThreads.Length)\n {\n var newActiveThreads = new int[_activeThreads.Length << 1];\n newActiveThreads.Fill(-1);\n for (var i = 0; i < _activeThreads.Length; ++i)\n if (_activeThreads[i] != -1)\n newActiveThreads[i] = _activeThreads[i];\n _activeThreads = newActiveThreads;\n }\n _activeThreads[_activeThreadPosition] = id;\n Interlocked.Increment(ref _activeThreadPosition);\n }\n return id;\n }\n public bool Contains(T item)\n {\n if (_activeThreads == null)\n throw new Exception(\"Initialization has not completed. Note: The constructor cannot be void.\");\n var eComparer = EqualityComparer<T>.Default;\n for (var i = 0; i < _activeThreads.Length; ++i)\n if (_activeThreads[i] != -1)\n if (_array[_activeThreads[i]].Allocated)\n for (var j = 0; j < _array[_activeThreads[i]].Count; ++j)\n if (eComparer.Equals(_array[_activeThreads[i]].Slots[j].Value, item))\n return true;\n return false;\n }\n public T[] ToArray()\n {\n if (_activeThreads == null)\n throw new Exception(\"Initialization has not completed. Note: The constructor cannot be void.\");\n var outputArray = new T[Count];\n var ptr = 0;\n for (var i = 0; i < _activeThreads.Length; ++i)\n if (_activeThreads[i] != -1)\n if (_array[_activeThreads[i]].Allocated)\n foreach (var v in _array[_activeThreads[i]])\n outputArray[ptr++] = v.Value;\n return outputArray;\n }\n private IEnumerator<T> GetEnum()\n {\n var array = ToArray();\n foreach (var i in array)\n yield return i;\n }\n [DebuggerDisplay(\"Count = {\" + nameof(Count) + \"}\")]\n internal struct IntSet18<T1>\n {\n private int[] _buckets;\n internal Slot[] Slots;\n public bool Allocated;\n public IntSet18(int size)\n {\n _buckets = new int[size];\n Slots = new Slot[size];\n Count = 0;\n Allocated = true;\n }\n public int Count\n {\n get;\n private set;\n }\n public IEnumerator<KeyValuePair<int, T1>> GetEnumerator()\n {\n for (var i = 0; i < Count; i++)\n if (Slots[i].Key >= 0)\n yield return new KeyValuePair<int, T1>(Slots[i].Key, Slots[i].Value);\n }\n public bool Add(int key, T1 value)\n {\n if (FindKey(key) != -1)\n return true;\n if (Count >= Slots.Length)\n Resize();\n var pos = key % _buckets.Length;\n Slots[Count].Next = _buckets[pos] - 1;\n Slots[Count].Key = key;\n Slots[Count].Value = value;\n _buckets[pos] = Count + 1;\n ++Count;\n return false;\n }\n private void Resize()\n {\n var newSize = _buckets.Length + _buckets.Length \/ 4 * 3;\n var newSlots = new Slot[newSize];\n var newBuckets = new int[newSize];\n var newCount = 0;\n var en = GetEnumerator();\n while (en.MoveNext())\n {\n var key = en.Current.Key;\n var value = en.Current.Value;\n var pos = key % newBuckets.Length;\n newSlots[newCount].Next = newBuckets[pos] - 1;\n newSlots[newCount].Key = key;\n newSlots[newCount].Value = value;\n newBuckets[pos] = newCount + 1;\n ++newCount;\n }\n Slots = newSlots;\n _buckets = newBuckets;\n Count = newCount;\n }\n public int FindKey(int key)\n {\n for (var position = _buckets[key % _buckets.Length] - 1; position >= 0; position = Slots[position].Next)\n if (Equals(Slots[position].Key, key))\n return position;\n return -1;\n }\n internal struct Slot\n {\n public int Next;\n public int Key;\n public T1 Value;\n }\n }\n}<\/pre>\n","protected":false},"excerpt":{"rendered":"Concurrent Array (Ordered) Class with Indexing, and without Blocking Updated: Aug-2,2021<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"open","sticky":false,"template":"","format":"standard","meta":[],"categories":[2],"tags":[35,84,207,209],"_links":{"self":[{"href":"https:\/\/michaeljohnsteiner.com\/index.php\/wp-json\/wp\/v2\/posts\/511"}],"collection":[{"href":"https:\/\/michaeljohnsteiner.com\/index.php\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/michaeljohnsteiner.com\/index.php\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/michaeljohnsteiner.com\/index.php\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/michaeljohnsteiner.com\/index.php\/wp-json\/wp\/v2\/comments?post=511"}],"version-history":[{"count":7,"href":"https:\/\/michaeljohnsteiner.com\/index.php\/wp-json\/wp\/v2\/posts\/511\/revisions"}],"predecessor-version":[{"id":520,"href":"https:\/\/michaeljohnsteiner.com\/index.php\/wp-json\/wp\/v2\/posts\/511\/revisions\/520"}],"wp:attachment":[{"href":"https:\/\/michaeljohnsteiner.com\/index.php\/wp-json\/wp\/v2\/media?parent=511"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/michaeljohnsteiner.com\/index.php\/wp-json\/wp\/v2\/categories?post=511"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/michaeljohnsteiner.com\/index.php\/wp-json\/wp\/v2\/tags?post=511"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}