ogb912/opensim
1
0
Fork 0
mirror of https://github.com/opensim/opensim.git synced 2026-08-02 23:18:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

smartthreadpool: replace its locked dictionary

Browse Source
This commit is contained in:
UbitUmarov
2021-09-03 14:36:57 +01:00
parent 30c479db94
commit 94d1cf1205
6 changed files with 59 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
ThirdParty/SmartThreadPool/SmartThreadPool.cs vendored
View File

@@ -99,6 +99,7 @@ using System;
using System.Security;
using System.Threading;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Runtime.CompilerServices;
@@ -202,7 +203,8 @@ namespace Amib.Threading
/// <summary>
/// Dictionary of all the threads in the thread pool.
/// </summary>
private readonly SynchronizedDictionary<Thread, ThreadEntry> _workerThreads = new SynchronizedDictionary<Thread, ThreadEntry>();
private readonly ConcurrentDictionary<int, ThreadEntry> _workerThreads = new ConcurrentDictionary<int, ThreadEntry>();
private readonly object _workerThreadsLock = new object();
/// <summary>
/// Queue of work items.
@@ -271,7 +273,7 @@ namespace Amib.Threading
/// work item int the SmartThreadPool
/// This variable is used in case of Shutdown
/// </summary>
private readonly SynchronizedDictionary<IWorkItemsGroup, IWorkItemsGroup> _workItemsGroups = new SynchronizedDictionary<IWorkItemsGroup, IWorkItemsGroup>();
private readonly ConcurrentDictionary<int, IWorkItemsGroup> _workItemsGroups = new ConcurrentDictionary<int, IWorkItemsGroup>();
/// <summary>
/// A common object for all the work items int the STP
@@ -434,7 +436,7 @@ namespace Amib.Threading
private void StartOptimalNumberOfThreads()
{
int threadsCount;
lock (_workerThreads.SyncRoot)
lock (_workerThreadsLock)
{
threadsCount = _workItemsQueue.Count;
if (threadsCount == _stpStartInfo.MinWorkerThreads)
@@ -553,17 +555,20 @@ namespace Amib.Threading
}
private int baseWorkIDs = Environment.TickCount;
internal void RegisterWorkItemsGroup(IWorkItemsGroup workItemsGroup)
{
_workItemsGroups[workItemsGroup] = workItemsGroup;
int localID = Interlocked.Increment(ref baseWorkIDs);
while (_workItemsGroups.ContainsKey(localID))
localID = Interlocked.Increment(ref baseWorkIDs);
workItemsGroup.localID = localID;
_workItemsGroups[localID] = workItemsGroup;
}
internal void UnregisterWorkItemsGroup(IWorkItemsGroup workItemsGroup)
{
if (_workItemsGroups.Contains(workItemsGroup))
{
_workItemsGroups.Remove(workItemsGroup);
}
_workItemsGroups.TryRemove(workItemsGroup.localID, out IWorkItemsGroup dummy);
}
/// <summary>
@@ -575,9 +580,9 @@ namespace Amib.Threading
// There is no need to lock the two methods together
// since only the current thread removes itself
// and the _workerThreads is a synchronized dictionary
if (_workerThreads.Contains(Thread.CurrentThread))
if (_workerThreads.TryRemove(Thread.CurrentThread.ManagedThreadId, out ThreadEntry te))
{
_workerThreads.Remove(Thread.CurrentThread);
te.Clean();
_windowsPCs.SampleThreads(_workerThreads.Count, _inUseWorkerThreads);
_localPCs.SampleThreads(_workerThreads.Count, _inUseWorkerThreads);
}
@@ -592,7 +597,7 @@ namespace Amib.Threading
if (_isSuspended)
return;
lock (_workerThreads.SyncRoot)
lock (_workerThreadsLock)
{
// Don't start threads on shut down
if (_shutdown)
@@ -648,7 +653,7 @@ namespace Amib.Threading
--threadsCount;
// Add it to the dictionary and update its creation time.
_workerThreads[workerThread] = new ThreadEntry(this);
_workerThreads[workerThread.ManagedThreadId] = new ThreadEntry(this, workerThread);
_windowsPCs.SampleThreads(_workerThreads.Count, _inUseWorkerThreads);
_localPCs.SampleThreads(_workerThreads.Count, _inUseWorkerThreads);
@@ -663,7 +668,7 @@ namespace Amib.Threading
{
// Keep the entry of the dictionary as thread's variable to avoid the synchronization locks
// of the dictionary.
CurrentThreadEntry = _workerThreads[Thread.CurrentThread];
CurrentThreadEntry = _workerThreads[Thread.CurrentThread.ManagedThreadId];
bool informedCompleted = false;
FireOnThreadInitialization();
@@ -682,7 +687,7 @@ namespace Amib.Threading
// Double lock for quit.
if (_workerThreads.Count > maxworkers)
{
lock (_workerThreads.SyncRoot)
lock (_workerThreadsLock)
{
if (_workerThreads.Count > maxworkers)
{
@@ -708,7 +713,7 @@ namespace Amib.Threading
// Double lock for quit.
if (_workerThreads.Count > minworkers)
{
lock (_workerThreads.SyncRoot)
lock (_workerThreadsLock)
{
if (_workerThreads.Count > minworkers)
{
@@ -914,8 +919,8 @@ namespace Amib.Threading
pcs.Dispose();
}
Thread[] threads;
lock (_workerThreads.SyncRoot)
ThreadEntry[] threadEntries;
lock (_workerThreadsLock)
{
// Shutdown the work items queue
_workItemsQueue.Dispose();
@@ -925,8 +930,9 @@ namespace Amib.Threading
_shuttingDownEvent.Set();
// Make a copy of the threads' references in the pool
threads = new Thread[_workerThreads.Count];
_workerThreads.Keys.CopyTo(threads, 0);
threadEntries = new ThreadEntry[_workerThreads.Count];
_workerThreads.Values.CopyTo(threadEntries, 0);
_workerThreads.Clear();
}
int millisecondsLeft = millisecondsTimeout;
@@ -936,8 +942,10 @@ namespace Amib.Threading
bool timeout = false;
// Each iteration we update the time left for the timeout.
foreach (Thread thread in threads)
foreach (ThreadEntry te in threadEntries)
{
Thread thread = te.WorkThread;
// Join don't work with negative numbers
if (!waitInfinitely && (millisecondsLeft < 0))
{
@@ -959,19 +967,21 @@ namespace Amib.Threading
//TimeSpan ts = DateTime.UtcNow - start;
millisecondsLeft = millisecondsTimeout - (int)stopwatch.ElapsedMilliseconds;
}
te.WorkThread = null;
}
if (timeout && forceAbort)
{
// Abort the threads in the pool
foreach (Thread thread in threads)
foreach (ThreadEntry te in threadEntries)
{
Thread thread = te.WorkThread;
if ((thread != null) && thread.IsAlive )
{
try
{
thread.Abort(); // Shutdown
te.WorkThread = null;
}
catch (SecurityException e)
{
@@ -1184,9 +1194,8 @@ namespace Amib.Threading
{
tih();
}
catch (Exception e)
catch
{
e.GetHashCode();
Debug.Assert(false);
throw;
}
@@ -1204,9 +1213,8 @@ namespace Amib.Threading
{
tth();
}
catch (Exception e)
catch
{
e.GetHashCode();
Debug.Assert(false);
throw;
}
@@ -1242,9 +1250,7 @@ namespace Amib.Threading
foreach (ThreadEntry threadEntry in _workerThreads.Values)
{
WorkItem workItem = threadEntry.CurrentWorkItem;
if (null != workItem &&
workItem.WasQueuedBy(wig) &&
!workItem.IsCanceled)
if (null != workItem && !workItem.IsCanceled && workItem.WasQueuedBy(wig))
{
threadEntry.CurrentWorkItem.GetWorkItemResult().Cancel(true);
}
@@ -1478,8 +1484,7 @@ namespace Amib.Threading
}
_isSuspended = false;
ICollection workItemsGroups = _workItemsGroups.Values;
foreach (WorkItemsGroup workItemsGroup in workItemsGroups)
foreach (WorkItemsGroup workItemsGroup in _workItemsGroups.Values)
{
workItemsGroup.OnSTPIsStarting();
}
@@ -1496,8 +1501,7 @@ namespace Amib.Threading
_canceledSmartThreadPool.IsCanceled = true;
_canceledSmartThreadPool = new CanceledWorkItemsGroup();
ICollection workItemsGroups = _workItemsGroups.Values;
foreach (WorkItemsGroup workItemsGroup in workItemsGroups)
foreach (WorkItemsGroup workItemsGroup in _workItemsGroups.Values)
{
workItemsGroup.Cancel(abortExecution);
}
@@ -1506,12 +1510,13 @@ namespace Amib.Threading
{
foreach (ThreadEntry threadEntry in _workerThreads.Values)
{
WorkItem workItem = threadEntry.CurrentWorkItem;
if (null != workItem &&
threadEntry.AssociatedSmartThreadPool == this &&
!workItem.IsCanceled)
if(threadEntry.AssociatedSmartThreadPool == this)
{
threadEntry.CurrentWorkItem.GetWorkItemResult().Cancel(true);
WorkItem workItem = threadEntry.CurrentWorkItem;
if (null != workItem && !workItem.IsCanceled)
{
threadEntry.CurrentWorkItem.GetWorkItemResult().Cancel(true);
}
}
}
}