fn poll(&self) -> Poll<FutItem, FutError> {
        let mut map_l = lock_c!(self.channels);
        if map_l.len() < self.position.load(Ordering::Relaxed) {
            self.position.store(0, Ordering::Relaxed);
        }

        let start_pos = self.position.load(Ordering::Relaxed);
        let mut pos = 0;

        let mut worker_counter = 0;
        let mut roundtrip = 0;
        let mut no_work = true;
        let mut idle = false;

        while self.workers_active.load(Ordering::Relaxed) < self.max_worker.load(Ordering::Relaxed)
            && !idle
        {
            map_l.retain(|_, channel| {
                // skip to position from last poll
                if roundtrip == 0 && pos < start_pos {
                    return true;
                }
                let mut connected = true;
                match channel.recv.consume() {
                    Ok(w) => {
                        channel.len.fetch_sub(1, Ordering::SeqCst);
                        no_work = false;
                        self.workers_active.fetch_add(1, Ordering::SeqCst);
                        worker_counter += 1;
                        let worker_c = self.workers_active.clone();
                        let task = task::current();
                        let work_fn = self.worker_fn.clone();
                        let work_fn_final = self.worker_fn_finalize.clone();
                        let mut cancel_recv = channel.cancel_bus.add_rx();
                        thread::spawn(move || {
                            let result: R = work_fn(w);
                            if cancel_recv.try_recv().is_err() {
                                if let Some(finalizer) = work_fn_final.as_ref() {
                                    finalizer(result);
                                }
                            }
                            worker_c.fetch_sub(1, Ordering::SeqCst);
                            task.notify();
                        });
                    }
                    Err(ConsumeError::Empty) => (),
                    Err(ConsumeError::Disconnected) => connected = false,
                }
                pos += 1;
                connected
            });
            pos = 0;

            if no_work && roundtrip >= 1 {
                idle = true;
            }
            roundtrip += 1;

            no_work = true;
        }
        let mut task_l = self.task.write().expect("Can't lock task!");
        *task_l = Some(task::current());
        drop(task_l);
        self.position.store(pos, Ordering::Relaxed);
        if self.exit_on_idle && map_l.len() == 0 && self.workers_active.load(Ordering::Relaxed) == 0
        {
            Ok(Async::Ready(()))
        } else {
            Ok(Async::NotReady)
        }
    }
/// Inner poll method, only to be called by future handler
Inner poll method, only to be called by future handler
Inner poll method only to be called by future handler
fn poll(&self) -> Poll<FutItem, FutError> {
        let mut map_l = lock_c!(self.channels);
        if map_l.len() < self.position.load(Ordering::Relaxed) {
            self.position.store(0, Ordering::Relaxed);
        }
        let start_pos = self.position.load(Ordering::Relaxed);
        let mut pos = 0;
        let mut worker_counter = 0;
        let mut roundtrip = 0;
        let mut no_work = true;
        let mut idle = false;
        while self.workers_active.load(Ordering::Relaxed) < self.max_worker.load(Ordering::Relaxed)
            && !idle
        {
            map_l.retain(|_, channel| {
                // skip to position from last poll
                if roundtrip == 0 && pos < start_pos {
                    return true;
                }
                let mut connected = true;
                match channel.recv.consume() {
                    Ok(w) => {
                        channel.len.fetch_sub(1, Ordering::SeqCst);
                        no_work = false;
                        self.workers_active.fetch_add(1, Ordering::SeqCst);
                        worker_counter += 1;
                        let worker_c = self.workers_active.clone();
                        let task = task::current();
                        let work_fn = self.worker_fn.clone();
                        let work_fn_final = self.worker_fn_finalize.clone();
                        let mut cancel_recv = channel.cancel_bus.add_rx();
                        thread::spawn(move || {
                            let result: R = work_fn(w);
                            if cancel_recv.try_recv().is_err() {
                                if let Some(finalizer) = work_fn_final.as_ref() {
                                    finalizer(result);
                                }
                            }
                            worker_c.fetch_sub(1, Ordering::SeqCst);
                            task.notify();
                        });
                    }
                    Err(ConsumeError::Empty) => (),
                    Err(ConsumeError::Disconnected) => connected = false,
                }
                pos += 1;
                connected
            });
            pos = 0;
            if no_work && roundtrip >= 1 {
                idle = true;
            }
            roundtrip += 1;
            no_work = true;
        }
        let mut task_l = self.task.write().expect("Can't lock task!");
        *task_l = Some(task::current());
        drop(task_l);
        self.position.store(pos, Ordering::Relaxed);
        if self.exit_on_idle && map_l.len() == 0 && self.workers_active.load(Ordering::Relaxed) == 0
        {
            Ok(Async::Ready(()))
        } else {
            Ok(Async::NotReady)
        }
    }
fn poll ( & self ) -> Poll < FutItem , FutError > { let mut map_l = lock_c ! ( self channels ) ; if map_l . len ( ) < self . position . load ( Ordering :: Relaxed ) { self . position . store ( 0 , Ordering :: Relaxed ) ; } let start_pos = self . position . load ( Ordering :: Relaxed ) ; let mut pos = 0 ; let mut worker_counter = 0 ; let mut roundtrip = 0 ; let mut no_work = true ; let mut idle = false ; while self . workers_active . load ( Ordering :: Relaxed ) < self . max_worker . load ( Ordering :: Relaxed ) && ! idle { map_l . retain ( | _ , channel | { // skip to position from last poll if roundtrip == 0 && pos < start_pos { return true ; } let mut connected = true ; match channel . recv . consume ( ) { Ok ( w ) => { channel . len . fetch_sub ( 1 , Ordering :: SeqCst ) ; no_work = false ; self . workers_active . fetch_add ( 1 , Ordering :: SeqCst ) ; worker_counter += 1 ; let worker_c = self . workers_active . clone ( ) ; let task = task :: current ( ) ; let work_fn = self . worker_fn . clone ( ) ; let work_fn_final = self . worker_fn_finalize . clone ( ) ; let mut cancel_recv = channel . cancel_bus . add_rx ( ) ; thread :: spawn ( move | | { let result : R = work_fn ( w ) ; if cancel_recv . try_recv ( ) . is_err ( ) { if let Some ( finalizer ) = work_fn_final . as_ref ( ) { finalizer ( result ) ; } } worker_c . fetch_sub ( 1 , Ordering :: SeqCst ) ; task . notify ( ) ; } ) ; } Err ( ConsumeError :: Empty ) => ( ) , Err ( ConsumeError :: Disconnected ) => connected = false , } pos += 1 ; connected } ) ; pos = 0 ; if no_work && roundtrip >= 1 { idle = true ; } roundtrip += 1 ; no_work = true ; } let mut task_l = self . task . write ( ) . expect ( " " ) ; * task_l = Some ( task :: current ( ) ) ; drop ( task_l ) ; self . position . store ( pos , Ordering :: Relaxed ) ; if self . exit_on_idle && map_l . len ( ) == 0 && self . workers_active . load ( Ordering :: Relaxed ) == 0 { Ok ( Async :: Ready ( ( ) ) ) } else { Ok ( Async :: NotReady ) } }
Inner poll method, only to be called by future handler
Inner poll method only to be called by future handler
&self
&self