#ifndef _GNU_SOURCE #define _GNU_SOURCE #endif #include "../../include/gateway/io_reactor.hpp" #include #include #include #include #include #if defined(__linux__) #include #elif defined(__APPLE__) #include #include #endif namespace quant::gateway { struct IOReactor::Impl { std::unordered_map callbacks; std::unordered_map registered_events; }; #if defined(__linux__) IOReactor::IOReactor() : pimpl_(std::make_unique()) { reactor_fd_ = epoll_create1(EPOLL_CLOEXEC); } IOReactor::~IOReactor() { if (reactor_fd_ >= 0) { close(reactor_fd_); } } bool IOReactor::add_socket(int fd, IOEvent events, EventCallback callback) { if (reactor_fd_ < 0 || fd < 0) return false; epoll_event ev{}; ev.data.fd = fd; ev.events = EPOLLET; // Edge-triggered if (events & IOEvent::READ) ev.events |= EPOLLIN | EPOLLRDHUP; if (events & IOEvent::WRITE) ev.events |= EPOLLOUT; if (epoll_ctl(reactor_fd_, EPOLL_CTL_ADD, fd, &ev) < 0) { return false; } pimpl_->callbacks[fd] = std::move(callback); pimpl_->registered_events[fd] = events; return true; } bool IOReactor::modify_socket(int fd, IOEvent events) { if (reactor_fd_ < 0 || fd < 0) return false; epoll_event ev{}; ev.data.fd = fd; ev.events = EPOLLET; if (events & IOEvent::READ) ev.events |= EPOLLIN | EPOLLRDHUP; if (events & IOEvent::WRITE) ev.events |= EPOLLOUT; if (epoll_ctl(reactor_fd_, EPOLL_CTL_MOD, fd, &ev) < 0) { return false; } pimpl_->registered_events[fd] = events; return true; } bool IOReactor::remove_socket(int fd) { if (reactor_fd_ < 0 || fd < 0) return false; epoll_ctl(reactor_fd_, EPOLL_CTL_DEL, fd, nullptr); pimpl_->callbacks.erase(fd); pimpl_->registered_events.erase(fd); return true; } int IOReactor::poll_events(int timeout_ms) { if (reactor_fd_ < 0 || !running_) return 0; constexpr int MAX_EVENTS = 64; std::array events; int nfds = epoll_wait(reactor_fd_, events.data(), MAX_EVENTS, timeout_ms); if (nfds <= 0) return nfds; for (int i = 0; i < nfds; ++i) { int fd = events[i].data.fd; auto it = pimpl_->callbacks.find(fd); if (it == pimpl_->callbacks.end()) continue; IOEvent revents = static_cast(0); if (events[i].events & (EPOLLIN | EPOLLPRI)) revents = revents | IOEvent::READ; if (events[i].events & EPOLLOUT) revents = revents | IOEvent::WRITE; if (events[i].events & (EPOLLERR)) revents = revents | IOEvent::ERROR; if (events[i].events & (EPOLLHUP | EPOLLRDHUP)) revents = revents | IOEvent::HANGUP; it->second(fd, revents); } return nfds; } #elif defined(__APPLE__) IOReactor::IOReactor() : pimpl_(std::make_unique()) { reactor_fd_ = kqueue(); } IOReactor::~IOReactor() { if (reactor_fd_ >= 0) { close(reactor_fd_); } } bool IOReactor::add_socket(int fd, IOEvent events, EventCallback callback) { if (reactor_fd_ < 0 || fd < 0) return false; std::array evs; int count = 0; if (events & IOEvent::READ) { EV_SET(&evs[count++], fd, EVFILT_READ, EV_ADD | EV_CLEAR, 0, 0, nullptr); } if (events & IOEvent::WRITE) { EV_SET(&evs[count++], fd, EVFILT_WRITE, EV_ADD | EV_CLEAR, 0, 0, nullptr); } if (kevent(reactor_fd_, evs.data(), count, nullptr, 0, nullptr) < 0) { return false; } pimpl_->callbacks[fd] = std::move(callback); pimpl_->registered_events[fd] = events; return true; } bool IOReactor::modify_socket(int fd, IOEvent events) { if (reactor_fd_ < 0 || fd < 0) return false; // Reset filters std::array evs; int count = 0; auto cur = pimpl_->registered_events[fd]; if (cur & IOEvent::READ && !(events & IOEvent::READ)) { EV_SET(&evs[count++], fd, EVFILT_READ, EV_DELETE, 0, 0, nullptr); } else if (!(cur & IOEvent::READ) && (events & IOEvent::READ)) { EV_SET(&evs[count++], fd, EVFILT_READ, EV_ADD | EV_CLEAR, 0, 0, nullptr); } if (cur & IOEvent::WRITE && !(events & IOEvent::WRITE)) { EV_SET(&evs[count++], fd, EVFILT_WRITE, EV_DELETE, 0, 0, nullptr); } else if (!(cur & IOEvent::WRITE) && (events & IOEvent::WRITE)) { EV_SET(&evs[count++], fd, EVFILT_WRITE, EV_ADD | EV_CLEAR, 0, 0, nullptr); } if (count > 0) { kevent(reactor_fd_, evs.data(), count, nullptr, 0, nullptr); } pimpl_->registered_events[fd] = events; return true; } bool IOReactor::remove_socket(int fd) { if (reactor_fd_ < 0 || fd < 0) return false; std::array evs; int count = 0; auto cur = pimpl_->registered_events[fd]; if (cur & IOEvent::READ) { EV_SET(&evs[count++], fd, EVFILT_READ, EV_DELETE, 0, 0, nullptr); } if (cur & IOEvent::WRITE) { EV_SET(&evs[count++], fd, EVFILT_WRITE, EV_DELETE, 0, 0, nullptr); } if (count > 0) { kevent(reactor_fd_, evs.data(), count, nullptr, 0, nullptr); } pimpl_->callbacks.erase(fd); pimpl_->registered_events.erase(fd); return true; } int IOReactor::poll_events(int timeout_ms) { if (reactor_fd_ < 0 || !running_) return 0; constexpr int MAX_EVENTS = 64; std::array events; struct timespec ts{}; struct timespec* timeout_ptr = nullptr; if (timeout_ms >= 0) { ts.tv_sec = timeout_ms / 1000; ts.tv_nsec = (timeout_ms % 1000) * 1000000L; timeout_ptr = &ts; } int nfds = kevent(reactor_fd_, nullptr, 0, events.data(), MAX_EVENTS, timeout_ptr); if (nfds <= 0) return nfds; for (int i = 0; i < nfds; ++i) { int fd = static_cast(events[i].ident); auto it = pimpl_->callbacks.find(fd); if (it == pimpl_->callbacks.end()) continue; IOEvent revents = static_cast(0); if (events[i].filter == EVFILT_READ) revents = revents | IOEvent::READ; if (events[i].filter == EVFILT_WRITE) revents = revents | IOEvent::WRITE; if (events[i].flags & EV_EOF) revents = revents | IOEvent::HANGUP; if (events[i].flags & EV_ERROR) revents = revents | IOEvent::ERROR; it->second(fd, revents); } return nfds; } #endif } // namespace quant::gateway