/// Steals a task from the queue.
Steals a task from the queue.
Steals a task from the queue .
pub fn steal(&self) -> Steal<T> {
        let mut head;
        let mut block;
        let mut offset;
        let backoff = Backoff::new();
        loop {
            head = self.head.index.load(Ordering::Acquire);
            block = self.head.block.load(Ordering::Acquire);
            // Calculate the offset of the index into the block.
            offset = (head >> SHIFT) % LAP;
            // If we reached the end of the block, wait until the next one is installed.
            if offset == BLOCK_CAP {
                backoff.snooze();
            } else {
                break;
            }
        }
        let mut new_head = head + (1 << SHIFT);
        if new_head & HAS_NEXT == 0 {
            atomic::fence(Ordering::SeqCst);
            let tail = self.tail.index.load(Ordering::Relaxed);
            // If the tail equals the head, that means the queue is empty.
            if head >> SHIFT == tail >> SHIFT {
                return Steal::Empty;
            }
            // If head and tail are not in the same block, set `HAS_NEXT` in head.
            if (head >> SHIFT) / LAP != (tail >> SHIFT) / LAP {
                new_head |= HAS_NEXT;
            }
        }
        // Try moving the head index forward.
        if self
            .head
            .index
            .compare_exchange_weak(head, new_head, Ordering::SeqCst, Ordering::Acquire)
            .is_err()
        {
            return Steal::Retry;
        }
        unsafe {
            // If we've reached the end of the block, move to the next one.
            if offset + 1 == BLOCK_CAP {
                let next = (*block).wait_next();
                let mut next_index = (new_head & !HAS_NEXT).wrapping_add(1 << SHIFT);
                if !(*next).next.load(Ordering::Relaxed).is_null() {
                    next_index |= HAS_NEXT;
                }
                self.head.block.store(next, Ordering::Release);
                self.head.index.store(next_index, Ordering::Release);
            }
            // Read the task.
            let slot = (*block).slots.get_unchecked(offset);
            slot.wait_write();
            let m = slot.task.get().read();
            let task = ManuallyDrop::into_inner(m);
            // Destroy the block if we've reached the end, or if another thread wanted to destroy
            // but couldn't because we were busy reading from the slot.
            if offset + 1 == BLOCK_CAP {
                Block::destroy(block, offset);
            } else if slot.state.fetch_or(READ, Ordering::AcqRel) & DESTROY != 0 {
                Block::destroy(block, offset);
            }
            Steal::Success(task)
        }
    }
pub fn steal ( & self ) -> Steal < T > { let mut head ; let mut block ; let mut offset ; let backoff = Backoff :: new ( ) ; loop { head = self . head . index . load ( Ordering :: Acquire ) ; block = self . head . block . load ( Ordering :: Acquire ) ; // Calculate the offset of the index into the block. offset = ( head >> SHIFT ) % LAP ; // If we reached the end of the block, wait until the next one is installed. if offset == BLOCK_CAP { backoff . snooze ( ) ; } else { break ; } } let mut new_head = head + ( 1 << SHIFT ) ; if new_head & HAS_NEXT == 0 { atomic :: fence ( Ordering :: SeqCst ) ; let tail = self . tail . index . load ( Ordering :: Relaxed ) ; // If the tail equals the head, that means the queue is empty. if head >> SHIFT == tail >> SHIFT { return Steal :: Empty ; } // If head and tail are not in the same block, set `HAS_NEXT` in head. if ( head >> SHIFT ) / LAP != ( tail >> SHIFT ) / LAP { new_head |= HAS_NEXT ; } } // Try moving the head index forward. if self . head . index . compare_exchange_weak ( head , new_head , Ordering :: SeqCst , Ordering :: Acquire ) . is_err ( ) { return Steal :: Retry ; } unsafe { // If we've reached the end of the block, move to the next one. if offset + 1 == BLOCK_CAP { let next = ( * block ) . wait_next ( ) ; let mut next_index = ( new_head & ! HAS_NEXT ) . wrapping_add ( 1 << SHIFT ) ; if ! ( * next ) . next . load ( Ordering :: Relaxed ) . is_null ( ) { next_index |= HAS_NEXT ; } self . head . block . store ( next , Ordering :: Release ) ; self . head . index . store ( next_index , Ordering :: Release ) ; } // Read the task. let slot = ( * block ) . slots . get_unchecked ( offset ) ; slot . wait_write ( ) ; let m = slot . task . get ( ) . read ( ) ; let task = ManuallyDrop :: into_inner ( m ) ; // Destroy the block if we've reached the end, or if another thread wanted to destroy // but couldn't because we were busy reading from the slot. if offset + 1 == BLOCK_CAP { Block :: destroy ( block , offset ) ; } else if slot . state . fetch_or ( READ , Ordering :: AcqRel ) & DESTROY != 0 { Block :: destroy ( block , offset ) ; } Steal :: Success ( task ) } }
Steals a task from the queue.
Steals a task from the queue .
&self