pub(crate) fn collect_edges(
    train: &SvmScanner,
    featurizer: &Featurizer,
) -> (Vec<Edge>, Vec<usize>, GraphStats) {
    let nthreads = train.nthreads();
    let ref writers: Vec<Mutex<EdgeMap>> = repeat_with(|| {
        Mutex::new(HashMap::with_hasher(
            BuildHasherDefault::<ThreadUnsafeHasher>::default(),
        ))
    })
    .take(nthreads)
    .collect();
    let ref shared_stats = Mutex::new(GraphStats::default());

    {
        train.for_each(
            || EdgeCollector::new(featurizer, writers, shared_stats, nthreads),
            EdgeCollector::consume_line,
            EdgeCollector::finish,
        );
    }

    let lens: Vec<usize> = writers
        .iter()
        .map(|mutex| {
            let writer = mutex.lock().unwrap();
            writer.len()
        })
        .collect();
    let mut edges = vec![Edge::default(); lens.iter().sum::<usize>()];
    let mut frequencies = vec![0usize; edges.len()];
    {
        let ptr = edges.as_mut_ptr();
        let fptr = frequencies.as_mut_ptr();

        let edge_slices: Vec<_> = lens
            .into_iter()
            .scan(0, |state, size| {
                let start = *state;
                let stop = start + size;
                *state = stop;
                // SAFETY: each slice refers to a disjoint region in the vec
                Some((
                    unsafe { slice::from_raw_parts_mut(ptr.add(start), size) },
                    unsafe { slice::from_raw_parts_mut(fptr.add(start), size) },
                ))
            })
            .collect();

        edge_slices
            .into_par_iter()
            // note this zipped iter is of shorter length (intentionally)
            .zip(writers.into_par_iter())
            .for_each(|((edge_slice, freq_slice), writer)| {
                let writer = writer.lock().unwrap();
                for (i, (e, s)) in writer.iter().enumerate() {
                    edge_slice[i] = e.clone();
                    freq_slice[i] = s.count;
                }
            });

        if cfg!(debug_assertions) {
            let in_vec: HashSet<Edge> = edges.iter().cloned().collect();
            for w in writers {
                let orig_set = w.lock().unwrap();
                for e in orig_set.keys() {
                    debug_assert!(in_vec.contains(e), "{} {}", e.left(), e.right());
                }
            }
        }
    }
    let unlocked_stats = shared_stats.lock().unwrap();
    (edges, frequencies, unlocked_stats.clone())
}
/// Return edge set, frequency of edges, and stats on features active across row (recall
/// features are feature hashes).
Return edge set, frequency of edges, and stats on features active across row (recall
features are feature hashes).
Return edge set frequency of edges and stats on features active across row ( recall features are feature hashes ) .
pub(crate) fn collect_edges(
    train: &SvmScanner,
    featurizer: &Featurizer,
) -> (Vec<Edge>, Vec<usize>, GraphStats) {
    let nthreads = train.nthreads();
    let ref writers: Vec<Mutex<EdgeMap>> = repeat_with(|| {
        Mutex::new(HashMap::with_hasher(
            BuildHasherDefault::<ThreadUnsafeHasher>::default(),
        ))
    })
    .take(nthreads)
    .collect();
    let ref shared_stats = Mutex::new(GraphStats::default());
    {
        train.for_each(
            || EdgeCollector::new(featurizer, writers, shared_stats, nthreads),
            EdgeCollector::consume_line,
            EdgeCollector::finish,
        );
    }
    let lens: Vec<usize> = writers
        .iter()
        .map(|mutex| {
            let writer = mutex.lock().unwrap();
            writer.len()
        })
        .collect();
    let mut edges = vec![Edge::default(); lens.iter().sum::<usize>()];
    let mut frequencies = vec![0usize; edges.len()];
    {
        let ptr = edges.as_mut_ptr();
        let fptr = frequencies.as_mut_ptr();
        let edge_slices: Vec<_> = lens
            .into_iter()
            .scan(0, |state, size| {
                let start = *state;
                let stop = start + size;
                *state = stop;
                // SAFETY: each slice refers to a disjoint region in the vec
                Some((
                    unsafe { slice::from_raw_parts_mut(ptr.add(start), size) },
                    unsafe { slice::from_raw_parts_mut(fptr.add(start), size) },
                ))
            })
            .collect();
        edge_slices
            .into_par_iter()
            // note this zipped iter is of shorter length (intentionally)
            .zip(writers.into_par_iter())
            .for_each(|((edge_slice, freq_slice), writer)| {
                let writer = writer.lock().unwrap();
                for (i, (e, s)) in writer.iter().enumerate() {
                    edge_slice[i] = e.clone();
                    freq_slice[i] = s.count;
                }
            });
        if cfg!(debug_assertions) {
            let in_vec: HashSet<Edge> = edges.iter().cloned().collect();
            for w in writers {
                let orig_set = w.lock().unwrap();
                for e in orig_set.keys() {
                    debug_assert!(in_vec.contains(e), "{} {}", e.left(), e.right());
                }
            }
        }
    }
    let unlocked_stats = shared_stats.lock().unwrap();
    (edges, frequencies, unlocked_stats.clone())
}
pub ( crate ) fn collect_edges ( train : & SvmScanner , featurizer : & Featurizer , ) -> ( Vec < Edge > , Vec < usize > , GraphStats ) { let nthreads = train . nthreads ( ) ; let ref writers : Vec < Mutex < EdgeMap > > = repeat_with ( | | { Mutex :: new ( HashMap :: with_hasher ( BuildHasherDefault :: < ThreadUnsafeHasher > :: default ( ) , ) ) } ) . take ( nthreads ) . collect ( ) ; let ref shared_stats = Mutex :: new ( GraphStats :: default ( ) ) ; { train . for_each ( | | EdgeCollector :: new ( featurizer , writers , shared_stats , nthreads ) , EdgeCollector :: consume_line , EdgeCollector :: finish , ) ; } let lens : Vec < usize > = writers . iter ( ) . map ( | mutex | { let writer = mutex . lock ( ) . unwrap ( ) ; writer . len ( ) } ) . collect ( ) ; let mut edges = vec ! [ Edge default ( ) lens iter ( ) sum usize ( ) ] ; let mut frequencies = vec ! [ 0usize edges len ( ) ] ; { let ptr = edges . as_mut_ptr ( ) ; let fptr = frequencies . as_mut_ptr ( ) ; let edge_slices : Vec < _ > = lens . into_iter ( ) . scan ( 0 , | state , size | { let start = * state ; let stop = start + size ; * state = stop ; // SAFETY: each slice refers to a disjoint region in the vec Some ( ( unsafe { slice :: from_raw_parts_mut ( ptr . add ( start ) , size ) } , unsafe { slice :: from_raw_parts_mut ( fptr . add ( start ) , size ) } , ) ) } ) . collect ( ) ; edge_slices . into_par_iter ( ) // note this zipped iter is of shorter length (intentionally) . zip ( writers . into_par_iter ( ) ) . for_each ( | ( ( edge_slice , freq_slice ) , writer ) | { let writer = writer . lock ( ) . unwrap ( ) ; for ( i , ( e , s ) ) in writer . iter ( ) . enumerate ( ) { edge_slice [ i ] = e . clone ( ) ; freq_slice [ i ] = s . count ; } } ) ; if cfg ! ( debug_assertions ) { let in_vec : HashSet < Edge > = edges . iter ( ) . cloned ( ) . collect ( ) ; for w in writers { let orig_set = w . lock ( ) . unwrap ( ) ; for e in orig_set . keys ( ) { debug_assert ! ( in_vec contains ( e ) " " e left ( ) e right ( ) ) ; } } } } let unlocked_stats = shared_stats . lock ( ) . unwrap ( ) ; ( edges , frequencies , unlocked_stats . clone ( ) ) }
Return edge set, frequency of edges, and stats on features active across row (recall
features are feature hashes).
Return edge set frequency of edges and stats on features active across row ( recall features are feature hashes ) .
train
&SvmScanner
featurizer
&Featurizer
train
&SvmScanner
featurizer
&Featurizer