Also fix on kqueue

wait_for_fd reap events when wait_for_fd in 0 timeout

fix kevent immeidate enqueue

KQueue event submitt always use immediate

fix

Fix kqueue response for wait_for_fd with interests=0

Remove unused code

io/epoll.cpp

@@ -299,11 +299,30 @@ ok: entry.interests |= eint;
  LOG_ERROR_RETURN(EINVAL, -1, "can not wait for multiple interests");
  if (unlikely(interest == 0))
  return rm_interest({fd, EVENT_RWE| ONE_SHOT, 0}); // remove fd from epoll
- int ret = add_interest({fd, interest | ONE_SHOT, CURRENT});
+ thread* current = CURRENT;
+ int ret = add_interest({fd, interest | ONE_SHOT, current});
  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.
- ret = thread_usleep(timeout.timeout() ? timeout : Timeout(10));
+ if (timeout.expired()) {
+ 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, interest, 0});
+ errno = ETIMEDOUT;
+ }
+ return ret;
+ }
+ ret = thread_usleep(timeout);
  ERRNO err;
  if (ret == -1 && err.no == EOK) {
  return 0; // Event arrived

io/kqueue.cpp

@@ -72,44 +72,23 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine, public Res
  close(_kq);
  }
 
- __attribute__((noinline))
- static void debug_breakpoint() {
- }
-
  int enqueue(int fd, short event, uint16_t action, uint32_t event_flags, void* udata, bool immediate = false) {
- // if (fd == _kq) debug_breakpoint();
- // immediate = true;
- // LOG_DEBUG(VALUE(_kq), VALUE(fd), VALUE(event), VALUE(action), VALUE(event_flags), VALUE(udata), VALUE(immediate));
  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);
+ struct timespec tm{0, 0};
+ int ret = kevent(_kq, _events, _n, nullptr, 0, &tm);
  if (ret < 0) {
- // debug_breakpoint();
  LOG_ERRNO_RETURN(0, -1, "failed to submit events with kevent()");
  }
  _n = 0;
  }
  return 0;
  }
 
- int wait_for_fd(int fd, uint32_t interests, Timeout timeout) override {
- if (unlikely(interests == 0))
- return 0;
- 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
- }
-
- errno = (ret == 0) ? ETIMEDOUT : err.no;
- return -1;
- }
-
- ssize_t wait_and_fire_events(uint64_t timeout) 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;
@@ -125,7 +104,7 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine, public Res
  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;
@@ -134,6 +113,44 @@ class KQueue : public MasterEventEngine, public CascadingEventEngine, public Res
  return nev;
  }
 
+ int wait_for_fd(int fd, uint32_t interests, Timeout 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.expired()) {
+ 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;