From b581418f2ef37499db0bddb14cc93bd2ad58e61c Mon Sep 17 00:00:00 2001 From: Fred Nicolson Date: Fri, 9 Dec 2016 23:52:16 +0000 Subject: [PATCH] Added support for blocking on multiple sockets simultaneously --- CMakeLists.txt | 2 +- include/NetworkEncoding.h | 2 +- include/Socket.h | 14 +++++- include/SocketSelector.h | 32 ++++++++++++++ include/TcpListener.h | 8 +++- include/TcpSocket.h | 1 + main.cpp | 92 +++++++++++++++++++++++++++------------ src/SocketSelector.cpp | 53 ++++++++++++++++++++++ 8 files changed, 171 insertions(+), 33 deletions(-) create mode 100644 include/SocketSelector.h create mode 100644 src/SocketSelector.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index a4d8866..9e1ced4 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -4,5 +4,5 @@ project(frnetlib) include_directories(include) set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11 -m64 -fPIC -pthread") -set(SOURCE_FILES main.cpp src/TcpSocket.cpp include/TcpSocket.h src/TcpListener.cpp include/TcpListener.h src/Socket.cpp include/Socket.h src/Packet.cpp include/Packet.h include/NetworkEncoding.h) +set(SOURCE_FILES main.cpp src/TcpSocket.cpp include/TcpSocket.h src/TcpListener.cpp include/TcpListener.h src/Socket.cpp include/Socket.h src/Packet.cpp include/Packet.h include/NetworkEncoding.h src/SocketSelector.cpp include/SocketSelector.h) add_executable(frnetlib ${SOURCE_FILES}) \ No newline at end of file diff --git a/include/NetworkEncoding.h b/include/NetworkEncoding.h index 53cd2f0..d6992e1 100644 --- a/include/NetworkEncoding.h +++ b/include/NetworkEncoding.h @@ -62,7 +62,7 @@ inline void *get_sin_addr(struct sockaddr *sa) #include #include #else -#define closesocket(x) close(x) +#define closesocket(x) ::close(x) #define INVALID_SOCKET 0 #define SOCKET_ERROR -1 #include diff --git a/include/Socket.h b/include/Socket.h index 66e0641..c083db8 100644 --- a/include/Socket.h +++ b/include/Socket.h @@ -73,13 +73,23 @@ namespace fr * * @return The string address */ - inline virtual const std::string &get_remote_address() + inline const std::string &get_remote_address() { return remote_address; } + /*! + * Gets the socket descriptor of the object + * + * @return The socket file descriptor + */ + inline int32_t get_socket_descriptor() const + { + return socket_descriptor; + } + protected: - int socket_descriptor; + int32_t socket_descriptor; std::string remote_address; }; } diff --git a/include/SocketSelector.h b/include/SocketSelector.h new file mode 100644 index 0000000..16ef23e --- /dev/null +++ b/include/SocketSelector.h @@ -0,0 +1,32 @@ +// +// Created by fred on 09/12/16. +// + +#ifndef FRNETLIB_SOCKETSELECTOR_H +#define FRNETLIB_SOCKETSELECTOR_H + +#include "NetworkEncoding.h" +#include "Socket.h" +#include "TcpListener.h" + +namespace fr +{ + class SocketSelector + { + public: + SocketSelector() noexcept; + + bool wait(); + void add(const Socket &socket); + bool is_ready(const Socket &socket); + void remove(const Socket &socket); + private: + + fd_set listen_set; + fd_set listen_read; + int32_t max_descriptor; + }; +} + + +#endif //FRNETLIB_SOCKETSELECTOR_H diff --git a/include/TcpListener.h b/include/TcpListener.h index a529df5..2a7cff5 100644 --- a/include/TcpListener.h +++ b/include/TcpListener.h @@ -13,7 +13,7 @@ namespace fr { -class TcpListener +class TcpListener : public Socket { public: /*! @@ -33,7 +33,11 @@ public: Socket::Status accept(TcpSocket &client); private: - int socket_descriptor; + //Stubs + virtual Status send(const Packet &packet){return Socket::Error;} + virtual Status receive(Packet &packet){return Socket::Error;} + virtual void close(){} + virtual Socket::Status connect(const std::string &address, const std::string &port){return Socket::Error;} }; } diff --git a/include/TcpSocket.h b/include/TcpSocket.h index 0de4e55..82046eb 100644 --- a/include/TcpSocket.h +++ b/include/TcpSocket.h @@ -18,6 +18,7 @@ public: TcpSocket() noexcept; virtual ~TcpSocket() noexcept; TcpSocket(TcpSocket &&) noexcept = default; + void operator=(const TcpSocket &other)=delete; /*! * Send a packet through the socket diff --git a/main.cpp b/main.cpp index 0ad89f8..09fae6c 100644 --- a/main.cpp +++ b/main.cpp @@ -2,26 +2,66 @@ #include #include #include +#include #include #include +#include -std::atomic data_sent; void server() { fr::TcpListener listener; - listener.listen("8081"); - - fr::TcpSocket socket; - listener.accept(socket); - - while(socket.connected()) + if(listener.listen("8081") != fr::Socket::Success) { - fr::Packet packet; - socket.receive(packet); + std::cout << "Failed to listen to port" << std::endl; + return; + } - std::string message; - packet >> message; - data_sent += 0x100000; + fr::SocketSelector selector; + std::vector> clients; + + selector.add(listener); + + while(selector.wait()) + { + if(selector.is_ready(listener)) + { + clients.emplace_back(new fr::TcpSocket()); + if(listener.accept(*clients.back()) != fr::Socket::Success) + { + clients.pop_back(); + continue; + } + + selector.add(*clients.back()); + std::cout << "Got new connection from: " << clients.back()->get_remote_address() << std::endl; + } + else + { + for(auto iter = clients.begin(); iter != clients.end();) + { + if(selector.is_ready(**iter)) + { + fr::Packet packet; + if((*iter)->receive(packet) == fr::Socket::Success) + { + std::string message; + packet >> message; + std::cout << (*iter)->get_remote_address() << " sent: " << message << std::endl; + iter++; + } + else + { + std::cout << (*iter)->get_remote_address() << " has disconnected." << std::endl; + selector.remove(*iter->get()); + iter = clients.erase(iter); + } + } + else + { + iter++; + } + } + } } } @@ -30,33 +70,31 @@ void client() fr::TcpSocket socket; socket.connect("127.0.0.1", "8081"); - std::string buffer(0x100000, 'c'); + fr::TcpSocket socket2; + socket2.connect("127.0.0.1", "8081"); fr::Packet packet; - packet << buffer; - while(true) - { - socket.send(packet); - } + packet << "Hello, World! - From socket 1"; + socket.send(packet); + + packet.clear(); + packet << "Hello, world! - From socket 2"; + socket2.send(packet); + return; + } int main() { - data_sent = 0; std::thread t1(&server); std::this_thread::sleep_for(std::chrono::milliseconds(100)); auto start = std::chrono::system_clock::now(); std::thread t2(&client); - while(true) - { - std::this_thread::sleep_for(std::chrono::seconds(1)); - - uint64_t mb_sent = data_sent / 0x100000; - std::cout << "Transfer speed: " << mb_sent / (uint64_t)std::chrono::duration(std::chrono::system_clock::now() - start).count() << "MBps" << std::endl; - } - t1.join(); + + + return 0; } \ No newline at end of file diff --git a/src/SocketSelector.cpp b/src/SocketSelector.cpp new file mode 100644 index 0000000..2b24644 --- /dev/null +++ b/src/SocketSelector.cpp @@ -0,0 +1,53 @@ +// +// Created by fred on 09/12/16. +// + +#include "SocketSelector.h" + +namespace fr +{ + + SocketSelector::SocketSelector() noexcept + { + //Zero out sets + FD_ZERO(&listen_set); + FD_ZERO(&listen_read); + + max_descriptor = 0; + } + + void SocketSelector::add(const Socket &socket) + { + //Add it to the set + FD_SET(socket.get_socket_descriptor(), &listen_set); + + if(socket.get_socket_descriptor() > max_descriptor) + max_descriptor = socket.get_socket_descriptor(); + } + + bool SocketSelector::wait() + { + listen_read = listen_set; + if(select(max_descriptor + 1, &listen_read, NULL, NULL, NULL) == SOCKET_ERROR) + { + //Check that we've not just timed out + if(errno == EAGAIN || errno == EWOULDBLOCK) + return true; + + //Oops + throw std::logic_error("select() returned -1"); + } + + return true; + } + + void SocketSelector::remove(const Socket &socket) + { + FD_CLR(socket.get_socket_descriptor(), &listen_set); + } + + bool SocketSelector::is_ready(const Socket &socket) + { + return (FD_ISSET(socket.get_socket_descriptor(), &listen_read)); + } +} \ No newline at end of file