Skip to content

Commit

Permalink
Fix
Browse files Browse the repository at this point in the history
  • Loading branch information
@Coldwings
Coldwings committed Sep 12, 2024
1 parent f30b83c commit b1734db
Show file tree
Hide file tree
Showing 2 changed files with 42 additions and 38 deletions.
27 changes: 13 additions & 14 deletions io/epoll.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -266,7 +266,8 @@ 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
Expand All @@ -275,30 +276,28 @@ class EventEngineEPoll : public MasterEventEngine, public CascadingEventEngine {
ret = -1;
wait_for_events(
0,
[&](void* data) __INLINE__ {
if ((thread*)data == CURRENT) {
[current, &ret](void* data) __INLINE__ {
if ((thread*)data == current) {
ret = 0;
} else {
thread_interrupt((thread*)data, EOK);
}
},
[&]() __INLINE__ { return true; });
if (ret < 0) errno = ETIMEDOUT;
return ret;
if (ret < 0) {
rm_interest({fd, interest, 0});
errno = ETIMEDOUT;
}
}
ret = thread_usleep(timeout ? timeout : 10);
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, interest, 0}); // no ONE_SHOT, to reconfig epoll
errno = (ret == 0) ? ETIMEDOUT : // Timeout
err.no; // Interrupted by other thread
return -1;
}
};

Expand Down
53 changes: 29 additions & 24 deletions io/kqueue.cpp
View file
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,21 +56,20 @@ 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;
Expand All @@ -93,7 +91,8 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine {
nev += ret;
for (int i = 0; i < ret; ++i) {
if (_events[i].filter == EVFILT_USER) continue;
event_callback(_events[i].udata);
auto th = (thread*) _events[i].udata;
if (th) event_callback(th);
}
if (ret == (int) LEN(_events)) { // there may be more events
tm.tv_sec = tm.tv_nsec = 0;
Expand All @@ -103,36 +102,41 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine {
}

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;
enqueue(fd, ev, EV_ADD | EV_ONESHOT, 0, CURRENT);
auto current = CURRENT;
int ret = enqueue(fd, ev, EV_ADD | EV_ONESHOT, 0, current);
if (ret < 0) return ret;
if (!timeout) {
int ret = -1;
do_wait_and_fire_events(0, [&](void* data) {
auto th = (thread*)data;
if (th == CURRENT)
ret = -1;
do_wait_and_fire_events(0, [current, &ret](thread* th) {
if (th == current)
ret = 0;
else if (th)
else
thread_interrupt(th, EOK);
});
if (ret < 0) errno = ETIMEDOUT;
if (ret <0) {
enqueue(fd, ev, EV_DELETE, 0, current, true);
errno = ETIMEDOUT;
}
return ret;
}
int ret = thread_usleep(timeout);
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
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 = -1) override {
return do_wait_and_fire_events(timeout, [&](void* data){
auto th = (thread*)data;
if (th) thread_interrupt(th, EOK);
});
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 {
Expand Down Expand Up @@ -183,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};
ret = kevent(_kq, _events, _n, _events, count, &_tm);
if (ret < 0)
LOG_ERRNO_RETURN(0, -1, "failed to call kevent()");
Expand All @@ -202,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

0 comments on commit b1734db

Please sign in to comment.