diff options
Diffstat (limited to '')
| -rw-r--r-- | xsk.cc | 1262 |
1 files changed, 1262 insertions, 0 deletions
@@ -0,0 +1,1262 @@ +/* + * This file is part of PowerDNS or dnsdist. + * Copyright -- PowerDNS.COM B.V. and its contributors + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of version 2 of the GNU General Public License as + * published by the Free Software Foundation. + * + * In addition, for the avoidance of any doubt, permission is granted to + * link this program with OpenSSL and to (re)distribute the binaries + * produced as the result of such linking. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + */ + +#include "config.h" + +#ifdef HAVE_XSK + +#include <algorithm> +#include <cstdint> +#include <cstring> +#include <fcntl.h> +#include <iterator> +#include <linux/bpf.h> +#include <linux/if_ether.h> +#include <linux/if_link.h> +#include <linux/if_xdp.h> +#include <linux/ip.h> +#include <linux/ipv6.h> +#include <linux/tcp.h> +#include <linux/types.h> +#include <linux/udp.h> +#include <net/if.h> +#include <net/if_arp.h> +#include <netinet/in.h> +#include <poll.h> +#include <stdexcept> +#include <sys/eventfd.h> +#include <sys/ioctl.h> +#include <sys/mman.h> +#include <sys/socket.h> +#include <sys/timerfd.h> +#include <unistd.h> +#include <vector> + +#include <bpf/bpf.h> +#include <bpf/libbpf.h> +extern "C" +{ +#include <xdp/libxdp.h> +} + +#include "gettime.hh" +#include "xsk.hh" + +#ifdef DEBUG_UMEM +namespace +{ +struct UmemEntryStatus +{ + enum class Status : uint8_t + { + Free, + FillQueue, + Received, + TXQueue + }; + Status status{Status::Free}; +}; + +LockGuarded<std::unordered_map<uint64_t, UmemEntryStatus>> s_umems; + +void checkUmemIntegrity(const char* function, int line, uint64_t offset, const std::set<UmemEntryStatus::Status>& validStatuses, UmemEntryStatus::Status newStatus) +{ + auto umems = s_umems.lock(); + if (validStatuses.count(umems->at(offset).status) == 0) { + std::cerr << "UMEM integrity check failed at " << function << ": " << line << ": status is " << static_cast<int>(umems->at(offset).status) << ", expected: "; + for (const auto status : validStatuses) { + std::cerr << static_cast<int>(status) << " "; + } + std::cerr << std::endl; + abort(); + } + (*umems)[offset].status = newStatus; +} +} +#endif /* DEBUG_UMEM */ + +constexpr bool XskSocket::isPowOfTwo(uint32_t value) noexcept +{ + return value != 0 && (value & (value - 1)) == 0; +} + +int XskSocket::firstTimeout() +{ + if (waitForDelay.empty()) { + return -1; + } + timespec now{}; + gettime(&now); + const auto& firstTime = waitForDelay.top().getSendTime(); + const auto res = timeDifference(now, firstTime); + if (res <= 0) { + return 0; + } + return res; +} + +XskSocket::XskSocket(size_t frameNum_, std::string ifName_, uint32_t queue_id, const std::string& xskMapPath) : + frameNum(frameNum_), ifName(std::move(ifName_)), socket(nullptr, xsk_socket__delete), sharedEmptyFrameOffset(std::make_shared<LockGuarded<vector<uint64_t>>>()) +{ + if (!isPowOfTwo(frameNum_) || !isPowOfTwo(frameSize) + || !isPowOfTwo(fqCapacity) || !isPowOfTwo(cqCapacity) || !isPowOfTwo(rxCapacity) || !isPowOfTwo(txCapacity)) { + throw std::runtime_error("The number of frame , the size of frame and the capacity of rings must is a pow of 2"); + } + getMACFromIfName(); + + memset(&cq, 0, sizeof(cq)); + memset(&fq, 0, sizeof(fq)); + memset(&tx, 0, sizeof(tx)); + memset(&rx, 0, sizeof(rx)); + + xsk_umem_config umemCfg{}; + umemCfg.fill_size = fqCapacity; + umemCfg.comp_size = cqCapacity; + umemCfg.frame_size = frameSize; + umemCfg.frame_headroom = XSK_UMEM__DEFAULT_FRAME_HEADROOM; + umemCfg.flags = 0; + umem.umemInit(frameNum_ * frameSize, &cq, &fq, &umemCfg); + + { + xsk_socket_config socketCfg{}; + socketCfg.rx_size = rxCapacity; + socketCfg.tx_size = txCapacity; + socketCfg.bind_flags = XDP_USE_NEED_WAKEUP; + socketCfg.xdp_flags = XDP_FLAGS_SKB_MODE; + socketCfg.libxdp_flags = XSK_LIBBPF_FLAGS__INHIBIT_PROG_LOAD; + xsk_socket* tmp = nullptr; + auto ret = xsk_socket__create(&tmp, ifName.c_str(), queue_id, umem.umem, &rx, &tx, &socketCfg); + if (ret != 0) { + throw std::runtime_error("Error creating a xsk socket of if_name " + ifName + ": " + stringerror(ret)); + } + socket = std::unique_ptr<xsk_socket, decltype(&xsk_socket__delete)>(tmp, xsk_socket__delete); + } + + uniqueEmptyFrameOffset.reserve(frameNum); + { + for (uint64_t idx = 0; idx < frameNum; idx++) { + uniqueEmptyFrameOffset.push_back(idx * frameSize + XDP_PACKET_HEADROOM); +#ifdef DEBUG_UMEM + { + auto umems = s_umems.lock(); + (*umems)[idx * frameSize + XDP_PACKET_HEADROOM] = UmemEntryStatus(); + } +#endif /* DEBUG_UMEM */ + } + } + + fillFq(fqCapacity); + + const auto xskfd = xskFd(); + fds.push_back(pollfd{ + .fd = xskfd, + .events = POLLIN, + .revents = 0}); + + const auto xskMapFd = FDWrapper(bpf_obj_get(xskMapPath.c_str())); + + if (xskMapFd.getHandle() < 0) { + throw std::runtime_error("Error getting BPF map from path '" + xskMapPath + "'"); + } + + auto ret = bpf_map_update_elem(xskMapFd.getHandle(), &queue_id, &xskfd, 0); + if (ret != 0) { + throw std::runtime_error("Error inserting into xsk_map '" + xskMapPath + "': " + std::to_string(ret)); + } +} + +// see xdp.h in contrib/ +struct IPv4AndPort +{ + uint32_t addr; + uint16_t port; +}; +struct IPv6AndPort +{ + struct in6_addr addr; + uint16_t port; +}; + +static FDWrapper getDestinationMap(const std::string& mapPath) +{ + auto destMapFd = FDWrapper(bpf_obj_get(mapPath.c_str())); + if (destMapFd.getHandle() < 0) { + throw std::runtime_error("Error getting the XSK destination addresses map path '" + mapPath + "'"); + } + return destMapFd; +} + +void XskSocket::clearDestinationMap(const std::string& mapPath, bool isV6) +{ + auto destMapFd = getDestinationMap(mapPath); + if (!isV6) { + IPv4AndPort prevKey{}; + IPv4AndPort key{}; + while (bpf_map_get_next_key(destMapFd.getHandle(), &prevKey, &key) == 0) { + bpf_map_delete_elem(destMapFd.getHandle(), &key); + prevKey = key; + } + } + else { + IPv6AndPort prevKey{}; + IPv6AndPort key{}; + while (bpf_map_get_next_key(destMapFd.getHandle(), &prevKey, &key) == 0) { + bpf_map_delete_elem(destMapFd.getHandle(), &key); + prevKey = key; + } + } +} + +void XskSocket::addDestinationAddress(const std::string& mapPath, const ComboAddress& destination) +{ + auto destMapFd = getDestinationMap(mapPath); + bool value = true; + if (destination.isIPv4()) { + IPv4AndPort key{}; + key.addr = destination.sin4.sin_addr.s_addr; + key.port = destination.sin4.sin_port; + auto ret = bpf_map_update_elem(destMapFd.getHandle(), &key, &value, 0); + if (ret != 0) { + throw std::runtime_error("Error inserting into xsk_map '" + mapPath + "': " + std::to_string(ret)); + } + } + else { + IPv6AndPort key{}; + key.addr = destination.sin6.sin6_addr; + key.port = destination.sin6.sin6_port; + auto ret = bpf_map_update_elem(destMapFd.getHandle(), &key, &value, 0); + if (ret != 0) { + throw std::runtime_error("Error inserting into XSK destination addresses map '" + mapPath + "': " + std::to_string(ret)); + } + } +} + +void XskSocket::removeDestinationAddress(const std::string& mapPath, const ComboAddress& destination) +{ + auto destMapFd = getDestinationMap(mapPath); + if (destination.isIPv4()) { + IPv4AndPort key{}; + key.addr = destination.sin4.sin_addr.s_addr; + key.port = destination.sin4.sin_port; + bpf_map_delete_elem(destMapFd.getHandle(), &key); + } + else { + IPv6AndPort key{}; + key.addr = destination.sin6.sin6_addr; + key.port = destination.sin6.sin6_port; + bpf_map_delete_elem(destMapFd.getHandle(), &key); + } +} + +void XskSocket::fillFq(uint32_t fillSize) noexcept +{ + { + // if we have less than holdThreshold frames in the shared queue (which might be an issue + // when the XskWorker needs empty frames), move frames from the unique container into the + // shared one. This might not be optimal right now. + auto frames = sharedEmptyFrameOffset->lock(); + if (frames->size() < holdThreshold) { + const auto moveSize = std::min(holdThreshold - frames->size(), uniqueEmptyFrameOffset.size()); + if (moveSize > 0) { + // NOLINTNEXTLINE(bugprone-narrowing-conversions,cppcoreguidelines-narrowing-conversions) + frames->insert(frames->end(), std::make_move_iterator(uniqueEmptyFrameOffset.end() - moveSize), std::make_move_iterator(uniqueEmptyFrameOffset.end())); + uniqueEmptyFrameOffset.resize(uniqueEmptyFrameOffset.size() - moveSize); + } + } + } + + if (uniqueEmptyFrameOffset.size() < fillSize) { + return; + } + + uint32_t idx{0}; + auto toFill = xsk_ring_prod__reserve(&fq, fillSize, &idx); + if (toFill == 0) { + return; + } + uint32_t processed = 0; + for (; processed < toFill; processed++) { + *xsk_ring_prod__fill_addr(&fq, idx++) = uniqueEmptyFrameOffset.back(); +#ifdef DEBUG_UMEM + checkUmemIntegrity(__PRETTY_FUNCTION__, __LINE__, uniqueEmptyFrameOffset.back(), {UmemEntryStatus::Status::Free}, UmemEntryStatus::Status::FillQueue); +#endif /* DEBUG_UMEM */ + uniqueEmptyFrameOffset.pop_back(); + } + + xsk_ring_prod__submit(&fq, processed); +} + +int XskSocket::wait(int timeout) +{ + auto waitAtMost = std::min(timeout, firstTimeout()); + return poll(fds.data(), fds.size(), waitAtMost); +} + +[[nodiscard]] uint64_t XskSocket::frameOffset(const XskPacket& packet) const noexcept +{ + return packet.getFrameOffsetFrom(umem.bufBase); +} + +[[nodiscard]] int XskSocket::xskFd() const noexcept +{ + return xsk_socket__fd(socket.get()); +} + +void XskSocket::send(std::vector<XskPacket>& packets) +{ + while (!packets.empty()) { + auto packetSize = packets.size(); + if (packetSize > std::numeric_limits<uint32_t>::max()) { + packetSize = std::numeric_limits<uint32_t>::max(); + } + size_t toSend = std::min(static_cast<uint32_t>(packetSize), txCapacity); + uint32_t idx{0}; + auto toFill = xsk_ring_prod__reserve(&tx, toSend, &idx); + if (toFill == 0) { + return; + } + + size_t queued = 0; + for (const auto& packet : packets) { + if (queued == toFill) { + break; + } + *xsk_ring_prod__tx_desc(&tx, idx++) = { + .addr = frameOffset(packet), + .len = packet.getFrameLen(), + .options = 0}; +#ifdef DEBUG_UMEM + checkUmemIntegrity(__PRETTY_FUNCTION__, __LINE__, frameOffset(packet), {UmemEntryStatus::Status::Free, UmemEntryStatus::Status::Received}, UmemEntryStatus::Status::TXQueue); +#endif /* DEBUG_UMEM */ + queued++; + } + xsk_ring_prod__submit(&tx, toFill); + packets.erase(packets.begin(), packets.begin() + toFill); + } +} + +std::vector<XskPacket> XskSocket::recv(uint32_t recvSizeMax, uint32_t* failedCount) +{ + uint32_t idx{0}; + std::vector<XskPacket> res; + // how many descriptors to packets have been filled + const auto recvSize = xsk_ring_cons__peek(&rx, recvSizeMax, &idx); + if (recvSize == 0) { + return res; + } + + // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast) + const auto baseAddr = reinterpret_cast<uint64_t>(umem.bufBase); + uint32_t failed = 0; + uint32_t processed = 0; + res.reserve(recvSize); + for (; processed < recvSize; processed++) { + try { + const auto* desc = xsk_ring_cons__rx_desc(&rx, idx++); + // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast,performance-no-int-to-ptr) + XskPacket packet = XskPacket(reinterpret_cast<uint8_t*>(desc->addr + baseAddr), desc->len, frameSize); +#ifdef DEBUG_UMEM + checkUmemIntegrity(__PRETTY_FUNCTION__, __LINE__, frameOffset(packet), {UmemEntryStatus::Status::Free, UmemEntryStatus::Status::FillQueue}, UmemEntryStatus::Status::Received); +#endif /* DEBUG_UMEM */ + + if (!packet.parse(false)) { + ++failed; + markAsFree(packet); + } + else { + res.push_back(packet); + } + } + catch (const std::exception& exp) { + std::cerr << "Exception while processing the XSK RX queue: " << exp.what() << std::endl; + break; + } + catch (...) { + std::cerr << "Exception while processing the XSK RX queue" << std::endl; + break; + } + } + + // this releases the descriptor, but not the packet (umem entries) + // which will only be made available again when pushed into the fill + // queue + xsk_ring_cons__release(&rx, processed); + if (failedCount != nullptr) { + *failedCount = failed; + } + + return res; +} + +void XskSocket::pickUpReadyPacket(std::vector<XskPacket>& packets) +{ + timespec now{}; + gettime(&now); + while (!waitForDelay.empty() && timeDifference(now, waitForDelay.top().getSendTime()) <= 0) { + // NOLINTNEXTLINE(cppcoreguidelines-pro-type-const-cast) + auto& top = const_cast<XskPacket&>(waitForDelay.top()); + packets.push_back(top); + waitForDelay.pop(); + } +} + +void XskSocket::recycle(size_t size) noexcept +{ + uint32_t idx{0}; + const auto completeSize = xsk_ring_cons__peek(&cq, size, &idx); + if (completeSize == 0) { + return; + } + uniqueEmptyFrameOffset.reserve(uniqueEmptyFrameOffset.size() + completeSize); + uint32_t processed = 0; + for (; processed < completeSize; ++processed) { + uniqueEmptyFrameOffset.push_back(*xsk_ring_cons__comp_addr(&cq, idx++)); +#ifdef DEBUG_UMEM + checkUmemIntegrity(__PRETTY_FUNCTION__, __LINE__, uniqueEmptyFrameOffset.back(), {UmemEntryStatus::Status::Received, UmemEntryStatus::Status::TXQueue}, UmemEntryStatus::Status::Free); +#endif /* DEBUG_UMEM */ + } + xsk_ring_cons__release(&cq, processed); +} + +void XskSocket::XskUmem::umemInit(size_t memSize, xsk_ring_cons* completionQueue, xsk_ring_prod* fillQueue, xsk_umem_config* config) +{ + size = memSize; + bufBase = static_cast<uint8_t*>(mmap(nullptr, size, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0)); + if (bufBase == MAP_FAILED) { + throw std::runtime_error("mmap failed"); + } + auto ret = xsk_umem__create(&umem, bufBase, size, fillQueue, completionQueue, config); + if (ret != 0) { + munmap(bufBase, size); + throw std::runtime_error("Error creating a umem of size " + std::to_string(size) + ": " + stringerror(ret)); + } +} + +std::string XskSocket::getMetrics() const +{ + xdp_statistics stats{}; + socklen_t optlen = sizeof(stats); + int err = getsockopt(xskFd(), SOL_XDP, XDP_STATISTICS, &stats, &optlen); + if (err != 0) { + return ""; + } + if (optlen != sizeof(struct xdp_statistics)) { + return ""; + } + + ostringstream ret; + ret << "RX dropped: " << std::to_string(stats.rx_dropped) << std::endl; + ret << "RX invalid descs: " << std::to_string(stats.rx_invalid_descs) << std::endl; + ret << "TX invalid descs: " << std::to_string(stats.tx_invalid_descs) << std::endl; + ret << "RX ring full: " << std::to_string(stats.rx_ring_full) << std::endl; + ret << "RX fill ring empty descs: " << std::to_string(stats.rx_fill_ring_empty_descs) << std::endl; + ret << "TX ring empty descs: " << std::to_string(stats.tx_ring_empty_descs) << std::endl; + return ret.str(); +} + +[[nodiscard]] std::string XskSocket::getXDPMode() const +{ +#ifdef HAVE_BPF_XDP_QUERY + unsigned int itfIdx = if_nametoindex(ifName.c_str()); + if (itfIdx == 0) { + return "unable to get interface index"; + } + bpf_xdp_query_opts info{}; + info.sz = sizeof(info); + int ret = bpf_xdp_query(static_cast<int>(itfIdx), 0, &info); + if (ret != 0) { + return {}; + } + switch (info.attach_mode) { + case XDP_ATTACHED_DRV: + case XDP_ATTACHED_HW: + return "native"; + case XDP_ATTACHED_SKB: + return "emulated"; + default: + return "unknown"; + } +#else /* HAVE_BPF_XDP_QUERY */ + return "undetected"; +#endif /* HAVE_BPF_XDP_QUERY */ +} + +void XskSocket::markAsFree(const XskPacket& packet) +{ + auto offset = frameOffset(packet); +#ifdef DEBUG_UMEM + checkUmemIntegrity(__PRETTY_FUNCTION__, __LINE__, offset, {UmemEntryStatus::Status::Received, UmemEntryStatus::Status::TXQueue}, UmemEntryStatus::Status::Free); +#endif /* DEBUG_UMEM */ + + uniqueEmptyFrameOffset.push_back(offset); +} + +XskSocket::XskUmem::~XskUmem() +{ + if (umem != nullptr) { + xsk_umem__delete(umem); + } + if (bufBase != nullptr) { + munmap(bufBase, size); + } +} + +[[nodiscard]] size_t XskPacket::getL4HeaderOffset() const noexcept +{ + return sizeof(ethhdr) + (v6 ? (sizeof(ipv6hdr)) : sizeof(iphdr)); +} + +[[nodiscard]] size_t XskPacket::getDataOffset() const noexcept +{ + return getL4HeaderOffset() + sizeof(udphdr); +} + +[[nodiscard]] size_t XskPacket::getDataSize() const noexcept +{ + return frameLength - getDataOffset(); +} + +[[nodiscard]] ethhdr XskPacket::getEthernetHeader() const noexcept +{ + ethhdr ethHeader{}; + if (frameLength >= sizeof(ethHeader)) { + memcpy(ðHeader, frame, sizeof(ethHeader)); + } + return ethHeader; +} + +void XskPacket::setEthernetHeader(const ethhdr& ethHeader) noexcept +{ + if (frameLength < sizeof(ethHeader)) { + frameLength = sizeof(ethHeader); + } + memcpy(frame, ðHeader, sizeof(ethHeader)); +} + +[[nodiscard]] iphdr XskPacket::getIPv4Header() const noexcept +{ + iphdr ipv4Header{}; + assert(frameLength >= (sizeof(ethhdr) + sizeof(ipv4Header))); + assert(!v6); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) + memcpy(&ipv4Header, frame + sizeof(ethhdr), sizeof(ipv4Header)); + return ipv4Header; +} + +void XskPacket::setIPv4Header(const iphdr& ipv4Header) noexcept +{ + assert(frameLength >= (sizeof(ethhdr) + sizeof(iphdr))); + assert(!v6); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) + memcpy(frame + sizeof(ethhdr), &ipv4Header, sizeof(ipv4Header)); +} + +[[nodiscard]] ipv6hdr XskPacket::getIPv6Header() const noexcept +{ + ipv6hdr ipv6Header{}; + assert(frameLength >= (sizeof(ethhdr) + sizeof(ipv6Header))); + assert(v6); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) + memcpy(&ipv6Header, frame + sizeof(ethhdr), sizeof(ipv6Header)); + return ipv6Header; +} + +void XskPacket::setIPv6Header(const ipv6hdr& ipv6Header) noexcept +{ + assert(frameLength >= (sizeof(ethhdr) + sizeof(ipv6Header))); + assert(v6); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) + memcpy(frame + sizeof(ethhdr), &ipv6Header, sizeof(ipv6Header)); +} + +[[nodiscard]] udphdr XskPacket::getUDPHeader() const noexcept +{ + udphdr udpHeader{}; + assert(frameLength >= (sizeof(ethhdr) + (v6 ? sizeof(ipv6hdr) : sizeof(iphdr)) + sizeof(udpHeader))); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) + memcpy(&udpHeader, frame + getL4HeaderOffset(), sizeof(udpHeader)); + return udpHeader; +} + +void XskPacket::setUDPHeader(const udphdr& udpHeader) noexcept +{ + assert(frameLength >= (sizeof(ethhdr) + (v6 ? sizeof(ipv6hdr) : sizeof(iphdr)) + sizeof(udpHeader))); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) + memcpy(frame + getL4HeaderOffset(), &udpHeader, sizeof(udpHeader)); +} + +bool XskPacket::parse(bool fromSetHeader) +{ + if (frameLength <= sizeof(ethhdr)) { + return false; + } + + auto ethHeader = getEthernetHeader(); + uint8_t l4Protocol{0}; + if (ethHeader.h_proto == htons(ETH_P_IP)) { + if (frameLength < (sizeof(ethhdr) + sizeof(iphdr) + sizeof(udphdr))) { + return false; + } + v6 = false; + auto ipHeader = getIPv4Header(); + if (ipHeader.ihl != (static_cast<uint8_t>(sizeof(iphdr) / 4))) { + // ip options is not supported now! + return false; + } + // check ip.check == ipv4Checksum() is not needed! + // We check it in BPF program + // we don't, actually. + // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast) + from = makeComboAddressFromRaw(4, reinterpret_cast<const char*>(&ipHeader.saddr), sizeof(ipHeader.saddr)); + // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast) + to = makeComboAddressFromRaw(4, reinterpret_cast<const char*>(&ipHeader.daddr), sizeof(ipHeader.daddr)); + l4Protocol = ipHeader.protocol; + if (!fromSetHeader && (frameLength - sizeof(ethhdr)) != ntohs(ipHeader.tot_len)) { + // too small, or too large (trailing data), go away + return false; + } + } + else if (ethHeader.h_proto == htons(ETH_P_IPV6)) { + if (frameLength < (sizeof(ethhdr) + sizeof(ipv6hdr) + sizeof(udphdr))) { + return false; + } + v6 = true; + auto ipHeader = getIPv6Header(); + // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast) + from = makeComboAddressFromRaw(6, reinterpret_cast<const char*>(&ipHeader.saddr), sizeof(ipHeader.saddr)); + // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast) + to = makeComboAddressFromRaw(6, reinterpret_cast<const char*>(&ipHeader.daddr), sizeof(ipHeader.daddr)); + l4Protocol = ipHeader.nexthdr; + if (!fromSetHeader && (frameLength - (sizeof(ethhdr) + sizeof(ipv6hdr))) != ntohs(ipHeader.payload_len)) { + return false; + } + } + else { + return false; + } + + if (l4Protocol != IPPROTO_UDP) { + return false; + } + + // check udp.check == ipv4Checksum() is not needed! + // We check it in BPF program + // we don't, actually. + auto udpHeader = getUDPHeader(); + if (!fromSetHeader) { + // Because of XskPacket::setHeader + if (getDataOffset() > frameLength) { + return false; + } + + if (getDataSize() + sizeof(udphdr) != ntohs(udpHeader.len)) { + return false; + } + } + + from.setPort(ntohs(udpHeader.source)); + to.setPort(ntohs(udpHeader.dest)); + return true; +} + +uint32_t XskPacket::getDataLen() const noexcept +{ + return getDataSize(); +} + +uint32_t XskPacket::getFrameLen() const noexcept +{ + return frameLength; +} + +size_t XskPacket::getCapacity() const noexcept +{ + return frameSize; +} + +void XskPacket::changeDirectAndUpdateChecksum() noexcept +{ + auto ethHeader = getEthernetHeader(); + { + std::array<uint8_t, ETH_ALEN> tmp{}; + static_assert(tmp.size() == sizeof(ethHeader.h_dest), "Size Error"); + static_assert(tmp.size() == sizeof(ethHeader.h_source), "Size Error"); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-array-to-pointer-decay) + memcpy(tmp.data(), ethHeader.h_dest, tmp.size()); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-array-to-pointer-decay) + memcpy(ethHeader.h_dest, ethHeader.h_source, tmp.size()); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-array-to-pointer-decay) + memcpy(ethHeader.h_source, tmp.data(), tmp.size()); + } + if (ethHeader.h_proto == htons(ETH_P_IPV6)) { + // IPV6 + auto ipv6 = getIPv6Header(); + std::swap(ipv6.daddr, ipv6.saddr); + assert(ipv6.nexthdr == IPPROTO_UDP); + + auto udp = getUDPHeader(); + std::swap(udp.dest, udp.source); + udp.len = htons(getDataSize() + sizeof(udp)); + udp.check = 0; + /* needed to get the correct checksum */ + setIPv6Header(ipv6); + setUDPHeader(udp); + udp.check = tcp_udp_v6_checksum(&ipv6); + rewriteIpv6Header(&ipv6, getFrameLen()); + setIPv6Header(ipv6); + setUDPHeader(udp); + } + else { + // IPV4 + auto ipv4 = getIPv4Header(); + std::swap(ipv4.daddr, ipv4.saddr); + assert(ipv4.protocol == IPPROTO_UDP); + + auto udp = getUDPHeader(); + std::swap(udp.dest, udp.source); + udp.len = htons(getDataSize() + sizeof(udp)); + udp.check = 0; + /* needed to get the correct checksum */ + setIPv4Header(ipv4); + setUDPHeader(udp); + udp.check = tcp_udp_v4_checksum(&ipv4); + rewriteIpv4Header(&ipv4, getFrameLen()); + setIPv4Header(ipv4); + setUDPHeader(udp); + } + setEthernetHeader(ethHeader); +} + +void XskPacket::rewriteIpv4Header(struct iphdr* ipv4header, size_t frameLen) noexcept +{ + ipv4header->version = 4; + ipv4header->ihl = sizeof(iphdr) / 4; + ipv4header->tos = 0; + ipv4header->tot_len = htons(frameLen - sizeof(ethhdr)); + ipv4header->id = 0; + ipv4header->frag_off = 0; + ipv4header->ttl = DefaultTTL; + ipv4header->check = 0; + ipv4header->check = ipv4Checksum(ipv4header); +} + +void XskPacket::rewriteIpv6Header(struct ipv6hdr* ipv6header, size_t frameLen) noexcept +{ + ipv6header->version = 6; + ipv6header->priority = 0; + ipv6header->payload_len = htons(frameLen - sizeof(ethhdr) - sizeof(ipv6hdr)); + ipv6header->hop_limit = DefaultTTL; + memset(&ipv6header->flow_lbl, 0, sizeof(ipv6header->flow_lbl)); +} + +bool XskPacket::isIPV6() const noexcept +{ + return v6; +} + +XskPacket::XskPacket(uint8_t* frame_, size_t dataSize, size_t frameSize_) : + frame(frame_), frameLength(dataSize), frameSize(frameSize_ - XDP_PACKET_HEADROOM) +{ +} + +PacketBuffer XskPacket::clonePacketBuffer() const +{ + const auto size = getDataSize(); + PacketBuffer tmp(size); + if (size > 0) { + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) + memcpy(tmp.data(), frame + getDataOffset(), size); + } + return tmp; +} + +bool XskPacket::setPayload(const PacketBuffer& buf) +{ + const auto bufSize = buf.size(); + const auto currentCapacity = getCapacity(); + if (bufSize == 0 || bufSize > currentCapacity) { + return false; + } + flags |= UPDATE; + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) + memcpy(frame + getDataOffset(), buf.data(), bufSize); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) + frameLength = getDataOffset() + bufSize; + return true; +} + +void XskPacket::addDelay(const int relativeMilliseconds) noexcept +{ + gettime(&sendTime); + sendTime.tv_nsec += static_cast<int64_t>(relativeMilliseconds) * 1000000L; + sendTime.tv_sec += sendTime.tv_nsec / 1000000000L; + sendTime.tv_nsec %= 1000000000L; +} + +bool operator<(const XskPacket& lhs, const XskPacket& rhs) noexcept +{ + return lhs.getSendTime() < rhs.getSendTime(); +} + +const ComboAddress& XskPacket::getFromAddr() const noexcept +{ + return from; +} + +const ComboAddress& XskPacket::getToAddr() const noexcept +{ + return to; +} + +void XskWorker::notify(int desc) +{ + uint64_t value = 1; + ssize_t res = 0; + while ((res = write(desc, &value, sizeof(value))) == EINTR) { + } + if (res != sizeof(value)) { + throw runtime_error("Unable Wake Up XskSocket Failed"); + } +} + +XskWorker::XskWorker() : + workerWaker(createEventfd()), xskSocketWaker(createEventfd()) +{ +} + +void XskWorker::pushToProcessingQueue(XskPacket& packet) +{ +#if defined(__SANITIZE_THREAD__) + if (!incomingPacketsQueue.lock()->push(packet)) { +#else + if (!incomingPacketsQueue.push(packet)) { +#endif + markAsFree(packet); + } +} + +void XskWorker::pushToSendQueue(XskPacket& packet) +{ +#if defined(__SANITIZE_THREAD__) + if (!outgoingPacketsQueue.lock()->push(packet)) { +#else + if (!outgoingPacketsQueue.push(packet)) { +#endif + markAsFree(packet); + } +} + +const void* XskPacket::getPayloadData() const +{ + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) + return frame + getDataOffset(); +} + +void XskPacket::setAddr(const ComboAddress& from_, MACAddr fromMAC, const ComboAddress& to_, MACAddr toMAC) noexcept +{ + auto ethHeader = getEthernetHeader(); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-array-to-pointer-decay) + memcpy(ethHeader.h_dest, toMAC.data(), toMAC.size()); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-array-to-pointer-decay) + memcpy(ethHeader.h_source, fromMAC.data(), fromMAC.size()); + setEthernetHeader(ethHeader); + to = to_; + from = from_; + v6 = !to.isIPv4(); + flags = 0; +} + +void XskPacket::rewrite() noexcept +{ + flags |= REWRITE; + auto ethHeader = getEthernetHeader(); + if (!v6) { + ethHeader.h_proto = htons(ETH_P_IP); + + auto ipHeader = getIPv4Header(); + ipHeader.daddr = to.sin4.sin_addr.s_addr; + ipHeader.saddr = from.sin4.sin_addr.s_addr; + + auto udpHeader = getUDPHeader(); + ipHeader.protocol = IPPROTO_UDP; + udpHeader.source = from.sin4.sin_port; + udpHeader.dest = to.sin4.sin_port; + udpHeader.len = htons(getDataSize() + sizeof(udpHeader)); + udpHeader.check = 0; + /* needed to get the correct checksum */ + setIPv4Header(ipHeader); + setUDPHeader(udpHeader); + udpHeader.check = tcp_udp_v4_checksum(&ipHeader); + rewriteIpv4Header(&ipHeader, getFrameLen()); + setIPv4Header(ipHeader); + setUDPHeader(udpHeader); + } + else { + ethHeader.h_proto = htons(ETH_P_IPV6); + + auto ipHeader = getIPv6Header(); + memcpy(&ipHeader.daddr, &to.sin6.sin6_addr, sizeof(ipHeader.daddr)); + memcpy(&ipHeader.saddr, &from.sin6.sin6_addr, sizeof(ipHeader.saddr)); + + auto udpHeader = getUDPHeader(); + ipHeader.nexthdr = IPPROTO_UDP; + udpHeader.source = from.sin6.sin6_port; + udpHeader.dest = to.sin6.sin6_port; + udpHeader.len = htons(getDataSize() + sizeof(udpHeader)); + udpHeader.check = 0; + /* needed to get the correct checksum */ + setIPv6Header(ipHeader); + setUDPHeader(udpHeader); + udpHeader.check = tcp_udp_v6_checksum(&ipHeader); + setIPv6Header(ipHeader); + setUDPHeader(udpHeader); + } + + setEthernetHeader(ethHeader); +} + +[[nodiscard]] __be16 XskPacket::ipv4Checksum(const struct iphdr* ipHeader) noexcept +{ + auto partial = ip_checksum_partial(ipHeader, sizeof(iphdr), 0); + return ip_checksum_fold(partial); +} + +[[nodiscard]] __be16 XskPacket::tcp_udp_v4_checksum(const struct iphdr* ipHeader) const noexcept +{ + // ip options is not supported !!! + const auto l4Length = static_cast<uint16_t>(getDataSize() + sizeof(udphdr)); + auto sum = tcp_udp_v4_header_checksum_partial(ipHeader->saddr, ipHeader->daddr, ipHeader->protocol, l4Length); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) + sum = ip_checksum_partial(frame + getL4HeaderOffset(), l4Length, sum); + return ip_checksum_fold(sum); +} + +[[nodiscard]] __be16 XskPacket::tcp_udp_v6_checksum(const struct ipv6hdr* ipv6) const noexcept +{ + const auto l4Length = static_cast<uint16_t>(getDataSize() + sizeof(udphdr)); + uint64_t sum = tcp_udp_v6_header_checksum_partial(&ipv6->saddr, &ipv6->daddr, ipv6->nexthdr, l4Length); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) + sum = ip_checksum_partial(frame + getL4HeaderOffset(), l4Length, sum); + return ip_checksum_fold(sum); +} + +[[nodiscard]] uint64_t XskPacket::ip_checksum_partial(const void* ptr, const size_t len, uint64_t sum) noexcept +{ + size_t position{0}; + /* Main loop: 32 bits at a time */ + for (position = 0; position < len; position += sizeof(uint32_t)) { + uint32_t value{}; + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) + memcpy(&value, static_cast<const uint8_t*>(ptr) + position, sizeof(value)); + sum += value; + } + + /* Handle un-32bit-aligned trailing bytes */ + if ((len - position) >= 2) { + uint16_t value{}; + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) + memcpy(&value, static_cast<const uint8_t*>(ptr) + position, sizeof(value)); + sum += value; + position += sizeof(value); + } + + if ((len - position) > 0) { + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) + const auto* ptr8 = static_cast<const uint8_t*>(ptr) + position; + sum += ntohs(*ptr8 << 8); /* RFC says pad last byte */ + } + + return sum; +} + +[[nodiscard]] __be16 XskPacket::ip_checksum_fold(uint64_t sum) noexcept +{ + while ((sum & ~0xffffffffULL) != 0U) { + sum = (sum >> 32) + (sum & 0xffffffffULL); + } + while ((sum & 0xffff0000ULL) != 0U) { + sum = (sum >> 16) + (sum & 0xffffULL); + } + + return static_cast<__be16>(~sum); +} + +#ifndef __packed +#define packed_attribute __attribute__((packed)) +#else +#define packed_attribute __packed +#endif + +// NOLINTNEXTLINE(bugprone-easily-swappable-parameters) +[[nodiscard]] uint64_t XskPacket::tcp_udp_v4_header_checksum_partial(__be32 src_ip, __be32 dst_ip, uint8_t protocol, uint16_t len) noexcept +{ + struct header + { + __be32 src_ip; + __be32 dst_ip; + __uint8_t mbz; + __uint8_t protocol; + __be16 length; + }; + /* The IPv4 pseudo-header is defined in RFC 793, Section 3.1. */ + struct ipv4_pseudo_header_t + { + /* We use a union here to avoid aliasing issues with gcc -O2 */ + union + { + header packed_attribute fields; + // NOLINTNEXTLINE(cppcoreguidelines-avoid-c-arrays,modernize-avoid-c-arrays) + uint32_t words[3]; + }; + }; + ipv4_pseudo_header_t pseudo_header{}; + static_assert(sizeof(pseudo_header) == 12, "IPv4 pseudo-header size is incorrect"); + + /* Fill in the pseudo-header. */ + pseudo_header.fields.src_ip = src_ip; + pseudo_header.fields.dst_ip = dst_ip; + pseudo_header.fields.mbz = 0; + pseudo_header.fields.protocol = protocol; + pseudo_header.fields.length = htons(len); + return ip_checksum_partial(&pseudo_header, sizeof(pseudo_header), 0); +} + +[[nodiscard]] uint64_t XskPacket::tcp_udp_v6_header_checksum_partial(const struct in6_addr* src_ip, const struct in6_addr* dst_ip, uint8_t protocol, uint32_t len) noexcept +{ + struct header + { + struct in6_addr src_ip; + struct in6_addr dst_ip; + __be32 length; + // NOLINTNEXTLINE(cppcoreguidelines-avoid-c-arrays,modernize-avoid-c-arrays) + __uint8_t mbz[3]; + __uint8_t next_header; + }; + /* The IPv6 pseudo-header is defined in RFC 2460, Section 8.1. */ + struct ipv6_pseudo_header_t + { + /* We use a union here to avoid aliasing issues with gcc -O2 */ + union + { + header packed_attribute fields; + // NOLINTNEXTLINE(cppcoreguidelines-avoid-c-arrays,modernize-avoid-c-arrays) + uint32_t words[10]; + }; + }; + ipv6_pseudo_header_t pseudo_header{}; + static_assert(sizeof(pseudo_header) == 40, "IPv6 pseudo-header size is incorrect"); + + /* Fill in the pseudo-header. */ + pseudo_header.fields.src_ip = *src_ip; + pseudo_header.fields.dst_ip = *dst_ip; + pseudo_header.fields.length = htonl(len); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-array-to-pointer-decay) + memset(pseudo_header.fields.mbz, 0, sizeof(pseudo_header.fields.mbz)); + pseudo_header.fields.next_header = protocol; + return ip_checksum_partial(&pseudo_header, sizeof(pseudo_header), 0); +} + +void XskPacket::setHeader(PacketBuffer& buf) +{ + memcpy(frame, buf.data(), buf.size()); + frameLength = buf.size(); + buf.clear(); + flags = 0; + if (!parse(true)) { + throw std::runtime_error("Error setting the XSK frame header"); + } +} + +PacketBuffer XskPacket::cloneHeaderToPacketBuffer() const +{ + const auto size = getFrameLen() - getDataSize(); + PacketBuffer tmp(size); + memcpy(tmp.data(), frame, size); + return tmp; +} + +int XskWorker::createEventfd() +{ + auto desc = ::eventfd(0, EFD_CLOEXEC); + if (desc < 0) { + throw runtime_error("Unable create eventfd"); + } + return desc; +} + +void XskWorker::waitForXskSocket() const noexcept +{ + uint64_t value = read(workerWaker, &value, sizeof(value)); +} + +void XskWorker::notifyXskSocket() const +{ + notify(xskSocketWaker); +} + +std::shared_ptr<XskWorker> XskWorker::create() +{ + return std::make_shared<XskWorker>(); +} + +void XskSocket::addWorker(std::shared_ptr<XskWorker> worker) +{ + const auto socketWaker = worker->xskSocketWaker.getHandle(); + worker->umemBufBase = umem.bufBase; + d_workers.insert({socketWaker, std::move(worker)}); + fds.push_back(pollfd{ + .fd = socketWaker, + .events = POLLIN, + .revents = 0}); +}; + +void XskSocket::addWorkerRoute(const std::shared_ptr<XskWorker>& worker, const ComboAddress& dest) +{ + d_workerRoutes.lock()->insert({dest, worker}); +} + +void XskSocket::removeWorkerRoute(const ComboAddress& dest) +{ + d_workerRoutes.lock()->erase(dest); +} + +uint64_t XskWorker::frameOffset(const XskPacket& packet) const noexcept +{ + return packet.getFrameOffsetFrom(umemBufBase); +} + +void XskWorker::notifyWorker() const +{ + notify(workerWaker); +} + +void XskSocket::getMACFromIfName() +{ + ifreq ifr{}; + auto desc = FDWrapper(::socket(AF_INET, SOCK_DGRAM, 0)); + if (desc < 0) { + throw std::runtime_error("Error creating a socket to get the MAC address of interface " + ifName); + } + + if (ifName.size() >= IFNAMSIZ) { + throw std::runtime_error("Unable to get MAC address for interface " + ifName + ": name too long"); + } + + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-array-to-pointer-decay) + strncpy(ifr.ifr_name, ifName.c_str(), ifName.length() + 1); + if (ioctl(desc.getHandle(), SIOCGIFHWADDR, &ifr) < 0 || ifr.ifr_hwaddr.sa_family != ARPHRD_ETHER) { + throw std::runtime_error("Error getting MAC address for interface " + ifName); + } + static_assert(sizeof(ifr.ifr_hwaddr.sa_data) >= std::tuple_size<decltype(source)>{}, "The size of an ARPHRD_ETHER MAC address is smaller than expected"); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-array-to-pointer-decay) + memcpy(source.data(), ifr.ifr_hwaddr.sa_data, source.size()); +} + +[[nodiscard]] int XskSocket::timeDifference(const timespec& lhs, const timespec& rhs) noexcept +{ + const auto res = lhs.tv_sec * 1000 + lhs.tv_nsec / 1000000L - (rhs.tv_sec * 1000 + rhs.tv_nsec / 1000000L); + return static_cast<int>(res); +} + +void XskWorker::cleanWorkerNotification() const noexcept +{ + uint64_t value = read(xskSocketWaker, &value, sizeof(value)); +} + +void XskWorker::cleanSocketNotification() const noexcept +{ + uint64_t value = read(workerWaker, &value, sizeof(value)); +} + +std::vector<pollfd> getPollFdsForWorker(XskWorker& info) +{ + std::vector<pollfd> fds; + int timerfd = timerfd_create(CLOCK_MONOTONIC, TFD_CLOEXEC); + if (timerfd < 0) { + throw std::runtime_error("create_timerfd failed"); + } + fds.push_back(pollfd{ + .fd = info.workerWaker, + .events = POLLIN, + .revents = 0, + }); + fds.push_back(pollfd{ + .fd = timerfd, + .events = POLLIN, + .revents = 0, + }); + return fds; +} + +void XskWorker::fillUniqueEmptyOffset() +{ + auto frames = sharedEmptyFrameOffset->lock(); + const auto moveSize = std::min(static_cast<size_t>(32), frames->size()); + if (moveSize > 0) { + // NOLINTNEXTLINE(bugprone-narrowing-conversions,cppcoreguidelines-narrowing-conversions) + uniqueEmptyFrameOffset.insert(uniqueEmptyFrameOffset.end(), std::make_move_iterator(frames->end() - moveSize), std::make_move_iterator(frames->end())); + frames->resize(frames->size() - moveSize); + } +} + +std::optional<XskPacket> XskWorker::getEmptyFrame() +{ + if (!uniqueEmptyFrameOffset.empty()) { + auto offset = uniqueEmptyFrameOffset.back(); + uniqueEmptyFrameOffset.pop_back(); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) + return XskPacket(offset + umemBufBase, 0, frameSize); + } + fillUniqueEmptyOffset(); + if (!uniqueEmptyFrameOffset.empty()) { + auto offset = uniqueEmptyFrameOffset.back(); + uniqueEmptyFrameOffset.pop_back(); + // NOLINTNEXTLINE(cppcoreguidelines-pro-bounds-pointer-arithmetic) + return XskPacket(offset + umemBufBase, 0, frameSize); + } + return std::nullopt; +} + +void XskWorker::markAsFree(const XskPacket& packet) +{ + auto offset = frameOffset(packet); +#ifdef DEBUG_UMEM + checkUmemIntegrity(__PRETTY_FUNCTION__, __LINE__, offset, {UmemEntryStatus::Status::Received, UmemEntryStatus::Status::TXQueue}, UmemEntryStatus::Status::Free); +#endif /* DEBUG_UMEM */ + uniqueEmptyFrameOffset.push_back(offset); +} + +uint32_t XskPacket::getFlags() const noexcept +{ + return flags; +} + +void XskPacket::updatePacket() noexcept +{ + if ((flags & UPDATE) == 0U) { + return; + } + if ((flags & REWRITE) == 0U) { + changeDirectAndUpdateChecksum(); + } +} +#endif /* HAVE_XSK */ |