datenlord / rdma-sys Goto Github PK

.derive_debug(true)
.no_debug("ibv_qp")
.no_debug("ibv_context")
.no_debug("ibv_cq_ex")
.no_debug("ibv_srq")
.no_debug("ibv_wq")
.no_debug("ibv_cq")

    pub fn init(&self) -> Result<()> {
        let mut attr = unsafe { std::mem::zeroed::<ibv_qp_attr>() };
        attr.qp_state = ibv_qp_state::IBV_QPS_INIT;
        attr.pkey_index = 0;
        attr.port_num = 1;
        attr.qp_access_flags = (ibv_access_flags::IBV_ACCESS_LOCAL_WRITE
            | ibv_access_flags::IBV_ACCESS_REMOTE_WRITE
            | ibv_access_flags::IBV_ACCESS_REMOTE_READ)
            .0.cast();
        let attr_mask = ibv_qp_attr_mask::IBV_QP_STATE
            | ibv_qp_attr_mask::IBV_QP_PKEY_INDEX
            | ibv_qp_attr_mask::IBV_QP_PORT
            | ibv_qp_attr_mask::IBV_QP_ACCESS_FLAGS;
        if unsafe { ibv_modify_qp(self.inner(), &mut attr, attr_mask.0.cast()) } != 0 {
            return Err(Error::last_os_error());
        }
        Ok(())
    }

    pub fn ready_to_receive(&self, remote_emp: EndPoint) -> Result<()> {
        let mut attr = unsafe { std::mem::zeroed::<ibv_qp_attr>() };
        attr.qp_state = ibv_qp_state::IBV_QPS_RTR;
        attr.path_mtu = ibv_mtu::IBV_MTU_512;
        attr.dest_qp_num = remote_emp.qpn;
        // qp_attr.rq_psn(X) must be equal to qp_attr.sq_psn(Y)
        attr.rq_psn = 0;
        attr.max_dest_rd_atomic = 1;
        attr.min_rnr_timer = 18;
        attr.ah_attr = ibv_ah_attr {
            dlid: remote_emp.lid,
            sl: 0,
            src_path_bits: 0,
            static_rate: 0,
            is_global: 1,
            port_num: 1,
            grh: ibv_global_route {
                sgid_index: 1,
                dgid: remote_emp.gid,
                hop_limit: 255,
                traffic_class: 0,
                flow_label: 0,
            }
        };
        let attr_mask = ibv_qp_attr_mask::IBV_QP_STATE
            | ibv_qp_attr_mask::IBV_QP_AV
            | ibv_qp_attr_mask::IBV_QP_PATH_MTU
            | ibv_qp_attr_mask::IBV_QP_DEST_QPN
            | ibv_qp_attr_mask::IBV_QP_RQ_PSN
            | ibv_qp_attr_mask::IBV_QP_MAX_DEST_RD_ATOMIC
            | ibv_qp_attr_mask::IBV_QP_MIN_RNR_TIMER;
        if unsafe { ibv_modify_qp(self.inner(), &mut attr, attr_mask.0.cast()) } != 0 {
            return Err(Error::last_os_error());
        }
        Ok(())
    }

    pub fn ready_to_send(&self) -> Result<()> {
        let mut attr = unsafe { std::mem::zeroed::<ibv_qp_attr>() };
        attr.qp_state = ibv_qp_state::IBV_QPS_RTS;
        attr.timeout = 14;
        attr.retry_cnt = 6;
        attr.rnr_retry = 6;
        attr.sq_psn = 0;
        attr.max_rd_atomic = 1;
        let attr_mask = ibv_qp_attr_mask::IBV_QP_STATE
            | ibv_qp_attr_mask::IBV_QP_TIMEOUT
            | ibv_qp_attr_mask::IBV_QP_RETRY_CNT
            | ibv_qp_attr_mask::IBV_QP_RNR_RETRY
            | ibv_qp_attr_mask::IBV_QP_SQ_PSN
            | ibv_qp_attr_mask::IBV_QP_MAX_QP_RD_ATOMIC;
        // SAFETY: ffi, and qp will not modify by other threads
        if unsafe { ibv_modify_qp(self.inner(), &mut attr, attr_mask.0.cast()) } != 0 {
            return Err(Error::last_os_error());
        }
        Ok(())
    }

lazy_static! {
    static ref DEVICE: Device = Device::new(default_device());
}

fn main() {
    let device = &DEVICE;

    let pd_client = Arc::new(PD::new(device));
    let pd_server = Arc::new(PD::new(device));

    let cq_client = Arc::new(CQ::new(device));
    let cq_server = Arc::new(CQ::new(device));

    let (tx1, rx1) = channel();
    let (tx2, rx2) = channel();

    println!("client start");
    let pd = pd_client.clone();
    let cq = cq_client.clone();
    let h1 = thread::spawn(move || client(pd, cq, tx1, rx2));

    println!("server start");
    let pd = pd_server.clone();
    let cq = cq_server.clone();
    let h2 = thread::spawn(move || server(pd, cq, tx2, rx1));
    h1.join().unwrap();
    h2.join().unwrap();
}

fn client(pd: Arc<PD>, cq: Arc<CQ>, tx: Sender<EndPoint>, rx: Receiver<EndPoint>) {
    let cap = QPCap::new(10, 10, 1, 1);
    // Create a QP
    let qp = QP::new(&pd, &cq, cap);
    if let Err(err) = qp.init() {
        println!("err: {}", err);
    }

    let enp = qp.endpoint();
    println!("client enp: {:?}", enp);

    // Exchange QP information with the remote side (e.g. using sockets)
    tx.send(enp);
    let remote_enp = rx.recv().unwrap();

    if let Err(err) = qp.ready_to_receive(remote_enp) {
        println!("client err : {}", err);
    }
    if let Err(err) = qp.ready_to_send() {...
}

fn server(pd: Arc<PD>, cq: Arc<CQ>, tx: Sender<EndPoint>, rx: Receiver<EndPoint>) {
    let cap = QPCap::new(10, 10, 1, 1);
    // Create a QP
    let qp = QP::new(&pd, &cq, cap);
    qp.init();

    let enp = qp.endpoint();
    println!("server enp: {:?}", enp);
    // Exchange QP information withw the remote side (e.g. using sockets)
    tx.send(enp);
    let remote_enp = rx.recv().unwrap();

    if let Err(err) = qp.ready_to_receive(remote_enp) {
        println!("server err : {}", err);
    }
    if let Err(err) = qp.ready_to_send() {
        println!("server err: {}", err);
    }
...
}