tom/hermes-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Changed various locking mechanism.

Browse Source
This commit is contained in:
Tom
2025-03-29 20:18:09 +00:00
parent c80f1f2aa0
commit c7904f120d
10 changed files with 282 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
Services/ChannelManager.cs
View File

@ -14,7 +14,7 @@ namespace HermesSocketServer.Services
private readonly ServerConfiguration _configuration;
private readonly Serilog.ILogger _logger;
private readonly IDictionary<string, Channel> _channels;
private readonly object _lock;
private readonly Mutex _mutex;
public ChannelManager(IStore<string, User> users, Database database, IStore<string, TTSVoice> voices, ServerConfiguration configuration, Serilog.ILogger logger)
{
@ -24,7 +24,7 @@ namespace HermesSocketServer.Services
_configuration = configuration;
_logger = logger;
_channels = new ConcurrentDictionary<string, Channel>();
_lock = new object();
_mutex = new Mutex();
}
@ -36,15 +36,16 @@ namespace HermesSocketServer.Services
return await Task.Run(() =>
{
lock (_lock)
try
{
_mutex.WaitOne();
if (_channels.TryGetValue(userId, out var channel))
return Task.FromResult<Channel?>(channel);
return channel;
var actionTable = _configuration.Database.Tables["Action"];
var chatterTable = _configuration.Database.Tables["Chatter"];
var connectionTable = _configuration.Database.Tables["Connection"];
//var chatterGroupTable = _configuration.Database.Tables["ChatterGroup"];
var connectionStateTable = _configuration.Database.Tables["ConnectionState"];
var groupTable = _configuration.Database.Tables["Group"];
var groupPermissionTable = _configuration.Database.Tables["GroupPermission"];
var policyTable = _configuration.Database.Tables["Policy"];
@ -93,22 +94,43 @@ namespace HermesSocketServer.Services
]);
_channels.Add(userId, channel);
return Task.FromResult<Channel?>(channel);
return channel;
}
finally
{
_mutex.ReleaseMutex();
}
});
}
public Channel? Get(string channelId)
{
try
{
_mutex.WaitOne();
if (_channels.TryGetValue(channelId, out var channel))
return channel;
return null;
}
finally
{
_mutex.ReleaseMutex();
}
}
public async Task Save(string userId)
{
if (!_channels.TryGetValue(userId, out var channel))
Channel? channel;
try
{
_mutex.WaitOne();
if (!_channels.TryGetValue(userId, out channel))
return;
}
finally
{
_mutex.ReleaseMutex();
}
_logger.Debug($"Saving channel data to database [channel id: {channel.Id}][channel name: {channel.User.Name}]");
await Task.WhenAll([

14
Socket/Handlers/ChatterHandler.cs
View File

@ -14,7 +14,7 @@ namespace HermesSocketServer.Socket.Handlers
private readonly long[] _array;
private readonly ILogger _logger;
private readonly object _lock;
private readonly Mutex _lock;
private int _index;
public ChatterHandler(Database database, ILogger logger)
@ -24,7 +24,7 @@ namespace HermesSocketServer.Socket.Handlers
_chatters = new HashSet<long>(CHATTER_BUFFER_SIZE);
_array = new long[CHATTER_BUFFER_SIZE];
_index = -1;
_lock = new object();
_lock = new Mutex();
}
public async Task Execute<T>(WebSocketUser sender, T message, HermesSocketManager sockets)
@ -32,8 +32,9 @@ namespace HermesSocketServer.Socket.Handlers
if (message is not ChatterMessage data || sender.Id == null)
return;
lock (_lock)
try
{
_lock.WaitOne();
if (_chatters.Contains(data.Id))
return;
@ -43,11 +44,16 @@ namespace HermesSocketServer.Socket.Handlers
_index = -1;
var previous = _array[++_index];
if (previous != 0) {
if (previous != 0)
{
_chatters.Remove(previous);
}
_array[_index] = data.Id;
}
finally
{
_lock.ReleaseMutex();
}
try
{

11
Socket/Handlers/EmoteDetailsHandler.cs
View File

@ -14,7 +14,7 @@ namespace HermesSocketServer.Socket.Handlers
private readonly HashSet<string> _emotes;
private readonly string[] _array;
private readonly ILogger _logger;
private readonly object _lock;
private readonly Mutex _mutex;
private int _index;
@ -24,7 +24,7 @@ namespace HermesSocketServer.Socket.Handlers
_emotes = new HashSet<string>(EMOTE_BUFFER_SIZE);
_array = new string[EMOTE_BUFFER_SIZE];
_logger = logger;
_lock = new object();
_mutex = new Mutex();
_index = -1;
}
@ -36,8 +36,9 @@ namespace HermesSocketServer.Socket.Handlers
if (data.Emotes == null || !data.Emotes.Any())
return;
lock (_lock)
try
{
_mutex.WaitOne();
foreach (var entry in data.Emotes)
{
if (_emotes.Contains(entry.Key))
@ -59,6 +60,10 @@ namespace HermesSocketServer.Socket.Handlers
_array[_index] = entry.Key;
}
}
finally
{
_mutex.ReleaseMutex();
}
if (!data.Emotes.Any())
return;

14
Socket/Handlers/EmoteUsageHandler.cs
View File

@ -13,7 +13,7 @@ namespace HermesSocketServer.Socket.Handlers
private readonly HashSet<string> _history;
private readonly EmoteUsageMessage[] _array;
private readonly ILogger _logger;
private readonly object _lock;
private readonly Mutex _mutex;
private int _index;
@ -21,9 +21,9 @@ namespace HermesSocketServer.Socket.Handlers
{
_database = database;
_logger = logger;
_history = new HashSet<string>(101);
_array = new EmoteUsageMessage[100];
_lock = new object();
_history = new HashSet<string>(1001);
_array = new EmoteUsageMessage[1000];
_mutex = new Mutex();
_index = -1;
}
@ -33,7 +33,7 @@ namespace HermesSocketServer.Socket.Handlers
if (message is not EmoteUsageMessage data || sender.Id == null)
return;
lock (_lock)
try
{
if (_history.Contains(data.MessageId))
{
@ -50,6 +50,10 @@ namespace HermesSocketServer.Socket.Handlers
_array[_index] = data;
}
finally
{
_mutex.ReleaseMutex();
}
int rows = 0;
string sql = "INSERT INTO \"EmoteUsageHistory\" (timestamp, \"broadcasterId\", \"emoteId\", \"chatterId\") VALUES (@time, @broadcaster, @emote, @chatter)";

15
Socket/Handlers/HermesLoginHandler.cs
View File

@ -17,7 +17,7 @@ namespace HermesSocketServer.Socket.Handlers
private readonly Database _database;
private readonly HermesSocketManager _sockets;
private readonly ILogger _logger;
private readonly object _lock;
private readonly SemaphoreSlim _semaphore;
public HermesLoginHandler(ChannelManager manager, IStore<string, TTSVoice> voices, ServerConfiguration configuration, Database database, HermesSocketManager sockets, ILogger logger)
{
@ -27,7 +27,7 @@ namespace HermesSocketServer.Socket.Handlers
_database = database;
_sockets = sockets;
_logger = logger;
_lock = new object();
_semaphore = new SemaphoreSlim(1);
}
@ -46,15 +46,15 @@ namespace HermesSocketServer.Socket.Handlers
return;
IEnumerable<WebSocketUser?> recipients = Enumerable.Empty<WebSocketUser?>();
lock (_lock)
try
{
await _semaphore.WaitAsync();
if (sender.Id != null)
return;
sender.Id = userId;
recipients = _sockets.GetSockets(userId).ToList().Where(s => s.SessionId != sender.SessionId);
sender.Slave = data.WebLogin || recipients.Where(r => r != null && !r.WebLogin).Any();
}
recipients = _sockets.GetSockets(userId).ToList().Where(s => s.SessionId != sender.SessionId);
sender.ApiKey = data.ApiKey;
sender.WebLogin = data.WebLogin;
@ -131,5 +131,10 @@ namespace HermesSocketServer.Socket.Handlers
}
await Task.WhenAll(tasks);
}
finally
{
_semaphore.Release();
}
}
}
}

7
Store/ChatterGroupStore.cs
View File

@ -30,7 +30,8 @@ namespace HermesSocketServer.Store
await _database.Execute(sql, data, (reader) =>
{
var chatterId = reader.GetInt32(0).ToString();
lock (_lock)
_rwls.EnterWriteLock();
try
{
_store.Add(chatterId, new GroupChatter()
{
@ -40,6 +41,10 @@ namespace HermesSocketServer.Store
ChatterLabel = reader.GetString(1),
});
}
finally
{
_rwls.ExitWriteLock();
}
});
_logger.Information($"Loaded {_store.Count} group chatters from database [group id: {_groupId}]");
}

1
Store/IStore.cs
View File

@ -2,6 +2,7 @@ namespace HermesSocketServer.Store
{
public interface IStore<K, V>
{
bool Exists(K key);
V? Get(K key);
IDictionary<K, V> Get();
Task Load();

15
Store/Internal/AutoSavedStore.cs
View File

@ -46,14 +46,27 @@ namespace HermesSocketServer.Store.Internal
private async Task GenerateQuery(IList<K> keys, Func<int, string> generate, Func<string, IEnumerable<K>, IEnumerable<V>, Task<int>> execute)
{
ImmutableList<K>? list = null;
lock (_lock)
_rwls.EnterUpgradeableReadLock();
try
{
if (!keys.Any())
return;
_rwls.EnterWriteLock();
try
{
list = keys.ToImmutableList();
keys.Clear();
}
finally
{
_rwls.ExitWriteLock();
}
}
finally
{
_rwls.ExitUpgradeableReadLock();
}
var query = generate(list.Count);
var values = list.Select(id => _store[id]).Where(v => v != null);

32
Store/Internal/ComplexAutoSavedStore.cs
View File

@ -51,14 +51,27 @@ namespace HermesSocketServer.Store.Internal
private async Task GenerateQuery(IList<K> keys, Func<int, string> generate, Func<string, IEnumerable<K>, IEnumerable<V?>, Task<int>> execute)
{
ImmutableList<K>? list = null;
lock (_lock)
_rwls.EnterUpgradeableReadLock();
try
{
if (!keys.Any())
return;
_rwls.EnterWriteLock();
try
{
list = keys.ToImmutableList();
keys.Clear();
}
finally
{
_rwls.ExitWriteLock();
}
}
finally
{
_rwls.ExitUpgradeableReadLock();
}
var query = generate(list.Count);
var values = list.Select(id => _store[id]).Where(v => v != null);
@ -77,17 +90,28 @@ namespace HermesSocketServer.Store.Internal
private async Task GenerateDeleteQuery(IList<K> keys, IList<V> values, Func<int, string> generate, Func<string, IEnumerable<V>, Task<int>> execute)
{
ImmutableList<V>? list = null;
lock (_lock)
_rwls.EnterUpgradeableReadLock();
try
{
if (!keys.Any() || !values.Any())
{
return;
}
_rwls.EnterWriteLock();
try
{
list = values.ToImmutableList();
values.Clear();
keys.Clear();
}
finally
{
_rwls.ExitWriteLock();
}
}
finally
{
_rwls.ExitUpgradeableReadLock();
}
var query = generate(list.Count);
int rowsAffected = await execute(query, list);

82
Store/Internal/GroupedSaveStore.cs
View File

@ -10,7 +10,7 @@ namespace HermesSocketServer.Store.Internal
protected readonly IList<K> _added;
protected readonly IList<K> _modified;
protected readonly IList<K> _deleted;
protected readonly object _lock;
protected readonly ReaderWriterLockSlim _rwls;
public GroupSaveStore()
@ -19,7 +19,7 @@ namespace HermesSocketServer.Store.Internal
_added = new List<K>();
_modified = new List<K>();
_deleted = new List<K>();
_lock = new object();
_rwls = new ReaderWriterLockSlim();
}
public abstract Task Load();
@ -28,22 +28,46 @@ namespace HermesSocketServer.Store.Internal
protected abstract void OnPostRemove(K key, V value);
public abstract Task Save();
public bool Exists(K key)
{
_rwls.EnterReadLock();
try
{
return _store.ContainsKey(key);
}
finally
{
_rwls.ExitReadLock();
}
}
public V? Get(K key)
{
lock (_lock)
_rwls.EnterReadLock();
try
{
if (_store.TryGetValue(key, out var value))
return value;
}
finally
{
_rwls.ExitReadLock();
}
return null;
}
public IDictionary<K, V> Get()
{
lock (_lock)
_rwls.EnterReadLock();
try
{
return _store.ToImmutableDictionary();
}
finally
{
_rwls.ExitReadLock();
}
}
public bool Modify(K? key, V value)
@ -51,9 +75,13 @@ namespace HermesSocketServer.Store.Internal
if (key == null)
return false;
lock (_lock)
_rwls.EnterUpgradeableReadLock();
try
{
if (_store.TryGetValue(key, out V? oldValue))
{
_rwls.EnterWriteLock();
try
{
OnInitialModify(key, oldValue, value);
_store[key] = value;
@ -63,6 +91,15 @@ namespace HermesSocketServer.Store.Internal
}
return true;
}
finally
{
_rwls.ExitWriteLock();
}
}
}
finally
{
_rwls.ExitUpgradeableReadLock();
}
return false;
}
@ -72,9 +109,13 @@ namespace HermesSocketServer.Store.Internal
if (key == null)
return false;
lock (_lock)
_rwls.EnterUpgradeableReadLock();
try
{
if (_store.TryGetValue(key, out V? value))
{
_rwls.EnterWriteLock();
try
{
modify(value);
if (!_added.Contains(key) && !_modified.Contains(key))
@ -83,6 +124,15 @@ namespace HermesSocketServer.Store.Internal
}
return true;
}
finally
{
_rwls.ExitWriteLock();
}
}
}
finally
{
_rwls.ExitUpgradeableReadLock();
}
return false;
}
@ -92,12 +142,12 @@ namespace HermesSocketServer.Store.Internal
if (key == null)
return false;
lock (_lock)
_rwls.EnterWriteLock();
try
{
if (_store.TryGetValue(key, out var value))
{
if (_store.Remove(key))
{
_store.Remove(key);
OnPostRemove(key, value);
if (!_added.Remove(key))
{
@ -110,6 +160,9 @@ namespace HermesSocketServer.Store.Internal
return true;
}
}
finally
{
_rwls.ExitWriteLock();
}
return false;
}
@ -119,11 +172,10 @@ namespace HermesSocketServer.Store.Internal
if (key == null)
return false;
lock (_lock)
_rwls.EnterWriteLock();
try
{
if (_store.TryGetValue(key, out V? fetched))
{
if (fetched != value)
{
OnInitialModify(key, fetched, value);
_store[key] = value;
@ -133,7 +185,6 @@ namespace HermesSocketServer.Store.Internal
}
return true;
}
}
else
{
OnInitialAdd(key, value);
@ -145,7 +196,10 @@ namespace HermesSocketServer.Store.Internal
return true;
}
}
return false;
finally
{
_rwls.ExitWriteLock();
}
}
}
}