sid
/
NazaraEngine
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Network/SocketPoller: Makes it possible to watch read and write states

This commit is contained in:
Lynix 2017-05-16 09:07:28 +02:00
parent 7425993d2d
commit 65d3b59e03
9 changed files with 192 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
include/Nazara/Network/Enums.hpp
View File

@ -7,7 +7,7 @@
#ifndef NAZARA_ENUMS_NETWORK_HPP
#define NAZARA_ENUMS_NETWORK_HPP
#include <Nazara/Prerequesites.hpp>
#include <Nazara/Core/Flags.hpp>
namespace Nz
{
@ -91,6 +91,23 @@ namespace Nz
SocketError_Max = SocketError_UnreachableHost
};
enum SocketPollEvent
{
SocketPollEvent_Read, //< One or more sockets is ready for a read operation
SocketPollEvent_Write, //< One or more sockets is ready for a write operation
SocketPollEvent_Max = SocketPollEvent_Write
};
template<>
struct EnumAsFlags<SocketPollEvent>
{
static constexpr bool value = true;
static constexpr int max = SocketPollEvent_Max;
};
using SocketPollEventFlags = Flags<SocketPollEvent>;
enum SocketState
{
SocketState_Bound, //< The socket is currently bound

7
include/Nazara/Network/SocketPoller.hpp
View File

@ -24,10 +24,11 @@ namespace Nz
void Clear();
bool IsReady(const AbstractSocket& socket) const;
bool IsReadyToRead(const AbstractSocket& socket) const;
bool IsReadyToWrite(const AbstractSocket& socket) const;
bool IsRegistered(const AbstractSocket& socket) const;
bool RegisterSocket(AbstractSocket& socket);
bool RegisterSocket(AbstractSocket& socket, SocketPollEventFlags eventFlags);
void UnregisterSocket(AbstractSocket& socket);
bool Wait(UInt64 msTimeout);
@ -41,4 +42,4 @@ namespace Nz
#include <Nazara/Network/SocketPoller.inl>
#endif // NAZARA_SOCKETPOLLER_HPP
#endif // NAZARA_SOCKETPOLLER_HPP

45
src/Nazara/Network/Linux/SocketPollerImpl.cpp
View File

@ -23,13 +23,19 @@ namespace Nz
void SocketPollerImpl::Clear()
{
m_activeSockets.clear();
m_readyToReadSockets.clear();
m_readyToWriteSockets.clear();
m_sockets.clear();
}
bool SocketPollerImpl::IsReady(SocketHandle socket) const
bool SocketPollerImpl::IsReadyToRead(SocketHandle socket) const
{
return m_activeSockets.count(socket) != 0;
return m_readyToReadSockets.count(socket) != 0;
}
bool SocketPollerImpl::IsReadyToWrite(SocketHandle socket) const
{
return m_readyToWriteSockets.count(socket) != 0;
}
bool SocketPollerImpl::IsRegistered(SocketHandle socket) const
@ -37,15 +43,21 @@ namespace Nz
return m_sockets.count(socket) != 0;
}
bool SocketPollerImpl::RegisterSocket(SocketHandle socket)
bool SocketPollerImpl::RegisterSocket(SocketHandle socket, SocketPollEventFlags eventFlags)
{
NazaraAssert(!IsRegistered(socket), "Socket is already registered");
epoll_event event;
event.events = EPOLLIN;
event.data.fd = socket;
epoll_event entry;
entry.events = 0;
entry.data.fd = socket;
if (epoll_ctl(m_handle, EPOLL_CTL_ADD, socket, &event) != 0)
if (eventFlags & SocketPollEvent_Read)
entry.events |= EPOLLIN;
if (eventFlags & SocketPollEvent_Write)
entry.events |= EPOLLOUT;
if (epoll_ctl(m_handle, EPOLL_CTL_ADD, socket, &entry) != 0)
{
NazaraError("Failed to add socket to epoll structure (errno " + String::Number(errno) + ": " + Error::GetLastSystemError() + ')');
return false;
@ -60,7 +72,8 @@ namespace Nz
{
NazaraAssert(IsRegistered(socket), "Socket is not registered");
m_activeSockets.erase(socket);
m_readyToReadSockets.erase(socket);
m_readyToWriteSockets.erase(socket);
m_sockets.erase(socket);
if (epoll_ctl(m_handle, EPOLL_CTL_DEL, socket, nullptr) != 0)
@ -84,21 +97,27 @@ namespace Nz
return 0;
}
m_activeSockets.clear();
m_readyToReadSockets.clear();
m_readyToWriteSockets.clear();
if (activeSockets > 0)
{
int socketCount = activeSockets;
for (int i = 0; i < socketCount; ++i)
{
if (m_events[i].events & (EPOLLIN | EPOLLHUP | EPOLLERR))
if (m_events[i].events & (EPOLLIN | EPOLLOUT | EPOLLHUP | EPOLLERR))
{
m_activeSockets.insert(m_events[i].data.fd);
if (m_events[i].events & (EPOLLIN | EPOLLHUP | EPOLLERR))
m_readyToReadSockets.insert(m_events[i].data.fd);
if (m_events[i].events & (EPOLLOUT | EPOLLERR))
m_readyToWriteSockets.insert(m_events[i].data.fd);
if (m_events[i].events & EPOLLERR)
NazaraWarning("Descriptor " + String::Number(m_events[i].data.fd) + " was returned by epoll with EPOLLERR status");
}
else
{
NazaraWarning("Descriptor " + String::Number(m_events[i].data.fd) + " was returned by epoll without EPOLLIN (events: 0x" + String::Number(m_events[i].events, 16) + ')');
NazaraWarning("Descriptor " + String::Number(m_events[i].data.fd) + " was returned by epoll without EPOLLIN nor EPOLLOUT flags (events: 0x" + String::Number(m_events[i].events, 16) + ')');
activeSockets--;
}
}

8
src/Nazara/Network/Linux/SocketPollerImpl.hpp
View File

@ -23,16 +23,18 @@ namespace Nz
void Clear();
bool IsReady(SocketHandle socket) const;
bool IsReadyToRead(SocketHandle socket) const;
bool IsReadyToWrite(SocketHandle socket) const;
bool IsRegistered(SocketHandle socket) const;
bool RegisterSocket(SocketHandle socket);
bool RegisterSocket(SocketHandle socket, SocketPollEventFlags eventFlags);
void UnregisterSocket(SocketHandle socket);
int Wait(UInt64 msTimeout, SocketError* error);
private:
std::unordered_set<SocketHandle> m_activeSockets;
std::unordered_set<SocketHandle> m_readyToReadSockets;
std::unordered_set<SocketHandle> m_readyToWriteSockets;
std::unordered_set<SocketHandle> m_sockets;
std::vector<epoll_event> m_events;
int m_handle;

46
src/Nazara/Network/Posix/SocketPollerImpl.cpp
View File

@ -10,14 +10,20 @@ namespace Nz
{
void SocketPollerImpl::Clear()
{
m_activeSockets.clear();
m_readyToReadSockets.clear();
m_readyToWriteSockets.clear();
m_allSockets.clear();
m_sockets.clear();
}
bool SocketPollerImpl::IsReady(SocketHandle socket) const
bool SocketPollerImpl::IsReadyToRead(SocketHandle socket) const
{
return m_activeSockets.count(socket) != 0;
return m_readyToReadSockets.count(socket) != 0;
}
bool SocketPollerImpl::IsReadyToWrite(SocketHandle socket) const
{
return m_readyToWriteSockets.count(socket) != 0;
}
bool SocketPollerImpl::IsRegistered(SocketHandle socket) const
@ -25,16 +31,22 @@ namespace Nz
return m_allSockets.count(socket) != 0;
}
bool SocketPollerImpl::RegisterSocket(SocketHandle socket)
bool SocketPollerImpl::RegisterSocket(SocketHandle socket, SocketPollEventFlags eventFlags)
{
NazaraAssert(!IsRegistered(socket), "Socket is already registered");
PollSocket entry = {
socket,
POLLRDNORM,
0,
0
};
if (eventFlags & SocketPollEvent_Read)
entry.events |= POLLRDNORM;
if (eventFlags & SocketPollEvent_Write)
entry.events |= POLLWRNORM;
m_allSockets[socket] = m_sockets.size();
m_sockets.emplace_back(entry);
@ -57,10 +69,11 @@ namespace Nz
// Now move it properly (lastElement is invalid after the following line) and pop it
m_sockets[entry] = std::move(m_sockets.back());
}
m_sockets.pop_back();
m_activeSockets.erase(socket);
m_allSockets.erase(socket);
m_readyToReadSockets.erase(socket);
m_readyToWriteSockets.erase(socket);
}
int SocketPollerImpl::Wait(UInt64 msTimeout, SocketError* error)
@ -68,20 +81,25 @@ namespace Nz
int activeSockets;
// Reset status of sockets
for (PollSocket& entry : m_sockets)
entry.revents = 0;
activeSockets = SocketImpl::Poll(m_sockets.data(), m_sockets.size(), static_cast<int>(msTimeout), error);
m_activeSockets.clear();
if (activeSockets > 0)
m_readyToReadSockets.clear();
m_readyToWriteSockets.clear();
if (activeSockets > 0U)
{
int socketRemaining = activeSockets;
for (PollSocket& entry : m_sockets)
{
if (entry.revents & POLLRDNORM)
if (entry.revents != 0)
{
m_activeSockets.insert(entry.fd);
if (entry.revents & POLLRDNORM)
m_readyToReadSockets.insert(entry.fd);
if (entry.revents & POLLWRNORM)
m_readyToWriteSockets.insert(entry.fd);
entry.revents = 0;
if (--socketRemaining == 0)
break;
}

5
src/Nazara/Network/Posix/SocketPollerImpl.hpp
View File

@ -27,13 +27,14 @@ namespace Nz
bool IsReady(SocketHandle socket) const;
bool IsRegistered(SocketHandle socket) const;
bool RegisterSocket(SocketHandle socket);
bool RegisterSocket(SocketHandle socket, SocketPollEventFlags eventFlags);
void UnregisterSocket(SocketHandle socket);
int Wait(UInt64 msTimeout, SocketError* error);
private:
std::unordered_set<SocketHandle> m_activeSockets;
std::unordered_set<SocketHandle> m_readyToReadSockets;
std::unordered_set<SocketHandle> m_readyToWriteSockets;
std::unordered_map<SocketHandle, std::size_t> m_allSockets;
std::vector<PollSocket> m_sockets;
};

43
src/Nazara/Network/SocketPoller.cpp
View File

@ -57,12 +57,13 @@ namespace Nz
/*!
* \brief Checks if a specific socket is ready to read data
*
* This function allows you to read the results of the last Wait operation and if a specific socket is ready.
* This function allows you to read the results of the last Wait operation and if a specific socket is ready to read (has incoming data).
*
* A socket in the ready state (with the exception of TcpServer) has incoming data and can be read without blocking.
* A socket in the ready to read state (with the exception of TcpServer) has incoming data and can be read without blocking.
*
* \remark When used on a TcpServer socket, this function returns true if the server is ready to accept a new client.
* \remark You must call Wait before using this function in order to refresh the state.
* \remark You must call Wait before using this function in order to refresh the read state.
* \remark A socket must be registered with SocketPollerEvent_Read event flag for its read state to be watched
* \remark A TcpServer socket becomes ready to read when it is ready to accept a new client.
*
* \param socket Reference to the socket to check
*
@ -70,11 +71,32 @@ namespace Nz
*
* \see Wait
*/
bool SocketPoller::IsReady(const AbstractSocket& socket) const
bool SocketPoller::IsReadyToRead(const AbstractSocket& socket) const
{
NazaraAssert(IsRegistered(socket), "Socket is not registered in the poller");
return m_impl->IsReady(socket.GetNativeHandle());
return m_impl->IsReadyToRead(socket.GetNativeHandle());
}
/*!
* \brief Checks if a specific socket is ready to write data
*
* This function allows you to read the results of the last Wait operation and if a specific socket is ready to write (can be written to without blocking).
*
* \remark You must call Wait before using this function in order to refresh the read state.
* \remark A socket must be registered with SocketPollerEvent_Write event flag for its read state to be watched
*
* \param socket Reference to the socket to check
*
* \return True if the socket is available for writing without blocking, false otherwise
*
* \see Wait
*/
bool SocketPoller::IsReadyToWrite(const AbstractSocket& socket) const
{
NazaraAssert(IsRegistered(socket), "Socket is not registered in the poller");
return m_impl->IsReadyToWrite(socket.GetNativeHandle());
}
/*!
@ -97,7 +119,7 @@ namespace Nz
/*!
* \brief Register a socket in the SocketPoller
*
* A registered socket is part of the SocketPoller and will be checked by the next Wait operations.
* A registered socket is part of the SocketPoller and will be checked by the next Wait operations according to the event flags passed when registered.
*
* The SocketPoller keeps a reference to the internal handle of registered socket, which should not be freed while it is registered in the SocketPooler.
*
@ -107,17 +129,18 @@ namespace Nz
* \remark The socket should not be freed while it is registered in the SocketPooler.
*
* \param socket Reference to the socket to register
* \param eventFlags Socket events to watch
*
* \return True if the socket is registered, false otherwise
*
* \see IsRegistered
* \see UnregisterSocket
*/
bool SocketPoller::RegisterSocket(AbstractSocket& socket)
bool SocketPoller::RegisterSocket(AbstractSocket& socket, SocketPollEventFlags eventFlags)
{
NazaraAssert(!IsRegistered(socket), "This socket is already registered in this SocketPoller");
return m_impl->RegisterSocket(socket.GetNativeHandle());
return m_impl->RegisterSocket(socket.GetNativeHandle(), eventFlags);
}
/*!
@ -145,7 +168,7 @@ namespace Nz
* \brief Wait until any registered socket switches to a ready state.
*
* Waits a specific/undetermined amount of time until at least one socket part of the SocketPoller becomes ready.
* To query the ready state of the registered socket, use the IsReady function.
* To query the ready state of the registered socket, use the IsReadyToRead or IsReadyToWrite functions.
*
* \param msTimeout Maximum time to wait in milliseconds, 0 for infinity
*

96
src/Nazara/Network/Win32/SocketPollerImpl.cpp
View File

@ -10,29 +10,43 @@ namespace Nz
SocketPollerImpl::SocketPollerImpl()
{
#if !NAZARA_NETWORK_POLL_SUPPORT
FD_ZERO(&m_activeSockets);
FD_ZERO(&m_sockets);
FD_ZERO(&m_readSockets);
FD_ZERO(&m_readyToReadSockets);
FD_ZERO(&m_readyToWriteSockets);
FD_ZERO(&m_writeSockets);
#endif
}
void SocketPollerImpl::Clear()
{
#if NAZARA_NETWORK_POLL_SUPPORT
m_activeSockets.clear();
m_allSockets.clear();
m_readyToReadSockets.clear();
m_readyToWriteSockets.clear();
m_sockets.clear();
#else
FD_ZERO(&m_activeSockets);
FD_ZERO(&m_sockets);
FD_ZERO(&m_readSockets);
FD_ZERO(&m_readyToReadSockets);
FD_ZERO(&m_readyToWriteSockets);
FD_ZERO(&m_writeSockets);
#endif
}
bool SocketPollerImpl::IsReady(SocketHandle socket) const
bool SocketPollerImpl::IsReadyToRead(SocketHandle socket) const
{
#if NAZARA_NETWORK_POLL_SUPPORT
return m_activeSockets.count(socket) != 0;
return m_readyToReadSockets.count(socket) != 0;
#else
return FD_ISSET(socket, &m_activeSockets) != 0;
return FD_ISSET(socket, &m_readyToReadSockets) != 0;
#endif
}
bool SocketPollerImpl::IsReadyToWrite(SocketHandle socket) const
{
#if NAZARA_NETWORK_POLL_SUPPORT
return m_readyToWriteSockets.count(socket) != 0;
#else
return FD_ISSET(socket, &m_readyToWriteSockets) != 0;
#endif
}
@ -41,31 +55,45 @@ namespace Nz
#if NAZARA_NETWORK_POLL_SUPPORT
return m_allSockets.count(socket) != 0;
#else
return FD_ISSET(socket, &m_sockets) != 0;
return FD_ISSET(socket, &m_readSockets) != 0 ||
FD_ISSET(socket, &m_writeSockets) != 0;
#endif
}
bool SocketPollerImpl::RegisterSocket(SocketHandle socket)
bool SocketPollerImpl::RegisterSocket(SocketHandle socket, SocketPollEventFlags eventFlags)
{
NazaraAssert(!IsRegistered(socket), "Socket is already registered");
#if NAZARA_NETWORK_POLL_SUPPORT
PollSocket entry = {
socket,
POLLRDNORM,
0,
0
};
if (eventFlags & SocketPollEvent_Read)
entry.events |= POLLRDNORM;
if (eventFlags & SocketPollEvent_Write)
entry.events |= POLLWRNORM;
m_allSockets[socket] = m_sockets.size();
m_sockets.emplace_back(entry);
#else
if (m_sockets.fd_count > FD_SETSIZE)
for (std::size_t i = 0; i < 2; ++i)
{
NazaraError("Socket count exceeding FD_SETSIZE (" + String::Number(FD_SETSIZE) + ")");
return false;
}
if (eventFlags & ((i == 0) ? SocketPollEvent_Read : SocketPollEvent_Write) == 0)
continue;
FD_SET(socket, &m_sockets);
fd_set& targetSet = (i == 0) ? m_readSockets : m_writeSockets;
if (targetSet.fd_count > FD_SETSIZE)
{
NazaraError("Socket count exceeding hard-coded FD_SETSIZE (" + String::Number(FD_SETSIZE) + ")");
return false;
}
FD_SET(socket, &targetSet);
}
#endif
return true;
@ -88,13 +116,16 @@ namespace Nz
// Now move it properly (lastElement is invalid after the following line) and pop it
m_sockets[entry] = std::move(m_sockets.back());
}
m_sockets.pop_back();
m_activeSockets.erase(socket);
m_allSockets.erase(socket);
m_readyToReadSockets.erase(socket);
m_readyToWriteSockets.erase(socket);
#else
FD_CLR(socket, &m_activeSockets);
FD_CLR(socket, &m_sockets);
FD_CLR(socket, &m_readSockets);
FD_CLR(socket, &m_readyToReadSockets);
FD_CLR(socket, &m_readyToWriteSockets);
FD_CLR(socket, &m_writeSockets);
#endif
}
@ -103,35 +134,18 @@ namespace Nz
int activeSockets;
#if NAZARA_NETWORK_POLL_SUPPORT
// Reset status of sockets
for (PollSocket& entry : m_sockets)
entry.revents = 0;
activeSockets = SocketImpl::Poll(m_sockets.data(), m_sockets.size(), static_cast<int>(msTimeout), error);
x
m_activeSockets.clear();
if (activeSockets > 0U)
{
int socketRemaining = activeSockets;
for (PollSocket& entry : m_sockets)
{
if (entry.revents & POLLRDNORM)
{
m_activeSockets.insert(entry.fd);
if (--socketRemaining == 0)
break;
}
}
}
#else
m_activeSockets = m_sockets;
m_readyToReadSockets = m_readSockets;
m_readyToWriteSockets = m_writeSockets;
timeval tv;
tv.tv_sec = static_cast<long>(msTimeout / 1000ULL);
tv.tv_usec = static_cast<long>((msTimeout % 1000ULL) * 1000ULL);
activeSockets = ::select(0xDEADBEEF, &m_activeSockets, nullptr, nullptr, (msTimeout > 0) ? &tv : nullptr); //< The first argument is ignored on Windows
activeSockets = ::select(0xDEADBEEF, &m_readyToReadSockets, &m_readyToWriteSockets, nullptr, (msTimeout > 0) ? &tv : nullptr); //< The first argument is ignored on Windows
if (activeSockets == SOCKET_ERROR)
{
if (error)

16
src/Nazara/Network/Win32/SocketPollerImpl.hpp
View File

@ -25,24 +25,28 @@ namespace Nz
void Clear();
bool IsReady(SocketHandle socket) const;
bool IsReadyToRead(SocketHandle socket) const;
bool IsReadyToWrite(SocketHandle socket) const;
bool IsRegistered(SocketHandle socket) const;
bool RegisterSocket(SocketHandle socket);
bool RegisterSocket(SocketHandle socket, SocketPollEventFlags eventFlags);
void UnregisterSocket(SocketHandle socket);
int Wait(UInt64 msTimeout, SocketError* error);
private:
#if NAZARA_NETWORK_POLL_SUPPORT
std::unordered_set<SocketHandle> m_activeSockets;
std::unordered_set<SocketHandle> m_readyToReadSockets;
std::unordered_set<SocketHandle> m_readyToWriteSockets;
std::unordered_map<SocketHandle, std::size_t> m_allSockets;
std::vector<PollSocket> m_sockets;
#else
fd_set m_sockets;
fd_set m_activeSockets;
fd_set m_readSockets;
fd_set m_readyToReadSockets;
fd_set m_readyToWriteSockets;
fd_set m_writeSockets;
#endif
};
}
#endif // NAZARA_SOCKETPOLLERIMPL_HPP
#endif // NAZARA_SOCKETPOLLERIMPL_HPP