github/Discord.Net
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Switched locks over to Nito.AsyncEx locks

Browse Source
This commit is contained in:
RogueException
2016-01-02 01:27:22 -04:00
parent 24c97a93f7
commit 6e1dbb0e0e
20 changed files with 177 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
src/Discord.Net.Audio/AudioClient.cs
View File

@@ -2,6 +2,7 @@
using Discord.Logging;
using Discord.Net.WebSockets;
using Newtonsoft.Json;
using Nito.AsyncEx;
using System;
using System.Threading;
using System.Threading.Tasks;
@@ -10,7 +11,7 @@ namespace Discord.Audio
{
internal class AudioClient : IAudioClient
{
private readonly Semaphore _connectionLock;
private readonly AsyncLock _connectionLock;
private readonly JsonSerializer _serializer;
private CancellationTokenSource _cancelTokenSource;
@@ -31,7 +32,7 @@ namespace Discord.Audio
GatewaySocket = gatewaySocket;
Logger = logger;
_connectionLock = new Semaphore(1, 1);
_connectionLock = new AsyncLock();
_serializer = new JsonSerializer();
_serializer.DateTimeZoneHandling = DateTimeZoneHandling.Utc;
@@ -92,8 +93,7 @@ namespace Discord.Audio
if (VoiceSocket.Server == null)
throw new InvalidOperationException("This client has been closed.");
_connectionLock.WaitOne();
try
using (await _connectionLock.LockAsync())
{
_cancelTokenSource = new CancellationTokenSource();
var cancelToken = _cancelTokenSource.Token;
@@ -106,26 +106,17 @@ namespace Discord.Audio
VoiceSocket.WaitForConnection(cancelToken);
});
}
finally
{
_connectionLock.Release();
}
}
public async Task Disconnect()
{
_connectionLock.WaitOne();
try
using (await _connectionLock.LockAsync())
{
Service.RemoveClient(VoiceSocket.Server, this);
VoiceSocket.Channel = null;
SendVoiceUpdate();
await VoiceSocket.Disconnect();
}
finally
{
_connectionLock.Release();
}
}
private async void OnReceivedDispatch(object sender, WebSocketEventEventArgs e)

2
src/Discord.Net.Audio/AudioServiceConfig.cs
View File

@@ -29,7 +29,7 @@ namespace Discord.Audio
public bool EnableMultiserver { get { return _enableMultiserver; } set { SetValue(ref _enableMultiserver, value); } }
private bool _enableMultiserver = false;
/// <summary> Gets or sets the max buffer length (in milliseconds) for outgoing voice packets. This value is the target maximum but is not guaranteed, the buffer will often go slightly above this value. </summary>
/// <summary> Gets or sets the max buffer length (in milliseconds) for outgoing voice packets. </summary>
public int BufferLength { get { return _bufferLength; } set { SetValue(ref _bufferLength, value); } }
private int _bufferLength = 1000;

48
src/Discord.Net.Audio/Net/WebSockets/VoiceWebSocket.cs
View File

@@ -29,7 +29,7 @@ namespace Discord.Net.WebSockets
private readonly ConcurrentDictionary<uint, OpusDecoder> _decoders;
private readonly AudioClient _audioClient;
private readonly AudioServiceConfig _config;
private Thread _sendThread, _receiveThread;
private Task _sendTask, _receiveTask;
private VoiceBuffer _sendBuffer;
private OpusEncoder _encoder;
private uint _ssrc;
@@ -95,20 +95,9 @@ namespace Discord.Net.WebSockets
_udp = new UdpClient(new IPEndPoint(IPAddress.Any, 0));
List<Task> tasks = new List<Task>();
if ((_config.Mode & AudioMode.Outgoing) != 0)
{
_sendThread = new Thread(new ThreadStart(() => SendVoiceAsync(CancelToken)));
_sendThread.IsBackground = true;
_sendThread.Start();
}
/*if ((_config.Mode & AudioMode.Incoming) != 0)
{*/
_receiveThread = new Thread(new ThreadStart(() => ReceiveVoiceAsync(CancelToken)));
_receiveThread.IsBackground = true;
_receiveThread.Start();
/*}
else
tasks.Add(Task.Run(() => ReceiveVoiceAsync(CancelToken)));*/
if (_config.Mode.HasFlag(AudioMode.Outgoing))
_sendTask = Task.Run(() => SendVoiceAsync(CancelToken));
_receiveTask = Task.Run(() => ReceiveVoiceAsync(CancelToken));
SendIdentify();
@@ -119,14 +108,15 @@ namespace Discord.Net.WebSockets
tasks.Add(HeartbeatAsync(CancelToken));
await _taskManager.Start(tasks, _cancelTokenSource).ConfigureAwait(false);
}
protected override Task Cleanup()
protected override async Task Cleanup()
{
if (_sendThread != null)
_sendThread.Join();
if (_receiveThread != null)
_receiveThread.Join();
_sendThread = null;
_receiveThread = null;
var sendThread = _sendTask;
if (sendThread != null) await sendThread;
_sendTask = null;
var receiveThread = _receiveTask;
if (receiveThread != null) await receiveThread;
_receiveTask = null;
OpusDecoder decoder;
foreach (var pair in _decoders)
@@ -138,10 +128,10 @@ namespace Discord.Net.WebSockets
ClearPCMFrames();
_udp = null;
return base.Cleanup();
await base.Cleanup();
}
private void ReceiveVoiceAsync(CancellationToken cancelToken)
private async Task ReceiveVoiceAsync(CancellationToken cancelToken)
{
var closeTask = cancelToken.Wait();
try
@@ -158,7 +148,7 @@ namespace Discord.Net.WebSockets
while (!cancelToken.IsCancellationRequested)
{
Thread.Sleep(1);
await Task.Delay(1);
if (_udp.Available > 0)
{
#if !DOTNET5_4
@@ -243,12 +233,12 @@ namespace Discord.Net.WebSockets
catch (InvalidOperationException) { } //Includes ObjectDisposedException
}
private void SendVoiceAsync(CancellationToken cancelToken)
private async Task SendVoiceAsync(CancellationToken cancelToken)
{
try
{
while (!cancelToken.IsCancellationRequested && State != ConnectionState.Connected)
Thread.Sleep(1);
await Task.Delay(1);
if (cancelToken.IsCancellationRequested)
return;
@@ -370,10 +360,10 @@ namespace Discord.Net.WebSockets
{
int time = (int)Math.Floor(ticksToNextFrame / ticksPerMillisecond);
if (time > 0)
Thread.Sleep(time);
await Task.Delay(time);
}
else
Thread.Sleep(1); //Give as much time to the encrypter as possible
await Task.Delay(1); //Give as much time to the encrypter as possible
}
}
}

19
src/Discord.Net.Audio/SimpleAudioClient.cs
View File

@@ -1,4 +1,5 @@
using Discord.Logging;
using Nito.AsyncEx;
using System.Threading;
using System.Threading.Tasks;
@@ -27,21 +28,20 @@ namespace Discord.Audio
void IAudioClient.Wait() => _client.Wait();
}
private readonly Semaphore _connectionLock;
private readonly AsyncLock _connectionLock;
internal VirtualClient CurrentClient { get; private set; }
public SimpleAudioClient(AudioService service, int id, Logger logger)
: base(service, id, null, service.Client.GatewaySocket, logger)
{
_connectionLock = new Semaphore(1, 1);
_connectionLock = new AsyncLock();
}
//Only disconnects if is current a member of this server
public async Task Leave(VirtualClient client)
{
_connectionLock.WaitOne();
try
using (await _connectionLock.LockAsync())
{
if (CurrentClient == client)
{
@@ -49,16 +49,11 @@ namespace Discord.Audio
await Disconnect();
}
}
finally
{
_connectionLock.Release();
}
}
internal async Task<IAudioClient> Connect(Channel channel)
{
_connectionLock.WaitOne();
try
using (await _connectionLock.LockAsync())
{
bool changeServer = channel.Server != VoiceSocket.Server;
if (changeServer || CurrentClient == null)
@@ -70,10 +65,6 @@ namespace Discord.Audio
await Join(channel);
return CurrentClient;
}
finally
{
_connectionLock.Release();
}
}
}
}

50
src/Discord.Net.Audio/VoiceBuffer.cs
View File

@@ -1,4 +1,5 @@
using System;
using Nito.AsyncEx;
using System;
using System.Threading;
namespace Discord.Audio
@@ -9,8 +10,9 @@ namespace Discord.Audio
private readonly byte[] _buffer;
private readonly byte[] _blankFrame;
private ushort _readCursor, _writeCursor;
private ManualResetEventSlim _notOverflowEvent;
private ManualResetEventSlim _notOverflowEvent;
private bool _isClearing;
private AsyncLock _lock;
public int FrameSize => _frameSize;
public int FrameCount => _frameCount;
@@ -27,6 +29,7 @@ namespace Discord.Audio
_buffer = new byte[_bufferSize];
_blankFrame = new byte[_frameSize];
_notOverflowEvent = new ManualResetEventSlim(); //Notifies when an overflow is solved
_lock = new AsyncLock();
}
public void Push(byte[] buffer, int bytes, CancellationToken cancelToken)
@@ -38,8 +41,8 @@ namespace Discord.Audio
int expectedBytes = wholeFrames * _frameSize;
int lastFrameSize = bytes - expectedBytes;
lock (this)
{
using (_lock.Lock())
{
for (int i = 0, pos = 0; i <= wholeFrames; i++, pos += _frameSize)
{
//If the read cursor is in the next position, wait for it to move.
@@ -83,35 +86,34 @@ namespace Discord.Audio
}
public bool Pop(byte[] buffer)
{
if (_writeCursor == _readCursor)
{
_notOverflowEvent.Set();
return false;
}
{
//using (_lock.Lock())
//{
if (_writeCursor == _readCursor)
{
_notOverflowEvent.Set();
return false;
}
bool isClearing = _isClearing;
if (!isClearing)
Buffer.BlockCopy(_buffer, _readCursor * _frameSize, buffer, 0, _frameSize);
bool isClearing = _isClearing;
if (!isClearing)
Buffer.BlockCopy(_buffer, _readCursor * _frameSize, buffer, 0, _frameSize);
//Advance the read cursor to the next position
AdvanceCursorPos(ref _readCursor);
_notOverflowEvent.Set();
return !isClearing;
//Advance the read cursor to the next position
AdvanceCursorPos(ref _readCursor);
_notOverflowEvent.Set();
return !isClearing;
//}
}
public void Clear(CancellationToken cancelToken)
{
lock (this)
{
using (_lock.Lock())
{
_isClearing = true;
for (int i = 0; i < _frameCount; i++)
Buffer.BlockCopy(_blankFrame, 0, _buffer, i * _frameCount, i++);
try
{
Wait(cancelToken);
}
catch (OperationCanceledException) { }
_writeCursor = 0;
_readCursor = 0;
_isClearing = false;