/// Steals a batch of tasks and pushes them into another 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 another 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 another 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<()> {
        // Load the front index.
        let mut f = self.inner.front.load(Ordering::Acquire);
        // A SeqCst fence is needed here.
        //
        // If the current thread is already pinned (reentrantly), we must manually issue the
        // fence. Otherwise, the following pinning will issue the fence anyway, so we don't
        // have to.
        if epoch::is_pinned() {
            atomic::fence(Ordering::SeqCst);
        }
        let guard = &epoch::pin();
        // Load the back index.
        let b = self.inner.back.load(Ordering::Acquire);
        // Is the queue empty?
        let len = b.wrapping_sub(f);
        if len <= 0 {
            return Steal::Empty;
        }
        // Reserve capacity for the stolen batch.
        let batch_size = cmp::min((len as usize + 1) / 2, MAX_BATCH);
        dest.reserve(batch_size);
        let mut batch_size = batch_size as isize;
        // Get the destination buffer and back index.
        let dest_buffer = dest.buffer.get();
        let mut dest_b = dest.inner.back.load(Ordering::Relaxed);
        // Load the buffer.
        let buffer = self.inner.buffer.load(Ordering::Acquire, guard);
        match self.flavor {
            // Steal a batch of tasks from the front at once.
            Flavor::Fifo => {
                // Copy the batch from the source to the destination buffer.
                match dest.flavor {
                    Flavor::Fifo => {
                        for i in 0..batch_size {
                            unsafe {
                                let task = buffer.deref().read(f.wrapping_add(i));
                                dest_buffer.write(dest_b.wrapping_add(i), task);
                            }
                        }
                    }
                    Flavor::Lifo => {
                        for i in 0..batch_size {
                            unsafe {
                                let task = buffer.deref().read(f.wrapping_add(i));
                                dest_buffer.write(dest_b.wrapping_add(batch_size - 1 - i), task);
                            }
                        }
                    }
                }
                // Try incrementing the front index to steal the batch.
                if self
                    .inner
                    .front
                    .compare_exchange(
                        f,
                        f.wrapping_add(batch_size),
                        Ordering::SeqCst,
                        Ordering::Relaxed,
                    )
                    .is_err()
                {
                    return Steal::Retry;
                }
                dest_b = dest_b.wrapping_add(batch_size);
            }
            // Steal a batch of tasks from the front one by one.
            Flavor::Lifo => {
                let size = batch_size;
                for i in 0..size {
                    // If this is not the first steal, check whether the queue is empty.
                    if i > 0 {
                        // We've already got the current front index. Now execute the fence to
                        // synchronize with other threads.
                        atomic::fence(Ordering::SeqCst);
                        // Load the back index.
                        let b = self.inner.back.load(Ordering::Acquire);
                        // Is the queue empty?
                        if b.wrapping_sub(f) <= 0 {
                            batch_size = i;
                            break;
                        }
                    }
                    // Read the task at the front.
                    let task = unsafe { buffer.deref().read(f) };
                    // Try incrementing the front index to steal the task.
                    if self
                        .inner
                        .front
                        .compare_exchange(f, f.wrapping_add(1), Ordering::SeqCst, Ordering::Relaxed)
                        .is_err()
                    {
                        // We didn't steal this task, forget it and break from the loop.
                        mem::forget(task);
                        batch_size = i;
                        break;
                    }
                    // Write the stolen task into the destination buffer.
                    unsafe {
                        dest_buffer.write(dest_b, task);
                    }
                    // Move the source front index and the destination back index one step forward.
                    f = f.wrapping_add(1);
                    dest_b = dest_b.wrapping_add(1);
                }
                // If we didn't steal anything, the operation needs to be retried.
                if batch_size == 0 {
                    return Steal::Retry;
                }
                // If stealing into a FIFO queue, stolen tasks need to be reversed.
                if dest.flavor == Flavor::Fifo {
                    for i in 0..batch_size / 2 {
                        unsafe {
                            let i1 = dest_b.wrapping_sub(batch_size - i);
                            let i2 = dest_b.wrapping_sub(i + 1);
                            let t1 = dest_buffer.read(i1);
                            let t2 = dest_buffer.read(i2);
                            dest_buffer.write(i1, t2);
                            dest_buffer.write(i2, t1);
                        }
                    }
                }
            }
        }
        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, Ordering::Release);
        // Return with success.
        Steal::Success(())
    }
pub fn steal_batch ( & self , dest : & Worker < T > ) -> Steal < ( ) > { // Load the front index. let mut f = self . inner . front . load ( Ordering :: Acquire ) ; // A SeqCst fence is needed here. // // If the current thread is already pinned (reentrantly), we must manually issue the // fence. Otherwise, the following pinning will issue the fence anyway, so we don't // have to. if epoch :: is_pinned ( ) { atomic :: fence ( Ordering :: SeqCst ) ; } let guard = & epoch :: pin ( ) ; // Load the back index. let b = self . inner . back . load ( Ordering :: Acquire ) ; // Is the queue empty? let len = b . wrapping_sub ( f ) ; if len <= 0 { return Steal :: Empty ; } // Reserve capacity for the stolen batch. let batch_size = cmp :: min ( ( len as usize + 1 ) / 2 , MAX_BATCH ) ; dest . reserve ( batch_size ) ; let mut batch_size = batch_size as isize ; // Get the destination buffer and back index. let dest_buffer = dest . buffer . get ( ) ; let mut dest_b = dest . inner . back . load ( Ordering :: Relaxed ) ; // Load the buffer. let buffer = self . inner . buffer . load ( Ordering :: Acquire , guard ) ; match self . flavor { // Steal a batch of tasks from the front at once. Flavor :: Fifo => { // Copy the batch from the source to the destination buffer. match dest . flavor { Flavor :: Fifo => { for i in 0 .. batch_size { unsafe { let task = buffer . deref ( ) . read ( f . wrapping_add ( i ) ) ; dest_buffer . write ( dest_b . wrapping_add ( i ) , task ) ; } } } Flavor :: Lifo => { for i in 0 .. batch_size { unsafe { let task = buffer . deref ( ) . read ( f . wrapping_add ( i ) ) ; dest_buffer . write ( dest_b . wrapping_add ( batch_size - 1 - i ) , task ) ; } } } } // Try incrementing the front index to steal the batch. if self . inner . front . compare_exchange ( f , f . wrapping_add ( batch_size ) , Ordering :: SeqCst , Ordering :: Relaxed , ) . is_err ( ) { return Steal :: Retry ; } dest_b = dest_b . wrapping_add ( batch_size ) ; } // Steal a batch of tasks from the front one by one. Flavor :: Lifo => { let size = batch_size ; for i in 0 .. size { // If this is not the first steal, check whether the queue is empty. if i > 0 { // We've already got the current front index. Now execute the fence to // synchronize with other threads. atomic :: fence ( Ordering :: SeqCst ) ; // Load the back index. let b = self . inner . back . load ( Ordering :: Acquire ) ; // Is the queue empty? if b . wrapping_sub ( f ) <= 0 { batch_size = i ; break ; } } // Read the task at the front. let task = unsafe { buffer . deref ( ) . read ( f ) } ; // Try incrementing the front index to steal the task. if self . inner . front . compare_exchange ( f , f . wrapping_add ( 1 ) , Ordering :: SeqCst , Ordering :: Relaxed ) . is_err ( ) { // We didn't steal this task, forget it and break from the loop. mem :: forget ( task ) ; batch_size = i ; break ; } // Write the stolen task into the destination buffer. unsafe { dest_buffer . write ( dest_b , task ) ; } // Move the source front index and the destination back index one step forward. f = f . wrapping_add ( 1 ) ; dest_b = dest_b . wrapping_add ( 1 ) ; } // If we didn't steal anything, the operation needs to be retried. if batch_size == 0 { return Steal :: Retry ; } // If stealing into a FIFO queue, stolen tasks need to be reversed. if dest . flavor == Flavor :: Fifo { for i in 0 .. batch_size / 2 { unsafe { let i1 = dest_b . wrapping_sub ( batch_size - i ) ; let i2 = dest_b . wrapping_sub ( i + 1 ) ; let t1 = dest_buffer . read ( i1 ) ; let t2 = dest_buffer . read ( i2 ) ; dest_buffer . write ( i1 , t2 ) ; dest_buffer . write ( i2 , t1 ) ; } } } } } 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 , Ordering :: Release ) ; // Return with success. Steal :: Success ( ( ) ) }
Steals a batch of tasks and pushes them into another worker.
Steals a batch of tasks and pushes them into another worker .
&self
dest
&Worker<T>