lib/ofproto: Add comments about races in ofproto_flush().

[cascardo/ovs.git] / ofproto / ofproto-dpif-upcall.c

diff --git a/ofproto/ofproto-dpif-upcall.c b/ofproto/ofproto-dpif-upcall.c
index cce89dd..1f9c484 100644 (file)
--- a/ofproto/ofproto-dpif-upcall.c
+++ b/ofproto/ofproto-dpif-upcall.c
@@ -45,7 +45,8 @@
 
 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,
@@ -576,6 +577,7 @@ recv_upcalls(struct handler *handler)
     struct ofpbuf recv_bufs[UPCALL_MAX_BATCH];
     struct dpif_upcall dupcalls[UPCALL_MAX_BATCH];
     struct upcall upcalls[UPCALL_MAX_BATCH];
+    struct flow flows[UPCALL_MAX_BATCH];
     size_t n_upcalls, i;
 
     n_upcalls = 0;
@@ -583,8 +585,8 @@ recv_upcalls(struct handler *handler)
         struct ofpbuf *recv_buf = &recv_bufs[n_upcalls];
         struct dpif_upcall *dupcall = &dupcalls[n_upcalls];
         struct upcall *upcall = &upcalls[n_upcalls];
+        struct flow *flow = &flows[n_upcalls];
         struct pkt_metadata md;
-        struct flow flow;
         int error;
 
         ofpbuf_use_stub(recv_buf, recv_stubs[n_upcalls],
@@ -594,13 +596,13 @@ recv_upcalls(struct handler *handler)
             break;
         }
 
-        if (odp_flow_key_to_flow(dupcall->key, dupcall->key_len, &flow)
+        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);
+                               dupcall->type, dupcall->userdata, flow);
         if (error) {
             if (error == ENODEV) {
                 /* Received packet on datapath port for which we couldn't
@@ -610,7 +612,7 @@ recv_upcalls(struct handler *handler)
                 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);
+                             "port %"PRIu32, flow->in_port.odp_port);
             }
             goto free_dupcall;
         }
@@ -620,12 +622,12 @@ recv_upcalls(struct handler *handler)
 
         upcall->out_tun_key = dupcall->out_tun_key;
 
-        if (vsp_adjust_flow(upcall->ofproto, &flow, &dupcall->packet)) {
+        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);
+        md = pkt_metadata_from_flow(flow);
+        flow_extract(&dupcall->packet, &md, flow);
 
         error = process_upcall(udpif, upcall, NULL);
         if (error) {
@@ -818,9 +820,10 @@ compose_slow_path(struct udpif *udpif, struct xlate_out *xout,
                              buf);
 }
 
-/* The upcall must be destroyed with upcall_uninit() before quiescing,
- * as the referred objects are guaranteed to exist only until the calling
- * thread quiesces. */
+/* 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,
@@ -944,7 +947,7 @@ upcall_cb(const struct ofpbuf *packet, const struct flow *flow,
     error = upcall_receive(&upcall, udpif->backer, packet, type, userdata,
                            flow);
     if (error) {
-        goto out;
+        return error;
     }
 
     error = process_upcall(udpif, &upcall, actions);
@@ -962,8 +965,7 @@ upcall_cb(const struct ofpbuf *packet, const struct flow *flow,
             /* XXX: This could be avoided with sufficient API changes. */
             *wc = upcall.xout.wc;
         } else {
-            memset(wc, 0xff, sizeof *wc);
-            flow_wildcards_clear_non_packet_fields(wc);
+            flow_wildcards_init_for_packet(wc, flow);
         }
     }
 
@@ -1136,6 +1138,7 @@ handle_upcalls(struct udpif *udpif, struct upcall *upcalls,
             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;
+            op->u.execute.probe = false;
         }
     }
 
@@ -1547,15 +1550,19 @@ revalidate(struct revalidator *revalidator)
                 /* 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;
             }
@@ -1749,7 +1756,7 @@ upcall_unixctl_dump_wait(struct unixctl_conn *conn,
                          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);