AuroraRuntime/Source/IO/Net/AuNetSocketChannelOutput.cpp

295 lines
8.3 KiB
C++

/***
Copyright (C) 2022 J Reece Wilson (a/k/a "Reece"). All rights reserved.
File: AuNetSocketChannelOutput.cpp
Date: 2022-8-21
Author: Reece
***/
#include "Networking.hpp"
#include "AuNetSocket.hpp"
#include "AuNetSocketChannelOutput.hpp"
#include "AuNetWorker.hpp"
#include <Source/IO/AuIOPipeProcessor.hpp>
#if defined(AURORA_IS_MODERNNT_DERIVED)
#include "AuNetStream.NT.hpp"
#else
#include "AuNetStream.Linux.hpp"
#endif
static const auto kDefaultBufferSize = 64 * 1024;
namespace Aurora::IO::Net
{
SocketChannelOutput::SocketChannelOutput(SocketBase *pParent, const AuSPtr<IAsyncTransaction> &stream) :
pParent_(pParent),
pNetWriteTransaction_(stream),
outputBuffer_(kDefaultBufferSize, true)
{
this->outputWriteQueue_.pBase = pParent;
}
bool SocketChannelOutput::IsValid()
{
return bool(this->pNetWriteTransaction_) &&
bool(this->outputBuffer_);
}
AuSPtr<IAsyncTransaction> SocketChannelOutput::ToWriteTransaction()
{
return AuSPtr<IAsyncTransaction>(this->pNetWriteTransaction_);
}
AuSPtr<Memory::ByteBuffer> SocketChannelOutput::AsWritableByteBuffer()
{
if (!this->pParent_)
{
return {};
}
return AuSPtr<Memory::ByteBuffer>(this->pParent_->SharedFromThis(),
&this->outputBuffer_);
}
void SocketChannelOutput::ScheduleOutOfFrameWrite()
{
SendIfData();
SchedWriteTick();
}
void SocketChannelOutput::SchedWriteTick()
{
if (!this->pParent_)
{
return;
}
auto pWorker = this->pParent_->ToWorkerEx();
//if (this->pParent_->bHasFinalized_)
{
if (pWorker->IsOnThread())
{
WriteTick();
}
return;
}
auto meShared = AuSPtr<SocketChannelOutput>(this->pParent_->SharedFromThis(), this);
if (!pWorker->TryScheduleInternalTemplate<AuNullS>([=](const AuSPtr<AuAsync::PromiseCallback<AuNullS>> &info)
{
meShared->WriteTick();
info->OnSuccess((void *)nullptr);
}, AuStaticCast<AuAsync::PromiseCallback<AuNullS, AuNullS>>(AuMakeShared<AuAsync::PromiseCallbackFunctional<AuNullS, AuNullS>>([=](const AuSPtr<AuNullS> &dumb)
{
}))))
{
SysPushErrorIO("Couldn't schedule write tick");
this->pParent_->SendErrorBeginShutdown(AuNet::ENetworkError::eAsyncError);
}
}
void SocketChannelOutput::SendIfData()
{
AU_LOCK_GUARD(this->lock_);
if (!this->pParent_)
{
return;
}
struct View : AuMemoryViewRead
{
View(const AuMemoryViewRead &in) : AuMemoryViewRead(in)
{
}
AuSPtr<void> pin;
};
while (this->pOutSendPointer_ != this->outputBuffer_.writePtr)
{
if (this->pOutSendPointer_ == this->outputBuffer_.writePtr)
{
break;
}
if (this->pOutSendPointer_ == this->outputBuffer_.base + this->outputBuffer_.length)
{
this->pOutSendPointer_ = this->outputBuffer_.base;
continue;
}
if (!this->pOutSendPointer_)
{
this->pOutSendPointer_ = this->outputBuffer_.base;
continue;
}
if ((AuUInt8 *)this->pOutSendPointer_ > this->outputBuffer_.writePtr)
{
auto pBase = (AuUInt8 *)this->pOutSendPointer_;
auto uLen = (AuUInt8 *)this->outputBuffer_.base + this->outputBuffer_.length - pBase;
auto pView = AuMakeSharedPanic<View>(AuMemoryViewRead(pBase, uLen));
SysAssert(pView);
pView->pin = this->pParent_->SharedFromThis();
this->outputWriteQueue_.Push(pView);
this->pOutSendPointer_ = (AuUInt8 *)this->pOutSendPointer_ + uLen;
}
else
{
auto pBase = (AuUInt8 *)this->pOutSendPointer_;
auto pWriteHead = (AuUInt8 *)this->outputBuffer_.writePtr;
auto uLen = pWriteHead - pBase;
auto pView = AuMakeSharedPanic<View>(AuMemoryViewRead(pBase, uLen));
SysAssert(pView);
pView->pin = this->pParent_->SharedFromThis();
this->outputWriteQueue_.Push(pView);
this->pOutSendPointer_ = pWriteHead ;
}
}
}
void SocketChannelOutput::WriteTick()
{
bool bShouldShutdown {};
{
AU_LOCK_GUARD(this->lock_);
bShutdownOnComplete = WriteTickLocked();
}
if (bShouldShutdown)
{
if (this->pParent_)
{
this->pParent_->Shutdown(true);
}
}
}
bool SocketChannelOutput::WriteTickLocked()
{
if (!this->pParent_)
{
return false;
}
#if defined(AURORA_IS_MODERNNT_DERIVED)
auto pHackTransaction =
AuStaticCast<NtAsyncNetworkTransaction>(this->pNetWriteTransaction_);
#else
auto pHackTransaction =
AuStaticCast<LinuxAsyncNetworkTransaction>(this->pNetWriteTransaction_);
#endif
if (pHackTransaction->bIsWriting)
{
// IsComplete?
if (!pHackTransaction->bLatch)
{
return false;
}
}
if (this->outputWriteQueue_.IsEmpty())
{
if (this->pParent_)// &&
///this->outputBuffer_.outputChannel.AsWritableByteBuffer()->GetNextLinearRead().length == 0)
{
this->pParent_->socketChannel_.DoReallocWriteTick();
}
}
// do not forcefully flush preemptive hello packets until the socket has properly connected
if (!this->pParent_->bHasConnected_)
{
return true;
}
if (auto pFrameToSend = this->outputWriteQueue_.Dequeue())
{
if (this->pParent_)
{
this->pNetWriteTransaction_->SetCallback(AuSPtr<IAsyncFinishedSubscriber>(this->pParent_->SharedFromThis(), this));
if (!this->pParent_->ToWorkerEx()->IncrementIOEventTaskCounter())
{
SysPushErrorIO("Couldn't begin wait");
this->pParent_->SendErrorBeginShutdown({});
return false;
}
if (!this->pNetWriteTransaction_->StartWrite(0, pFrameToSend))
{
this->pParent_->ToWorkerEx()->DecrementIOEventTaskCounter();
SysPushErrorIO("Couldn't dispatch the to-send frame, had: {} bytes remaining to send", pFrameToSend->length);
this->pParent_->SendErrorBeginShutdown({});
return false;
}
}
}
else
{
this->pNetWriteTransaction_->SetCallback({});
if (this->outputBuffer_.HasStreamError())
{
this->pParent_->SendErrorBeginShutdown(AuNet::ENetworkError::eBrokenByteBuffer);
return true;
}
if (this->bShutdownOnComplete)
{
return true;
}
}
return false;
}
void SocketChannelOutput::OnEndOfReadTick()
{
SendIfData();
WriteTick();
}
AuByteBuffer &SocketChannelOutput::GetByteBuffer()
{
return this->outputBuffer_;
}
bool SocketChannelOutput::CanResize()
{
return this->outputWriteQueue_.IsEmpty();
}
void SocketChannelOutput::OnAsyncFileOpFinished(AuUInt64 offset, AuUInt32 length)
{
if (!this->pParent_)
{
return;
}
this->pParent_->ToWorkerEx()->DecrementIOEventTaskCounter();
AuStaticCast<SocketChannel>(this->pParent_->ToChannel())->GetSendStatsEx().AddBytes(length);
if (!length)
{
this->pParent_->SendErrorBeginShutdown({});
return;
}
this->outputWriteQueue_.NotifyBytesWritten(length); // does the resend bit
SysAssert(this->outputBuffer_.ReaderTryGoForward(length));
WriteTick();
}
}