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,
            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 (conn && force && ctx->reply) {
        conn_clean(ct, conn, &ct->buckets[bucket]);
        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;
            ctx2.key = conn->rev_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;
    bool ftp_ctl = is_ftp_ctl(pkt);

    if (OVS_LIKELY(conn)) {
        if (ftp_ctl) {
            /* Keep sequence tracking in sync with the source of the
             * sequence skew. */
            if (ctx->reply != conn->seq_skew_dir) {
                handle_ftp_ctl(ct, ctx, pkt, conn, now, CT_FTP_CTL_OTHER,
                               !!nat_action_info);
                create_new_conn = conn_update_state(ct, pkt, ctx, &conn, now,
                                                    bucket);
            } else {
                create_new_conn = conn_update_state(ct, pkt, ctx, &conn, now,
                                                    bucket);

                if (!create_new_conn) {
                    handle_ftp_ctl(ct, ctx, pkt, conn, now, CT_FTP_CTL_OTHER,
                                   !!nat_action_info);
                }
            }
        } else {
            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);
        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);
    }

    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]);
    }

    bool tftp_ctl = is_tftp_ctl(pkt);
    struct conn conn_for_expectation;
    if (conn && (ftp_ctl || tftp_ctl)) {
        conn_for_expectation = *conn;
    }

    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);
    }

    /* FTP control packet handling with expectation creation. */
    if (OVS_UNLIKELY(conn && ftp_ctl)) {
        handle_ftp_ctl(ct, ctx, pkt, &conn_for_expectation,
                       now, CT_FTP_CTL_INTEREST, !!nat_action_info);
    } else if (OVS_UNLIKELY(conn && tftp_ctl)) {
        handle_tftp_ctl(ct, &conn_for_expectation, now);
    }
}