all windows fixes

This commit is contained in:
Robin Müller 2021-09-28 17:26:02 +02:00
parent 6e88f8f400
commit 32b5060c62
2 changed files with 162 additions and 13 deletions

View File

@ -41,6 +41,17 @@ ReturnValue_t TcpTmTcServer::initialize() {
return result; return result;
} }
switch(receptionMode) {
case(ReceptionModes::SPACE_PACKETS): {
spacePacketParser = new SpacePacketParser(validPacketIds);
if(spacePacketParser == nullptr) {
return HasReturnvaluesIF::RETURN_FAILED;
}
#if defined PLATFORM_UNIX
tcpConfig.tcpFlags |= MSG_DONTWAIT;
#endif
}
}
tcStore = ObjectManager::instance()->get<StorageManagerIF>(objects::TC_STORE); tcStore = ObjectManager::instance()->get<StorageManagerIF>(objects::TC_STORE);
if (tcStore == nullptr) { if (tcStore == nullptr) {
#if FSFW_CPP_OSTREAM_ENABLED == 1 #if FSFW_CPP_OSTREAM_ENABLED == 1
@ -144,11 +155,14 @@ ReturnValue_t TcpTmTcServer::initializeAfterTaskCreation() {
return HasReturnvaluesIF::RETURN_OK; return HasReturnvaluesIF::RETURN_OK;
} }
void TcpTmTcServer::handleServerOperation(socket_t connSocket) { void TcpTmTcServer::handleServerOperation(socket_t& connSocket) {
int retval = 0; #if defined PLATFORM_WIN
do { setSocketNonBlocking(connSocket);
// Read all telecommands sent by the client #endif
retval = recv(connSocket,
while (true) {
int retval = recv(
connSocket,
reinterpret_cast<char*>(receptionBuffer.data()), reinterpret_cast<char*>(receptionBuffer.data()),
receptionBuffer.capacity(), receptionBuffer.capacity(),
tcpFlags); tcpFlags);
@ -159,9 +173,34 @@ void TcpTmTcServer::handleServerOperation(socket_t connSocket) {
// Client has finished sending telecommands, send telemetry now // Client has finished sending telecommands, send telemetry now
handleTmSending(connSocket); handleTmSending(connSocket);
} }
else { else if(retval < 0) {
// Should not happen int errorValue = GetLastError();
tcpip::handleError(tcpip::Protocol::TCP, tcpip::ErrorSources::RECV_CALL); #if defined PLATFORM_UNIX
int wouldBlockValue = EAGAIN;
#elif defined PLATFORM_WIN
int wouldBlockValue = WSAEWOULDBLOCK;
#endif
if(errorValue == wouldBlockValue) {
// No data available. Check whether any packets have been read, then send back
// telemetry if available
bool tcAvailable = false;
bool tmSent = false;
size_t availableReadData = ringBuffer.getAvailableReadData();
if(availableReadData > lastRingBufferSize) {
tcAvailable = true;
handleTcRingBufferData(availableReadData);
}
ReturnValue_t result = handleTmSending(connSocket, tmSent);
if(result == CONN_BROKEN) {
return;
}
if(not tcAvailable and not tmSent) {
TaskFactory::delayTask(tcpConfig.tcpLoopDelay);
}
}
else {
tcpip::handleError(tcpip::Protocol::TCP, tcpip::ErrorSources::RECV_CALL, 300);
}
} }
} while(retval > 0); } while(retval > 0);
} }
@ -228,3 +267,106 @@ ReturnValue_t TcpTmTcServer::handleTmSending(socket_t connSocket) {
} }
return HasReturnvaluesIF::RETURN_OK; return HasReturnvaluesIF::RETURN_OK;
} }
ReturnValue_t TcpTmTcServer::handleTcRingBufferData(size_t availableReadData) {
ReturnValue_t status = HasReturnvaluesIF::RETURN_OK;
ReturnValue_t result = HasReturnvaluesIF::RETURN_OK;
size_t readAmount = availableReadData;
lastRingBufferSize = availableReadData;
if(readAmount >= ringBuffer.getMaxSize()) {
#if FSFW_VERBOSE_LEVEL >= 1
#if FSFW_CPP_OSTREAM_ENABLED == 1
// Possible configuration error, too much data or/and data coming in too fast,
// requiring larger buffers
sif::warning << "TcpTmTcServer::handleServerOperation: Ring buffer reached " <<
"fill count" << std::endl;
#else
sif::printWarning("TcpTmTcServer::handleServerOperation: Ring buffer reached "
"fill count");
#endif
#endif
}
if(readAmount >= receptionBuffer.size()) {
#if FSFW_VERBOSE_LEVEL >= 1
#if FSFW_CPP_OSTREAM_ENABLED == 1
// Possible configuration error, too much data or/and data coming in too fast,
// requiring larger buffers
sif::warning << "TcpTmTcServer::handleServerOperation: "
"Reception buffer too small " << std::endl;
#else
sif::printWarning("TcpTmTcServer::handleServerOperation: Reception buffer too small\n");
#endif
#endif
readAmount = receptionBuffer.size();
}
ringBuffer.readData(receptionBuffer.data(), readAmount, true);
const uint8_t* bufPtr = receptionBuffer.data();
const uint8_t** bufPtrPtr = &bufPtr;
size_t startIdx = 0;
size_t foundSize = 0;
size_t readLen = 0;
while(readLen < readAmount) {
result = spacePacketParser->parseSpacePackets(bufPtrPtr, readAmount,
startIdx, foundSize, readLen);
switch(result) {
case(SpacePacketParser::NO_PACKET_FOUND):
case(SpacePacketParser::SPLIT_PACKET): {
break;
}
case(HasReturnvaluesIF::RETURN_OK): {
result = handleTcReception(receptionBuffer.data() + startIdx, foundSize);
if(result != HasReturnvaluesIF::RETURN_OK) {
status = result;
}
}
}
ringBuffer.deleteData(foundSize);
lastRingBufferSize = ringBuffer.getAvailableReadData();
std::memset(receptionBuffer.data() + startIdx, 0, foundSize);
}
return status;
}
void TcpTmTcServer::enableWiretapping(bool enable) {
this->wiretappingEnabled = enable;
}
void TcpTmTcServer::handleSocketError(ConstStorageAccessor &accessor) {
// Don't delete data
accessor.release();
auto socketError = getLastSocketError();
switch(socketError) {
#if defined PLATFORM_WIN
case(WSAECONNRESET): {
// See https://docs.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-send
// Remote client might have shut down connection
return;
}
#else
case(EPIPE): {
// See https://man7.org/linux/man-pages/man2/send.2.html
// Remote client might have shut down connection
return;
}
#endif
default: {
tcpip::handleError(tcpip::Protocol::TCP, tcpip::ErrorSources::SEND_CALL);
}
}
}
void TcpTmTcServer::setSocketNonBlocking(socket_t &connSocket) {
u_long iMode = 1;
int iResult = ioctlsocket(connSocket, FIONBIO, &iMode);
if(iResult != NO_ERROR) {
#if FSFW_VERBOSE_LEVEL >= 1
#if FSFW_CPP_OSTREAM_ENABLED == 1
sif::warning << "TcpTmTcServer::handleServerOperation: Setting socket"
" non-blocking failed with error " << iResult;
#else
sif::printWarning("TcpTmTcServer::handleServerOperation: Setting socket"
" non-blocking failed with error %d\n", iResult);
#endif
#endif
}
}

View File

@ -84,12 +84,19 @@ private:
int tcpBacklog = 3; int tcpBacklog = 3;
std::vector<uint8_t> receptionBuffer; std::vector<uint8_t> receptionBuffer;
int tcpSockOpt = 0; SimpleRingBuffer ringBuffer;
int tcpTmFlags = 0; std::vector<uint16_t> validPacketIds;
SpacePacketParser* spacePacketParser = nullptr;
uint8_t lastRingBufferSize = 0;
void handleServerOperation(socket_t connSocket); virtual void handleServerOperation(socket_t& connSocket);
ReturnValue_t handleTcReception(size_t bytesRecvd); ReturnValue_t handleTcReception(uint8_t* spacePacket, size_t packetSize);
ReturnValue_t handleTmSending(socket_t connSocket); ReturnValue_t handleTmSending(socket_t connSocket, bool& tmSent);
ReturnValue_t handleTcRingBufferData(size_t availableReadData);
void handleSocketError(ConstStorageAccessor& accessor);
#if defined PLATFORM_WIN
void setSocketNonBlocking(socket_t& connSocket);
#endif
}; };
#endif /* FSFW_OSAL_COMMON_TCP_TMTC_SERVER_H_ */ #endif /* FSFW_OSAL_COMMON_TCP_TMTC_SERVER_H_ */