/// Steals a batch of tasks and pushes them into a worker.
///
/// How many tasks exactly will be stolen is not specified. That said, this method will try to
/// steal around half of the tasks in the queue, but also not more than some constant limit.
Steals a batch of tasks and pushes them into a worker.
How many tasks exactly will be stolen is not specified. That said, this method will try to
steal around half of the tasks in the queue, but also not more than some constant limit.
Steals a batch of tasks and pushes them into a worker . How many tasks exactly will be stolen is not specified . That said this method will try to steal around half of the tasks in the queue but also not more than some constant limit .
pub fn steal_batch(&self, dest: &Worker<T>) -> Steal<()> {
        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;
        let advance;
        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. Also, calculate
            // the right batch size to steal.
            if (head >> SHIFT) / LAP != (tail >> SHIFT) / LAP {
                new_head |= HAS_NEXT;
                // We can steal all tasks till the end of the block.
                advance = (BLOCK_CAP - offset).min(MAX_BATCH);
            } else {
                let len = (tail - head) >> SHIFT;
                // Steal half of the available tasks.
                advance = ((len + 1) / 2).min(MAX_BATCH);
            }
        } else {
            // We can steal all tasks till the end of the block.
            advance = (BLOCK_CAP - offset).min(MAX_BATCH);
        }
        new_head += advance << SHIFT;
        let new_offset = offset + advance;
        // 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;
        }
        // Reserve capacity for the stolen batch.
        let batch_size = new_offset - offset;
        dest.reserve(batch_size);
        // Get the destination buffer and back index.
        let dest_buffer = dest.buffer.get();
        let dest_b = dest.inner.back.load(Ordering::Relaxed);
        unsafe {
            // If we've reached the end of the block, move to the next one.
            if new_offset == 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);
            }
            // Copy values from the injector into the destination queue.
            match dest.flavor {
                Flavor::Fifo => {
                    for i in 0..batch_size {
                        // Read the task.
                        let slot = (*block).slots.get_unchecked(offset + i);
                        slot.wait_write();
                        let m = slot.task.get().read();
                        let task = ManuallyDrop::into_inner(m);
                        // Write it into the destination queue.
                        dest_buffer.write(dest_b.wrapping_add(i as isize), task);
                    }
                }
                Flavor::Lifo => {
                    for i in 0..batch_size {
                        // Read the task.
                        let slot = (*block).slots.get_unchecked(offset + i);
                        slot.wait_write();
                        let m = slot.task.get().read();
                        let task = ManuallyDrop::into_inner(m);
                        // Write it into the destination queue.
                        dest_buffer.write(dest_b.wrapping_add((batch_size - 1 - i) as isize), task);
                    }
                }
            }
            atomic::fence(Ordering::Release);
            // Update the back index in the destination queue.
            //
            // This ordering could be `Relaxed`, but then thread sanitizer would falsely report
            // data races because it doesn't understand fences.
            dest.inner
                .back
                .store(dest_b.wrapping_add(batch_size as isize), Ordering::Release);
            // 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 new_offset == BLOCK_CAP {
                Block::destroy(block, offset);
            } else {
                for i in offset..new_offset {
                    let slot = (*block).slots.get_unchecked(i);
                    if slot.state.fetch_or(READ, Ordering::AcqRel) & DESTROY != 0 {
                        Block::destroy(block, offset);
                        break;
                    }
                }
            }
            Steal::Success(())
        }
    }
pub fn steal_batch ( & self , dest : & Worker < T > ) -> Steal < ( ) > { 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 ; let advance ; 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. Also, calculate // the right batch size to steal. if ( head >> SHIFT ) / LAP != ( tail >> SHIFT ) / LAP { new_head |= HAS_NEXT ; // We can steal all tasks till the end of the block. advance = ( BLOCK_CAP - offset ) . min ( MAX_BATCH ) ; } else { let len = ( tail - head ) >> SHIFT ; // Steal half of the available tasks. advance = ( ( len + 1 ) / 2 ) . min ( MAX_BATCH ) ; } } else { // We can steal all tasks till the end of the block. advance = ( BLOCK_CAP - offset ) . min ( MAX_BATCH ) ; } new_head += advance << SHIFT ; let new_offset = offset + advance ; // 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 ; } // Reserve capacity for the stolen batch. let batch_size = new_offset - offset ; dest . reserve ( batch_size ) ; // Get the destination buffer and back index. let dest_buffer = dest . buffer . get ( ) ; let dest_b = dest . inner . back . load ( Ordering :: Relaxed ) ; unsafe { // If we've reached the end of the block, move to the next one. if new_offset == 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 ) ; } // Copy values from the injector into the destination queue. match dest . flavor { Flavor :: Fifo => { for i in 0 .. batch_size { // Read the task. let slot = ( * block ) . slots . get_unchecked ( offset + i ) ; slot . wait_write ( ) ; let m = slot . task . get ( ) . read ( ) ; let task = ManuallyDrop :: into_inner ( m ) ; // Write it into the destination queue. dest_buffer . write ( dest_b . wrapping_add ( i as isize ) , task ) ; } } Flavor :: Lifo => { for i in 0 .. batch_size { // Read the task. let slot = ( * block ) . slots . get_unchecked ( offset + i ) ; slot . wait_write ( ) ; let m = slot . task . get ( ) . read ( ) ; let task = ManuallyDrop :: into_inner ( m ) ; // Write it into the destination queue. dest_buffer . write ( dest_b . wrapping_add ( ( batch_size - 1 - i ) as isize ) , task ) ; } } } atomic :: fence ( Ordering :: Release ) ; // Update the back index in the destination queue. // // This ordering could be `Relaxed`, but then thread sanitizer would falsely report // data races because it doesn't understand fences. dest . inner . back . store ( dest_b . wrapping_add ( batch_size as isize ) , Ordering :: Release ) ; // 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 new_offset == BLOCK_CAP { Block :: destroy ( block , offset ) ; } else { for i in offset .. new_offset { let slot = ( * block ) . slots . get_unchecked ( i ) ; if slot . state . fetch_or ( READ , Ordering :: AcqRel ) & DESTROY != 0 { Block :: destroy ( block , offset ) ; break ; } } } Steal :: Success ( ( ) ) } }
Steals a batch of tasks and pushes them into a worker.
Steals a batch of tasks and pushes them into a worker .
&self
dest
&Worker<T>