unsafe fn init_splitters(&mut self, inputs: &InputMap) -> (SplitterMap, i64) {
        let mut num_items = vec![0; inputs.len()];
        let splitters = inputs
            .iter()
            .enumerate()
            .map(|(i, (arg_id, (task_args, split_type_param)))| {
                if split_type_param.ty.is_broadcast() {
                    trace!("No splitter for arg {:?} (broadcast)", arg_id);
                    num_items[i] = COMPOSER_INFINITE_ITEMS;
                    return (*arg_id, None);
                } else if self.futures.contains(arg_id) {
                    // don't split values that were returned by a function in the pipeline -
                    // they're already split.
                    trace!("No splitter for arg {:?} (returned from pipeline)", arg_id);
                    num_items[i] = COMPOSER_INFINITE_ITEMS;
                    return (*arg_id, None);
                }

                let runtime = split_type_param.ty.runtime_info();
                let splitter_init_fn = mem::transmute::<usize, SplitterInitFn>(runtime.initializer);
                let argval = self.arguments.get(arg_id).unwrap();

                // Construct the argument struct for the splitter initializer.
                // Note that this struct needs to be packed.
                let mut splitter_arg_struct = vec![];
                for arg_index in split_type_param.arguments.iter() {
                    let ref splitter_arg_id = task_args[*arg_index];
                    let ref data = self.arguments[splitter_arg_id];
                    splitter_arg_struct.extend(data.iter().cloned());
                }

                trace!("About to call splitter init for arg {:?} at pointer {:x}",
                       arg_id,
                       runtime.initializer);

                let arg_val_ptr = argval.as_ptr() as *const c_void;
                let init_args_ptr = splitter_arg_struct.as_ptr() as *const c_void;

                // Initialize the splitter.
                let splitter = (splitter_init_fn)(arg_val_ptr,
                                                  init_args_ptr,
                                                  num_items.as_mut_ptr().offset(i as isize));
                (*arg_id, Some(SendSyncPointer(splitter)))
            })
        .collect();

        // TODO this check should eventually be removed.
        let num_items = {
            let mut final_items = None;
            for item in num_items {
                if item != COMPOSER_INFINITE_ITEMS {
                    if let Some(final_items) = final_items {
                        assert_eq!(final_items, item);
                    } else {
                        final_items = Some(item);
                    }
                }
            }
            final_items.unwrap()
        };
        (splitters, num_items)
    }
/// Returns initialized splitters for each argument and the number of items they produce.
///
/// If the argument is constructed inside the pipeline or has a split type of `broadcast`, it
/// won't have a splitter.
///
/// Currently, this method also checks that each splitter returns the same number of items,
/// though this requirement is not strictly necessary.
Returns initialized splitters for each argument and the number of items they produce.
If the argument is constructed inside the pipeline or has a split type of `broadcast`, it
won't have a splitter.

Currently, this method also checks that each splitter returns the same number of items,
though this requirement is not strictly necessary.
Returns initialized splitters for each argument and the number of items they produce . If the argument is constructed inside the pipeline or has a split type of ` broadcast ` it won ' t have a splitter . Currently this method also checks that each splitter returns the same number of items though this requirement is not strictly necessary .
unsafe fn init_splitters(&mut self, inputs: &InputMap) -> (SplitterMap, i64) {
        let mut num_items = vec![0; inputs.len()];
        let splitters = inputs
            .iter()
            .enumerate()
            .map(|(i, (arg_id, (task_args, split_type_param)))| {
                if split_type_param.ty.is_broadcast() {
                    trace!("No splitter for arg {:?} (broadcast)", arg_id);
                    num_items[i] = COMPOSER_INFINITE_ITEMS;
                    return (*arg_id, None);
                } else if self.futures.contains(arg_id) {
                    // don't split values that were returned by a function in the pipeline -
                    // they're already split.
                    trace!("No splitter for arg {:?} (returned from pipeline)", arg_id);
                    num_items[i] = COMPOSER_INFINITE_ITEMS;
                    return (*arg_id, None);
                }
                let runtime = split_type_param.ty.runtime_info();
                let splitter_init_fn = mem::transmute::<usize, SplitterInitFn>(runtime.initializer);
                let argval = self.arguments.get(arg_id).unwrap();
                // Construct the argument struct for the splitter initializer.
                // Note that this struct needs to be packed.
                let mut splitter_arg_struct = vec![];
                for arg_index in split_type_param.arguments.iter() {
                    let ref splitter_arg_id = task_args[*arg_index];
                    let ref data = self.arguments[splitter_arg_id];
                    splitter_arg_struct.extend(data.iter().cloned());
                }
                trace!("About to call splitter init for arg {:?} at pointer {:x}",
                       arg_id,
                       runtime.initializer);
                let arg_val_ptr = argval.as_ptr() as *const c_void;
                let init_args_ptr = splitter_arg_struct.as_ptr() as *const c_void;
                // Initialize the splitter.
                let splitter = (splitter_init_fn)(arg_val_ptr,
                                                  init_args_ptr,
                                                  num_items.as_mut_ptr().offset(i as isize));
                (*arg_id, Some(SendSyncPointer(splitter)))
            })
        .collect();
        // TODO this check should eventually be removed.
        let num_items = {
            let mut final_items = None;
            for item in num_items {
                if item != COMPOSER_INFINITE_ITEMS {
                    if let Some(final_items) = final_items {
                        assert_eq!(final_items, item);
                    } else {
                        final_items = Some(item);
                    }
                }
            }
            final_items.unwrap()
        };
        (splitters, num_items)
    }
unsafe fn init_splitters ( & mut self , inputs : & InputMap ) -> ( SplitterMap , i64 ) { let mut num_items = vec ! [ 0 inputs len ( ) ] ; let splitters = inputs . iter ( ) . enumerate ( ) . map ( | ( i , ( arg_id , ( task_args , split_type_param ) ) ) | { if split_type_param . ty . is_broadcast ( ) { trace ! ( " " arg_id ) ; num_items [ i ] = COMPOSER_INFINITE_ITEMS ; return ( * arg_id , None ) ; } else if self . futures . contains ( arg_id ) { // don't split values that were returned by a function in the pipeline - // they're already split. trace ! ( " " arg_id ) ; num_items [ i ] = COMPOSER_INFINITE_ITEMS ; return ( * arg_id , None ) ; } let runtime = split_type_param . ty . runtime_info ( ) ; let splitter_init_fn = mem :: transmute :: < usize , SplitterInitFn > ( runtime . initializer ) ; let argval = self . arguments . get ( arg_id ) . unwrap ( ) ; // Construct the argument struct for the splitter initializer. // Note that this struct needs to be packed. let mut splitter_arg_struct = vec ! [ ] ; for arg_index in split_type_param . arguments . iter ( ) { let ref splitter_arg_id = task_args [ * arg_index ] ; let ref data = self . arguments [ splitter_arg_id ] ; splitter_arg_struct . extend ( data . iter ( ) . cloned ( ) ) ; } trace ! ( " " arg_id runtime initializer ) ; let arg_val_ptr = argval . as_ptr ( ) as * const c_void ; let init_args_ptr = splitter_arg_struct . as_ptr ( ) as * const c_void ; // Initialize the splitter. let splitter = ( splitter_init_fn ) ( arg_val_ptr , init_args_ptr , num_items . as_mut_ptr ( ) . offset ( i as isize ) ) ; ( * arg_id , Some ( SendSyncPointer ( splitter ) ) ) } ) . collect ( ) ; // TODO this check should eventually be removed. let num_items = { let mut final_items = None ; for item in num_items { if item != COMPOSER_INFINITE_ITEMS { if let Some ( final_items ) = final_items { assert_eq ! ( final_items item ) ; } else { final_items = Some ( item ) ; } } } final_items . unwrap ( ) } ; ( splitters , num_items ) }
Returns initialized splitters for each argument and the number of items they produce.
Returns initialized splitters for each argument and the number of items they produce .
&mut self
inputs
&InputMap