generic-library/poco
1
0
Fork 0
mirror of https://github.com/pocoproject/poco.git synced 2026-01-12 10:02:36 +01:00
Code Issues Projects Releases Wiki Activity

Feature net udp (#2347)

Browse Source 
* add PMTU discovery #2329

* add socket gather/scatter capabilities #2330 (win, udp)

* enable WSAPoll

* add FastMemoryPool

* add receiveFrom() with native args

* allow copying of StringTokenizer

* add AtomicFlag and SpinlockMutex

* update .gitignore

* UDPServer and client #2343 (windows)

* fix warnings

* fix warnings

* regenerate Net VS solutions

* regenerate CppUnit projects/solutions

* clang fixes

* gcc fixes

* try to fix travis

* more travis fixes

* more travis fixes

* handle UDPClient exception

* fix makefiles and init order warnings

* add UNIX gather/scatter sendto/recvfrom implementations and tests

* run travis tests as sudo

* try to run tests as sudo, 2nd attempt

* fix warning

* use mutex in reactor

* lock-order-inversion in SocketReactor #2346

* add PMTU discovery #2329 (linux)

* ICMPSocket does not check reply address #1921

* remove some ignored tests

* add PMTU discovery #2329 (reconcile logic with #1921)

* fix native receiveFrome()

* reinstate ignoring of proxy errors

* add testMTU to ignore list

* add include atomic

* NTPClient not checking reply address #2348

* some ICMP/MTU fixes

* UDPSocketReader cleanup

* resolve some socket inheritance warnings

* add NTP time sync to ignored tests

* SocketNotifier not thread-safe #2345

* prevent x64 samples build attempt for win32

* build TestApp and Library

* fix ICMP tests

* regen VS projects

* regen VS projects and add missing 2012 files

* remove debug prints
This commit is contained in:
Aleksandar Fabijanic
2018-06-02 14:02:33 -05:00
committed by GitHub
parent da15142f69
commit c4e676d36d
127 changed files with 5540 additions and 1408 deletions
Download Patch File Download Diff File Expand all files Collapse all files

279
Net/src/SocketImpl.cpp
View File

@@ -18,11 +18,26 @@
#include "Poco/NumberFormatter.h"
#include "Poco/Timestamp.h"
#include <string.h> // FD_SET needs memset on some platforms, so we can't use <cstring>
#if defined(_WIN32) && _WIN32_WINNT >= 0x0600
#ifndef POCO_HAVE_FD_POLL
#define POCO_HAVE_FD_POLL 1
#endif
#elif defined(POCO_OS_FAMILY_BSD)
#ifndef POCO_HAVE_FD_POLL
#define POCO_HAVE_FD_POLL 1
#endif
#endif
#if defined(POCO_HAVE_FD_EPOLL)
#include <sys/epoll.h>
#elif defined(POCO_HAVE_FD_POLL)
#ifndef _WIN32
#include <poll.h>
#endif
#endif
#if defined(sun) || defined(__sun) || defined(__sun__)
@@ -56,7 +71,7 @@ bool checkIsBrokenTimeout()
// https://social.msdn.microsoft.com/Forums/en-US/76620f6d-22b1-4872-aaf0-833204f3f867/minimum-timeout-value-for-sorcvtimeo
OSVERSIONINFO vi;
vi.dwOSVersionInfoSize = sizeof(vi);
if (GetVersionEx(&vi) == 0) return true; //throw SystemException("Cannot get OS version information");
if (GetVersionEx(&vi) == 0) return true;
return vi.dwMajorVersion < 6 || (vi.dwMajorVersion == 6 && vi.dwMinorVersion < 2);
#endif
return false;
@@ -84,7 +99,7 @@ SocketImpl::~SocketImpl()
close();
}
SocketImpl* SocketImpl::acceptConnection(SocketAddress& clientAddr)
{
if (_sockfd == POCO_INVALID_SOCKET) throw InvalidSocketException();
@@ -189,7 +204,7 @@ void SocketImpl::connectNB(const SocketAddress& address)
void SocketImpl::bind(const SocketAddress& address, bool reuseAddress)
{
bind(address, reuseAddress, true);
bind(address, reuseAddress, true);
}
@@ -201,7 +216,7 @@ void SocketImpl::bind(const SocketAddress& address, bool reuseAddress, bool reus
}
if (reuseAddress)
setReuseAddress(true);
if (reusePort)
if (reusePort)
setReusePort(true);
#if defined(POCO_VXWORKS)
int rc = ::bind(_sockfd, (sockaddr*) address.addr(), address.length());
@@ -214,7 +229,7 @@ void SocketImpl::bind(const SocketAddress& address, bool reuseAddress, bool reus
void SocketImpl::bind6(const SocketAddress& address, bool reuseAddress, bool ipV6Only)
{
bind6(address, reuseAddress, true, ipV6Only);
bind6(address, reuseAddress, true, ipV6Only);
}
@@ -235,7 +250,7 @@ void SocketImpl::bind6(const SocketAddress& address, bool reuseAddress, bool reu
#endif
if (reuseAddress)
setReuseAddress(true);
if (reusePort)
if (reusePort)
setReusePort(true);
int rc = ::bind(_sockfd, address.addr(), address.length());
if (rc != 0) error(address.toString());
@@ -244,7 +259,7 @@ void SocketImpl::bind6(const SocketAddress& address, bool reuseAddress, bool reu
#endif
}
void SocketImpl::listen(int backlog)
{
if (_sockfd == POCO_INVALID_SOCKET) throw InvalidSocketException();
@@ -272,7 +287,7 @@ void SocketImpl::shutdownReceive()
if (rc != 0) error();
}
void SocketImpl::shutdownSend()
{
if (_sockfd == POCO_INVALID_SOCKET) throw InvalidSocketException();
@@ -281,7 +296,7 @@ void SocketImpl::shutdownSend()
if (rc != 0) error();
}
void SocketImpl::shutdown()
{
if (_sockfd == POCO_INVALID_SOCKET) throw InvalidSocketException();
@@ -291,16 +306,22 @@ void SocketImpl::shutdown()
}
int SocketImpl::sendBytes(const void* buffer, int length, int flags)
void SocketImpl::checkBrokenTimeout()
{
if (_isBrokenTimeout)
{
if (_sndTimeout.totalMicroseconds() != 0)
if (_recvTimeout.totalMicroseconds() != 0)
{
if (!poll(_sndTimeout, SELECT_WRITE))
if (!poll(_recvTimeout, SELECT_READ))
throw TimeoutException();
}
}
}
int SocketImpl::sendBytes(const void* buffer, int length, int flags)
{
checkBrokenTimeout();
int rc;
do
@@ -314,17 +335,35 @@ int SocketImpl::sendBytes(const void* buffer, int length, int flags)
}
int SocketImpl::sendBytes(const SocketBufVec& buffers, int flags)
{
checkBrokenTimeout();
int rc = 0;
do
{
if (_sockfd == POCO_INVALID_SOCKET) throw InvalidSocketException();
#if defined(POCO_OS_FAMILY_WINDOWS)
DWORD sent = 0;
rc = WSASend(_sockfd, const_cast<LPWSABUF>(&buffers[0]),
static_cast<DWORD>(buffers.size()), &sent,
static_cast<DWORD>(flags), 0, 0);
if (rc == SOCKET_ERROR) error();
rc = sent;
#elif defined(POCO_OS_FAMILY_UNIX)
rc = writev(_sockfd, &buffers[0], static_cast<int>(buffers.size()));
#endif
}
while (_blocking && rc < 0 && lastError() == POCO_EINTR);
if (rc < 0) error();
return rc;
}
int SocketImpl::receiveBytes(void* buffer, int length, int flags)
{
if (_isBrokenTimeout)
{
if (_recvTimeout.totalMicroseconds() != 0)
{
if (!poll(_recvTimeout, SELECT_READ))
throw TimeoutException();
}
}
checkBrokenTimeout();
int rc;
do
{
@@ -346,12 +385,76 @@ int SocketImpl::receiveBytes(void* buffer, int length, int flags)
}
int SocketImpl::receiveBytes(SocketBufVec& buffers, int flags)
{
checkBrokenTimeout();
int rc = 0;
do
{
if (_sockfd == POCO_INVALID_SOCKET) throw InvalidSocketException();
#if defined(POCO_OS_FAMILY_WINDOWS)
DWORD recvd = 0;
DWORD dwFlags = static_cast<DWORD>(flags);
rc = WSARecv(_sockfd, &buffers[0], static_cast<DWORD>(buffers.size()),
&recvd, &dwFlags, 0, 0);
if (rc == SOCKET_ERROR) error();
rc = recvd;
#elif defined(POCO_OS_FAMILY_UNIX)
rc = readv(_sockfd, &buffers[0], static_cast<int>(buffers.size()));
#endif
}
while (_blocking && rc < 0 && lastError() == POCO_EINTR);
if (rc < 0)
{
int err = lastError();
if (err == POCO_EAGAIN && !_blocking)
;
else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
throw TimeoutException(err);
else
error(err);
}
return rc;
}
int SocketImpl::receiveBytes(Poco::Buffer<char>& buffer, int flags, const Poco::Timespan& timeout)
{
int rc = 0;
if (poll(timeout, SELECT_READ))
{
int avail = available();
if (buffer.size() < avail) buffer.resize(avail);
do
{
if (_sockfd == POCO_INVALID_SOCKET) throw InvalidSocketException();
rc = ::recv(_sockfd, buffer.begin(), static_cast<int>(buffer.size()), flags);
}
while (_blocking && rc < 0 && lastError() == POCO_EINTR);
if (rc < 0)
{
int err = lastError();
if (err == POCO_EAGAIN && !_blocking)
;
else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
throw TimeoutException(err);
else
error(err);
}
if (rc < buffer.size()) buffer.resize(rc);
}
return rc;
}
int SocketImpl::sendTo(const void* buffer, int length, const SocketAddress& address, int flags)
{
int rc;
do
{
if (_sockfd == POCO_INVALID_SOCKET) throw InvalidSocketException();
if (_sockfd == POCO_INVALID_SOCKET) init(address.af());
#if defined(POCO_VXWORKS)
rc = ::sendto(_sockfd, (char*) buffer, length, flags, (sockaddr*) address.addr(), address.length());
#else
@@ -364,32 +467,121 @@ int SocketImpl::sendTo(const void* buffer, int length, const SocketAddress& addr
}
int SocketImpl::sendTo(const SocketBufVec& buffers, const SocketAddress& address, int flags)
{
int rc = 0;
do
{
if (_sockfd == POCO_INVALID_SOCKET) init(address.af());
#if defined(POCO_OS_FAMILY_WINDOWS)
DWORD sent = 0;
rc = WSASendTo(_sockfd, const_cast<LPWSABUF>(&buffers[0]),
static_cast<DWORD>(buffers.size()), &sent,
static_cast<DWORD>(flags),
address.addr(), address.length(), 0, 0);
if (rc == SOCKET_ERROR) error();
rc = sent;
#elif defined(POCO_OS_FAMILY_UNIX)
struct msghdr msgHdr;
msgHdr.msg_name = const_cast<sockaddr*>(address.addr());
msgHdr.msg_namelen = address.length();
msgHdr.msg_iov = const_cast<iovec*>(&buffers[0]);
msgHdr.msg_iovlen = buffers.size();
msgHdr.msg_control = 0;
msgHdr.msg_controllen = 0;
msgHdr.msg_flags = flags;
rc = sendmsg(_sockfd, &msgHdr, flags);
#endif
}
while (_blocking && rc < 0 && lastError() == POCO_EINTR);
if (rc < 0) error();
return rc;
}
int SocketImpl::receiveFrom(void* buffer, int length, SocketAddress& address, int flags)
{
if (_isBrokenTimeout)
{
if (_recvTimeout.totalMicroseconds() != 0)
{
if (!poll(_recvTimeout, SELECT_READ))
throw TimeoutException();
}
}
char abuffer[SocketAddress::MAX_ADDRESS_LENGTH];
struct sockaddr* pSA = reinterpret_cast<struct sockaddr*>(abuffer);
poco_socklen_t saLen = sizeof(abuffer);
int rc;
do
{
if (_sockfd == POCO_INVALID_SOCKET) throw InvalidSocketException();
rc = ::recvfrom(_sockfd, reinterpret_cast<char*>(buffer), length, flags, pSA, &saLen);
}
while (_blocking && rc < 0 && lastError() == POCO_EINTR);
poco_socklen_t* pSALen = &saLen;
int rc = receiveFrom(buffer, length, &pSA, &pSALen, flags);
if (rc >= 0)
{
address = SocketAddress(pSA, saLen);
}
else
return rc;
}
int SocketImpl::receiveFrom(void* buffer, int length, struct sockaddr** ppSA, poco_socklen_t** ppSALen, int flags)
{
checkBrokenTimeout();
int rc;
do
{
if (_sockfd == POCO_INVALID_SOCKET) throw InvalidSocketException();
rc = ::recvfrom(_sockfd, reinterpret_cast<char*>(buffer), length, flags, *ppSA, *ppSALen);
}
while (_blocking && rc < 0 && lastError() == POCO_EINTR);
if (rc < 0)
{
int err = lastError();
if (err == POCO_EAGAIN && !_blocking)
;
else if (err == POCO_EAGAIN || err == POCO_ETIMEDOUT)
throw TimeoutException(err);
else
error(err);
}
return rc;
}
int SocketImpl::receiveFrom(SocketBufVec& buffers, SocketAddress& address, int flags)
{
char abuffer[SocketAddress::MAX_ADDRESS_LENGTH];
struct sockaddr* pSA = reinterpret_cast<struct sockaddr*>(abuffer);
poco_socklen_t saLen = sizeof(abuffer);
poco_socklen_t* pSALen = &saLen;
int rc = receiveFrom(buffers, &pSA, &pSALen, flags);
if(rc >= 0)
{
address = SocketAddress(pSA, saLen);
}
return rc;
}
int SocketImpl::receiveFrom(SocketBufVec& buffers, struct sockaddr** pSA, poco_socklen_t** ppSALen, int flags)
{
checkBrokenTimeout();
int rc = 0;
do
{
if (_sockfd == POCO_INVALID_SOCKET) throw InvalidSocketException();
#if defined(POCO_OS_FAMILY_WINDOWS)
DWORD recvd = 0;
DWORD dwFlags = static_cast<DWORD>(flags);
rc = WSARecvFrom(_sockfd, &buffers[0], static_cast<DWORD>(buffers.size()),
&recvd, &dwFlags, *pSA, *ppSALen, 0, 0);
if (rc == SOCKET_ERROR) error();
rc = recvd;
#elif defined(POCO_OS_FAMILY_UNIX)
struct msghdr msgHdr;
msgHdr.msg_name = *pSA;
msgHdr.msg_namelen = **ppSALen;
msgHdr.msg_iov = &buffers[0];
msgHdr.msg_iovlen = buffers.size();
msgHdr.msg_control = 0;
msgHdr.msg_controllen = 0;
msgHdr.msg_flags = flags;
rc = recvmsg(_sockfd, &msgHdr, flags);
if (rc >= 0) **ppSALen = msgHdr.msg_namelen;
#endif
}
while (_blocking && rc < 0 && lastError() == POCO_EINTR);
if (rc < 0)
{
int err = lastError();
if (err == POCO_EAGAIN && !_blocking)
@@ -414,7 +606,7 @@ void SocketImpl::sendUrgent(unsigned char data)
int SocketImpl::available()
{
int result;
int result = 0;
ioctl(FIONREAD, result);
return result;
}
@@ -494,8 +686,11 @@ bool SocketImpl::poll(const Poco::Timespan& timeout, int mode)
do
{
Poco::Timestamp start;
#ifdef _WIN32
rc = WSAPoll(&pollBuf, 1, static_cast<INT>(timeout.totalMilliseconds()));
#else
rc = ::poll(&pollBuf, 1, remainingTime.totalMilliseconds());
#endif
if (rc < 0 && lastError() == POCO_EINTR)
{
Poco::Timestamp end;
@@ -939,7 +1134,7 @@ void SocketImpl::initSocket(int af, int type, int proto)
// will crash the process. This only happens on UNIX, and not Linux.
//
// In order to have POCO sockets behave the same across platforms, it is
// best to just ignore SIGPIPE all together.
// best to just ignore SIGPIPE altogether.
setOption(SOL_SOCKET, SO_NOSIGPIPE, 1);
#endif
}