kapitan/IXWebSocket
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

linux fix / linux still use event fd for now

Browse Source
This commit is contained in:
Benjamin Sergeant 2019-03-13 17:23:05 -07:00
parent d1a7b9a985
commit 922d58eb59
10 changed files with 107 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
docker/Dockerfile.debian
View File

@ -12,11 +12,19 @@ RUN apt-get -y install libz-dev
RUN apt-get -y install vim
RUN apt-get -y install make
RUN apt-get -y install cmake
RUN apt-get -y install curl
RUN apt-get -y install python
# debian strech cmake is too old for building with Docker
COPY makefile .
RUN ["make", "install_cmake_for_linux"]
COPY . .
WORKDIR ws
RUN ["sh", "docker_build.sh"]
ARG CMAKE_BIN_PATH=/tmp/cmake/cmake-3.14.0-rc4-Linux-x86_64/bin
ENV PATH="${CMAKE_BIN_PATH}:${PATH}"
# RUN ["make"]
EXPOSE 8765
CMD ["/ws/ws", "transfer", "--port", "8765", "--hostname", "0.0.0.0"]
CMD ["/ws/ws", "transfer", "--port", "8765", "--host", "0.0.0.0"]

57
ixwebsocket/IXEventFd.cpp
View File

@ -17,6 +17,8 @@
// cf Android/Kernel table here
// https://android.stackexchange.com/questions/51651/which-android-runs-which-linux-kernel
//
// On macOS we use UNIX pipes to wake up select.
//
#include "IXEventFd.h"
@ -24,17 +26,24 @@
# include <sys/eventfd.h>
#endif
#ifndef _WIN32
#include <unistd.h> // for write
#endif
#include <fcntl.h>
namespace ix
{
EventFd::EventFd() :
_eventfd(-1)
EventFd::EventFd()
{
#ifdef __linux__
_eventfd = -1;
_eventfd = eventfd(0, 0);
fcntl(_eventfd, F_SETFL, O_NONBLOCK);
#else
_fildes[0] = -1;
_fildes[1] = -1;
pipe(_fildes);
fcntl(_fildes[0], F_SETFL, O_NONBLOCK);
fcntl(_fildes[1], F_SETFL, O_NONBLOCK);
#endif
}
@ -42,22 +51,44 @@ namespace ix
{
#ifdef __linux__
::close(_eventfd);
#else
::close(_fildes[0]);
::close(_fildes[1]);
_fildes[0] = -1;
_fildes[1] = -1;
#endif
}
bool EventFd::notify()
bool EventFd::notify(uint64_t value)
{
#if defined(__linux__)
if (_eventfd == -1) return false;
int fd;
// select will wake up when a non-zero value is written to our eventfd
uint64_t value = 1;
#if defined(__linux__)
fd = _eventfd;
#else
// File descriptor at index 1 in _fildes is the write end of the pipe
fd = _fildes[1];
#endif
if (fd == -1) return false;
// we should write 8 bytes for an uint64_t
return write(_eventfd, &value, sizeof(value)) == 8;
return write(fd, &value, sizeof(value)) == 8;
}
// TODO: return max uint64_t for errors ?
uint64_t EventFd::read()
{
int fd;
#if defined(__linux__)
fd = _eventfd;
#else
return true;
fd = _fildes[0];
#endif
uint64_t value = 0;
::read(fd, &value, sizeof(value));
return value;
}
bool EventFd::clear()
@ -77,6 +108,10 @@ namespace ix
int EventFd::getFd()
{
#if defined(__linux__)
return _eventfd;
#else
return _fildes[0];
#endif
}
}

12
ixwebsocket/IXEventFd.h
View File

@ -6,6 +6,8 @@
#pragma once
#include <stdint.h>
namespace ix
{
class EventFd {
@ -13,11 +15,19 @@ namespace ix
EventFd();
virtual ~EventFd();
bool notify();
bool notify(uint64_t value);
bool clear();
uint64_t read();
int getFd();
private:
#if defined(__linux__)
int _eventfd;
#else
// Store file descriptors used by the communication pipe. Communication
// happens between a control thread and a background thread, which is
// blocked on select.
int _fildes[2];
#endif
};
}

39
ixwebsocket/IXSocket.cpp
View File

@ -23,15 +23,14 @@ namespace ix
{
const int Socket::kDefaultPollNoTimeout = -1; // No poll timeout by default
const int Socket::kDefaultPollTimeout = kDefaultPollNoTimeout;
const int Socket::kSendRequest = 1;
const int Socket::kCloseRequest = 2;
const uint8_t Socket::kSendRequest = 1;
const uint8_t Socket::kCloseRequest = 2;
constexpr size_t Socket::kChunkSize;
Socket::Socket(int fd) :
_sockfd(fd)
{
_fildes[0] = -1;
_fildes[1] = -1;
;
}
Socket::~Socket()
@ -59,9 +58,10 @@ namespace ix
FD_SET(_sockfd, &rfds);
// File descriptor at index 0 in _fildes is the read end of the pipe
if (_fildes[0] != -1)
int eventfd = _eventfd.getFd();
if (eventfd != -1)
{
FD_SET(_fildes[0], &rfds);
FD_SET(eventfd, &rfds);
}
struct timeval timeout;
@ -70,7 +70,7 @@ namespace ix
// Compute the highest fd.
int sockfd = _sockfd;
int nfds = (std::max)(sockfd, _fildes[0]);
int nfds = (std::max)(sockfd, eventfd);
int ret = ::select(nfds + 1, &rfds, nullptr, nullptr,
(timeoutSecs < 0) ? nullptr : &timeout);
@ -84,10 +84,9 @@ namespace ix
{
pollResult = PollResultType_Timeout;
}
else if (_fildes[0] != -1 && FD_ISSET(_fildes[0], &rfds))
else if (eventfd != -1 && FD_ISSET(eventfd, &rfds))
{
uint64_t value = 0;
read(_fildes[0], &value, sizeof(value));
uint8_t value = _eventfd.read();
if (value == kSendRequest)
{
@ -102,14 +101,10 @@ namespace ix
return pollResult;
}
// Wake up from poll/select by writing to the pipe which is is watched by select
bool Socket::wakeUpFromPoll(int wakeUpCode)
// Wake up from poll/select by writing to the pipe which is watched by select
bool Socket::wakeUpFromPoll(uint8_t wakeUpCode)
{
// File descriptor at index 1 in _fildes is the write end of the pipe
if (_fildes[1] == -1) return false;
int value = wakeUpCode;
return ::write(_fildes[1], &value, sizeof(value)) == 4;
return _eventfd.notify(wakeUpCode);
}
bool Socket::connect(const std::string& host,
@ -119,10 +114,7 @@ namespace ix
{
std::lock_guard<std::mutex> lock(_socketMutex);
if (pipe(_fildes) < 0) return false;
fcntl(_fildes[0], F_SETFL, O_NONBLOCK);
fcntl(_fildes[1], F_SETFL, O_NONBLOCK);
if (!_eventfd.clear()) return false;
_sockfd = SocketConnect::connect(host, port, errMsg, isCancellationRequested);
return _sockfd != -1;
@ -136,11 +128,6 @@ namespace ix
closeSocket(_sockfd);
_sockfd = -1;
::close(_fildes[0]);
::close(_fildes[1]);
_fildes[0] = -1;
_fildes[1] = -1;
}
ssize_t Socket::send(char* buffer, size_t length)

12
ixwebsocket/IXSocket.h
View File

@ -20,6 +20,7 @@ typedef SSIZE_T ssize_t;
#include "IXCancellationRequest.h"
#include "IXProgressCallback.h"
#include "IXEventFd.h"
namespace ix
{
@ -44,7 +45,7 @@ namespace ix
PollResultType select(int timeoutSecs, int timeoutMs);
virtual void poll(const OnPollCallback& onPollCallback,
int timeoutSecs = kDefaultPollTimeout);
virtual bool wakeUpFromPoll(int wakeUpCode);
virtual bool wakeUpFromPoll(uint8_t wakeUpCode);
// Virtual methods
virtual bool connect(const std::string& url,
@ -76,8 +77,8 @@ namespace ix
static void cleanup(); // Required on Windows to cleanup WinSocket
// Used as special codes for pipe communication
static const int kSendRequest;
static const int kCloseRequest;
static const uint8_t kSendRequest;
static const uint8_t kCloseRequest;
protected:
void closeSocket(int fd);
@ -93,9 +94,6 @@ namespace ix
std::vector<uint8_t> _readBuffer;
static constexpr size_t kChunkSize = 1 << 15;
// Store file descriptors used by the communication pipe. Communication
// happens between a control thread and a background thread, which is
// blocked on select.
int _fildes[2];
EventFd _eventfd;
};
}

8
makefile
View File

@ -8,10 +8,10 @@ brew:
.PHONY: docker
docker:
docker build -t broadcast_server:latest .
docker build -t ws:latest .
run:
docker run --cap-add sys_ptrace -it broadcast_server:latest bash
docker run --cap-add sys_ptrace -it ws:latest
# this is helpful to remove trailing whitespaces
trail:
@ -43,5 +43,9 @@ rebase_upstream:
git reset --hard upstream/master
git push origin master --force
install_cmake_for_linux:
mkdir -p /tmp/cmake
(cd /tmp/cmake ; curl -L -O https://github.com/Kitware/CMake/releases/download/v3.14.0-rc4/cmake-3.14.0-rc4-Linux-x86_64.tar.gz ; tar zxf cmake-3.14.0-rc4-Linux-x86_64.tar.gz)
.PHONY: test
.PHONY: build

12
ws/ws.cpp
View File

@ -51,6 +51,7 @@ int main(int argc, char** argv)
CLI::App* transferApp = app.add_subcommand("transfer", "Broadcasting server");
transferApp->add_option("--port", port, "Connection url");
transferApp->add_option("--host", hostname, "Hostname");
CLI::App* connectApp = app.add_subcommand("connect", "Connect to a remote server");
connectApp->add_option("url", url, "Connection url")->required();
@ -60,11 +61,12 @@ int main(int argc, char** argv)
chatApp->add_option("user", user, "User name")->required();
CLI::App* echoServerApp = app.add_subcommand("echo_server", "Echo server");
echoServerApp->add_option("--port", port, "Connection url");
echoServerApp->add_option("--port", port, "Port");
echoServerApp->add_option("--host", hostname, "Hostname");
CLI::App* broadcastServerApp = app.add_subcommand("broadcast_server", "Broadcasting server");
broadcastServerApp->add_option("--port", port, "Connection url");
broadcastServerApp->add_option("--hostname", hostname, "Hostname");
broadcastServerApp->add_option("--port", port, "Port");
broadcastServerApp->add_option("--host", hostname, "Hostname");
CLI::App* pingPongApp = app.add_subcommand("ping", "Ping pong");
pingPongApp->add_option("url", url, "Connection url")->required();
@ -90,7 +92,7 @@ int main(int argc, char** argv)
if (app.got_subcommand("transfer"))
{
return ix::ws_transfer_main(port);
return ix::ws_transfer_main(port, hostname);
}
else if (app.got_subcommand("send"))
{
@ -111,7 +113,7 @@ int main(int argc, char** argv)
}
else if (app.got_subcommand("echo_server"))
{
return ix::ws_echo_server_main(port);
return ix::ws_echo_server_main(port, hostname);
}
else if (app.got_subcommand("broadcast_server"))
{

6
ws/ws.h
View File

@ -24,9 +24,9 @@ namespace ix
int ws_ping_pong_main(const std::string& url);
int ws_echo_server_main(int port);
int ws_echo_server_main(int port, const std::string& hostname);
int ws_broadcast_server_main(int port, const std::string& hostname);
int ws_transfer_main(int port, const std::string& hostname);
int ws_chat_main(const std::string& url,
const std::string& user);
@ -36,8 +36,6 @@ namespace ix
int ws_receive_main(const std::string& url,
bool enablePerMessageDeflate);
int ws_transfer_main(int port);
int ws_send_main(const std::string& url,
const std::string& path);
}

6
ws/ws_echo_server.cpp
View File

@ -10,11 +10,11 @@
namespace ix
{
int ws_echo_server_main(int port)
int ws_echo_server_main(int port, const std::string& hostname)
{
std::cout << "Listening on port " << port << std::endl;
std::cout << "Listening on " << hostname << ":" << port << std::endl;
ix::WebSocketServer server(port);
ix::WebSocketServer server(port, hostname);
server.setOnConnectionCallback(
[](std::shared_ptr<ix::WebSocket> webSocket)

6
ws/ws_transfer.cpp
View File

@ -10,11 +10,11 @@
namespace ix
{
int ws_transfer_main(int port)
int ws_transfer_main(int port, const std::string& hostname)
{
std::cout << "Listening on port " << port << std::endl;
std::cout << "Listening on " << hostname << ":" << port << std::endl;
ix::WebSocketServer server(port);
ix::WebSocketServer server(port, hostname);
server.setOnConnectionCallback(
[&server](std::shared_ptr<ix::WebSocket> webSocket)