--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+/*
+ * This file is open source software, licensed to you under the terms
+ * of the Apache License, Version 2.0 (the "License"). See the NOTICE file
+ * distributed with this work for additional information regarding copyright
+ * ownership. You may not use this file except in compliance with the License.
+ *
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+/*
+ * Copyright (C) 2014 Cloudius Systems, Ltd.
+ */
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2015 XSky <haomai@xsky.com>
+ *
+ * Author: Haomai Wang <haomaiwang@gmail.com>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation. See file COPYING.
+ *
+ */
+
+#include "net.h"
+#include "DPDK.h"
+#include "DPDKStack.h"
+
+#include "common/dout.h"
+#include "include/assert.h"
+
+#define dout_subsys ceph_subsys_dpdk
+#undef dout_prefix
+#define dout_prefix *_dout << "net "
+
+interface::interface(CephContext *c, std::shared_ptr<DPDKDevice> dev, EventCenter *center)
+ : cct(c), _dev(dev),
+ _rx(_dev->receive(
+ center->get_id(),
+ [center, this] (Packet p) {
+ return dispatch_packet(center, std::move(p));
+ }
+ )),
+ _hw_address(_dev->hw_address()),
+ _hw_features(_dev->get_hw_features()) {
+ auto idx = 0u;
+ unsigned qid = center->get_id();
+ dev->queue_for_cpu(center->get_id()).register_packet_provider([this, idx, qid] () mutable {
+ Tub<Packet> p;
+ for (size_t i = 0; i < _pkt_providers.size(); i++) {
+ auto l3p = _pkt_providers[idx++]();
+ if (idx == _pkt_providers.size())
+ idx = 0;
+ if (l3p) {
+ auto l3pv = std::move(*l3p);
+ auto eh = l3pv.p.prepend_header<eth_hdr>();
+ eh->dst_mac = l3pv.to;
+ eh->src_mac = _hw_address;
+ eh->eth_proto = uint16_t(l3pv.proto_num);
+ *eh = eh->hton();
+ ldout(cct, 10) << "=== tx === proto " << std::hex << uint16_t(l3pv.proto_num)
+ << " " << _hw_address << " -> " << l3pv.to
+ << " length " << std::dec << l3pv.p.len() << dendl;
+ p = std::move(l3pv.p);
+ return p;
+ }
+ }
+ return p;
+ });
+}
+
+subscription<Packet, ethernet_address> interface::register_l3(
+ eth_protocol_num proto_num,
+ std::function<int (Packet p, ethernet_address from)> next,
+ std::function<bool (forward_hash&, Packet& p, size_t)> forward)
+{
+ auto i = _proto_map.emplace(std::piecewise_construct, std::make_tuple(uint16_t(proto_num)), std::forward_as_tuple(std::move(forward)));
+ assert(i.second);
+ l3_rx_stream& l3_rx = i.first->second;
+ return l3_rx.packet_stream.listen(std::move(next));
+}
+
+unsigned interface::hash2cpu(uint32_t hash) {
+ return _dev->hash2cpu(hash);
+}
+
+const rss_key_type& interface::rss_key() const {
+ return _dev->rss_key();
+}
+
+uint16_t interface::hw_queues_count() const {
+ return _dev->hw_queues_count();
+}
+
+class C_handle_l2forward : public EventCallback {
+ std::shared_ptr<DPDKDevice> sdev;
+ unsigned &queue_depth;
+ Packet p;
+ unsigned dst;
+
+ public:
+ C_handle_l2forward(std::shared_ptr<DPDKDevice> &p, unsigned &qd, Packet pkt, unsigned target)
+ : sdev(p), queue_depth(qd), p(std::move(pkt)), dst(target) {}
+ void do_request(int fd) {
+ sdev->l2receive(dst, std::move(p));
+ queue_depth--;
+ delete this;
+ }
+};
+
+void interface::forward(EventCenter *source, unsigned target, Packet p) {
+ static __thread unsigned queue_depth;
+
+ if (queue_depth < 1000) {
+ queue_depth++;
+ // FIXME: need ensure this event not be called after EventCenter destruct
+ _dev->workers[target]->center.dispatch_event_external(
+ new C_handle_l2forward(_dev, queue_depth, std::move(p.free_on_cpu(source)), target));
+ }
+}
+
+int interface::dispatch_packet(EventCenter *center, Packet p) {
+ auto eh = p.get_header<eth_hdr>();
+ if (eh) {
+ auto i = _proto_map.find(ntoh(eh->eth_proto));
+ auto hwrss = p.rss_hash();
+ if (hwrss) {
+ ldout(cct, 10) << __func__ << " === rx === proto " << std::hex << ::ntoh(eh->eth_proto)
+ << " "<< eh->src_mac.ntoh() << " -> " << eh->dst_mac.ntoh()
+ << " length " << std::dec << p.len() << " rss_hash " << *p.rss_hash() << dendl;
+ } else {
+ ldout(cct, 10) << __func__ << " === rx === proto " << std::hex << ::ntoh(eh->eth_proto)
+ << " "<< eh->src_mac.ntoh() << " -> " << eh->dst_mac.ntoh()
+ << " length " << std::dec << p.len() << dendl;
+ }
+ if (i != _proto_map.end()) {
+ l3_rx_stream& l3 = i->second;
+ auto fw = _dev->forward_dst(center->get_id(), [&p, &l3, this] () {
+ auto hwrss = p.rss_hash();
+ if (hwrss) {
+ return *hwrss;
+ } else {
+ forward_hash data;
+ if (l3.forward(data, p, sizeof(eth_hdr))) {
+ return toeplitz_hash(rss_key(), data);
+ }
+ return 0u;
+ }
+ });
+ if (fw != center->get_id()) {
+ ldout(cct, 1) << __func__ << " forward to " << fw << dendl;
+ forward(center, fw, std::move(p));
+ } else {
+ auto h = eh->ntoh();
+ auto from = h.src_mac;
+ p.trim_front(sizeof(*eh));
+ // avoid chaining, since queue length is unlimited
+ // drop instead.
+ if (l3.ready()) {
+ return l3.packet_stream.produce(std::move(p), from);
+ }
+ }
+ }
+ }
+ return 0;
+}
+
+class C_arp_learn : public EventCallback {
+ DPDKWorker *worker;
+ ethernet_address l2_addr;
+ ipv4_address l3_addr;
+
+ public:
+ C_arp_learn(DPDKWorker *w, ethernet_address l2, ipv4_address l3)
+ : worker(w), l2_addr(l2), l3_addr(l3) {}
+ void do_request(int id) {
+ worker->arp_learn(l2_addr, l3_addr);
+ delete this;
+ }
+};
+
+void interface::arp_learn(ethernet_address l2, ipv4_address l3)
+{
+ for (auto &&w : _dev->workers) {
+ w->center.dispatch_event_external(
+ new C_arp_learn(w, l2, l3));
+ }
+}
+
+l3_protocol::l3_protocol(interface* netif, eth_protocol_num proto_num, packet_provider_type func)
+ : _netif(netif), _proto_num(proto_num) {
+ _netif->register_packet_provider(std::move(func));
+}
+
+subscription<Packet, ethernet_address> l3_protocol::receive(
+ std::function<int (Packet, ethernet_address)> rx_fn,
+ std::function<bool (forward_hash &h, Packet &p, size_t s)> forward) {
+ return _netif->register_l3(_proto_num, std::move(rx_fn), std::move(forward));
+};
Code Review / stor4nfv.git / blobdiff