[ovs-dev,v3,1/2] ofproto: Allow xlate_actions() to fail.

diff --git a/ofproto/ofproto-dpif-upcall.c b/ofproto/ofproto-dpif-upcall.c
index 91648f5..9fbadf1 100644
--- a/ofproto/ofproto-dpif-upcall.c
+++ b/ofproto/ofproto-dpif-upcall.c
@@ -1098,7 +1098,7 @@  upcall_xlate(struct udpif *udpif, struct upcall *upcall,
         ofpbuf_use_const(&upcall->put_actions,
                          odp_actions->data, odp_actions->size);
     } else {
-        ofpbuf_init(&upcall->put_actions, 0);
+        /* upcall->put_actions already initialized by upcall_receive(). */
         compose_slow_path(udpif, &upcall->xout, upcall->flow,
                           upcall->flow->in_port.odp_port,
                           &upcall->put_actions);
diff --git a/ofproto/ofproto-dpif-xlate.c b/ofproto/ofproto-dpif-xlate.c
index 1738394..36a6fbc 100644
--- a/ofproto/ofproto-dpif-xlate.c
+++ b/ofproto/ofproto-dpif-xlate.c
@@ -313,8 +313,33 @@  struct xlate_ctx {
      * datapath actions. */
     bool action_set_has_group;  /* Action set contains OFPACT_GROUP? */
     struct ofpbuf action_set;   /* Action set. */
+
+    enum xlate_error error;     /* Translation failed. */
 };
 
+const char *xlate_strerror(enum xlate_error error)
+{
+    switch (error) {
+    case XLATE_OK:
+        return "OK";
+    case XLATE_BRIDGE_NOT_FOUND:
+        return "Bridge not found";
+    case XLATE_RECURSION_TOO_DEEP:
+        return "Recursion too deep";
+    case XLATE_TOO_MANY_RESUBMITS:
+        return "Too many resubmits";
+    case XLATE_STACK_TOO_DEEP:
+        return "Stack too deep";
+    case XLATE_NO_RECIRCULATION_CONTEXT:
+        return "No recirculation context";
+    case XLATE_RECIRCULATION_CONFLICT:
+        return "Recirculation conflict";
+    case XLATE_TOO_MANY_MPLS_LABELS:
+        return "Too many MPLS labels";
+    }
+    return "Unknown error";
+}
+
 static void xlate_action_set(struct xlate_ctx *ctx);
 static void xlate_commit_actions(struct xlate_ctx *ctx);
 
@@ -536,6 +561,17 @@  xlate_report(struct xlate_ctx *ctx, const char *format, ...)
     }
 }
 
+static struct vlog_rate_limit error_report_rl = VLOG_RATE_LIMIT_INIT(1, 5);
+
+#define XLATE_REPORT_ERROR(CTX, ...)                    \
+    do {                                                \
+        if (OVS_UNLIKELY((CTX)->xin->report_hook)) {    \
+            xlate_report(CTX, __VA_ARGS__);             \
+        } else {                                        \
+            VLOG_ERR_RL(&error_report_rl, __VA_ARGS__); \
+        }                                               \
+    } while (0)
+
 static inline void
 xlate_report_actions(struct xlate_ctx *ctx, const char *title,
                      const struct ofpact *ofpacts, size_t ofpacts_len)
@@ -2949,6 +2985,9 @@  compose_output_action__(struct xlate_ctx *ctx, ofp_port_t ofp_port,
          * recirculated packet! */
         ctx->exit = false;
 
+        /* Peer bridge errors do not propagate back. */
+        ctx->error = XLATE_OK;
+
         if (ctx->xin->resubmit_stats) {
             netdev_vport_inc_tx(xport->netdev, ctx->xin->resubmit_stats);
             netdev_vport_inc_rx(peer->netdev, ctx->xin->resubmit_stats);
@@ -3126,17 +3165,20 @@  xlate_recursively(struct xlate_ctx *ctx, struct rule_dpif *rule)
 static bool
 xlate_resubmit_resource_check(struct xlate_ctx *ctx)
 {
-    static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 1);
-
     if (ctx->recurse >= MAX_RESUBMIT_RECURSION + MAX_INTERNAL_RESUBMITS) {
-        VLOG_ERR_RL(&rl, "resubmit actions recursed over %d times",
-                    MAX_RESUBMIT_RECURSION);
+        XLATE_REPORT_ERROR(ctx, "resubmit actions recursed over %d times",
+                           MAX_RESUBMIT_RECURSION);
+        ctx->error = XLATE_RECURSION_TOO_DEEP;
     } else if (ctx->resubmits >= MAX_RESUBMITS + MAX_INTERNAL_RESUBMITS) {
-        VLOG_ERR_RL(&rl, "over %d resubmit actions", MAX_RESUBMITS);
+        XLATE_REPORT_ERROR(ctx, "over %d resubmit actions", MAX_RESUBMITS);
+        ctx->error = XLATE_TOO_MANY_RESUBMITS;
     } else if (ctx->odp_actions->size > UINT16_MAX) {
-        VLOG_ERR_RL(&rl, "resubmits yielded over 64 kB of actions");
+        XLATE_REPORT_ERROR(ctx, "resubmits yielded over 64 kB of actions");
+        /* NOT an error, as we'll be slow-pathing the flow in this case? */
+        ctx->exit = true; /* XXX: translation still terminated! */
     } else if (ctx->stack.size >= 65536) {
-        VLOG_ERR_RL(&rl, "resubmits yielded over 64 kB of stack");
+        XLATE_REPORT_ERROR(ctx, "resubmits yielded over 64 kB of stack");
+        ctx->error = XLATE_STACK_TOO_DEEP;
     } else {
         return true;
     }
@@ -3188,8 +3230,6 @@  xlate_table_action(struct xlate_ctx *ctx, ofp_port_t in_port, uint8_t table_id,
         ctx->table_id = old_table_id;
         return;
     }
-
-    ctx->exit = true;
 }
 
 static void
@@ -3559,8 +3599,8 @@  compose_recirculate_action__(struct xlate_ctx *ctx, uint8_t table)
          * with the udpif key ('ukey') created for each new datapath flow. */
         id = recirc_alloc_id_ctx(&state);
         if (!id) {
-            static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
-            VLOG_ERR_RL(&rl, "Failed to allocate recirculation id");
+            XLATE_REPORT_ERROR(ctx, "Failed to allocate recirculation id");
+            ctx->error = XLATE_NO_RECIRCULATION_CONTEXT;
             return;
         }
         xlate_out_add_recirc(ctx->xout, id);
@@ -3568,11 +3608,13 @@  compose_recirculate_action__(struct xlate_ctx *ctx, uint8_t table)
         /* Look up an existing recirc id for the given metadata state in the
          * flow.  No new reference is taken, as the ID is RCU protected and is
          * only required temporarily for verification.
-         *
-         * This might fail and return 0.  We let zero 'id' to be used in the
-         * RECIRC action below, which will fail all revalidations as zero is
-         * not a valid recirculation ID. */
+         * If flow tables have changed sufficiently this can fail and we will
+         * delete the old datapath flow. */
         id = recirc_find_id(&state);
+        if (!id) {
+            ctx->error = XLATE_NO_RECIRCULATION_CONTEXT;
+            return;
+        }
     }
 
     nl_msg_put_u32(ctx->odp_actions, OVS_ACTION_ATTR_RECIRC, id);
@@ -3615,13 +3657,12 @@  compose_mpls_push_action(struct xlate_ctx *ctx, struct ofpact_push_mpls *mpls)
         xlate_commit_actions(ctx);
     } else if (n >= FLOW_MAX_MPLS_LABELS) {
         if (ctx->xin->packet != NULL) {
-            static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
-            VLOG_WARN_RL(&rl, "bridge %s: dropping packet on which an "
+            XLATE_REPORT_ERROR(ctx, "bridge %s: dropping packet on which an "
                          "MPLS push action can't be performed as it would "
                          "have more MPLS LSEs than the %d supported.",
                          ctx->xbridge->name, FLOW_MAX_MPLS_LABELS);
         }
-        ctx->exit = true;
+        ctx->error = XLATE_TOO_MANY_MPLS_LABELS;
         return;
     }
 
@@ -3640,13 +3681,12 @@  compose_mpls_pop_action(struct xlate_ctx *ctx, ovs_be16 eth_type)
         }
     } else if (n >= FLOW_MAX_MPLS_LABELS) {
         if (ctx->xin->packet != NULL) {
-            static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
-            VLOG_WARN_RL(&rl, "bridge %s: dropping packet on which an "
+            XLATE_REPORT_ERROR(ctx, "bridge %s: dropping packet on which an "
                          "MPLS pop action can't be performed as it has "
                          "more MPLS LSEs than the %d supported.",
                          ctx->xbridge->name, FLOW_MAX_MPLS_LABELS);
         }
-        ctx->exit = true;
+        ctx->error = XLATE_TOO_MANY_MPLS_LABELS;
         ofpbuf_clear(ctx->odp_actions);
     }
 }
@@ -4274,6 +4314,10 @@  do_xlate_actions(const struct ofpact *ofpacts, size_t ofpacts_len,
         const struct ofpact_set_field *set_field;
         const struct mf_field *mf;
 
+        if (ctx->error) {
+            break;
+        }
+
         if (ctx->exit) {
             /* Check if need to store the remaining actions for later
              * execution. */
@@ -4632,7 +4676,7 @@  do_xlate_actions(const struct ofpact *ofpacts, size_t ofpacts_len,
 
         /* Check if need to store this and the remaining actions for later
          * execution. */
-        if (ctx->exit && ctx_first_recirculation_action(ctx)) {
+        if (!ctx->error && ctx->exit && ctx_first_recirculation_action(ctx)) {
             recirc_unroll_actions(a, OFPACT_ALIGN(ofpacts_len -
                                                   ((uint8_t *)a -
                                                    (uint8_t *)ofpacts)),
@@ -4688,8 +4732,15 @@  void
 xlate_actions_for_side_effects(struct xlate_in *xin)
 {
     struct xlate_out xout;
+    enum xlate_error error;
+
+    error = xlate_actions(xin, &xout);
+    if (error) {
+        static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
+
+        VLOG_WARN_RL(&rl, "xlate_actions failed (%s)!", xlate_strerror(error));
+    }
 
-    xlate_actions(xin, &xout);
     xlate_out_uninit(&xout);
 }
 
@@ -4878,8 +4929,12 @@  xlate_wc_finish(struct xlate_ctx *ctx)
 /* Translates the flow, actions, or rule in 'xin' into datapath actions in
  * 'xout'.
  * The caller must take responsibility for eventually freeing 'xout', with
- * xlate_out_uninit(). */
-void
+ * xlate_out_uninit().
+ * Returns 'XLATE_OK' if translation was successful.  In case of an error an
+ * empty set of actions will be returned in 'xin->odp_actions' (if non-NULL),
+ * so that most callers may ignore the return value and transparently install a
+ * drop flow when the translation fails. */
+enum xlate_error
 xlate_actions(struct xlate_in *xin, struct xlate_out *xout)
 {
     *xout = (struct xlate_out) {
@@ -4891,7 +4946,7 @@  xlate_actions(struct xlate_in *xin, struct xlate_out *xout)
     struct xlate_cfg *xcfg = ovsrcu_get(struct xlate_cfg *, &xcfgp);
     struct xbridge *xbridge = xbridge_lookup(xcfg, xin->ofproto);
     if (!xbridge) {
-        return;
+        return XLATE_BRIDGE_NOT_FOUND;
     }
 
     struct flow *flow = &xin->flow;
@@ -4924,6 +4979,7 @@  xlate_actions(struct xlate_in *xin, struct xlate_out *xout)
         .sflow_odp_port = 0,
         .nf_output_iface = NF_OUT_DROP,
         .exit = false,
+        .error = XLATE_OK,
         .mirrors = 0,
 
         .recirc_action_offset = -1,
@@ -4976,6 +5032,7 @@  xlate_actions(struct xlate_in *xin, struct xlate_out *xout)
 
             VLOG_WARN_RL(&rl, "Recirculation conflict (%s)!", conflict);
             xlate_report(&ctx, "- Recirculation conflict (%s)!", conflict);
+            ctx.error = XLATE_RECIRCULATION_CONFLICT;
             goto exit;
         }
 
@@ -4990,6 +5047,7 @@  xlate_actions(struct xlate_in *xin, struct xlate_out *xout)
                 static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 1);
                 VLOG_WARN_RL(&rl, "Recirculation bridge no longer exists.");
                 xlate_report(&ctx, "- Recirculation bridge no longer exists.");
+                ctx.error = XLATE_BRIDGE_NOT_FOUND;
                 goto exit;
             }
             ctx.xbridge = new_bridge;
@@ -5049,6 +5107,7 @@  xlate_actions(struct xlate_in *xin, struct xlate_out *xout)
 
         VLOG_WARN_RL(&rl, "Recirculation context not found for ID %"PRIx32,
                      flow->recirc_id);
+        ctx.error = XLATE_NO_RECIRCULATION_CONTEXT;
         goto exit;
     }
     /* The bridge is now known so obtain its table version. */
@@ -5140,6 +5199,9 @@  xlate_actions(struct xlate_in *xin, struct xlate_out *xout)
 
             mirror_ingress_packet(&ctx);
             do_xlate_actions(ofpacts, ofpacts_len, &ctx);
+            if (ctx.error) {
+                goto exit;
+            }
 
             /* We've let OFPP_NORMAL and the learning action look at the
              * packet, so drop it now if forwarding is disabled. */
@@ -5219,6 +5281,15 @@  exit:
     ofpbuf_uninit(&ctx.stack);
     ofpbuf_uninit(&ctx.action_set);
     ofpbuf_uninit(&scratch_actions);
+
+    /* Make sure we return a "drop flow" in case of an error. */
+    if (ctx.error) {
+        xout->slow = 0;
+        if (xin->odp_actions) {
+            ofpbuf_clear(xin->odp_actions);
+        }
+    }
+    return ctx.error;
 }
 
 /* Sends 'packet' out 'ofport'.
diff --git a/ofproto/ofproto-dpif-xlate.h b/ofproto/ofproto-dpif-xlate.h
index 585650c..2e23771 100644
--- a/ofproto/ofproto-dpif-xlate.h
+++ b/ofproto/ofproto-dpif-xlate.h
@@ -239,7 +239,21 @@  int xlate_lookup(const struct dpif_backer *, const struct flow *,
                  struct dpif_sflow **, struct netflow **,
                  ofp_port_t *ofp_in_port);
 
-void xlate_actions(struct xlate_in *, struct xlate_out *);
+enum xlate_error {
+    XLATE_OK = 0,
+    XLATE_BRIDGE_NOT_FOUND,
+    XLATE_RECURSION_TOO_DEEP,
+    XLATE_TOO_MANY_RESUBMITS,
+    XLATE_STACK_TOO_DEEP,
+    XLATE_NO_RECIRCULATION_CONTEXT,
+    XLATE_RECIRCULATION_CONFLICT,
+    XLATE_TOO_MANY_MPLS_LABELS,
+};
+
+const char *xlate_strerror(enum xlate_error error);
+
+enum xlate_error xlate_actions(struct xlate_in *, struct xlate_out *);
+
 void xlate_in_init(struct xlate_in *, struct ofproto_dpif *,
                    const struct flow *, ofp_port_t in_port, struct rule_dpif *,
                    uint16_t tcp_flags, const struct dp_packet *packet,
diff --git a/ofproto/ofproto-dpif.c b/ofproto/ofproto-dpif.c
index 9fe565e..b924af5 100644
--- a/ofproto/ofproto-dpif.c
+++ b/ofproto/ofproto-dpif.c
@@ -3730,7 +3730,10 @@  ofproto_dpif_execute_actions__(struct ofproto_dpif *ofproto,
     xin.resubmit_stats = &stats;
     xin.recurse = recurse;
     xin.resubmits = resubmits;
-    xlate_actions(&xin, &xout);
+    if (xlate_actions(&xin, &xout) != XLATE_OK) {
+        error = EINVAL;
+        goto out;
+    }
 
     execute.actions = odp_actions.data;
     execute.actions_len = odp_actions.size;
@@ -3749,7 +3752,7 @@  ofproto_dpif_execute_actions__(struct ofproto_dpif *ofproto,
     execute.packet->md.in_port.odp_port = ofp_port_to_odp_port(ofproto, in_port);
 
     error = dpif_execute(ofproto->backer->dpif, &execute);
-
+out:
     xlate_out_uninit(&xout);
     ofpbuf_uninit(&odp_actions);
 
@@ -4993,6 +4996,7 @@  ofproto_trace(struct ofproto_dpif *ofproto, struct flow *flow,
               struct ds *ds)
 {
     struct trace_ctx trace;
+    enum xlate_error error;
 
     ds_put_format(ds, "Bridge: %s\n", ofproto->up.name);
     ds_put_cstr(ds, "Flow: ");
@@ -5012,8 +5016,7 @@  ofproto_trace(struct ofproto_dpif *ofproto, struct flow *flow,
     trace.xin.resubmit_hook = trace_resubmit;
     trace.xin.report_hook = trace_report_valist;
 
-    xlate_actions(&trace.xin, &trace.xout);
-
+    error = xlate_actions(&trace.xin, &trace.xout);
     ds_put_char(ds, '\n');
     trace_format_flow(ds, 0, "Final flow", &trace);
     trace_format_megaflow(ds, 0, "Megaflow", &trace);
@@ -5021,7 +5024,10 @@  ofproto_trace(struct ofproto_dpif *ofproto, struct flow *flow,
     ds_put_cstr(ds, "Datapath actions: ");
     format_odp_actions(ds, trace.odp_actions.data, trace.odp_actions.size);
 
-    if (trace.xout.slow) {
+    if (error != XLATE_OK) {
+        ds_put_format(ds, "\nTranslation failed (%s), packet is dropped.\n",
+                      xlate_strerror(error));
+    } else if (trace.xout.slow) {
         enum slow_path_reason slow;
 
         ds_put_cstr(ds, "\nThis flow is handled by the userspace "
diff --git a/tests/ofproto-dpif.at b/tests/ofproto-dpif.at
index f1c7cb6..aa456c7 100644
--- a/tests/ofproto-dpif.at
+++ b/tests/ofproto-dpif.at
@@ -6697,9 +6697,10 @@  OVS_VSWITCHD_START
 AT_CHECK([ovs-ofctl add-flow br0 actions=resubmit:1,resubmit:2,output:3])
 AT_CHECK([ovs-appctl -vpoll_loop:off ofproto/trace br0 'eth_dst=ff:ff:ff:ff:ff:ff'],
   [0], [stdout])
-AT_CHECK([tail -1 stdout], [0], [Datapath actions: drop
+AT_CHECK([tail -1 stdout], [0],
+  [Translation failed (Recursion too deep), packet is dropped.
 ])
-AT_CHECK([grep -c 'resubmit actions recursed over 64 times' ovs-vswitchd.log],
+AT_CHECK([grep -c 'resubmit actions recursed over 64 times' stdout],
   [0], [1
 ])
 OVS_VSWITCHD_STOP(["/resubmit actions recursed/d"])
@@ -6715,7 +6716,10 @@  ADD_OF_PORTS([br0], 1)
  echo "in_port=65, actions=local") > flows
  AT_CHECK([ovs-ofctl add-flows br0 flows])
 AT_CHECK([ovs-appctl -vpoll_loop:off ofproto/trace br0 'in_port=1'], [0], [stdout])
-AT_CHECK([grep -c 'over 4096 resubmit actions' ovs-vswitchd.log], [0], [1
+AT_CHECK([tail -1 stdout], [0],
+  [Translation failed (Too many resubmits), packet is dropped.
+])
+AT_CHECK([grep -c 'over 4096 resubmit actions' stdout], [0], [1
 ])
 OVS_VSWITCHD_STOP(["/over.*resubmit actions/d"])
 AT_CLEANUP
@@ -6733,7 +6737,7 @@  AT_CHECK([ovs-appctl -vpoll_loop:off ofproto/trace br0 'in_port=1'], [0], [stdou
 AT_CHECK([grep -c -e '- Uses action(s) not supported by datapath' stdout],
   [0], [1
 ])
-AT_CHECK([grep -c 'resubmits yielded over 64 kB of actions' ovs-vswitchd.log], [0], [1
+AT_CHECK([grep -c 'resubmits yielded over 64 kB of actions' stdout], [0], [1
 ])
 OVS_VSWITCHD_STOP(["/resubmits yielded over 64 kB of actions/d"])
 AT_CLEANUP
@@ -6749,7 +6753,10 @@  ADD_OF_PORTS([br0], 1)
  echo "in_port=13, actions=$push,$push,$push,$push,$push,$push,$push,$push") > flows
  AT_CHECK([ovs-ofctl add-flows br0 flows])
 AT_CHECK([ovs-appctl -vpoll_loop:off ofproto/trace br0 'in_port=1'], [0], [stdout])
-AT_CHECK([grep -c 'resubmits yielded over 64 kB of stack' ovs-vswitchd.log], [0], [1
+AT_CHECK([tail -1 stdout], [0],
+  [Translation failed (Stack too deep), packet is dropped.
+])
+AT_CHECK([grep -c 'resubmits yielded over 64 kB of stack' stdout], [0], [1
 ])
 OVS_VSWITCHD_STOP(["/resubmits yielded over 64 kB of stack/d"])
 AT_CLEANUP