pub fn steal_batch_and_pop_with_amount(&self, dest: &Worker<T>, amount: usize) -> Steal<T> {
        // 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();

        // Shadow the requested amount;
        let mut amount = amount;

        // 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;
        }

        let default_time_ahead = (len as usize - 1) / 2;
        if amount > default_time_ahead as usize {
            amount = default_time_ahead;
        }

        // Reserve capacity for the stolen batch.
        let batch_size = cmp::min(amount, MAX_BATCH - 1);
        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);

        // Read the task at the front.
        let mut task = unsafe { buffer.deref().read(f) };

        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 + 1));
                                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 + 1));
                                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 + 1),
                        Ordering::SeqCst,
                        Ordering::Relaxed,
                    )
                    .is_err()
                {
                    // We didn't steal this task, forget it.
                    mem::forget(task);
                    return Steal::Retry;
                }

                dest_b = dest_b.wrapping_add(batch_size);
            }

            // Steal a batch of tasks from the front one by one.
            Flavor::Lifo => {
                // 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.
                    mem::forget(task);
                    return Steal::Retry;
                }

                // Move the front index one step forward.
                f = f.wrapping_add(1);

                // Repeat the same procedure for the batch steals.
                let size = batch_size;
                for i in 0..size {
                    // 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 tmp = 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(tmp);
                        batch_size = i;
                        break;
                    }

                    // Write the previously stolen task into the destination buffer.
                    unsafe {
                        dest_buffer.write(dest_b, mem::replace(&mut task, tmp));
                    }

                    // 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 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(task)
    }
///
/// Steals a batch of tasks and amend to the given worker's run queue with the specified amount.
/// If requested amount is above the threshold of time-ahead it defaults to time-ahead amount.
Steals a batch of tasks and amend to the given worker's run queue with the specified amount.
If requested amount is above the threshold of time-ahead it defaults to time-ahead amount.
Steals a batch of tasks and amend to the given worker ' s run queue with the specified amount . If requested amount is above the threshold of time - ahead it defaults to time - ahead amount .
pub fn steal_batch_and_pop_with_amount(&self, dest: &Worker<T>, amount: usize) -> Steal<T> {
        // 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();
        // Shadow the requested amount;
        let mut amount = amount;
        // 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;
        }
        let default_time_ahead = (len as usize - 1) / 2;
        if amount > default_time_ahead as usize {
            amount = default_time_ahead;
        }
        // Reserve capacity for the stolen batch.
        let batch_size = cmp::min(amount, MAX_BATCH - 1);
        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);
        // Read the task at the front.
        let mut task = unsafe { buffer.deref().read(f) };
        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 + 1));
                                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 + 1));
                                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 + 1),
                        Ordering::SeqCst,
                        Ordering::Relaxed,
                    )
                    .is_err()
                {
                    // We didn't steal this task, forget it.
                    mem::forget(task);
                    return Steal::Retry;
                }
                dest_b = dest_b.wrapping_add(batch_size);
            }
            // Steal a batch of tasks from the front one by one.
            Flavor::Lifo => {
                // 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.
                    mem::forget(task);
                    return Steal::Retry;
                }
                // Move the front index one step forward.
                f = f.wrapping_add(1);
                // Repeat the same procedure for the batch steals.
                let size = batch_size;
                for i in 0..size {
                    // 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 tmp = 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(tmp);
                        batch_size = i;
                        break;
                    }
                    // Write the previously stolen task into the destination buffer.
                    unsafe {
                        dest_buffer.write(dest_b, mem::replace(&mut task, tmp));
                    }
                    // 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 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(task)
    }
pub fn steal_batch_and_pop_with_amount ( & self , dest : & Worker < T > , amount : usize ) -> Steal < T > { // 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 ( ) ; // Shadow the requested amount; let mut amount = amount ; // 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 ; } let default_time_ahead = ( len as usize - 1 ) / 2 ; if amount > default_time_ahead as usize { amount = default_time_ahead ; } // Reserve capacity for the stolen batch. let batch_size = cmp :: min ( amount , MAX_BATCH - 1 ) ; 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 ) ; // Read the task at the front. let mut task = unsafe { buffer . deref ( ) . read ( f ) } ; 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 + 1 ) ) ; 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 + 1 ) ) ; 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 + 1 ) , Ordering :: SeqCst , Ordering :: Relaxed , ) . is_err ( ) { // We didn't steal this task, forget it. mem :: forget ( task ) ; return Steal :: Retry ; } dest_b = dest_b . wrapping_add ( batch_size ) ; } // Steal a batch of tasks from the front one by one. Flavor :: Lifo => { // 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. mem :: forget ( task ) ; return Steal :: Retry ; } // Move the front index one step forward. f = f . wrapping_add ( 1 ) ; // Repeat the same procedure for the batch steals. let size = batch_size ; for i in 0 .. size { // 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 tmp = 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 ( tmp ) ; batch_size = i ; break ; } // Write the previously stolen task into the destination buffer. unsafe { dest_buffer . write ( dest_b , mem :: replace ( & mut task , tmp ) ) ; } // 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 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 ( task ) }
Steals a batch of tasks and amend to the given worker's run queue with the specified amount.
Steals a batch of tasks and amend to the given worker ' s run queue with the specified amount .
&self
dest
&Worker<T>
amount
usize
&self
dest
&Worker<T>
amount
usize