classifier: Support table versioning

This patch allows classifier rules to become visible and invisible in
specific versions.  A 'version' is defined as a positive monotonically
increasing integer, which never wraps around.

The new 'visibility' attribute replaces the prior 'to_be_removed' and
'visible' attributes.

When versioning is not used, the 'version' parameter should be passed
as 'CLS_MIN_VERSION' when creating rules, and 'CLS_MAX_VERSION' when
looking up flows.

This feature enables the support for atomic OpenFlow bundles without
significant performance penalty on 64-bit systems. There is a
performance decrease in 32-bit systems due to 64-bit atomics used.

Signed-off-by: Jarno Rajahalme <jrajahalme@nicira.com>
Acked-by: Ben Pfaff <blp@nicira.com>

9 files changed, 347 insertions, 195 deletions

diff --git a/lib/classifier-private.h b/lib/classifier-private.h
index a7edbe93b..2703b75a7 100644
--- a/lib/classifier-private.h
+++ b/lib/classifier-private.h
@@ -79,13 +79,63 @@ struct cls_match {
                                                     * 'indices'. */
     /* Accessed by all readers. */
     struct cmap_node cmap_node; /* Within struct cls_subtable 'rules'. */
-    bool visible;
+
+    /* Controls rule's visibility to lookups.
+     *
+     * When 'visibility' is:
+     *
+     * > 0  - rule is visible starting from version 'visibility'
+     * <= 0 - rule is invisible starting from version '-(visibility)'
+     *
+     * The minimum version number used in lookups is 1 (== CLS_NO_VERSION),
+     * which implies that when 'visibility' is:
+     *
+     * 1    - rule is visible in all lookup versions
+     * 0    - rule is invisible in all lookup versions. */
+    atomic_llong visibility;
+
     const struct cls_rule *cls_rule;
     OVSRCU_TYPE(struct cls_conjunction_set *) conj_set;
     const struct miniflow flow; /* Matching rule. Mask is in the subtable. */
     /* 'flow' must be the last field. */
 };
 
+static inline void
+cls_match_set_visibility(struct cls_match *rule, long long version)
+{
+    atomic_store_relaxed(&rule->visibility, version);
+}
+
+static inline bool
+cls_match_visible_in_version(const struct cls_match *rule, long long version)
+{
+    long long visibility;
+
+    /* C11 does not want to access an atomic via a const object pointer. */
+    atomic_read_relaxed(&CONST_CAST(struct cls_match *, rule)->visibility,
+                        &visibility);
+
+    if (OVS_LIKELY(visibility > 0)) {
+        /* Rule is visible starting from version 'visibility'. */
+        return version >= visibility;
+    } else {
+        /* Rule is invisible starting from version '-visibility'. */
+        return version < -visibility;
+    }
+}
+
+static inline bool
+cls_match_is_eventually_invisible(const struct cls_match *rule)
+{
+    long long visibility;
+
+    /* C11 does not want to access an atomic via a const object pointer. */
+    atomic_read_relaxed(&CONST_CAST(struct cls_match *, rule)->visibility,
+                        &visibility);
+
+    return visibility <= 0;
+}
+
 /* A longest-prefix match tree. */
 struct trie_node {
     uint32_t prefix;           /* Prefix bits for this node, MSB first. */
diff --git a/lib/classifier.c b/lib/classifier.c
index 6075cf785..2b2d3f646 100644
--- a/lib/classifier.c
+++ b/lib/classifier.c
@@ -99,7 +99,7 @@ cls_match_alloc(const struct cls_rule *rule,
     rculist_init(&cls_match->list);
     *CONST_CAST(const struct cls_rule **, &cls_match->cls_rule) = rule;
     *CONST_CAST(int *, &cls_match->priority) = rule->priority;
-    cls_match->visible = false;
+    atomic_init(&cls_match->visibility, 0);   /* Initially invisible. */
     miniflow_clone_inline(CONST_CAST(struct miniflow *, &cls_match->flow),
                           &rule->match.flow, count);
     ovsrcu_set_hidden(&cls_match->conj_set,
@@ -115,6 +115,7 @@ static struct cls_subtable *insert_subtable(struct classifier *cls,
 static void destroy_subtable(struct classifier *cls, struct cls_subtable *);
 
 static const struct cls_match *find_match_wc(const struct cls_subtable *,
+                                             long long version,
                                              const struct flow *,
                                              struct trie_ctx *,
                                              unsigned int n_tries,
@@ -139,12 +140,12 @@ next_rule_in_list(const struct cls_match *rule, const struct cls_match *head)
 
 /* Return the next lower-priority rule in the list that is visible.  Multiple
  * identical rules with the same priority may exist transitionally.  In that
- * case the first rule of a given priority has been marked as 'to_be_removed',
- * and the later rules are marked as '!visible'.  This gets a bit complex if
- * there are two rules of the same priority in the list, as in that case the
- * head and tail of the list will have the same priority. */
+ * case the first rule of a given priority has been marked as visible in one
+ * version and the later rules are marked as visible on the other version.
+ * This makes it possible to for the head and tail of the list have the same
+ * priority. */
 static inline const struct cls_match *
-next_visible_rule_in_list(const struct cls_match *rule)
+next_visible_rule_in_list(const struct cls_match *rule, long long version)
 {
     const struct cls_match *next = rule;
 
@@ -154,7 +155,7 @@ next_visible_rule_in_list(const struct cls_match *rule)
             /* We have reached the head of the list, stop. */
             return NULL;
         }
-    } while (!next->visible);
+    } while (!cls_match_visible_in_version(next, version));
 
     return next;
 }
@@ -206,11 +207,14 @@ static bool mask_prefix_bits_set(const struct flow_wildcards *,
 /* cls_rule. */
 
 static inline void
-cls_rule_init__(struct cls_rule *rule, unsigned int priority)
+cls_rule_init__(struct cls_rule *rule, unsigned int priority,
+                long long version)
 {
+    ovs_assert(version > 0);
+
     rculist_init(&rule->node);
-    rule->priority = priority;
-    rule->to_be_removed = false;
+    *CONST_CAST(int *, &rule->priority) = priority;
+    *CONST_CAST(long long *, &rule->version) = version;
     rule->cls_match = NULL;
 }
 
@@ -223,19 +227,21 @@ cls_rule_init__(struct cls_rule *rule, unsigned int priority)
  * Clients should not use priority INT_MIN.  (OpenFlow uses priorities between
  * 0 and UINT16_MAX, inclusive.) */
 void
-cls_rule_init(struct cls_rule *rule, const struct match *match, int priority)
+cls_rule_init(struct cls_rule *rule, const struct match *match, int priority,
+              long long version)
 {
-    cls_rule_init__(rule, priority);
-    minimatch_init(&rule->match, match);
+    cls_rule_init__(rule, priority, version);
+    minimatch_init(CONST_CAST(struct minimatch *, &rule->match), match);
 }
 
 /* Same as cls_rule_init() for initialization from a "struct minimatch". */
 void
 cls_rule_init_from_minimatch(struct cls_rule *rule,
-                             const struct minimatch *match, int priority)
+                             const struct minimatch *match, int priority,
+                             long long version)
 {
-    cls_rule_init__(rule, priority);
-    minimatch_clone(&rule->match, match);
+    cls_rule_init__(rule, priority, version);
+    minimatch_clone(CONST_CAST(struct minimatch *, &rule->match), match);
 }
 
 /* Initializes 'dst' as a copy of 'src'.
@@ -244,20 +250,21 @@ cls_rule_init_from_minimatch(struct cls_rule *rule,
 void
 cls_rule_clone(struct cls_rule *dst, const struct cls_rule *src)
 {
-    cls_rule_init__(dst, src->priority);
-    minimatch_clone(&dst->match, &src->match);
+    cls_rule_init__(dst, src->priority, src->version);
+    minimatch_clone(CONST_CAST(struct minimatch *, &dst->match), &src->match);
 }
 
 /* Initializes 'dst' with the data in 'src', destroying 'src'.
+ *
  * 'src' must be a cls_rule NOT in a classifier.
  *
  * The caller must eventually destroy 'dst' with cls_rule_destroy(). */
 void
 cls_rule_move(struct cls_rule *dst, struct cls_rule *src)
 {
-    ovs_assert(!src->cls_match);   /* Must not be in a classifier. */
-    cls_rule_init__(dst, src->priority);
-    minimatch_move(&dst->match, &src->match);
+    cls_rule_init__(dst, src->priority, src->version);
+    minimatch_move(CONST_CAST(struct minimatch *, &dst->match),
+                   CONST_CAST(struct minimatch *, &src->match));
 }
 
 /* Frees memory referenced by 'rule'.  Doesn't free 'rule' itself (it's
@@ -275,7 +282,7 @@ cls_rule_destroy(struct cls_rule *rule)
     ovs_assert(rculist_next_protected(&rule->node) == RCULIST_POISON
                || rculist_is_empty(&rule->node));
 
-    minimatch_destroy(&rule->match);
+    minimatch_destroy(CONST_CAST(struct minimatch *, &rule->match));
 }
 
 void
@@ -327,15 +334,53 @@ cls_rule_is_catchall(const struct cls_rule *rule)
     return minimask_is_catchall(&rule->match.mask);
 }
 
-/* Rules inserted during classifier_defer() need to be made visible before
- * calling classifier_publish().
+/* Makes rule invisible after 'version'.  Once that version is made invisible
+ * (by changing the version parameter used in lookups), the rule should be
+ * actually removed via ovsrcu_postpone().
+ *
+ * 'rule_' must be in a classifier. */
+void
+cls_rule_make_invisible_in_version(const struct cls_rule *rule_,
+                                   long long version, long long lookup_version)
+{
+    struct cls_match *rule = rule_->cls_match;
+
+    /* XXX: Adjust when versioning is actually used. */
+    ovs_assert(version >= rule_->version && version >= lookup_version);
+
+    /* Normally, we call this when deleting a rule that is already visible to
+     * lookups.  However, sometimes a bundle transaction will add a rule and
+     * then delete it before the rule has ever become visible.  If we set such
+     * a rule to become invisible in a future 'version', it would become
+     * visible to all prior versions.  So, in this case we must set the rule
+     * visibility to 0 (== never visible). */
+    if (cls_match_visible_in_version(rule, lookup_version)) {
+        /* Make invisible starting at 'version'. */
+        atomic_store_relaxed(&rule->visibility, -version);
+    } else {
+        /* Rule has not yet been visible to lookups, make invisible in all
+         * version. */
+        atomic_store_relaxed(&rule->visibility, 0);
+    }
+}
+
+/* This undoes the change made by cls_rule_make_invisible_after_version().
  *
  * 'rule' must be in a classifier. */
-void cls_rule_make_visible(const struct cls_rule *rule)
+void
+cls_rule_restore_visibility(const struct cls_rule *rule)
 {
-    rule->cls_match->visible = true;
+    atomic_store_relaxed(&rule->cls_match->visibility, rule->version);
 }
 
+/* Return true if 'rule' is visible in 'version'.
+ *
+ * 'rule' must be in a classifier. */
+bool
+cls_rule_visible_in_version(const struct cls_rule *rule, long long version)
+{
+    return cls_match_visible_in_version(rule->cls_match, version);
+}
 
 /* Initializes 'cls' as a classifier that initially contains no classification
  * rules. */
@@ -597,7 +642,7 @@ const struct cls_rule *
 classifier_replace(struct classifier *cls, const struct cls_rule *rule,
                    const struct cls_conjunction *conjs, size_t n_conjs)
 {
-    struct cls_match *new = cls_match_alloc(rule, conjs, n_conjs);
+    struct cls_match *new;
     struct cls_subtable *subtable;
     uint32_t ihash[CLS_MAX_INDICES];
     uint8_t prev_be64ofs = 0;
@@ -607,6 +652,11 @@ classifier_replace(struct classifier *cls, const struct cls_rule *rule,
     uint32_t hash;
     int i;
 
+    ovs_assert(rule->version > 0);
+
+    /* 'new' is initially invisible to lookups. */
+    new = cls_match_alloc(rule, conjs, n_conjs);
+
     CONST_CAST(struct cls_rule *, rule)->cls_match = new;
 
     subtable = find_subtable(cls, &rule->match.mask);
@@ -673,12 +723,12 @@ classifier_replace(struct classifier *cls, const struct cls_rule *rule,
         struct cls_match *iter;
 
         /* Scan the list for the insertion point that will keep the list in
-         * order of decreasing priority.
-         * Insert after 'to_be_removed' rules of the same priority. */
+         * order of decreasing priority.  Insert after rules marked invisible
+         * in any version of the same priority. */
         FOR_EACH_RULE_IN_LIST_PROTECTED (iter, head) {
             if (rule->priority > iter->priority
                 || (rule->priority == iter->priority
-                    && !iter->cls_rule->to_be_removed)) {
+                    && !cls_match_is_eventually_invisible(iter))) {
                 break;
             }
         }
@@ -716,8 +766,8 @@ classifier_replace(struct classifier *cls, const struct cls_rule *rule,
 
                 /* No change in subtable's max priority or max count. */
 
-                /* Make rule visible to lookups? */
-                new->visible = cls->publish;
+                /* Make 'new' visible to lookups in the appropriate version. */
+                cls_match_set_visibility(new, rule->version);
 
                 /* Make rule visible to iterators (immediately). */
                 rculist_replace(CONST_CAST(struct rculist *, &rule->node),
@@ -732,8 +782,8 @@ classifier_replace(struct classifier *cls, const struct cls_rule *rule,
         }
     }
 
-    /* Make rule visible to lookups? */
-    new->visible = cls->publish;
+    /* Make 'new' visible to lookups in the appropriate version. */
+    cls_match_set_visibility(new, rule->version);
 
     /* Make rule visible to iterators (immediately). */
     rculist_push_back(&subtable->rules_list,
@@ -1026,8 +1076,9 @@ free_conjunctive_matches(struct hmap *matches,
  * 'flow' is non-const to allow for temporary modifications during the lookup.
  * Any changes are restored before returning. */
 static const struct cls_rule *
-classifier_lookup__(const struct classifier *cls, struct flow *flow,
-                    struct flow_wildcards *wc, bool allow_conjunctive_matches)
+classifier_lookup__(const struct classifier *cls, long long version,
+                    struct flow *flow, struct flow_wildcards *wc,
+                    bool allow_conjunctive_matches)
 {
     const struct cls_partition *partition;
     struct trie_ctx trie_ctx[CLS_MAX_TRIES];
@@ -1094,7 +1145,8 @@ classifier_lookup__(const struct classifier *cls, struct flow *flow,
 
         /* Skip subtables with no match, or where the match is lower-priority
          * than some certain match we've already found. */
-        match = find_match_wc(subtable, flow, trie_ctx, cls->n_tries, wc);
+        match = find_match_wc(subtable, version, flow, trie_ctx, cls->n_tries,
+                              wc);
         if (!match || match->priority <= hard_pri) {
             continue;
         }
@@ -1218,7 +1270,7 @@ classifier_lookup__(const struct classifier *cls, struct flow *flow,
                 const struct cls_rule *rule;
 
                 flow->conj_id = id;
-                rule = classifier_lookup__(cls, flow, wc, false);
+                rule = classifier_lookup__(cls, version, flow, wc, false);
                 flow->conj_id = saved_conj_id;
 
                 if (rule) {
@@ -1246,7 +1298,7 @@ classifier_lookup__(const struct classifier *cls, struct flow *flow,
             }
 
             /* Find next-lower-priority flow with identical flow match. */
-            match = next_visible_rule_in_list(soft[i]->match);
+            match = next_visible_rule_in_list(soft[i]->match, version);
             if (match) {
                 soft[i] = ovsrcu_get(struct cls_conjunction_set *,
                                      &match->conj_set);
@@ -1271,9 +1323,10 @@ classifier_lookup__(const struct classifier *cls, struct flow *flow,
     return hard ? hard->cls_rule : NULL;
 }
 
-/* Finds and returns the highest-priority rule in 'cls' that matches 'flow'.
- * Returns a null pointer if no rules in 'cls' match 'flow'.  If multiple rules
- * of equal priority match 'flow', returns one arbitrarily.
+/* Finds and returns the highest-priority rule in 'cls' that matches 'flow' and
+ * that is visible in 'version'.  Returns a null pointer if no rules in 'cls'
+ * match 'flow'.  If multiple rules of equal priority match 'flow', returns one
+ * arbitrarily.
  *
  * If a rule is found and 'wc' is non-null, bitwise-OR's 'wc' with the
  * set of bits that were significant in the lookup.  At some point
@@ -1283,18 +1336,16 @@ classifier_lookup__(const struct classifier *cls, struct flow *flow,
  * 'flow' is non-const to allow for temporary modifications during the lookup.
  * Any changes are restored before returning. */
 const struct cls_rule *
-classifier_lookup(const struct classifier *cls, struct flow *flow,
-                  struct flow_wildcards *wc)
+classifier_lookup(const struct classifier *cls, long long version,
+                  struct flow *flow, struct flow_wildcards *wc)
 {
-    return classifier_lookup__(cls, flow, wc, true);
+    return classifier_lookup__(cls, version, flow, wc, true);
 }
 
 /* Finds and returns a rule in 'cls' with exactly the same priority and
- * matching criteria as 'target'.  Returns a null pointer if 'cls' doesn't
- * contain an exact match.
- *
- * Returns the first matching rule that is not 'to_be_removed'.  Only one such
- * rule may exist. */
+ * matching criteria as 'target', and that is visible in 'target->version.
+ * Only one such rule may ever exist.  Returns a null pointer if 'cls' doesn't
+ * contain an exact match. */
 const struct cls_rule *
 classifier_find_rule_exactly(const struct classifier *cls,
                              const struct cls_rule *target)
@@ -1318,7 +1369,7 @@ classifier_find_rule_exactly(const struct classifier *cls,
             break; /* Not found. */
         }
         if (rule->priority == target->priority
-            && !rule->cls_rule->to_be_removed) {
+            && cls_match_visible_in_version(rule, target->version)) {
             return rule->cls_rule;
         }
     }
@@ -1326,16 +1377,18 @@ classifier_find_rule_exactly(const struct classifier *cls,
 }
 
 /* Finds and returns a rule in 'cls' with priority 'priority' and exactly the
- * same matching criteria as 'target'.  Returns a null pointer if 'cls' doesn't
- * contain an exact match. */
+ * same matching criteria as 'target', and that is visible in 'version'.
+ * Returns a null pointer if 'cls' doesn't contain an exact match visible in
+ * 'version'. */
 const struct cls_rule *
 classifier_find_match_exactly(const struct classifier *cls,
-                              const struct match *target, int priority)
+                              const struct match *target, int priority,
+                              long long version)
 {
     const struct cls_rule *retval;
     struct cls_rule cr;
 
-    cls_rule_init(&cr, target, priority);
+    cls_rule_init(&cr, target, priority, version);
     retval = classifier_find_rule_exactly(cls, &cr);
     cls_rule_destroy(&cr);
 
@@ -1344,16 +1397,12 @@ classifier_find_match_exactly(const struct classifier *cls,
 
 /* Checks if 'target' would overlap any other rule in 'cls'.  Two rules are
  * considered to overlap if both rules have the same priority and a packet
- * could match both.
+ * could match both, and if both rules are visible in the same version.
  *
  * A trivial example of overlapping rules is two rules matching disjoint sets
  * of fields. E.g., if one rule matches only on port number, while another only
  * on dl_type, any packet from that specific port and with that specific
- * dl_type could match both, if the rules also have the same priority.
- *
- * 'target' is not considered to overlap with a rule that has been marked
- * as 'to_be_removed'.
- */
+ * dl_type could match both, if the rules also have the same priority. */
 bool
 classifier_rule_overlaps(const struct classifier *cls,
                          const struct cls_rule *target)
@@ -1371,9 +1420,10 @@ classifier_rule_overlaps(const struct classifier *cls,
 
         RCULIST_FOR_EACH (rule, node, &subtable->rules_list) {
             if (rule->priority == target->priority
-                && !rule->to_be_removed
                 && miniflow_equal_in_minimask(&target->match.flow,
-                                              &rule->match.flow, &mask)) {
+                                              &rule->match.flow, &mask)
+                && cls_match_visible_in_version(rule->cls_match,
+                                                target->version)) {
                 return true;
             }
         }
@@ -1425,16 +1475,17 @@ cls_rule_is_loose_match(const struct cls_rule *rule,
 
 /* Iteration. */
 
+/* Rule may only match a target if it is visible in target's version.  For NULL
+ * target we only return rules that are not invisible in any version. */
 static bool
 rule_matches(const struct cls_rule *rule, const struct cls_rule *target)
 {
-    /* Iterators never see rules that have been marked for removal.
-     * This allows them to be oblivious of duplicate rules. */
-    return (!rule->to_be_removed &&
-            (!target
-             || miniflow_equal_in_minimask(&rule->match.flow,
-                                           &target->match.flow,
-                                           &target->match.mask)));
+    /* Iterators never see duplicate rules with the same priority. */
+    return target
+        ? (miniflow_equal_in_minimask(&rule->match.flow, &target->match.flow,
+                                      &target->match.mask)
+           && cls_match_visible_in_version(rule->cls_match, target->version))
+        : !cls_match_is_eventually_invisible(rule->cls_match);
 }
 
 static const struct cls_rule *
@@ -1457,10 +1508,13 @@ search_subtable(const struct cls_subtable *subtable,
 /* Initializes 'cursor' for iterating through rules in 'cls', and returns the
  * first matching cls_rule via '*pnode', or NULL if there are no matches.
  *
- *     - If 'target' is null, the cursor will visit every rule in 'cls'.
+ *     - If 'target' is null, or if the 'target' is a catchall target and the
+ *       target's version is CLS_NO_VERSION, the cursor will visit every rule
+ *       in 'cls' that is not invisible in any version.
  *
  *     - If 'target' is nonnull, the cursor will visit each 'rule' in 'cls'
- *       such that cls_rule_is_loose_match(rule, target) returns true.
+ *       such that cls_rule_is_loose_match(rule, target) returns true and that
+ *       the rule is visible in 'target->version'.
  *
  * Ignores target->priority. */
 struct cls_cursor
@@ -1470,7 +1524,9 @@ cls_cursor_start(const struct classifier *cls, const struct cls_rule *target)
     struct cls_subtable *subtable;
 
     cursor.cls = cls;
-    cursor.target = target && !cls_rule_is_catchall(target) ? target : NULL;
+    cursor.target = target && (!cls_rule_is_catchall(target)
+                               || target->version != CLS_MAX_VERSION)
+        ? target : NULL;
     cursor.rule = NULL;
 
     /* Find first rule. */
@@ -1722,8 +1778,8 @@ miniflow_and_mask_matches_flow(const struct miniflow *flow,
 }
 
 static inline const struct cls_match *
-find_match(const struct cls_subtable *subtable, const struct flow *flow,
-           uint32_t hash)
+find_match(const struct cls_subtable *subtable, long long version,
+           const struct flow *flow, uint32_t hash)
 {
     const struct cls_match *head, *rule;
 
@@ -1733,7 +1789,7 @@ find_match(const struct cls_subtable *subtable, const struct flow *flow,
                                                       flow))) {
             /* Return highest priority rule that is visible. */
             FOR_EACH_RULE_IN_LIST(rule, head) {
-                if (OVS_LIKELY(rule->visible)) {
+                if (OVS_LIKELY(cls_match_visible_in_version(rule, version))) {
                     return rule;
                 }
             }
@@ -1791,9 +1847,9 @@ fill_range_wc(const struct cls_subtable *subtable, struct flow_wildcards *wc,
 }
 
 static const struct cls_match *
-find_match_wc(const struct cls_subtable *subtable, const struct flow *flow,
-              struct trie_ctx trie_ctx[CLS_MAX_TRIES], unsigned int n_tries,
-              struct flow_wildcards *wc)
+find_match_wc(const struct cls_subtable *subtable, long long version,
+              const struct flow *flow, struct trie_ctx trie_ctx[CLS_MAX_TRIES],
+              unsigned int n_tries, struct flow_wildcards *wc)
 {
     uint32_t basis = 0, hash;
     const struct cls_match *rule = NULL;
@@ -1801,7 +1857,7 @@ find_match_wc(const struct cls_subtable *subtable, const struct flow *flow,
     struct range ofs;
 
     if (OVS_UNLIKELY(!wc)) {
-        return find_match(subtable, flow,
+        return find_match(subtable, version, flow,
                           flow_hash_in_minimask(flow, &subtable->mask, 0));
     }
 
@@ -1842,7 +1898,8 @@ find_match_wc(const struct cls_subtable *subtable, const struct flow *flow,
                                                   flow, wc)) {
                 /* Return highest priority rule that is visible. */
                 FOR_EACH_RULE_IN_LIST(rule, head) {
-                    if (OVS_LIKELY(rule->visible)) {
+                    if (OVS_LIKELY(cls_match_visible_in_version(rule,
+                                                                version))) {
                         return rule;
                     }
                 }
@@ -1859,7 +1916,7 @@ find_match_wc(const struct cls_subtable *subtable, const struct flow *flow,
     }
     hash = flow_hash_in_minimask_range(flow, &subtable->mask, ofs.start,
                                        ofs.end, &basis);
-    rule = find_match(subtable, flow, hash);
+    rule = find_match(subtable, version, flow, hash);
     if (!rule && subtable->ports_mask_len) {
         /* Ports are always part of the final range, if any.
          * No match was found for the ports.  Use the ports trie to figure out
diff --git a/lib/classifier.h b/lib/classifier.h
index d69c20191..cb0030abd 100644
--- a/lib/classifier.h
+++ b/lib/classifier.h
@@ -210,46 +210,98 @@
  * Each eliminated subtable lookup also reduces the amount of un-wildcarding.
  *
  *
- * Tentative Modifications
- * =======================
- *
- * When a new rule is added to a classifier, it can optionally be "invisible".
- * That means that lookups won't find the rule, although iterations through
- * the classifier will see it.
- *
- * Similarly, deletions from a classifier can be "tentative", by setting
- * 'to_be_removed' to true within the rule.  A rule that is tentatively deleted
- * will not appear in iterations, although it will still be found by lookups.
+ * Classifier Versioning
+ * =====================
+ *
+ * Classifier lookups are always done in a specific classifier version, where
+ * a version is defined to be a natural number.
+ *
+ * When a new rule is added to a classifier, it is set to become visible in a
+ * specific version.  If the version number used at insert time is larger than
+ * any version number currently used in lookups, the new rule is said to be
+ * invisible to lookups.  This means that lookups won't find the rule, but the
+ * rule is immediately available to classifier iterations.
+ *
+ * Similarly, a rule can be marked as to be deleted in a future version, or
+ * more precisely, to be visible upto a given version number.  To delete a rule
+ * in a way to not remove the rule before all ongoing lookups are finished, the
+ * rule should be marked as "to be deleted" by setting the rule's visibility to
+ * the negation of the last version number in which it should be visible.
+ * Then, when all the lookups use a later version number, the rule can be
+ * actually deleted from the classifier.  A rule that is marked for deletion
+ * after a future version will not appear in iterations, although it will still
+ * be found by lookups using a lookup version number up to that future version
+ * number.
  *
  * Classifiers can hold duplicate rules (rules with the same match criteria and
- * priority) when tentative modifications are involved: one (or more) identical
- * tentatively deleted rules can coexist in a classifier with at most one
- * identical invisible rule.
- *
- * The classifier supports tentative modifications for two reasons:
- *
- *     1. Performance: Adding (or deleting) a rule can, in pathological cases,
- *        have a cost proportional to the number of rules already in the
- *        classifier.  When multiple rules are being added (or deleted) in one
- *        go, though, this cost can be paid just once, not once per addition
- *        (or deletion), as long as it is OK for any new rules to be invisible
- *        until the batch change is complete.
- *
- *     2. Staging additions and deletions: Invisibility allows a rule to be
- *        added tentatively, to possibly be modified or removed before it
- *        becomes visible.  Tentatively deletion allows a rule to be scheduled
- *        for deletion before it is certain that the deletion is desirable.
+ * priority) when at most one of the duplicates with the same priority is
+ * visible in any given lookup version.  The caller responsible for classifier
+ * modifications must maintain this invariant.
+ *
+ * The classifier supports versioning for two reasons:
+ *
+ *     1. Support for versioned modifications makes it possible to perform an
+ *        arbitraty series of classifier changes as one atomic transaction,
+ *        where intermediate versions of the classifier are not visible to any
+ *        lookups.  Also, when a rule is added for a future version, or marked
+ *        for removal after the current version, such modifications can be
+ *        reverted without any visible effects to any of the current lookups.
+ *
+ *     2. Performance: Adding (or deleting) a large set of rules can, in
+ *        pathological cases, have a cost proportional to the number of rules
+ *        already in the classifier.  When multiple rules are being added (or
+ *        deleted) in one go, though, this pathological case cost can be
+ *        typically avoided, as long as it is OK for any new rules to be
+ *        invisible until the batch change is complete.
+ *
+ * Note that the classifier_replace() function replaces a rule immediately, and
+ * is therefore not safe to use with versioning.  It is still available for the
+ * users that do not use versioning.
+ *
+ *
+ * Deferred Publication
+ * ====================
+ *
+ * Removing large number of rules from classifier can be costly, as the
+ * supporting data structures are teared down, in many cases just to be
+ * re-instantiated right after.  In the worst case, as when each rule has a
+ * different match pattern (mask), the maintenance of the match patterns can
+ * have cost O(N^2), where N is the number of different match patterns.  To
+ * alleviate this, the classifier supports a "deferred mode", in which changes
+ * in internal data structures needed for future version lookups may not be
+ * fully computed yet.  The computation is finalized when the deferred mode is
+ * turned off.
+ *
+ * This feature can be used with versioning such that all changes to future
+ * versions are made in the deferred mode.  Then, right before making the new
+ * version visible to lookups, the deferred mode is turned off so that all the
+ * data structures are ready for lookups with the new version number.
  *
  * To use deferred publication, first call classifier_defer().  Then, modify
- * the classifier via additions and deletions.  Call cls_rule_make_visible() on
- * each new rule at an appropriate time.  Finally, call classifier_publish().
+ * the classifier via additions (classifier_insert() with a specific, future
+ * version number) and deletions (use cls_rule_make_removable_after_version()).
+ * Then call classifier_publish(), and after that, announce the new version
+ * number to be used in lookups.
  *
  *
  * Thread-safety
  * =============
  *
- * The classifier may safely be accessed by many reader threads concurrently or
- * by a single writer. */
+ * The classifier may safely be accessed by many reader threads concurrently
+ * and by a single writer, or by multiple writers when they guarantee mutually
+ * exlucive access to classifier modifications.
+ *
+ * Since the classifier rules are RCU protected, the rule destruction after
+ * removal from the classifier must be RCU postponed.  Also, when versioning is
+ * used, the rule removal itself needs to be typically RCU postponed.  In this
+ * case the rule destruction is doubly RCU postponed, i.e., the second
+ * ovsrcu_postpone() call to destruct the rule is called from the first RCU
+ * callback that removes the rule.
+ *
+ * Rules that have never been visible to lookups are an exeption to the above
+ * rule.  Such rules can be removed immediately, but their destruction must
+ * still be RCU postponed, as the rule's visibility attribute may be examined
+ * parallel to the rule's removal. */
 
 #include "cmap.h"
 #include "match.h"
@@ -275,6 +327,8 @@ struct cls_trie {
 };
 
 enum {
+    CLS_MIN_VERSION = 1,   /* Default version number to use. */
+    CLS_MAX_VERSION = LLONG_MAX, /* Last possible version number. */
     CLS_MAX_INDICES = 3,   /* Maximum number of lookup indices per subtable. */
     CLS_MAX_TRIES = 3      /* Maximum number of prefix trees per classifier. */
 };
@@ -301,22 +355,17 @@ struct cls_conjunction {
 
 /* A rule to be inserted to the classifier. */
 struct cls_rule {
-    struct rculist node;         /* In struct cls_subtable 'rules_list'. */
-    int priority;                /* Larger numbers are higher priorities. */
-    bool to_be_removed;          /* Rule will be deleted.
-                                  * This is the only field that may be
-                                  * modified after the rule has been added to
-                                  * a classifier.  Modifications are to be
-                                  * done only under same locking as all other
-                                  * classifier modifications.  This field may
-                                  * not be examined by lookups. */
-    struct cls_match *cls_match; /* NULL if not in a classifier. */
-    struct minimatch match;      /* Matching rule. */
+    struct rculist node;          /* In struct cls_subtable 'rules_list'. */
+    const int priority;           /* Larger numbers are higher priorities. */
+    const long long version;      /* Version in which the rule was added. */
+    struct cls_match *cls_match;  /* NULL if not in a classifier. */
+    const struct minimatch match; /* Matching rule. */
 };
 
-void cls_rule_init(struct cls_rule *, const struct match *, int priority);
+void cls_rule_init(struct cls_rule *, const struct match *, int priority,
+                   long long version);
 void cls_rule_init_from_minimatch(struct cls_rule *, const struct minimatch *,
-                                  int priority);
+                                  int priority, long long version);
 void cls_rule_clone(struct cls_rule *, const struct cls_rule *);
 void cls_rule_move(struct cls_rule *dst, struct cls_rule *src);
 void cls_rule_destroy(struct cls_rule *);
@@ -330,7 +379,11 @@ void cls_rule_format(const struct cls_rule *, struct ds *);
 bool cls_rule_is_catchall(const struct cls_rule *);
 bool cls_rule_is_loose_match(const struct cls_rule *rule,
                              const struct minimatch *criteria);
-void cls_rule_make_visible(const struct cls_rule *rule);
+bool cls_rule_visible_in_version(const struct cls_rule *, long long version);
+void cls_rule_make_invisible_in_version(const struct cls_rule *,
+                                        long long version,
+                                        long long lookup_version);
+void cls_rule_restore_visibility(const struct cls_rule *);
 
 /* Constructor/destructor.  Must run single-threaded. */
 void classifier_init(struct classifier *, const uint8_t *flow_segments);
@@ -354,7 +407,7 @@ static inline void classifier_publish(struct classifier *);
 /* Lookups.  These are RCU protected and may run concurrently with modifiers
  * and each other. */
 const struct cls_rule *classifier_lookup(const struct classifier *,
-                                         struct flow *,
+                                         long long version, struct flow *,
                                          struct flow_wildcards *);
 bool classifier_rule_overlaps(const struct classifier *,
                               const struct cls_rule *);
@@ -362,7 +415,8 @@ const struct cls_rule *classifier_find_rule_exactly(const struct classifier *,
                                                     const struct cls_rule *);
 const struct cls_rule *classifier_find_match_exactly(const struct classifier *,
                                                      const struct match *,
-                                                     int priority);
+                                                     int priority,
+                                                     long long version);
 bool classifier_is_empty(const struct classifier *);
 int classifier_count(const struct classifier *);
 
diff --git a/lib/ovs-router.c b/lib/ovs-router.c
index bf205d6b4..532487e8f 100644
--- a/lib/ovs-router.c
+++ b/lib/ovs-router.c
@@ -68,7 +68,7 @@ ovs_router_lookup(ovs_be32 ip_dst, char output_bridge[], ovs_be32 *gw)
     const struct cls_rule *cr;
     struct flow flow = {.nw_dst = ip_dst};
 
-    cr = classifier_lookup(&cls, &flow, NULL);
+    cr = classifier_lookup(&cls, CLS_MAX_VERSION, &flow, NULL);
     if (cr) {
         struct ovs_router_entry *p = ovs_router_entry_cast(cr);
 
@@ -115,7 +115,8 @@ ovs_router_insert__(uint8_t priority, ovs_be32 ip_dst, uint8_t plen,
     p->nw_addr = match.flow.nw_dst;
     p->plen = plen;
     p->priority = priority;
-    cls_rule_init(&p->cr, &match, priority); /* Longest prefix matches first. */
+    /* Longest prefix matches first. */
+    cls_rule_init(&p->cr, &match, priority, CLS_MIN_VERSION);
 
     ovs_mutex_lock(&mutex);
     cr = classifier_replace(&cls, &p->cr, NULL, 0);
@@ -144,7 +145,7 @@ rt_entry_delete(uint8_t priority, ovs_be32 ip_dst, uint8_t plen)
 
     rt_init_match(&match, ip_dst, plen);
 
-    cls_rule_init(&rule, &match, priority);
+    cls_rule_init(&rule, &match, priority, CLS_MIN_VERSION);
 
     /* Find the exact rule. */
     cr = classifier_find_rule_exactly(&cls, &rule);
diff --git a/lib/tnl-ports.c b/lib/tnl-ports.c
index 759d6ba33..2602db543 100644
--- a/lib/tnl-ports.c
+++ b/lib/tnl-ports.c
@@ -84,7 +84,7 @@ tnl_port_map_insert(odp_port_t port, ovs_be32 ip_dst, ovs_be16 udp_port,
 
     ovs_mutex_lock(&mutex);
     do {
-        cr = classifier_lookup(&cls, &match.flow, NULL);
+        cr = classifier_lookup(&cls, CLS_MAX_VERSION, &match.flow, NULL);
         p = tnl_port_cast(cr);
         /* Try again if the rule was released before we get the reference. */
     } while (p && !ovs_refcount_try_ref_rcu(&p->ref_cnt));
@@ -99,7 +99,7 @@ tnl_port_map_insert(odp_port_t port, ovs_be32 ip_dst, ovs_be16 udp_port,
         match.wc.masks.tp_dst = OVS_BE16_MAX;
         match.wc.masks.nw_src = OVS_BE32_MAX;
 
-        cls_rule_init(&p->cr, &match, 0);   /* Priority == 0. */
+        cls_rule_init(&p->cr, &match, 0, CLS_MIN_VERSION); /* Priority == 0. */
         ovs_refcount_init(&p->ref_cnt);
         ovs_strlcpy(p->dev_name, dev_name, sizeof p->dev_name);
 
@@ -130,7 +130,7 @@ tnl_port_map_delete(ovs_be32 ip_dst, ovs_be16 udp_port)
 
     tnl_port_init_flow(&flow, ip_dst, udp_port);
 
-    cr = classifier_lookup(&cls, &flow, NULL);
+    cr = classifier_lookup(&cls, CLS_MAX_VERSION, &flow, NULL);
     tnl_port_unref(cr);
 }
 
@@ -139,7 +139,8 @@ tnl_port_map_delete(ovs_be32 ip_dst, ovs_be16 udp_port)
 odp_port_t
 tnl_port_map_lookup(struct flow *flow, struct flow_wildcards *wc)
 {
-    const struct cls_rule *cr = classifier_lookup(&cls, flow, wc);
+    const struct cls_rule *cr = classifier_lookup(&cls, CLS_MAX_VERSION, flow,
+                                                  wc);
 
     return (cr) ? tnl_port_cast(cr)->portno : ODPP_NONE;
 }
diff --git a/ofproto/ofproto-dpif.c b/ofproto/ofproto-dpif.c
index c4cafe00b..81beca049 100644
--- a/ofproto/ofproto-dpif.c
+++ b/ofproto/ofproto-dpif.c
@@ -3725,7 +3725,7 @@ rule_dpif_lookup_in_table(struct ofproto_dpif *ofproto, uint8_t table_id,
     struct rule_dpif *rule;
 
     do {
-        cls_rule = classifier_lookup(cls, flow, wc);
+        cls_rule = classifier_lookup(cls, CLS_MAX_VERSION, flow, wc);
 
         rule = rule_dpif_cast(rule_from_cls_rule(cls_rule));
 
diff --git a/ofproto/ofproto.c b/ofproto/ofproto.c
index f2e9557c6..b5424b9f4 100644
--- a/ofproto/ofproto.c
+++ b/ofproto/ofproto.c
@@ -153,6 +153,7 @@ struct rule_criteria {
 
 static void rule_criteria_init(struct rule_criteria *, uint8_t table_id,
                                const struct match *match, int priority,
+                               long long version,
                                ovs_be64 cookie, ovs_be64 cookie_mask,
                                ofp_port_t out_port, uint32_t out_group);
 static void rule_criteria_require_rw(struct rule_criteria *,
@@ -2043,7 +2044,8 @@ ofproto_add_flow(struct ofproto *ofproto, const struct match *match,
     /* First do a cheap check whether the rule we're looking for already exists
      * with the actions that we want.  If it does, then we're done. */
     rule = rule_from_cls_rule(classifier_find_match_exactly(
-                                  &ofproto->tables[0].cls, match, priority));
+                                  &ofproto->tables[0].cls, match, priority,
+                                  CLS_MAX_VERSION));
     if (rule) {
         const struct rule_actions *actions = rule_get_actions(rule);
         must_add = !ofpacts_equal(actions->ofpacts, actions->ofpacts_len,
@@ -2080,9 +2082,9 @@ ofproto_flow_mod(struct ofproto *ofproto, struct ofputil_flow_mod *fm)
         struct rule *rule;
         bool done = false;
 
-        rule = rule_from_cls_rule(classifier_find_match_exactly(&table->cls,
-                                                                &fm->match,
-                                                                fm->priority));
+        rule = rule_from_cls_rule(classifier_find_match_exactly(
+                                      &table->cls, &fm->match,
+                                      fm->priority, CLS_MAX_VERSION));
         if (rule) {
             /* Reading many of the rule fields and writing on 'modified'
              * requires the rule->mutex.  Also, rule->actions may change
@@ -2129,7 +2131,8 @@ ofproto_delete_flow(struct ofproto *ofproto,
     /* First do a cheap check whether the rule we're looking for has already
      * been deleted.  If so, then we're done. */
     rule = rule_from_cls_rule(classifier_find_match_exactly(cls, target,
-                                                            priority));
+                                                            priority,
+                                                            CLS_MAX_VERSION));
     if (!rule) {
         return;
     }
@@ -3038,7 +3041,7 @@ learned_cookies_flush(struct ofproto *ofproto, struct ovs_list *dead_cookies)
         struct match match;
 
         match_init_catchall(&match);
-        rule_criteria_init(&criteria, c->table_id, &match, 0,
+        rule_criteria_init(&criteria, c->table_id, &match, 0, CLS_MAX_VERSION,
                            c->cookie, OVS_BE64_MAX, OFPP_ANY, OFPG_ANY);
         rule_criteria_require_rw(&criteria, false);
         collect_rules_loose(ofproto, &criteria, &rules);
@@ -3676,12 +3679,12 @@ next_matching_table(const struct ofproto *ofproto,
  * supplied as 0. */
 static void
 rule_criteria_init(struct rule_criteria *criteria, uint8_t table_id,
-                   const struct match *match, int priority,
+                   const struct match *match, int priority, long long version,
                    ovs_be64 cookie, ovs_be64 cookie_mask,
                    ofp_port_t out_port, uint32_t out_group)
 {
     criteria->table_id = table_id;
-    cls_rule_init(&criteria->cr, match, priority);
+    cls_rule_init(&criteria->cr, match, priority, version);
     criteria->cookie = cookie;
     criteria->cookie_mask = cookie_mask;
     criteria->out_port = out_port;
@@ -3785,7 +3788,7 @@ rule_collection_destroy(struct rule_collection *rules)
  * function verifies most of the criteria in 'c' itself, but the caller must
  * check 'c->cr' itself.
  *
- * Rules that have already been marked as 'to_be_removed' are not collected.
+ * Rules that have already been marked for removal are not collected.
  *
  * Increments '*n_readonly' if 'rule' wasn't added because it's read-only (and
  * 'c' only includes modifiable rules). */
@@ -3799,7 +3802,7 @@ collect_rule(struct rule *rule, const struct rule_criteria *c,
         && ofproto_rule_has_out_group(rule, c->out_group)
         && !((rule->flow_cookie ^ c->cookie) & c->cookie_mask)
         && (!rule_is_hidden(rule) || c->include_hidden)
-        && !rule->cr.to_be_removed) {
+        && cls_rule_visible_in_version(&rule->cr, c->cr.version)) {
         /* Rule matches all the criteria... */
         if (!rule_is_readonly(rule) || c->include_readonly) {
             /* ...add it. */
@@ -3951,8 +3954,9 @@ handle_flow_stats_request(struct ofconn *ofconn,
         return error;
     }
 
-    rule_criteria_init(&criteria, fsr.table_id, &fsr.match, 0, fsr.cookie,
-                       fsr.cookie_mask, fsr.out_port, fsr.out_group);
+    rule_criteria_init(&criteria, fsr.table_id, &fsr.match, 0, CLS_MAX_VERSION,
+                       fsr.cookie, fsr.cookie_mask, fsr.out_port,
+                       fsr.out_group);
 
     ovs_mutex_lock(&ofproto_mutex);
     error = collect_rules_loose(ofproto, &criteria, &rules);
@@ -4115,7 +4119,7 @@ handle_aggregate_stats_request(struct ofconn *ofconn,
     }
 
     rule_criteria_init(&criteria, request.table_id, &request.match, 0,
-                       request.cookie, request.cookie_mask,
+                       CLS_MAX_VERSION, request.cookie, request.cookie_mask,
                        request.out_port, request.out_group);
 
     ovs_mutex_lock(&ofproto_mutex);
@@ -4404,10 +4408,10 @@ add_flow_start(struct ofproto *ofproto, struct ofputil_flow_mod *fm,
         return OFPERR_OFPBRC_EPERM;
     }
 
-    cls_rule_init(&cr, &fm->match, fm->priority);
+    cls_rule_init(&cr, &fm->match, fm->priority, CLS_MIN_VERSION);
 
     /* Check for the existence of an identical rule.
-     * This will not return rules earlier marked as 'to_be_removed'. */
+     * This will not return rules earlier marked for removal. */
     rule = rule_from_cls_rule(classifier_find_rule_exactly(&table->cls, &cr));
     if (rule) {
         /* Transform "add" into "modify" of an existing identical flow. */
@@ -4506,7 +4510,6 @@ add_flow_finish(struct ofproto *ofproto, struct ofputil_flow_mod *fm,
     } else {
         struct oftable *table = &ofproto->tables[rule->table_id];
 
-        cls_rule_make_visible(&rule->cr);
         classifier_publish(&table->cls);
 
         learned_cookies_inc(ofproto, rule_get_actions(rule));
@@ -4573,7 +4576,7 @@ modify_flows_check__(struct ofproto *ofproto, struct ofputil_flow_mod *fm,
     return 0;
 }
 
-/* Modifies the 'rule', changing them to match 'fm'. */
+/* Modifies the 'rule', changing it to match 'fm'. */
 static void
 modify_flow__(struct ofproto *ofproto, struct ofputil_flow_mod *fm,
               const struct flow_mod_requester *req, struct rule *rule,
@@ -4742,7 +4745,7 @@ modify_flows_start_loose(struct ofproto *ofproto, struct ofputil_flow_mod *fm,
     struct rule_criteria criteria;
     enum ofperr error;
 
-    rule_criteria_init(&criteria, fm->table_id, &fm->match, 0,
+    rule_criteria_init(&criteria, fm->table_id, &fm->match, 0, CLS_MAX_VERSION,
                        fm->cookie, fm->cookie_mask, OFPP_ANY, OFPG11_ANY);
     rule_criteria_require_rw(&criteria,
                              (fm->flags & OFPUTIL_FF_NO_READONLY) != 0);
@@ -4796,7 +4799,8 @@ modify_flow_start_strict(struct ofproto *ofproto, struct ofputil_flow_mod *fm,
     enum ofperr error;
 
     rule_criteria_init(&criteria, fm->table_id, &fm->match, fm->priority,
-                       fm->cookie, fm->cookie_mask, OFPP_ANY, OFPG11_ANY);
+                       CLS_MAX_VERSION, fm->cookie, fm->cookie_mask, OFPP_ANY,
+                       OFPG11_ANY);
     rule_criteria_require_rw(&criteria,
                              (fm->flags & OFPUTIL_FF_NO_READONLY) != 0);
     error = collect_rules_strict(ofproto, &criteria, rules);
@@ -4826,11 +4830,12 @@ delete_flows__(const struct rule_collection *rules,
         struct ovs_list dead_cookies = OVS_LIST_INITIALIZER(&dead_cookies);
         struct ofproto *ofproto = rules->rules[0]->ofproto;
         struct rule *rule, *next;
+        uint8_t prev_table = UINT8_MAX;
         size_t i;
 
         for (i = 0, next = rules->rules[0];
              rule = next, next = (++i < rules->n) ? rules->rules[i] : NULL,
-                 rule; ) {
+                 rule; prev_table = rule->table_id) {
             struct classifier *cls = &ofproto->tables[rule->table_id].cls;
             uint8_t next_table = next ? next->table_id : UINT8_MAX;
 
@@ -4840,7 +4845,8 @@ delete_flows__(const struct rule_collection *rules,
                              req ? req->ofconn : NULL,
                              req ? req->request->xid : 0, NULL);
 
-            if (next_table == rule->table_id) {
+            /* Defer once for each new table. */
+            if (rule->table_id != prev_table) {
                 classifier_defer(cls);
             }
             if (!classifier_remove(cls, &rule->cr)) {
@@ -4873,7 +4879,7 @@ delete_flows_start_loose(struct ofproto *ofproto,
     struct rule_criteria criteria;
     enum ofperr error;
 
-    rule_criteria_init(&criteria, fm->table_id, &fm->match, 0,
+    rule_criteria_init(&criteria, fm->table_id, &fm->match, 0, CLS_MAX_VERSION,
                        fm->cookie, fm->cookie_mask,
                        fm->out_port, fm->out_group);
     rule_criteria_require_rw(&criteria,
@@ -4885,7 +4891,10 @@ delete_flows_start_loose(struct ofproto *ofproto,
         for (size_t i = 0; i < rules->n; i++) {
             struct rule *rule = rules->rules[i];
 
-            CONST_CAST(struct cls_rule *, &rule->cr)->to_be_removed = true;
+            cls_rule_make_invisible_in_version(CONST_CAST(struct cls_rule *,
+                                                          &rule->cr),
+                                               CLS_MIN_VERSION,
+                                               CLS_MIN_VERSION);
         }
     }
 
@@ -4897,9 +4906,7 @@ delete_flows_revert(struct rule_collection *rules)
     OVS_REQUIRES(ofproto_mutex)
 {
     for (size_t i = 0; i < rules->n; i++) {
-        struct rule *rule = rules->rules[i];
-
-        CONST_CAST(struct cls_rule *, &rule->cr)->to_be_removed = false;
+        cls_rule_restore_visibility(&rules->rules[i]->cr);
     }
     rule_collection_destroy(rules);
 }
@@ -4925,7 +4932,7 @@ delete_flow_start_strict(struct ofproto *ofproto,
     enum ofperr error;
 
     rule_criteria_init(&criteria, fm->table_id, &fm->match, fm->priority,
-                       fm->cookie, fm->cookie_mask,
+                       CLS_MAX_VERSION, fm->cookie, fm->cookie_mask,
                        fm->out_port, fm->out_group);
     rule_criteria_require_rw(&criteria,
                              (fm->flags & OFPUTIL_FF_NO_READONLY) != 0);
@@ -4936,7 +4943,10 @@ delete_flow_start_strict(struct ofproto *ofproto,
         for (size_t i = 0; i < rules->n; i++) {
             struct rule *rule = rules->rules[i];
 
-            CONST_CAST(struct cls_rule *, &rule->cr)->to_be_removed = true;
+            cls_rule_make_invisible_in_version(CONST_CAST(struct cls_rule *,
+                                                          &rule->cr),
+                                               CLS_MIN_VERSION,
+                                               CLS_MIN_VERSION);
         }
     }
 
@@ -5340,7 +5350,7 @@ ofproto_collect_ofmonitor_refresh_rules(const struct ofmonitor *m,
     const struct oftable *table;
     struct cls_rule target;
 
-    cls_rule_init_from_minimatch(&target, &m->match, 0);
+    cls_rule_init_from_minimatch(&target, &m->match, 0, CLS_MAX_VERSION);
     FOR_EACH_MATCHING_TABLE (table, m->table_id, ofproto) {
         struct rule *rule;
 
@@ -5877,8 +5887,8 @@ group_get_ref_count(struct ofgroup *group)
     uint32_t count;
 
     match_init_catchall(&match);
-    rule_criteria_init(&criteria, 0xff, &match, 0, htonll(0), htonll(0),
-                       OFPP_ANY, group->group_id);
+    rule_criteria_init(&criteria, 0xff, &match, 0, CLS_MAX_VERSION, htonll(0),
+                       htonll(0), OFPP_ANY, group->group_id);
     ovs_mutex_lock(&ofproto_mutex);
     error = collect_rules_loose(ofproto, &criteria, &rules);
     ovs_mutex_unlock(&ofproto_mutex);
@@ -6518,26 +6528,6 @@ do_bundle_flow_mod_finish(struct ofproto *ofproto, struct ofputil_flow_mod *fm,
     }
 }
 
-/* Commit phases (all while locking ofproto_mutex):
- *
- * 1. Gather resources - do not send any events or notifications.
- *
- * add: Check conflicts, check for a displaced flow. If no displaced flow
- *      exists, add the new flow, but mark it as "invisible".
- * mod: Collect affected flows, Do not modify yet.
- * del: Collect affected flows, Do not delete yet.
- *
- * 2a. Fail if any errors are found.  After this point no errors are possible.
- * No visible changes were made, so rollback is minimal (remove added invisible
- * flows, revert 'to_be_removed' status of flows).
- *
- * 2b. Commit the changes
- *
- * add: if have displaced flow, modify it, otherwise mark the new flow as
- *      "visible".
- * mod: Modify the collected flows.
- * del: Delete the collected flows.
- */
 static enum ofperr
 do_bundle_commit(struct ofconn *ofconn, uint32_t id, uint16_t flags)
 {
@@ -7434,7 +7424,7 @@ ofproto_get_vlan_usage(struct ofproto *ofproto, unsigned long int *vlan_bitmap)
 
     match_init_catchall(&match);
     match_set_vlan_vid_masked(&match, htons(VLAN_CFI), htons(VLAN_CFI));
-    cls_rule_init(&target, &match, 0);
+    cls_rule_init(&target, &match, 0, CLS_MAX_VERSION);
 
     free(ofproto->vlan_bitmap);
     ofproto->vlan_bitmap = bitmap_allocate(4096);
diff --git a/tests/test-classifier.c b/tests/test-classifier.c
index a615438c0..24fc5eb11 100644
--- a/tests/test-classifier.c
+++ b/tests/test-classifier.c
@@ -433,7 +433,7 @@ compare_classifiers(struct classifier *cls, struct tcls *tcls)
         /* This assertion is here to suppress a GCC 4.9 array-bounds warning */
         ovs_assert(cls->n_tries <= CLS_MAX_TRIES);
 
-        cr0 = classifier_lookup(cls, &flow, &wc);
+        cr0 = classifier_lookup(cls, CLS_MAX_VERSION, &flow, &wc);
         cr1 = tcls_lookup(tcls, &flow);
         assert((cr0 == NULL) == (cr1 == NULL));
         if (cr0 != NULL) {
@@ -443,7 +443,7 @@ compare_classifiers(struct classifier *cls, struct tcls *tcls)
             assert(cls_rule_equal(cr0, cr1));
             assert(tr0->aux == tr1->aux);
         }
-        cr2 = classifier_lookup(cls, &flow, NULL);
+        cr2 = classifier_lookup(cls, CLS_MAX_VERSION, &flow, NULL);
         assert(cr2 == cr0);
     }
 }
@@ -635,7 +635,7 @@ make_rule(int wc_fields, int priority, int value_pat)
     rule = xzalloc(sizeof *rule);
     cls_rule_init(&rule->cls_rule, &match, wc_fields
                   ? (priority == INT_MIN ? priority + 1 : priority)
-                  : INT_MAX);
+                  : INT_MAX, CLS_MIN_VERSION);
     return rule;
 }
 
diff --git a/utilities/ovs-ofctl.c b/utilities/ovs-ofctl.c
index 10d44e0ef..812ce7fff 100644
--- a/utilities/ovs-ofctl.c
+++ b/utilities/ovs-ofctl.c
@@ -2457,7 +2457,7 @@ fte_insert(struct classifier *cls, const struct match *match,
     struct fte *old, *fte;
 
     fte = xzalloc(sizeof *fte);
-    cls_rule_init(&fte->rule, match, priority);
+    cls_rule_init(&fte->rule, match, priority, CLS_MIN_VERSION);
     fte->versions[index] = version;
 
     old = fte_from_cls_rule(classifier_replace(cls, &fte->rule, NULL, 0));
@@ -2467,7 +2467,6 @@ fte_insert(struct classifier *cls, const struct match *match,
 
         ovsrcu_postpone(fte_free, old);
     }
-    cls_rule_make_visible(&fte->rule);
 }
 
 /* Reads the flows in 'filename' as flow table entries in 'cls' for the version