feat: Adds a memory mirrored ring buffer for networking. (#1533)
## Breaking Changes
Incoming packet registration signature has changed to:
```cs
delegate* void OnReceiveCallback(NetState state, SpanReader reader, int packetLength);
IncomingPackets.Register(int packetID, int length, bool ingame, OnReceiveCallback onReceive);
```
For example, an incoming packet handler signature would now look like this:
```cs
public static void SomeIncomingPacket(NetState state, SpanReader reader, int packetLength)
{
// Parse the data
}
```
## Summary
Updates the network Pipe class to use a mirrored memory technique. This technique involves mapping the same physical memory to two contiguous virtual memory spaces so the byte buffer appears duplicated. This allows writing to a double-sized array to wrap around without the need for the `CircularBuffer` classes.
In practice this allows us to use `Span<byte>` as if the buffer was a regular array.
### Bug Fixes
- [X] Fixes bad fixed length string parsing
This commit is contained in:
Kamron Batman
GitHub
parent
629a5008c3
commit
10a69bf754
83 changed files with 958 additions and 2432 deletions
|
|
@ -13,15 +13,17 @@
|
|||
* along with this program. If not, see <http://www.gnu.org/licenses/>. *
|
||||
*************************************************************************/
|
||||
|
||||
using System.Buffers;
|
||||
|
||||
namespace Server.Network;
|
||||
|
||||
public ref struct EncodedReader
|
||||
{
|
||||
private CircularBufferReader _reader;
|
||||
private SpanReader _reader;
|
||||
|
||||
public EncodedReader(CircularBufferReader reader) => _reader = reader;
|
||||
public EncodedReader(SpanReader reader) => _reader = reader;
|
||||
|
||||
public void Trace(NetState state) => _reader.Trace(state);
|
||||
public void Trace(NetState state) => state.Trace(_reader.Buffer);
|
||||
|
||||
public int ReadInt32() => _reader.ReadByte() != 0 ? 0 : _reader.ReadInt32();
|
||||
|
||||
|
|
|
|||
|
|
@ -34,8 +34,8 @@ namespace Server.Network;
|
|||
|
||||
public delegate void NetStateCreatedCallback(NetState ns);
|
||||
|
||||
public delegate void DecodePacket(CircularBuffer<byte> buffer, ref int length);
|
||||
public delegate void EncodePacket(ReadOnlySpan<byte> inputBuffer, CircularBuffer<byte> outputBuffer, out int length);
|
||||
public delegate void DecodePacket(Span<byte> buffer, ref int length);
|
||||
public delegate int EncodePacket(ReadOnlySpan<byte> inputBuffer, Span<byte> outputBuffer);
|
||||
|
||||
public partial class NetState : IComparable<NetState>
|
||||
{
|
||||
|
|
@ -123,8 +123,8 @@ public partial class NetState : IComparable<NetState>
|
|||
HuePickers = new List<HuePicker>();
|
||||
Menus = new List<IMenu>();
|
||||
Trades = new List<SecureTrade>();
|
||||
RecvPipe = new Pipe<byte>(GC.AllocateUninitializedArray<byte>(RecvPipeSize));
|
||||
SendPipe = new Pipe<byte>(GC.AllocateUninitializedArray<byte>(SendPipeSize));
|
||||
RecvPipe = new Pipe(RecvPipeSize);
|
||||
SendPipe = new Pipe(SendPipeSize);
|
||||
_nextActivityCheck = Core.TickCount + 30000;
|
||||
ConnectedOn = Core.Now;
|
||||
|
||||
|
|
@ -202,9 +202,9 @@ public partial class NetState : IComparable<NetState>
|
|||
|
||||
public bool Seeded { get; set; }
|
||||
|
||||
public Pipe<byte> RecvPipe { get; }
|
||||
public Pipe RecvPipe { get; }
|
||||
|
||||
public Pipe<byte> SendPipe { get; }
|
||||
public Pipe SendPipe { get; }
|
||||
|
||||
public bool Running => _running;
|
||||
|
||||
|
|
@ -454,7 +454,7 @@ public partial class NetState : IComparable<NetState>
|
|||
|
||||
public override string ToString() => _toString;
|
||||
|
||||
public bool GetSendBuffer(out CircularBuffer<byte> cBuffer)
|
||||
public bool GetSendBuffer(out Span<byte> buffer)
|
||||
{
|
||||
#if THREADGUARD
|
||||
if (Thread.CurrentThread != Core.Thread)
|
||||
|
|
@ -466,10 +466,8 @@ public partial class NetState : IComparable<NetState>
|
|||
return;
|
||||
}
|
||||
#endif
|
||||
var result = SendPipe.Writer.TryGetMemory();
|
||||
cBuffer = new CircularBuffer<byte>(result.Buffer);
|
||||
|
||||
return !(result.IsClosed || result.Length <= 0);
|
||||
buffer = SendPipe.Writer.AvailableToWrite();
|
||||
return !(SendPipe.Writer.IsClosed || buffer.Length <= 0);
|
||||
}
|
||||
|
||||
public void Send(ReadOnlySpan<byte> span)
|
||||
|
|
@ -497,16 +495,16 @@ public partial class NetState : IComparable<NetState>
|
|||
|
||||
if (_packetEncoder != null)
|
||||
{
|
||||
_packetEncoder(span, buffer, out length);
|
||||
length = _packetEncoder(span, buffer);
|
||||
}
|
||||
else
|
||||
{
|
||||
buffer.CopyFrom(span);
|
||||
span.CopyTo(buffer);
|
||||
}
|
||||
|
||||
if (PacketLogging)
|
||||
{
|
||||
LogPacket(span, ReadOnlySpan<byte>.Empty, span.Length, false);
|
||||
LogPacket(span, false);
|
||||
}
|
||||
|
||||
SendPipe.Writer.Advance((uint)length);
|
||||
|
|
@ -545,7 +543,7 @@ public partial class NetState : IComparable<NetState>
|
|||
}
|
||||
}
|
||||
|
||||
private void LogPacket(ReadOnlySpan<byte> first, ReadOnlySpan<byte> second, int totalLength, bool incoming)
|
||||
private void LogPacket(ReadOnlySpan<byte> buffer, bool incoming)
|
||||
{
|
||||
try
|
||||
{
|
||||
|
|
@ -557,8 +555,8 @@ public partial class NetState : IComparable<NetState>
|
|||
const string outgoingStr = "Server -> Client";
|
||||
|
||||
using var sw = new StreamWriter(logPath, true);
|
||||
sw.WriteLine($"{Core.Now:HH:mm:ss.ffff}: {(incoming ? incomingStr : outgoingStr)} 0x{first[0]:X2} (Length: {totalLength})");
|
||||
sw.FormatBuffer(first, second, totalLength);
|
||||
sw.WriteLine($"{Core.Now:HH:mm:ss.ffff}: {(incoming ? incomingStr : outgoingStr)} 0x{buffer[0]:X2} (Length: {buffer.Length})");
|
||||
sw.FormatBuffer(buffer);
|
||||
sw.WriteLine();
|
||||
sw.WriteLine();
|
||||
}
|
||||
|
|
@ -587,15 +585,15 @@ public partial class NetState : IComparable<NetState>
|
|||
// Process as many packets as we can synchronously
|
||||
while (_running && _parserState != ParserState.Error && _protocolState != ProtocolState.Error)
|
||||
{
|
||||
var result = reader.TryRead();
|
||||
var length = result.Length;
|
||||
var buffer = reader.AvailableToRead();
|
||||
var length = buffer.Length;
|
||||
|
||||
if (length <= 0)
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
var packetReader = new CircularBufferReader(result.Buffer);
|
||||
var packetReader = new SpanReader(buffer);
|
||||
var packetId = packetReader.ReadByte();
|
||||
int packetLength = length;
|
||||
|
||||
|
|
@ -744,8 +742,6 @@ public partial class NetState : IComparable<NetState>
|
|||
break;
|
||||
}
|
||||
}
|
||||
|
||||
reader.Commit();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
|
|
@ -771,7 +767,7 @@ public partial class NetState : IComparable<NetState>
|
|||
* length is the total buffer length. We might be able to use packetReader.Capacity() instead.
|
||||
* packetLength is the length of the packet that this function actually found.
|
||||
*/
|
||||
private unsafe ParserState HandlePacket(CircularBufferReader packetReader, byte packetId, out int packetLength)
|
||||
private unsafe ParserState HandlePacket(SpanReader packetReader, byte packetId, out int packetLength)
|
||||
{
|
||||
PacketHandler handler = IncomingPackets.GetHandler(packetId);
|
||||
int length = packetReader.Length;
|
||||
|
|
@ -842,7 +838,7 @@ public partial class NetState : IComparable<NetState>
|
|||
|
||||
if (PacketLogging)
|
||||
{
|
||||
LogPacket(packetReader.First, packetReader.Second, packetLength, true);
|
||||
LogPacket(packetReader.Buffer[..packetLength], true);
|
||||
}
|
||||
|
||||
handler.OnReceive(this, packetReader, packetLength);
|
||||
|
|
@ -861,12 +857,10 @@ public partial class NetState : IComparable<NetState>
|
|||
return true;
|
||||
}
|
||||
|
||||
SendPipe.Writer.Flush();
|
||||
|
||||
var reader = SendPipe.Reader;
|
||||
var result = reader.TryRead();
|
||||
var buffer = reader.AvailableToRead();
|
||||
|
||||
if (result.IsClosed || result.Length == 0)
|
||||
if (reader.IsClosed || buffer.Length == 0)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
|
@ -875,7 +869,7 @@ public partial class NetState : IComparable<NetState>
|
|||
|
||||
try
|
||||
{
|
||||
bytesWritten = Connection.Send(result.Buffer, SocketFlags.None);
|
||||
bytesWritten = Connection.Send(buffer, SocketFlags.None);
|
||||
}
|
||||
catch (SocketException ex)
|
||||
{
|
||||
|
|
@ -897,21 +891,20 @@ public partial class NetState : IComparable<NetState>
|
|||
reader.Advance((uint)bytesWritten);
|
||||
}
|
||||
|
||||
return bytesWritten == result.Length;
|
||||
return bytesWritten == buffer.Length;
|
||||
}
|
||||
|
||||
private void DecodePacket(ArraySegment<byte>[] buffer, ref int length)
|
||||
private void DecodePacket(Span<byte> buffer, ref int length)
|
||||
{
|
||||
CircularBuffer<byte> cBuffer = new CircularBuffer<byte>(buffer);
|
||||
_packetDecoder?.Invoke(cBuffer, ref length);
|
||||
_packetDecoder?.Invoke(buffer, ref length);
|
||||
}
|
||||
|
||||
private void ReceiveData()
|
||||
{
|
||||
var writer = RecvPipe.Writer;
|
||||
var result = writer.TryGetMemory();
|
||||
var buffer = writer.AvailableToWrite();
|
||||
|
||||
if (result.IsClosed || result.Length == 0)
|
||||
if (writer.IsClosed || buffer.Length == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
|
@ -920,7 +913,7 @@ public partial class NetState : IComparable<NetState>
|
|||
|
||||
try
|
||||
{
|
||||
bytesWritten = Connection.Receive(result.Buffer, SocketFlags.None);
|
||||
bytesWritten = Connection.Receive(buffer, SocketFlags.None);
|
||||
}
|
||||
catch (SocketException ex)
|
||||
{
|
||||
|
|
@ -943,7 +936,7 @@ public partial class NetState : IComparable<NetState>
|
|||
return;
|
||||
}
|
||||
|
||||
DecodePacket(result.Buffer, ref bytesWritten);
|
||||
DecodePacket(buffer, ref bytesWritten);
|
||||
|
||||
writer.Advance((uint)bytesWritten);
|
||||
_nextActivityCheck = Core.TickCount + 90000;
|
||||
|
|
@ -1042,6 +1035,28 @@ public partial class NetState : IComparable<NetState>
|
|||
}
|
||||
}
|
||||
|
||||
public void Trace(ReadOnlySpan<byte> buffer)
|
||||
{
|
||||
// We don't have data, so nothing to trace
|
||||
if (buffer.Length == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
using var sw = new StreamWriter("unhandled-packets.log", true);
|
||||
sw.WriteLine("Client: {0}: Unhandled packet 0x{1:X2}", this, buffer[0]);
|
||||
sw.FormatBuffer(buffer);
|
||||
sw.WriteLine();
|
||||
sw.WriteLine();
|
||||
}
|
||||
catch
|
||||
{
|
||||
// ignored
|
||||
}
|
||||
}
|
||||
|
||||
public static void TraceException(Exception ex)
|
||||
{
|
||||
try
|
||||
|
|
@ -1148,6 +1163,8 @@ public partial class NetState : IComparable<NetState>
|
|||
|
||||
Connection.Close();
|
||||
_handle.Free();
|
||||
RecvPipe.Dispose();
|
||||
SendPipe.Dispose();
|
||||
|
||||
Mobile = null;
|
||||
|
||||
|
|
|
|||
|
|
@ -61,129 +61,6 @@ public static class NetworkCompression
|
|||
0x4, 0x00D
|
||||
};
|
||||
|
||||
public static void Compress(ReadOnlySpan<byte> input, CircularBuffer<byte> output, out int length)
|
||||
{
|
||||
length = Compress(input, output);
|
||||
}
|
||||
|
||||
public static int Compress(ReadOnlySpan<byte> input, CircularBuffer<byte> output)
|
||||
{
|
||||
if (input.Length > DefiniteOverflow)
|
||||
{
|
||||
return 0;
|
||||
}
|
||||
|
||||
int bitCount = 0;
|
||||
int bitValue = 0;
|
||||
|
||||
int inputIdx = 0;
|
||||
int outputIdx = 0;
|
||||
|
||||
while (inputIdx < input.Length)
|
||||
{
|
||||
int i = input[inputIdx++] << 1;
|
||||
|
||||
bitCount += _huffmanTable[i];
|
||||
bitValue = (bitValue << _huffmanTable[i]) | _huffmanTable[i + 1];
|
||||
|
||||
while (bitCount >= 8)
|
||||
{
|
||||
bitCount -= 8;
|
||||
|
||||
if (output.Length < outputIdx + 1)
|
||||
{
|
||||
return 0;
|
||||
}
|
||||
|
||||
output[outputIdx++] = (byte)(bitValue >> bitCount);
|
||||
}
|
||||
}
|
||||
|
||||
// terminal code
|
||||
bitCount += _huffmanTable[0x200];
|
||||
bitValue = (bitValue << _huffmanTable[0x200]) | _huffmanTable[0x201];
|
||||
|
||||
// align on byte boundary
|
||||
if ((bitCount & 7) != 0)
|
||||
{
|
||||
bitValue <<= 8 - (bitCount & 7);
|
||||
bitCount += 8 - (bitCount & 7);
|
||||
}
|
||||
|
||||
while (bitCount >= 8)
|
||||
{
|
||||
bitCount -= 8;
|
||||
|
||||
if (output.Length < outputIdx + 1)
|
||||
{
|
||||
return 0;
|
||||
}
|
||||
|
||||
output[outputIdx++] = (byte)(bitValue >> bitCount);
|
||||
}
|
||||
|
||||
return outputIdx;
|
||||
}
|
||||
|
||||
public static int Compress(CircularBuffer<byte> input, CircularBuffer<byte> output)
|
||||
{
|
||||
if (input.Length > DefiniteOverflow)
|
||||
{
|
||||
return 0;
|
||||
}
|
||||
|
||||
int bitCount = 0;
|
||||
int bitValue = 0;
|
||||
|
||||
int inputIdx = 0;
|
||||
int outputIdx = 0;
|
||||
|
||||
while (inputIdx < input.Length)
|
||||
{
|
||||
int i = input[inputIdx++] << 1;
|
||||
|
||||
bitCount += _huffmanTable[i];
|
||||
bitValue = (bitValue << _huffmanTable[i]) | _huffmanTable[i + 1];
|
||||
|
||||
while (bitCount >= 8)
|
||||
{
|
||||
bitCount -= 8;
|
||||
|
||||
if (output.Length < outputIdx + 1)
|
||||
{
|
||||
return 0;
|
||||
}
|
||||
|
||||
output[outputIdx++] = (byte)(bitValue >> bitCount);
|
||||
}
|
||||
}
|
||||
|
||||
// terminal code
|
||||
bitCount += _huffmanTable[0x200];
|
||||
bitValue = (bitValue << _huffmanTable[0x200]) | _huffmanTable[0x201];
|
||||
|
||||
// align on byte boundary
|
||||
if ((bitCount & 7) != 0)
|
||||
{
|
||||
bitValue <<= 8 - (bitCount & 7);
|
||||
bitCount += 8 - (bitCount & 7);
|
||||
}
|
||||
|
||||
while (bitCount >= 8)
|
||||
{
|
||||
bitCount -= 8;
|
||||
|
||||
if (output.Length < outputIdx + 1)
|
||||
{
|
||||
return 0;
|
||||
}
|
||||
|
||||
output[outputIdx++] = (byte)(bitValue >> bitCount);
|
||||
}
|
||||
|
||||
return outputIdx;
|
||||
}
|
||||
|
||||
public static int Compress(ReadOnlySpan<byte> input, Span<byte> output)
|
||||
{
|
||||
if (input.Length > DefiniteOverflow)
|
||||
|
|
|
|||
|
|
@ -13,13 +13,15 @@
|
|||
* along with this program. If not, see <http://www.gnu.org/licenses/>. *
|
||||
*************************************************************************/
|
||||
|
||||
using System.Buffers;
|
||||
|
||||
namespace Server.Network;
|
||||
|
||||
public unsafe class PacketHandler
|
||||
{
|
||||
private readonly int _length;
|
||||
|
||||
public PacketHandler(int packetID, int length, bool ingame, delegate*<NetState, CircularBufferReader, int, void> onReceive)
|
||||
public PacketHandler(int packetID, int length, bool ingame, delegate*<NetState, SpanReader, int, void> onReceive)
|
||||
{
|
||||
_length = length;
|
||||
PacketID = packetID;
|
||||
|
|
@ -31,7 +33,7 @@ public unsafe class PacketHandler
|
|||
|
||||
public virtual int GetLength(NetState ns) => _length;
|
||||
|
||||
public delegate*<NetState, CircularBufferReader, int, void> OnReceive { get; }
|
||||
public delegate*<NetState, SpanReader, int, void> OnReceive { get; }
|
||||
|
||||
public delegate*<int, NetState, out bool, bool> ThrottleCallback { get; set; }
|
||||
|
||||
|
|
|
|||
|
|
@ -22,15 +22,6 @@ namespace Server.Network;
|
|||
|
||||
public static class PacketUtilities
|
||||
{
|
||||
[MethodImpl(MethodImplOptions.AggressiveInlining)]
|
||||
public static void WritePacketLength(this ref CircularBufferWriter writer)
|
||||
{
|
||||
var length = writer.Position;
|
||||
writer.Seek(1, SeekOrigin.Begin);
|
||||
writer.Write((ushort)length);
|
||||
writer.Seek(length, SeekOrigin.Begin);
|
||||
}
|
||||
|
||||
[MethodImpl(MethodImplOptions.AggressiveInlining)]
|
||||
public static void WritePacketLength(this ref SpanWriter writer)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@
|
|||
* along with this program. If not, see <http://www.gnu.org/licenses/>. *
|
||||
*************************************************************************/
|
||||
|
||||
using System.Buffers;
|
||||
using System.Runtime.CompilerServices;
|
||||
|
||||
namespace Server.Network;
|
||||
|
|
@ -25,7 +26,7 @@ public static class IncomingPackets
|
|||
|
||||
[MethodImpl(MethodImplOptions.AggressiveInlining)]
|
||||
public static unsafe void Register(int packetID, int length, bool ingame,
|
||||
delegate*<NetState, CircularBufferReader, int, void> onReceive) =>
|
||||
delegate*<NetState, SpanReader, int, void> onReceive) =>
|
||||
Register(new PacketHandler(packetID, length, ingame, onReceive));
|
||||
|
||||
public static void Register(PacketHandler packetHandler)
|
||||
|
|
|
|||
|
|
@ -14,137 +14,35 @@
|
|||
*************************************************************************/
|
||||
|
||||
using System;
|
||||
using System.Runtime.CompilerServices;
|
||||
using System.Threading;
|
||||
using System.IO;
|
||||
using System.Runtime.InteropServices;
|
||||
|
||||
namespace Server.Network;
|
||||
|
||||
public interface IPipeTask<T> : INotifyCompletion
|
||||
public partial class Pipe : IDisposable
|
||||
{
|
||||
public IPipeTask<T> GetAwaiter();
|
||||
|
||||
public bool IsCompleted { get; }
|
||||
|
||||
public T GetResult();
|
||||
}
|
||||
|
||||
public class Pipe<T>
|
||||
{
|
||||
public struct Result
|
||||
public class PipeWriter
|
||||
{
|
||||
public ArraySegment<T>[] Buffer { get; }
|
||||
public bool IsClosed { get; set; }
|
||||
private readonly Pipe _pipe;
|
||||
|
||||
public int Length
|
||||
{
|
||||
get
|
||||
{
|
||||
var length = 0;
|
||||
for (int i = 0; i < Buffer.Length; i++)
|
||||
{
|
||||
length += Buffer[i].Count;
|
||||
}
|
||||
internal PipeWriter(Pipe pipe) => _pipe = pipe;
|
||||
|
||||
return length;
|
||||
}
|
||||
}
|
||||
|
||||
public void CopyFrom(ReadOnlySpan<T> bytes)
|
||||
{
|
||||
var remaining = bytes.Length;
|
||||
var offset = 0;
|
||||
|
||||
if (remaining == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
for (int i = 0; i < 2; i++)
|
||||
{
|
||||
var buffer = Buffer[i];
|
||||
var sz = Math.Min(remaining, buffer.Count);
|
||||
bytes.Slice(offset, sz).CopyTo(buffer);
|
||||
|
||||
remaining -= sz;
|
||||
offset += sz;
|
||||
|
||||
if (remaining == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
throw new OutOfMemoryException();
|
||||
}
|
||||
|
||||
public Result(int segments)
|
||||
{
|
||||
IsClosed = false;
|
||||
Buffer = new ArraySegment<T>[segments];
|
||||
}
|
||||
}
|
||||
|
||||
public class PipeWriter : IPipeTask<Result>
|
||||
{
|
||||
private readonly Pipe<T> _pipe;
|
||||
|
||||
private Result _result = new(2);
|
||||
|
||||
internal PipeWriter(Pipe<T> pipe) => _pipe = pipe;
|
||||
|
||||
[MethodImpl(MethodImplOptions.AggressiveInlining)]
|
||||
public uint GetAvailable()
|
||||
public unsafe Span<byte> AvailableToWrite()
|
||||
{
|
||||
var read = _pipe._readIdx;
|
||||
var write = _pipe._writeIdx;
|
||||
|
||||
uint sz;
|
||||
if (read <= write)
|
||||
{
|
||||
if (read == 0)
|
||||
{
|
||||
return _pipe.Size - write - 1;
|
||||
}
|
||||
|
||||
return _pipe.Size - write + (read - 1);
|
||||
}
|
||||
|
||||
return read - write - 1;
|
||||
}
|
||||
|
||||
public Result TryGetMemory()
|
||||
{
|
||||
var read = _pipe._readIdx;
|
||||
var write = _pipe._writeIdx;
|
||||
|
||||
_result.IsClosed = _pipe._closed;
|
||||
|
||||
if (read <= write)
|
||||
{
|
||||
var readZero = read == 0;
|
||||
var sz = _pipe.Size - write - (readZero ? 1 : 0);
|
||||
|
||||
_result.Buffer[0] = sz == 0 ? ArraySegment<T>.Empty : new ArraySegment<T>(_pipe._buffer, (int)write, (int)sz);
|
||||
_result.Buffer[1] = readZero ? ArraySegment<T>.Empty : new ArraySegment<T>(_pipe._buffer, 0, (int)read - 1);
|
||||
sz = _pipe.Size - write + read - 1;
|
||||
}
|
||||
else
|
||||
{
|
||||
var sz = read - write - 1;
|
||||
|
||||
_result.Buffer[0] = sz == 0 ? ArraySegment<T>.Empty : new ArraySegment<T>(_pipe._buffer, (int)write, (int)sz);
|
||||
_result.Buffer[1] = ArraySegment<T>.Empty;
|
||||
sz = read - write - 1;
|
||||
}
|
||||
|
||||
return _result;
|
||||
}
|
||||
|
||||
public IPipeTask<Result> GetMemory()
|
||||
{
|
||||
if (_pipe._writeAwaitBeginning)
|
||||
{
|
||||
throw new Exception("Double await on writer");
|
||||
}
|
||||
|
||||
return this;
|
||||
return new Span<byte>((void*)(_pipe._buffer + write), (int)sz);
|
||||
}
|
||||
|
||||
public void Advance(uint count)
|
||||
|
|
@ -159,14 +57,14 @@ public class Pipe<T>
|
|||
|
||||
if (count > _pipe.Size - 1)
|
||||
{
|
||||
throw new InvalidOperationException();
|
||||
throw new EndOfPipeException("Unable to advance beyond the end of the pipe.");
|
||||
}
|
||||
|
||||
if (read <= write)
|
||||
{
|
||||
if (count > read + _pipe.Size - write - 1)
|
||||
{
|
||||
throw new InvalidOperationException();
|
||||
throw new EndOfPipeException("Unable to advance beyond the end of the pipe.");
|
||||
}
|
||||
|
||||
var sz = Math.Min(count, _pipe.Size - write);
|
||||
|
|
@ -182,7 +80,7 @@ public class Pipe<T>
|
|||
{
|
||||
if (count >= read)
|
||||
{
|
||||
throw new InvalidOperationException();
|
||||
throw new EndOfPipeException("Unable to advance beyond the end of the pipe.");
|
||||
}
|
||||
|
||||
write = count;
|
||||
|
|
@ -192,7 +90,7 @@ public class Pipe<T>
|
|||
{
|
||||
if (count > read - write - 1)
|
||||
{
|
||||
throw new InvalidOperationException();
|
||||
throw new EndOfPipeException("Unable to advance beyond the end of the pipe.");
|
||||
}
|
||||
|
||||
write += count;
|
||||
|
|
@ -202,146 +100,39 @@ public class Pipe<T>
|
|||
// the read pointer. Check that here.
|
||||
if (write == read)
|
||||
{
|
||||
throw new InvalidOperationException("Write index equals read index after advance");
|
||||
throw new EndOfPipeException("Unable to advance beyond the end of the pipe.");
|
||||
}
|
||||
|
||||
_pipe._writeIdx = write;
|
||||
}
|
||||
|
||||
public void Close()
|
||||
{
|
||||
_pipe._closed = true;
|
||||
public void Close() => _pipe._closed = true;
|
||||
|
||||
var waiting = _pipe._readAwaitBeginning;
|
||||
|
||||
if (!waiting)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
Action continuation;
|
||||
|
||||
do
|
||||
{
|
||||
continuation = _pipe._readContinuation;
|
||||
} while (continuation == null);
|
||||
|
||||
_pipe._readContinuation = null;
|
||||
_pipe._readAwaitBeginning = false;
|
||||
|
||||
ThreadPool.UnsafeQueueUserWorkItem(_ => continuation(), true);
|
||||
}
|
||||
|
||||
public void Flush()
|
||||
{
|
||||
if (_pipe._readIdx == _pipe._writeIdx)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var waiting = _pipe._readAwaitBeginning;
|
||||
|
||||
if (!waiting)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
Action continuation;
|
||||
|
||||
do
|
||||
{
|
||||
continuation = _pipe._readContinuation;
|
||||
} while (continuation == null);
|
||||
|
||||
_pipe._readContinuation = null;
|
||||
_pipe._readAwaitBeginning = false;
|
||||
|
||||
ThreadPool.UnsafeQueueUserWorkItem(_ => continuation(), true);
|
||||
}
|
||||
|
||||
#region Awaitable
|
||||
|
||||
// The following makes it possible to await the writer. Do not use any of this directly.
|
||||
|
||||
public IPipeTask<Result> GetAwaiter() => this;
|
||||
|
||||
public bool IsCompleted
|
||||
{
|
||||
get
|
||||
{
|
||||
if (GetAvailable() > 0)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
if (_pipe._closed)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
_pipe._writeAwaitBeginning = true;
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
public Result GetResult() => TryGetMemory();
|
||||
|
||||
public void OnCompleted(Action continuation) => _pipe._writeContinuation = continuation;
|
||||
|
||||
#endregion
|
||||
public bool IsClosed => _pipe._closed;
|
||||
}
|
||||
|
||||
public class PipeReader : IPipeTask<Result>
|
||||
public class PipeReader
|
||||
{
|
||||
private readonly Pipe<T> _pipe;
|
||||
private readonly Pipe _pipe;
|
||||
|
||||
private Result _result = new(2);
|
||||
internal PipeReader(Pipe pipe) => _pipe = pipe;
|
||||
|
||||
internal PipeReader(Pipe<T> pipe) => _pipe = pipe;
|
||||
|
||||
[MethodImpl(MethodImplOptions.AggressiveInlining)]
|
||||
public uint GetAvailable()
|
||||
public unsafe Span<byte> AvailableToRead()
|
||||
{
|
||||
var read = _pipe._readIdx;
|
||||
var write = _pipe._writeIdx;
|
||||
|
||||
uint sz;
|
||||
if (read <= write)
|
||||
{
|
||||
return write - read;
|
||||
}
|
||||
|
||||
return write + _pipe.Size - read;
|
||||
}
|
||||
|
||||
public Result TryRead()
|
||||
{
|
||||
var read = _pipe._readIdx;
|
||||
var write = _pipe._writeIdx;
|
||||
|
||||
_result.IsClosed = _pipe._closed;
|
||||
|
||||
if (read <= write)
|
||||
{
|
||||
_result.Buffer[0] = write - read == 0 ? ArraySegment<T>.Empty : new ArraySegment<T>(_pipe._buffer, (int)read, (int)(write - read));
|
||||
_result.Buffer[1] = ArraySegment<T>.Empty;
|
||||
sz = write - read;
|
||||
}
|
||||
else
|
||||
{
|
||||
_result.Buffer[0] = _pipe.Size - read == 0 ? ArraySegment<T>.Empty : new ArraySegment<T>(_pipe._buffer, (int)read, (int)(_pipe.Size - read));
|
||||
_result.Buffer[1] = write == 0 ? ArraySegment<T>.Empty : new ArraySegment<T>(_pipe._buffer, 0, (int)write);
|
||||
sz = _pipe.Size - read + write;
|
||||
}
|
||||
|
||||
return _result;
|
||||
}
|
||||
|
||||
public IPipeTask<Result> Read()
|
||||
{
|
||||
if (_pipe._readAwaitBeginning)
|
||||
{
|
||||
throw new Exception("Double await on reader");
|
||||
}
|
||||
|
||||
return this;
|
||||
return new Span<byte>((void*)(_pipe._buffer + read), (int)sz);
|
||||
}
|
||||
|
||||
public void Advance(uint count)
|
||||
|
|
@ -353,7 +144,7 @@ public class Pipe<T>
|
|||
{
|
||||
if (count > write - read)
|
||||
{
|
||||
throw new InvalidOperationException();
|
||||
throw new EndOfPipeException("Unable to advance beyond the end of the pipe.");
|
||||
}
|
||||
|
||||
read += count;
|
||||
|
|
@ -373,118 +164,199 @@ public class Pipe<T>
|
|||
{
|
||||
if (count > write)
|
||||
{
|
||||
throw new InvalidOperationException();
|
||||
throw new EndOfPipeException("Unable to advance beyond the end of the pipe.");
|
||||
}
|
||||
|
||||
read = count;
|
||||
}
|
||||
}
|
||||
|
||||
_pipe._readIdx = read;
|
||||
}
|
||||
|
||||
public void Commit()
|
||||
{
|
||||
if (_pipe._readIdx == ((_pipe._writeIdx + 1) & (_pipe.Size - 1)))
|
||||
if (read == write)
|
||||
{
|
||||
return;
|
||||
// If the read pointer catches up to the write pointer, then the pipe is empty.
|
||||
// As a performance optimization, set both to 0. This should improve the chances cache lines are hit.
|
||||
_pipe._readIdx = 0;
|
||||
_pipe._writeIdx = 0;
|
||||
}
|
||||
|
||||
var waiting = _pipe._writeAwaitBeginning;
|
||||
|
||||
if (!waiting)
|
||||
else
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
Action continuation;
|
||||
|
||||
do
|
||||
{
|
||||
continuation = _pipe._writeContinuation;
|
||||
} while (continuation == null);
|
||||
|
||||
_pipe._writeContinuation = null;
|
||||
_pipe._writeAwaitBeginning = false;
|
||||
|
||||
ThreadPool.UnsafeQueueUserWorkItem(_ => continuation(), true);
|
||||
}
|
||||
|
||||
public void Close()
|
||||
{
|
||||
_pipe._closed = true;
|
||||
|
||||
var waiting = _pipe._writeAwaitBeginning;
|
||||
|
||||
if (!waiting)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
Action continuation;
|
||||
|
||||
do
|
||||
{
|
||||
continuation = _pipe._writeContinuation;
|
||||
} while (continuation == null);
|
||||
|
||||
_pipe._writeContinuation = null;
|
||||
_pipe._writeAwaitBeginning = false;
|
||||
|
||||
ThreadPool.UnsafeQueueUserWorkItem(_ => continuation(), true);
|
||||
}
|
||||
|
||||
#region Awaitable
|
||||
|
||||
// The following makes it possible to await the reader. Do not use any of this directly.
|
||||
|
||||
public IPipeTask<Result> GetAwaiter() => this;
|
||||
|
||||
public bool IsCompleted
|
||||
{
|
||||
get
|
||||
{
|
||||
if (GetAvailable() > 0)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
if (_pipe._closed)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
_pipe._readAwaitBeginning = true;
|
||||
return false;
|
||||
_pipe._readIdx = read;
|
||||
}
|
||||
}
|
||||
|
||||
public Result GetResult() => TryRead();
|
||||
public void Close() => _pipe._closed = true;
|
||||
|
||||
public void OnCompleted(Action continuation) => _pipe._readContinuation = continuation;
|
||||
|
||||
#endregion
|
||||
public bool IsClosed => _pipe._closed;
|
||||
}
|
||||
|
||||
private readonly T[] _buffer;
|
||||
private volatile uint _writeIdx;
|
||||
private volatile uint _readIdx;
|
||||
private IntPtr _handle; // Doubles as the file descriptor for linux/darwin
|
||||
private IntPtr _buffer;
|
||||
private readonly uint _bufferSize;
|
||||
private uint _writeIdx;
|
||||
private uint _readIdx;
|
||||
private bool _closed;
|
||||
|
||||
public PipeWriter Writer { get; }
|
||||
public PipeReader Reader { get; }
|
||||
|
||||
public uint Size => (uint)_buffer.Length;
|
||||
public uint Size => _bufferSize;
|
||||
|
||||
public Pipe(T[] buf)
|
||||
public bool Closed => _closed;
|
||||
|
||||
public Pipe(uint size)
|
||||
{
|
||||
// Test if the buffer is a power of two
|
||||
if (buf.Length == 0 || (buf.Length & (buf.Length - 1)) != 0)
|
||||
var pageSize = (uint)Environment.SystemPageSize;
|
||||
|
||||
// Virtual allocation requires multiples of system page size
|
||||
// So let's adjust the requested size rounded to the next available page size
|
||||
var adjustedSize = (size + pageSize - 1) & ~(pageSize - 1);
|
||||
|
||||
if (Core.IsWindows)
|
||||
{
|
||||
throw new ArgumentOutOfRangeException(nameof(buf), "Pipe buffers must have a length that is a power of two");
|
||||
// Reserve a region of virtual memory. We need twice the size so we can later mirror.
|
||||
var region = NativeMethods_Windows.VirtualAlloc2(
|
||||
IntPtr.Zero,
|
||||
IntPtr.Zero,
|
||||
adjustedSize * 2,
|
||||
NativeMethods_Windows.MEM_RESERVE | NativeMethods_Windows.MEM_RESERVE_PLACEHOLDER,
|
||||
NativeMethods_Windows.PAGE_NOACCESS,
|
||||
IntPtr.Zero,
|
||||
0
|
||||
);
|
||||
|
||||
if (region == IntPtr.Zero)
|
||||
{
|
||||
throw new InvalidOperationException($"Allocating virtual memory failed. ({Marshal.GetLastPInvokeError()})");
|
||||
}
|
||||
|
||||
// Releases half of the region so we can map the same memory region twice
|
||||
var freed = NativeMethods_Windows.VirtualFree(
|
||||
region,
|
||||
adjustedSize,
|
||||
NativeMethods_Windows.MEM_RELEASE | NativeMethods_Windows.MEM_PRESERVE_PLACEHOLDER
|
||||
);
|
||||
|
||||
if (!freed)
|
||||
{
|
||||
throw new InvalidOperationException($"Creating virtual placeholder failed. ({Marshal.GetLastPInvokeError()})");
|
||||
}
|
||||
|
||||
// Create a file descriptor
|
||||
_handle = NativeMethods_Windows.CreateFileMappingW(
|
||||
NativeMethods_Windows.InvalidHandleValue,
|
||||
IntPtr.Zero,
|
||||
NativeMethods_Windows.PAGE_READWRITE,
|
||||
0,
|
||||
adjustedSize,
|
||||
null
|
||||
);
|
||||
|
||||
if (_handle == IntPtr.Zero)
|
||||
{
|
||||
throw new InvalidOperationException($"Creating file mapping failed. ({Marshal.GetLastPInvokeError()})");
|
||||
}
|
||||
|
||||
// Map the region to the first half of the virtual space
|
||||
_buffer = NativeMethods_Windows.MapViewOfFile3(
|
||||
_handle,
|
||||
IntPtr.Zero,
|
||||
region,
|
||||
0,
|
||||
adjustedSize,
|
||||
NativeMethods_Windows.MEM_REPLACE_PLACEHOLDER,
|
||||
NativeMethods_Windows.PAGE_READWRITE,
|
||||
IntPtr.Zero,
|
||||
0
|
||||
);
|
||||
|
||||
if (_buffer == IntPtr.Zero)
|
||||
{
|
||||
throw new InvalidOperationException($"Mapping file view failed. ({Marshal.GetLastPInvokeError()})");
|
||||
}
|
||||
|
||||
// Map the same region to the second half of the virtual space
|
||||
var view2 = NativeMethods_Windows.MapViewOfFile3(
|
||||
_handle,
|
||||
IntPtr.Zero,
|
||||
new IntPtr(_buffer + adjustedSize),
|
||||
0,
|
||||
adjustedSize,
|
||||
NativeMethods_Windows.MEM_REPLACE_PLACEHOLDER,
|
||||
NativeMethods_Windows.PAGE_READWRITE,
|
||||
IntPtr.Zero,
|
||||
0
|
||||
);
|
||||
|
||||
if (view2 == IntPtr.Zero)
|
||||
{
|
||||
throw new InvalidOperationException($"Mapping file view mirror failed. ({Marshal.GetLastPInvokeError()})");
|
||||
}
|
||||
}
|
||||
else if (Core.IsLinux || Core.IsDarwin)
|
||||
{
|
||||
var anon = Core.IsLinux ? NativeMethods_Linux.MAP_ANONYMOUS : NativeMethods_Linux.MAP_ANON;
|
||||
|
||||
int fd;
|
||||
|
||||
if (Core.IsLinux)
|
||||
{
|
||||
// Create a memory-backed file descriptor
|
||||
fd = NativeMethods_Linux.memfd_create("mirrored_ring_buffer", 0);
|
||||
}
|
||||
else
|
||||
{
|
||||
var fdName = $"/muo/ring/{GetHashCode()}";
|
||||
fd = NativeMethods_Linux.shm_open(fdName, NativeMethods_Linux.O_CREAT | NativeMethods_Linux.O_RDWR, 0600);
|
||||
|
||||
// Unlink immediately to emulate memfd_create() functionality
|
||||
NativeMethods_Linux.shm_unlink(fdName);
|
||||
}
|
||||
|
||||
if (fd == NativeMethods_Linux.InvalidPtrValue)
|
||||
{
|
||||
throw new InvalidOperationException($"Creating file descriptor failed. ({Marshal.GetLastPInvokeError()})");
|
||||
}
|
||||
|
||||
// Set the size of the file descriptor
|
||||
if (NativeMethods_Linux.ftruncate(fd, (int)adjustedSize) != 0)
|
||||
{
|
||||
throw new InvalidOperationException($"Setting file descriptor size failed. ({Marshal.GetLastPInvokeError()})");
|
||||
}
|
||||
|
||||
// Get virtual address space, must be double the size so we can map twice
|
||||
_buffer = NativeMethods_Linux.mmap(IntPtr.Zero, adjustedSize * 2,
|
||||
NativeMethods_Linux.PROT_READ | NativeMethods_Linux.PROT_WRITE,
|
||||
NativeMethods_Linux.MAP_PRIVATE | anon, NativeMethods_Linux.InvalidFileDescriptor, 0);
|
||||
|
||||
if (_buffer == NativeMethods_Linux.InvalidPtrValue)
|
||||
{
|
||||
throw new InsufficientMemoryException($"Allocating virtual memory failed. ({Marshal.GetLastPInvokeError()})");
|
||||
}
|
||||
|
||||
// Map the file descriptor to the first half of the virtual space
|
||||
var view1 = NativeMethods_Linux.mmap(_buffer, adjustedSize,
|
||||
NativeMethods_Linux.PROT_READ | NativeMethods_Linux.PROT_WRITE,
|
||||
NativeMethods_Linux.MAP_SHARED | NativeMethods_Linux.MAP_FIXED, fd, 0);
|
||||
|
||||
if (view1 == NativeMethods_Linux.InvalidPtrValue)
|
||||
{
|
||||
throw new InvalidOperationException($"Mapping memory failed. ({Marshal.GetLastPInvokeError()})");
|
||||
}
|
||||
|
||||
// Map the file descriptor to the second half of the virtual space
|
||||
var view2 = NativeMethods_Linux.mmap(new IntPtr(_buffer + adjustedSize), adjustedSize,
|
||||
NativeMethods_Linux.PROT_READ | NativeMethods_Linux.PROT_WRITE,
|
||||
NativeMethods_Linux.MAP_SHARED | NativeMethods_Linux.MAP_FIXED, fd, 0);
|
||||
|
||||
if (view2 == NativeMethods_Linux.InvalidPtrValue)
|
||||
{
|
||||
throw new InvalidOperationException($"Mapping mirrored memory failed. ({Marshal.GetLastPInvokeError()})");
|
||||
}
|
||||
|
||||
_handle = fd;
|
||||
}
|
||||
|
||||
_buffer = buf;
|
||||
_bufferSize = adjustedSize;
|
||||
_writeIdx = 0;
|
||||
_readIdx = 0;
|
||||
_closed = false;
|
||||
|
|
@ -493,12 +365,155 @@ public class Pipe<T>
|
|||
Reader = new PipeReader(this);
|
||||
}
|
||||
|
||||
#region Awaitable
|
||||
private volatile bool _readAwaitBeginning;
|
||||
private volatile Action _readContinuation;
|
||||
private static partial class NativeMethods_Windows
|
||||
{
|
||||
private const string Kernel32 = "kernel32.dll";
|
||||
private const string KernelBase = "kernelbase.dll";
|
||||
public const IntPtr InvalidHandleValue = -1;
|
||||
|
||||
private volatile bool _writeAwaitBeginning;
|
||||
private volatile Action _writeContinuation;
|
||||
[LibraryImport(Kernel32, SetLastError = true, StringMarshalling = StringMarshalling.Utf16)]
|
||||
public static partial IntPtr CreateFileMappingW(
|
||||
IntPtr hFile, IntPtr lpFileMappingAttributes, uint flProtect, uint dwMaximumSizeHigh, uint dwMaximumSizeLow,
|
||||
string lpName
|
||||
);
|
||||
|
||||
#endregion
|
||||
[LibraryImport(KernelBase, SetLastError = true)]
|
||||
public static partial IntPtr MapViewOfFile3(
|
||||
IntPtr hFileMappingObject, IntPtr processHandle, IntPtr pvBaseAddress, ulong ullOffset, ulong ullSize,
|
||||
uint allocFlags, uint dwDesiredAccess,
|
||||
IntPtr hExtendedParameter, int parameterCount
|
||||
);
|
||||
|
||||
[LibraryImport(Kernel32, SetLastError = true)]
|
||||
[return: MarshalAs(UnmanagedType.Bool)]
|
||||
public static partial bool UnmapViewOfFile(IntPtr lpBaseAddress);
|
||||
|
||||
[LibraryImport(Kernel32, SetLastError = true)]
|
||||
[return: MarshalAs(UnmanagedType.Bool)]
|
||||
public static partial bool CloseHandle(IntPtr hObject);
|
||||
|
||||
[LibraryImport(KernelBase, SetLastError = true)]
|
||||
public static partial IntPtr VirtualAlloc2(
|
||||
IntPtr process,
|
||||
IntPtr address,
|
||||
ulong size,
|
||||
uint allocationType,
|
||||
uint protect,
|
||||
IntPtr extendedParameters,
|
||||
uint parameterCount
|
||||
);
|
||||
|
||||
[LibraryImport(Kernel32, SetLastError = true)]
|
||||
[return: MarshalAs(UnmanagedType.Bool)]
|
||||
public static partial bool VirtualFree(IntPtr lpAddress, uint dwSize, uint dwFreeType);
|
||||
|
||||
public const uint MEM_PRESERVE_PLACEHOLDER = 0x02;
|
||||
public const uint MEM_RESERVE = 0x2000;
|
||||
public const uint MEM_REPLACE_PLACEHOLDER = 0x4000;
|
||||
public const uint MEM_RELEASE = 0x8000;
|
||||
public const uint MEM_RESERVE_PLACEHOLDER = 0x40000;
|
||||
public const uint PAGE_NOACCESS = 0x01;
|
||||
public const uint PAGE_READWRITE = 0x04;
|
||||
}
|
||||
|
||||
private static partial class NativeMethods_Linux
|
||||
{
|
||||
private const string LibC = "libc";
|
||||
public const IntPtr InvalidPtrValue = -1;
|
||||
public const int InvalidFileDescriptor = -1;
|
||||
|
||||
// For MacOS
|
||||
[LibraryImport(LibC, SetLastError = true, StringMarshalling = StringMarshalling.Utf8)]
|
||||
public static partial int shm_open(string name, int oflag, int mode);
|
||||
|
||||
[LibraryImport(LibC, SetLastError = true, StringMarshalling = StringMarshalling.Utf8)]
|
||||
public static partial int shm_unlink(string name);
|
||||
|
||||
[LibraryImport(LibC, SetLastError = true, StringMarshalling = StringMarshalling.Utf8)]
|
||||
public static partial int memfd_create(string name, uint flags);
|
||||
|
||||
[LibraryImport(LibC, SetLastError = true)]
|
||||
public static partial int ftruncate(int fd, int length);
|
||||
|
||||
[LibraryImport(LibC, SetLastError = true)]
|
||||
public static partial int close(int fd);
|
||||
|
||||
[LibraryImport(LibC, SetLastError = true)]
|
||||
public static partial IntPtr mmap(IntPtr addr, ulong length, int prot, int flags, int fd, int offset);
|
||||
|
||||
[LibraryImport(LibC, SetLastError = true)]
|
||||
public static partial int munmap(IntPtr addr, ulong length);
|
||||
|
||||
public const int PROT_READ = 0x1;
|
||||
public const int PROT_WRITE = 0x2;
|
||||
public const int MAP_PRIVATE = 0x02;
|
||||
public const int MAP_SHARED = 0x01;
|
||||
public const int MAP_FIXED = 0x10;
|
||||
public const int MAP_ANONYMOUS = 0x20;
|
||||
|
||||
// Darwin
|
||||
public const int O_RDWR = 0x2;
|
||||
public const int O_CREAT = 0x200;
|
||||
public const int MAP_ANON = 0x1000;
|
||||
}
|
||||
|
||||
private void ReleaseUnmanagedResources()
|
||||
{
|
||||
if (_buffer == IntPtr.Zero)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
if (Core.IsWindows)
|
||||
{
|
||||
if (_handle != IntPtr.Zero)
|
||||
{
|
||||
NativeMethods_Windows.CloseHandle(_handle);
|
||||
_handle = IntPtr.Zero;
|
||||
}
|
||||
|
||||
if (_buffer != IntPtr.Zero)
|
||||
{
|
||||
NativeMethods_Windows.UnmapViewOfFile(_buffer);
|
||||
NativeMethods_Windows.UnmapViewOfFile(new IntPtr(_buffer + _bufferSize));
|
||||
}
|
||||
}
|
||||
else if (Core.IsLinux || Core.IsDarwin)
|
||||
{
|
||||
if (_handle != NativeMethods_Linux.InvalidFileDescriptor)
|
||||
{
|
||||
#pragma warning disable CA2020
|
||||
NativeMethods_Linux.close((int)_handle);
|
||||
#pragma warning restore CA2020
|
||||
|
||||
_handle = NativeMethods_Linux.InvalidFileDescriptor;
|
||||
}
|
||||
|
||||
if (_buffer != IntPtr.Zero)
|
||||
{
|
||||
NativeMethods_Linux.munmap(_buffer, _bufferSize);
|
||||
NativeMethods_Linux.munmap(new IntPtr(_buffer + _bufferSize), _bufferSize);
|
||||
}
|
||||
}
|
||||
|
||||
_buffer = IntPtr.Zero;
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
ReleaseUnmanagedResources();
|
||||
GC.SuppressFinalize(this);
|
||||
}
|
||||
|
||||
~Pipe()
|
||||
{
|
||||
ReleaseUnmanagedResources();
|
||||
}
|
||||
}
|
||||
|
||||
public class EndOfPipeException : IOException
|
||||
{
|
||||
public EndOfPipeException(string message) : base(message)
|
||||
{
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue