Suppose the following series of events occurs:
EPOLLIN | EPOLLEXCLUSIVE
EPOLLIN | EPOLLEXCLUSIVE
accept
, a second incoming connection arrives at the listening socket.Here, the socket is already readable, so the second connection doesn't change that. This is level-triggered epoll, so according to the normal rules, the second connection can be treated as a no-op, and the second thread doesn't need to be awoken. ...Of course, not waking up the second thread would kind of defeat the whole purpose of EPOLLEXCLUSIVE
? But my trust in API designers doing the right thing is not as strong as it once was, and I can't find anything in the documentation to rule this out.
Questions
a) Is the above scenario possible, where two connections arrive but only thread is woken? Or is it guaranteed that every distinct incoming connection on a listening socket will wake another thread?
b) Is there a general rule to predict how EPOLLEXCLUSIVE
and level-triggered epoll interact?
b) What about EPOLLIN | EPOLLEXCLUSIVE
and EPOLLOUT | EPOLLEXCLUSIVE
for byte-stream fds, like a connected TCP socket or a pipe? E.g. what happens if more data arrives while a pipe is already readable?
epoll provides both edge-triggered and level-triggered modes. In edge-triggered mode, a call to epoll_wait will return only when a new event is enqueued with the epoll object, while in level-triggered mode, epoll_wait will return as long as the condition holds.
epoll monitors I/O events for multiple file descriptors. epoll supports edge trigger (ET) or level trigger (LT), which waits for I/O events via epoll_wait and blocks the calling thread if no events are currently available. select and poll only support LT working mode, and the default working mode of epoll is LT mode.
epoll stands for event poll and is a Linux specific construct. It allows for a process to monitor multiple file descriptors and get notifications when I/O is possible on them. It allows for both edge-triggered as well as level-triggered notifications.
The main difference between epoll and select is that in select() the list of file descriptors to wait on only exists for the duration of a single select() call, and the calling task only stays on the sockets' wait queues for the duration of a single call.
Edited (original answer is after the code used for testing)
To make sure things are clear, I'll go over EPOLLEXCLUSIVE
as it relates to edge triggered events (EPOLLET
) as well as level-triggered events, to show how these effect expected behavior.
As you well know:
Edge Triggered: Once you set EPOLLET
, events are triggered only if they change the state of the fd
- meaning that only the first event is triggered and no new events will get triggered until that event is fully handled.
This design is explicitly meant to prevent epoll_wait
from returning due to an event that is in the process of being handled (i.e., when new data arrives while the EPOLLIN
was already raised but read
hadn't been called or not all of the data was read).
The edge-triggered event rule is simple all same-type (i.e. EPOLLIN
) events are merged until all available data was processed.
In the case of a listening socket, the EPOLLIN
event won't be triggered again until all existing listen
"backlog" sockets have been accepted using accept
.
In the case of a byte stream, new events won't be triggered until all the the available bytes have been read from the stream (the buffer was emptied).
Level Triggered: On the other hand, level triggered events will behave closer to how legacy select
(or poll
) operates, allowing epoll
to be used with older code.
The event-merger rule is more complex: events of the same type are only merged if no one is waiting for an event (no one is waiting for epoll_wait
to return), or if multiple events happen before epoll_wait
can return... otherwise any event causes epoll_wait
to return.
In the case of a listening socket, the EPOLLIN
event will be triggered every time a client connects... unless no one is waiting for epoll_wait
to return, in which case the next call for epoll_wait
will return immediately and all the EPOLLIN
events that occurred during that time will have been merged into a single event.
In the case of a byte stream, new events will be triggered every time new data comes in... unless, of course, no one is waiting for epoll_wait
to return, in which case the next call will return immediately for all the data that arrive util epoll_wait
returned (even if it arrived in different chunks / events).
Exclusive return: The EPOLLEXCLUSIVE
flag is used to prevent the "thundering heard" behavior, so only a single epoll_wait
caller is woken up for each fd
wake-up event.
As I pointed out before, for edge-triggered states, an fd
wake-up event is a change in the fd
state. So all EPOLLIN
events will be raised until all data was read (the listening socket's backlog was emptied).
On the other hand, for level triggered events, each EPOLLIN
will invoke a wake up event. If no one is waiting, these events will be merged.
Following the example in your question:
For level triggered events: every time a client connects, a single thread will return from epoll_wait
... BUT, if two more clients were to connect while both threads were busy accepting the first two clients, these EPOLLIN
events would merge into a single event and the next call to epoll_wait
will return immediately with that merged event.
In the context of the example given in the question, thread B is expected to "wake up" due to epoll_wait
returning.
In this case, both threads will "race" towards accept
.
However, this doesn't defeat the EPOLLEXCLUSIVE
directive or intent.
The EPOLLEXCLUSIVE
directive is meant to prevent the "thundering heard" phenomenon. In this case, two threads are racing to accept two connections. Each thread can (presumably) call accept
safely, with no errors. If three threads were used, the third would keep on sleeping.
If the EPOLLEXCLUSIVE
weren't used, all the epoll_wait
threads would have been woken up whenever a connection was available, meaning that as soon as the first connection arrived, both threads would have been racing to accept a single connection (resulting in a possible error for one of them).
For edge triggered events: only one thread is expected to receive the "wake up" call. That thread is expected to accept
all waiting connections (empty the listen
"backlog"). No more EPOLLIN
events will be raised for that socket until the backlog is emptied.
The same applies to readable sockets and pipes. The thread that was woken up is expected to deal with all the readable data. This prevents to waiting threads from attempting to read the data concurrently and experiencing file lock race conditions.
I would recommend (and this is what I do) to set the listening socket to non-blocking mode and calling accept
in a loop until an EAGAIN
(or EWOULDBLOCK
) error is raised, indicating that the backlog is empty. There is no way to avoid the risk of events being merged. The same is true for reading from a socket.
Testing this with code:
I wrote a simple test, with some sleep
commands and blocking sockets. Client sockets are initiated only after both threads start waiting for epoll
.
Client thread initiation is delayed, so client 1 and client 2 start a second apart.
Once a server thread is woken up, it will sleep for a second (allowing the second client to do it's thing) before calling accept
. Maybe the servers should sleep a little more, but it seems close enough to manage the scheduler without resorting to conditional variables.
Here are the results of my test code (which might be a mess, I'm not the best person for test design)...
On Ubuntu 16.10, which supports EPOLLEXCLUSIVE
, the test results show that the listening threads are woken up one after the other, in response to the clients. In the example in the question, thread B is woken up.
Test address: <null>:8000
Server thread 2 woke up with 1 events
Server thread 2 will sleep for a second, to let things happen.
client number 1 connected
Server thread 1 woke up with 1 events
Server thread 1 will sleep for a second, to let things happen.
client number 2 connected
Server thread 2 accepted a connection and saying hello.
client 1: Hello World - from server thread 2.
Server thread 1 accepted a connection and saying hello.
client 2: Hello World - from server thread 1.
To compare with Ubuntu 16.04 (without EPOLLEXCLUSIVE
support), than both threads are woken up for the first connection. Since I use blocking sockets, the second thread hangs on accept
until client # 2 connects.
main.c:178:2: warning: #warning EPOLLEXCLUSIVE undeclared, test is futile [-Wcpp]
#warning EPOLLEXCLUSIVE undeclared, test is futile
^
Test address: <null>:8000
Server thread 1 woke up with 1 events
Server thread 1 will sleep for a second, to let things happen.
Server thread 2 woke up with 1 events
Server thread 2 will sleep for a second, to let things happen.
client number 1 connected
Server thread 1 accepted a connection and saying hello.
client 1: Hello World - from server thread 1.
client number 2 connected
Server thread 2 accepted a connection and saying hello.
client 2: Hello World - from server thread 2.
For one more comparison, the results for level triggered kqueue
show that both threads are awoken for the first connection. Since I use blocking sockets, the second thread hangs on accept
until client # 2 connects.
Test address: <null>:8000
client number 1 connected
Server thread 2 woke up with 1 events
Server thread 1 woke up with 1 events
Server thread 2 will sleep for a second, to let things happen.
Server thread 1 will sleep for a second, to let things happen.
Server thread 2 accepted a connection and saying hello.
client 1: Hello World - from server thread 2.
client number 2 connected
Server thread 1 accepted a connection and saying hello.
client 2: Hello World - from server thread 1.
My test code was (sorry for the lack of comments and the messy code, I wasn't writing for future maintenance):
#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#define ADD_EPOLL_OPTION 0 // define as EPOLLET or 0
#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <netdb.h>
#include <pthread.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/resource.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <sys/types.h>
#include <time.h>
#include <unistd.h>
#if !defined(__linux__) && !defined(__CYGWIN__)
#include <sys/event.h>
#define reactor_epoll 0
#else
#define reactor_epoll 1
#include <sys/epoll.h>
#include <sys/timerfd.h>
#endif
int sock_listen(const char *address, const char *port);
void *listen_threard(void *arg);
void *client_thread(void *arg);
int server_fd;
char const *address = NULL;
char const *port = "8000";
int main(int argc, char const *argv[]) {
if (argc == 2) {
port = argv[1];
} else if (argc == 3) {
port = argv[2];
address = argv[1];
}
fprintf(stderr, "Test address: %s:%s\n", address ? address : "<null>", port);
server_fd = sock_listen(address, port);
/* code */
pthread_t threads[4];
for (size_t i = 0; i < 2; i++) {
if (pthread_create(threads + i, NULL, listen_threard, (void *)i))
perror("couldn't initiate server thread"), exit(-1);
}
for (size_t i = 2; i < 4; i++) {
sleep(1);
if (pthread_create(threads + i, NULL, client_thread, (void *)i))
perror("couldn't initiate client thread"), exit(-1);
}
// join only server threads.
for (size_t i = 0; i < 2; i++) {
pthread_join(threads[i], NULL);
}
close(server_fd);
sleep(1);
return 0;
}
/**
Sets a socket to non blocking state.
*/
inline int sock_set_non_block(int fd) // Thanks to Bjorn Reese
{
/* If they have O_NONBLOCK, use the Posix way to do it */
#if defined(O_NONBLOCK)
/* Fixme: O_NONBLOCK is defined but broken on SunOS 4.1.x and AIX 3.2.5. */
int flags;
if (-1 == (flags = fcntl(fd, F_GETFL, 0)))
flags = 0;
// printf("flags initial value was %d\n", flags);
return fcntl(fd, F_SETFL, flags | O_NONBLOCK);
#else
/* Otherwise, use the old way of doing it */
static int flags = 1;
return ioctl(fd, FIOBIO, &flags);
#endif
}
/* open a listenning socket */
int sock_listen(const char *address, const char *port) {
int srvfd;
// setup the address
struct addrinfo hints;
struct addrinfo *servinfo; // will point to the results
memset(&hints, 0, sizeof hints); // make sure the struct is empty
hints.ai_family = AF_UNSPEC; // don't care IPv4 or IPv6
hints.ai_socktype = SOCK_STREAM; // TCP stream sockets
hints.ai_flags = AI_PASSIVE; // fill in my IP for me
if (getaddrinfo(address, port, &hints, &servinfo)) {
perror("addr err");
return -1;
}
// get the file descriptor
srvfd =
socket(servinfo->ai_family, servinfo->ai_socktype, servinfo->ai_protocol);
if (srvfd <= 0) {
perror("socket err");
freeaddrinfo(servinfo);
return -1;
}
// // keep the server socket blocking for the test.
// // make sure the socket is non-blocking
// if (sock_set_non_block(srvfd) < 0) {
// perror("couldn't set socket as non blocking! ");
// freeaddrinfo(servinfo);
// close(srvfd);
// return -1;
// }
// avoid the "address taken"
{
int optval = 1;
setsockopt(srvfd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));
}
// bind the address to the socket
{
int bound = 0;
for (struct addrinfo *p = servinfo; p != NULL; p = p->ai_next) {
if (!bind(srvfd, p->ai_addr, p->ai_addrlen))
bound = 1;
}
if (!bound) {
// perror("bind err");
freeaddrinfo(servinfo);
close(srvfd);
return -1;
}
}
freeaddrinfo(servinfo);
// listen in
if (listen(srvfd, SOMAXCONN) < 0) {
perror("couldn't start listening");
close(srvfd);
return -1;
}
return srvfd;
}
/* will start listenning, sleep for 5 seconds, then accept all the backlog and
* finish */
void *listen_threard(void *arg) {
int epoll_fd;
ssize_t event_count;
#if reactor_epoll
#ifndef EPOLLEXCLUSIVE
#warning EPOLLEXCLUSIVE undeclared, test is futile
#define EPOLLEXCLUSIVE 0
#endif
// create the epoll wait fd
epoll_fd = epoll_create1(0);
if (epoll_fd < 0)
perror("couldn't create epoll fd"), exit(1);
// add the server fd to the epoll watchlist
{
struct epoll_event chevent = {0};
chevent.data.ptr = (void *)((uintptr_t)server_fd);
chevent.events =
EPOLLOUT | EPOLLIN | EPOLLERR | EPOLLEXCLUSIVE | ADD_EPOLL_OPTION;
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, server_fd, &chevent);
}
// wait with epoll
struct epoll_event events[10];
event_count = epoll_wait(epoll_fd, events, 10, 5000);
#else
// testing on BSD, use kqueue
epoll_fd = kqueue();
if (epoll_fd < 0)
perror("couldn't create kqueue fd"), exit(1);
// add the server fd to the kqueue watchlist
{
struct kevent chevent[2];
EV_SET(chevent, server_fd, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0,
(void *)((uintptr_t)server_fd));
EV_SET(chevent + 1, server_fd, EVFILT_WRITE, EV_ADD | EV_ENABLE, 0, 0,
(void *)((uintptr_t)server_fd));
kevent(epoll_fd, chevent, 2, NULL, 0, NULL);
}
// wait with kqueue
static struct timespec reactor_timeout = {.tv_sec = 5, .tv_nsec = 0};
struct kevent events[10];
event_count = kevent(epoll_fd, NULL, 0, events, 10, &reactor_timeout);
#endif
close(epoll_fd);
if (event_count <= 0) {
fprintf(stderr, "Server thread %lu wakeup no events / error\n",
(size_t)arg + 1);
perror("errno ");
return NULL;
}
fprintf(stderr, "Server thread %lu woke up with %lu events\n",
(size_t)arg + 1, event_count);
fprintf(stderr,
"Server thread %lu will sleep for a second, to let things happen.\n",
(size_t)arg + 1);
sleep(1);
int connfd;
struct sockaddr_storage client_addr;
socklen_t client_addrlen = sizeof client_addr;
/* accept up all connections. we're non-blocking, -1 == no more connections */
if ((connfd = accept(server_fd, (struct sockaddr *)&client_addr,
&client_addrlen)) >= 0) {
fprintf(stderr,
"Server thread %lu accepted a connection and saying hello.\n",
(size_t)arg + 1);
if (write(connfd, arg ? "Hello World - from server thread 2."
: "Hello World - from server thread 1.",
35) < 35)
perror("server write failed");
close(connfd);
} else {
fprintf(stderr, "Server thread %lu failed to accept a connection",
(size_t)arg + 1);
perror(": ");
}
return NULL;
}
void *client_thread(void *arg) {
int fd;
// setup the address
struct addrinfo hints;
struct addrinfo *addrinfo; // will point to the results
memset(&hints, 0, sizeof hints); // make sure the struct is empty
hints.ai_family = AF_UNSPEC; // don't care IPv4 or IPv6
hints.ai_socktype = SOCK_STREAM; // TCP stream sockets
hints.ai_flags = AI_PASSIVE; // fill in my IP for me
if (getaddrinfo(address, port, &hints, &addrinfo)) {
perror("client couldn't initiate address");
return NULL;
}
// get the file descriptor
fd =
socket(addrinfo->ai_family, addrinfo->ai_socktype, addrinfo->ai_protocol);
if (fd <= 0) {
perror("client couldn't create socket");
freeaddrinfo(addrinfo);
return NULL;
}
// // // Leave the socket blocking for the test.
// // make sure the socket is non-blocking
// if (sock_set_non_block(fd) < 0) {
// freeaddrinfo(addrinfo);
// close(fd);
// return -1;
// }
if (connect(fd, addrinfo->ai_addr, addrinfo->ai_addrlen) < 0 &&
errno != EINPROGRESS) {
fprintf(stderr, "client number %lu FAILED\n", (size_t)arg - 1);
perror("client connect failure");
close(fd);
freeaddrinfo(addrinfo);
return NULL;
}
freeaddrinfo(addrinfo);
fprintf(stderr, "client number %lu connected\n", (size_t)arg - 1);
char buffer[128];
if (read(fd, buffer, 35) < 35) {
perror("client: read error");
close(fd);
} else {
buffer[35] = 0;
fprintf(stderr, "client %lu: %s\n", (size_t)arg - 1, buffer);
close(fd);
}
return NULL;
}
P.S.
As a final recommendation, I would consider having no more than a single thread and a single epoll fd
per process. This way the "thundering heard" is a non-issue and EPOLLEXCLUSIVE
(which is still very new and isn't widely supported) can be disregarded... the only "thundering heard" this still exposes is for the limited amount of shared sockets, where the race condition might be good for load balancing.
Original Answer
I'm not sure I understand the confusion, so I'll go over EPOLLET
and EPOLLEXCLUSIVE
to show their combined expected behavior.
As you well know:
Once you set EPOLLET
(edge triggered), events are triggered on fd
state changes rather than fd
events.
This design is explicitly meant to prevent epoll_wait
from returning due to an event that is in the process of being handled (i.e., when new data arrives while the EPOLLIN
was already raised but read
hadn't been called or not all of the data was read).
In the case of a listening socket, the EPOLLIN
event won't be triggered again until all existing listen
"backlog" sockets have been accepted using accept
.
The EPOLLEXCLUSIVE
flag is used to prevent the "thundering heard" behavior, so only a single epoll_wait
caller is woken up for each fd
wake-up event.
As I pointed out before, for edge-triggered states, an fd
wake-up event is a change in the fd
state. So all EPOLLIN
events will be raised until all data was read (the listening socket's backlog was emptied).
When merging these behaviors, and following the example in your question, only one thread is expected to receive the "wake up" call. That thread is expected to accept
all waiting connections (empty the listen
"backlog") or no more EPOLLIN
events will be raised for that socket.
The same applies to readable sockets and pipes. The thread that was woken up is expected to deal with all the readable data. This prevents to waiting threads from attempting to read the data concurrently and experiencing file lock race conditions.
I would recommend that you consider avoiding the edge triggered events if you mean to call accept
only once for each epoll_wait
wake-up event. Regardless of using EPOLLEXCLUSIVE
, you run the risk of not emptying the existing "backlog", so that no new wake-up events will be raised.
Alternatively, I would recommend (and this is what I do) to set the listening socket to non-blocking mode and calling accept
in a loop until and an EAGAIN
(or EWOULDBLOCK
) error is raised, indicating that the backlog is empty.
EDIT 1: Level Triggered Events
It seems, as Nathaniel pointed out in the comment, that I totally misunderstood the question... I guess I'm used to EPOLLET
being the misunderstood element.
So, what happens with normal, level-triggered, events (NOT EPOLLET
)?
Well... the expected behavior is the exact mirror image (opposite) of edge triggered events.
For listenning sockets, the epoll_wait
is expected return whenever a new connected is available, whether accept
was called after a previous event or not.
Events are only "merged" if no-one is waiting with epoll_wait
... in which case the next call for epoll_wait
will return immediately.
In the context of the example given in the question, thread B is expected to "wake up" due to epoll_wait
returning.
In this case, both threads will "race" towards accept
.
However, this doesn't defeat the EPOLLEXCLUSIVE
directive or intent.
The EPOLLEXCLUSIVE
directive is meant to prevent the "thundering heard" phenomenon. In this case, two threads are racing to accept two connections. Each thread can (presumably) call accept
safely, with no errors. If three threads were used, the third would keep on sleeping.
If the EPOLLEXCLUSIVE
weren't used, all the epoll_wait
threads would have been woken up whenever a connection was available, meaning that as soon as the first connection arrived, both threads would have been racing to accept a single connection (resulting in a possible error for one of them).
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With