diff --git a/lib/conntrack-private.h b/lib/conntrack-private.h
index a9546ee9c..fd501d574 100644
--- a/lib/conntrack-private.h
+++ b/lib/conntrack-private.h
@@ -95,6 +95,7 @@ struct alg_exp_node {
 struct conn_key_node {
     enum key_dir dir;
     struct conn_key key;
+    uint32_t key_hash;
     struct cmap_node cm_node;
 };
 
@@ -102,7 +103,6 @@ struct conn {
     /* Immutable data. */
     struct conn_key_node key_node[CT_DIR_MAX];
     struct conn_key parent_key; /* Only used for orig_tuple support. */
-    struct ovs_list exp_node;
 
     uint16_t nat_action;
     char *alg;
@@ -121,7 +121,9 @@ struct conn {
     /* Mutable data. */
     bool seq_skew_dir; /* TCP sequence skew direction due to NATTing of FTP
                         * control messages; true if reply direction. */
-    bool cleaned; /* True if cleaned from expiry lists. */
+    atomic_flag cleaned; /* True if the entry was stale and one of the
+                          * cleaner (i.e. packet path or sweeper) took
+                          * charge of it. */
 
     /* Immutable data. */
     bool alg_related; /* True if alg data connection. */
@@ -192,10 +194,25 @@ enum ct_timeout {
     N_CT_TM
 };
 
-struct conntrack {
-    struct ovs_mutex ct_lock; /* Protects 2 following fields. */
+#define CONNTRACK_BUCKETS_SHIFT 9
+#define CONNTRACK_BUCKETS (1 << CONNTRACK_BUCKETS_SHIFT)
+
+struct ct_bucket {
+    /* Protects 'conns'. In case of natted conns, there's a high
+     * chance that the forward and the reverse key stand in different
+     * buckets. buckets_lock() should be the preferred way to acquire
+     * these locks (unless otherwise needed), as it deals with the
+     * acquisition order. */
+    struct ovs_mutex lock;
+    /* Contains the connections in the bucket, indexed by
+     * 'struct conn_key'. */
     struct cmap conns OVS_GUARDED;
-    struct ovs_list exp_lists[N_CT_TM] OVS_GUARDED;
+};
+
+struct conntrack {
+    struct ct_bucket buckets[CONNTRACK_BUCKETS];
+    unsigned int next_bucket;
+    struct ovs_mutex ct_lock;
     struct cmap zone_limits OVS_GUARDED;
     struct cmap timeout_policies OVS_GUARDED;
     uint32_t hash_basis; /* Salt for hashing a connection key. */
@@ -220,9 +237,10 @@ struct conntrack {
 };
 
 /* Lock acquisition order:
- *    1. 'ct_lock'
- *    2. 'conn->lock'
- *    3. 'resources_lock'
+ *    1. 'buckets[p1]->lock'
+ *    2  'buckets[p2]->lock' (with p1 < p2)
+ *    3. 'conn->lock'
+ *    4. 'resources_lock'
  */
 
 extern struct ct_l4_proto ct_proto_tcp;
diff --git a/lib/conntrack-tp.c b/lib/conntrack-tp.c
index 9ecb06978..117810528 100644
--- a/lib/conntrack-tp.c
+++ b/lib/conntrack-tp.c
@@ -236,27 +236,6 @@ tm_to_ct_dpif_tp(enum ct_timeout tm)
     return CT_DPIF_TP_ATTR_MAX;
 }
 
-static void
-conn_update_expiration__(struct conntrack *ct, struct conn *conn,
-                         enum ct_timeout tm, long long now,
-                         uint32_t tp_value)
-    OVS_REQUIRES(conn->lock)
-{
-    ovs_mutex_unlock(&conn->lock);
-
-    ovs_mutex_lock(&ct->ct_lock);
-    ovs_mutex_lock(&conn->lock);
-    if (!conn->cleaned) {
-        conn->expiration = now + tp_value * 1000;
-        ovs_list_remove(&conn->exp_node);
-        ovs_list_push_back(&ct->exp_lists[tm], &conn->exp_node);
-    }
-    ovs_mutex_unlock(&conn->lock);
-    ovs_mutex_unlock(&ct->ct_lock);
-
-    ovs_mutex_lock(&conn->lock);
-}
-
 /* The conn entry lock must be held on entry and exit. */
 void
 conn_update_expiration(struct conntrack *ct, struct conn *conn,
@@ -266,42 +245,25 @@ conn_update_expiration(struct conntrack *ct, struct conn *conn,
     struct timeout_policy *tp;
     uint32_t val;
 
-    ovs_mutex_unlock(&conn->lock);
-
-    ovs_mutex_lock(&ct->ct_lock);
-    ovs_mutex_lock(&conn->lock);
     tp = timeout_policy_lookup(ct, conn->tp_id);
     if (tp) {
         val = tp->policy.attrs[tm_to_ct_dpif_tp(tm)];
     } else {
         val = ct_dpif_netdev_tp_def[tm_to_ct_dpif_tp(tm)];
     }
-    ovs_mutex_unlock(&conn->lock);
-    ovs_mutex_unlock(&ct->ct_lock);
 
-    ovs_mutex_lock(&conn->lock);
     VLOG_DBG_RL(&rl, "Update timeout %s zone=%u with policy id=%d "
                 "val=%u sec.",
                 ct_timeout_str[tm], conn->key_node[CT_DIR_FWD].key.zone,
                 conn->tp_id, val);
 
-    conn_update_expiration__(ct, conn, tm, now, val);
-}
-
-static void
-conn_init_expiration__(struct conntrack *ct, struct conn *conn,
-                       enum ct_timeout tm, long long now,
-                       uint32_t tp_value)
-{
-    conn->expiration = now + tp_value * 1000;
-    ovs_list_push_back(&ct->exp_lists[tm], &conn->exp_node);
+    conn->expiration = now + val * 1000;
 }
 
 /* ct_lock must be held. */
 void
 conn_init_expiration(struct conntrack *ct, struct conn *conn,
                      enum ct_timeout tm, long long now)
-    OVS_REQUIRES(ct->ct_lock)
 {
     struct timeout_policy *tp;
     uint32_t val;
@@ -317,5 +279,5 @@ conn_init_expiration(struct conntrack *ct, struct conn *conn,
                 ct_timeout_str[tm], conn->key_node[CT_DIR_FWD].key.zone,
                 conn->tp_id, val);
 
-    conn_init_expiration__(ct, conn, tm, now, val);
+    conn->expiration = now + val * 1000;
 }
diff --git a/lib/conntrack.c b/lib/conntrack.c
index 371a34b89..fb784d066 100644
--- a/lib/conntrack.c
+++ b/lib/conntrack.c
@@ -44,7 +44,6 @@
 VLOG_DEFINE_THIS_MODULE(conntrack);
 
 COVERAGE_DEFINE(conntrack_full);
-COVERAGE_DEFINE(conntrack_long_cleanup);
 COVERAGE_DEFINE(conntrack_l3csum_err);
 COVERAGE_DEFINE(conntrack_l4csum_err);
 COVERAGE_DEFINE(conntrack_lookup_natted_miss);
@@ -85,6 +84,8 @@ struct zone_limit {
     struct conntrack_zone_limit czl;
 };
 
+static unsigned hash_to_bucket(uint32_t hash);
+static void conn_clean(struct conntrack *ct, struct conn *conn);
 static bool conn_key_extract(struct conntrack *, struct dp_packet *,
                              ovs_be16 dl_type, struct conn_lookup_ctx *,
                              uint16_t zone);
@@ -109,8 +110,9 @@ static void set_label(struct dp_packet *, struct conn *,
 static void *clean_thread_main(void *f_);
 
 static bool
-nat_get_unique_tuple(struct conntrack *ct, struct conn *conn,
-                     const struct nat_action_info_t *nat_info);
+nat_get_unique_tuple_lock(struct conntrack *ct, struct conn *conn,
+                          const struct nat_action_info_t *nat_info,
+                          uint32_t *rev_hash);
 
 static uint8_t
 reverse_icmp_type(uint8_t type);
@@ -249,16 +251,17 @@ conntrack_init(void)
     ovs_rwlock_unlock(&ct->resources_lock);
 
     ovs_mutex_init_adaptive(&ct->ct_lock);
-    ovs_mutex_lock(&ct->ct_lock);
-    cmap_init(&ct->conns);
-    for (unsigned i = 0; i < ARRAY_SIZE(ct->exp_lists); i++) {
-        ovs_list_init(&ct->exp_lists[i]);
+
+    ct->next_bucket = 0;
+    for (unsigned i = 0; i < CONNTRACK_BUCKETS; i++) {
+        struct ct_bucket *bucket = &ct->buckets[i];
+        cmap_init(&bucket->conns);
+        ovs_mutex_init_adaptive(&bucket->lock);
     }
+
     cmap_init(&ct->zone_limits);
     ct->zone_limit_seq = 0;
     timeout_policy_init(ct);
-    ovs_mutex_unlock(&ct->ct_lock);
-
     atomic_count_init(&ct->n_conn, 0);
     atomic_init(&ct->n_conn_limit, DEFAULT_N_CONN_LIMIT);
     atomic_init(&ct->tcp_seq_chk, true);
@@ -410,9 +413,9 @@ zone_limit_delete(struct conntrack *ct, uint16_t zone)
 }
 
 static void
-conn_clean(struct conntrack *ct, struct conn *conn)
-    OVS_REQUIRES(ct->ct_lock)
+conn_clean__(struct conntrack *ct, struct conn *conn)
 {
+    struct ct_bucket *bucket;
     struct zone_limit *zl;
     uint32_t hash;
 
@@ -420,8 +423,9 @@ conn_clean(struct conntrack *ct, struct conn *conn)
         expectation_clean(ct, &conn->key_node[CT_DIR_FWD].key);
     }
 
-    hash = conn_key_hash(&conn->key_node[CT_DIR_FWD].key, ct->hash_basis);
-    cmap_remove(&ct->conns, &conn->key_node[CT_DIR_FWD].cm_node, hash);
+    hash = conn->key_node[CT_DIR_FWD].key_hash;
+    bucket = &ct->buckets[hash_to_bucket(hash)];
+    cmap_remove(&bucket->conns, &conn->key_node[CT_DIR_FWD].cm_node, hash);
 
     zl = zone_limit_lookup(ct, conn->admit_zone);
     if (zl && zl->czl.zone_limit_seq == conn->zone_limit_seq) {
@@ -429,12 +433,10 @@ conn_clean(struct conntrack *ct, struct conn *conn)
     }
 
     if (conn->nat_action) {
-        hash = conn_key_hash(&conn->key_node[CT_DIR_REV].key,
-                             ct->hash_basis);
-        cmap_remove(&ct->conns, &conn->key_node[CT_DIR_REV].cm_node, hash);
+        hash = conn->key_node[CT_DIR_REV].key_hash;
+        bucket = &ct->buckets[hash_to_bucket(hash)];
+        cmap_remove(&bucket->conns, &conn->key_node[CT_DIR_REV].cm_node, hash);
     }
-    ovs_list_remove(&conn->exp_node);
-    conn->cleaned = true;
     ovsrcu_postpone(delete_conn, conn);
     atomic_count_dec(&ct->n_conn);
 }
@@ -446,22 +448,32 @@ void
 conntrack_destroy(struct conntrack *ct)
 {
     struct conn_key_node *keyn;
+    struct ct_bucket *bucket;
     struct conn *conn;
+    int i;
 
     latch_set(&ct->clean_thread_exit);
     pthread_join(ct->clean_thread, NULL);
     latch_destroy(&ct->clean_thread_exit);
 
-    ovs_mutex_lock(&ct->ct_lock);
-    CMAP_FOR_EACH (keyn, cm_node, &ct->conns) {
-        if (keyn->dir != CT_DIR_FWD) {
-            continue;
+    for (i = 0; i < CONNTRACK_BUCKETS; i++) {
+        bucket = &ct->buckets[i];
+        CMAP_FOR_EACH (keyn, cm_node, &bucket->conns) {
+            if (keyn->dir != CT_DIR_FWD) {
+                continue;
+            }
+            conn = CONTAINER_OF(keyn, struct conn, key_node[keyn->dir]);
+            conn_clean(ct, conn);
         }
-        conn = CONTAINER_OF(keyn, struct conn, key_node[keyn->dir]);
-        conn_clean(ct, conn);
     }
-    cmap_destroy(&ct->conns);
 
+    for (i = 0; i < CONNTRACK_BUCKETS; i++) {
+        bucket = &ct->buckets[i];
+        ovs_mutex_destroy(&bucket->lock);
+        cmap_destroy(&ct->buckets[i].conns);
+    }
+
+    ovs_mutex_lock(&ct->ct_lock);
     struct zone_limit *zl;
     CMAP_FOR_EACH (zl, node, &ct->zone_limits) {
         uint32_t hash = zone_key_hash(zl->czl.zone, ct->hash_basis);
@@ -498,45 +510,132 @@ conntrack_destroy(struct conntrack *ct)
 }
 
 
+static unsigned hash_to_bucket(uint32_t hash)
+{
+    return hash >> (32 - CONNTRACK_BUCKETS_SHIFT);
+}
+
 static bool
-conn_key_lookup(struct conntrack *ct, const struct conn_key *key,
-                uint32_t hash, long long now, struct conn **conn_out,
-                bool *reply)
+conn_key_lookup__(struct conntrack *ct, unsigned bucket,
+                  const struct conn_key *key, uint32_t hash,
+                  long long now, struct conn **conn_out,
+                  bool *reply)
 {
+    struct ct_bucket *ctb = &ct->buckets[bucket];
     struct conn_key_node *keyn;
-    struct conn *conn = NULL;
     bool found = false;
+    struct conn *conn;
 
-    CMAP_FOR_EACH_WITH_HASH (keyn, cm_node, hash, &ct->conns) {
+    CMAP_FOR_EACH_WITH_HASH (keyn, cm_node, hash, &ctb->conns) {
         conn = CONTAINER_OF(keyn, struct conn, key_node[keyn->dir]);
+        if (conn_expired(conn, now)) {
+            continue;
+        }
+
         for (int i = CT_DIR_FWD; i < CT_DIR_MAX; i++) {
-            if (!conn_key_cmp(&conn->key_node[i].key, key) &&
-                !conn_expired(conn, now)) {
+            if (!conn_key_cmp(&conn->key_node[i].key, key)) {
                 found = true;
                 if (reply) {
                     *reply = i;
                 }
-                goto out_found;
+
+                goto conn_out_found;
             }
         }
     }
 
-out_found:
-    if (found && conn_out) {
-        *conn_out = conn;
-    } else if (conn_out) {
-        *conn_out = NULL;
+conn_out_found:
+    if (conn_out) {
+        *conn_out = found ? conn : NULL;
     }
 
     return found;
 }
 
+static bool
+conn_key_lookup(struct conntrack *ct, unsigned bucket,
+                const struct conn_key *key, uint32_t hash,
+                long long now, struct conn **conn_out,
+                bool *reply)
+{
+    return conn_key_lookup__(ct, bucket, key, hash,
+                             now, conn_out, reply);
+}
+
+static void
+ct_buckets_unlock__(struct conntrack *ct, uint32_t p1, uint32_t p2)
+    OVS_NO_THREAD_SAFETY_ANALYSIS
+{
+    if (p1 < p2) {
+        ovs_mutex_unlock(&ct->buckets[p2].lock);
+        ovs_mutex_unlock(&ct->buckets[p1].lock);
+    } else if (p1 > p2) {
+        ovs_mutex_unlock(&ct->buckets[p1].lock);
+        ovs_mutex_unlock(&ct->buckets[p2].lock);
+    } else {
+        ovs_mutex_unlock(&ct->buckets[p1].lock);
+    }
+}
+
+static void
+ct_buckets_unlock(struct conntrack *ct, uint32_t h1, uint32_t h2)
+{
+    unsigned p1 = hash_to_bucket(h1),
+        p2 = hash_to_bucket(h2);
+
+    ct_buckets_unlock__(ct, p1, p2);
+}
+
+/* Acquires both locks in an ordered way. */
+static void
+ct_buckets_lock__(struct conntrack *ct, unsigned p1, unsigned p2)
+    OVS_NO_THREAD_SAFETY_ANALYSIS
+{
+    if (p1 < p2) {
+        ovs_mutex_lock(&ct->buckets[p1].lock);
+        ovs_mutex_lock(&ct->buckets[p2].lock);
+    } else if (p1 > p2) {
+        ovs_mutex_lock(&ct->buckets[p2].lock);
+        ovs_mutex_lock(&ct->buckets[p1].lock);
+    } else {
+        ovs_mutex_lock(&ct->buckets[p1].lock);
+    }
+}
+
+static void
+ct_buckets_lock(struct conntrack *ct, uint32_t h1, uint32_t h2)
+{
+    unsigned p1 = hash_to_bucket(h1),
+        p2 = hash_to_bucket(h2);
+
+    ct_buckets_lock__(ct, p1, p2);
+}
+
+static void
+conn_clean(struct conntrack *ct, struct conn *conn)
+{
+    uint32_t h1, h2;
+
+    if (atomic_flag_test_and_set(&conn->cleaned)) {
+        return;
+    }
+
+    h1 = conn->key_node[CT_DIR_FWD].key_hash;
+    h2 = conn->key_node[CT_DIR_REV].key_hash;
+    ct_buckets_lock(ct, h1, h2);
+    conn_clean__(ct, conn);
+    ct_buckets_unlock(ct, h1, h2);
+}
+
 static bool
 conn_lookup(struct conntrack *ct, const struct conn_key *key,
             long long now, struct conn **conn_out, bool *reply)
 {
     uint32_t hash = conn_key_hash(key, ct->hash_basis);
-    return conn_key_lookup(ct, key, hash, now, conn_out, reply);
+    unsigned bucket = hash_to_bucket(hash);
+
+    return conn_key_lookup__(ct, bucket, key, hash,
+                             now, conn_out, reply);
 }
 
 static void
@@ -944,7 +1043,6 @@ conn_not_found(struct conntrack *ct, struct dp_packet *pkt,
                const struct nat_action_info_t *nat_action_info,
                const char *helper, const struct alg_exp_node *alg_exp,
                enum ct_alg_ctl_type ct_alg_ctl, uint32_t tp_id)
-    OVS_REQUIRES(ct->ct_lock)
 {
     struct conn *nc = NULL;
     uint32_t rev_hash = ctx->hash;
@@ -961,10 +1059,11 @@ conn_not_found(struct conntrack *ct, struct dp_packet *pkt,
     }
 
     if (commit) {
-        struct conn_key_node *fwd_key_node, *rev_key_node;
-
         struct zone_limit *zl = zone_limit_lookup_or_default(ct,
                                                              ctx->key.zone);
+        struct conn_key_node *fwd_key_node, *rev_key_node;
+        bool handle_tuple = false;
+
         if (zl && atomic_count_get(&zl->czl.count) >= zl->czl.limit) {
             return nc;
         }
@@ -1007,22 +1106,40 @@ conn_not_found(struct conntrack *ct, struct dp_packet *pkt,
                     nc->nat_action = NAT_ACTION_DST;
                 }
             } else {
-                bool nat_res = nat_get_unique_tuple(ct, nc, nat_action_info);
-
-                if (!nat_res) {
-                    goto nat_res_exhaustion;
-                }
+                handle_tuple = true;
             }
-
-            nat_packet(pkt, nc, ctx->icmp_related);
-            rev_hash = conn_key_hash(&rev_key_node->key, ct->hash_basis);
-            rev_key_node->dir = CT_DIR_REV;
-            cmap_insert(&ct->conns, &rev_key_node->cm_node, rev_hash);
         }
 
         ovs_mutex_init_adaptive(&nc->lock);
+        atomic_flag_clear(&nc->cleaned);
         fwd_key_node->dir = CT_DIR_FWD;
-        cmap_insert(&ct->conns, &fwd_key_node->cm_node, ctx->hash);
+        rev_key_node->dir = CT_DIR_REV;
+
+        if (handle_tuple) {
+            bool nat_res = nat_get_unique_tuple_lock(ct, nc, nat_action_info,
+                                                     &rev_hash);
+
+            if (!nat_res) {
+                goto out_error;
+            }
+        } else {
+            rev_hash = conn_key_hash(&rev_key_node->key, ct->hash_basis);
+            ct_buckets_lock(ct, ctx->hash, rev_hash);
+        }
+
+        if (conn_lookup(ct, &ctx->key, now, NULL, NULL)) {
+            goto out_error_unlock;
+        }
+
+        fwd_key_node->key_hash = ctx->hash;
+        cmap_insert(&ct->buckets[hash_to_bucket(ctx->hash)].conns,
+                    &fwd_key_node->cm_node, ctx->hash);
+        if (nat_action_info) {
+            rev_key_node->key_hash = rev_hash;
+            cmap_insert(&ct->buckets[hash_to_bucket(rev_hash)].conns,
+                        &rev_key_node->cm_node, rev_hash);
+            nat_packet(pkt, nc, ctx->icmp_related);
+        }
 
         atomic_count_inc(&ct->n_conn);
         ctx->conn = nc; /* For completeness. */
@@ -1033,21 +1150,23 @@ conn_not_found(struct conntrack *ct, struct dp_packet *pkt,
         } else {
             nc->admit_zone = INVALID_ZONE;
         }
+        ct_buckets_unlock(ct, ctx->hash, rev_hash);
     }
 
     return nc;
 
+out_error_unlock:
+    ct_buckets_unlock(ct, ctx->hash, rev_hash);
     /* This would be a user error or a DOS attack.  A user error is prevented
      * by allocating enough combinations of NAT addresses when combined with
      * ephemeral ports.  A DOS attack should be protected against with
      * firewall rules or a separate firewall.  Also using zone partitioning
      * can limit DoS impact. */
-nat_res_exhaustion:
-    ovs_list_remove(&nc->exp_node);
+out_error:
+    ovs_mutex_destroy(&nc->lock);
     delete_conn_cmn(nc);
     static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
-    VLOG_WARN_RL(&rl, "Unable to NAT due to tuple space exhaustion - "
-                 "if DoS attack, use firewalling and/or zone partitioning.");
+    VLOG_WARN_RL(&rl, "Unable to insert a new connection.");
     return NULL;
 }
 
@@ -1082,12 +1201,7 @@ conn_update_state(struct conntrack *ct, struct dp_packet *pkt,
             pkt->md.ct_state = CS_INVALID;
             break;
         case CT_UPDATE_NEW:
-            ovs_mutex_lock(&ct->ct_lock);
-            if (conn_lookup(ct, &conn->key_node[CT_DIR_FWD].key,
-                            now, NULL, NULL)) {
-                conn_clean(ct, conn);
-            }
-            ovs_mutex_unlock(&ct->ct_lock);
+            conn_clean(ct, conn);
             create_new_conn = true;
             break;
         case CT_UPDATE_VALID_NEW:
@@ -1253,6 +1367,8 @@ static void
 initial_conn_lookup(struct conntrack *ct, struct conn_lookup_ctx *ctx,
                     long long now, bool natted)
 {
+    unsigned bucket = hash_to_bucket(ctx->hash);
+
     if (natted) {
         /* If the packet has been already natted (e.g. a previous
          * action took place), retrieve it performing a lookup of its
@@ -1260,7 +1376,8 @@ initial_conn_lookup(struct conntrack *ct, struct conn_lookup_ctx *ctx,
         conn_key_reverse(&ctx->key);
     }
 
-    conn_key_lookup(ct, &ctx->key, ctx->hash, now, &ctx->conn, &ctx->reply);
+    conn_key_lookup(ct, bucket, &ctx->key, ctx->hash,
+                    now, &ctx->conn, &ctx->reply);
 
     if (natted) {
         if (OVS_LIKELY(ctx->conn)) {
@@ -1287,24 +1404,20 @@ process_one(struct conntrack *ct, struct dp_packet *pkt,
             ovs_be16 tp_src, ovs_be16 tp_dst, const char *helper,
             uint32_t tp_id)
 {
+    bool create_new_conn = false;
+
     /* Reset ct_state whenever entering a new zone. */
     if (pkt->md.ct_state && pkt->md.ct_zone != zone) {
         pkt->md.ct_state = 0;
     }
 
-    bool create_new_conn = false;
     initial_conn_lookup(ct, ctx, now, !!(pkt->md.ct_state &
                                          (CS_SRC_NAT | CS_DST_NAT)));
     struct conn *conn = ctx->conn;
 
     /* Delete found entry if in wrong direction. 'force' implies commit. */
     if (OVS_UNLIKELY(force && ctx->reply && conn)) {
-        ovs_mutex_lock(&ct->ct_lock);
-        if (conn_lookup(ct, &conn->key_node[CT_DIR_FWD].key,
-                        now, NULL, NULL)) {
-            conn_clean(ct, conn);
-        }
-        ovs_mutex_unlock(&ct->ct_lock);
+        conn_clean(ct, conn);
         conn = NULL;
     }
 
@@ -1338,7 +1451,6 @@ process_one(struct conntrack *ct, struct dp_packet *pkt,
     struct alg_exp_node alg_exp_entry;
 
     if (OVS_UNLIKELY(create_new_conn)) {
-
         ovs_rwlock_rdlock(&ct->resources_lock);
         alg_exp = expectation_lookup(&ct->alg_expectations, &ctx->key,
                                      ct->hash_basis,
@@ -1349,12 +1461,9 @@ process_one(struct conntrack *ct, struct dp_packet *pkt,
         }
         ovs_rwlock_unlock(&ct->resources_lock);
 
-        ovs_mutex_lock(&ct->ct_lock);
-        if (!conn_lookup(ct, &ctx->key, now, NULL, NULL)) {
-            conn = conn_not_found(ct, pkt, ctx, commit, now, nat_action_info,
-                                  helper, alg_exp, ct_alg_ctl, tp_id);
-        }
-        ovs_mutex_unlock(&ct->ct_lock);
+        conn = conn_not_found(ct, pkt, ctx, commit, now,
+                              nat_action_info, helper, alg_exp,
+                              ct_alg_ctl, tp_id);
     }
 
     write_ct_md(pkt, zone, conn, &ctx->key, alg_exp);
@@ -1467,83 +1576,92 @@ set_label(struct dp_packet *pkt, struct conn *conn,
 }
 
 
-/* Delete the expired connections from 'ctb', up to 'limit'. Returns the
- * earliest expiration time among the remaining connections in 'ctb'.  Returns
- * LLONG_MAX if 'ctb' is empty.  The return value might be smaller than 'now',
- * if 'limit' is reached */
+/* Delete the expired connections from 'bucket', up to 'limit'.
+ * Returns the earliest expiration time among the remaining
+ * connections in 'bucket'.  Returns LLONG_MAX if 'bucket' is empty.
+ * The return value might be smaller than 'now', if 'limit' is
+ * reached. */
 static long long
-ct_sweep(struct conntrack *ct, long long now, size_t limit)
+sweep_bucket(struct conntrack *ct, struct ct_bucket *bucket,
+             long long now)
 {
+    struct conn_key_node *keyn;
+    unsigned int conn_count = 0;
     struct conn *conn;
-    long long min_expiration = LLONG_MAX;
-    size_t count = 0;
+    long long expiration;
 
-    ovs_mutex_lock(&ct->ct_lock);
+    CMAP_FOR_EACH (keyn, cm_node, &bucket->conns) {
+        if (keyn->dir != CT_DIR_FWD) {
+            continue;
+        }
 
-    for (unsigned i = 0; i < N_CT_TM; i++) {
-        LIST_FOR_EACH_SAFE (conn, exp_node, &ct->exp_lists[i]) {
-            ovs_mutex_lock(&conn->lock);
-            if (now < conn->expiration || count >= limit) {
-                min_expiration = MIN(min_expiration, conn->expiration);
-                ovs_mutex_unlock(&conn->lock);
-                if (count >= limit) {
-                    /* Do not check other lists. */
-                    COVERAGE_INC(conntrack_long_cleanup);
-                    goto out;
-                }
-                break;
-            } else {
-                ovs_mutex_unlock(&conn->lock);
-                conn_clean(ct, conn);
-            }
-            count++;
+        conn = CONTAINER_OF(keyn, struct conn, key_node[keyn->dir]);
+        ovs_mutex_lock(&conn->lock);
+        expiration = conn->expiration;
+        ovs_mutex_unlock(&conn->lock);
+
+        if (now >= expiration) {
+            conn_clean(ct, conn);
         }
+
+        conn_count++;
     }
 
-out:
-    VLOG_DBG("conntrack cleanup %"PRIuSIZE" entries in %lld msec", count,
-             time_msec() - now);
-    ovs_mutex_unlock(&ct->ct_lock);
-    return min_expiration;
+    return conn_count;
 }
 
-/* Cleans up old connection entries from 'ct'.  Returns the time when the
- * next expiration might happen.  The return value might be smaller than
- * 'now', meaning that an internal limit has been reached, and some expired
- * connections have not been deleted. */
+/* Cleans up old connection entries from 'ct'.  Returns the the next
+ * wake up time.  The return value might be smaller than 'now', meaning
+ * that an internal limit has been reached, that is, the table
+ * hasn't been entirely scanned. */
 static long long
 conntrack_clean(struct conntrack *ct, long long now)
 {
-    unsigned int n_conn_limit;
+    long long next_wakeup = now + 30 * 1000;
+    unsigned int n_conn_limit, i, count = 0;
+    size_t clean_end;
+
     atomic_read_relaxed(&ct->n_conn_limit, &n_conn_limit);
-    size_t clean_max = n_conn_limit > 10 ? n_conn_limit / 10 : 1;
-    long long min_exp = ct_sweep(ct, now, clean_max);
-    long long next_wakeup = MIN(min_exp, now + CT_DPIF_NETDEV_TP_MIN);
+    clean_end = n_conn_limit / 64;
+
+    for (i = ct->next_bucket; i < CONNTRACK_BUCKETS; i++) {
+        struct ct_bucket *bucket = &ct->buckets[i];
+
+        count += sweep_bucket(ct, bucket, now);
+
+        if (count > clean_end) {
+            next_wakeup = 0;
+            break;
+        }
+    }
+
+    ct->next_bucket = (i < CONNTRACK_BUCKETS) ? i : 0;
 
     return next_wakeup;
 }
 
 /* Cleanup:
  *
- * We must call conntrack_clean() periodically.  conntrack_clean() return
- * value gives an hint on when the next cleanup must be done (either because
- * there is an actual connection that expires, or because a new connection
- * might be created with the minimum timeout).
+ * We must call conntrack_clean() periodically.  conntrack_clean()
+ * return value gives an hint on when the next cleanup must be done
+ * (either because there is still work to do, or because a new
+ * connection might be created).
  *
  * The logic below has two goals:
  *
- * - We want to reduce the number of wakeups and batch connection cleanup
- *   when the load is not very high.  CT_CLEAN_INTERVAL ensures that if we
- *   are coping with the current cleanup tasks, then we wait at least
- *   5 seconds to do further cleanup.
+ * - When the load is high, we want to avoid to hog the CPU scanning
+ *   all the buckets and their respective CMAPs "at once". For this
+ *   reason, every batch cleanup aims to scan at most n_conn_limit /
+ *   64 entries (more if the buckets contains many entrie) before
+ *   yielding the CPU. In this case, the next wake up will happen in
+ *   CT_CLEAN_MIN_INTERVAL_MS and the scan will resume starting from
+ *   the first bucket not scanned.
  *
- * - We don't want to keep the map locked too long, as we might prevent
- *   traffic from flowing.  CT_CLEAN_MIN_INTERVAL ensures that if cleanup is
- *   behind, there is at least some 200ms blocks of time when the map will be
- *   left alone, so the datapath can operate unhindered.
- */
-#define CT_CLEAN_INTERVAL 5000 /* 5 seconds */
-#define CT_CLEAN_MIN_INTERVAL 200  /* 0.2 seconds */
+ * - We also don't want to scan the buckets so frequently, as going
+ *   through all the connections, during high loads, may be costly in
+ *   terms of CPU time. In this case the next wake up is set to 90
+ *   seconds. */
+#define CT_CLEAN_MIN_INTERVAL_MS 100  /* 0.1 seconds */
 
 static void *
 clean_thread_main(void *f_)
@@ -1556,9 +1674,9 @@ clean_thread_main(void *f_)
         next_wake = conntrack_clean(ct, now);
 
         if (next_wake < now) {
-            poll_timer_wait_until(now + CT_CLEAN_MIN_INTERVAL);
+            poll_timer_wait_until(now + CT_CLEAN_MIN_INTERVAL_MS);
         } else {
-            poll_timer_wait_until(MAX(next_wake, now + CT_CLEAN_INTERVAL));
+            poll_timer_wait_until(next_wake);
         }
         latch_wait(&ct->clean_thread_exit);
         poll_block();
@@ -2301,9 +2419,10 @@ store_addr_to_key(union ct_addr *addr, struct conn_key *key,
 }
 
 static bool
-nat_get_unique_l4(struct conntrack *ct, struct conn_key *rev_key,
-                  ovs_be16 *port, uint16_t curr, uint16_t min,
-                  uint16_t max)
+nat_get_unique_l4(struct conntrack *ct, struct conn_key *fwd_key,
+                  struct conn_key *rev_key, ovs_be16 *port,
+                  uint16_t curr, uint16_t min,
+                  uint16_t max, uint32_t *rev_hash)
 {
     static const unsigned int max_attempts = 128;
     uint16_t range = max - min + 1;
@@ -2324,10 +2443,15 @@ another_round:
         }
 
         *port = htons(curr);
-        if (!conn_lookup(ct, rev_key, time_msec(),
-                         NULL, NULL)) {
+        uint32_t key_hash = conn_key_hash(fwd_key, ct->hash_basis);
+        *rev_hash = conn_key_hash(rev_key, ct->hash_basis);
+
+        ct_buckets_lock(ct, key_hash, *rev_hash);
+        if (!conn_lookup(ct, rev_key,
+                         time_msec(), NULL, NULL)) {
             return true;
         }
+        ct_buckets_unlock(ct, key_hash, *rev_hash);
     }
 
     if (attempts < range && attempts >= 16) {
@@ -2361,10 +2485,13 @@ another_round:
  *          tries to find a source port in the ephemeral
  *          range (after testing the port used by the sender).
  *
- * If none can be found, return exhaustion to the caller. */
+ * If none can be found, return exhaustion to the caller.
+ * If successful, this function returns with the locks
+ * acquired. */
 static bool
-nat_get_unique_tuple(struct conntrack *ct, struct conn *conn,
-                     const struct nat_action_info_t *nat_info)
+nat_get_unique_tuple_lock(struct conntrack *ct, struct conn *conn,
+                          const struct nat_action_info_t *nat_info,
+                          uint32_t *rev_hash)
 {
     union ct_addr min_addr = {0}, max_addr = {0}, addr = {0};
     struct conn_key *fwd_key = &conn->key_node[CT_DIR_FWD].key,
@@ -2397,23 +2524,30 @@ nat_get_unique_tuple(struct conntrack *ct, struct conn *conn,
                       nat_info->nat_action);
 
     if (!pat_proto) {
+        uint32_t key_hash = conn_key_hash(fwd_key, ct->hash_basis);
+        *rev_hash = conn_key_hash(rev_key, ct->hash_basis);
+
+        ct_buckets_lock(ct, key_hash, *rev_hash);
         if (!conn_lookup(ct, rev_key,
                          time_msec(), NULL, NULL)) {
             return true;
         }
+        ct_buckets_unlock(ct, key_hash, *rev_hash);
 
         return false;
     }
 
     bool found = false;
     if (nat_info->nat_action & NAT_ACTION_DST_PORT) {
-        found = nat_get_unique_l4(ct, rev_key, &rev_key->src.port,
-                                  curr_dport, min_dport, max_dport);
+        found = nat_get_unique_l4(ct, fwd_key, rev_key, &rev_key->src.port,
+                                  curr_dport, min_dport, max_dport,
+                                  rev_hash);
     }
 
     if (!found) {
-        found = nat_get_unique_l4(ct, rev_key, &rev_key->dst.port,
-                                  curr_sport, min_sport, max_sport);
+        found = nat_get_unique_l4(ct, fwd_key, rev_key, &rev_key->dst.port,
+                                  curr_sport, min_sport, max_sport,
+                                  rev_hash);
     }
 
     if (found) {
@@ -2615,20 +2749,38 @@ conntrack_dump_next(struct conntrack_dump *dump, struct ct_dpif_entry *entry)
 {
     struct conntrack *ct = dump->ct;
     long long now = time_msec();
+    struct ct_bucket *bucket;
 
-    for (;;) {
-        struct cmap_node *cm_node = cmap_next_position(&ct->conns,
-                                                       &dump->cm_pos);
-        if (!cm_node) {
-            break;
+    while (dump->bucket < CONNTRACK_BUCKETS) {
+        struct cmap_node *cm_node;
+        bucket = &ct->buckets[dump->bucket];
+
+        for (;;) {
+            cm_node = cmap_next_position(&bucket->conns,
+                                         &dump->cm_pos);
+            if (!cm_node) {
+                break;
+            }
+            struct conn_key_node *keyn;
+            struct conn *conn;
+            INIT_CONTAINER(keyn, cm_node, cm_node);
+            conn = CONTAINER_OF(keyn, struct conn, key_node[keyn->dir]);
+
+            if (conn_expired(conn, now)) {
+                continue;
+            }
+
+            if ((!dump->filter_zone || keyn->key.zone == dump->zone) &&
+                (keyn->dir == CT_DIR_FWD)) {
+                conn_to_ct_dpif_entry(conn, entry, now);
+                break;
+            }
         }
-        struct conn_key_node *keyn;
-        struct conn *conn;
-        INIT_CONTAINER(keyn, cm_node, cm_node);
-        conn = CONTAINER_OF(keyn, struct conn, key_node[keyn->dir]);
-        if ((!dump->filter_zone || keyn->key.zone == dump->zone) &&
-            (keyn->dir == CT_DIR_FWD)) {
-            conn_to_ct_dpif_entry(conn, entry, now);
+
+        if (!cm_node) {
+            memset(&dump->cm_pos, 0, sizeof dump->cm_pos);
+            dump->bucket++;
+        } else {
             return 0;
         }
     }
@@ -2648,17 +2800,18 @@ conntrack_flush(struct conntrack *ct, const uint16_t *zone)
     struct conn_key_node *keyn;
     struct conn *conn;
 
-    ovs_mutex_lock(&ct->ct_lock);
-    CMAP_FOR_EACH (keyn, cm_node, &ct->conns) {
-        if (keyn->dir != CT_DIR_FWD) {
-            continue;
-        }
-        conn = CONTAINER_OF(keyn, struct conn, key_node[keyn->dir]);
-        if (!zone || *zone == keyn->key.zone) {
-            conn_clean(ct, conn);
+    for (unsigned i = 0; i < CONNTRACK_BUCKETS; i++) {
+        CMAP_FOR_EACH (keyn, cm_node, &ct->buckets[i].conns) {
+            if (keyn->dir != CT_DIR_FWD) {
+                continue;
+            }
+
+            conn = CONTAINER_OF(keyn, struct conn, key_node[keyn->dir]);
+            if (!zone || *zone == keyn->key.zone) {
+                conn_clean(ct, conn);
+            }
         }
     }
-    ovs_mutex_unlock(&ct->ct_lock);
 
     return 0;
 }
@@ -2667,15 +2820,19 @@ int
 conntrack_flush_tuple(struct conntrack *ct, const struct ct_dpif_tuple *tuple,
                       uint16_t zone)
 {
-    int error = 0;
     struct conn_key key;
-    struct conn *conn;
+    struct conn *conn = NULL;
+    unsigned bucket;
+    uint32_t hash;
+    int error = 0;
 
     memset(&key, 0, sizeof(key));
     tuple_to_conn_key(tuple, zone, &key);
-    ovs_mutex_lock(&ct->ct_lock);
-    conn_lookup(ct, &key, time_msec(), &conn, NULL);
 
+    hash = conn_key_hash(&key, ct->hash_basis);
+    bucket = hash_to_bucket(hash);
+
+    conn_key_lookup(ct, bucket, &key, hash, time_msec(), &conn, NULL);
     if (conn) {
         conn_clean(ct, conn);
     } else {
@@ -2683,7 +2840,6 @@ conntrack_flush_tuple(struct conntrack *ct, const struct ct_dpif_tuple *tuple,
         error = ENOENT;
     }
 
-    ovs_mutex_unlock(&ct->ct_lock);
     return error;
 }
 
diff --git a/tests/system-traffic.at b/tests/system-traffic.at
index 36e10aa4a..90794571a 100644
--- a/tests/system-traffic.at
+++ b/tests/system-traffic.at
@@ -5131,9 +5131,8 @@ AT_CHECK([ovs-appctl dpctl/dump-conntrack | FORMAT_CT(10.1.1.2) | sed -e 's/dst=
 tcp,orig=(src=10.1.1.1,dst=10.1.1.2,sport=<cleared>,dport=<cleared>),reply=(src=10.1.1.2,dst=10.1.1.2XX,sport=<cleared>,dport=<cleared>),zone=1,protoinfo=(state=<cleared>)
 ])
 
-OVS_TRAFFIC_VSWITCHD_STOP(["dnl
-/Unable to NAT due to tuple space exhaustion - if DoS attack, use firewalling and\/or zone partitioning./d
-/Dropped .* log messages in last .* seconds \(most recently, .* seconds ago\) due to excessive rate/d"])
+OVS_TRAFFIC_VSWITCHD_STOP(["/dnl
+Unable to insert a new connection./d"])
 AT_CLEANUP
 
 AT_SETUP([conntrack - more complex SNAT])
