OVS_REQUIRES(ct->ct_lock)
{
    struct conn *nc = NULL;
    struct conn *nat_conn = NULL;

    if (!valid_new(pkt, &ctx->key)) {
        pkt->md.ct_state = CS_INVALID;
        return nc;
    }

    pkt->md.ct_state = CS_NEW;

    if (alg_exp) {
        pkt->md.ct_state |= CS_RELATED;
    }

    if (commit) {
        struct zone_limit *zl = zone_limit_lookup_or_default(ct,
                                                             ctx->key.zone);
        if (zl && zl->czl.count >= zl->czl.limit) {
            return nc;
        }

        unsigned int n_conn_limit;
        atomic_read_relaxed(&ct->n_conn_limit, &n_conn_limit);
        if (atomic_count_get(&ct->n_conn) >= n_conn_limit) {
            COVERAGE_INC(conntrack_full);
            return nc;
        }

        nc = new_conn(ct, pkt, &ctx->key, now, tp_id);
        memcpy(&nc->key, &ctx->key, sizeof nc->key);
        memcpy(&nc->rev_key, &nc->key, sizeof nc->rev_key);
        conn_key_reverse(&nc->rev_key);

        if (ct_verify_helper(helper, ct_alg_ctl)) {
            nc->alg = nullable_xstrdup(helper);
        }

        if (alg_exp) {
            nc->alg_related = true;
            nc->mark = alg_exp->parent_mark;
            nc->label = alg_exp->parent_label;
            nc->parent_key = alg_exp->parent_key;
        }

        if (nat_action_info) {
            nc->nat_info = xmemdup(nat_action_info, sizeof *nc->nat_info);
            nat_conn = xzalloc(sizeof *nat_conn);

            if (alg_exp) {
                if (alg_exp->nat_rpl_dst) {
                    nc->rev_key.dst.addr = alg_exp->alg_nat_repl_addr;
                    nc->nat_info->nat_action = NAT_ACTION_SRC;
                } else {
                    nc->rev_key.src.addr = alg_exp->alg_nat_repl_addr;
                    nc->nat_info->nat_action = NAT_ACTION_DST;
                }
            } else {
                memcpy(nat_conn, nc, sizeof *nat_conn);
                bool nat_res = nat_select_range_tuple(ct, nc, nat_conn);

                if (!nat_res) {
                    goto nat_res_exhaustion;
                }

                /* Update nc with nat adjustments made to nat_conn by
                 * nat_select_range_tuple(). */
                memcpy(nc, nat_conn, sizeof *nc);
            }

            nat_packet(pkt, nc, ctx->icmp_related);
            memcpy(&nat_conn->key, &nc->rev_key, sizeof nat_conn->key);
            memcpy(&nat_conn->rev_key, &nc->key, sizeof nat_conn->rev_key);
            nat_conn->conn_type = CT_CONN_TYPE_UN_NAT;
            nat_conn->nat_info = NULL;
            nat_conn->alg = NULL;
            nat_conn->nat_conn = NULL;
            uint32_t nat_hash = conn_key_hash(&nat_conn->key, ct->hash_basis);
            cmap_insert(&ct->conns, &nat_conn->cm_node, nat_hash);
        }

        nc->nat_conn = nat_conn;
        ovs_mutex_init_adaptive(&nc->lock);
        nc->conn_type = CT_CONN_TYPE_DEFAULT;
        cmap_insert(&ct->conns, &nc->cm_node, ctx->hash);
        atomic_count_inc(&ct->n_conn);
        ctx->conn = nc; /* For completeness. */
        if (zl) {
            nc->admit_zone = zl->czl.zone;
            nc->zone_limit_seq = zl->czl.zone_limit_seq;
            zl->czl.count++;
        } else {
            nc->admit_zone = INVALID_ZONE;
        }
    }

    return nc;

    /* 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:
    free(nat_conn);
    ovs_list_remove(&nc->exp_node);
    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.");
    return NULL;
}