mir/ovs
2
0
Fork 0
mirror of https://github.com/openvswitch/ovs synced 2025-09-03 07:45:30 +00:00
Code Issues Releases Wiki Activity

dpif-netdev: Add poll-mode-device thread.

Browse Source 
This patch adds PMD type netdev for netdevice with poll-mode
drivers.  Since there is no way to get signal on a packet recv
from these devices we need to poll them in busy loop.  So minimize
system call overhead this patch uses dpif-thread exclusively
for PMD devices and rest of devices which needs system calls to
do IO are moved to dpif-netdev-run().
PMD device like DPDK work in userspace so there is no system call
overhead for them.

Signed-off-by: Pravin B Shelar <pshelar@nicira.com>
Acked-by: Thomas Graf <tgraf@redhat.com>
This commit is contained in:
Pravin
2014-03-20 10:57:41 -07:00
committed by Pravin B Shelar
parent b284085e55
commit e4cfed38b1
3 changed files with 223 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

315
lib/dpif-netdev.c
View File

@@ -66,6 +66,8 @@ VLOG_DEFINE_THIS_MODULE(dpif_netdev);
/* By default, choose a priority in the middle. */ /* By default, choose a priority in the middle. */
#define NETDEV_RULE_PRIORITY 0x8000 #define NETDEV_RULE_PRIORITY 0x8000
#define NR_THREADS 1
/* Configuration parameters. */ /* Configuration parameters. */
enum { MAX_FLOWS = 65536 }; /* Maximum number of flows in flow table. */ enum { MAX_FLOWS = 65536 }; /* Maximum number of flows in flow table. */
@@ -161,8 +163,9 @@ struct dp_netdev {
/* Forwarding threads. */ /* Forwarding threads. */
struct latch exit_latch; struct latch exit_latch;
struct dp_forwarder *forwarders; struct pmd_thread *pmd_threads;
size_t n_forwarders; size_t n_pmd_threads;
int pmd_count;
}; };
static struct dp_netdev_port *dp_netdev_lookup_port(const struct dp_netdev *dp, static struct dp_netdev_port *dp_netdev_lookup_port(const struct dp_netdev *dp,
@@ -300,13 +303,23 @@ struct dp_netdev_actions *dp_netdev_flow_get_actions(
const struct dp_netdev_flow *); const struct dp_netdev_flow *);
static void dp_netdev_actions_free(struct dp_netdev_actions *); static void dp_netdev_actions_free(struct dp_netdev_actions *);
/* A thread that receives packets from some ports, looks them up in the flow /* PMD: Poll modes drivers. PMD accesses devices via polling to eliminate
* table, and executes the actions it finds. */ * the performance overhead of interrupt processing. Therefore netdev can
struct dp_forwarder { * not implement rx-wait for these devices. dpif-netdev needs to poll
* these device to check for recv buffer. pmd-thread does polling for
* devices assigned to itself thread.
*
* DPDK used PMD for accessing NIC.
*
* A thread that receives packets from PMD ports, looks them up in the flow
* table, and executes the actions it finds.
**/
struct pmd_thread {
struct dp_netdev *dp; struct dp_netdev *dp;
pthread_t thread; pthread_t thread;
int id;
atomic_uint change_seq;
char *name; char *name;
uint32_t min_hash, max_hash;
}; };
/* Interface to netdev-based datapath. */ /* Interface to netdev-based datapath. */
@@ -337,18 +350,16 @@ static int dpif_netdev_open(const struct dpif_class *, const char *name,
static int dp_netdev_output_userspace(struct dp_netdev *dp, struct ofpbuf *, static int dp_netdev_output_userspace(struct dp_netdev *dp, struct ofpbuf *,
int queue_no, int type, int queue_no, int type,
const struct flow *, const struct flow *,
const struct nlattr *userdata) const struct nlattr *userdata);
OVS_EXCLUDED(dp->queue_rwlock);
static void dp_netdev_execute_actions(struct dp_netdev *dp, static void dp_netdev_execute_actions(struct dp_netdev *dp,
const struct flow *, struct ofpbuf *, bool may_steal, const struct flow *, struct ofpbuf *, bool may_steal,
struct pkt_metadata *, struct pkt_metadata *,
const struct nlattr *actions, const struct nlattr *actions,
size_t actions_len) size_t actions_len);
OVS_REQ_RDLOCK(dp->port_rwlock);
static void dp_netdev_port_input(struct dp_netdev *dp, struct ofpbuf *packet, static void dp_netdev_port_input(struct dp_netdev *dp, struct ofpbuf *packet,
struct pkt_metadata *) struct pkt_metadata *);
OVS_REQ_RDLOCK(dp->port_rwlock);
static void dp_netdev_set_threads(struct dp_netdev *, int n); static void dp_netdev_set_pmd_threads(struct dp_netdev *, int n);
static struct dpif_netdev * static struct dpif_netdev *
dpif_netdev_cast(const struct dpif *dpif) dpif_netdev_cast(const struct dpif *dpif)
@@ -485,7 +496,6 @@ create_dp_netdev(const char *name, const struct dpif_class *class,
dp_netdev_free(dp); dp_netdev_free(dp);
return error; return error;
} }
dp_netdev_set_threads(dp, 2);
*dpp = dp; *dpp = dp;
return 0; return 0;
@@ -546,8 +556,8 @@ dp_netdev_free(struct dp_netdev *dp)
shash_find_and_delete(&dp_netdevs, dp->name); shash_find_and_delete(&dp_netdevs, dp->name);
dp_netdev_set_threads(dp, 0); dp_netdev_set_pmd_threads(dp, 0);
free(dp->forwarders); free(dp->pmd_threads);
dp_netdev_flow_flush(dp); dp_netdev_flow_flush(dp);
ovs_rwlock_wrlock(&dp->port_rwlock); ovs_rwlock_wrlock(&dp->port_rwlock);
@@ -641,6 +651,19 @@ dpif_netdev_get_stats(const struct dpif *dpif, struct dpif_dp_stats *stats)
return 0; return 0;
} }
static void
dp_netdev_reload_pmd_threads(struct dp_netdev *dp)
{
int i;
for (i = 0; i < dp->n_pmd_threads; i++) {
struct pmd_thread *f = &dp->pmd_threads[i];
int id;
atomic_add(&f->change_seq, 1, &id);
}
}
static int static int
do_add_port(struct dp_netdev *dp, const char *devname, const char *type, do_add_port(struct dp_netdev *dp, const char *devname, const char *type,
odp_port_t port_no) odp_port_t port_no)
@@ -649,7 +672,6 @@ do_add_port(struct dp_netdev *dp, const char *devname, const char *type,
struct netdev_saved_flags *sf; struct netdev_saved_flags *sf;
struct dp_netdev_port *port; struct dp_netdev_port *port;
struct netdev *netdev; struct netdev *netdev;
struct netdev_rx *rx;
enum netdev_flags flags; enum netdev_flags flags;
const char *open_type; const char *open_type;
int error; int error;
@@ -671,7 +693,11 @@ do_add_port(struct dp_netdev *dp, const char *devname, const char *type,
return EINVAL; return EINVAL;
} }
error = netdev_rx_open(netdev, &rx); port = xzalloc(sizeof *port);
port->port_no = port_no;
port->netdev = netdev;
port->type = xstrdup(type);
error = netdev_rx_open(netdev, &port->rx);
if (error if (error
&& !(error == EOPNOTSUPP && dpif_netdev_class_is_dummy(dp->class))) { && !(error == EOPNOTSUPP && dpif_netdev_class_is_dummy(dp->class))) {
VLOG_ERR("%s: cannot receive packets on this network device (%s)", VLOG_ERR("%s: cannot receive packets on this network device (%s)",
@@ -682,21 +708,23 @@ do_add_port(struct dp_netdev *dp, const char *devname, const char *type,
error = netdev_turn_flags_on(netdev, NETDEV_PROMISC, &sf); error = netdev_turn_flags_on(netdev, NETDEV_PROMISC, &sf);
if (error) { if (error) {
netdev_rx_close(rx); netdev_rx_close(port->rx);
netdev_close(netdev); netdev_close(netdev);
free(port->rx);
free(port);
return error; return error;
} }
port = xmalloc(sizeof *port);
port->port_no = port_no;
port->netdev = netdev;
port->sf = sf; port->sf = sf;
port->rx = rx;
port->type = xstrdup(type); if (netdev_is_pmd(netdev)) {
dp->pmd_count++;
dp_netdev_set_pmd_threads(dp, NR_THREADS);
dp_netdev_reload_pmd_threads(dp);
}
ovs_refcount_init(&port->ref_cnt);
hmap_insert(&dp->ports, &port->node, hash_int(odp_to_u32(port_no), 0)); hmap_insert(&dp->ports, &port->node, hash_int(odp_to_u32(port_no), 0));
seq_change(dp->port_seq); seq_change(dp->port_seq);
ovs_refcount_init(&port->ref_cnt);
return 0; return 0;
} }
@@ -827,6 +855,9 @@ do_del_port(struct dp_netdev *dp, odp_port_t port_no)
hmap_remove(&dp->ports, &port->node); hmap_remove(&dp->ports, &port->node);
seq_change(dp->port_seq); seq_change(dp->port_seq);
if (netdev_is_pmd(port->netdev)) {
dp_netdev_reload_pmd_threads(dp);
}
port_unref(port); port_unref(port);
return 0; return 0;
@@ -1690,116 +1721,199 @@ dp_netdev_actions_free(struct dp_netdev_actions *actions)
free(actions); free(actions);
} }
static void *
dp_forwarder_main(void *f_) inline static void
dp_netdev_process_rx_port(struct dp_netdev *dp,
struct dp_netdev_port *port,
struct netdev_rx *queue)
{ {
struct dp_forwarder *f = f_; struct ofpbuf *packet[NETDEV_MAX_RX_BATCH];
struct dp_netdev *dp = f->dp; int error, c;
f->name = xasprintf("forwarder_%u", ovsthread_id_self()); error = netdev_rx_recv(queue, packet, &c);
set_subprogram_name("%s", f->name); if (!error) {
struct pkt_metadata md = PKT_METADATA_INITIALIZER(port->port_no);
while (!latch_is_set(&dp->exit_latch)) {
bool received_anything;
int i; int i;
ovs_rwlock_rdlock(&dp->port_rwlock); for (i = 0; i < c; i++) {
for (i = 0; i < 50; i++) { dp_netdev_port_input(dp, packet[i], &md);
struct dp_netdev_port *port; }
} else if (error != EAGAIN && error != EOPNOTSUPP) {
static struct vlog_rate_limit rl
= VLOG_RATE_LIMIT_INIT(1, 5);
received_anything = false; VLOG_ERR_RL(&rl, "error receiving data from %s: %s",
HMAP_FOR_EACH (port, node, &f->dp->ports) { netdev_get_name(port->netdev),
if (port->rx ovs_strerror(error));
&& port->node.hash >= f->min_hash }
&& port->node.hash <= f->max_hash) { }
struct ofpbuf *packets[NETDEV_MAX_RX_BATCH];
int count;
int error;
error = netdev_rx_recv(port->rx, packets, &count); static void
if (!error) { dpif_netdev_run(struct dpif *dpif)
int i; {
struct pkt_metadata md struct dp_netdev_port *port;
= PKT_METADATA_INITIALIZER(port->port_no); struct dp_netdev *dp = get_dp_netdev(dpif);
for (i = 0; i < count; i++) { ovs_rwlock_rdlock(&dp->port_rwlock);
dp_netdev_port_input(dp, packets[i], &md);
}
received_anything = true;
} else if (error != EAGAIN && error != EOPNOTSUPP) {
static struct vlog_rate_limit rl
= VLOG_RATE_LIMIT_INIT(1, 5);
VLOG_ERR_RL(&rl, "error receiving data from %s: %s", HMAP_FOR_EACH (port, node, &dp->ports) {
netdev_get_name(port->netdev), if (port->rx && !netdev_is_pmd(port->netdev)) {
ovs_strerror(error)); dp_netdev_process_rx_port(dp, port, port->rx);
} }
} }
ovs_rwlock_unlock(&dp->port_rwlock);
}
static void
dpif_netdev_wait(struct dpif *dpif)
{
struct dp_netdev_port *port;
struct dp_netdev *dp = get_dp_netdev(dpif);
ovs_rwlock_rdlock(&dp->port_rwlock);
HMAP_FOR_EACH (port, node, &dp->ports) {
if (port->rx && !netdev_is_pmd(port->netdev)) {
netdev_rx_wait(port->rx);
}
}
ovs_rwlock_unlock(&dp->port_rwlock);
}
struct rx_poll {
struct dp_netdev_port *port;
};
static int
pmd_load_queues(struct pmd_thread *f,
struct rx_poll **ppoll_list, int poll_cnt)
{
struct dp_netdev *dp = f->dp;
struct rx_poll *poll_list = *ppoll_list;
struct dp_netdev_port *port;
int id = f->id;
int index;
int i;
/* Simple scheduler for netdev rx polling. */
ovs_rwlock_rdlock(&dp->port_rwlock);
for (i = 0; i < poll_cnt; i++) {
port_unref(poll_list[i].port);
}
poll_cnt = 0;
index = 0;
HMAP_FOR_EACH (port, node, &f->dp->ports) {
if (netdev_is_pmd(port->netdev)) {
if ((index % dp->n_pmd_threads) == id) {
poll_list = xrealloc(poll_list, sizeof *poll_list * (poll_cnt + 1));
port_ref(port);
poll_list[poll_cnt++].port = port;
} }
index++;
}
}
if (!received_anything) { ovs_rwlock_unlock(&dp->port_rwlock);
*ppoll_list = poll_list;
return poll_cnt;
}
static void *
pmd_thread_main(void *f_)
{
struct pmd_thread *f = f_;
struct dp_netdev *dp = f->dp;
unsigned int lc = 0;
struct rx_poll *poll_list;
unsigned int port_seq;
int poll_cnt;
int i;
f->name = xasprintf("pmd_%u", ovsthread_id_self());
set_subprogram_name("%s", f->name);
poll_cnt = 0;
poll_list = NULL;
reload:
poll_cnt = pmd_load_queues(f, &poll_list, poll_cnt);
atomic_read(&f->change_seq, &port_seq);
for (;;) {
unsigned int c_port_seq;
int i;
for (i = 0; i < poll_cnt; i++) {
dp_netdev_process_rx_port(dp, poll_list[i].port, poll_list[i].port->rx);
}
if (lc++ > 1024) {
ovsrcu_quiesce();
/* TODO: need completely userspace based signaling method.
* to keep this thread entirely in userspace.
* For now using atomic counter. */
lc = 0;
atomic_read_explicit(&f->change_seq, &c_port_seq, memory_order_consume);
if (c_port_seq != port_seq) {
break; break;
} }
} }
if (received_anything) {
poll_immediate_wake();
} else {
struct dp_netdev_port *port;
HMAP_FOR_EACH (port, node, &f->dp->ports)
if (port->rx
&& port->node.hash >= f->min_hash
&& port->node.hash <= f->max_hash) {
netdev_rx_wait(port->rx);
}
seq_wait(dp->port_seq, seq_read(dp->port_seq));
latch_wait(&dp->exit_latch);
}
ovs_rwlock_unlock(&dp->port_rwlock);
poll_block();
} }
free(f->name); if (!latch_is_set(&f->dp->exit_latch)){
goto reload;
}
for (i = 0; i < poll_cnt; i++) {
port_unref(poll_list[i].port);
}
free(poll_list);
free(f->name);
return NULL; return NULL;
} }
static void static void
dp_netdev_set_threads(struct dp_netdev *dp, int n) dp_netdev_set_pmd_threads(struct dp_netdev *dp, int n)
{ {
int i; int i;
if (n == dp->n_forwarders) { if (n == dp->n_pmd_threads) {
return; return;
} }
/* Stop existing threads. */ /* Stop existing threads. */
latch_set(&dp->exit_latch); latch_set(&dp->exit_latch);
for (i = 0; i < dp->n_forwarders; i++) { dp_netdev_reload_pmd_threads(dp);
struct dp_forwarder *f = &dp->forwarders[i]; for (i = 0; i < dp->n_pmd_threads; i++) {
struct pmd_thread *f = &dp->pmd_threads[i];
xpthread_join(f->thread, NULL); xpthread_join(f->thread, NULL);
} }
latch_poll(&dp->exit_latch); latch_poll(&dp->exit_latch);
free(dp->forwarders); free(dp->pmd_threads);
/* Start new threads. */ /* Start new threads. */
dp->forwarders = xmalloc(n * sizeof *dp->forwarders); dp->pmd_threads = xmalloc(n * sizeof *dp->pmd_threads);
dp->n_forwarders = n; dp->n_pmd_threads = n;
for (i = 0; i < n; i++) { for (i = 0; i < n; i++) {
struct dp_forwarder *f = &dp->forwarders[i]; struct pmd_thread *f = &dp->pmd_threads[i];
f->dp = dp; f->dp = dp;
f->min_hash = UINT32_MAX / n * i; f->id = i;
f->max_hash = UINT32_MAX / n * (i + 1) - 1; atomic_store(&f->change_seq, 1);
if (i == n - 1) {
f->max_hash = UINT32_MAX; /* Each thread will distribute all devices rx-queues among
} * themselves. */
xpthread_create(&f->thread, NULL, dp_forwarder_main, f); xpthread_create(&f->thread, NULL, pmd_thread_main, f);
} }
} }
static void * static void *
dp_netdev_flow_stats_new_cb(void) dp_netdev_flow_stats_new_cb(void)
@@ -1851,7 +1965,6 @@ dp_netdev_count_packet(struct dp_netdev *dp, enum dp_stat_type type)
static void static void
dp_netdev_port_input(struct dp_netdev *dp, struct ofpbuf *packet, dp_netdev_port_input(struct dp_netdev *dp, struct ofpbuf *packet,
struct pkt_metadata *md) struct pkt_metadata *md)
OVS_REQ_RDLOCK(dp->port_rwlock)
{ {
struct dp_netdev_flow *netdev_flow; struct dp_netdev_flow *netdev_flow;
struct flow key; struct flow key;
@@ -1884,7 +1997,6 @@ static int
dp_netdev_output_userspace(struct dp_netdev *dp, struct ofpbuf *packet, dp_netdev_output_userspace(struct dp_netdev *dp, struct ofpbuf *packet,
int queue_no, int type, const struct flow *flow, int queue_no, int type, const struct flow *flow,
const struct nlattr *userdata) const struct nlattr *userdata)
OVS_EXCLUDED(dp->queue_rwlock)
{ {
struct dp_netdev_queue *q; struct dp_netdev_queue *q;
int error; int error;
@@ -1992,7 +2104,6 @@ dp_netdev_execute_actions(struct dp_netdev *dp, const struct flow *key,
struct ofpbuf *packet, bool may_steal, struct ofpbuf *packet, bool may_steal,
struct pkt_metadata *md, struct pkt_metadata *md,
const struct nlattr *actions, size_t actions_len) const struct nlattr *actions, size_t actions_len)
OVS_REQ_RDLOCK(dp->port_rwlock)
{ {
struct dp_netdev_execute_aux aux = {dp, key}; struct dp_netdev_execute_aux aux = {dp, key};
@@ -2007,8 +2118,8 @@ const struct dpif_class dpif_netdev_class = {
dpif_netdev_open, dpif_netdev_open,
dpif_netdev_close, dpif_netdev_close,
dpif_netdev_destroy, dpif_netdev_destroy,
NULL, /* run */ dpif_netdev_run,
NULL, /* wait */ dpif_netdev_wait,
dpif_netdev_get_stats, dpif_netdev_get_stats,
dpif_netdev_port_add, dpif_netdev_port_add,
dpif_netdev_port_del, dpif_netdev_port_del,

6
lib/netdev.c
View File

@@ -90,6 +90,12 @@ static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 20);
static void restore_all_flags(void *aux OVS_UNUSED); static void restore_all_flags(void *aux OVS_UNUSED);
void update_device_args(struct netdev *, const struct shash *args); void update_device_args(struct netdev *, const struct shash *args);
bool
netdev_is_pmd(const struct netdev *netdev)
{
return !strcmp(netdev->netdev_class->type, "dpdk");
}
static void static void
netdev_initialize(void) netdev_initialize(void)
OVS_EXCLUDED(netdev_class_mutex, netdev_mutex) OVS_EXCLUDED(netdev_class_mutex, netdev_mutex)

5
lib/netdev.h
View File

@@ -134,8 +134,11 @@ void netdev_wait(void);
void netdev_enumerate_types(struct sset *types); void netdev_enumerate_types(struct sset *types);
bool netdev_is_reserved_name(const char *name); bool netdev_is_reserved_name(const char *name);
bool netdev_is_pmd(const struct netdev *netdev);
/* Open and close. */ /* Open and close. */
int netdev_open(const char *name, const char *type, struct netdev **); int netdev_open(const char *name, const char *type, struct netdev **netdevp);
struct netdev *netdev_ref(const struct netdev *); struct netdev *netdev_ref(const struct netdev *);
void netdev_close(struct netdev *); void netdev_close(struct netdev *);