+++ /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.
- *
- */
-
-#ifndef CEPH_DPDK_DEV_H
-#define CEPH_DPDK_DEV_H
-
-#include <memory>
-#include <functional>
-#include <rte_config.h>
-#include <rte_common.h>
-#include <rte_ethdev.h>
-#include <rte_malloc.h>
-#include <rte_version.h>
-
-#include "include/page.h"
-#include "common/Tub.h"
-#include "common/perf_counters.h"
-#include "msg/async/Event.h"
-#include "const.h"
-#include "circular_buffer.h"
-#include "ethernet.h"
-#include "memory.h"
-#include "Packet.h"
-#include "stream.h"
-#include "net.h"
-#include "toeplitz.h"
-
-
-struct free_deleter {
- void operator()(void* p) { ::free(p); }
-};
-
-
-enum {
- l_dpdk_dev_first = 58800,
- l_dpdk_dev_rx_mcast,
- l_dpdk_dev_rx_total_errors,
- l_dpdk_dev_tx_total_errors,
- l_dpdk_dev_rx_badcrc_errors,
- l_dpdk_dev_rx_dropped_errors,
- l_dpdk_dev_rx_nombuf_errors,
- l_dpdk_dev_last
-};
-
-enum {
- l_dpdk_qp_first = 58900,
- l_dpdk_qp_rx_packets,
- l_dpdk_qp_tx_packets,
- l_dpdk_qp_rx_bad_checksum_errors,
- l_dpdk_qp_rx_no_memory_errors,
- l_dpdk_qp_rx_bytes,
- l_dpdk_qp_tx_bytes,
- l_dpdk_qp_rx_last_bunch,
- l_dpdk_qp_tx_last_bunch,
- l_dpdk_qp_rx_fragments,
- l_dpdk_qp_tx_fragments,
- l_dpdk_qp_rx_copy_ops,
- l_dpdk_qp_tx_copy_ops,
- l_dpdk_qp_rx_copy_bytes,
- l_dpdk_qp_tx_copy_bytes,
- l_dpdk_qp_rx_linearize_ops,
- l_dpdk_qp_tx_linearize_ops,
- l_dpdk_qp_tx_queue_length,
- l_dpdk_qp_last
-};
-
-class DPDKDevice;
-class DPDKWorker;
-
-class DPDKQueuePair {
- using packet_provider_type = std::function<Tub<Packet> ()>;
- public:
- void configure_proxies(const std::map<unsigned, float>& cpu_weights);
- // build REdirection TAble for cpu_weights map: target cpu -> weight
- void build_sw_reta(const std::map<unsigned, float>& cpu_weights);
- void proxy_send(Packet p) {
- _proxy_packetq.push_back(std::move(p));
- }
- void register_packet_provider(packet_provider_type func) {
- _pkt_providers.push_back(std::move(func));
- }
- bool poll_tx();
- friend class DPDKDevice;
-
- class tx_buf_factory;
-
- class tx_buf {
- friend class DPDKQueuePair;
- public:
- static tx_buf* me(rte_mbuf* mbuf) {
- return reinterpret_cast<tx_buf*>(mbuf);
- }
-
- private:
- /**
- * Checks if the original packet of a given cluster should be linearized
- * due to HW limitations.
- *
- * @param head head of a cluster to check
- *
- * @return TRUE if a packet should be linearized.
- */
- static bool i40e_should_linearize(rte_mbuf *head);
-
- /**
- * Sets the offload info in the head buffer of an rte_mbufs cluster.
- *
- * @param p an original packet the cluster is built for
- * @param qp QP handle
- * @param head a head of an rte_mbufs cluster
- */
- static void set_cluster_offload_info(const Packet& p, const DPDKQueuePair& qp, rte_mbuf* head);
-
- /**
- * Creates a tx_buf cluster representing a given packet in a "zero-copy"
- * way.
- *
- * @param p packet to translate
- * @param qp DPDKQueuePair handle
- *
- * @return the HEAD tx_buf of the cluster or nullptr in case of a
- * failure
- */
- static tx_buf* from_packet_zc(
- CephContext *cct, Packet&& p, DPDKQueuePair& qp);
-
- /**
- * Copy the contents of the "packet" into the given cluster of
- * rte_mbuf's.
- *
- * @note Size of the cluster has to be big enough to accommodate all the
- * contents of the given packet.
- *
- * @param p packet to copy
- * @param head head of the rte_mbuf's cluster
- */
- static void copy_packet_to_cluster(const Packet& p, rte_mbuf* head);
-
- /**
- * Creates a tx_buf cluster representing a given packet in a "copy" way.
- *
- * @param p packet to translate
- * @param qp DPDKQueuePair handle
- *
- * @return the HEAD tx_buf of the cluster or nullptr in case of a
- * failure
- */
- static tx_buf* from_packet_copy(Packet&& p, DPDKQueuePair& qp);
-
- /**
- * Zero-copy handling of a single fragment.
- *
- * @param do_one_buf Functor responsible for a single rte_mbuf
- * handling
- * @param qp DPDKQueuePair handle (in)
- * @param frag Fragment to copy (in)
- * @param head Head of the cluster (out)
- * @param last_seg Last segment of the cluster (out)
- * @param nsegs Number of segments in the cluster (out)
- *
- * @return TRUE in case of success
- */
- template <class DoOneBufFunc>
- static bool do_one_frag(DoOneBufFunc do_one_buf, DPDKQueuePair& qp,
- fragment& frag, rte_mbuf*& head,
- rte_mbuf*& last_seg, unsigned& nsegs) {
- size_t len, left_to_set = frag.size;
- char* base = frag.base;
-
- rte_mbuf* m;
-
- // TODO: assert() in a fast path! Remove me ASAP!
- assert(frag.size);
-
- // Create a HEAD of mbufs' cluster and set the first bytes into it
- len = do_one_buf(qp, head, base, left_to_set);
- if (!len) {
- return false;
- }
-
- left_to_set -= len;
- base += len;
- nsegs = 1;
-
- //
- // Set the rest of the data into the new mbufs and chain them to
- // the cluster.
- //
- rte_mbuf* prev_seg = head;
- while (left_to_set) {
- len = do_one_buf(qp, m, base, left_to_set);
- if (!len) {
- me(head)->recycle();
- return false;
- }
-
- left_to_set -= len;
- base += len;
- nsegs++;
-
- prev_seg->next = m;
- prev_seg = m;
- }
-
- // Return the last mbuf in the cluster
- last_seg = prev_seg;
-
- return true;
- }
-
- /**
- * Zero-copy handling of a single fragment.
- *
- * @param qp DPDKQueuePair handle (in)
- * @param frag Fragment to copy (in)
- * @param head Head of the cluster (out)
- * @param last_seg Last segment of the cluster (out)
- * @param nsegs Number of segments in the cluster (out)
- *
- * @return TRUE in case of success
- */
- static bool translate_one_frag(DPDKQueuePair& qp, fragment& frag,
- rte_mbuf*& head, rte_mbuf*& last_seg,
- unsigned& nsegs) {
- return do_one_frag(set_one_data_buf, qp, frag, head,
- last_seg, nsegs);
- }
-
- /**
- * Copies one fragment into the cluster of rte_mbuf's.
- *
- * @param qp DPDKQueuePair handle (in)
- * @param frag Fragment to copy (in)
- * @param head Head of the cluster (out)
- * @param last_seg Last segment of the cluster (out)
- * @param nsegs Number of segments in the cluster (out)
- *
- * We return the "last_seg" to avoid traversing the cluster in order to get
- * it.
- *
- * @return TRUE in case of success
- */
- static bool copy_one_frag(DPDKQueuePair& qp, fragment& frag,
- rte_mbuf*& head, rte_mbuf*& last_seg,
- unsigned& nsegs) {
- return do_one_frag(copy_one_data_buf, qp, frag, head,
- last_seg, nsegs);
- }
-
- /**
- * Allocates a single rte_mbuf and sets it to point to a given data
- * buffer.
- *
- * @param qp DPDKQueuePair handle (in)
- * @param m New allocated rte_mbuf (out)
- * @param va virtual address of a data buffer (in)
- * @param buf_len length of the data to copy (in)
- *
- * @return The actual number of bytes that has been set in the mbuf
- */
- static size_t set_one_data_buf(
- DPDKQueuePair& qp, rte_mbuf*& m, char* va, size_t buf_len) {
- static constexpr size_t max_frag_len = 15 * 1024; // 15K
-
- // FIXME: current all tx buf is alloced without rte_malloc
- return copy_one_data_buf(qp, m, va, buf_len);
- //
- // Currently we break a buffer on a 15K boundary because 82599
- // devices have a 15.5K limitation on a maximum single fragment
- // size.
- //
- phys_addr_t pa = rte_malloc_virt2phy(va);
- if (!pa)
- return copy_one_data_buf(qp, m, va, buf_len);
-
- assert(buf_len);
- tx_buf* buf = qp.get_tx_buf();
- if (!buf) {
- return 0;
- }
-
- size_t len = std::min(buf_len, max_frag_len);
-
- buf->set_zc_info(va, pa, len);
- m = buf->rte_mbuf_p();
-
- return len;
- }
-
- /**
- * Allocates a single rte_mbuf and copies a given data into it.
- *
- * @param qp DPDKQueuePair handle (in)
- * @param m New allocated rte_mbuf (out)
- * @param data Data to copy from (in)
- * @param buf_len length of the data to copy (in)
- *
- * @return The actual number of bytes that has been copied
- */
- static size_t copy_one_data_buf(
- DPDKQueuePair& qp, rte_mbuf*& m, char* data, size_t buf_len);
-
- /**
- * Checks if the first fragment of the given packet satisfies the
- * zero-copy flow requirement: its first 128 bytes should not cross the
- * 4K page boundary. This is required in order to avoid splitting packet
- * headers.
- *
- * @param p packet to check
- *
- * @return TRUE if packet is ok and FALSE otherwise.
- */
- static bool check_frag0(Packet& p)
- {
- //
- // First frag is special - it has headers that should not be split.
- // If the addressing is such that the first fragment has to be
- // split, then send this packet in a (non-zero) copy flow. We'll
- // check if the first 128 bytes of the first fragment reside in the
- // physically contiguous area. If that's the case - we are good to
- // go.
- //
- if (p.frag(0).size < 128)
- return false;
-
- return true;
- }
-
- public:
- tx_buf(tx_buf_factory& fc) : _fc(fc) {
-
- _buf_physaddr = _mbuf.buf_physaddr;
- _data_off = _mbuf.data_off;
- }
-
- rte_mbuf* rte_mbuf_p() { return &_mbuf; }
-
- void set_zc_info(void* va, phys_addr_t pa, size_t len) {
- // mbuf_put()
- _mbuf.data_len = len;
- _mbuf.pkt_len = len;
-
- // Set the mbuf to point to our data
- _mbuf.buf_addr = va;
- _mbuf.buf_physaddr = pa;
- _mbuf.data_off = 0;
- _is_zc = true;
- }
-
- void reset_zc() {
-
- //
- // If this mbuf was the last in a cluster and contains an
- // original packet object then call the destructor of the
- // original packet object.
- //
- if (_p) {
- //
- // Reset the std::optional. This in particular is going
- // to call the "packet"'s destructor and reset the
- // "optional" state to "nonengaged".
- //
- _p.destroy();
-
- } else if (!_is_zc) {
- return;
- }
-
- // Restore the rte_mbuf fields we trashed in set_zc_info()
- _mbuf.buf_physaddr = _buf_physaddr;
- _mbuf.buf_addr = rte_mbuf_to_baddr(&_mbuf);
- _mbuf.data_off = _data_off;
-
- _is_zc = false;
- }
-
- void recycle() {
- struct rte_mbuf *m = &_mbuf, *m_next;
-
- while (m != nullptr) {
- m_next = m->next;
- rte_pktmbuf_reset(m);
- _fc.put(me(m));
- m = m_next;
- }
- }
-
- void set_packet(Packet&& p) {
- _p = std::move(p);
- }
-
- private:
- struct rte_mbuf _mbuf;
- MARKER private_start;
- Tub<Packet> _p;
- phys_addr_t _buf_physaddr;
- uint16_t _data_off;
- // TRUE if underlying mbuf has been used in the zero-copy flow
- bool _is_zc = false;
- // buffers' factory the buffer came from
- tx_buf_factory& _fc;
- MARKER private_end;
- };
-
- class tx_buf_factory {
- //
- // Number of buffers to free in each GC iteration:
- // We want the buffers to be allocated from the mempool as many as
- // possible.
- //
- // On the other hand if there is no Tx for some time we want the
- // completions to be eventually handled. Thus we choose the smallest
- // possible packets count number here.
- //
- static constexpr int gc_count = 1;
- public:
- tx_buf_factory(CephContext *c, DPDKDevice *dev, uint8_t qid);
- ~tx_buf_factory() {
- // put all mbuf back into mempool in order to make the next factory work
- while (gc());
- rte_mempool_put_bulk(_pool, (void**)_ring.data(),
- _ring.size());
- }
-
-
- /**
- * @note Should not be called if there are no free tx_buf's
- *
- * @return a free tx_buf object
- */
- tx_buf* get() {
- // Take completed from the HW first
- tx_buf *pkt = get_one_completed();
- if (pkt) {
- pkt->reset_zc();
- return pkt;
- }
-
- //
- // If there are no completed at the moment - take from the
- // factory's cache.
- //
- if (_ring.empty()) {
- return nullptr;
- }
-
- pkt = _ring.back();
- _ring.pop_back();
-
- return pkt;
- }
-
- void put(tx_buf* buf) {
- buf->reset_zc();
- _ring.push_back(buf);
- }
-
- bool gc() {
- for (int cnt = 0; cnt < gc_count; ++cnt) {
- auto tx_buf_p = get_one_completed();
- if (!tx_buf_p) {
- return false;
- }
-
- put(tx_buf_p);
- }
-
- return true;
- }
- private:
- /**
- * Fill the mbufs circular buffer: after this the _pool will become
- * empty. We will use it to catch the completed buffers:
- *
- * - Underlying PMD drivers will "free" the mbufs once they are
- * completed.
- * - We will poll the _pktmbuf_pool_tx till it's empty and release
- * all the buffers from the freed mbufs.
- */
- void init_factory() {
- while (rte_mbuf* mbuf = rte_pktmbuf_alloc(_pool)) {
- _ring.push_back(new(tx_buf::me(mbuf)) tx_buf{*this});
- }
- }
-
- /**
- * PMD puts the completed buffers back into the mempool they have
- * originally come from.
- *
- * @note rte_pktmbuf_alloc() resets the mbuf so there is no need to call
- * rte_pktmbuf_reset() here again.
- *
- * @return a single tx_buf that has been completed by HW.
- */
- tx_buf* get_one_completed() {
- return tx_buf::me(rte_pktmbuf_alloc(_pool));
- }
-
- private:
- CephContext *cct;
- std::vector<tx_buf*> _ring;
- rte_mempool* _pool = nullptr;
- };
-
- public:
- explicit DPDKQueuePair(CephContext *c, EventCenter *cen, DPDKDevice* dev, uint8_t qid);
- ~DPDKQueuePair() {
- if (device_stat_time_fd) {
- center->delete_time_event(device_stat_time_fd);
- }
- rx_gc(true);
- }
-
- void rx_start() {
- _rx_poller.construct(this);
- }
-
- uint32_t send(circular_buffer<Packet>& pb) {
- // Zero-copy send
- return _send(pb, [&] (Packet&& p) {
- return tx_buf::from_packet_zc(cct, std::move(p), *this);
- });
- }
-
- DPDKDevice& port() const { return *_dev; }
- tx_buf* get_tx_buf() { return _tx_buf_factory.get(); }
-
- void handle_stats();
-
- private:
- template <class Func>
- uint32_t _send(circular_buffer<Packet>& pb, Func &&packet_to_tx_buf_p) {
- if (_tx_burst.size() == 0) {
- for (auto&& p : pb) {
- // TODO: assert() in a fast path! Remove me ASAP!
- assert(p.len());
-
- tx_buf* buf = packet_to_tx_buf_p(std::move(p));
- if (!buf) {
- break;
- }
-
- _tx_burst.push_back(buf->rte_mbuf_p());
- }
- }
-
- uint16_t sent = rte_eth_tx_burst(_dev_port_idx, _qid,
- _tx_burst.data() + _tx_burst_idx,
- _tx_burst.size() - _tx_burst_idx);
-
- uint64_t nr_frags = 0, bytes = 0;
-
- for (int i = 0; i < sent; i++) {
- rte_mbuf* m = _tx_burst[_tx_burst_idx + i];
- bytes += m->pkt_len;
- nr_frags += m->nb_segs;
- pb.pop_front();
- }
-
- perf_logger->inc(l_dpdk_qp_tx_fragments, nr_frags);
- perf_logger->inc(l_dpdk_qp_tx_bytes, bytes);
-
- _tx_burst_idx += sent;
-
- if (_tx_burst_idx == _tx_burst.size()) {
- _tx_burst_idx = 0;
- _tx_burst.clear();
- }
-
- return sent;
- }
-
- /**
- * Allocate a new data buffer and set the mbuf to point to it.
- *
- * Do some DPDK hacks to work on PMD: it assumes that the buf_addr
- * points to the private data of RTE_PKTMBUF_HEADROOM before the actual
- * data buffer.
- *
- * @param m mbuf to update
- */
- static bool refill_rx_mbuf(rte_mbuf* m, size_t size,
- std::vector<void*> &datas) {
- if (datas.empty())
- return false;
- void *data = datas.back();
- datas.pop_back();
-
- //
- // Set the mbuf to point to our data.
- //
- // Do some DPDK hacks to work on PMD: it assumes that the buf_addr
- // points to the private data of RTE_PKTMBUF_HEADROOM before the
- // actual data buffer.
- //
- m->buf_addr = (char*)data - RTE_PKTMBUF_HEADROOM;
- m->buf_physaddr = rte_malloc_virt2phy(data) - RTE_PKTMBUF_HEADROOM;
- return true;
- }
-
- static bool init_noninline_rx_mbuf(rte_mbuf* m, size_t size,
- std::vector<void*> &datas) {
- if (!refill_rx_mbuf(m, size, datas)) {
- return false;
- }
- // The below fields stay constant during the execution.
- m->buf_len = size + RTE_PKTMBUF_HEADROOM;
- m->data_off = RTE_PKTMBUF_HEADROOM;
- return true;
- }
-
- bool init_rx_mbuf_pool();
- bool rx_gc(bool force=false);
- bool refill_one_cluster(rte_mbuf* head);
-
- /**
- * Polls for a burst of incoming packets. This function will not block and
- * will immediately return after processing all available packets.
- *
- */
- bool poll_rx_once();
-
- /**
- * Translates an rte_mbuf's into packet and feeds them to _rx_stream.
- *
- * @param bufs An array of received rte_mbuf's
- * @param count Number of buffers in the bufs[]
- */
- void process_packets(struct rte_mbuf **bufs, uint16_t count);
-
- /**
- * Translate rte_mbuf into the "packet".
- * @param m mbuf to translate
- *
- * @return a "optional" object representing the newly received data if in an
- * "engaged" state or an error if in a "disengaged" state.
- */
- Tub<Packet> from_mbuf(rte_mbuf* m);
-
- /**
- * Transform an LRO rte_mbuf cluster into the "packet" object.
- * @param m HEAD of the mbufs' cluster to transform
- *
- * @return a "optional" object representing the newly received LRO packet if
- * in an "engaged" state or an error if in a "disengaged" state.
- */
- Tub<Packet> from_mbuf_lro(rte_mbuf* m);
-
- private:
- CephContext *cct;
- std::vector<packet_provider_type> _pkt_providers;
- Tub<std::array<uint8_t, 128>> _sw_reta;
- circular_buffer<Packet> _proxy_packetq;
- stream<Packet> _rx_stream;
- circular_buffer<Packet> _tx_packetq;
- std::vector<void*> _alloc_bufs;
-
- PerfCounters *perf_logger;
- DPDKDevice* _dev;
- uint8_t _dev_port_idx;
- EventCenter *center;
- uint8_t _qid;
- rte_mempool *_pktmbuf_pool_rx;
- std::vector<rte_mbuf*> _rx_free_pkts;
- std::vector<rte_mbuf*> _rx_free_bufs;
- std::vector<fragment> _frags;
- std::vector<char*> _bufs;
- size_t _num_rx_free_segs = 0;
- uint64_t device_stat_time_fd = 0;
-
-#ifdef CEPH_PERF_DEV
- uint64_t rx_cycles = 0;
- uint64_t rx_count = 0;
- uint64_t tx_cycles = 0;
- uint64_t tx_count = 0;
-#endif
-
- class DPDKTXPoller : public EventCenter::Poller {
- DPDKQueuePair *qp;
-
- public:
- explicit DPDKTXPoller(DPDKQueuePair *qp)
- : EventCenter::Poller(qp->center, "DPDK::DPDKTXPoller"), qp(qp) {}
-
- virtual int poll() {
- return qp->poll_tx();
- }
- } _tx_poller;
-
- class DPDKRXGCPoller : public EventCenter::Poller {
- DPDKQueuePair *qp;
-
- public:
- explicit DPDKRXGCPoller(DPDKQueuePair *qp)
- : EventCenter::Poller(qp->center, "DPDK::DPDKRXGCPoller"), qp(qp) {}
-
- virtual int poll() {
- return qp->rx_gc();
- }
- } _rx_gc_poller;
- tx_buf_factory _tx_buf_factory;
- class DPDKRXPoller : public EventCenter::Poller {
- DPDKQueuePair *qp;
-
- public:
- explicit DPDKRXPoller(DPDKQueuePair *qp)
- : EventCenter::Poller(qp->center, "DPDK::DPDKRXPoller"), qp(qp) {}
-
- virtual int poll() {
- return qp->poll_rx_once();
- }
- };
- Tub<DPDKRXPoller> _rx_poller;
- class DPDKTXGCPoller : public EventCenter::Poller {
- DPDKQueuePair *qp;
-
- public:
- explicit DPDKTXGCPoller(DPDKQueuePair *qp)
- : EventCenter::Poller(qp->center, "DPDK::DPDKTXGCPoller"), qp(qp) {}
-
- virtual int poll() {
- return qp->_tx_buf_factory.gc();
- }
- } _tx_gc_poller;
- std::vector<rte_mbuf*> _tx_burst;
- uint16_t _tx_burst_idx = 0;
-};
-
-class DPDKDevice {
- public:
- CephContext *cct;
- PerfCounters *perf_logger;
- std::vector<std::unique_ptr<DPDKQueuePair>> _queues;
- std::vector<DPDKWorker*> workers;
- size_t _rss_table_bits = 0;
- uint8_t _port_idx;
- uint16_t _num_queues;
- unsigned cores;
- hw_features _hw_features;
- uint8_t _queues_ready = 0;
- unsigned _home_cpu;
- bool _use_lro;
- bool _enable_fc;
- std::vector<uint8_t> _redir_table;
- rss_key_type _rss_key;
- bool _is_i40e_device = false;
- bool _is_vmxnet3_device = false;
-
- public:
- rte_eth_dev_info _dev_info = {};
-
- /**
- * The final stage of a port initialization.
- * @note Must be called *after* all queues from stage (2) have been
- * initialized.
- */
- int init_port_fini();
-
- private:
- /**
- * Port initialization consists of 3 main stages:
- * 1) General port initialization which ends with a call to
- * rte_eth_dev_configure() where we request the needed number of Rx and
- * Tx queues.
- * 2) Individual queues initialization. This is done in the constructor of
- * DPDKQueuePair class. In particular the memory pools for queues are allocated
- * in this stage.
- * 3) The final stage of the initialization which starts with the call of
- * rte_eth_dev_start() after which the port becomes fully functional. We
- * will also wait for a link to get up in this stage.
- */
-
-
- /**
- * First stage of the port initialization.
- *
- * @return 0 in case of success and an appropriate error code in case of an
- * error.
- */
- int init_port_start();
-
- /**
- * Check the link status of out port in up to 9s, and print them finally.
- */
- int check_port_link_status();
-
- /**
- * Configures the HW Flow Control
- */
- void set_hw_flow_control();
-
- public:
- DPDKDevice(CephContext *c, uint8_t port_idx, uint16_t num_queues, bool use_lro, bool enable_fc):
- cct(c), _port_idx(port_idx), _num_queues(num_queues),
- _home_cpu(0), _use_lro(use_lro),
- _enable_fc(enable_fc) {
- _queues = std::vector<std::unique_ptr<DPDKQueuePair>>(_num_queues);
- /* now initialise the port we will use */
- int ret = init_port_start();
- if (ret != 0) {
- rte_exit(EXIT_FAILURE, "Cannot initialise port %u\n", _port_idx);
- }
- string name(std::string("port") + std::to_string(port_idx));
- PerfCountersBuilder plb(cct, name, l_dpdk_dev_first, l_dpdk_dev_last);
-
- plb.add_u64_counter(l_dpdk_dev_rx_mcast, "dpdk_device_receive_multicast_packets", "DPDK received multicast packets");
- plb.add_u64_counter(l_dpdk_dev_rx_badcrc_errors, "dpdk_device_receive_badcrc_errors", "DPDK received bad crc errors");
-
- plb.add_u64_counter(l_dpdk_dev_rx_total_errors, "dpdk_device_receive_total_errors", "DPDK received total_errors");
- plb.add_u64_counter(l_dpdk_dev_tx_total_errors, "dpdk_device_send_total_errors", "DPDK sendd total_errors");
- plb.add_u64_counter(l_dpdk_dev_rx_dropped_errors, "dpdk_device_receive_dropped_errors", "DPDK received dropped errors");
- plb.add_u64_counter(l_dpdk_dev_rx_nombuf_errors, "dpdk_device_receive_nombuf_errors", "DPDK received RX mbuf allocation errors");
-
- perf_logger = plb.create_perf_counters();
- cct->get_perfcounters_collection()->add(perf_logger);
- }
-
- ~DPDKDevice() {
- rte_eth_dev_stop(_port_idx);
- }
-
- DPDKQueuePair& queue_for_cpu(unsigned cpu) { return *_queues[cpu]; }
- void l2receive(int qid, Packet p) {
- _queues[qid]->_rx_stream.produce(std::move(p));
- }
- subscription<Packet> receive(unsigned cpuid, std::function<int (Packet)> next_packet) {
- auto sub = _queues[cpuid]->_rx_stream.listen(std::move(next_packet));
- _queues[cpuid]->rx_start();
- return std::move(sub);
- }
- ethernet_address hw_address() {
- struct ether_addr mac;
- rte_eth_macaddr_get(_port_idx, &mac);
-
- return mac.addr_bytes;
- }
- hw_features get_hw_features() {
- return _hw_features;
- }
- const rss_key_type& rss_key() const { return _rss_key; }
- uint16_t hw_queues_count() { return _num_queues; }
- std::unique_ptr<DPDKQueuePair> init_local_queue(CephContext *c, EventCenter *center, string hugepages, uint16_t qid) {
- std::unique_ptr<DPDKQueuePair> qp;
- qp = std::unique_ptr<DPDKQueuePair>(new DPDKQueuePair(c, center, this, qid));
- return std::move(qp);
- }
- unsigned hash2qid(uint32_t hash) {
- // return hash % hw_queues_count();
- return _redir_table[hash & (_redir_table.size() - 1)];
- }
- void set_local_queue(unsigned i, std::unique_ptr<DPDKQueuePair> qp) {
- assert(!_queues[i]);
- _queues[i] = std::move(qp);
- }
- void unset_local_queue(unsigned i) {
- assert(_queues[i]);
- _queues[i].reset();
- }
- template <typename Func>
- unsigned forward_dst(unsigned src_cpuid, Func&& hashfn) {
- auto& qp = queue_for_cpu(src_cpuid);
- if (!qp._sw_reta)
- return src_cpuid;
-
- assert(!qp._sw_reta);
- auto hash = hashfn() >> _rss_table_bits;
- auto& reta = *qp._sw_reta;
- return reta[hash % reta.size()];
- }
- unsigned hash2cpu(uint32_t hash) {
- // there is an assumption here that qid == get_id() which will
- // not necessary be true in the future
- return forward_dst(hash2qid(hash), [hash] { return hash; });
- }
-
- hw_features& hw_features_ref() { return _hw_features; }
-
- const rte_eth_rxconf* def_rx_conf() const {
- return &_dev_info.default_rxconf;
- }
-
- const rte_eth_txconf* def_tx_conf() const {
- return &_dev_info.default_txconf;
- }
-
- /**
- * Set the RSS table in the device and store it in the internal vector.
- */
- void set_rss_table();
-
- uint8_t port_idx() { return _port_idx; }
- bool is_i40e_device() const {
- return _is_i40e_device;
- }
- bool is_vmxnet3_device() const {
- return _is_vmxnet3_device;
- }
-};
-
-
-std::unique_ptr<DPDKDevice> create_dpdk_net_device(
- CephContext *c, unsigned cores, uint8_t port_idx = 0,
- bool use_lro = true, bool enable_fc = true);
-
-
-/**
- * @return Number of bytes needed for mempool objects of each QP.
- */
-uint32_t qp_mempool_obj_size();
-
-#endif // CEPH_DPDK_DEV_H
Code Review / stor4nfv.git / blobdiff