AuroraSupport/AuroraRuntime
2
1
Fork 0
Code Issues 27 Pull Requests Releases Wiki Activity

[*] QOL / Hardening / Optimizations / Bug fixes

Browse Source 
(but not really. this is just some patchwork)
This commit is contained in:
Reece Wilson 2024-10-02 00:24:34 +01:00
parent ebec613f66
commit 978693559e
27 changed files with 729 additions and 132 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Include/Aurora/IO/Net/INetSrvDatagram.hpp
View File

@ -15,8 +15,8 @@ namespace Aurora::IO::Net
struct NetDatagramBind struct NetDatagramBind
{ {
IPAddress ip; IPAddress ip;
AuUInt16 uPort; AuUInt16 uPort {};
AuUInt32 uDefaultInputStreamSize; AuUInt32 uDefaultInputStreamSize {};
AuSPtr<IDatagramDriver> pDriver; AuSPtr<IDatagramDriver> pDriver;
}; };

36
Include/Aurora/IO/Net/ISocketChannel.hpp
View File

@ -208,6 +208,42 @@ namespace Aurora::IO::Net
*/ */
virtual AuUInt GetOutputBufferSize() = 0; virtual AuUInt GetOutputBufferSize() = 0;
/**
* Recommended: 0, consume everything within the constraints of GetInputBufferSize/SpecifyInputBufferSize.
* use SingleFrame stack pieces to recursively tick a stack of protocol handlers.
* use DynamicBuffer protocol stack pieces to dynamically scale the output size.
*
* When not zero, this is the packet length per io processor frame.
* The io read is capped to segment of this size (uPageSize out of GetInputBufferSize/SpecifyInputBufferSize).
* In theory it is possible to boost max connection throughput (fair bandwidth between higher socket count) at the cost of lower bandwidth, increased tickrate, and higher cpu usage.
*
* Let say you can afford to buffer 10 frames.
* Should an aggressively written application dequeue of all of that at once, you would need to:
* >allocate x10 the amount of memory in a single io tick,
* >need to worry about how much cpu work time a single request or batch of requests take,
* >peak memory usage requirements for decompression and encryption handlers (you need x10 the peak memory usage).
* uPageSize can retard connections down into keeping their pending to-read memory elsewhere in the network stack.
*
* if you know the MTU is 32k, SpecifyPageLength(32k) for 1:1 op/tick.
* if you know the target is one tcp frame tick, SpecifyPageLength(64k) for 1:1 op/tick.
* udp may need splitting up across 576byte frames.
* you may want to bulk read multiple frames from kernel allocated network packets, over hammering tickrate.
* on the otherhand, you may want to limit how many segments you dequeue in a single io processor tick and ::read/ReadFromEx/ReadFileEx/io_submit([pread])(uNBytes) to a multiplier of your max expected packet size.
*/
virtual void SpecifyPageLength(AuUInt uPageSize) = 0;
/**
* @brief
* @return
*/
virtual AuUInt GetPageLength() = 0;
//
// BREAK! Ending preestablishment bias
// BREAK! Ending preestablishment bias
//
/** /**
* @brief * @brief
* @return * @return

55
Include/Aurora/Memory/ByteBuffer.hpp
View File

@ -88,6 +88,7 @@ namespace Aurora::Memory
AuUInt8 flagNoRealloc : 1; /// Prevents a subset of free options, specifically realloc, operations AuUInt8 flagNoRealloc : 1; /// Prevents a subset of free options, specifically realloc, operations
AuUInt8 flagAlwaysExpandable : 1; /// Internal flag. Do not use. AuUInt8 flagAlwaysExpandable : 1; /// Internal flag. Do not use.
AuUInt8 flagReserveA : 1; /// Placeholder AuUInt8 flagReserveA : 1; /// Placeholder
AuUInt8 flagReservedB;
/////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////
// Special flags/values // Special flags/values
@ -465,18 +466,7 @@ namespace Aurora::Memory
{ {
AuSPtr<void> memory; AuSPtr<void> memory;
inline SharedByteBuffer(AuSPtr<MemoryViewWrite> pWriteView) : ByteBuffer() inline SharedByteBuffer(const MemoryViewWrite &view, AuSPtr<void> pRAIIParentOwner) :
{
this->allocSize = 0;
this->base = (AuUInt8 *)pWriteView->ptr;
this->length = pWriteView->length;
this->readPtr = this->base;
this->writePtr = this->base + this->length;
this->flagNoFree = true;
this->memory = pWriteView;
}
inline SharedByteBuffer(AuSPtr<void> pRAIIParentOwner, MemoryViewWrite view) :
ByteBuffer(), ByteBuffer(),
__view(view) __view(view)
{ {
@ -484,11 +474,50 @@ namespace Aurora::Memory
this->base = (AuUInt8 *)view.ptr; this->base = (AuUInt8 *)view.ptr;
this->length = view.length; this->length = view.length;
this->readPtr = this->base; this->readPtr = this->base;
this->writePtr = this->base + this->length; this->writePtr = this->base;
this->flagNoFree = true; this->flagNoFree = true;
this->memory = pRAIIParentOwner; this->memory = pRAIIParentOwner;
} }
inline SharedByteBuffer(const MemoryViewWrite &view) :
ByteBuffer(),
__view(view)
{
this->allocSize = 0;
this->base = (AuUInt8 *)view.ptr;
this->length = view.length;
this->readPtr = this->base;
this->writePtr = this->base;
this->flagNoFree = true;
}
inline SharedByteBuffer(const MemoryViewRead &view, AuSPtr<void> pRAIIParentOwner) :
ByteBuffer(),
__view(*(MemoryViewWrite *)&view)
{
this->allocSize = 0;
this->base = (AuUInt8 *)view.ptr;
this->length = view.length;
this->readPtr = this->base;
this->writePtr = this->base + this->length;
this->flagNoFree = true;
this->flagWriteError = true;
this->memory = pRAIIParentOwner;
}
inline SharedByteBuffer(const MemoryViewRead &view) :
ByteBuffer(),
__view(*(MemoryViewWrite *)&view)
{
this->allocSize = 0;
this->base = (AuUInt8 *)view.ptr;
this->length = view.length;
this->readPtr = this->base;
this->writePtr = this->base + this->length;
this->flagNoFree = true;
this->flagWriteError = true;
}
private: private:
MemoryViewWrite __view; MemoryViewWrite __view;
}; };

2
Include/Aurora/Memory/ByteBuffer_Memory.inl
View File

@ -274,7 +274,7 @@ namespace Aurora::Memory
return true; return true;
} }
auto newLength = AuMax(length, AuPageRoundUp(this->allocSize + (this->allocSize / 3), AuUInt(64))); auto newLength = AuMax(length, AuPageRoundUp(this->allocSize + (this->allocSize / 3), AuUInt(128)));
AuUInt8 *pNext {}; AuUInt8 *pNext {};
if (this->alignment) if (this->alignment)

23
Include/Aurora/Memory/ByteBuffer_Utils.inl
View File

@ -197,7 +197,7 @@ namespace Aurora::Memory
ByteBuffer::ByteBuffer(ByteBuffer &&buffer) : ByteBuffer::ByteBuffer(ByteBuffer &&buffer) :
flagCircular {}, flagExpandable {}, flagReadError {}, flagWriteError {}, flagCircular {}, flagExpandable {}, flagReadError {}, flagWriteError {},
flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {} flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA{}, flagReservedB {}
{ {
this->base = buffer.base; this->base = buffer.base;
this->length = buffer.length; this->length = buffer.length;
@ -210,6 +210,7 @@ namespace Aurora::Memory
this->flagReadError = buffer.flagReadError; this->flagReadError = buffer.flagReadError;
this->flagWriteError = buffer.flagWriteError; this->flagWriteError = buffer.flagWriteError;
this->alignment = buffer.alignment; this->alignment = buffer.alignment;
this->flagReservedB = buffer.flagReservedB;
buffer.base = {}; buffer.base = {};
buffer.length = {}; buffer.length = {};
buffer.allocSize = {}; buffer.allocSize = {};
@ -223,7 +224,7 @@ namespace Aurora::Memory
} }
ByteBuffer::ByteBuffer(const ByteBuffer &buffer, bool preservePointers) : ByteBuffer::ByteBuffer(const ByteBuffer &buffer, bool preservePointers) :
flagCircular {}, flagExpandable { }, flagReadError {}, flagWriteError {}, flagCircular {}, flagExpandable { }, flagReadError {}, flagWriteError {}, flagReservedB {},
flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {} flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {}
{ {
if (buffer.length) if (buffer.length)
@ -262,7 +263,7 @@ namespace Aurora::Memory
} }
ByteBuffer::ByteBuffer(const void *in, AuUInt length, bool circular, bool expandable) : ByteBuffer::ByteBuffer(const void *in, AuUInt length, bool circular, bool expandable) :
flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {}, flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {}, flagReservedB {},
flagCircular(circular), flagExpandable(expandable), flagReadError(0), flagWriteError(0) flagCircular(circular), flagExpandable(expandable), flagReadError(0), flagWriteError(0)
{ {
this->scaleSize = kBufferInitialPower; this->scaleSize = kBufferInitialPower;
@ -285,7 +286,7 @@ namespace Aurora::Memory
} }
ByteBuffer::ByteBuffer(const MemoryViewRead &readView, bool circular, bool expandable) : ByteBuffer::ByteBuffer(const MemoryViewRead &readView, bool circular, bool expandable) :
flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {}, flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {}, flagReservedB {},
flagCircular(circular), flagExpandable(expandable), flagReadError(0), flagWriteError(0) flagCircular(circular), flagExpandable(expandable), flagReadError(0), flagWriteError(0)
{ {
this->scaleSize = kBufferInitialPower; this->scaleSize = kBufferInitialPower;
@ -303,7 +304,7 @@ namespace Aurora::Memory
} }
ByteBuffer::ByteBuffer(const AuList<AuUInt8> &vector, bool circular, bool expandable) : ByteBuffer::ByteBuffer(const AuList<AuUInt8> &vector, bool circular, bool expandable) :
flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {}, flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {}, flagReservedB {},
flagCircular(circular), flagExpandable(expandable), flagReadError(0), flagWriteError(0) flagCircular(circular), flagExpandable(expandable), flagReadError(0), flagWriteError(0)
{ {
this->scaleSize = kBufferInitialPower; this->scaleSize = kBufferInitialPower;
@ -321,7 +322,7 @@ namespace Aurora::Memory
} }
ByteBuffer::ByteBuffer(const MemoryViewRead &readView, AuUInt uAlignment, bool circular, bool expandable) : ByteBuffer::ByteBuffer(const MemoryViewRead &readView, AuUInt uAlignment, bool circular, bool expandable) :
flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {}, flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {}, flagReservedB {},
flagCircular(circular), flagExpandable(expandable), flagReadError(0), flagWriteError(0) flagCircular(circular), flagExpandable(expandable), flagReadError(0), flagWriteError(0)
{ {
this->scaleSize = kBufferInitialPower; this->scaleSize = kBufferInitialPower;
@ -344,7 +345,7 @@ namespace Aurora::Memory
} }
ByteBuffer::ByteBuffer(AuUInt length, bool circular, bool expandable) : ByteBuffer::ByteBuffer(AuUInt length, bool circular, bool expandable) :
flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {}, flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {}, flagReservedB {},
flagCircular(circular), flagExpandable(expandable), flagReadError(0), flagWriteError(0) flagCircular(circular), flagExpandable(expandable), flagReadError(0), flagWriteError(0)
{ {
this->scaleSize = kBufferInitialPower; this->scaleSize = kBufferInitialPower;
@ -366,7 +367,7 @@ namespace Aurora::Memory
} }
ByteBuffer::ByteBuffer(AuUInt length, AuUInt alignment, bool circular, bool expandable) : ByteBuffer::ByteBuffer(AuUInt length, AuUInt alignment, bool circular, bool expandable) :
flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {}, flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {}, flagReservedB {},
flagCircular(circular), flagExpandable(expandable), flagReadError(0), flagWriteError(0) flagCircular(circular), flagExpandable(expandable), flagReadError(0), flagWriteError(0)
{ {
if (!length) if (!length)
@ -390,7 +391,7 @@ namespace Aurora::Memory
template<typename T> template<typename T>
ByteBuffer::ByteBuffer(T *base, T *end, bool circular, bool expandable) : ByteBuffer::ByteBuffer(T *base, T *end, bool circular, bool expandable) :
flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {}, flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {}, flagReservedB {},
flagCircular(circular), flagExpandable(expandable), flagReadError(0), flagWriteError(0) flagCircular(circular), flagExpandable(expandable), flagReadError(0), flagWriteError(0)
{ {
if (!base) if (!base)
@ -419,7 +420,7 @@ namespace Aurora::Memory
} }
ByteBuffer::ByteBuffer() : ByteBuffer::ByteBuffer() :
flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {}, flagNoFree {}, flagNoRealloc {}, flagAlwaysExpandable {}, flagReserveA {}, flagReservedB {},
flagCircular(0), flagExpandable(true), flagReadError(0), flagWriteError(0) flagCircular(0), flagExpandable(true), flagReadError(0), flagWriteError(0)
{ {
this->base = {}; this->base = {};
@ -444,6 +445,7 @@ namespace Aurora::Memory
this->flagNoRealloc = other.flagNoRealloc; this->flagNoRealloc = other.flagNoRealloc;
this->flagNoFree = other.flagNoFree; this->flagNoFree = other.flagNoFree;
this->scaleSize = other.scaleSize; this->scaleSize = other.scaleSize;
this->flagReservedB = other.flagReservedB;
other.base = {}; other.base = {};
other.length = {}; other.length = {};
other.allocSize = {}; other.allocSize = {};
@ -483,6 +485,7 @@ namespace Aurora::Memory
this->flagCircular = buffer.flagCircular; this->flagCircular = buffer.flagCircular;
this->flagExpandable = buffer.flagExpandable; this->flagExpandable = buffer.flagExpandable;
this->flagAlwaysExpandable = buffer.flagAlwaysExpandable; this->flagAlwaysExpandable = buffer.flagAlwaysExpandable;
this->flagReservedB = buffer.flagReservedB;
return *this; return *this;
} }

5
Include/Aurora/RuntimeConfig.hpp
View File

@ -277,7 +277,12 @@ namespace Aurora
struct IOConfig struct IOConfig
{ {
AuUInt32 uProtocolStackDefaultBufferSize { 64 * 1024 }; AuUInt32 uProtocolStackDefaultBufferSize { 64 * 1024 };
#if defined(AURORA_IS_SERVER)
bool bIsVeryLargeIOApplication { true };
#else
bool bIsVeryLargeIOApplication { false }; bool bIsVeryLargeIOApplication { false };
#endif
// On Win32, NewLSTimer has very bad resolution. // On Win32, NewLSTimer has very bad resolution.
// On linux, timerfd is good enough. // On linux, timerfd is good enough.

46
Source/Compression/AuCompression.hpp
View File

@ -9,7 +9,51 @@
namespace Aurora::Compression namespace Aurora::Compression
{ {
static const AuUInt64 kChunkSize = 128 * 1024; #if defined(AURORA_IS_SERVER) || defined(AURORA_16_KIB_IN_OUT_BUFFERS_32K)
// servers can have a multiple of these per client, if streams are not to be recycled.
// XMPP, for example, does not reset. zlib and others expect genuine compression streams.
// HTTP, libdeflate, and others expect flush framesm and then for the stateless stream to be recreated each IO tick.
// let's assume we are a server;
// most of our IO is done under protocol stacks that'll handle almost all of the buffering,
// changing this value by 2-8 will apply to the law of large numbers on a server,
// changing this value by an order of magnitude is a cost of 10x memory per client (tho you can afford the memory these days),
// the underlying compressor object may not even be accessing these temp[kChunkSize] optimizations - some can directly read from AuByteBuffers,
// some compression libraries are internally alloc heavy, which delgates the usage area to malloc/free callbacks...
// ...these, like zstd, demand almost zero buffered data. a frame fragment is almost nothing.
// therefore, it makes sense to keep these values down as much as possible
static const AuUInt64 kChunkSize = 16 * 1024;
#elif defined(AURORA_IS_CLIENT)
// for builds explicitly earmarked to go out to end users, 2x the usual buffer size
static const AuUInt64 kChunkSize = 256 * 1024;
#else
// and general purpose.
static const AuUInt64 kChunkSize = 128 * 1024;
// also note: zlib docs state up to 256k is efficient, but you can start at around 16k
// https://www.zlib.net/zlib_how.html Last modified 8 February 2023. approx 30 Oct 2004 - 11 Dec 2005.
// general infrastructure and requirements havent changed in 2 decades.
// in 2001 and 2021 you can afford to buy a server with a few tens of gigabytes of ram
// in 2001 and 2021 you can afford to buy a gigabit switch
// in 2001 and 2021 you're still running on deflate algorithms
// in 2004 and 2024 you're still struggling to parallelize compression without N x tar balls on N x cores (N ~= 2-6 would be reasonable)
// in 2012-2014 you could get yourself an ivybridge or a sandbridge, in 2024 you could get yourself an AMD page aligned atomic acker or Oxide Inside 14th gen.
// with ~4th gen being, in theory, multiple single digits slower than 14th gen. and, yes, comparing 4th gen to 9th gen compilation times will result in 4-8x boosts,
// on a very fundamental level, we havent moved that much in terms of single threaded performance.
// that also applies to compression. it's stil the same old slow task calculating bit patterns, looking through big precomputed (or not) datasets, and making in-order & non-parallelizable changes
// With hardware not progressing **that** much, the fundamental algorithms staying the exact same, and the same libraries being used, call me a bit insane to think maybe an article from 2005 isnt that far off from reality.
// 16k*128k (clients) ~= 2gb (x2 for local in/out? [= 4GiB] x4 for additional in/out stream pairing? [= 8GiB]) of ram for a heavyweight server (plus many other user implementation overheads)
// 16k per stream / 256MB on an embedded tablet or games console
// 16k per stream / 256 - 2048MB per app / 1024 - 4096 per home appliance mid 2000s (laptop, pc, etc)
// 16k per stream / 256 - 2048MB per app / 4096 - 32768 per home appliance mid 2020s (laptop, pc, etc)
// 64k (2^16) IPv4 UDP ports, including system region and others
// 2x IPv6 address = 128k
// AAA software can see peaks of hundreds of thousands to millions per clients per platform, so x2, x10, x20, x100 this (more responsive stateful, proprietary, xmpp, etc protocols)
// WebShid "technology" works on stateless packets, so your memory requirements are basically just this in the network stream buffers, and with the rest being temporary allocations (more expensive & wasteful for stream-like connections).
// Seems fair...? seems like the world we should be still living in...? maybe we can 10x or 100x it no matter the era for increased throughput...? would zlib even care...?
#endif
// sometimes on stack
static const AuUInt64 kChunkSize2 = 128 * 1024;
/// * compression type + bits -> internal zlib windowBits /// * compression type + bits -> internal zlib windowBits
static bool CompressionLevelFromExternalApi(const DecompressInfo &info, AuInt8 &out) static bool CompressionLevelFromExternalApi(const DecompressInfo &info, AuInt8 &out)

32
Source/Compression/StreamCompression.cpp
View File

@ -256,8 +256,8 @@ namespace Aurora::Compression
{ {
int ret, flush; int ret, flush;
z_stream strm {}; z_stream strm {};
unsigned char in[kChunkSize]; unsigned char in[kChunkSize2];
unsigned char out[kChunkSize]; unsigned char out[kChunkSize2];
if (!stream.pWritePipe) if (!stream.pWritePipe)
{ {
@ -348,8 +348,8 @@ namespace Aurora::Compression
{ {
int ret; int ret;
z_stream strm {}; z_stream strm {};
unsigned char in[kChunkSize]; unsigned char in[kChunkSize2];
unsigned char out[kChunkSize]; unsigned char out[kChunkSize2];
if (!stream.pWritePipe) if (!stream.pWritePipe)
{ {
@ -437,8 +437,8 @@ namespace Aurora::Compression
#if defined(_AUHAS_BZIP2) #if defined(_AUHAS_BZIP2)
int ret, flush; int ret, flush;
bz_stream strm {}; bz_stream strm {};
char in[kChunkSize]; char in[kChunkSize2];
char out[kChunkSize]; char out[kChunkSize2];
if (!stream.pWritePipe) if (!stream.pWritePipe)
{ {
@ -462,7 +462,7 @@ namespace Aurora::Compression
do do
{ {
AuIO::EStreamError error; AuIO::EStreamError error;
AuUInt read{ kChunkSize }; AuUInt read{ kChunkSize2 };
if (((error = stream.pReadPipe->Read(AuMemory::MemoryViewStreamWrite(in, read))) != AuIO::EStreamError::eErrorNone) && if (((error = stream.pReadPipe->Read(AuMemory::MemoryViewStreamWrite(in, read))) != AuIO::EStreamError::eErrorNone) &&
(error != AuIO::EStreamError::eErrorEndOfStream)) (error != AuIO::EStreamError::eErrorEndOfStream))
{ {
@ -528,8 +528,8 @@ namespace Aurora::Compression
#if defined(_AUHAS_BZIP2) #if defined(_AUHAS_BZIP2)
int ret; int ret;
bz_stream strm {}; bz_stream strm {};
char in[kChunkSize]; char in[kChunkSize2];
char out[kChunkSize]; char out[kChunkSize2];
if (!stream.pWritePipe) if (!stream.pWritePipe)
{ {
@ -832,7 +832,7 @@ namespace Aurora::Compression
} }
DecompressInfo meta3 = meta2; DecompressInfo meta3 = meta2;
meta3.uInternalStreamSize = kChunkSize * 10; // stupid, but required for BaseStream::Write to not blow up. not optimizing old apis. meta3.uInternalStreamSize = kChunkSize2 * 10; // stupid, but required for BaseStream::Write to not blow up. not optimizing old apis.
auto pCompressor = DecompressorUnique(stream.pReadPipe, meta3); auto pCompressor = DecompressorUnique(stream.pReadPipe, meta3);
if (!pCompressor) if (!pCompressor)
{ {
@ -840,13 +840,13 @@ namespace Aurora::Compression
return false; return false;
} }
AuByteBuffer buffer(kChunkSize * 2, true); AuByteBuffer buffer(kChunkSize2 * 2, true);
AuStreamReadWrittenPair_t pair; AuStreamReadWrittenPair_t pair;
do do
{ {
auto view = buffer.GetNextLinearRead(); auto view = buffer.GetNextLinearRead();
view.length = AuMin<AuUInt>(view.length, kChunkSize / 2); view.length = AuMin<AuUInt>(view.length, kChunkSize2 / 2);
auto pair = pCompressor->ReadEx(AuMemory::MemoryViewWrite(buffer), true); auto pair = pCompressor->ReadEx(AuMemory::MemoryViewWrite(buffer), true);
while (pair.first || pair.second) while (pair.first || pair.second)
@ -914,7 +914,7 @@ namespace Aurora::Compression
} }
CompressInfo meta3 = meta2; CompressInfo meta3 = meta2;
meta3.uInternalStreamSize = kChunkSize / 2; meta3.uInternalStreamSize = kChunkSize2 / 2;
auto pCompressor = CompressorUnique(stream.pReadPipe, meta3); auto pCompressor = CompressorUnique(stream.pReadPipe, meta3);
if (!pCompressor) if (!pCompressor)
{ {
@ -922,13 +922,13 @@ namespace Aurora::Compression
return false; return false;
} }
AuByteBuffer buffer(kChunkSize / 2, true); AuByteBuffer buffer(kChunkSize2 / 2, true);
AuStreamReadWrittenPair_t pair; AuStreamReadWrittenPair_t pair;
do do
{ {
auto view = buffer.GetNextLinearRead(); auto view = buffer.GetNextLinearRead();
view.length = AuMin(view.length, kChunkSize / 2); view.length = AuMin(view.length, kChunkSize2 / 2);
auto pair = pCompressor->ReadEx(AuMemory::MemoryViewWrite(buffer), true); auto pair = pCompressor->ReadEx(AuMemory::MemoryViewWrite(buffer), true);
while (pair.first || pair.second) while (pair.first || pair.second)
@ -968,7 +968,7 @@ namespace Aurora::Compression
} }
auto view = buffer.GetNextLinearRead(); auto view = buffer.GetNextLinearRead();
view.length = AuMin<AuUInt>(view.length, kChunkSize / 2); view.length = AuMin<AuUInt>(view.length, kChunkSize2 / 2);
auto pair = pCompressor->ReadEx(AuMemory::MemoryViewWrite(buffer), false); auto pair = pCompressor->ReadEx(AuMemory::MemoryViewWrite(buffer), false);
while (pair.first || pair.second) while (pair.first || pair.second)

6
Source/Crypto/ECC/ECCX25519Private.cpp
View File

@ -12,14 +12,16 @@
namespace Aurora::Crypto::ECC namespace Aurora::Crypto::ECC
{ {
PrivateCurve25519Impl::PrivateCurve25519Impl(bool isX25519, curve25519_key &&key) : bIsX25519_(isX25519), key_(key) PrivateCurve25519Impl::PrivateCurve25519Impl(bool isX25519, curve25519_key &&key) :
bIsX25519_(isX25519),
key_(key)
{ {
} }
PrivateCurve25519Impl::~PrivateCurve25519Impl() PrivateCurve25519Impl::~PrivateCurve25519Impl()
{ {
AuMemset(&this->key_, 'N', AuSizeOf(this->key_));
} }
bool PrivateCurve25519Impl::Sign(const Memory::MemoryViewRead &plainText, bool PrivateCurve25519Impl::Sign(const Memory::MemoryViewRead &plainText,

1
Source/Crypto/ECC/ECCX25519Public.cpp
View File

@ -18,6 +18,7 @@ namespace Aurora::Crypto::ECC
PublicCurve25519Impl::~PublicCurve25519Impl() PublicCurve25519Impl::~PublicCurve25519Impl()
{ {
AuMemset(&this->key_, 'N', AuSizeOf(this->key_));
} }
bool PublicCurve25519Impl::Verify(const Memory::MemoryViewRead &hash, bool PublicCurve25519Impl::Verify(const Memory::MemoryViewRead &hash,

5
Source/Crypto/ECC/PrivateECCImpl.cpp
View File

@ -17,7 +17,9 @@
namespace Aurora::Crypto::ECC namespace Aurora::Crypto::ECC
{ {
PrivateECCImpl::PrivateECCImpl(EECCCurve type, ecc_key &key) : _key(key), _type(type) PrivateECCImpl::PrivateECCImpl(EECCCurve type, ecc_key &key) :
_key(key),
_type(type)
{ {
} }
@ -25,6 +27,7 @@ namespace Aurora::Crypto::ECC
PrivateECCImpl::~PrivateECCImpl() PrivateECCImpl::~PrivateECCImpl()
{ {
ecc_free(&this->_key); ecc_free(&this->_key);
AuMemset(&this->_key, 'N', AuSizeOf(this->_key));
} }
EECCCurve PrivateECCImpl::GetType() EECCCurve PrivateECCImpl::GetType()

3
Source/Crypto/ECC/PublicECCImpl.cpp
View File

@ -23,7 +23,8 @@ namespace Aurora::Crypto::ECC
PublicECCImpl::~PublicECCImpl() PublicECCImpl::~PublicECCImpl()
{ {
ecc_free(&_key); ecc_free(&this->_key);
AuMemset(&this->_key, 'N', AuSizeOf(this->_key));
} }
EECCCurve PublicECCImpl::GetType() EECCCurve PublicECCImpl::GetType()

4
Source/Crypto/RSA/RSAPrivate.cpp
View File

@ -13,7 +13,8 @@
namespace Aurora::Crypto::RSA namespace Aurora::Crypto::RSA
{ {
PrivateRSA::PrivateRSA(rsa_key &key) : key_(key) PrivateRSA::PrivateRSA(const rsa_key &key) :
key_(key)
{ {
} }
@ -21,6 +22,7 @@ namespace Aurora::Crypto::RSA
PrivateRSA::~PrivateRSA() PrivateRSA::~PrivateRSA()
{ {
::rsa_free(&this->key_); ::rsa_free(&this->key_);
AuMemset(&this->key_, 'N', AuSizeOf(this->key_));
} }
bool PrivateRSA::Sign(const Memory::MemoryViewRead & payload, bool PrivateRSA::Sign(const Memory::MemoryViewRead & payload,

2
Source/Crypto/RSA/RSAPrivate.hpp
View File

@ -11,7 +11,7 @@ namespace Aurora::Crypto::RSA
{ {
struct PrivateRSA : IRSAPrivate struct PrivateRSA : IRSAPrivate
{ {
PrivateRSA(rsa_key &key); PrivateRSA(const rsa_key &key);
~PrivateRSA(); ~PrivateRSA();
bool Sign(const Memory::MemoryViewRead & payload, bool Sign(const Memory::MemoryViewRead & payload,

4
Source/Crypto/RSA/RSAPublic.cpp
View File

@ -12,7 +12,8 @@
namespace Aurora::Crypto::RSA namespace Aurora::Crypto::RSA
{ {
PublicRSA::PublicRSA(rsa_key &key) : key_(key) PublicRSA::PublicRSA(const rsa_key &key) :
key_(key)
{ {
} }
@ -20,6 +21,7 @@ namespace Aurora::Crypto::RSA
PublicRSA::~PublicRSA() PublicRSA::~PublicRSA()
{ {
::rsa_free(&this->key_); ::rsa_free(&this->key_);
AuMemset(&this->key_, 'N', AuSizeOf(this->key_));
} }
bool PublicRSA::Verify(const AuMemoryViewRead &payload, bool PublicRSA::Verify(const AuMemoryViewRead &payload,

3
Source/Crypto/RSA/RSAPublic.hpp
View File

@ -12,7 +12,7 @@ namespace Aurora::Crypto::RSA
{ {
struct PublicRSA : IRSAPublic struct PublicRSA : IRSAPublic
{ {
PublicRSA(rsa_key &key); PublicRSA(const rsa_key &key);
~PublicRSA(); ~PublicRSA();
bool Verify(const Memory::MemoryViewRead & payload, bool Verify(const Memory::MemoryViewRead & payload,
@ -33,6 +33,5 @@ namespace Aurora::Crypto::RSA
private: private:
rsa_key key_; rsa_key key_;
bool bOwned_;
}; };
} }

2
Source/IO/FS/FSPlatformDevices.NT.cpp
View File

@ -338,7 +338,7 @@ namespace Aurora::IO::FS
} }
} }
AUKN_SYM AuUInt64 GetDeviceSizeInBytes(const AuString &physicalDevicePath) AUKN_SYM AuUInt64 GetDeviceSizeInBytes(const AuROString &physicalDevicePath)
{ {
DISK_GEOMETRY_EX geo {}; DISK_GEOMETRY_EX geo {};
DWORD cbBytesReturned; DWORD cbBytesReturned;

79
Source/IO/Net/AuNetSocketChannel.cpp
View File

@ -271,6 +271,16 @@ namespace Aurora::IO::Net
return pProtocol; return pProtocol;
} }
void SocketChannel::SpecifyPageLength(AuUInt uPageSize)
{
this->uBytesPerFrame = uPageSize;
}
AuUInt SocketChannel::GetPageLength()
{
return this->uBytesPerFrame;
}
void SocketChannel::SpecifyRecvProtocol(const AuSPtr<Protocol::IProtocolStack> &pRecvProtocol) void SocketChannel::SpecifyRecvProtocol(const AuSPtr<Protocol::IProtocolStack> &pRecvProtocol)
{ {
this->pRecvProtocol = pRecvProtocol; this->pRecvProtocol = pRecvProtocol;
@ -629,7 +639,14 @@ namespace Aurora::IO::Net
if (pCallbackOptional) if (pCallbackOptional)
{ {
pCallbackOptional->OnSuccess((void *)nullptr); try
{
pCallbackOptional->OnSuccess((void *)nullptr);
}
catch (...)
{
SysPushErrorCatch();
}
} }
return; return;
@ -651,10 +668,14 @@ namespace Aurora::IO::Net
if (bHasCurrentSuccess) if (bHasCurrentSuccess)
{ {
if (pCallbackOptional) try
{ {
pCallbackOptional->OnSuccess((void *)nullptr); pCallbackOptional->OnSuccess((void *)nullptr);
} }
catch (...)
{
SysPushErrorCatch();
}
return; return;
} }
@ -668,7 +689,14 @@ namespace Aurora::IO::Net
{ {
if (auto pOutput = AuExchange(this->pRetargetOutput, {})) if (auto pOutput = AuExchange(this->pRetargetOutput, {}))
{ {
pOutput->OnFailure((void *)nullptr); try
{
pOutput->OnFailure((void *)nullptr);
}
catch (...)
{
SysPushErrorCatch();
}
if (this->pRetargetInput == pOutput) if (this->pRetargetInput == pOutput)
{ {
AuResetMember(this->pRetargetInput); AuResetMember(this->pRetargetInput);
@ -686,7 +714,14 @@ namespace Aurora::IO::Net
{ {
if (auto pInput = AuExchange(this->pRetargetInput, {})) if (auto pInput = AuExchange(this->pRetargetInput, {}))
{ {
pInput->OnFailure((void *)nullptr); try
{
pInput->OnFailure((void *)nullptr);
}
catch (...)
{
SysPushErrorCatch();
}
if (this->pRetargetOutput == pInput) if (this->pRetargetOutput == pInput)
{ {
AuResetMember(this->pRetargetOutput); AuResetMember(this->pRetargetOutput);
@ -715,7 +750,14 @@ namespace Aurora::IO::Net
if (auto pOutput = AuExchange(this->pRetargetOutput, {})) if (auto pOutput = AuExchange(this->pRetargetOutput, {}))
{ {
pOutput->OnFailure((void *)nullptr); try
{
pOutput->OnFailure((void *)nullptr);
}
catch (...)
{
SysPushErrorCatch();
}
if (this->pRetargetInput == pOutput) if (this->pRetargetInput == pOutput)
{ {
AuResetMember(this->pRetargetInput); AuResetMember(this->pRetargetInput);
@ -730,7 +772,14 @@ namespace Aurora::IO::Net
{ {
if (this->pRetargetInput != pOutput) if (this->pRetargetInput != pOutput)
{ {
pOutput->OnSuccess((void *)nullptr); try
{
pOutput->OnSuccess((void *)nullptr);
}
catch (...)
{
SysPushErrorCatch();
}
} }
} }
} }
@ -750,7 +799,14 @@ namespace Aurora::IO::Net
if (auto pInput = AuExchange(this->pRetargetInput, {})) if (auto pInput = AuExchange(this->pRetargetInput, {}))
{ {
pInput->OnFailure((void *)nullptr); try
{
pInput->OnFailure((void *)nullptr);
}
catch (...)
{
SysPushErrorCatch();
}
if (this->pRetargetOutput == pInput) if (this->pRetargetOutput == pInput)
{ {
@ -767,7 +823,14 @@ namespace Aurora::IO::Net
{ {
if (this->pRetargetOutput != pInput) if (this->pRetargetOutput != pInput)
{ {
pInput->OnSuccess((void *)nullptr); try
{
pInput->OnSuccess((void *)nullptr);
}
catch (...)
{
SysPushErrorCatch();
}
} }
} }
} }

3
Source/IO/Net/AuNetSocketChannel.hpp
View File

@ -69,6 +69,8 @@ namespace Aurora::IO::Net
AuUInt GetNextFrameTargetLength() override; AuUInt GetNextFrameTargetLength() override;
void SpecifyPageLength(AuUInt uPageSize) override;
AuUInt GetPageLength() override;
void SpecifyRecvProtocol(const AuSPtr<Protocol::IProtocolStack> &pRecvProtocol) override; void SpecifyRecvProtocol(const AuSPtr<Protocol::IProtocolStack> &pRecvProtocol) override;
@ -106,6 +108,7 @@ namespace Aurora::IO::Net
AuUInt uBytesToFlip { 0 }; AuUInt uBytesToFlip { 0 };
AuUInt uBytesInputBuffer { 0 }; AuUInt uBytesInputBuffer { 0 };
AuUInt uBytesOutputBuffer { 0 }; AuUInt uBytesOutputBuffer { 0 };
AuUInt uBytesPerFrame { 0 };
AuUInt uBytesInputBufferRetarget { 0 }; AuUInt uBytesInputBufferRetarget { 0 };
AuSPtr<AuAsync::PromiseCallback<AuNullS, AuNullS>> pRetargetInput; AuSPtr<AuAsync::PromiseCallback<AuNullS, AuNullS>> pRetargetInput;

9
Source/IO/Net/AuNetSocketChannelInput.cpp
View File

@ -86,6 +86,15 @@ namespace Aurora::IO::Net
req.bIsStream = true; req.bIsStream = true;
req.pListener = sharedThis; req.pListener = sharedThis;
req.uBufferLengthOrZero = AuStaticCast<SocketChannel>(this->pParent_->ToChannel())->uBytesInputBuffer; req.uBufferLengthOrZero = AuStaticCast<SocketChannel>(this->pParent_->ToChannel())->uBytesInputBuffer;
if (auto uFrame = AuStaticCast<SocketChannel>(this->pParent_->ToChannel())->uBytesPerFrame)
{
req.uPageLengthOrZero = uFrame;
req.bReadEntireAllocation = false;
}
else
{
req.bReadEntireAllocation = true;
}
this->pNetReader = this->pParent_->ToWorker()->ToProcessor()->ToPipeProcessor()->NewAIOPipe(req); this->pNetReader = this->pParent_->ToWorker()->ToProcessor()->ToPipeProcessor()->NewAIOPipe(req);
} }

25
Source/IO/Net/AuNetSocketChannelOutput.cpp
View File

@ -126,10 +126,12 @@ namespace Aurora::IO::Net
auto pBase = (AuUInt8 *)this->pOutSendPointer_; auto pBase = (AuUInt8 *)this->pOutSendPointer_;
auto uLen = (AuUInt8 *)this->outputBuffer_.base + this->outputBuffer_.length - pBase; auto uLen = (AuUInt8 *)this->outputBuffer_.base + this->outputBuffer_.length - pBase;
// do not lock!
auto mem = AuMemoryViewRead { AuMemoryViewRead { pBase, (AuUInt)uLen }, this->pParent_->SharedFromThis() }; auto mem = AuMemoryViewRead { AuMemoryViewRead { pBase, (AuUInt)uLen }, this->pParent_->SharedFromThis() };
this->pOutSendPointer_ = (AuUInt8 *)this->pOutSendPointer_ + uLen; this->pOutSendPointer_ = (AuUInt8 *)this->pOutSendPointer_ + uLen;
this->outputWriteQueue_.Push(mem); this->outputWriteQueue_.Push(mem);
AuAtomicAdd(&this->outputBuffer_.uInUseCounter, 1u);
} }
else else
@ -138,10 +140,12 @@ namespace Aurora::IO::Net
auto pWriteHead = (AuUInt8 *)this->outputBuffer_.writePtr; auto pWriteHead = (AuUInt8 *)this->outputBuffer_.writePtr;
auto uLen = pWriteHead - pBase; auto uLen = pWriteHead - pBase;
// do not lock!
auto mem = AuMemoryViewRead { AuMemoryViewRead { pBase, (AuUInt)uLen }, this->pParent_->SharedFromThis() }; auto mem = AuMemoryViewRead { AuMemoryViewRead { pBase, (AuUInt)uLen }, this->pParent_->SharedFromThis() };
this->pOutSendPointer_ = pWriteHead; this->pOutSendPointer_ = pWriteHead;
this->outputWriteQueue_.Push(mem); this->outputWriteQueue_.Push(mem);
AuAtomicAdd(&this->outputBuffer_.uInUseCounter, 1u);
} }
} }
} }
@ -149,9 +153,10 @@ namespace Aurora::IO::Net
void SocketChannelOutput::WriteTick() void SocketChannelOutput::WriteTick()
{ {
bool bShouldShutdown {}; bool bShouldShutdown {};
{ {
AU_LOCK_GUARD(this->lock_); AU_LOCK_GUARD(this->lock_);
bShutdownOnComplete = WriteTickLocked(); bShouldShutdown = WriteTickLocked();
} }
if (bShouldShutdown) if (bShouldShutdown)
@ -187,19 +192,16 @@ namespace Aurora::IO::Net
} }
} }
if (this->outputWriteQueue_.IsEmpty()) if (this->CanResize() &&
this->pParent_)
{ {
if (this->pParent_)// && this->pParent_->socketChannel_.DoReallocWriteTick();
///this->outputBuffer_.outputChannel.AsWritableByteBuffer()->GetNextLinearRead().length == 0)
{
this->pParent_->socketChannel_.DoReallocWriteTick();
}
} }
// do not forcefully flush preemptive hello packets until the socket has properly connected // do not forcefully flush preemptive hello packets until the socket has properly connected
if (!this->pParent_->bHasConnected_) if (!this->pParent_->bHasConnected_)
{ {
return true; return false;
} }
if (auto pFrameToSend = this->outputWriteQueue_.Dequeue()) if (auto pFrameToSend = this->outputWriteQueue_.Dequeue())
@ -222,6 +224,8 @@ namespace Aurora::IO::Net
this->pParent_->SendErrorBeginShutdown({}); this->pParent_->SendErrorBeginShutdown({});
return false; return false;
} }
AuAtomicAdd(&this->uCompleteCounter_, 1u);
} }
} }
else else
@ -256,11 +260,14 @@ namespace Aurora::IO::Net
bool SocketChannelOutput::CanResize() bool SocketChannelOutput::CanResize()
{ {
return this->outputWriteQueue_.IsEmpty(); return this->outputWriteQueue_.IsEmpty() && AuAtomicLoad(&this->uCompleteCounter_) == 0;
} }
void SocketChannelOutput::OnAsyncFileOpFinished(AuUInt64 offset, AuUInt32 length) void SocketChannelOutput::OnAsyncFileOpFinished(AuUInt64 offset, AuUInt32 length)
{ {
AuAtomicSub(&this->outputBuffer_.uInUseCounter, 1u);
AuAtomicSub(&this->uCompleteCounter_, 1u);
if (!this->pParent_) if (!this->pParent_)
{ {
return; return;

1
Source/IO/Net/AuNetSocketChannelOutput.hpp
View File

@ -45,6 +45,7 @@ namespace Aurora::IO::Net
void * pOutSendPointer_ {}; void * pOutSendPointer_ {};
AuByteBuffer outputBuffer_; AuByteBuffer outputBuffer_;
NetWriteQueue outputWriteQueue_; NetWriteQueue outputWriteQueue_;
AuUInt32 uCompleteCounter_ {};
AuThreadPrimitives::SpinLock lock_; AuThreadPrimitives::SpinLock lock_;
}; };
} }

8
Source/IO/Net/AuNetSrvSockets.cpp
View File

@ -143,6 +143,7 @@ namespace Aurora::IO::Net
AuSPtr<ISocketServer> NetSrvSockets::NewServer(const NetSocketBind &netBind) AuSPtr<ISocketServer> NetSrvSockets::NewServer(const NetSocketBind &netBind)
{ {
auto uMaxSockets = netBind.uMaxConnections ? netBind.uMaxConnections : 512; auto uMaxSockets = netBind.uMaxConnections ? netBind.uMaxConnections : 512;
auto uMaxDatagramSize = netBind.uDefaultInputStreamSize ? netBind.uDefaultInputStreamSize : kDefaultStreamSize;
auto pWorker = this->pParent_->TryScheduleEx(); auto pWorker = this->pParent_->TryScheduleEx();
if (!pWorker) if (!pWorker)
@ -162,7 +163,7 @@ namespace Aurora::IO::Net
AuSToMS<AuUInt32>(60), AuSToMS<AuUInt32>(60),
netBind.bMultiThreaded, netBind.bMultiThreaded,
false, false,
kDefaultStreamSize); uMaxDatagramSize);
if (!pSocket) if (!pSocket)
{ {
SysPushErrorNet("No Memory"); SysPushErrorNet("No Memory");
@ -235,7 +236,8 @@ namespace Aurora::IO::Net
AuSPtr<ISocketServer> NetSrvSockets::NewServerEx(const NetSocketBindEx &netBindEx) AuSPtr<ISocketServer> NetSrvSockets::NewServerEx(const NetSocketBindEx &netBindEx)
{ {
auto uMaxSockets = netBindEx.uMaxConnections ? netBindEx.uMaxConnections : 512; auto uMaxSockets = netBindEx.uMaxConnections ? netBindEx.uMaxConnections : 512;
auto uMaxDatagramSize = netBindEx.uDefaultInputStreamSize ? netBindEx.uDefaultInputStreamSize : kDefaultStreamSize;
auto pWorker = this->pParent_->TryScheduleEx(); auto pWorker = this->pParent_->TryScheduleEx();
if (!pWorker) if (!pWorker)
@ -255,7 +257,7 @@ namespace Aurora::IO::Net
AuSToMS<AuUInt32>(netBindEx.uUDPTimeoutMs), AuSToMS<AuUInt32>(netBindEx.uUDPTimeoutMs),
netBindEx.bMultiThreaded, netBindEx.bMultiThreaded,
false, false,
kDefaultStreamSize); uMaxDatagramSize);
if (!pSocket) if (!pSocket)
{ {
SysPushErrorNet("No Memory"); SysPushErrorNet("No Memory");

372
Source/IO/Net/SocketOverDatagram/AuNetDatagramSocketServerChannel.cpp
View File

@ -21,6 +21,15 @@ static const auto kDefaultBufferSize = 64 * 1024;
namespace Aurora::IO::Net namespace Aurora::IO::Net
{ {
static bool ForceResize(AuByteBuffer &buffer, AuUInt uLength)
{
auto old = buffer.flagNoRealloc;
buffer.flagNoRealloc = false;
bool bRet = buffer.Resize(uLength);
buffer.flagNoRealloc = old;
return bRet;
}
NetDatagramSocketServerChannel::NetDatagramSocketServerChannel(NetDatagramSocketServerSession *pParent) : NetDatagramSocketServerChannel::NetDatagramSocketServerChannel(NetDatagramSocketServerSession *pParent) :
pParent_(pParent), pParent_(pParent),
outputBuffer(kDefaultBufferSize) outputBuffer(kDefaultBufferSize)
@ -81,8 +90,10 @@ namespace Aurora::IO::Net
} }
} }
void NetDatagramSocketServerChannel::OnComplete() void NetDatagramSocketServerChannel::OnComplete(AuUInt32 uNBytes)
{ {
AuAtomicSub(&this->outputBuffer.uInUseCounter, 1u);
if (AuAtomicSub(&this->uFence_, 1u)) if (AuAtomicSub(&this->uFence_, 1u))
{ {
if (this->bRateLimitPerTick) if (this->bRateLimitPerTick)
@ -109,21 +120,36 @@ namespace Aurora::IO::Net
{ {
if (this->outputBuffer.Resize(this->uNextOutputLengthRealloc_)) if (this->outputBuffer.Resize(this->uNextOutputLengthRealloc_))
{ {
if (this->pCallbackOptional_) try
{ {
this->pCallbackOptional_->OnSuccess((void *)nullptr); if (this->pCallbackOptional_)
{
this->pCallbackOptional_->OnSuccess((void *)nullptr);
}
}
catch (...)
{
SysPushErrorCatch();
} }
} }
else else
{ {
if (this->pCallbackOptional_) try
{ {
this->pCallbackOptional_->OnFailure((void *)nullptr); if (this->pCallbackOptional_)
{
this->pCallbackOptional_->OnFailure((void *)nullptr);
}
}
catch (...)
{
SysPushErrorCatch();
} }
} }
AuResetMember(this->pCallbackOptional_); AuResetMember(this->pCallbackOptional_);
AuResetMember(this->uNextOutputLengthRealloc_); AuResetMember(this->uNextOutputLengthRealloc_);
AuResetMember(this->uNextInputLengthRealloc_);
} }
if (this->bRateLimitPerTick) if (this->bRateLimitPerTick)
@ -149,13 +175,9 @@ namespace Aurora::IO::Net
auto pWriteTransaction = this->ToWriteTransaction(); auto pWriteTransaction = this->ToWriteTransaction();
struct DatagramMemory : AuMemoryViewRead
{
AuSPtr<void> pHold;
};
if (auto uDelta = uOffset - uStartOffset) if (auto uDelta = uOffset - uStartOffset)
{ {
// do not lock!
auto mem = AuMemoryViewRead { AuMemoryViewRead { this->outputBuffer.readPtr, uDelta }, this->pParent_->SharedFromThis() }; auto mem = AuMemoryViewRead { AuMemoryViewRead { this->outputBuffer.readPtr, uDelta }, this->pParent_->SharedFromThis() };
this->sendStats_.AddBytes(uDelta); this->sendStats_.AddBytes(uDelta);
@ -171,12 +193,13 @@ namespace Aurora::IO::Net
pWriteTransaction->SetCallback(AuMakeSharedThrow<AuIO::IAsyncFinishedSubscriberFunctional>([=](AuUInt64 uOffset, AuUInt32 uLength) pWriteTransaction->SetCallback(AuMakeSharedThrow<AuIO::IAsyncFinishedSubscriberFunctional>([=](AuUInt64 uOffset, AuUInt32 uLength)
{ {
this->OnComplete(); this->OnComplete(uLength);
})); }));
if (pWriteTransaction->StartWrite(0, mem)) if (pWriteTransaction->StartWrite(0, mem))
{ {
AuAtomicAdd(&this->uFence_, 1u); AuAtomicAdd(&this->uFence_, 1u);
AuAtomicAdd(&this->outputBuffer.uInUseCounter, 1u);
} }
else else
{ {
@ -229,12 +252,77 @@ namespace Aurora::IO::Net
bool NetDatagramSocketServerChannel::SpecifyBufferSize(AuUInt uBytes, bool NetDatagramSocketServerChannel::SpecifyBufferSize(AuUInt uBytes,
const AuSPtr<AuAsync::PromiseCallback<AuNullS, AuNullS>> &pCallbackOptional) const AuSPtr<AuAsync::PromiseCallback<AuNullS, AuNullS>> &pCallbackOptional)
{ {
return {}; if (!this->pParent_->bIsReady)
{
this->uBytesOutputBuffer = uBytes;
this->uBytesInputBuffer = uBytes;
if (pCallbackOptional)
{
try
{
pCallbackOptional->OnSuccess((void *)nullptr);
}
catch (...)
{
SysPushErrorCatch();
}
}
return true;
}
auto pWorker = this->pParent_->pWorker;
if (!pWorker)
{
return false;
}
if (pWorker->IsOnThread())
{
this->DoBufferResizeOnThread(true,
true,
uBytes,
pCallbackOptional);
return true;
}
auto pThat = AuSPtr<NetDatagramSocketServerChannel>(this->pParent_->SharedFromThis(), this);
if (!pWorker->TryScheduleInternalTemplate<AuNullS>([=](const AuSPtr<AuAsync::PromiseCallback<AuNullS>> &info)
{
pThat->DoBufferResizeOnThread(true,
true,
uBytes,
pCallbackOptional);
}, AuSPtr<AuAsync::PromiseCallback<AuNullS, AuNullS>>{}))
{
return false;
}
return true;
} }
bool NetDatagramSocketServerChannel::SpecifyOutputBufferSize(AuUInt uBytes, bool NetDatagramSocketServerChannel::SpecifyOutputBufferSize(AuUInt uBytes,
const AuSPtr<AuAsync::PromiseCallback<AuNullS, AuNullS>> &pCallbackOptional) const AuSPtr<AuAsync::PromiseCallback<AuNullS, AuNullS>> &pCallbackOptional)
{ {
if (!this->pParent_->bIsReady)
{
this->uBytesOutputBuffer = uBytes;
if (pCallbackOptional)
{
try
{
pCallbackOptional->OnSuccess((void *)nullptr);
}
catch (...)
{
SysPushErrorCatch();
}
}
return true;
}
auto pWorker = this->pParent_->pWorker; auto pWorker = this->pParent_->pWorker;
if (!pWorker) if (!pWorker)
{ {
@ -255,9 +343,9 @@ namespace Aurora::IO::Net
if (!pWorker->TryScheduleInternalTemplate<AuNullS>([=](const AuSPtr<AuAsync::PromiseCallback<AuNullS>> &info) if (!pWorker->TryScheduleInternalTemplate<AuNullS>([=](const AuSPtr<AuAsync::PromiseCallback<AuNullS>> &info)
{ {
pThat->DoBufferResizeOnThread(false, pThat->DoBufferResizeOnThread(false,
true, true,
uBytes, uBytes,
pCallbackOptional); pCallbackOptional);
}, AuSPtr<AuAsync::PromiseCallback<AuNullS, AuNullS>>{})) }, AuSPtr<AuAsync::PromiseCallback<AuNullS, AuNullS>>{}))
{ {
return false; return false;
@ -269,7 +357,53 @@ namespace Aurora::IO::Net
bool NetDatagramSocketServerChannel::SpecifyInputBufferSize(AuUInt uBytes, bool NetDatagramSocketServerChannel::SpecifyInputBufferSize(AuUInt uBytes,
const AuSPtr<AuAsync::PromiseCallback<AuNullS, AuNullS>> &pCallbackOptional) const AuSPtr<AuAsync::PromiseCallback<AuNullS, AuNullS>> &pCallbackOptional)
{ {
return {}; if (!this->pParent_->bIsReady)
{
this->uBytesInputBuffer = uBytes;
if (pCallbackOptional)
{
try
{
pCallbackOptional->OnSuccess((void *)nullptr);
}
catch (...)
{
SysPushErrorCatch();
}
}
return true;
}
auto pWorker = this->pParent_->pWorker;
if (!pWorker)
{
return false;
}
if (pWorker->IsOnThread())
{
this->DoBufferResizeOnThread(true,
false,
uBytes,
pCallbackOptional);
return true;
}
auto pThat = AuSPtr<NetDatagramSocketServerChannel>(this->pParent_->SharedFromThis(), this);
if (!pWorker->TryScheduleInternalTemplate<AuNullS>([=](const AuSPtr<AuAsync::PromiseCallback<AuNullS>> &info)
{
pThat->DoBufferResizeOnThread(true,
false,
uBytes,
pCallbackOptional);
}, AuSPtr<AuAsync::PromiseCallback<AuNullS, AuNullS>>{}))
{
return false;
}
return true;
} }
bool NetDatagramSocketServerChannel::SpecifyManualWrite(bool bEnableDirectAIOWrite) bool NetDatagramSocketServerChannel::SpecifyManualWrite(bool bEnableDirectAIOWrite)
@ -292,6 +426,16 @@ namespace Aurora::IO::Net
return true /*technically not nodelay. if users care to ask for nagle, no, you're sending frames. */; return true /*technically not nodelay. if users care to ask for nagle, no, you're sending frames. */;
} }
void NetDatagramSocketServerChannel::SpecifyPageLength(AuUInt uPageSize)
{
}
AuUInt NetDatagramSocketServerChannel::GetPageLength()
{
return this->pParent_->ToParent()->uDefaultPacketSize;
}
AuUInt NetDatagramSocketServerChannel::GetInputBufferSize() AuUInt NetDatagramSocketServerChannel::GetInputBufferSize()
{ {
if (this->pBuffer) if (this->pBuffer)
@ -470,7 +614,10 @@ namespace Aurora::IO::Net
if (!this->bOwnsBuffer) if (!this->bOwnsBuffer)
{ {
AuSPtr<AuByteBuffer> pNewBuffer; AuSPtr<AuByteBuffer> pNewBuffer;
if (!((pNewBuffer = AuMakeShared<AuByteBuffer>(*pBuffer)) && (pNewBuffer->IsValid())))
pNewBuffer = AuMakeShared<AuByteBuffer>(*pBuffer);
if (!pNewBuffer || !pNewBuffer->IsValid())
{ {
SysPushErrorMemory(); SysPushErrorMemory();
this->pParent_->OnError(AuNet::ENetworkError::eResourceConstraint); this->pParent_->OnError(AuNet::ENetworkError::eResourceConstraint);
@ -492,6 +639,82 @@ namespace Aurora::IO::Net
this->pBuffer = pBuffer; this->pBuffer = pBuffer;
this->bOwnsBuffer = bOwn; this->bOwnsBuffer = bOwn;
} }
this->DoReallocRecvTick();
}
void NetDatagramSocketServerChannel::DoReallocRecvTick()
{
if (this->uNextInputLengthRealloc_)
{
bool bFailed {};
if (this->bOwnsBuffer && this->pBuffer)
{
if (!ForceResize(*this->pBuffer, this->uNextInputLengthRealloc_))
{
bFailed = true;
}
}
else
{
AuSPtr<AuByteBuffer> pNewBuffer;
if (this->pBuffer)
{
pNewBuffer = AuMakeShared<AuByteBuffer>(*this->pBuffer);
if (pNewBuffer && pNewBuffer->IsValid() && this->uNextInputLengthRealloc_ > pNewBuffer->size())
{
pNewBuffer->flagNoRealloc = false;
pNewBuffer->flagWriteError = pNewBuffer->Resize(this->uNextInputLengthRealloc_);
}
else
{
AuResetMember(pNewBuffer);
}
}
else
{
pNewBuffer = AuMakeShared<AuByteBuffer>(this->uNextInputLengthRealloc_);
}
if (!pNewBuffer || !pNewBuffer->IsValid())
{
bFailed = true;
}
else
{
pNewBuffer->flagCircular = true;
this->pBuffer = pNewBuffer;
this->bOwnsBuffer = true;
}
}
try
{
if (this->pCallbackOptional_)
{
if (!bFailed)
{
this->pCallbackOptional_->OnSuccess((void *)nullptr);
}
else
{
this->pCallbackOptional_->OnFailure((void *)nullptr);
}
}
}
catch (...)
{
SysPushErrorCatch();
}
AuResetMember(this->uNextInputLengthRealloc_);
AuResetMember(this->uNextOutputLengthRealloc_);
AuResetMember(this->pCallbackOptional_);
}
} }
void NetDatagramSocketServerChannel::DoSendTick() void NetDatagramSocketServerChannel::DoSendTick()
@ -527,7 +750,7 @@ namespace Aurora::IO::Net
if (this->pBuffer) if (this->pBuffer)
{ {
if ((this->pBuffer->readPtr == this->pBuffer->writePtr) || if ((this->pBuffer->readPtr == this->pBuffer->writePtr) ||
((!this->pBuffer->flagCircular) && ((!this->pBuffer->flagCircular) && // ok?
(this->pBuffer->readPtr == this->pBuffer->base + this->pBuffer->length))) (this->pBuffer->readPtr == this->pBuffer->base + this->pBuffer->length)))
{ {
this->pBuffer.reset();; this->pBuffer.reset();;
@ -542,25 +765,111 @@ namespace Aurora::IO::Net
{ {
bool bFail {}; bool bFail {};
if (this->uFence_) if (bOutput && (AuAtomicLoad(&this->uFence_) || this->heads_.size()))
{ {
if (this->pCallbackOptional_) try
{ {
this->pCallbackOptional_->OnFailure((void *)nullptr); if (this->pCallbackOptional_)
{
this->pCallbackOptional_->OnFailure((void *)nullptr);
}
}
catch (...)
{
SysPushErrorCatch();
} }
this->uNextOutputLengthRealloc_ = uBytes; if (bOutput)
{
this->uNextOutputLengthRealloc_ = uBytes;
}
if (bInput)
{
this->uNextInputLengthRealloc_ = uBytes;
}
this->pCallbackOptional_ = pCallbackOptional; this->pCallbackOptional_ = pCallbackOptional;
return; return;
} }
if (bInput) if (bInput)
{ {
bFail = true; if (this->bOwnsBuffer &&
this->pBuffer)
{
// consumer read views should lock this into failing
// internally, we bypass read/writes views.
if (!ForceResize(*this->pBuffer, uBytes))
{
bFail = true;
}
}
else if (!this->bOwnsBuffer)
{
AuSPtr<AuByteBuffer> pNewBuffer;
if (this->pBuffer)
{
if (this->uNextInputLengthRealloc_ == this->pBuffer->size())
{
// NOP
}
else
{
bFail = true;
pNewBuffer = AuMakeShared<AuByteBuffer>(*this->pBuffer);
if (pNewBuffer && pNewBuffer->IsValid())
{
if (this->uNextInputLengthRealloc_ > pNewBuffer->size())
{
pNewBuffer->flagNoRealloc = false;
if (pNewBuffer->Resize(this->uNextInputLengthRealloc_))
{
pNewBuffer->flagCircular = true;
this->pBuffer = pNewBuffer;
this->bOwnsBuffer = true;
bFail = false;
}
}
}
}
}
else
{
pNewBuffer = AuMakeShared<AuByteBuffer>(this->uNextInputLengthRealloc_);
if (pNewBuffer && pNewBuffer->IsValid())
{
pNewBuffer->flagCircular = true;
this->pBuffer = pNewBuffer;
this->bOwnsBuffer = true;
}
else
{
bFail = true;
}
}
}
else
{
bFail = true;
}
} }
else if (bOutput) else if (bOutput)
{ {
if (!this->outputBuffer.Resize(uBytes)) if (this->heads_.size())
{
bFail = true;
}
else if (AuAtomicLoad(&this->uFence_))
{
bFail = true;
}
else if (!this->outputBuffer.Resize(uBytes))
{ {
bFail = true; bFail = true;
} }
@ -568,13 +877,20 @@ namespace Aurora::IO::Net
if (pCallbackOptional) if (pCallbackOptional)
{ {
if (!bFail) try
{ {
pCallbackOptional->OnSuccess((void *)nullptr); if (!bFail)
{
pCallbackOptional->OnSuccess((void *)nullptr);
}
else
{
pCallbackOptional->OnFailure((void *)nullptr);
}
} }
else catch (...)
{ {
pCallbackOptional->OnFailure((void *)nullptr); SysPushErrorCatch();
} }
} }
} }

9
Source/IO/Net/SocketOverDatagram/AuNetDatagramSocketServerChannel.hpp
View File

@ -76,6 +76,9 @@ namespace Aurora::IO::Net
AuSPtr<ISocketChannelLimits> GetChannelLimits() override; AuSPtr<ISocketChannelLimits> GetChannelLimits() override;
void SpecifyPageLength(AuUInt uPageSize) override;
AuUInt GetPageLength() override;
SocketStats &GetSendStatsEx(); SocketStats &GetSendStatsEx();
SocketStats &GetRecvStatsEx(); SocketStats &GetRecvStatsEx();
@ -87,6 +90,7 @@ namespace Aurora::IO::Net
const AuSPtr<AuAsync::PromiseCallback<AuNullS, AuNullS>> &pCallbackOptional); const AuSPtr<AuAsync::PromiseCallback<AuNullS, AuNullS>> &pCallbackOptional);
void ExpandRecv(const AuSPtr<AuByteBuffer> &pBuffer, bool bOwn); void ExpandRecv(const AuSPtr<AuByteBuffer> &pBuffer, bool bOwn);
void DoReallocRecvTick();
void DoSendTick(); void DoSendTick();
void UpdateReadPointers(); void UpdateReadPointers();
@ -96,7 +100,7 @@ namespace Aurora::IO::Net
void SchedWriteTick(); void SchedWriteTick();
void WriteTick(); void WriteTick();
void OnComplete(); void OnComplete(AuUInt32 uNBytes);
AuSPtr<AuByteBuffer> pBuffer; AuSPtr<AuByteBuffer> pBuffer;
@ -106,6 +110,7 @@ namespace Aurora::IO::Net
AuUInt uBytesToFlip { 0 }; AuUInt uBytesToFlip { 0 };
AuUInt uBytesInputBuffer { 0 }; AuUInt uBytesInputBuffer { 0 };
AuUInt uBytesOutputBuffer { 0 }; AuUInt uBytesOutputBuffer { 0 };
AuUInt uSpecifyPageLength { 0 };
AuSPtr<Protocol::IProtocolStack> pRecvProtocol; AuSPtr<Protocol::IProtocolStack> pRecvProtocol;
AuSPtr<Protocol::IProtocolStack> pSendProtocol; AuSPtr<Protocol::IProtocolStack> pSendProtocol;
@ -118,7 +123,9 @@ namespace Aurora::IO::Net
AuList<AuUInt> heads_; AuList<AuUInt> heads_;
AuThreadPrimitives::Mutex mutex_; AuThreadPrimitives::Mutex mutex_;
AuUInt32 uFence_ {}; AuUInt32 uFence_ {};
AuUInt uBytesCompleted_ {};
AuUInt32 uNextOutputLengthRealloc_ {}; AuUInt32 uNextOutputLengthRealloc_ {};
AuUInt32 uNextInputLengthRealloc_ {};
AuSPtr<AuAsync::PromiseCallback<AuNullS, AuNullS>> pCallbackOptional_; AuSPtr<AuAsync::PromiseCallback<AuNullS, AuNullS>> pCallbackOptional_;
bool bRateLimitPerTick { false }; bool bRateLimitPerTick { false };
}; };

83
Source/IO/Net/SocketOverDatagram/AuNetDatagramSocketServerSession.cpp
View File

@ -47,14 +47,6 @@ namespace Aurora::IO::Net
this->pDriver = pDriverFactory->NewSocketDriver(); this->pDriver = pDriverFactory->NewSocketDriver();
if (!this->channel.outputBuffer.Resize(this->ToParent()->uDefaultPacketSize))
{
return;
}
//this->channel.outputBuffer.flagCircular = true;
this->channel.outputBuffer.flagExpandable = false;
if (this->pDriver) if (this->pDriver)
{ {
try try
@ -67,6 +59,77 @@ namespace Aurora::IO::Net
this->bIsReady = true; this->bIsReady = true;
// strictly uBytesInputBuffer bytebuffer ring bytes
if (auto uBytes = this->channel.uBytesInputBuffer)
{
// [+] Realtime
// [-] Slower read
auto pBytebuffer = AuMakeShared<AuMemory::ByteBuffer>(uBytes);
if (pBytebuffer)
{
// TODO: not even TCP uses circular buffers
#if 1
// (datagram channel) DoEndReadTick will not break me with this flag
// > keep alive
// > do not fail to segmented buffer space
pBytebuffer->flagCircular = true;
#endif
pBytebuffer->flagNoRealloc = true;
this->channel.outputBuffer.flagExpandable = false;
}
if (!pBytebuffer || !(*pBytebuffer))
{
this->Destroy();
return;
}
this->channel.pBuffer = pBytebuffer;
this->channel.bOwnsBuffer = true;
}
else
{
// [+] Faster read
// [+] Per frame optimized
// [-] Dynamic memory allocation
// [old behaviour] dynamically scale if no channel resize call was made
// pass the buck onto the user to prevent abuse
}
{
if (auto uBytes = this->channel.uBytesOutputBuffer)
{
if (!this->channel.outputBuffer.Resize(uBytes))
{
this->Destroy();
return;
}
}
else
{
if (!this->channel.outputBuffer.Resize(this->ToParent()->uDefaultPacketSize))
{
this->Destroy();
return;
}
}
// ring buffers dont scale so well to arbitrary send order acknowledgement
// oh well.
// we would need a third lock head to prevent overwrite into pending submit regions.
// the main issue is for linear timelines we can just += -= pointer read/write heads.
// when those guarantees are chucked away, in this example, we cannot increment the read head until all of the requests have completed.
// if another write tick comes along and we're not done? guess what? we're resending from frame[0] again
// if an async write acknowledgement comes in out of order? guess what? now we've probably seeked to the wrong read head location
// it's just easier to not use ring buffers here, and reset the heads to the base pointer when starved of pending async io.
this->channel.outputBuffer.flagExpandable = false;
this->channel.outputBuffer.flagNoRealloc = true;
// On tcp implementation parity, we would need to flip these two flagCircular states to be inline with AuNetSocketChannelOutput.cpp and AuNetSocketChannelInput.cpp
}
this->channel.InsertFrameFence(); this->channel.InsertFrameFence();
this->pDriver->OnEstablish(); this->pDriver->OnEstablish();
@ -230,7 +293,7 @@ namespace Aurora::IO::Net
{ {
this->bIsDead = true; this->bIsDead = true;
if (pWorker->IsOnThread()) if (this->pWorker->IsOnThread())
{ {
if (auto pDriver = this->pDriver) if (auto pDriver = this->pDriver)
{ {
@ -247,7 +310,7 @@ namespace Aurora::IO::Net
return; return;
} }
pWorker->TryScheduleInternalTemplate<AuNullS>([pThat = this->SharedFromThis()](const AuSPtr<AuAsync::PromiseCallback<AuNullS>> &info) this->pWorker->TryScheduleInternalTemplate<AuNullS>([pThat = this->SharedFromThis()](const AuSPtr<AuAsync::PromiseCallback<AuNullS>> &info)
{ {
pThat->Destroy(); pThat->Destroy();
}, AuStaticCast<AuAsync::PromiseCallback<AuNullS, AuNullS>>(AuMakeShared<AuAsync::PromiseCallbackFunctional<AuNullS, AuNullS>>([=](const AuSPtr<AuNullS> &dumb) }, AuStaticCast<AuAsync::PromiseCallback<AuNullS, AuNullS>>(AuMakeShared<AuAsync::PromiseCallbackFunctional<AuNullS, AuNullS>>([=](const AuSPtr<AuNullS> &dumb)

27
Source/IO/Protocol/AuProtocolStack.cpp
View File

@ -132,6 +132,11 @@ namespace Aurora::IO::Protocol
// Circular ref // Circular ref
pNew->pOuputWriter = AuMakeShared<AuIO::Buffered::BlobWriter>(AuSPtr<Memory::ByteBuffer>(pNew, &pNew->outputBuffer)); pNew->pOuputWriter = AuMakeShared<AuIO::Buffered::BlobWriter>(AuSPtr<Memory::ByteBuffer>(pNew, &pNew->outputBuffer));
if (!pNew->pOuputWriter)
{
SysPushErrorNet("Out of memory");
return {};
}
struct StreamWrapper : IStreamWriter, IProtocolNext, AuEnableSharedFromThis<StreamWrapper> struct StreamWrapper : IStreamWriter, IProtocolNext, AuEnableSharedFromThis<StreamWrapper>
{ {
@ -297,6 +302,11 @@ namespace Aurora::IO::Protocol
// Circular ref // Circular ref
pNew->pOuputWriter = AuMakeShared<AuIO::Buffered::BlobWriter>(AuSPtr<Memory::ByteBuffer>(pNew, &pNew->outputBuffer)); pNew->pOuputWriter = AuMakeShared<AuIO::Buffered::BlobWriter>(AuSPtr<Memory::ByteBuffer>(pNew, &pNew->outputBuffer));
if (!pNew->pOuputWriter)
{
SysPushErrorNet("Out of memory");
return {};
}
struct StreamWrapper : IStreamWriter, IProtocolNext, AuEnableSharedFromThis<StreamWrapper> struct StreamWrapper : IStreamWriter, IProtocolNext, AuEnableSharedFromThis<StreamWrapper>
{ {
@ -756,22 +766,11 @@ namespace Aurora::IO::Protocol
if (target >= pCurrent->outputBuffer.allocSize || if (target >= pCurrent->outputBuffer.allocSize ||
((pCurrent->outputBuffer.length > target) && (bytesRem < pCurrent->uStartingSize))) ((pCurrent->outputBuffer.length > target) && (bytesRem < pCurrent->uStartingSize)))
{ {
AuByteBuffer replacement(target, true, false); if (pCurrent->outputBuffer.Resize(target))
if (!replacement)
{ {
this->Terminate(); pCurrent->outputBuffer.flagAlwaysExpandable =
return false; pCurrent->outputBuffer.flagExpandable = 1;
} }
if (!replacement.WriteFrom(pCurrent->outputBuffer))
{
this->Terminate();
return false;
}
pCurrent->outputBuffer = AuMove(replacement);
pCurrent->outputBuffer.flagAlwaysExpandable =
pCurrent->outputBuffer.flagExpandable = 1;
} }
} }
} }