内核 qdiscs 模块中的网络数据包缓冲

Question

我想缓冲来自容器网络接口的输出数据包。 这个名为 sch_plug.c https://code.woboq.org/linux/linux/net/sched/sch_plug.c.html 的 netlink 库文件看起来可以解决问题，但我发现它很难使用？我应该如何调用这些函数才能使其真正起作用？如何获取传递给源代码中定义的函数的struct netlink_ext_ack *extack、struct sk_buff *skb等参数？

Answer 1

命令行

可以使用命令nl-qdisc-add、nl-qdisc-delete、nl-qdisc-list（libnl 的一部分）来控制 qdiscs。 --help 标志可用于显示一些用法示例（link）：

• 为网络接口eth0创建一个大小为32KB的plug qdisc：

  # nl-qdisc-add --dev=eth0 --parent=root plug --limit=32768
  

• 默认情况下，插件 qdisc 将处于缓冲模式（这意味着它会阻止所有传出流量）。您可以使用以下命令在缓冲模式和释放模式之间切换：

  • 切换到释放模式：

    # nl-qdisc-add --dev=eth0 --parent=root --update plug --release-indefinite
    

  • 切换回缓冲模式：

    # nl-qdisc-add --dev=eth0 --parent=root --update plug --buffer
    

• 您可以通过以下方式检查活动的 qdisc：

    # nl-qdisc-list --kind=plug --details --stats
  

  这也会告诉你每个 qdisc 的 id。

• 根据id，你可以再次删除一个qdisc：

   # nl-qdisc-delete --id <id>
  

来自代码

可以检查上面使用的工具的代码以编写自定义实现（link）：

#include <linux/netlink.h>

#include <netlink/netlink.h>
#include <netlink/route/qdisc.h>
#include <netlink/route/qdisc/plug.h>
#include <netlink/socket.h>

#include <atomic>
#include <csignal>
#include <iostream>
#include <stdexcept>

/**
 * Netlink route socket.
 */
struct Socket {
  Socket() : handle{nl_socket_alloc()} {

    if (handle == nullptr) {
      throw std::runtime_error{"Failed to allocate socket!"};
    }

    if (int err = nl_connect(handle, NETLINK_ROUTE); err < 0) {
      throw std::runtime_error{"Unable to connect netlink socket: " +
                               std::string{nl_geterror(err)}};
    }
  }

  Socket(const Socket &) = delete;
  Socket &operator=(const Socket &) = delete;
  Socket(Socket &&) = delete;
  Socket &operator=(Socket &&) = delete;

  ~Socket() { nl_socket_free(handle); }

  struct nl_sock *handle;
};

/**
 * Read all links from netlink socket.
 */
struct LinkCache {
  explicit LinkCache(Socket *socket) : handle{nullptr} {
    if (int err = rtnl_link_alloc_cache(socket->handle, AF_UNSPEC, &handle);
        err < 0) {
      throw std::runtime_error{"Unable to allocate link cache: " +
                               std::string{nl_geterror(err)}};
    }
  }

  LinkCache(const LinkCache &) = delete;
  LinkCache &operator=(const LinkCache &) = delete;
  LinkCache(LinkCache &&) = delete;
  LinkCache &operator=(LinkCache &&) = delete;

  ~LinkCache() { nl_cache_free(handle); }

  struct nl_cache *handle;
};

/**
 * Link (such as "eth0" or "wlan0").
 */
struct Link {
  Link(LinkCache *link_cache, const std::string &iface)
      : handle{rtnl_link_get_by_name(link_cache->handle, iface.c_str())} {

    if (handle == nullptr) {
      throw std::runtime_error{"Link does not exist:" + iface};
    }
  }

  Link(const Link &) = delete;
  Link &operator=(const Link &) = delete;
  Link(Link &&) = delete;
  Link &operator=(Link &&) = delete;

  ~Link() { rtnl_link_put(handle); }

  struct rtnl_link *handle;
};

/**
 * Queuing discipline.
 */
struct QDisc {
  QDisc(const std::string &iface, const std::string &kind)
      : handle{rtnl_qdisc_alloc()} {
    if (handle == nullptr) {
      throw std::runtime_error{"Failed to allocate qdisc!"};
    }

    struct rtnl_tc *tc = TC_CAST(handle);

    // Set link
    LinkCache link_cache{&socket};
    Link link{&link_cache, iface};
    rtnl_tc_set_link(tc, link.handle);

    // Set parent qdisc
    uint32_t parent = 0;

    if (int err = rtnl_tc_str2handle("root", &parent); err < 0) {
      throw std::runtime_error{"Unable to parse handle: " +
                               std::string{nl_geterror(err)}};
    }

    rtnl_tc_set_parent(tc, parent);

    // Set kind (e.g. "plug")
    if (int err = rtnl_tc_set_kind(tc, kind.c_str()); err < 0) {
      throw std::runtime_error{"Unable to set kind: " +
                               std::string{nl_geterror(err)}};
    }
  }

  QDisc(const QDisc &) = delete;
  QDisc &operator=(const QDisc &) = delete;
  QDisc(QDisc &&) = delete;
  QDisc &operator=(QDisc &&) = delete;

  ~QDisc() {
    if (int err = rtnl_qdisc_delete(socket.handle, handle); err < 0) {
      std::cerr << "Unable to delete qdisc: " << nl_geterror(err) << std::endl;
    }

    rtnl_qdisc_put(handle);
  }

  void send_msg() {
    int flags = NLM_F_CREATE;

    if (int err = rtnl_qdisc_add(socket.handle, handle, flags); err < 0) {
      throw std::runtime_error{"Unable to add qdisc: " +
                               std::string{nl_geterror(err)}};
    }
  }

  Socket socket;
  struct rtnl_qdisc *handle;
};

/**
 * Queuing discipline for plugging traffic.
 */
class Plug {
public:
  Plug(const std::string &iface, uint32_t limit, bool enabled)
      : qdisc_{iface, "plug"}, enabled_{enabled} {

    rtnl_qdisc_plug_set_limit(qdisc_.handle, limit);
    qdisc_.send_msg();

    set_enabled(enabled_);
  }

  void set_enabled(bool enabled) {
    if (enabled) {
      rtnl_qdisc_plug_buffer(qdisc_.handle);
    } else {
      rtnl_qdisc_plug_release_indefinite(qdisc_.handle);
    }

    qdisc_.send_msg();
    enabled_ = enabled;
  }

  bool is_enabled() const { return enabled_; }

private:
  QDisc qdisc_;

  bool enabled_;
};

std::atomic<bool> quit{false};

void exit_handler(int /*signal*/) { quit = true; }

int main() {
  std::string iface{"eth0"};
  constexpr uint32_t buffer_size = 32768;
  bool enabled = true;

  Plug plug{iface, buffer_size, enabled};

  /**
   * Set custom exit handler to ensure destructor runs to delete qdisc.
   */
  struct sigaction sa {};
  sa.sa_handler = exit_handler;
  sigfillset(&sa.sa_mask);
  sigaction(SIGINT, &sa, nullptr);

  while (!quit) {
    std::cout << "Plug set to " << plug.is_enabled() << std::endl;
    std::cout << "Press <Enter> to continue.";
    std::cin.get();

    plug.set_enabled(!plug.is_enabled());
  }

  return EXIT_SUCCESS;
}

在main 函数中设置您要使用的网络接口（例如eth0 或wlan0）。然后该程序可以用于：

# g++ -std=c++17 -Wall -Wextra -pedantic netbuf.cpp $( pkg-config --cflags --libs libnl-3.0 libnl-route-3.0 )
# ./a.out 
Plug set to 1
Press <Enter> to continue.
Plug set to 0
Press <Enter> to continue.
Plug set to 1
Press <Enter> to continue.

（使用 Ctrl+c 退出。）