Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fix sockpool wait_for_fd for old releases #554

Closed
wants to merge 6 commits into from
Closed

Fix sockpool wait_for_fd for old releases #554

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
8aefb11
FIX: wait_for_fd in epoll should atleast put into sleep for a tiny mo…
Coldwings Sep 9, 2024
5b063be
Also fix on kqueue
Coldwings Sep 9, 2024
bdf2350
For old thread_usleep API
Coldwings Sep 10, 2024
f30b83c
wait_for_fd(0) reap events directly
Coldwings Sep 11, 2024
596440a
Fix
Coldwings Sep 12, 2024
38a05c7
minor fix
Coldwings Sep 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 25 additions & 9 deletions io/epoll.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -266,22 +266,38 @@ class EventEngineEPoll : public MasterEventEngine, public CascadingEventEngine {
virtual int cancel_wait() override { return eventfd_write(_evfd, 1); }

int wait_for_fd(int fd, uint32_t interests, uint64_t timeout) override {
Event event{fd, interests | ONE_SHOT, CURRENT};
thread* current = CURRENT;
Event event{fd, interests | ONE_SHOT, current};
int ret = add_interest(event);
if (ret < 0) LOG_ERROR_RETURN(0, -1, "failed to add event interest");
// if timeout is just simple 0, wait for a tiny little moment
// so that events can be collect.
if (!timeout) {
ret = -1;
wait_for_events(
0,
[current, &ret](void* data) __INLINE__ {
if ((thread*)data == current) {
ret = 0;
} else {
thread_interrupt((thread*)data, EOK);
}
},
[&]() __INLINE__ { return true; });
if (ret < 0) {
rm_interest({fd, interests, 0});
errno = ETIMEDOUT;
}
}
ret = thread_usleep(timeout);
ERRNO err;
if (ret == -1 && err.no == EOK) {
return 0; // Event arrived
} else if (ret == 0) {
rm_interest(event); // Timeout
errno = ETIMEDOUT;
return -1;
} else {
rm_interest(event); // Interrupted by other thread
errno = err.no;
return -1;
}
rm_interest({fd, interests, 0}); // no ONE_SHOT, to reconfig epoll
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it's might be safer to rm_interest(event) as before.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done

errno = (ret == 0) ? ETIMEDOUT : // Timeout
err.no; // Interrupted by other thread
return -1;
}
};

Expand Down
74 changes: 49 additions & 25 deletions io/kqueue.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,6 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine {
struct kevent _events[32];
int _kq = -1;
uint32_t _n = 0; // # of events to submit
struct timespec _tm = {0, 0}; // used for poll

int init() {
if (_kq >= 0)
Expand All @@ -57,41 +56,27 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine {

~KQueue() override {
LOG_INFO("Finish event engine: kqueue");
// if (_n > 0) LOG_INFO(VALUE(_events[0].ident), VALUE(_events[0].filter), VALUE(_events[0].flags));
// assert(_n == 0);
if (_kq >= 0)
close(_kq);
}

int enqueue(int fd, short event, uint16_t action, uint32_t event_flags, void* udata, bool immediate = false) {
// LOG_INFO("enqueue _kq: `, fd: `, event: `, action: `", _kq, fd, event, action);
assert(_n < LEN(_events));
auto entry = &_events[_n++];
EV_SET(entry, fd, event, action, event_flags, 0, udata);
if (immediate || _n == LEN(_events)) {
int ret = kevent(_kq, _events, _n, nullptr, 0, nullptr);
if (ret < 0)
struct timespec tm{0, 0};
int ret = kevent(_kq, _events, _n, nullptr, 0, &tm);
if (ret < 0) {
LOG_ERRNO_RETURN(0, -1, "failed to submit events with kevent()");
}
_n = 0;
}
return 0;
}

int wait_for_fd(int fd, uint32_t interests, uint64_t timeout) override {
short ev = (interests == EVENT_READ) ? EVFILT_READ : EVFILT_WRITE;
enqueue(fd, ev, EV_ADD | EV_ONESHOT, 0, CURRENT);
int ret = thread_usleep(timeout);
ERRNO err;
if (ret == -1 && err.no == EOK) {
return 0; // event arrived
}

// enqueue(fd, ev, EV_DELETE, 0, CURRENT, true); // immediately
errno = (ret == 0) ? ETIMEDOUT : err.no;
return -1;
}

ssize_t wait_and_fire_events(uint64_t timeout = -1) override {
template<typename EVCB>
ssize_t do_wait_and_fire_events(uint64_t timeout, EVCB&& event_callback) {
ssize_t nev = 0;
struct timespec tm;
tm.tv_sec = timeout / 1000 / 1000;
Expand All @@ -107,7 +92,7 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine {
for (int i = 0; i < ret; ++i) {
if (_events[i].filter == EVFILT_USER) continue;
auto th = (thread*) _events[i].udata;
if (th) thread_interrupt(th, EOK);
if (th) event_callback(th);
}
if (ret == (int) LEN(_events)) { // there may be more events
tm.tv_sec = tm.tv_nsec = 0;
Expand All @@ -116,6 +101,44 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine {
return nev;
}

int wait_for_fd(int fd, uint32_t interests, uint64_t timeout) override {
if (unlikely(interests == 0)) {
errno = ENOSYS;
return -1;
}
short ev = (interests == EVENT_READ) ? EVFILT_READ : EVFILT_WRITE;
auto current = CURRENT;
int ret = enqueue(fd, ev, EV_ADD | EV_ONESHOT, 0, current);
if (ret < 0) return ret;
if (!timeout) {
ret = -1;
do_wait_and_fire_events(0, [current, &ret](thread* th) {
if (th == current)
ret = 0;
else
thread_interrupt(th, EOK);
});
if (ret <0) {
enqueue(fd, ev, EV_DELETE, 0, current, true);
errno = ETIMEDOUT;
}
return ret;
}
ret = thread_usleep(timeout);
ERRNO err;
if (ret == -1 && err.no == EOK) {
return 0; // event arrived
}

enqueue(fd, ev, EV_DELETE, 0, current, true);
errno = (ret == 0) ? ETIMEDOUT : err.no;
return -1;
}

ssize_t wait_and_fire_events(uint64_t timeout) override {
return do_wait_and_fire_events(timeout, [](thread *th) { thread_interrupt(th, EOK); });
}

int cancel_wait() override {
enqueue(_kq, EVFILT_USER, EV_ONESHOT, NOTE_TRIGGER, nullptr, true);
return 0;
Expand Down Expand Up @@ -164,11 +187,12 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine {
}

ssize_t wait_for_events(void** data,
size_t count, uint64_t timeout = -1) override {
int ret = get_vcpu()->master_event_engine->wait_for_fd_readable(_kq, timeout);
size_t count, uint64_t timeout) override {
int ret = ::photon::wait_for_fd_readable(_kq, timeout);
if (ret < 0) return errno == ETIMEDOUT ? 0 : -1;
if (count > LEN(_events))
count = LEN(_events);
static const struct timespec _tm = {0, 0};
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why static here, while not in enqueue()?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Use on-stack temporary timespec is fine. Changed in latest commit.

ret = kevent(_kq, _events, _n, _events, count, &_tm);
if (ret < 0)
LOG_ERRNO_RETURN(0, -1, "failed to call kevent()");
Expand All @@ -183,7 +207,7 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine {
};

__attribute__((noinline))
KQueue* new_kqueue_engine() {
static KQueue* new_kqueue_engine() {
LOG_INFO("Init event engine: kqueue");
return NewObj<KQueue>()->init();
}
Expand Down
9 changes: 5 additions & 4 deletions net/pooled_socket.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,9 +121,10 @@ class TCPSocketPool : public ForwardSocketClient {
photon::Timer timer;

// all fd < 0 treated as socket not based on fd
// and always alive. Using such socket needs user
// and always reuseable. Using such socket needs user
// to check if connected socket is still usable.
bool stream_alive(int fd) {
// if there still have unread bytes in strema, it should be closed.
bool stream_reusable(int fd) {
return (fd < 0) || (wait_for_fd_readable(fd, 0) != 0);
}

Expand Down Expand Up @@ -196,7 +197,7 @@ class TCPSocketPool : public ForwardSocketClient {
if (!stream) {
stream = m_underlay->connect(remote, local);
if (!stream) return nullptr;
} else if (!stream_alive(stream->get_underlay_fd())) {
} else if (!stream_reusable(stream->get_underlay_fd())) {
delete stream;
goto again;
}
Expand Down Expand Up @@ -229,7 +230,7 @@ class TCPSocketPool : public ForwardSocketClient {

bool release(EndPoint ep, ISocketStream* stream) {
auto fd = stream->get_underlay_fd();
if (!stream_alive(fd)) return false;
if (!stream_reusable(fd)) return false;
auto node = new StreamListNode(ep, stream, fd, expiration);
push_into_pool(node);
return true;
Expand Down
Loading