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 <[email protected]> --- lib/dpif-netdev.c | 311 ++++++++++++++++++++++++++++++++++++----------------- lib/dpif-netdev.h | 2 + lib/netdev.c | 6 ++ lib/netdev.h | 5 +- 4 files changed, 225 insertions(+), 99 deletions(-) diff --git a/lib/dpif-netdev.c b/lib/dpif-netdev.c index dda8fe5..6467874 100644 --- a/lib/dpif-netdev.c +++ b/lib/dpif-netdev.c @@ -159,8 +159,9 @@ struct dp_netdev { /* Forwarding threads. */ struct latch exit_latch; - struct dp_forwarder *forwarders; - size_t n_forwarders; + struct pmd_thread *pmd_threads; + size_t n_pmd_threads; + int pmd_count; }; static struct dp_netdev_port *dp_netdev_lookup_port(const struct dp_netdev *dp, @@ -169,13 +170,14 @@ static struct dp_netdev_port *dp_netdev_lookup_port(const struct dp_netdev *dp, /* A port in a netdev-based datapath. */ struct dp_netdev_port { - struct hmap_node node; /* Node in dp_netdev's 'ports'. */ - odp_port_t port_no; + struct pkt_metadata md; + struct netdev_rx *rx; struct netdev *netdev; + odp_port_t port_no; struct netdev_saved_flags *sf; - struct netdev_rx *rx; - struct ovs_refcount ref_cnt; char *type; /* Port type as requested by user. */ + struct ovs_refcount ref_cnt; + struct hmap_node node; /* Node in dp_netdev's 'ports'. */ }; /* A flow in dp_netdev's 'flow_table'. @@ -287,11 +289,12 @@ void dp_netdev_actions_unref(struct dp_netdev_actions *); /* A thread that receives packets from some ports, looks them up in the flow * table, and executes the actions it finds. */ -struct dp_forwarder { +struct pmd_thread { struct dp_netdev *dp; - pthread_t thread; + int id; + atomic_uint change_seq; char *name; - uint32_t min_hash, max_hash; + pthread_t thread; }; /* Interface to netdev-based datapath. */ @@ -330,7 +333,7 @@ static void dp_netdev_execute_actions(struct dp_netdev *dp, static void dp_netdev_port_input(struct dp_netdev *dp, struct ofpbuf *packet, 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 * dpif_netdev_cast(const struct dpif *dpif) @@ -476,7 +479,6 @@ create_dp_netdev(const char *name, const struct dpif_class *class, dp_netdev_free(dp); return error; } - dp_netdev_set_threads(dp, 2); *dpp = dp; return 0; @@ -534,8 +536,8 @@ dp_netdev_free(struct dp_netdev *dp) shash_find_and_delete(&dp_netdevs, dp->name); - dp_netdev_set_threads(dp, 0); - free(dp->forwarders); + dp_netdev_set_pmd_threads(dp, 0); + free(dp->pmd_threads); dp_netdev_flow_flush(dp); ovs_rwlock_wrlock(&dp->port_rwlock); @@ -617,6 +619,19 @@ dpif_netdev_get_stats(const struct dpif *dpif, struct dpif_dp_stats *stats) 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 do_add_port(struct dp_netdev *dp, const char *devname, const char *type, odp_port_t port_no) @@ -625,7 +640,6 @@ do_add_port(struct dp_netdev *dp, const char *devname, const char *type, struct netdev_saved_flags *sf; struct dp_netdev_port *port; struct netdev *netdev; - struct netdev_rx *rx; enum netdev_flags flags; const char *open_type; int error; @@ -647,7 +661,12 @@ do_add_port(struct dp_netdev *dp, const char *devname, const char *type, return EINVAL; } - error = netdev_rx_open(netdev, &rx); + port = xzalloc(sizeof *port); + port->port_no = port_no; + port->md = PKT_METADATA_INITIALIZER(port->port_no); + port->netdev = netdev; + port->type = xstrdup(type); + error = netdev_rx_open(netdev, &port->rx); if (error && !(error == EOPNOTSUPP && dpif_netdev_class_is_dummy(dp->class))) { VLOG_ERR("%s: cannot receive packets on this network device (%s)", @@ -658,21 +677,23 @@ do_add_port(struct dp_netdev *dp, const char *devname, const char *type, error = netdev_turn_flags_on(netdev, NETDEV_PROMISC, &sf); if (error) { - netdev_rx_close(rx); + netdev_rx_close(port->rx); netdev_close(netdev); + free(port->rx); + free(port); return error; } - - port = xmalloc(sizeof *port); - port->port_no = port_no; - port->netdev = netdev; 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)); seq_change(dp->port_seq); - ovs_refcount_init(&port->ref_cnt); return 0; } @@ -766,9 +787,10 @@ port_unref(struct dp_netdev_port *port) { if (port && ovs_refcount_unref(&port->ref_cnt) == 1) { - netdev_close(port->netdev); - netdev_restore_flags(port->sf); netdev_rx_close(port->rx); + + netdev_restore_flags(port->sf); + netdev_close(port->netdev); free(port->type); free(port); } @@ -804,6 +826,9 @@ do_del_port(struct dp_netdev *dp, odp_port_t port_no) hmap_remove(&dp->ports, &port->node); seq_change(dp->port_seq); + if (netdev_is_pmd(port->netdev)) { + dp_netdev_reload_pmd_threads(dp); + } port_unref(port); return 0; @@ -1598,116 +1623,206 @@ dp_netdev_actions_unref(struct dp_netdev_actions *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 ofpbuf *packet[MAX_RX_BATCH]; + struct pkt_metadata *md = &port->md; + int error, c; + + error = netdev_rx_recv(queue, packet, &c); + if (!error) { + int i; + + for (i = 0; i < c; i++) { + dp_netdev_port_input(dp, packet[i], md); + } + + } 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", + netdev_get_name(port->netdev), + ovs_strerror(error)); + } +} + +static void +dpif_netdev_run(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)) { + 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_forwarder *f = f_; 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; - f->name = xasprintf("forwarder_%u", ovsthread_id_self()); - set_subprogram_name("%s", f->name); + /* 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); + } - while (!latch_is_set(&dp->exit_latch)) { - bool received_anything; - int i; + free(poll_list); + poll_cnt = 0; + index = 0; - ovs_rwlock_rdlock(&dp->port_rwlock); - for (i = 0; i < 50; i++) { - struct dp_netdev_port *port; - - received_anything = false; - HMAP_FOR_EACH (port, node, &f->dp->ports) { - if (port->rx - && port->node.hash >= f->min_hash - && port->node.hash <= f->max_hash) { - struct ofpbuf *packets[MAX_RX_BATCH]; - int count; - int error; - - error = netdev_rx_recv(port->rx, packets, &count); - if (!error) { - int i; - struct pkt_metadata md - = PKT_METADATA_INITIALIZER(port->port_no); - - for (i = 0; i < count; i++) { - 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", - netdev_get_name(port->netdev), - ovs_strerror(error)); - } - } + HMAP_FOR_EACH (port, node, &f->dp->ports) { + if (netdev_is_pmd(port->netdev)) { + if ((index % dp->n_pmd_threads) == id) { + port_ref(port); + poll_cnt++; } + index++; + } + } - if (!received_anything) { - break; + poll_list = xzalloc(sizeof *poll_list * poll_cnt); + 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[poll_cnt].port = port; + poll_cnt++; + VLOG_INFO("poll_cnt %d port = %d i = %d",poll_cnt,port->port_no, i); } + index++; } + } - if (received_anything) { - poll_immediate_wake(); - } else { - struct dp_netdev_port *port; + ovs_rwlock_unlock(&dp->port_rwlock); + *ppoll_list = poll_list; + return poll_cnt; +} - 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); +static void * +pmd_thread_main(void *f_) +{ + struct pmd_thread *f = f_; + struct dp_netdev *dp = f->dp; + unsigned long lc = 0; + struct rx_poll *poll_list; + unsigned int port_seq; + int poll_cnt; + + 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); + + while (1) { + 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++ > (64 * 1024 * 1024)) { + /* TODO: need completely userspace based signaling method. + * to keep this thread entirely in userspace. + * For now using atomic counter. */ + lc = 0; + atomic_read(&f->change_seq, &c_port_seq); + if (c_port_seq != port_seq) { + break; + } } - ovs_rwlock_unlock(&dp->port_rwlock); + } - poll_block(); + if (!latch_is_set(&f->dp->exit_latch)){ + goto reload; } + free(poll_list); free(f->name); - return NULL; } 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; - if (n == dp->n_forwarders) { + if (n == dp->n_pmd_threads) { return; } /* Stop existing threads. */ latch_set(&dp->exit_latch); - for (i = 0; i < dp->n_forwarders; i++) { - struct dp_forwarder *f = &dp->forwarders[i]; + dp_netdev_reload_pmd_threads(dp); + for (i = 0; i < dp->n_pmd_threads; i++) { + struct pmd_thread *f = &dp->pmd_threads[i]; xpthread_join(f->thread, NULL); } latch_poll(&dp->exit_latch); - free(dp->forwarders); + free(dp->pmd_threads); /* Start new threads. */ - dp->forwarders = xmalloc(n * sizeof *dp->forwarders); - dp->n_forwarders = n; + dp->pmd_threads = xmalloc(n * sizeof *dp->pmd_threads); + dp->n_pmd_threads = n; + 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->min_hash = UINT32_MAX / n * i; - f->max_hash = UINT32_MAX / n * (i + 1) - 1; - if (i == n - 1) { - f->max_hash = UINT32_MAX; - } - xpthread_create(&f->thread, NULL, dp_forwarder_main, f); + f->id = i; + atomic_store(&f->change_seq, 1); + + /* Each thread will distribute all devices rx-queues among + * themselves. */ + xpthread_create(&f->thread, NULL, pmd_thread_main, f); } } + static void dp_netdev_flow_used(struct dp_netdev_flow *netdev_flow, @@ -1872,8 +1987,8 @@ const struct dpif_class dpif_netdev_class = { dpif_netdev_open, dpif_netdev_close, dpif_netdev_destroy, - NULL, /* run */ - NULL, /* wait */ + dpif_netdev_run, + dpif_netdev_wait, dpif_netdev_get_stats, dpif_netdev_port_add, dpif_netdev_port_del, diff --git a/lib/dpif-netdev.h b/lib/dpif-netdev.h index 64c93ee..76f261b 100644 --- a/lib/dpif-netdev.h +++ b/lib/dpif-netdev.h @@ -33,6 +33,8 @@ enum { DP_NETDEV_HEADROOM = 2 + VLAN_HEADER_LEN }; enum { MAX_RX_BATCH = 256 }; /* Maximum number of flows in flow table. */ +#define NR_THREADS 1 + #ifdef __cplusplus } #endif diff --git a/lib/netdev.c b/lib/netdev.c index 295e121..058874f 100644 --- a/lib/netdev.c +++ b/lib/netdev.c @@ -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); 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 netdev_initialize(void) OVS_EXCLUDED(netdev_class_mutex, netdev_mutex) diff --git a/lib/netdev.h b/lib/netdev.h index 2b4e2a9..636bc8c 100644 --- a/lib/netdev.h +++ b/lib/netdev.h @@ -134,8 +134,11 @@ void netdev_wait(void); void netdev_enumerate_types(struct sset *types); bool netdev_is_reserved_name(const char *name); +bool netdev_is_pmd(const struct netdev *netdev); + /* 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 *); void netdev_close(struct netdev *); -- 1.7.9.5 _______________________________________________ dev mailing list [email protected] http://openvswitch.org/mailman/listinfo/dev
