X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=lib%2Fdpif-linux.c;h=e2300d6c297987354d6aa3687f276133334f3842;hb=b3dcb73cc58a7ee089cfaec623818142196e1172;hp=3cf1225ff29f73549dbd9470b484da4a4e33109c;hpb=26508d9a5572cf4e980573552175d620c8f09d3e;p=sliver-openvswitch.git

diff --git a/lib/dpif-linux.c b/lib/dpif-linux.c
index 3cf1225ff..e2300d6c2 100644
--- a/lib/dpif-linux.c
+++ b/lib/dpif-linux.c
@@ -46,8 +46,6 @@
 #include "netlink.h"
 #include "odp-util.h"
 #include "ofpbuf.h"
-#include "openvswitch/datapath-compat.h"
-#include "openvswitch/tunnel.h"
 #include "packets.h"
 #include "poll-loop.h"
 #include "random.h"
@@ -106,6 +104,8 @@ struct dpif_linux_flow {
      * the Netlink version of the command, even if actions_len is zero. */
     const struct nlattr *key;           /* OVS_FLOW_ATTR_KEY. */
     size_t key_len;
+    const struct nlattr *mask;          /* OVS_FLOW_ATTR_MASK. */
+    size_t mask_len;
     const struct nlattr *actions;       /* OVS_FLOW_ATTR_ACTIONS. */
     size_t actions_len;
     const struct ovs_flow_stats *stats; /* OVS_FLOW_ATTR_STATS. */
@@ -139,6 +139,7 @@ struct dpif_linux {
     int dp_ifindex;
 
     /* Upcall messages. */
+    struct ovs_mutex upcall_lock;
     int uc_array_size;          /* Size of 'channels' and 'epoll_events'. */
     struct dpif_channel *channels;
     struct epoll_event *epoll_events;
@@ -147,28 +148,28 @@ struct dpif_linux {
     int event_offset;           /* Offset into 'epoll_events'. */
 
     /* Change notification. */
-    struct sset changed_ports;  /* Ports that have changed. */
-    struct nln_notifier *port_notifier;
-    bool change_error;
+    struct nl_sock *port_notifier; /* vport multicast group subscriber. */
 };
 
 static struct vlog_rate_limit error_rl = VLOG_RATE_LIMIT_INIT(9999, 5);
 
-/* Generic Netlink family numbers for OVS. */
+/* Generic Netlink family numbers for OVS.
+ *
+ * Initialized by dpif_linux_init(). */
 static int ovs_datapath_family;
 static int ovs_vport_family;
 static int ovs_flow_family;
 static int ovs_packet_family;
 
-/* Generic Netlink socket. */
-static struct nl_sock *genl_sock;
-static struct nln *nln = NULL;
+/* Generic Netlink multicast groups for OVS.
+ *
+ * Initialized by dpif_linux_init(). */
+static unsigned int ovs_vport_mcgroup;
 
 static int dpif_linux_init(void);
-static void open_dpif(const struct dpif_linux_dp *, struct dpif **);
-static bool dpif_linux_nln_parse(struct ofpbuf *, void *);
-static void dpif_linux_port_changed(const void *vport, void *dpif);
-static uint32_t dpif_linux_port_get_pid(const struct dpif *, uint32_t port_no);
+static int open_dpif(const struct dpif_linux_dp *, struct dpif **);
+static uint32_t dpif_linux_port_get_pid(const struct dpif *,
+                                        odp_port_t port_no);
 
 static void dpif_linux_vport_to_ofpbuf(const struct dpif_linux_vport *,
                                        struct ofpbuf *);
@@ -234,33 +235,34 @@ dpif_linux_open(const struct dpif_class *class OVS_UNUSED, const char *name,
         return error;
     }
 
-    open_dpif(&dp, dpifp);
+    error = open_dpif(&dp, dpifp);
     ofpbuf_delete(buf);
-    return 0;
+    return error;
 }
 
-static void
+static int
 open_dpif(const struct dpif_linux_dp *dp, struct dpif **dpifp)
 {
     struct dpif_linux *dpif;
 
     dpif = xzalloc(sizeof *dpif);
-    dpif->port_notifier = nln_notifier_create(nln, dpif_linux_port_changed,
-                                              dpif);
+    dpif->port_notifier = NULL;
+    ovs_mutex_init(&dpif->upcall_lock);
     dpif->epoll_fd = -1;
 
     dpif_init(&dpif->dpif, &dpif_linux_class, dp->name,
               dp->dp_ifindex, dp->dp_ifindex);
 
     dpif->dp_ifindex = dp->dp_ifindex;
-    sset_init(&dpif->changed_ports);
     *dpifp = &dpif->dpif;
+
+    return 0;
 }
 
 static void
 destroy_channels(struct dpif_linux *dpif)
 {
-    int i;
+    unsigned int i;
 
     if (dpif->epoll_fd < 0) {
         return;
@@ -275,11 +277,13 @@ destroy_channels(struct dpif_linux *dpif)
             continue;
         }
 
+        epoll_ctl(dpif->epoll_fd, EPOLL_CTL_DEL, nl_sock_fd(ch->sock), NULL);
+
         /* Turn off upcalls. */
         dpif_linux_vport_init(&vport_request);
         vport_request.cmd = OVS_VPORT_CMD_SET;
         vport_request.dp_ifindex = dpif->dp_ifindex;
-        vport_request.port_no = i;
+        vport_request.port_no = u32_to_odp(i);
         vport_request.upcall_pid = &upcall_pid;
         dpif_linux_vport_transact(&vport_request, NULL, NULL);
 
@@ -294,14 +298,15 @@ destroy_channels(struct dpif_linux *dpif)
     dpif->epoll_events = NULL;
     dpif->n_events = dpif->event_offset = 0;
 
-    close(dpif->epoll_fd);
-    dpif->epoll_fd = -1;
+    /* Don't close dpif->epoll_fd since that would cause other threads that
+     * call dpif_recv_wait(dpif) to wait on an arbitrary fd or a closed fd. */
 }
 
 static int
-add_channel(struct dpif_linux *dpif, uint32_t port_no, struct nl_sock *sock)
+add_channel(struct dpif_linux *dpif, odp_port_t port_no, struct nl_sock *sock)
 {
     struct epoll_event event;
+    uint32_t port_idx = odp_to_u32(port_no);
 
     if (dpif->epoll_fd < 0) {
         return 0;
@@ -309,11 +314,11 @@ add_channel(struct dpif_linux *dpif, uint32_t port_no, struct nl_sock *sock)
 
     /* We assume that the datapath densely chooses port numbers, which
      * can therefore be used as an index into an array of channels. */
-    if (port_no >= dpif->uc_array_size) {
-        int new_size = port_no + 1;
-        int i;
+    if (port_idx >= dpif->uc_array_size) {
+        uint32_t new_size = port_idx + 1;
+        uint32_t i;
 
-        if (new_size > 65535) {
+        if (new_size > MAX_PORTS) {
             VLOG_WARN_RL(&error_rl, "%s: datapath port %"PRIu32" too big",
                          dpif_name(&dpif->dpif), port_no);
             return EFBIG;
@@ -332,33 +337,36 @@ add_channel(struct dpif_linux *dpif, uint32_t port_no, struct nl_sock *sock)
 
     memset(&event, 0, sizeof event);
     event.events = EPOLLIN;
-    event.data.u32 = port_no;
+    event.data.u32 = port_idx;
     if (epoll_ctl(dpif->epoll_fd, EPOLL_CTL_ADD, nl_sock_fd(sock),
                   &event) < 0) {
         return errno;
     }
 
-    dpif->channels[port_no].sock = sock;
-    dpif->channels[port_no].last_poll = LLONG_MIN;
+    nl_sock_destroy(dpif->channels[port_idx].sock);
+    dpif->channels[port_idx].sock = sock;
+    dpif->channels[port_idx].last_poll = LLONG_MIN;
 
     return 0;
 }
 
 static void
-del_channel(struct dpif_linux *dpif, uint32_t port_no)
+del_channel(struct dpif_linux *dpif, odp_port_t port_no)
 {
     struct dpif_channel *ch;
+    uint32_t port_idx = odp_to_u32(port_no);
 
-    if (dpif->epoll_fd < 0 || port_no >= dpif->uc_array_size) {
+    if (dpif->epoll_fd < 0 || port_idx >= dpif->uc_array_size) {
         return;
     }
 
-    ch = &dpif->channels[port_no];
+    ch = &dpif->channels[port_idx];
     if (!ch->sock) {
         return;
     }
 
     epoll_ctl(dpif->epoll_fd, EPOLL_CTL_DEL, nl_sock_fd(ch->sock), NULL);
+    dpif->event_offset = dpif->n_events = 0;
 
     nl_sock_destroy(ch->sock);
     ch->sock = NULL;
@@ -369,9 +377,12 @@ dpif_linux_close(struct dpif *dpif_)
 {
     struct dpif_linux *dpif = dpif_linux_cast(dpif_);
 
-    nln_notifier_destroy(dpif->port_notifier);
+    nl_sock_destroy(dpif->port_notifier);
     destroy_channels(dpif);
-    sset_destroy(&dpif->changed_ports);
+    if (dpif->epoll_fd >= 0) {
+        close(dpif->epoll_fd);
+    }
+    ovs_mutex_destroy(&dpif->upcall_lock);
     free(dpif);
 }
 
@@ -387,22 +398,6 @@ dpif_linux_destroy(struct dpif *dpif_)
     return dpif_linux_dp_transact(&dp, NULL, NULL);
 }
 
-static void
-dpif_linux_run(struct dpif *dpif_ OVS_UNUSED)
-{
-    if (nln) {
-        nln_run(nln);
-    }
-}
-
-static void
-dpif_linux_wait(struct dpif *dpif OVS_UNUSED)
-{
-    if (nln) {
-        nln_wait(nln);
-    }
-}
-
 static int
 dpif_linux_get_stats(const struct dpif *dpif_, struct dpif_dp_stats *stats)
 {
@@ -439,12 +434,12 @@ get_vport_type(const struct dpif_linux_vport *vport)
     case OVS_VPORT_TYPE_GRE64:
         return "gre64";
 
-    case OVS_VPORT_TYPE_CAPWAP:
-        return "capwap";
-
     case OVS_VPORT_TYPE_VXLAN:
         return "vxlan";
 
+    case OVS_VPORT_TYPE_LISP:
+        return "lisp";
+
     case OVS_VPORT_TYPE_UNSPEC:
     case __OVS_VPORT_TYPE_MAX:
         break;
@@ -468,22 +463,24 @@ netdev_to_ovs_vport_type(const struct netdev *netdev)
         return OVS_VPORT_TYPE_GRE64;
     } else if (strstr(type, "gre")) {
         return OVS_VPORT_TYPE_GRE;
-    } else if (!strcmp(type, "capwap")) {
-        return OVS_VPORT_TYPE_CAPWAP;
     } else if (!strcmp(type, "vxlan")) {
         return OVS_VPORT_TYPE_VXLAN;
+    } else if (!strcmp(type, "lisp")) {
+        return OVS_VPORT_TYPE_LISP;
     } else {
         return OVS_VPORT_TYPE_UNSPEC;
     }
 }
 
 static int
-dpif_linux_port_add(struct dpif *dpif_, struct netdev *netdev,
-                    uint32_t *port_nop)
+dpif_linux_port_add__(struct dpif *dpif_, struct netdev *netdev,
+                      odp_port_t *port_nop)
 {
     struct dpif_linux *dpif = dpif_linux_cast(dpif_);
     const struct netdev_tunnel_config *tnl_cfg;
-    const char *name = netdev_vport_get_dpif_port(netdev);
+    char namebuf[NETDEV_VPORT_NAME_BUFSIZE];
+    const char *name = netdev_vport_get_dpif_port(netdev,
+                                                  namebuf, sizeof namebuf);
     const char *type = netdev_get_type(netdev);
     struct dpif_linux_vport request, reply;
     struct nl_sock *sock = NULL;
@@ -521,7 +518,7 @@ dpif_linux_port_add(struct dpif *dpif_, struct netdev *netdev,
     if (tnl_cfg && tnl_cfg->dst_port != 0) {
         ofpbuf_use_stack(&options, options_stub, sizeof options_stub);
         nl_msg_put_u16(&options, OVS_TUNNEL_ATTR_DST_PORT,
-                       htons(tnl_cfg->dst_port));
+                       ntohs(tnl_cfg->dst_port));
         request.options = options.data;
         request.options_len = options.size;
     }
@@ -536,7 +533,7 @@ dpif_linux_port_add(struct dpif *dpif_, struct netdev *netdev,
         VLOG_DBG("%s: assigning port %"PRIu32" to netlink pid %"PRIu32,
                  dpif_name(dpif_), reply.port_no, upcall_pid);
     } else {
-        if (error == EBUSY && *port_nop != UINT32_MAX) {
+        if (error == EBUSY && *port_nop != ODPP_NONE) {
             VLOG_INFO("%s: requested port %"PRIu32" is in use",
                       dpif_name(dpif_), *port_nop);
         }
@@ -568,7 +565,21 @@ dpif_linux_port_add(struct dpif *dpif_, struct netdev *netdev,
 }
 
 static int
-dpif_linux_port_del(struct dpif *dpif_, uint32_t port_no)
+dpif_linux_port_add(struct dpif *dpif_, struct netdev *netdev,
+                    odp_port_t *port_nop)
+{
+    struct dpif_linux *dpif = dpif_linux_cast(dpif_);
+    int error;
+
+    ovs_mutex_lock(&dpif->upcall_lock);
+    error = dpif_linux_port_add__(dpif_, netdev, port_nop);
+    ovs_mutex_unlock(&dpif->upcall_lock);
+
+    return error;
+}
+
+static int
+dpif_linux_port_del__(struct dpif *dpif_, odp_port_t port_no)
 {
     struct dpif_linux *dpif = dpif_linux_cast(dpif_);
     struct dpif_linux_vport vport;
@@ -586,7 +597,20 @@ dpif_linux_port_del(struct dpif *dpif_, uint32_t port_no)
 }
 
 static int
-dpif_linux_port_query__(const struct dpif *dpif, uint32_t port_no,
+dpif_linux_port_del(struct dpif *dpif_, odp_port_t port_no)
+{
+    struct dpif_linux *dpif = dpif_linux_cast(dpif_);
+    int error;
+
+    ovs_mutex_lock(&dpif->upcall_lock);
+    error = dpif_linux_port_del__(dpif_, port_no);
+    ovs_mutex_unlock(&dpif->upcall_lock);
+
+    return error;
+}
+
+static int
+dpif_linux_port_query__(const struct dpif *dpif, odp_port_t port_no,
                         const char *port_name, struct dpif_port *dpif_port)
 {
     struct dpif_linux_vport request;
@@ -617,7 +641,7 @@ dpif_linux_port_query__(const struct dpif *dpif, uint32_t port_no,
 }
 
 static int
-dpif_linux_port_query_by_number(const struct dpif *dpif, uint32_t port_no,
+dpif_linux_port_query_by_number(const struct dpif *dpif, odp_port_t port_no,
                                 struct dpif_port *dpif_port)
 {
     return dpif_linux_port_query__(dpif, port_no, NULL, dpif_port);
@@ -630,31 +654,35 @@ dpif_linux_port_query_by_name(const struct dpif *dpif, const char *devname,
     return dpif_linux_port_query__(dpif, 0, devname, dpif_port);
 }
 
-static int
+static uint32_t
 dpif_linux_get_max_ports(const struct dpif *dpif OVS_UNUSED)
 {
     return MAX_PORTS;
 }
 
 static uint32_t
-dpif_linux_port_get_pid(const struct dpif *dpif_, uint32_t port_no)
+dpif_linux_port_get_pid(const struct dpif *dpif_, odp_port_t port_no)
 {
     struct dpif_linux *dpif = dpif_linux_cast(dpif_);
+    uint32_t port_idx = odp_to_u32(port_no);
+    uint32_t pid = 0;
 
-    if (dpif->epoll_fd < 0) {
-        return 0;
-    } else {
-        /* The UINT32_MAX "reserved" port number uses the "ovs-system"'s
+    ovs_mutex_lock(&dpif->upcall_lock);
+    if (dpif->epoll_fd >= 0) {
+        /* The ODPP_NONE "reserved" port number uses the "ovs-system"'s
          * channel, since it is not heavily loaded. */
-        int idx = (port_no >= dpif->uc_array_size) ? 0 : port_no;
-        return nl_sock_pid(dpif->channels[idx].sock);
+        uint32_t idx = port_idx >= dpif->uc_array_size ? 0 : port_idx;
+        pid = nl_sock_pid(dpif->channels[idx].sock);
     }
+    ovs_mutex_unlock(&dpif->upcall_lock);
+
+    return pid;
 }
 
 static int
 dpif_linux_flow_flush(struct dpif *dpif_)
 {
-    struct dpif_linux *dpif = dpif_linux_cast(dpif_);
+    const struct dpif_linux *dpif = dpif_linux_cast(dpif_);
     struct dpif_linux_flow flow;
 
     dpif_linux_flow_init(&flow);
@@ -670,7 +698,7 @@ struct dpif_linux_port_state {
 static int
 dpif_linux_port_dump_start(const struct dpif *dpif_, void **statep)
 {
-    struct dpif_linux *dpif = dpif_linux_cast(dpif_);
+    const struct dpif_linux *dpif = dpif_linux_cast(dpif_);
     struct dpif_linux_port_state *state;
     struct dpif_linux_vport request;
     struct ofpbuf *buf;
@@ -683,7 +711,7 @@ dpif_linux_port_dump_start(const struct dpif *dpif_, void **statep)
 
     buf = ofpbuf_new(1024);
     dpif_linux_vport_to_ofpbuf(&request, buf);
-    nl_dump_start(&state->dump, genl_sock, buf);
+    nl_dump_start(&state->dump, NETLINK_GENERIC, buf);
     ofpbuf_delete(buf);
 
     return 0;
@@ -728,23 +756,74 @@ dpif_linux_port_poll(const struct dpif *dpif_, char **devnamep)
 {
     struct dpif_linux *dpif = dpif_linux_cast(dpif_);
 
-    if (dpif->change_error) {
-        dpif->change_error = false;
-        sset_clear(&dpif->changed_ports);
+    /* Lazily create the Netlink socket to listen for notifications. */
+    if (!dpif->port_notifier) {
+        struct nl_sock *sock;
+        int error;
+
+        error = nl_sock_create(NETLINK_GENERIC, &sock);
+        if (error) {
+            return error;
+        }
+
+        error = nl_sock_join_mcgroup(sock, ovs_vport_mcgroup);
+        if (error) {
+            nl_sock_destroy(sock);
+            return error;
+        }
+        dpif->port_notifier = sock;
+
+        /* We have no idea of the current state so report that everything
+         * changed. */
         return ENOBUFS;
-    } else if (!sset_is_empty(&dpif->changed_ports)) {
-        *devnamep = sset_pop(&dpif->changed_ports);
-        return 0;
-    } else {
-        return EAGAIN;
+    }
+
+    for (;;) {
+        static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
+        uint64_t buf_stub[4096 / 8];
+        struct ofpbuf buf;
+        int error;
+
+        ofpbuf_use_stub(&buf, buf_stub, sizeof buf_stub);
+        error = nl_sock_recv(dpif->port_notifier, &buf, false);
+        if (!error) {
+            struct dpif_linux_vport vport;
+
+            error = dpif_linux_vport_from_ofpbuf(&vport, &buf);
+            if (!error) {
+                if (vport.dp_ifindex == dpif->dp_ifindex
+                    && (vport.cmd == OVS_VPORT_CMD_NEW
+                        || vport.cmd == OVS_VPORT_CMD_DEL
+                        || vport.cmd == OVS_VPORT_CMD_SET)) {
+                    VLOG_DBG("port_changed: dpif:%s vport:%s cmd:%"PRIu8,
+                             dpif->dpif.full_name, vport.name, vport.cmd);
+                    *devnamep = xstrdup(vport.name);
+                    ofpbuf_uninit(&buf);
+                    return 0;
+                }
+            }
+        } else if (error != EAGAIN) {
+            VLOG_WARN_RL(&rl, "error reading or parsing netlink (%s)",
+                         ovs_strerror(error));
+            nl_sock_drain(dpif->port_notifier);
+            error = ENOBUFS;
+        }
+
+        ofpbuf_uninit(&buf);
+        if (error) {
+            return error;
+        }
     }
 }
 
 static void
 dpif_linux_port_poll_wait(const struct dpif *dpif_)
 {
-    struct dpif_linux *dpif = dpif_linux_cast(dpif_);
-    if (!sset_is_empty(&dpif->changed_ports) || dpif->change_error) {
+    const struct dpif_linux *dpif = dpif_linux_cast(dpif_);
+
+    if (dpif->port_notifier) {
+        nl_sock_wait(dpif->port_notifier, POLLIN);
+    } else {
         poll_immediate_wake();
     }
 }
@@ -754,7 +833,7 @@ dpif_linux_flow_get__(const struct dpif *dpif_,
                       const struct nlattr *key, size_t key_len,
                       struct dpif_linux_flow *reply, struct ofpbuf **bufp)
 {
-    struct dpif_linux *dpif = dpif_linux_cast(dpif_);
+    const struct dpif_linux *dpif = dpif_linux_cast(dpif_);
     struct dpif_linux_flow request;
 
     dpif_linux_flow_init(&request);
@@ -794,9 +873,9 @@ static void
 dpif_linux_init_flow_put(struct dpif *dpif_, const struct dpif_flow_put *put,
                          struct dpif_linux_flow *request)
 {
-    static struct nlattr dummy_action;
+    static const struct nlattr dummy_action;
 
-    struct dpif_linux *dpif = dpif_linux_cast(dpif_);
+    const struct dpif_linux *dpif = dpif_linux_cast(dpif_);
 
     dpif_linux_flow_init(request);
     request->cmd = (put->flags & DPIF_FP_CREATE
@@ -804,8 +883,12 @@ dpif_linux_init_flow_put(struct dpif *dpif_, const struct dpif_flow_put *put,
     request->dp_ifindex = dpif->dp_ifindex;
     request->key = put->key;
     request->key_len = put->key_len;
+    request->mask = put->mask;
+    request->mask_len = put->mask_len;
     /* Ensure that OVS_FLOW_ATTR_ACTIONS will always be included. */
-    request->actions = put->actions ? put->actions : &dummy_action;
+    request->actions = (put->actions
+                        ? put->actions
+                        : CONST_CAST(struct nlattr *, &dummy_action));
     request->actions_len = put->actions_len;
     if (put->flags & DPIF_FP_ZERO_STATS) {
         request->clear = true;
@@ -835,7 +918,7 @@ static void
 dpif_linux_init_flow_del(struct dpif *dpif_, const struct dpif_flow_del *del,
                          struct dpif_linux_flow *request)
 {
-    struct dpif_linux *dpif = dpif_linux_cast(dpif_);
+    const struct dpif_linux *dpif = dpif_linux_cast(dpif_);
 
     dpif_linux_flow_init(request);
     request->cmd = OVS_FLOW_CMD_DEL;
@@ -872,7 +955,7 @@ struct dpif_linux_flow_state {
 static int
 dpif_linux_flow_dump_start(const struct dpif *dpif_, void **statep)
 {
-    struct dpif_linux *dpif = dpif_linux_cast(dpif_);
+    const struct dpif_linux *dpif = dpif_linux_cast(dpif_);
     struct dpif_linux_flow_state *state;
     struct dpif_linux_flow request;
     struct ofpbuf *buf;
@@ -885,7 +968,7 @@ dpif_linux_flow_dump_start(const struct dpif *dpif_, void **statep)
 
     buf = ofpbuf_new(1024);
     dpif_linux_flow_to_ofpbuf(&request, buf);
-    nl_dump_start(&state->dump, genl_sock, buf);
+    nl_dump_start(&state->dump, NETLINK_GENERIC, buf);
     ofpbuf_delete(buf);
 
     state->buf = NULL;
@@ -896,6 +979,7 @@ dpif_linux_flow_dump_start(const struct dpif *dpif_, void **statep)
 static int
 dpif_linux_flow_dump_next(const struct dpif *dpif_ OVS_UNUSED, void *state_,
                           const struct nlattr **key, size_t *key_len,
+                          const struct nlattr **mask, size_t *mask_len,
                           const struct nlattr **actions, size_t *actions_len,
                           const struct dpif_flow_stats **stats)
 {
@@ -923,7 +1007,8 @@ dpif_linux_flow_dump_next(const struct dpif *dpif_ OVS_UNUSED, void *state_,
             if (error == ENOENT) {
                 VLOG_DBG("dumped flow disappeared on get");
             } else if (error) {
-                VLOG_WARN("error fetching dumped flow: %s", strerror(error));
+                VLOG_WARN("error fetching dumped flow: %s",
+                          ovs_strerror(error));
             }
         }
     } while (error);
@@ -936,6 +1021,10 @@ dpif_linux_flow_dump_next(const struct dpif *dpif_ OVS_UNUSED, void *state_,
         *key = state->flow.key;
         *key_len = state->flow.key_len;
     }
+    if (mask) {
+        *mask = state->flow.mask;
+        *mask_len = state->flow.mask ? state->flow.mask_len : 0;
+    }
     if (stats) {
         dpif_linux_flow_get_stats(&state->flow, &state->stats);
         *stats = &state->stats;
@@ -986,7 +1075,7 @@ dpif_linux_execute__(int dp_ifindex, const struct dpif_execute *execute)
 
     ofpbuf_use_stub(&request, request_stub, sizeof request_stub);
     dpif_linux_encode_execute(dp_ifindex, execute, &request);
-    error = nl_sock_transact(genl_sock, &request, NULL);
+    error = nl_transact(NETLINK_GENERIC, &request, NULL);
     ofpbuf_uninit(&request);
 
     return error;
@@ -995,7 +1084,7 @@ dpif_linux_execute__(int dp_ifindex, const struct dpif_execute *execute)
 static int
 dpif_linux_execute(struct dpif *dpif_, const struct dpif_execute *execute)
 {
-    struct dpif_linux *dpif = dpif_linux_cast(dpif_);
+    const struct dpif_linux *dpif = dpif_linux_cast(dpif_);
 
     return dpif_linux_execute__(dpif->dp_ifindex, execute);
 }
@@ -1005,7 +1094,7 @@ dpif_linux_execute(struct dpif *dpif_, const struct dpif_execute *execute)
 static void
 dpif_linux_operate__(struct dpif *dpif_, struct dpif_op **ops, size_t n_ops)
 {
-    struct dpif_linux *dpif = dpif_linux_cast(dpif_);
+    const struct dpif_linux *dpif = dpif_linux_cast(dpif_);
 
     struct op_auxdata {
         struct nl_transaction txn;
@@ -1071,7 +1160,7 @@ dpif_linux_operate__(struct dpif *dpif_, struct dpif_op **ops, size_t n_ops)
     for (i = 0; i < n_ops; i++) {
         txnsp[i] = &auxes[i].txn;
     }
-    nl_sock_transact_multiple(genl_sock, txnsp, n_ops);
+    nl_transact_multiple(NETLINK_GENERIC, txnsp, n_ops);
 
     for (i = 0; i < n_ops; i++) {
         struct op_auxdata *aux = &auxes[i];
@@ -1145,7 +1234,7 @@ dpif_linux_operate(struct dpif *dpif, struct dpif_op **ops, size_t n_ops)
 }
 
 static int
-dpif_linux_recv_set(struct dpif *dpif_, bool enable)
+dpif_linux_recv_set__(struct dpif *dpif_, bool enable)
 {
     struct dpif_linux *dpif = dpif_linux_cast(dpif_);
 
@@ -1159,9 +1248,11 @@ dpif_linux_recv_set(struct dpif *dpif_, bool enable)
         struct dpif_port_dump port_dump;
         struct dpif_port port;
 
-        dpif->epoll_fd = epoll_create(10);
         if (dpif->epoll_fd < 0) {
-            return errno;
+            dpif->epoll_fd = epoll_create(10);
+            if (dpif->epoll_fd < 0) {
+                return errno;
+            }
         }
 
         DPIF_PORT_FOR_EACH (&port, &port_dump, &dpif->dpif) {
@@ -1190,7 +1281,7 @@ dpif_linux_recv_set(struct dpif *dpif_, bool enable)
             } else {
                 VLOG_WARN_RL(&error_rl,
                              "%s: failed to set upcall pid on port: %s",
-                             dpif_name(&dpif->dpif), strerror(error));
+                             dpif_name(&dpif->dpif), ovs_strerror(error));
                 nl_sock_destroy(sock);
 
                 if (error == ENODEV || error == ENOENT) {
@@ -1214,6 +1305,19 @@ dpif_linux_recv_set(struct dpif *dpif_, bool enable)
     return 0;
 }
 
+static int
+dpif_linux_recv_set(struct dpif *dpif_, bool enable)
+{
+    struct dpif_linux *dpif = dpif_linux_cast(dpif_);
+    int error;
+
+    ovs_mutex_lock(&dpif->upcall_lock);
+    error = dpif_linux_recv_set__(dpif_, enable);
+    ovs_mutex_unlock(&dpif->upcall_lock);
+
+    return error;
+}
+
 static int
 dpif_linux_queue_to_priority(const struct dpif *dpif OVS_UNUSED,
                              uint32_t queue_id, uint32_t *priority)
@@ -1237,7 +1341,7 @@ parse_odp_packet(struct ofpbuf *buf, struct dpif_upcall *upcall,
         [OVS_PACKET_ATTR_KEY] = { .type = NL_A_NESTED },
 
         /* OVS_PACKET_CMD_ACTION only. */
-        [OVS_PACKET_ATTR_USERDATA] = { .type = NL_A_U64, .optional = true },
+        [OVS_PACKET_ATTR_USERDATA] = { .type = NL_A_UNSPEC, .optional = true },
     };
 
     struct ovs_header *ovs_header;
@@ -1275,17 +1379,15 @@ parse_odp_packet(struct ofpbuf *buf, struct dpif_upcall *upcall,
     upcall->key = CONST_CAST(struct nlattr *,
                              nl_attr_get(a[OVS_PACKET_ATTR_KEY]));
     upcall->key_len = nl_attr_get_size(a[OVS_PACKET_ATTR_KEY]);
-    upcall->userdata = (a[OVS_PACKET_ATTR_USERDATA]
-                        ? nl_attr_get_u64(a[OVS_PACKET_ATTR_USERDATA])
-                        : 0);
+    upcall->userdata = a[OVS_PACKET_ATTR_USERDATA];
     *dp_ifindex = ovs_header->dp_ifindex;
 
     return 0;
 }
 
 static int
-dpif_linux_recv(struct dpif *dpif_, struct dpif_upcall *upcall,
-                struct ofpbuf *buf)
+dpif_linux_recv__(struct dpif *dpif_, struct dpif_upcall *upcall,
+                  struct ofpbuf *buf)
 {
     struct dpif_linux *dpif = dpif_linux_cast(dpif_);
     int read_tries = 0;
@@ -1305,7 +1407,7 @@ dpif_linux_recv(struct dpif *dpif_, struct dpif_upcall *upcall,
         } while (retval < 0 && errno == EINTR);
         if (retval < 0) {
             static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 1);
-            VLOG_WARN_RL(&rl, "epoll_wait failed (%s)", strerror(errno));
+            VLOG_WARN_RL(&rl, "epoll_wait failed (%s)", ovs_strerror(errno));
         } else if (retval > 0) {
             dpif->n_events = retval;
         }
@@ -1355,33 +1457,49 @@ dpif_linux_recv(struct dpif *dpif_, struct dpif_upcall *upcall,
     return EAGAIN;
 }
 
+static int
+dpif_linux_recv(struct dpif *dpif_, struct dpif_upcall *upcall,
+                struct ofpbuf *buf)
+{
+    struct dpif_linux *dpif = dpif_linux_cast(dpif_);
+    int error;
+
+    ovs_mutex_lock(&dpif->upcall_lock);
+    error = dpif_linux_recv__(dpif_, upcall, buf);
+    ovs_mutex_unlock(&dpif->upcall_lock);
+
+    return error;
+}
+
 static void
 dpif_linux_recv_wait(struct dpif *dpif_)
 {
     struct dpif_linux *dpif = dpif_linux_cast(dpif_);
 
-    if (dpif->epoll_fd < 0) {
-       return;
+    ovs_mutex_lock(&dpif->upcall_lock);
+    if (dpif->epoll_fd >= 0) {
+        poll_fd_wait(dpif->epoll_fd, POLLIN);
     }
-
-    poll_fd_wait(dpif->epoll_fd, POLLIN);
+    ovs_mutex_unlock(&dpif->upcall_lock);
 }
 
 static void
 dpif_linux_recv_purge(struct dpif *dpif_)
 {
     struct dpif_linux *dpif = dpif_linux_cast(dpif_);
-    struct dpif_channel *ch;
 
-    if (dpif->epoll_fd < 0) {
-       return;
-    }
+    ovs_mutex_lock(&dpif->upcall_lock);
+    if (dpif->epoll_fd >= 0) {
+        struct dpif_channel *ch;
 
-    for (ch = dpif->channels; ch < &dpif->channels[dpif->uc_array_size]; ch++) {
-        if (ch->sock) {
-            nl_sock_drain(ch->sock);
+        for (ch = dpif->channels; ch < &dpif->channels[dpif->uc_array_size];
+             ch++) {
+            if (ch->sock) {
+                nl_sock_drain(ch->sock);
+            }
         }
     }
+    ovs_mutex_unlock(&dpif->upcall_lock);
 }
 
 const struct dpif_class dpif_linux_class = {
@@ -1391,8 +1509,8 @@ const struct dpif_class dpif_linux_class = {
     dpif_linux_open,
     dpif_linux_close,
     dpif_linux_destroy,
-    dpif_linux_run,
-    dpif_linux_wait,
+    NULL,                       /* run */
+    NULL,                       /* wait */
     dpif_linux_get_stats,
     dpif_linux_port_add,
     dpif_linux_port_del,
@@ -1424,11 +1542,10 @@ const struct dpif_class dpif_linux_class = {
 static int
 dpif_linux_init(void)
 {
-    static int error = -1;
-
-    if (error < 0) {
-        unsigned int ovs_vport_mcgroup;
+    static struct ovsthread_once once = OVSTHREAD_ONCE_INITIALIZER;
+    static int error;
 
+    if (ovsthread_once_start(&once)) {
         error = nl_lookup_genl_family(OVS_DATAPATH_FAMILY,
                                       &ovs_datapath_family);
         if (error) {
@@ -1446,19 +1563,12 @@ dpif_linux_init(void)
             error = nl_lookup_genl_family(OVS_PACKET_FAMILY,
                                           &ovs_packet_family);
         }
-        if (!error) {
-            error = nl_sock_create(NETLINK_GENERIC, &genl_sock);
-        }
         if (!error) {
             error = nl_lookup_genl_mcgroup(OVS_VPORT_FAMILY, OVS_VPORT_MCGROUP,
-                                           &ovs_vport_mcgroup,
-                                           OVS_VPORT_MCGROUP_FALLBACK_ID);
-        }
-        if (!error) {
-            static struct dpif_linux_vport vport;
-            nln = nln_create(NETLINK_GENERIC, ovs_vport_mcgroup,
-                             dpif_linux_nln_parse, &vport);
+                                           &ovs_vport_mcgroup);
         }
+
+        ovsthread_once_done(&once);
     }
 
     return error;
@@ -1476,38 +1586,11 @@ dpif_linux_is_internal_device(const char *name)
         ofpbuf_delete(buf);
     } else if (error != ENODEV && error != ENOENT) {
         VLOG_WARN_RL(&error_rl, "%s: vport query failed (%s)",
-                     name, strerror(error));
+                     name, ovs_strerror(error));
     }
 
     return reply.type == OVS_VPORT_TYPE_INTERNAL;
 }
-
-static bool
-dpif_linux_nln_parse(struct ofpbuf *buf, void *vport_)
-{
-    struct dpif_linux_vport *vport = vport_;
-    return dpif_linux_vport_from_ofpbuf(vport, buf) == 0;
-}
-
-static void
-dpif_linux_port_changed(const void *vport_, void *dpif_)
-{
-    const struct dpif_linux_vport *vport = vport_;
-    struct dpif_linux *dpif = dpif_;
-
-    if (vport) {
-        if (vport->dp_ifindex == dpif->dp_ifindex
-            && (vport->cmd == OVS_VPORT_CMD_NEW
-                || vport->cmd == OVS_VPORT_CMD_DEL
-                || vport->cmd == OVS_VPORT_CMD_SET)) {
-            VLOG_DBG("port_changed: dpif:%s vport:%s cmd:%"PRIu8,
-                     dpif->dpif.full_name, vport->name, vport->cmd);
-            sset_add(&dpif->changed_ports, vport->name);
-        }
-    } else {
-        dpif->change_error = true;
-    }
-}
 
 /* Parses the contents of 'buf', which contains a "struct ovs_header" followed
  * by Netlink attributes, into 'vport'.  Returns 0 if successful, otherwise a
@@ -1550,7 +1633,7 @@ dpif_linux_vport_from_ofpbuf(struct dpif_linux_vport *vport,
 
     vport->cmd = genl->cmd;
     vport->dp_ifindex = ovs_header->dp_ifindex;
-    vport->port_no = nl_attr_get_u32(a[OVS_VPORT_ATTR_PORT_NO]);
+    vport->port_no = nl_attr_get_odp_port(a[OVS_VPORT_ATTR_PORT_NO]);
     vport->type = nl_attr_get_u32(a[OVS_VPORT_ATTR_TYPE]);
     vport->name = nl_attr_get_string(a[OVS_VPORT_ATTR_NAME]);
     if (a[OVS_VPORT_ATTR_UPCALL_PID]) {
@@ -1580,8 +1663,8 @@ dpif_linux_vport_to_ofpbuf(const struct dpif_linux_vport *vport,
     ovs_header = ofpbuf_put_uninit(buf, sizeof *ovs_header);
     ovs_header->dp_ifindex = vport->dp_ifindex;
 
-    if (vport->port_no != UINT32_MAX) {
-        nl_msg_put_u32(buf, OVS_VPORT_ATTR_PORT_NO, vport->port_no);
+    if (vport->port_no != ODPP_NONE) {
+        nl_msg_put_odp_port(buf, OVS_VPORT_ATTR_PORT_NO, vport->port_no);
     }
 
     if (vport->type != OVS_VPORT_TYPE_UNSPEC) {
@@ -1612,7 +1695,7 @@ void
 dpif_linux_vport_init(struct dpif_linux_vport *vport)
 {
     memset(vport, 0, sizeof *vport);
-    vport->port_no = UINT32_MAX;
+    vport->port_no = ODPP_NONE;
 }
 
 /* Executes 'request' in the kernel datapath.  If the command fails, returns a
@@ -1642,7 +1725,7 @@ dpif_linux_vport_transact(const struct dpif_linux_vport *request,
 
     request_buf = ofpbuf_new(1024);
     dpif_linux_vport_to_ofpbuf(request, request_buf);
-    error = nl_sock_transact(genl_sock, request_buf, bufp);
+    error = nl_transact(NETLINK_GENERIC, request_buf, bufp);
     ofpbuf_delete(request_buf);
 
     if (reply) {
@@ -1763,7 +1846,7 @@ dpif_linux_dp_dump_start(struct nl_dump *dump)
 
     buf = ofpbuf_new(1024);
     dpif_linux_dp_to_ofpbuf(&request, buf);
-    nl_dump_start(dump, genl_sock, buf);
+    nl_dump_start(dump, NETLINK_GENERIC, buf);
     ofpbuf_delete(buf);
 }
 
@@ -1784,7 +1867,7 @@ dpif_linux_dp_transact(const struct dpif_linux_dp *request,
 
     request_buf = ofpbuf_new(1024);
     dpif_linux_dp_to_ofpbuf(request, request_buf);
-    error = nl_sock_transact(genl_sock, request_buf, bufp);
+    error = nl_transact(NETLINK_GENERIC, request_buf, bufp);
     ofpbuf_delete(request_buf);
 
     if (reply) {
@@ -1829,6 +1912,7 @@ dpif_linux_flow_from_ofpbuf(struct dpif_linux_flow *flow,
 {
     static const struct nl_policy ovs_flow_policy[] = {
         [OVS_FLOW_ATTR_KEY] = { .type = NL_A_NESTED },
+        [OVS_FLOW_ATTR_MASK] = { .type = NL_A_NESTED, .optional = true },
         [OVS_FLOW_ATTR_ACTIONS] = { .type = NL_A_NESTED, .optional = true },
         [OVS_FLOW_ATTR_STATS] = { NL_POLICY_FOR(struct ovs_flow_stats),
                                   .optional = true },
@@ -1860,6 +1944,11 @@ dpif_linux_flow_from_ofpbuf(struct dpif_linux_flow *flow,
     flow->dp_ifindex = ovs_header->dp_ifindex;
     flow->key = nl_attr_get(a[OVS_FLOW_ATTR_KEY]);
     flow->key_len = nl_attr_get_size(a[OVS_FLOW_ATTR_KEY]);
+
+    if (a[OVS_FLOW_ATTR_MASK]) {
+        flow->mask = nl_attr_get(a[OVS_FLOW_ATTR_MASK]);
+        flow->mask_len = nl_attr_get_size(a[OVS_FLOW_ATTR_MASK]);
+    }
     if (a[OVS_FLOW_ATTR_ACTIONS]) {
         flow->actions = nl_attr_get(a[OVS_FLOW_ATTR_ACTIONS]);
         flow->actions_len = nl_attr_get_size(a[OVS_FLOW_ATTR_ACTIONS]);
@@ -1895,6 +1984,10 @@ dpif_linux_flow_to_ofpbuf(const struct dpif_linux_flow *flow,
         nl_msg_put_unspec(buf, OVS_FLOW_ATTR_KEY, flow->key, flow->key_len);
     }
 
+    if (flow->mask_len) {
+        nl_msg_put_unspec(buf, OVS_FLOW_ATTR_MASK, flow->mask, flow->mask_len);
+    }
+
     if (flow->actions || flow->actions_len) {
         nl_msg_put_unspec(buf, OVS_FLOW_ATTR_ACTIONS,
                           flow->actions, flow->actions_len);
@@ -1938,7 +2031,7 @@ dpif_linux_flow_transact(struct dpif_linux_flow *request,
 
     request_buf = ofpbuf_new(1024);
     dpif_linux_flow_to_ofpbuf(request, request_buf);
-    error = nl_sock_transact(genl_sock, request_buf, bufp);
+    error = nl_transact(NETLINK_GENERIC, request_buf, bufp);
     ofpbuf_delete(request_buf);
 
     if (reply) {