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

send can fail silently when sending would block (EWOULDBLOCK return for send) (#18)

Browse Source 
* try to use a pipe for communication

* flush send buffer on the background thread

* cleanup

* linux fix / linux still use event fd for now

* cleanup
This commit is contained in:
Benjamin Sergeant
2019-03-13 23:09:45 -07:00
committed by GitHub
parent dedbeb3eab
commit 08c2cdbf1d
19 changed files with 277 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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
};
}

73
ixwebsocket/IXSocket.cpp
View File

@ -23,6 +23,8 @@ namespace ix
{
const int Socket::kDefaultPollNoTimeout = -1; // No poll timeout by default
const int Socket::kDefaultPollTimeout = kDefaultPollNoTimeout;
const uint8_t Socket::kSendRequest = 1;
const uint8_t Socket::kCloseRequest = 2;
constexpr size_t Socket::kChunkSize;
Socket::Socket(int fd) :
@ -44,7 +46,34 @@ namespace ix
return;
}
int ret = select(timeoutSecs, 0);
PollResultType pollResult = select(timeoutSecs, 0);
if (onPollCallback) onPollCallback(pollResult);
}
PollResultType Socket::select(int timeoutSecs, int timeoutMs)
{
fd_set rfds;
FD_ZERO(&rfds);
FD_SET(_sockfd, &rfds);
// File descriptor at index 0 in _fildes is the read end of the pipe
int eventfd = _eventfd.getFd();
if (eventfd != -1)
{
FD_SET(eventfd, &rfds);
}
struct timeval timeout;
timeout.tv_sec = timeoutSecs;
timeout.tv_usec = 1000 * timeoutMs;
// Compute the highest fd.
int sockfd = _sockfd;
int nfds = (std::max)(sockfd, eventfd);
int ret = ::select(nfds + 1, &rfds, nullptr, nullptr,
(timeoutSecs < 0) ? nullptr : &timeout);
PollResultType pollResult = PollResultType_ReadyForRead;
if (ret < 0)
@ -55,35 +84,27 @@ namespace ix
{
pollResult = PollResultType_Timeout;
}
else if (eventfd != -1 && FD_ISSET(eventfd, &rfds))
{
uint8_t value = _eventfd.read();
if (onPollCallback) onPollCallback(pollResult);
if (value == kSendRequest)
{
pollResult = PollResultType_SendRequest;
}
else if (value == kCloseRequest)
{
pollResult = PollResultType_CloseRequest;
}
}
return pollResult;
}
int Socket::select(int timeoutSecs, int timeoutMs)
// Wake up from poll/select by writing to the pipe which is watched by select
bool Socket::wakeUpFromPoll(uint8_t wakeUpCode)
{
fd_set rfds;
FD_ZERO(&rfds);
FD_SET(_sockfd, &rfds);
#ifdef __linux__
FD_SET(_eventfd.getFd(), &rfds);
#endif
struct timeval timeout;
timeout.tv_sec = timeoutSecs;
timeout.tv_usec = 1000 * timeoutMs;
int sockfd = _sockfd;
int nfds = (std::max)(sockfd, _eventfd.getFd());
int ret = ::select(nfds + 1, &rfds, nullptr, nullptr,
(timeoutSecs < 0) ? nullptr : &timeout);
return ret;
}
void Socket::wakeUpFromPoll()
{
// this will wake up the thread blocked on select, only needed on Linux
_eventfd.notify();
return _eventfd.notify(wakeUpCode);
}
bool Socket::connect(const std::string& host,

17
ixwebsocket/IXSocket.h
View File

@ -17,9 +17,9 @@
typedef SSIZE_T ssize_t;
#endif
#include "IXEventFd.h"
#include "IXCancellationRequest.h"
#include "IXProgressCallback.h"
#include "IXEventFd.h"
namespace ix
{
@ -27,7 +27,9 @@ namespace ix
{
PollResultType_ReadyForRead = 0,
PollResultType_Timeout = 1,
PollResultType_Error = 2
PollResultType_Error = 2,
PollResultType_SendRequest = 3,
PollResultType_CloseRequest = 4
};
class Socket {
@ -39,10 +41,10 @@ namespace ix
void configure();
int select(int timeoutSecs, int timeoutMs);
PollResultType select(int timeoutSecs, int timeoutMs);
virtual void poll(const OnPollCallback& onPollCallback,
int timeoutSecs = kDefaultPollTimeout);
virtual void wakeUpFromPoll();
virtual bool wakeUpFromPoll(uint8_t wakeUpCode);
// Virtual methods
virtual bool connect(const std::string& url,
@ -73,12 +75,15 @@ namespace ix
static bool init(); // Required on Windows to initialize WinSocket
static void cleanup(); // Required on Windows to cleanup WinSocket
// Used as special codes for pipe communication
static const uint8_t kSendRequest;
static const uint8_t kCloseRequest;
protected:
void closeSocket(int fd);
std::atomic<int> _sockfd;
std::mutex _socketMutex;
EventFd _eventfd;
private:
static const int kDefaultPollTimeout;
@ -87,5 +92,7 @@ namespace ix
// Buffer for reading from our socket. That buffer is never resized.
std::vector<uint8_t> _readBuffer;
static constexpr size_t kChunkSize = 1 << 15;
EventFd _eventfd;
};
}

5
ixwebsocket/IXWebSocket.cpp
View File

@ -379,4 +379,9 @@ namespace ix
{
_automaticReconnection = false;
}
size_t WebSocket::bufferedAmount() const
{
return _ws.bufferedAmount();
}
}

1
ixwebsocket/IXWebSocket.h
View File

@ -112,6 +112,7 @@ namespace ix
const std::string& getUrl() const;
const WebSocketPerMessageDeflateOptions& getPerMessageDeflateOptions() const;
int getHeartBeatPeriod() const;
size_t bufferedAmount() const;
void enableAutomaticReconnection();
void disableAutomaticReconnection();

120
ixwebsocket/IXWebSocketTransport.cpp
View File

@ -1,7 +1,31 @@
/*
* The MIT License (MIT)
*
* Copyright (c) 2012, 2013 <dhbaird@gmail.com>
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
*/
/*
* IXWebSocketTransport.cpp
* Author: Benjamin Sergeant
* Copyright (c) 2017-2018 Machine Zone, Inc. All rights reserved.
* Copyright (c) 2017-2019 Machine Zone, Inc. All rights reserved.
*/
//
@ -14,14 +38,6 @@
#include "IXUrlParser.h"
#include "IXSocketFactory.h"
#ifdef IXWEBSOCKET_USE_TLS
# ifdef __APPLE__
# include "IXSocketAppleSSL.h"
# else
# include "IXSocketOpenSSL.h"
# endif
#endif
#include <string.h>
#include <stdlib.h>
@ -80,16 +96,6 @@ namespace ix
std::string("Could not parse URL ") + url);
}
if (protocol != "ws" && protocol != "wss")
{
std::stringstream ss;
ss << "Invalid protocol: " << protocol
<< " for url " << url
<< " . Supported protocols are ws and wss";
return WebSocketInitResult(false, 0, ss.str());
}
bool tls = protocol == "wss";
std::string errorMsg;
_socket = createSocket(tls, errorMsg);
@ -184,38 +190,51 @@ namespace ix
std::stringstream ss;
ss << kHeartBeatPingMessage << "::" << _heartBeatPeriod << "s";
sendPing(ss.str());
return;
}
while (true)
// Make sure we send all the buffered data
// there can be a lot of it for large messages.
else if (pollResult == PollResultType_SendRequest)
{
ssize_t ret = _socket->recv((char*)&_readbuf[0], _readbuf.size());
if (ret < 0 && (_socket->getErrno() == EWOULDBLOCK ||
_socket->getErrno() == EAGAIN))
while (!isSendBufferEmpty() && !_requestInitCancellation)
{
break;
}
else if (ret <= 0)
{
_rxbuf.clear();
_socket->close();
setReadyState(CLOSED);
break;
}
else
{
_rxbuf.insert(_rxbuf.end(),
_readbuf.begin(),
_readbuf.begin() + ret);
sendOnSocket();
}
}
else if (pollResult == PollResultType_ReadyForRead)
{
while (true)
{
ssize_t ret = _socket->recv((char*)&_readbuf[0], _readbuf.size());
if (isSendBufferEmpty() && _readyState == CLOSING)
if (ret < 0 && (_socket->getErrno() == EWOULDBLOCK ||
_socket->getErrno() == EAGAIN))
{
break;
}
else if (ret <= 0)
{
_rxbuf.clear();
_socket->close();
setReadyState(CLOSED);
break;
}
else
{
_rxbuf.insert(_rxbuf.end(),
_readbuf.begin(),
_readbuf.begin() + ret);
}
}
}
else if (pollResult == PollResultType_Error)
{
_socket->close();
setReadyState(CLOSED);
}
else if (pollResult == PollResultType_CloseRequest)
{
;
}
},
_heartBeatPeriod);
}
@ -586,11 +605,7 @@ namespace ix
}
}
// Make sure we send all the buffered data ; there can be a lot of it
// for large messages.
// TODO / this will block the sending thread ; we need to eval whether
// this is the right fix
while (!isSendBufferEmpty()) sendOnSocket();
_socket->wakeUpFromPoll(Socket::kSendRequest);
return WebSocketSendInfo(true, compressionError, payloadSize, wireSize);
}
@ -737,8 +752,17 @@ namespace ix
sendData(wsheader_type::CLOSE, normalClosure, compress);
setReadyState(CLOSING);
_socket->wakeUpFromPoll();
_socket->wakeUpFromPoll(Socket::kCloseRequest);
_socket->close();
_closeCode = 1000;
setReadyState(CLOSED);
}
size_t WebSocketTransport::bufferedAmount() const
{
std::lock_guard<std::mutex> lock(_txbufMutex);
return _txbuf.size();
}
} // namespace ix

1
ixwebsocket/IXWebSocketTransport.h
View File

@ -77,6 +77,7 @@ namespace ix
void setReadyState(ReadyStateValues readyStateValue);
void setOnCloseCallback(const OnCloseCallback& onCloseCallback);
void dispatch(const OnMessageCallback& onMessageCallback);
size_t bufferedAmount() const;
private:
std::string _url;