Skip to content

Commit

Permalink
fix(Net): Non-blocking sockets support #4773
Browse files Browse the repository at this point in the history
  • Loading branch information
@obiltschnig
obiltschnig committed Nov 16, 2024
1 parent cb91880 commit 5d69308
Show file tree
Hide file tree
Showing 4 changed files with 93 additions and 20 deletions.
88 changes: 69 additions & 19 deletions Net/src/SocketImpl.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -361,24 +361,37 @@ void SocketImpl::checkBrokenTimeout(SelectMode mode)

int SocketImpl::sendBytes(const void* buffer, int length, int flags)
{
checkBrokenTimeout(SELECT_WRITE);

if (_blocking)
{
checkBrokenTimeout(SELECT_WRITE);
}
int rc;
do
{
if (_sockfd == POCO_INVALID_SOCKET) throw InvalidSocketException();
rc = ::send(_sockfd, reinterpret_cast<const char*>(buffer), length, flags);
}
while (_blocking && rc < 0 && lastError() == POCO_EINTR);
if (rc < 0) error();
if (rc < 0)
{
int err = lastError();
if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
;
else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
throw TimeoutException(err);
else
error(err);
}
return rc;
}


int SocketImpl::sendBytes(const SocketBufVec& buffers, int flags)
{
checkBrokenTimeout(SELECT_WRITE);

if (_blocking)
{
checkBrokenTimeout(SELECT_WRITE);
}
int rc = 0;
do
{
Expand All @@ -395,15 +408,26 @@ int SocketImpl::sendBytes(const SocketBufVec& buffers, int flags)
#endif
}
while (_blocking && rc < 0 && lastError() == POCO_EINTR);
if (rc < 0) error();
if (rc < 0)
{
int err = lastError();
if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
;
else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
throw TimeoutException(err);
else
error(err);
}
return rc;
}


int SocketImpl::receiveBytes(void* buffer, int length, int flags)
{
checkBrokenTimeout(SELECT_READ);

if (_blocking)
{
checkBrokenTimeout(SELECT_READ);
}
int rc;
do
{
Expand All @@ -414,7 +438,7 @@ int SocketImpl::receiveBytes(void* buffer, int length, int flags)
if (rc < 0)
{
int err = lastError();
if (err == POCO_EAGAIN && !_blocking)
if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
;
else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
throw TimeoutException(err);
Expand All @@ -427,8 +451,10 @@ int SocketImpl::receiveBytes(void* buffer, int length, int flags)

int SocketImpl::receiveBytes(SocketBufVec& buffers, int flags)
{
checkBrokenTimeout(SELECT_READ);

if (_blocking)
{
checkBrokenTimeout(SELECT_READ);
}
int rc = 0;
do
{
Expand All @@ -448,7 +474,7 @@ int SocketImpl::receiveBytes(SocketBufVec& buffers, int flags)
if (rc < 0)
{
int err = lastError();
if (err == POCO_EAGAIN && !_blocking)
if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
;
else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
throw TimeoutException(err);
Expand Down Expand Up @@ -476,7 +502,7 @@ int SocketImpl::receiveBytes(Poco::Buffer<char>& buffer, int flags, const Poco::
if (rc < 0)
{
int err = lastError();
if (err == POCO_EAGAIN && !_blocking)
if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
;
else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
throw TimeoutException(err);
Expand All @@ -502,7 +528,16 @@ int SocketImpl::sendTo(const void* buffer, int length, const SocketAddress& addr
#endif
}
while (_blocking && rc < 0 && lastError() == POCO_EINTR);
if (rc < 0) error();
if (rc < 0)
{
int err = lastError();
if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
;
else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
throw TimeoutException(err);
else
error(err);
}
return rc;
}

Expand Down Expand Up @@ -534,7 +569,16 @@ int SocketImpl::sendTo(const SocketBufVec& buffers, const SocketAddress& address
#endif
}
while (_blocking && rc < 0 && lastError() == POCO_EINTR);
if (rc < 0) error();
if (rc < 0)
{
int err = lastError();
if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
;
else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
throw TimeoutException(err);
else
error(err);
}
return rc;
}

Expand All @@ -556,7 +600,10 @@ int SocketImpl::receiveFrom(void* buffer, int length, SocketAddress& address, in

int SocketImpl::receiveFrom(void* buffer, int length, struct sockaddr** ppSA, poco_socklen_t** ppSALen, int flags)
{
checkBrokenTimeout(SELECT_READ);
if (_blocking)
{
checkBrokenTimeout(SELECT_READ);
}
int rc;
do
{
Expand All @@ -567,7 +614,7 @@ int SocketImpl::receiveFrom(void* buffer, int length, struct sockaddr** ppSA, po
if (rc < 0)
{
int err = lastError();
if (err == POCO_EAGAIN && !_blocking)
if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
;
else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
throw TimeoutException(err);
Expand Down Expand Up @@ -595,7 +642,10 @@ int SocketImpl::receiveFrom(SocketBufVec& buffers, SocketAddress& address, int f

int SocketImpl::receiveFrom(SocketBufVec& buffers, struct sockaddr** pSA, poco_socklen_t** ppSALen, int flags)
{
checkBrokenTimeout(SELECT_READ);
if (_blocking)
{
checkBrokenTimeout(SELECT_READ);
}
int rc = 0;
do
{
Expand Down Expand Up @@ -624,7 +674,7 @@ int SocketImpl::receiveFrom(SocketBufVec& buffers, struct sockaddr** pSA, poco_s
if (rc < 0)
{
int err = lastError();
if (err == POCO_EAGAIN && !_blocking)
if (!_blocking && (err == POCO_EAGAIN || err == POCO_EWOULDBLOCK))
;
else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
throw TimeoutException(err);
Expand Down
2 changes: 1 addition & 1 deletion Net/src/StreamSocketImpl.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ int StreamSocketImpl::sendBytes(const void* buffer, int length, int flags)
while (remaining > 0)
{
int n = SocketImpl::sendBytes(p, remaining, flags);
poco_assert_dbg (n >= 0);
poco_assert_dbg (!blocking || n >= 0);
p += n;
sent += n;
remaining -= n;
Expand Down
22 changes: 22 additions & 0 deletions Net/testsuite/src/SocketTest.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,27 @@ void SocketTest::testEcho()
}


void SocketTest::testPeek()
{
EchoServer echoServer;
StreamSocket ss;
ss.connect(SocketAddress("127.0.0.1", echoServer.port()));
int n = ss.sendBytes("hello, world!", 13);
assertTrue (n == 13);
char buffer[256];
n = ss.receiveBytes(buffer, 5, MSG_PEEK);
assertTrue (n == 5);
assertTrue (std::string(buffer, n) == "hello");
n = ss.receiveBytes(buffer, sizeof(buffer), MSG_PEEK);
assertTrue (n == 13);
assertTrue (std::string(buffer, n) == "hello, world!");
n = ss.receiveBytes(buffer, sizeof(buffer));
assertTrue (n == 13);
assertTrue (std::string(buffer, n) == "hello, world!");
ss.close();
}


void SocketTest::testMoveStreamSocket()
{
EchoServer echoServer;
Expand Down Expand Up @@ -679,6 +700,7 @@ CppUnit::Test* SocketTest::suite()
CppUnit::TestSuite* pSuite = new CppUnit::TestSuite("SocketTest");

CppUnit_addTest(pSuite, SocketTest, testEcho);
CppUnit_addTest(pSuite, SocketTest, testPeek);
CppUnit_addTest(pSuite, SocketTest, testMoveStreamSocket);
CppUnit_addTest(pSuite, SocketTest, testPoll);
CppUnit_addTest(pSuite, SocketTest, testAvailable);
Expand Down
1 change: 1 addition & 0 deletions Net/testsuite/src/SocketTest.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ class SocketTest: public CppUnit::TestCase
~SocketTest() override;

void testEcho();
void testPeek();
void testMoveStreamSocket();
void testPoll();
void testAvailable();
Expand Down

0 comments on commit 5d69308

Please sign in to comment.