#include "vlog.h"
#define MAX_QUEUE_LENGTH 512
-#define UPCALL_MAX_BATCH 50
+#define UPCALL_MAX_BATCH 64
#define REVALIDATE_MAX_BATCH 50
VLOG_DEFINE_THIS_MODULE(ofproto_dpif_upcall);
-COVERAGE_DEFINE(upcall_duplicate_flow);
+COVERAGE_DEFINE(dumped_duplicate_flow);
+COVERAGE_DEFINE(dumped_new_flow);
+COVERAGE_DEFINE(revalidate_missed_dp_flow);
/* A thread that reads upcalls from dpif, forwards each upcall's packet,
* and possibly sets up a kernel flow as a cache. */
atomic_uint flow_limit; /* Datapath flow hard limit. */
/* n_flows_mutex prevents multiple threads updating these concurrently. */
- atomic_ulong n_flows; /* Number of flows in the datapath. */
+ atomic_uint n_flows; /* Number of flows in the datapath. */
atomic_llong n_flows_timestamp; /* Last time n_flows was updated. */
struct ovs_mutex n_flows_mutex;
};
struct upcall {
- struct ofproto_dpif *ofproto;
+ struct ofproto_dpif *ofproto; /* Parent ofproto. */
- struct flow flow;
- const struct nlattr *key;
- size_t key_len;
- enum dpif_upcall_type upcall_type;
- struct dpif_flow_stats stats;
- odp_port_t odp_in_port;
+ /* The flow and packet are only required to be constant when using
+ * dpif-netdev. If a modification is absolutely necessary, a const cast
+ * may be used with other datapaths. */
+ const struct flow *flow; /* Parsed representation of the packet. */
+ const struct ofpbuf *packet; /* Packet associated with this upcall. */
+ ofp_port_t in_port; /* OpenFlow in port, or OFPP_NONE. */
- uint64_t slow_path_buf[128 / 8];
- struct odputil_keybuf mask_buf;
+ enum dpif_upcall_type type; /* Datapath type of the upcall. */
+ const struct nlattr *userdata; /* Userdata for DPIF_UC_ACTION Upcalls. */
- struct xlate_out xout;
+ bool xout_initialized; /* True if 'xout' must be uninitialized. */
+ struct xlate_out xout; /* Result of xlate_actions(). */
+ struct ofpbuf put_actions; /* Actions 'put' in the fastapath. */
- /* Raw upcall plus data for keeping track of the memory backing it. */
- struct dpif_upcall dpif_upcall; /* As returned by dpif_recv() */
- struct ofpbuf upcall_buf; /* Owns some data in 'dpif_upcall'. */
- uint64_t upcall_stub[512 / 8]; /* Buffer to reduce need for malloc(). */
+ struct dpif_ipfix *ipfix; /* IPFIX pointer or NULL. */
+ struct dpif_sflow *sflow; /* SFlow pointer or NULL. */
+
+ bool vsp_adjusted; /* 'packet' and 'flow' were adjusted for
+ VLAN splinters if true. */
+
+ /* Not used by the upcall callback interface. */
+ const struct nlattr *key; /* Datapath flow key. */
+ size_t key_len; /* Datapath flow key length. */
+ const struct nlattr *out_tun_key; /* Datapath output tunnel key. */
};
/* 'udpif_key's are responsible for tracking the little bit of state udpif
struct xlate_cache *xcache OVS_GUARDED; /* Cache for xlate entries that
* are affected by this ukey.
* Used for stats and learning.*/
- struct odputil_keybuf key_buf; /* Memory for 'key'. */
+ union {
+ struct odputil_keybuf key_buf; /* Memory for 'key'. */
+ struct nlattr key_buf_nla;
+ };
};
static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
static struct list all_udpifs = LIST_INITIALIZER(&all_udpifs);
-static size_t read_upcalls(struct handler *,
- struct upcall upcalls[UPCALL_MAX_BATCH]);
-static void handle_upcalls(struct handler *, struct upcall *, size_t n_upcalls);
+static size_t recv_upcalls(struct handler *);
+static int process_upcall(struct udpif *, struct upcall *,
+ struct ofpbuf *odp_actions);
+static void handle_upcalls(struct udpif *, struct upcall *, size_t n_upcalls);
static void udpif_stop_threads(struct udpif *);
static void udpif_start_threads(struct udpif *, size_t n_handlers,
size_t n_revalidators);
size_t key_len, long long int used,
struct udpif_key **result);
static void ukey_delete(struct revalidator *, struct udpif_key *);
+static enum upcall_type classify_upcall(enum dpif_upcall_type type,
+ const struct nlattr *userdata);
+
+static int upcall_receive(struct upcall *, const struct dpif_backer *,
+ const struct ofpbuf *packet, enum dpif_upcall_type,
+ const struct nlattr *userdata, const struct flow *);
+static void upcall_uninit(struct upcall *);
+
+static upcall_callback upcall_cb;
static atomic_bool enable_megaflows = ATOMIC_VAR_INIT(true);
atomic_init(&udpif->n_flows_timestamp, LLONG_MIN);
ovs_mutex_init(&udpif->n_flows_mutex);
+ dpif_register_upcall_cb(dpif, upcall_cb, udpif);
+
return udpif;
}
xpthread_join(udpif->revalidators[i].thread, NULL);
}
+ dpif_disable_upcall(udpif->dpif);
+
for (i = 0; i < udpif->n_revalidators; i++) {
struct revalidator *revalidator = &udpif->revalidators[i];
"handler", udpif_upcall_handler, handler);
}
+ dpif_enable_upcall(udpif->dpif);
+
ovs_barrier_init(&udpif->reval_barrier, udpif->n_revalidators);
udpif->reval_exit = false;
udpif->revalidators = xzalloc(udpif->n_revalidators
unsigned long flow_count;
now = time_msec();
- atomic_read(&udpif->n_flows_timestamp, &time);
+ atomic_read_relaxed(&udpif->n_flows_timestamp, &time);
if (time < now - 100 && !ovs_mutex_trylock(&udpif->n_flows_mutex)) {
struct dpif_dp_stats stats;
- atomic_store(&udpif->n_flows_timestamp, now);
+ atomic_store_relaxed(&udpif->n_flows_timestamp, now);
dpif_get_dp_stats(udpif->dpif, &stats);
flow_count = stats.n_flows;
- atomic_store(&udpif->n_flows, flow_count);
+ atomic_store_relaxed(&udpif->n_flows, flow_count);
ovs_mutex_unlock(&udpif->n_flows_mutex);
} else {
- atomic_read(&udpif->n_flows, &flow_count);
+ atomic_read_relaxed(&udpif->n_flows, &flow_count);
}
return flow_count;
}
struct udpif *udpif = handler->udpif;
while (!latch_is_set(&handler->udpif->exit_latch)) {
- struct upcall upcalls[UPCALL_MAX_BATCH];
- size_t n_upcalls, i;
-
- n_upcalls = read_upcalls(handler, upcalls);
- if (!n_upcalls) {
+ if (!recv_upcalls(handler)) {
dpif_recv_wait(udpif->dpif, handler->handler_id);
latch_wait(&udpif->exit_latch);
poll_block();
- } else {
- handle_upcalls(handler, upcalls, n_upcalls);
-
- for (i = 0; i < n_upcalls; i++) {
- xlate_out_uninit(&upcalls[i].xout);
- ofpbuf_uninit(&upcalls[i].dpif_upcall.packet);
- ofpbuf_uninit(&upcalls[i].upcall_buf);
- }
}
coverage_clear();
}
return NULL;
}
+static size_t
+recv_upcalls(struct handler *handler)
+{
+ struct udpif *udpif = handler->udpif;
+ uint64_t recv_stubs[UPCALL_MAX_BATCH][512 / 8];
+ struct ofpbuf recv_bufs[UPCALL_MAX_BATCH];
+ struct dpif_upcall dupcalls[UPCALL_MAX_BATCH];
+ struct upcall upcalls[UPCALL_MAX_BATCH];
+ size_t n_upcalls, i;
+
+ n_upcalls = 0;
+ while (n_upcalls < UPCALL_MAX_BATCH) {
+ struct ofpbuf *recv_buf = &recv_bufs[n_upcalls];
+ struct dpif_upcall *dupcall = &dupcalls[n_upcalls];
+ struct upcall *upcall = &upcalls[n_upcalls];
+ struct pkt_metadata md;
+ struct flow flow;
+ int error;
+
+ ofpbuf_use_stub(recv_buf, recv_stubs[n_upcalls],
+ sizeof recv_stubs[n_upcalls]);
+ if (dpif_recv(udpif->dpif, handler->handler_id, dupcall, recv_buf)) {
+ ofpbuf_uninit(recv_buf);
+ break;
+ }
+
+ if (odp_flow_key_to_flow(dupcall->key, dupcall->key_len, &flow)
+ == ODP_FIT_ERROR) {
+ goto free_dupcall;
+ }
+
+ error = upcall_receive(upcall, udpif->backer, &dupcall->packet,
+ dupcall->type, dupcall->userdata, &flow);
+ if (error) {
+ if (error == ENODEV) {
+ /* Received packet on datapath port for which we couldn't
+ * associate an ofproto. This can happen if a port is removed
+ * while traffic is being received. Print a rate-limited
+ * message in case it happens frequently. */
+ dpif_flow_put(udpif->dpif, DPIF_FP_CREATE, dupcall->key,
+ dupcall->key_len, NULL, 0, NULL, 0, NULL);
+ VLOG_INFO_RL(&rl, "received packet on unassociated datapath "
+ "port %"PRIu32, flow.in_port.odp_port);
+ }
+ goto free_dupcall;
+ }
+
+ upcall->key = dupcall->key;
+ upcall->key_len = dupcall->key_len;
+
+ upcall->out_tun_key = dupcall->out_tun_key;
+
+ if (vsp_adjust_flow(upcall->ofproto, &flow, &dupcall->packet)) {
+ upcall->vsp_adjusted = true;
+ }
+
+ md = pkt_metadata_from_flow(&flow);
+ flow_extract(&dupcall->packet, &md, &flow);
+
+ error = process_upcall(udpif, upcall, NULL);
+ if (error) {
+ goto cleanup;
+ }
+
+ n_upcalls++;
+ continue;
+
+cleanup:
+ upcall_uninit(upcall);
+free_dupcall:
+ ofpbuf_uninit(&dupcall->packet);
+ ofpbuf_uninit(recv_buf);
+ }
+
+ if (n_upcalls) {
+ handle_upcalls(handler->udpif, upcalls, n_upcalls);
+ for (i = 0; i < n_upcalls; i++) {
+ ofpbuf_uninit(&dupcalls[i].packet);
+ ofpbuf_uninit(&recv_bufs[i]);
+ upcall_uninit(&upcalls[i]);
+ }
+ }
+
+ return n_upcalls;
+}
+
static void *
udpif_revalidator(void *arg)
{
/* Used only by the leader. */
long long int start_time = 0;
uint64_t last_reval_seq = 0;
- unsigned int flow_limit = 0;
size_t n_flows = 0;
revalidator->id = ovsthread_id_self();
ovs_barrier_block(&udpif->reval_barrier);
if (leader) {
+ unsigned int flow_limit;
long long int duration;
+ atomic_read_relaxed(&udpif->flow_limit, &flow_limit);
+
dpif_flow_dump_destroy(udpif->dump);
seq_change(udpif->dump_seq);
duration = MAX(time_msec() - start_time, 1);
- atomic_read(&udpif->flow_limit, &flow_limit);
udpif->dump_duration = duration;
if (duration > 2000) {
flow_limit /= duration / 1000;
flow_limit += 1000;
}
flow_limit = MIN(ofproto_flow_limit, MAX(flow_limit, 1000));
- atomic_store(&udpif->flow_limit, flow_limit);
+ atomic_store_relaxed(&udpif->flow_limit, flow_limit);
if (duration > 2000) {
VLOG_INFO("Spent an unreasonably long %lldms dumping flows",
}
\f
static enum upcall_type
-classify_upcall(const struct upcall *upcall)
+classify_upcall(enum dpif_upcall_type type, const struct nlattr *userdata)
{
- const struct dpif_upcall *dpif_upcall = &upcall->dpif_upcall;
union user_action_cookie cookie;
size_t userdata_len;
/* First look at the upcall type. */
- switch (dpif_upcall->type) {
+ switch (type) {
case DPIF_UC_ACTION:
break;
case DPIF_N_UC_TYPES:
default:
- VLOG_WARN_RL(&rl, "upcall has unexpected type %"PRIu32,
- dpif_upcall->type);
+ VLOG_WARN_RL(&rl, "upcall has unexpected type %"PRIu32, type);
return BAD_UPCALL;
}
/* "action" upcalls need a closer look. */
- if (!dpif_upcall->userdata) {
+ if (!userdata) {
VLOG_WARN_RL(&rl, "action upcall missing cookie");
return BAD_UPCALL;
}
- userdata_len = nl_attr_get_size(dpif_upcall->userdata);
+ userdata_len = nl_attr_get_size(userdata);
if (userdata_len < sizeof cookie.type
|| userdata_len > sizeof cookie) {
VLOG_WARN_RL(&rl, "action upcall cookie has unexpected size %"PRIuSIZE,
return BAD_UPCALL;
}
memset(&cookie, 0, sizeof cookie);
- memcpy(&cookie, nl_attr_get(dpif_upcall->userdata), userdata_len);
+ memcpy(&cookie, nl_attr_get(userdata), userdata_len);
if (userdata_len == MAX(8, sizeof cookie.sflow)
&& cookie.type == USER_ACTION_COOKIE_SFLOW) {
return SFLOW_UPCALL;
* initialized with at least 128 bytes of space. */
static void
compose_slow_path(struct udpif *udpif, struct xlate_out *xout,
- struct flow *flow, odp_port_t odp_in_port,
+ const struct flow *flow, odp_port_t odp_in_port,
struct ofpbuf *buf)
{
union user_action_cookie cookie;
? ODPP_NONE
: odp_in_port;
pid = dpif_port_get_pid(udpif->dpif, port, flow_hash_5tuple(flow, 0));
- odp_put_userspace_action(pid, &cookie, sizeof cookie.slow_path, buf);
+ odp_put_userspace_action(pid, &cookie, sizeof cookie.slow_path, ODPP_NONE,
+ buf);
+}
+
+/* If there is no error, the upcall must be destroyed with upcall_uninit()
+ * before quiescing, as the referred objects are guaranteed to exist only
+ * until the calling thread quiesces. Otherwise, do not call upcall_uninit()
+ * since the 'upcall->put_actions' remains uninitialized. */
+static int
+upcall_receive(struct upcall *upcall, const struct dpif_backer *backer,
+ const struct ofpbuf *packet, enum dpif_upcall_type type,
+ const struct nlattr *userdata, const struct flow *flow)
+{
+ int error;
+
+ error = xlate_lookup(backer, flow, &upcall->ofproto, &upcall->ipfix,
+ &upcall->sflow, NULL, &upcall->in_port);
+ if (error) {
+ return error;
+ }
+
+ upcall->flow = flow;
+ upcall->packet = packet;
+ upcall->type = type;
+ upcall->userdata = userdata;
+ ofpbuf_init(&upcall->put_actions, 0);
+
+ upcall->xout_initialized = false;
+ upcall->vsp_adjusted = false;
+
+ upcall->key = NULL;
+ upcall->key_len = 0;
+
+ upcall->out_tun_key = NULL;
+
+ return 0;
}
static void
-upcall_init(struct upcall *upcall, struct flow *flow, struct ofpbuf *packet,
- struct ofproto_dpif *ofproto, struct dpif_upcall *dupcall,
- odp_port_t odp_in_port)
+upcall_xlate(struct udpif *udpif, struct upcall *upcall,
+ struct ofpbuf *odp_actions)
{
- struct pkt_metadata md = pkt_metadata_from_flow(flow);
+ struct dpif_flow_stats stats;
struct xlate_in xin;
- flow_extract(packet, &md, &upcall->flow);
-
- upcall->ofproto = ofproto;
- upcall->key = dupcall->key;
- upcall->key_len = dupcall->key_len;
- upcall->upcall_type = dupcall->type;
- upcall->stats.n_packets = 1;
- upcall->stats.n_bytes = ofpbuf_size(packet);
- upcall->stats.used = time_msec();
- upcall->stats.tcp_flags = ntohs(upcall->flow.tcp_flags);
- upcall->odp_in_port = odp_in_port;
+ stats.n_packets = 1;
+ stats.n_bytes = ofpbuf_size(upcall->packet);
+ stats.used = time_msec();
+ stats.tcp_flags = ntohs(upcall->flow->tcp_flags);
- xlate_in_init(&xin, upcall->ofproto, &upcall->flow, NULL,
- upcall->stats.tcp_flags, packet);
+ xlate_in_init(&xin, upcall->ofproto, upcall->flow, upcall->in_port, NULL,
+ stats.tcp_flags, upcall->packet);
+ xin.odp_actions = odp_actions;
- if (upcall->upcall_type == DPIF_UC_MISS) {
- xin.resubmit_stats = &upcall->stats;
+ if (upcall->type == DPIF_UC_MISS) {
+ xin.resubmit_stats = &stats;
} else {
/* For non-miss upcalls, there's a flow in the datapath which this
* packet was accounted to. Presumably the revalidators will deal
}
xlate_actions(&xin, &upcall->xout);
+ upcall->xout_initialized = true;
+
+ /* Special case for fail-open mode.
+ *
+ * If we are in fail-open mode, but we are connected to a controller too,
+ * then we should send the packet up to the controller in the hope that it
+ * will try to set up a flow and thereby allow us to exit fail-open.
+ *
+ * See the top-level comment in fail-open.c for more information.
+ *
+ * Copy packets before they are modified by execution. */
+ if (upcall->xout.fail_open) {
+ const struct ofpbuf *packet = upcall->packet;
+ struct ofproto_packet_in *pin;
+
+ pin = xmalloc(sizeof *pin);
+ pin->up.packet = xmemdup(ofpbuf_data(packet), ofpbuf_size(packet));
+ pin->up.packet_len = ofpbuf_size(packet);
+ pin->up.reason = OFPR_NO_MATCH;
+ pin->up.table_id = 0;
+ pin->up.cookie = OVS_BE64_MAX;
+ flow_get_metadata(upcall->flow, &pin->up.fmd);
+ pin->send_len = 0; /* Not used for flow table misses. */
+ pin->miss_type = OFPROTO_PACKET_IN_NO_MISS;
+ ofproto_dpif_send_packet_in(upcall->ofproto, pin);
+ }
+
+ if (!upcall->xout.slow) {
+ ofpbuf_use_const(&upcall->put_actions,
+ ofpbuf_data(upcall->xout.odp_actions),
+ ofpbuf_size(upcall->xout.odp_actions));
+ } else {
+ ofpbuf_init(&upcall->put_actions, 0);
+ compose_slow_path(udpif, &upcall->xout, upcall->flow,
+ upcall->flow->in_port.odp_port,
+ &upcall->put_actions);
+ }
}
-/* Reads and classifies upcalls. Returns the number of upcalls successfully
- * read. */
-static size_t
-read_upcalls(struct handler *handler,
- struct upcall upcalls[UPCALL_MAX_BATCH])
+static void
+upcall_uninit(struct upcall *upcall)
{
- struct udpif *udpif = handler->udpif;
- size_t i;
- size_t n_upcalls = 0;
+ if (upcall) {
+ if (upcall->xout_initialized) {
+ xlate_out_uninit(&upcall->xout);
+ }
+ ofpbuf_uninit(&upcall->put_actions);
+ }
+}
- /* Try reading UPCALL_MAX_BATCH upcalls from dpif. */
- for (i = 0; i < UPCALL_MAX_BATCH; i++) {
- struct upcall *upcall = &upcalls[n_upcalls];
- struct dpif_upcall *dupcall;
- struct ofpbuf *packet;
- struct ofproto_dpif *ofproto;
- struct dpif_sflow *sflow;
- struct dpif_ipfix *ipfix;
- struct flow flow;
- enum upcall_type type;
- odp_port_t odp_in_port;
- int error;
+static int
+upcall_cb(const struct ofpbuf *packet, const struct flow *flow,
+ enum dpif_upcall_type type, const struct nlattr *userdata,
+ struct ofpbuf *actions, struct flow_wildcards *wc,
+ struct ofpbuf *put_actions, void *aux)
+{
+ struct udpif *udpif = aux;
+ unsigned int flow_limit;
+ struct upcall upcall;
+ bool megaflow;
+ int error;
- ofpbuf_use_stub(&upcall->upcall_buf, upcall->upcall_stub,
- sizeof upcall->upcall_stub);
- error = dpif_recv(udpif->dpif, handler->handler_id,
- &upcall->dpif_upcall, &upcall->upcall_buf);
- if (error) {
- ofpbuf_uninit(&upcall->upcall_buf);
- break;
- }
+ atomic_read_relaxed(&enable_megaflows, &megaflow);
+ atomic_read_relaxed(&udpif->flow_limit, &flow_limit);
- dupcall = &upcall->dpif_upcall;
- packet = &dupcall->packet;
- error = xlate_receive(udpif->backer, packet, dupcall->key,
- dupcall->key_len, &flow,
- &ofproto, &ipfix, &sflow, NULL, &odp_in_port);
- if (error) {
- if (error == ENODEV) {
- /* Received packet on datapath port for which we couldn't
- * associate an ofproto. This can happen if a port is removed
- * while traffic is being received. Print a rate-limited
- * message in case it happens frequently. Install a drop flow
- * so that future packets of the flow are inexpensively dropped
- * in the kernel. */
- VLOG_INFO_RL(&rl, "received packet on unassociated datapath "
- "port %"PRIu32, odp_in_port);
- dpif_flow_put(udpif->dpif, DPIF_FP_CREATE | DPIF_FP_MODIFY,
- dupcall->key, dupcall->key_len, NULL, 0, NULL, 0,
- NULL);
- }
- goto destroy_upcall;
+ error = upcall_receive(&upcall, udpif->backer, packet, type, userdata,
+ flow);
+ if (error) {
+ return error;
+ }
+
+ error = process_upcall(udpif, &upcall, actions);
+ if (error) {
+ goto out;
+ }
+
+ if (upcall.xout.slow && put_actions) {
+ ofpbuf_put(put_actions, ofpbuf_data(&upcall.put_actions),
+ ofpbuf_size(&upcall.put_actions));
+ }
+
+ if (OVS_LIKELY(wc)) {
+ if (megaflow) {
+ /* XXX: This could be avoided with sufficient API changes. */
+ *wc = upcall.xout.wc;
+ } else {
+ flow_wildcards_init_for_packet(wc, flow);
}
+ }
+
+ if (udpif_get_n_flows(udpif) >= flow_limit) {
+ error = ENOSPC;
+ }
+
+out:
+ upcall_uninit(&upcall);
+ return error;
+}
- type = classify_upcall(upcall);
- if (type == MISS_UPCALL) {
- upcall_init(upcall, &flow, packet, ofproto, dupcall, odp_in_port);
- n_upcalls++;
- continue;
+static int
+process_upcall(struct udpif *udpif, struct upcall *upcall,
+ struct ofpbuf *odp_actions)
+{
+ const struct nlattr *userdata = upcall->userdata;
+ const struct ofpbuf *packet = upcall->packet;
+ const struct flow *flow = upcall->flow;
+
+ switch (classify_upcall(upcall->type, userdata)) {
+ case MISS_UPCALL:
+ upcall_xlate(udpif, upcall, odp_actions);
+ return 0;
+
+ case SFLOW_UPCALL:
+ if (upcall->sflow) {
+ union user_action_cookie cookie;
+
+ memset(&cookie, 0, sizeof cookie);
+ memcpy(&cookie, nl_attr_get(userdata), sizeof cookie.sflow);
+ dpif_sflow_received(upcall->sflow, packet, flow,
+ flow->in_port.odp_port, &cookie);
}
+ break;
- switch (type) {
- case SFLOW_UPCALL:
- if (sflow) {
- union user_action_cookie cookie;
+ case IPFIX_UPCALL:
+ if (upcall->ipfix) {
+ union user_action_cookie cookie;
+ struct flow_tnl output_tunnel_key;
- memset(&cookie, 0, sizeof cookie);
- memcpy(&cookie, nl_attr_get(dupcall->userdata),
- sizeof cookie.sflow);
- dpif_sflow_received(sflow, packet, &flow, odp_in_port,
- &cookie);
- }
- break;
- case IPFIX_UPCALL:
- if (ipfix) {
- dpif_ipfix_bridge_sample(ipfix, packet, &flow);
- }
- break;
- case FLOW_SAMPLE_UPCALL:
- if (ipfix) {
- union user_action_cookie cookie;
-
- memset(&cookie, 0, sizeof cookie);
- memcpy(&cookie, nl_attr_get(dupcall->userdata),
- sizeof cookie.flow_sample);
-
- /* The flow reflects exactly the contents of the packet.
- * Sample the packet using it. */
- dpif_ipfix_flow_sample(ipfix, packet, &flow,
- cookie.flow_sample.collector_set_id,
- cookie.flow_sample.probability,
- cookie.flow_sample.obs_domain_id,
- cookie.flow_sample.obs_point_id);
+ memset(&cookie, 0, sizeof cookie);
+ memcpy(&cookie, nl_attr_get(userdata), sizeof cookie.ipfix);
+
+ if (upcall->out_tun_key) {
+ memset(&output_tunnel_key, 0, sizeof output_tunnel_key);
+ odp_tun_key_from_attr(upcall->out_tun_key,
+ &output_tunnel_key);
}
- break;
- case BAD_UPCALL:
- break;
- case MISS_UPCALL:
- OVS_NOT_REACHED();
+ dpif_ipfix_bridge_sample(upcall->ipfix, packet, flow,
+ flow->in_port.odp_port,
+ cookie.ipfix.output_odp_port,
+ upcall->out_tun_key ?
+ &output_tunnel_key : NULL);
}
+ break;
+
+ case FLOW_SAMPLE_UPCALL:
+ if (upcall->ipfix) {
+ union user_action_cookie cookie;
+
+ memset(&cookie, 0, sizeof cookie);
+ memcpy(&cookie, nl_attr_get(userdata), sizeof cookie.flow_sample);
- dpif_ipfix_unref(ipfix);
- dpif_sflow_unref(sflow);
+ /* The flow reflects exactly the contents of the packet.
+ * Sample the packet using it. */
+ dpif_ipfix_flow_sample(upcall->ipfix, packet, flow,
+ cookie.flow_sample.collector_set_id,
+ cookie.flow_sample.probability,
+ cookie.flow_sample.obs_domain_id,
+ cookie.flow_sample.obs_point_id);
+ }
+ break;
-destroy_upcall:
- ofpbuf_uninit(&upcall->dpif_upcall.packet);
- ofpbuf_uninit(&upcall->upcall_buf);
+ case BAD_UPCALL:
+ break;
}
- return n_upcalls;
+ return EAGAIN;
}
static void
-handle_upcalls(struct handler *handler, struct upcall *upcalls,
+handle_upcalls(struct udpif *udpif, struct upcall *upcalls,
size_t n_upcalls)
{
- struct udpif *udpif = handler->udpif;
+ struct odputil_keybuf mask_bufs[UPCALL_MAX_BATCH];
struct dpif_op *opsp[UPCALL_MAX_BATCH * 2];
struct dpif_op ops[UPCALL_MAX_BATCH * 2];
- size_t n_ops, i;
unsigned int flow_limit;
- bool fail_open, may_put;
+ size_t n_ops, i;
+ bool may_put;
+ bool megaflow;
+
+ atomic_read_relaxed(&udpif->flow_limit, &flow_limit);
+ atomic_read_relaxed(&enable_megaflows, &megaflow);
- atomic_read(&udpif->flow_limit, &flow_limit);
may_put = udpif_get_n_flows(udpif) < flow_limit;
/* Handle the packets individually in order of arrival.
*
* The loop fills 'ops' with an array of operations to execute in the
* datapath. */
- fail_open = false;
n_ops = 0;
for (i = 0; i < n_upcalls; i++) {
struct upcall *upcall = &upcalls[i];
- struct ofpbuf *packet = &upcall->dpif_upcall.packet;
+ const struct ofpbuf *packet = upcall->packet;
struct dpif_op *op;
- fail_open = fail_open || upcall->xout.fail_open;
-
- if (upcall->flow.in_port.ofp_port
- != vsp_realdev_to_vlandev(upcall->ofproto,
- upcall->flow.in_port.ofp_port,
- upcall->flow.vlan_tci)) {
- /* This packet was received on a VLAN splinter port. We
- * added a VLAN to the packet to make the packet resemble
- * the flow, but the actions were composed assuming that
- * the packet contained no VLAN. So, we must remove the
- * VLAN header from the packet before trying to execute the
- * actions. */
- if (ofpbuf_size(&upcall->xout.odp_actions)) {
- eth_pop_vlan(packet);
+ if (upcall->vsp_adjusted) {
+ /* This packet was received on a VLAN splinter port. We added a
+ * VLAN to the packet to make the packet resemble the flow, but the
+ * actions were composed assuming that the packet contained no
+ * VLAN. So, we must remove the VLAN header from the packet before
+ * trying to execute the actions. */
+ if (ofpbuf_size(upcall->xout.odp_actions)) {
+ eth_pop_vlan(CONST_CAST(struct ofpbuf *, upcall->packet));
}
/* Remove the flow vlan tags inserted by vlan splinter logic
* to ensure megaflow masks generated match the data path flow. */
- upcall->flow.vlan_tci = 0;
+ CONST_CAST(struct flow *, upcall->flow)->vlan_tci = 0;
}
/* Do not install a flow into the datapath if:
*
* - We received this packet via some flow installed in the kernel
* already. */
- if (may_put
- && upcall->dpif_upcall.type == DPIF_UC_MISS) {
+ if (may_put && upcall->type == DPIF_UC_MISS) {
struct ofpbuf mask;
- bool megaflow;
- atomic_read(&enable_megaflows, &megaflow);
- ofpbuf_use_stack(&mask, &upcall->mask_buf, sizeof upcall->mask_buf);
+ ofpbuf_use_stack(&mask, &mask_bufs[i], sizeof mask_bufs[i]);
+
if (megaflow) {
size_t max_mpls;
bool recirc;
recirc = ofproto_dpif_get_enable_recirc(upcall->ofproto);
max_mpls = ofproto_dpif_get_max_mpls_depth(upcall->ofproto);
odp_flow_key_from_mask(&mask, &upcall->xout.wc.masks,
- &upcall->flow, UINT32_MAX, max_mpls,
+ upcall->flow, UINT32_MAX, max_mpls,
recirc);
}
op = &ops[n_ops++];
op->type = DPIF_OP_FLOW_PUT;
- op->u.flow_put.flags = DPIF_FP_CREATE | DPIF_FP_MODIFY;
+ op->u.flow_put.flags = DPIF_FP_CREATE;
op->u.flow_put.key = upcall->key;
op->u.flow_put.key_len = upcall->key_len;
op->u.flow_put.mask = ofpbuf_data(&mask);
op->u.flow_put.mask_len = ofpbuf_size(&mask);
op->u.flow_put.stats = NULL;
-
- if (!upcall->xout.slow) {
- op->u.flow_put.actions = ofpbuf_data(&upcall->xout.odp_actions);
- op->u.flow_put.actions_len = ofpbuf_size(&upcall->xout.odp_actions);
- } else {
- struct ofpbuf buf;
-
- ofpbuf_use_stack(&buf, upcall->slow_path_buf,
- sizeof upcall->slow_path_buf);
- compose_slow_path(udpif, &upcall->xout, &upcall->flow,
- upcall->odp_in_port, &buf);
- op->u.flow_put.actions = ofpbuf_data(&buf);
- op->u.flow_put.actions_len = ofpbuf_size(&buf);
- }
+ op->u.flow_put.actions = ofpbuf_data(&upcall->put_actions);
+ op->u.flow_put.actions_len = ofpbuf_size(&upcall->put_actions);
}
- if (ofpbuf_size(&upcall->xout.odp_actions)) {
-
+ if (ofpbuf_size(upcall->xout.odp_actions)) {
op = &ops[n_ops++];
op->type = DPIF_OP_EXECUTE;
- op->u.execute.packet = packet;
+ op->u.execute.packet = CONST_CAST(struct ofpbuf *, packet);
odp_key_to_pkt_metadata(upcall->key, upcall->key_len,
&op->u.execute.md);
- op->u.execute.actions = ofpbuf_data(&upcall->xout.odp_actions);
- op->u.execute.actions_len = ofpbuf_size(&upcall->xout.odp_actions);
+ op->u.execute.actions = ofpbuf_data(upcall->xout.odp_actions);
+ op->u.execute.actions_len = ofpbuf_size(upcall->xout.odp_actions);
op->u.execute.needs_help = (upcall->xout.slow & SLOW_ACTION) != 0;
- }
- }
-
- /* Special case for fail-open mode.
- *
- * If we are in fail-open mode, but we are connected to a controller too,
- * then we should send the packet up to the controller in the hope that it
- * will try to set up a flow and thereby allow us to exit fail-open.
- *
- * See the top-level comment in fail-open.c for more information.
- *
- * Copy packets before they are modified by execution. */
- if (fail_open) {
- for (i = 0; i < n_upcalls; i++) {
- struct upcall *upcall = &upcalls[i];
- struct ofpbuf *packet = &upcall->dpif_upcall.packet;
- struct ofproto_packet_in *pin;
-
- pin = xmalloc(sizeof *pin);
- pin->up.packet = xmemdup(ofpbuf_data(packet), ofpbuf_size(packet));
- pin->up.packet_len = ofpbuf_size(packet);
- pin->up.reason = OFPR_NO_MATCH;
- pin->up.table_id = 0;
- pin->up.cookie = OVS_BE64_MAX;
- flow_get_metadata(&upcall->flow, &pin->up.fmd);
- pin->send_len = 0; /* Not used for flow table misses. */
- pin->miss_type = OFPROTO_PACKET_IN_NO_MISS;
- ofproto_dpif_send_packet_in(upcall->ofproto, pin);
+ op->u.execute.probe = false;
}
}
struct udpif_key *ukey = xmalloc(sizeof *ukey);
ovs_mutex_init(&ukey->mutex);
- ukey->key = (struct nlattr *) &ukey->key_buf;
+ ukey->key = &ukey->key_buf_nla;
memcpy(&ukey->key_buf, key, key_len);
ukey->key_len = key_len;
}
static bool
-should_revalidate(uint64_t packets, long long int used)
+should_revalidate(const struct udpif *udpif, uint64_t packets,
+ long long int used)
{
long long int metric, now, duration;
+ if (udpif->dump_duration < 200) {
+ /* We are likely to handle full revalidation for the flows. */
+ return true;
+ }
+
/* Calculate the mean time between seeing these packets. If this
* exceeds the threshold, then delete the flow rather than performing
* costly revalidation for flows that aren't being hit frequently.
duration = now - used;
metric = duration / packets;
- if (metric > 200) {
- return false;
+ if (metric < 200) {
+ /* The flow is receiving more than ~5pps, so keep it. */
+ return true;
}
- return true;
+ return false;
}
static bool
struct ofpbuf xout_actions;
struct flow flow, dp_mask;
uint32_t *dp32, *xout32;
- odp_port_t odp_in_port;
+ ofp_port_t ofp_in_port;
struct xlate_in xin;
long long int last_used;
int error;
size_t i;
- bool may_learn, ok;
+ bool ok;
ok = false;
xoutp = NULL;
: 0);
if (udpif->need_revalidate && last_used
- && !should_revalidate(push.n_packets, last_used)) {
+ && !should_revalidate(udpif, push.n_packets, last_used)) {
ok = false;
goto exit;
}
goto exit;
}
- may_learn = push.n_packets > 0;
if (ukey->xcache && !udpif->need_revalidate) {
- xlate_push_stats(ukey->xcache, may_learn, &push);
+ xlate_push_stats(ukey->xcache, &push);
ok = true;
goto exit;
}
- error = xlate_receive(udpif->backer, NULL, ukey->key, ukey->key_len, &flow,
- &ofproto, NULL, NULL, &netflow, &odp_in_port);
+ if (odp_flow_key_to_flow(ukey->key, ukey->key_len, &flow)
+ == ODP_FIT_ERROR) {
+ goto exit;
+ }
+
+ error = xlate_lookup(udpif->backer, &flow, &ofproto, NULL, NULL, &netflow,
+ &ofp_in_port);
if (error) {
goto exit;
}
ukey->xcache = xlate_cache_new();
}
- xlate_in_init(&xin, ofproto, &flow, NULL, push.tcp_flags, NULL);
- xin.resubmit_stats = push.n_packets ? &push : NULL;
+ xlate_in_init(&xin, ofproto, &flow, ofp_in_port, NULL, push.tcp_flags,
+ NULL);
+ if (push.n_packets) {
+ xin.resubmit_stats = &push;
+ xin.may_learn = true;
+ }
xin.xcache = ukey->xcache;
- xin.may_learn = may_learn;
xin.skip_wildcards = !udpif->need_revalidate;
xlate_actions(&xin, &xout);
xoutp = &xout;
}
if (!xout.slow) {
- ofpbuf_use_const(&xout_actions, ofpbuf_data(&xout.odp_actions),
- ofpbuf_size(&xout.odp_actions));
+ ofpbuf_use_const(&xout_actions, ofpbuf_data(xout.odp_actions),
+ ofpbuf_size(xout.odp_actions));
} else {
ofpbuf_use_stack(&xout_actions, slow_path_buf, sizeof slow_path_buf);
- compose_slow_path(udpif, &xout, &flow, odp_in_port, &xout_actions);
+ compose_slow_path(udpif, &xout, &flow, flow.in_port.odp_port,
+ &xout_actions);
}
if (f->actions_len != ofpbuf_size(&xout_actions)
ok = true;
exit:
- if (netflow) {
- if (!ok) {
- netflow_flow_clear(netflow, &flow);
- }
- netflow_unref(netflow);
+ if (netflow && !ok) {
+ netflow_flow_clear(netflow, &flow);
}
xlate_out_uninit(xoutp);
return ok;
struct dpif_flow_stats *push, *stats, push_buf;
stats = op->op.u.flow_del.stats;
- if (op->ukey) {
- push = &push_buf;
- ovs_mutex_lock(&op->ukey->mutex);
- push->used = MAX(stats->used, op->ukey->stats.used);
- push->tcp_flags = stats->tcp_flags | op->ukey->stats.tcp_flags;
- push->n_packets = stats->n_packets - op->ukey->stats.n_packets;
- push->n_bytes = stats->n_bytes - op->ukey->stats.n_bytes;
- ovs_mutex_unlock(&op->ukey->mutex);
- } else {
- push = stats;
- }
+ push = &push_buf;
+
+ ovs_mutex_lock(&op->ukey->mutex);
+ push->used = MAX(stats->used, op->ukey->stats.used);
+ push->tcp_flags = stats->tcp_flags | op->ukey->stats.tcp_flags;
+ push->n_packets = stats->n_packets - op->ukey->stats.n_packets;
+ push->n_bytes = stats->n_bytes - op->ukey->stats.n_bytes;
+ ovs_mutex_unlock(&op->ukey->mutex);
if (push->n_packets || netflow_exists()) {
struct ofproto_dpif *ofproto;
struct netflow *netflow;
+ ofp_port_t ofp_in_port;
struct flow flow;
- bool may_learn;
-
- may_learn = push->n_packets > 0;
- if (op->ukey) {
- ovs_mutex_lock(&op->ukey->mutex);
- if (op->ukey->xcache) {
- xlate_push_stats(op->ukey->xcache, may_learn, push);
- ovs_mutex_unlock(&op->ukey->mutex);
- continue;
- }
+ int error;
+
+ ovs_mutex_lock(&op->ukey->mutex);
+ if (op->ukey->xcache) {
+ xlate_push_stats(op->ukey->xcache, push);
ovs_mutex_unlock(&op->ukey->mutex);
+ continue;
+ }
+ ovs_mutex_unlock(&op->ukey->mutex);
+
+ if (odp_flow_key_to_flow(op->op.u.flow_del.key,
+ op->op.u.flow_del.key_len, &flow)
+ == ODP_FIT_ERROR) {
+ continue;
}
- if (!xlate_receive(udpif->backer, NULL, op->op.u.flow_del.key,
- op->op.u.flow_del.key_len, &flow, &ofproto,
- NULL, NULL, &netflow, NULL)) {
+ error = xlate_lookup(udpif->backer, &flow, &ofproto,
+ NULL, NULL, &netflow, &ofp_in_port);
+ if (!error) {
struct xlate_in xin;
- xlate_in_init(&xin, ofproto, &flow, NULL, push->tcp_flags,
- NULL);
+ xlate_in_init(&xin, ofproto, &flow, ofp_in_port, NULL,
+ push->tcp_flags, NULL);
xin.resubmit_stats = push->n_packets ? push : NULL;
- xin.may_learn = may_learn;
+ xin.may_learn = push->n_packets > 0;
xin.skip_wildcards = true;
xlate_actions_for_side_effects(&xin);
if (netflow) {
netflow_flow_clear(netflow, &flow);
- netflow_unref(netflow);
}
}
}
unsigned int flow_limit;
dump_seq = seq_read(udpif->dump_seq);
- atomic_read(&udpif->flow_limit, &flow_limit);
+ atomic_read_relaxed(&udpif->flow_limit, &flow_limit);
dump_thread = dpif_flow_dump_thread_create(udpif->dump);
for (;;) {
struct dump_op ops[REVALIDATE_MAX_BATCH];
/* We couldn't acquire the ukey. This means that
* another revalidator is processing this flow
* concurrently, so don't bother processing it. */
- COVERAGE_INC(upcall_duplicate_flow);
+ COVERAGE_INC(dumped_duplicate_flow);
continue;
}
already_dumped = ukey->dump_seq == dump_seq;
if (already_dumped) {
- /* The flow has already been dumped and handled by another
- * revalidator during this flow dump operation. Skip it. */
- COVERAGE_INC(upcall_duplicate_flow);
+ /* The flow has already been handled during this flow dump
+ * operation. Skip it. */
+ if (ukey->xcache) {
+ COVERAGE_INC(dumped_duplicate_flow);
+ } else {
+ COVERAGE_INC(dumped_new_flow);
+ }
ovs_mutex_unlock(&ukey->mutex);
continue;
}
dpif_flow_dump_thread_destroy(dump_thread);
}
+/* Called with exclusive access to 'revalidator' and 'ukey'. */
+static bool
+handle_missed_revalidation(struct revalidator *revalidator,
+ struct udpif_key *ukey)
+ OVS_NO_THREAD_SAFETY_ANALYSIS
+{
+ struct udpif *udpif = revalidator->udpif;
+ struct dpif_flow flow;
+ struct ofpbuf buf;
+ uint64_t stub[DPIF_FLOW_BUFSIZE / 8];
+ bool keep = false;
+
+ COVERAGE_INC(revalidate_missed_dp_flow);
+
+ ofpbuf_use_stub(&buf, &stub, sizeof stub);
+ if (!dpif_flow_get(udpif->dpif, ukey->key, ukey->key_len, &buf, &flow)) {
+ keep = revalidate_ukey(udpif, ukey, &flow);
+ }
+ ofpbuf_uninit(&buf);
+
+ return keep;
+}
+
static void
revalidator_sweep__(struct revalidator *revalidator, bool purge)
OVS_NO_THREAD_SAFETY_ANALYSIS
/* During garbage collection, this revalidator completely owns its ukeys
* map, and therefore doesn't need to do any locking. */
HMAP_FOR_EACH_SAFE (ukey, next, hmap_node, revalidator->ukeys) {
- if (!ukey->flow_exists) {
- ukey_delete(revalidator, ukey);
- } else if (purge || ukey->dump_seq != dump_seq) {
+ if (ukey->flow_exists
+ && (purge
+ || (ukey->dump_seq != dump_seq
+ && revalidator->udpif->need_revalidate
+ && !handle_missed_revalidation(revalidator, ukey)))) {
struct dump_op *op = &ops[n_ops++];
- /* If we have previously seen a flow in the datapath, but it
- * hasn't been seen in the current dump, delete it. */
dump_op_init(op, ukey->key, ukey->key_len, ukey);
if (n_ops == REVALIDATE_MAX_BATCH) {
push_dump_ops(revalidator, ops, n_ops);
n_ops = 0;
}
+ } else if (!ukey->flow_exists) {
+ ukey_delete(revalidator, ukey);
}
}
unsigned int flow_limit;
size_t i;
- atomic_read(&udpif->flow_limit, &flow_limit);
+ atomic_read_relaxed(&udpif->flow_limit, &flow_limit);
ds_put_format(&ds, "%s:\n", dpif_name(udpif->dpif));
ds_put_format(&ds, "\tflows : (current %lu)"
const char *argv[] OVS_UNUSED,
void *aux OVS_UNUSED)
{
- atomic_store(&enable_megaflows, false);
+ atomic_store_relaxed(&enable_megaflows, false);
udpif_flush_all_datapaths();
unixctl_command_reply(conn, "megaflows disabled");
}
const char *argv[] OVS_UNUSED,
void *aux OVS_UNUSED)
{
- atomic_store(&enable_megaflows, true);
+ atomic_store_relaxed(&enable_megaflows, true);
udpif_flush_all_datapaths();
unixctl_command_reply(conn, "megaflows enabled");
}
unsigned int flow_limit = atoi(argv[1]);
LIST_FOR_EACH (udpif, list_node, &all_udpifs) {
- atomic_store(&udpif->flow_limit, flow_limit);
+ atomic_store_relaxed(&udpif->flow_limit, flow_limit);
}
ds_put_format(&ds, "set flow_limit to %u\n", flow_limit);
unixctl_command_reply(conn, ds_cstr(&ds));
void *aux OVS_UNUSED)
{
if (list_is_singleton(&all_udpifs)) {
- struct udpif *udpif;
+ struct udpif *udpif = NULL;
size_t len;
udpif = OBJECT_CONTAINING(list_front(&all_udpifs), udpif, list_node);
projects / cascardo / ovs.git / blobdiff