pub fn push(&self, task: T) {
        let backoff = Backoff::new();
        let mut tail = self.tail.index.load(Ordering::Acquire);
        let mut block = self.tail.block.load(Ordering::Acquire);
        let mut next_block = None;

        loop {
            // Calculate the offset of the index into the block.
            let offset = (tail >> SHIFT) % LAP;

            // If we reached the end of the block, wait until the next one is installed.
            if offset == BLOCK_CAP {
                backoff.snooze();
                tail = self.tail.index.load(Ordering::Acquire);
                block = self.tail.block.load(Ordering::Acquire);
                continue;
            }

            // If we're going to have to install the next block, allocate it in advance in order to
            // make the wait for other threads as short as possible.
            if offset + 1 == BLOCK_CAP && next_block.is_none() {
                next_block = Some(Box::new(Block::<T>::new()));
            }

            let new_tail = tail + (1 << SHIFT);

            // Try advancing the tail forward.
            match self.tail.index.compare_exchange_weak(
                tail,
                new_tail,
                Ordering::SeqCst,
                Ordering::Acquire,
            ) {
                Ok(_) => unsafe {
                    // If we've reached the end of the block, install the next one.
                    if offset + 1 == BLOCK_CAP {
                        let next_block = Box::into_raw(next_block.unwrap());
                        let next_index = new_tail.wrapping_add(1 << SHIFT);

                        self.tail.block.store(next_block, Ordering::Release);
                        self.tail.index.store(next_index, Ordering::Release);
                        (*block).next.store(next_block, Ordering::Release);
                    }

                    // Write the task into the slot.
                    let slot = (*block).slots.get_unchecked(offset);
                    slot.task.get().write(ManuallyDrop::new(task));
                    slot.state.fetch_or(WRITE, Ordering::Release);

                    return;
                },
                Err(t) => {
                    tail = t;
                    block = self.tail.block.load(Ordering::Acquire);
                    backoff.spin();
                }
            }
        }
    }
/// Pushes a task into the queue.
Pushes a task into the queue.
Pushes a task into the queue .
pub fn push(&self, task: T) {
        let backoff = Backoff::new();
        let mut tail = self.tail.index.load(Ordering::Acquire);
        let mut block = self.tail.block.load(Ordering::Acquire);
        let mut next_block = None;
        loop {
            // Calculate the offset of the index into the block.
            let offset = (tail >> SHIFT) % LAP;
            // If we reached the end of the block, wait until the next one is installed.
            if offset == BLOCK_CAP {
                backoff.snooze();
                tail = self.tail.index.load(Ordering::Acquire);
                block = self.tail.block.load(Ordering::Acquire);
                continue;
            }
            // If we're going to have to install the next block, allocate it in advance in order to
            // make the wait for other threads as short as possible.
            if offset + 1 == BLOCK_CAP && next_block.is_none() {
                next_block = Some(Box::new(Block::<T>::new()));
            }
            let new_tail = tail + (1 << SHIFT);
            // Try advancing the tail forward.
            match self.tail.index.compare_exchange_weak(
                tail,
                new_tail,
                Ordering::SeqCst,
                Ordering::Acquire,
            ) {
                Ok(_) => unsafe {
                    // If we've reached the end of the block, install the next one.
                    if offset + 1 == BLOCK_CAP {
                        let next_block = Box::into_raw(next_block.unwrap());
                        let next_index = new_tail.wrapping_add(1 << SHIFT);
                        self.tail.block.store(next_block, Ordering::Release);
                        self.tail.index.store(next_index, Ordering::Release);
                        (*block).next.store(next_block, Ordering::Release);
                    }
                    // Write the task into the slot.
                    let slot = (*block).slots.get_unchecked(offset);
                    slot.task.get().write(ManuallyDrop::new(task));
                    slot.state.fetch_or(WRITE, Ordering::Release);
                    return;
                },
                Err(t) => {
                    tail = t;
                    block = self.tail.block.load(Ordering::Acquire);
                    backoff.spin();
                }
            }
        }
    }
pub fn push ( & self , task : T ) { let backoff = Backoff :: new ( ) ; let mut tail = self . tail . index . load ( Ordering :: Acquire ) ; let mut block = self . tail . block . load ( Ordering :: Acquire ) ; let mut next_block = None ; loop { // Calculate the offset of the index into the block. let offset = ( tail >> SHIFT ) % LAP ; // If we reached the end of the block, wait until the next one is installed. if offset == BLOCK_CAP { backoff . snooze ( ) ; tail = self . tail . index . load ( Ordering :: Acquire ) ; block = self . tail . block . load ( Ordering :: Acquire ) ; continue ; } // If we're going to have to install the next block, allocate it in advance in order to // make the wait for other threads as short as possible. if offset + 1 == BLOCK_CAP && next_block . is_none ( ) { next_block = Some ( Box :: new ( Block :: < T > :: new ( ) ) ) ; } let new_tail = tail + ( 1 << SHIFT ) ; // Try advancing the tail forward. match self . tail . index . compare_exchange_weak ( tail , new_tail , Ordering :: SeqCst , Ordering :: Acquire , ) { Ok ( _ ) => unsafe { // If we've reached the end of the block, install the next one. if offset + 1 == BLOCK_CAP { let next_block = Box :: into_raw ( next_block . unwrap ( ) ) ; let next_index = new_tail . wrapping_add ( 1 << SHIFT ) ; self . tail . block . store ( next_block , Ordering :: Release ) ; self . tail . index . store ( next_index , Ordering :: Release ) ; ( * block ) . next . store ( next_block , Ordering :: Release ) ; } // Write the task into the slot. let slot = ( * block ) . slots . get_unchecked ( offset ) ; slot . task . get ( ) . write ( ManuallyDrop :: new ( task ) ) ; slot . state . fetch_or ( WRITE , Ordering :: Release ) ; return ; } , Err ( t ) => { tail = t ; block = self . tail . block . load ( Ordering :: Acquire ) ; backoff . spin ( ) ; } } } }
Pushes a task into the queue.
Pushes a task into the queue .
&self
task
T
&self
task
T