process_one(struct conntrack *ct, struct dp_packet *pkt,
            struct conn_lookup_ctx *ctx, uint16_t zone,
            bool force, bool commit, long long now, const uint32_t *setmark,
            const struct ovs_key_ct_labels *setlabel,
            const struct nat_action_info_t *nat_action_info,
            ovs_be16 tp_src, ovs_be16 tp_dst, const char *helper)
{
    struct conn *conn;
    unsigned bucket = hash_to_bucket(ctx->hash);
    ct_lock_lock(&ct->buckets[bucket].lock);
    conn_key_lookup(&ct->buckets[bucket], ctx, now);
    conn = ctx->conn;

    /* Reset ct_state whenever entering a new zone. */
    if (pkt->md.ct_state && pkt->md.ct_zone != zone) {
        pkt->md.ct_state = 0;
    }

    /* Delete found entry if in wrong direction. 'force' implies commit. */
    if (OVS_UNLIKELY(force && ctx->reply && conn)) {
        struct conn lconn;
        memcpy(&lconn, conn, sizeof lconn);
        ct_lock_unlock(&ct->buckets[bucket].lock);
        conn_clean_safe(ct, &lconn, &ct->buckets[bucket], ctx->hash);
        ct_lock_lock(&ct->buckets[bucket].lock);
        conn = NULL;
    }

    if (OVS_LIKELY(conn)) {
        if (conn->conn_type == CT_CONN_TYPE_UN_NAT) {

            ctx->reply = true;

            struct conn_lookup_ctx ctx2;
            ctx2.conn = NULL;
            memcpy(&ctx2.key, &conn->rev_key, sizeof ctx2.key);
            ctx2.hash = conn_key_hash(&conn->rev_key, ct->hash_basis);

            ct_lock_unlock(&ct->buckets[bucket].lock);
            bucket = hash_to_bucket(ctx2.hash);

            ct_lock_lock(&ct->buckets[bucket].lock);
            conn_key_lookup(&ct->buckets[bucket], &ctx2, now);

            if (ctx2.conn) {
                conn = ctx2.conn;
            } else {
                /* It is a race condition where conn has timed out and removed
                 * between unlock of the rev_conn and lock of the forward conn;
                 * nothing to do. */
                pkt->md.ct_state |= CS_INVALID;
                write_ct_md(pkt, zone, NULL, NULL, NULL);
                ct_lock_unlock(&ct->buckets[bucket].lock);
                return;
            }
        }
    }

    bool create_new_conn = false;
    struct conn conn_for_un_nat_copy;
    conn_for_un_nat_copy.conn_type = CT_CONN_TYPE_DEFAULT;

    enum ct_alg_ctl_type ct_alg_ctl = get_alg_ctl_type(pkt, tp_src, tp_dst,
                                                       helper);

    if (OVS_LIKELY(conn)) {
        if (OVS_LIKELY(!conn_update_state_alg(ct, pkt, ctx, conn,
                                              nat_action_info,
                                              ct_alg_ctl, now, bucket,
                                              &create_new_conn))) {
            create_new_conn = conn_update_state(ct, pkt, ctx, &conn, now,
                                                bucket);
        }
        if (nat_action_info && !create_new_conn) {
            handle_nat(pkt, conn, zone, ctx->reply, ctx->icmp_related);
        }

    } else if (check_orig_tuple(ct, pkt, ctx, now, &bucket, &conn,
                               nat_action_info)) {
        create_new_conn = conn_update_state(ct, pkt, ctx, &conn, now, bucket);
    } else {
        if (ctx->icmp_related) {
            /* An icmp related conn should always be found; no new
               connection is created based on an icmp related packet. */
            pkt->md.ct_state = CS_INVALID;
        } else {
            create_new_conn = true;
        }
    }

    const struct alg_exp_node *alg_exp = NULL;
    struct alg_exp_node alg_exp_entry;

    if (OVS_UNLIKELY(create_new_conn)) {

        ct_rwlock_rdlock(&ct->resources_lock);
        alg_exp = expectation_lookup(&ct->alg_expectations, &ctx->key,
                                     ct->hash_basis,
                                     alg_src_ip_wc(ct_alg_ctl));
        if (alg_exp) {
            alg_exp_entry = *alg_exp;
            alg_exp = &alg_exp_entry;
        }
        ct_rwlock_unlock(&ct->resources_lock);

        conn = conn_not_found(ct, pkt, ctx, commit, now, nat_action_info,
                              &conn_for_un_nat_copy, helper, alg_exp,
                              ct_alg_ctl);
    }

    write_ct_md(pkt, zone, conn, &ctx->key, alg_exp);

    if (conn && setmark) {
        set_mark(pkt, conn, setmark[0], setmark[1]);
    }

    if (conn && setlabel) {
        set_label(pkt, conn, &setlabel[0], &setlabel[1]);
    }

    struct conn conn_for_expectation;
    if (OVS_UNLIKELY((ct_alg_ctl != CT_ALG_CTL_NONE) && conn)) {
        memcpy(&conn_for_expectation, conn, sizeof conn_for_expectation);
    }

    ct_lock_unlock(&ct->buckets[bucket].lock);

    if (is_un_nat_conn_valid(&conn_for_un_nat_copy)) {
        create_un_nat_conn(ct, &conn_for_un_nat_copy, now, !!alg_exp);
    }

    handle_alg_ctl(ct, ctx, pkt, ct_alg_ctl, conn, now, !!nat_action_info,
                   &conn_for_expectation);
}