Compare commits

...

2 Commits

Author SHA1 Message Date
Benjamin Sergeant
0834198e74 do not create a kqueue everytime we call poll 2020-08-05 15:00:01 -07:00
Benjamin Sergeant
22dd32d4e9 wip kqueue, with selectinterrupt not implemented + we create a queue every time 2020-08-05 14:47:03 -07:00
4 changed files with 129 additions and 7 deletions

View File

@ -35,19 +35,122 @@ namespace ix
: _sockfd(fd) : _sockfd(fd)
, _selectInterrupt(createSelectInterrupt()) , _selectInterrupt(createSelectInterrupt())
{ {
; #if defined(__APPLE__)
_kqueuefd = kqueue();
#endif
} }
Socket::~Socket() Socket::~Socket()
{ {
close(); close();
#if defined(__APPLE__)
::close(_kqueuefd);
#endif
} }
PollResultType Socket::poll(bool readyToRead, PollResultType Socket::poll(bool readyToRead,
int timeoutMs, int timeoutMs,
int sockfd, int sockfd,
const SelectInterruptPtr& selectInterrupt) const SelectInterruptPtr& selectInterrupt,
int kqueuefd)
{ {
#if defined(__APPLE__)
// FIXME int kqueuefd = kqueue();
struct kevent ke;
EV_SET(&ke, sockfd, (readyToRead) ? EVFILT_READ : EVFILT_WRITE, EV_ADD, 0, 0, NULL);
if (kevent(kqueuefd, &ke, 1, NULL, 0, NULL) == -1) return PollResultType::Error;
int retval, numevents = 0;
int nfds = 1;
#if 0
if (selectInterrupt)
{
nfds = 2;
int interruptFd = selectInterrupt->getFd();
struct kevent ke;
EV_SET(&ke, interruptFd, EVFILT_READ, EV_ADD, 0, 0, NULL);
if (kevent(kqueuefd, &ke, 1, NULL, 0, NULL) == -1) return PollResultType::Error;
}
#endif
struct kevent *events;
events = (struct kevent*) malloc(sizeof(struct kevent) * nfds);
if (timeoutMs != 0)
{
struct timespec timeout;
timeout.tv_sec = timeoutMs / 1000;
timeout.tv_nsec = (timeoutMs % 1000) * 1000 * 1000;
retval = kevent(kqueuefd, NULL, 0, events, nfds, &timeout);
}
else
{
retval = kevent(kqueuefd, NULL, 0, events, nfds, NULL);
}
#if 0
if (retval > 0) {
int j;
numevents = retval;
for(j = 0; j < numevents; j++) {
int mask = 0;
struct kevent *e = events+j;
if (e->filter == EVFILT_READ) mask |= AE_READABLE;
if (e->filter == EVFILT_WRITE) mask |= AE_WRITABLE;
eventLoop->fired[j].fd = e->ident;
eventLoop->fired[j].mask = mask;
}
}
#else
PollResultType pollResult = PollResultType::ReadyForRead;
if (retval < 0)
{
pollResult = PollResultType::Error;
}
if (retval > 0) {
struct kevent *e = events;
if (e->filter == EVFILT_READ)
{
pollResult = PollResultType::ReadyForRead;
}
else if (e->filter == EVFILT_WRITE)
{
pollResult = PollResultType::ReadyForWrite;
int optval = -1;
socklen_t optlen = sizeof(optval);
// getsockopt() puts the errno value for connect into optval so 0
// means no-error.
if (getsockopt(sockfd, SOL_SOCKET, SO_ERROR, &optval, &optlen) == -1 || optval != 0)
{
pollResult = PollResultType::Error;
// set errno to optval so that external callers can have an
// appropriate error description when calling strerror
errno = optval;
}
}
}
else
{
pollResult = PollResultType::Timeout;
}
#endif
free(events);
// ::close(kqueuefd); //FMXE
return pollResult;
#else
// //
// We used to use ::select to poll but on Android 9 we get large fds out of // We used to use ::select to poll but on Android 9 we get large fds out of
// ::connect which crash in FD_SET as they are larger than FD_SETSIZE. Switching // ::connect which crash in FD_SET as they are larger than FD_SETSIZE. Switching
@ -142,6 +245,7 @@ namespace ix
} }
return pollResult; return pollResult;
#endif
} }
PollResultType Socket::isReadyToRead(int timeoutMs) PollResultType Socket::isReadyToRead(int timeoutMs)
@ -152,7 +256,7 @@ namespace ix
} }
bool readyToRead = true; bool readyToRead = true;
return poll(readyToRead, timeoutMs, _sockfd, _selectInterrupt); return poll(readyToRead, timeoutMs, _sockfd, _selectInterrupt, _kqueuefd);
} }
PollResultType Socket::isReadyToWrite(int timeoutMs) PollResultType Socket::isReadyToWrite(int timeoutMs)
@ -163,7 +267,7 @@ namespace ix
} }
bool readyToRead = false; bool readyToRead = false;
return poll(readyToRead, timeoutMs, _sockfd, _selectInterrupt); return poll(readyToRead, timeoutMs, _sockfd, _selectInterrupt, _kqueuefd);
} }
// Wake up from poll/select by writing to the pipe which is watched by select // Wake up from poll/select by writing to the pipe which is watched by select

View File

@ -13,6 +13,13 @@
#include <string> #include <string>
#include <vector> #include <vector>
// For kqueue
#if defined(__APPLE__)
#include <sys/types.h>
#include <sys/event.h>
#include <sys/time.h>
#endif
#ifdef _WIN32 #ifdef _WIN32
#include <BaseTsd.h> #include <BaseTsd.h>
typedef SSIZE_T ssize_t; typedef SSIZE_T ssize_t;
@ -94,7 +101,8 @@ namespace ix
static PollResultType poll(bool readyToRead, static PollResultType poll(bool readyToRead,
int timeoutMs, int timeoutMs,
int sockfd, int sockfd,
const SelectInterruptPtr& selectInterrupt); const SelectInterruptPtr& selectInterrupt,
int kqueuefd);
// Used as special codes for pipe communication // Used as special codes for pipe communication
@ -114,5 +122,9 @@ namespace ix
static constexpr size_t kChunkSize = 1 << 15; static constexpr size_t kChunkSize = 1 << 15;
SelectInterruptPtr _selectInterrupt; SelectInterruptPtr _selectInterrupt;
#if defined(__APPLE__)
int _kqueuefd;
#endif
}; };
} // namespace ix } // namespace ix

View File

@ -66,7 +66,10 @@ namespace ix
int timeoutMs = 10; int timeoutMs = 10;
bool readyToRead = false; bool readyToRead = false;
auto selectInterrupt = std::make_unique<SelectInterrupt>(); auto selectInterrupt = std::make_unique<SelectInterrupt>();
PollResultType pollResult = Socket::poll(readyToRead, timeoutMs, fd, selectInterrupt);
int kqueuefd = kqueue();
PollResultType pollResult = Socket::poll(readyToRead, timeoutMs, fd, selectInterrupt, kqueuefd);
::close(kqueuefd);
if (pollResult == PollResultType::Timeout) if (pollResult == PollResultType::Timeout)
{ {

View File

@ -259,8 +259,11 @@ namespace ix
int timeoutMs = 10; int timeoutMs = 10;
bool readyToRead = true; bool readyToRead = true;
auto selectInterrupt = std::make_unique<SelectInterrupt>(); auto selectInterrupt = std::make_unique<SelectInterrupt>();
int kqueuefd = kqueue();
PollResultType pollResult = PollResultType pollResult =
Socket::poll(readyToRead, timeoutMs, _serverFd, selectInterrupt); Socket::poll(readyToRead, timeoutMs, _serverFd, selectInterrupt, kqueuefd);
::close(kqueuefd);
if (pollResult == PollResultType::Error) if (pollResult == PollResultType::Error)
{ {