ember/mediabias-plotter - Change OK4C3RQCIRKVBU4CQ2OOVBT4EUMSSTU4BIZQA6TUVFS7AGKEZI2AC

attempt at brotli phase...

Created by ember on January 10, 2021

OK4C3RQCIRKVBU4CQ2OOVBT4EUMSSTU4BIZQA6TUVFS7AGKEZI2AC

Dependencies

In channels

main

Change contents

Replacement in src/main.rs at line 4 [4.10]

B:BD[3.70] → [2.0:15]

    sync::Arc,

[3.70]

    future::Future,
    sync::{atomic::Ordering, Arc},

Insertion in src/main.rs at line 11 [4.10]
[5.20]
[2.16]
```
use futures::prelude::*;
```

Replacement in src/main.rs at line 17 [4.10]

B:BD[3.188] → [2.42:90]

use tokio::{sync::broadcast, task::JoinHandle};

[3.188]

[2.90]

use tokio::{select, sync::broadcast, task::JoinHandle};

Insertion in src/main.rs at line 114 [4.10]

[5.1794]

[3.643]


    let all_tasks = Arc::new(std::sync::atomic::AtomicIsize::new(pending.len() as isize));

Replacement in src/main.rs at line 125 [4.10]

B:BD[3.827] → [2.1227:1517]

    while let Some(next_url) = pending_rx.recv().await {
        let cl = cl.clone();
        let confirmed = confirmed.clone();
        let pending_tx = pending_tx.clone();
        let _: JoinHandle<anyhow::Result<()>> = tokio::task::spawn(async move {
            required_sleep().await;

[3.827]

[3.916]

    let done = Arc::new(tokio::sync::Notify::new());
    loop {
        select! {
            _ = done.notified() => {
                break;
            }
            Some(next_url) = pending_rx.recv() => {
                all_tasks.fetch_sub(1, Ordering::SeqCst);
                if confirmed.read().contains_key(&next_url) {continue; } // already saw it, dupe in the crawling pipeline
                let cl = cl.clone();
                let confirmed = confirmed.clone();
                let pending_tx = pending_tx.clone();
                let all_tasks = all_tasks.clone();
                let done = done.clone();
                let _: JoinHandle<anyhow::Result<_>> = tokio::task::spawn(async move {
                    required_sleep().await;
                    let pg = cl.get(next_url.clone()).send().await?;
                    let final_url = pg.url().clone();
                    let data = pg.bytes().await?;

Replacement in src/main.rs at line 147 [4.10]

B:BD[3.917] → [2.1518:1625]

            let pg = cl.get(next_url.clone()).send().await?;
            let final_url = pg.url().clone();

[3.917]

[5.1886]

                    let mut pending = HashSet::new();
                    if let Ok(utf8) = std::str::from_utf8(&data) {
                        if let Err(e) = find_new_links(utf8, &confirmed.read(), &mut pending) {
                            warn!("error finding links on {}: {}", final_url, e);
                        }
                    }

Replacement in src/main.rs at line 154 [4.10]

B:BD[5.1887] → [2.1626:1668]

            let data = pg.bytes().await?;

[5.1887]

[5.1888]

                    confirmed
                        .write()
                        .insert(next_url.clone(), PageCandidate { final_url, data });

Replacement in src/main.rs at line 158 [4.10]

B:BD[5.1889] → [2.1669:1954]

            let mut pending = HashSet::new();
            if let Ok(utf8) = std::str::from_utf8(&data) {
                if let Err(e) = find_new_links(utf8, &confirmed.read(), &mut pending) {
                    warn!("error finding links on {}: {}", final_url, e);
                }

[5.1889]

[2.1954]

                    let num_seen_here = pending.len() as isize;
                    let prev_remaining = all_tasks.fetch_add(num_seen_here, Ordering::SeqCst);
                    pending
                        .drain()
                        .for_each(|pnd| drop(pending_tx.send(pnd.clone()).map_err(|e| println!("couldn't send ): {} ", e))));
                    let num_confirmed = confirmed.read().len();
                        println!(
                            "finished {}/{}: {}",
                            num_confirmed,
                            num_confirmed as isize + all_tasks.load(Ordering::SeqCst),
                            next_url
                        );
                    if num_seen_here == 0 && prev_remaining == 0 {
                        done.notify_one();
                        println!("finished crawling site!")
                    }
                    Ok(())
                });
            },
            _ = stop.recv() => {
                break;

Deletion in src/main.rs at line 183 [4.10]

B:BD[2.1968] → [2.1968:2455]

∅:D[6.98] → [7.1393:1449]

∅:D[2.2455] → [7.1393:1449]

∅:D[5.1933] → [7.1393:1449]

B:BD[7.1393] → [7.1393:1449]

            pending
                .drain()
                .for_each(|pnd| pending_tx.send(pnd.clone()).expect("couldn't send ):"));
            confirmed
                .write()
                .insert(next_url.clone(), PageCandidate { final_url, data });
            println!(
                "finished {}/{}: {}",
                confirmed.read().len(),
                confirmed.read().len() + pending.len(),
                next_url
            );
            Ok(())
        });
        if stop.try_recv().is_ok() {
            break;

Deletion in src/main.rs at line 185 [4.10]
B:BD[3.1717] → [3.1717:1718]

Insertion in src/main.rs at line 196 [4.10]

[2.2732]

}
async fn phase<T, U>(fname: &str, func: impl FnOnce(T) -> anyhow::Result<U>) -> anyhow::Result<T>
where
    T: for<'de> Deserialize<'de> + Serialize + Default,
    U: Future<Output = anyhow::Result<T>>,
{
    std::fs::copy(fname, format!("{}.bak", fname))?;
    let f = std::fs::File::open(fname).map_err(|e| anyhow!(e));
    let rehydrated = f
        .and_then(|f| {
            bincode::deserialize_from(brotli::CompressorReader::new(
                std::io::BufReader::new(f),
                4096 * 64,
                8,
                22,
            ))
            .map_err(|e| anyhow!(e))
        })
        .unwrap_or(T::default());
    let res = func(rehydrated)?.await?;
    bincode::serialize_into(
        brotli::CompressorWriter::new(
            std::io::BufWriter::new(
                std::fs::File::create(fname).expect("couldn't create phase output file ):"),
            ),
            4096 * 64,
            8,
            22,
        ),
        &res,
    )?;
    Ok(res)

Replacement in src/main.rs at line 266 [4.10]

B:BD[7.2123] → [7.2123:2164]

B:BD[7.2164] → [2.2736:3188]

    let mut cl = reqwest::Client::new();
    fn phase<T, U>(fname: &str, func: impl FnOnce(&mut T) -> anyhow::Result<U>) -> anyhow::Result<U>
    where
        T: for<'de> Deserialize<'de> + Serialize + Default,
    {
        let f = std::fs::File::open(fname).map_err(|e| anyhow!(e));
        let mut rehydrated = f
            .and_then(|f| {
                bincode::deserialize_from(std::io::BufReader::new(f)).map_err(|e| anyhow!(e))
            })
            .unwrap_or(T::default());

[7.2123]

[7.2203]

    let cl = reqwest::Client::new();

Deletion in src/main.rs at line 268 [4.10]

B:BD[7.2204] → [2.3189:3466]

        let res = func(&mut rehydrated);
        bincode::serialize_into(
            std::io::BufWriter::new(
                std::fs::File::create(fname).expect("couldn't create phase output file ):"),
            ),
            &rehydrated,
        )?;
        res
    }

Deletion in src/main.rs at line 271 [4.10]

B:BD[7.2396] → [7.2396:2443]

B:BD[7.2443] → [2.3467:3493]

∅:D[2.3493] → [5.2341:2569]

B:BD[5.2341] → [5.2341:2569]

∅:D[5.2569] → [7.2590:2626]

B:BD[7.2590] → [7.2590:2626]

B:BD[7.2626] → [2.3494:3642]

∅:D[2.3642] → [5.2652:2742]

B:BD[5.2652] → [5.2652:2742]

        }
        // load prev state from blob
        let confirmed = {
            let f = std::fs::File::open("phase1.blob").map_err(|e| anyhow!(e));
            f.and_then(|f| {
                bincode::deserialize_from(std::io::BufReader::new(f)).map_err(|e| anyhow!(e))
            })
        }
        .unwrap_or(HashMap::new());
        let confirmed = Arc::new(RwLock::new(confirmed));
        let r = all_urls_on_site_everywhere(stop, Arc::new(cl), confirmed.clone()).await;
        match r {
            Err(e) => error!("big sad: {}", e),
            Ok(_) => {}

Replacement in src/main.rs at line 272 [4.10]

∅:D[5.2752] → [7.2701:2734]

B:BD[7.2701] → [7.2701:2734]

B:BD[7.2734] → [2.3643:3899]

∅:D[2.3899] → [7.2847:2859]

B:BD[7.2847] → [7.2847:2859]

        bincode::serialize_into(
            brotli::CompressorWriter::new(
                std::io::BufWriter::new(std::fs::File::create("phase1.blob.brotli").expect("sad")),
                4096 * 64,
                8,
                22,
            ),
            &*confirmed.read(),
        )?;

[5.2752]

[7.2859]

        phase("phase1.blob.brotli", |confirmed| {
            let confirmed = Arc::new(RwLock::new(confirmed));
            let r = all_urls_on_site_everywhere(stop, Arc::new(cl), confirmed.clone());
            Ok(r.map(move |r| match r {
                Err(e) => Err(anyhow!("big sad: {}", e)),
                Ok(v) => Ok(std::mem::replace(&mut *confirmed.write(), HashMap::new())),
            }))
        })
        .await?;

Replacement in src/main.rs at line 291 [4.10]

B:BD[7.2866] → [2.4261:4668]

        let sel = Selector::parse("header.entry-header > h2 > img + img")
            .expect("fix the phase2 selector");
        let mut outlets: HashMap<Url, MediaOutlet> = {
            let f = std::fs::File::open("phase2.blob").map_err(|e| anyhow!(e));
            f.and_then(|f| {
                bincode::deserialize_from(std::io::BufReader::new(f)).map_err(|e| anyhow!(e))
            })?
        };

[7.2866]

[2.4668]

        phase(
            "phase2.blob.brotli",
            |mut outlets: HashMap<Url, MediaOutlet>| {
                let sel = Selector::parse("header.entry-header > h2 > img + img")
                    .expect("fix the phase2 selector");

Replacement in src/main.rs at line 297 [4.10]

B:BD[2.4669] → [2.4669:5061]

        for (orig_url, candidate) in all_pages.into_iter() {
            if !outlets.contains_key(&orig_url) {
                continue;
            } // a filter would borrow outlets during the iterations
            if let Some(outlet) = consider_page(&sel, &candidate)? {
                println!("found outlet!");
                outlets.insert(orig_url, outlet);
            }
        }

[2.4669]

[2.5061]

                for (orig_url, candidate) in all_pages.into_iter() {
                    if outlets.contains_key(&orig_url) {
                        continue;
                    } // a filter would borrow outlets during the iterations
                    if let Some(outlet) = consider_page(&sel, &candidate)? {
                        println!("found outlet!");
                        outlets.insert(orig_url, outlet);
                    }
                }
                Ok(std::future::ready(Ok(outlets)))
                /*
                let confirmed = Arc::new(RwLock::new(confirmed));
                let r = all_urls_on_site_everywhere(stop, Arc::new(cl), confirmed.clone());
                r.map(move |r| match r {
                    Err(e) => Err(anyhow!("big sad: {}", e)),
                    Ok(v) => Ok(std::mem::replace(&mut *confirmed.write(), HashMap::new())),
                })*/
            },
        )
        .await?;

Insertion in Cargo.toml at line 27 [8.20]
[2.5511]
```
futures = "0.3"
```

Insertion in Cargo.lock at line 337 [8.450]

[3.3836]

]
[[package]]
name = "futures"
version = "0.3.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c70be434c505aee38639abccb918163b63158a4b4bb791b45b7023044bdc3c9c"
dependencies = [
 "futures-channel",
 "futures-core",
 "futures-executor",
 "futures-io",
 "futures-sink",
 "futures-task",
 "futures-util",

Insertion in Cargo.lock at line 361 [8.450]
[3.3979]
[8.9999]
```
 "futures-sink",
```

Insertion in Cargo.lock at line 369 [8.450]

[3.4099]

[8.10203]


[[package]]
name = "futures-executor"
version = "0.3.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9ee9ca2f7eb4475772cf39dd1cd06208dce2670ad38f4d9c7262b3e15f127068"
dependencies = [
 "futures-core",
 "futures-task",
 "futures-util",
]

Insertion in Cargo.lock at line 382 [8.450]

[8.10216]

[3.4100]

name = "futures-io"
version = "0.3.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e37c1a51b037b80922864b8eed90692c5cd8abd4c71ce49b77146caa47f3253b"
[[package]]
name = "futures-macro"
version = "0.3.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0f8719ca0e1f3c5e34f3efe4570ef2c0610ca6da85ae7990d472e9cbfba13664"
dependencies = [
 "proc-macro-hack",
 "proc-macro2",
 "quote",
 "syn",
]
[[package]]

Insertion in Cargo.lock at line 410 [8.450]
[3.4339]
[8.10636]
```
dependencies = [
 "once_cell",
]
```
Insertion in Cargo.lock at line 420 [8.450]
[8.10852]
[3.4460]
```
 "futures-channel",
```
Insertion in Cargo.lock at line 422 [8.450]
[3.4477]
[3.4477]
```
 "futures-io",
 "futures-macro",
 "futures-sink",
```
Insertion in Cargo.lock at line 426 [8.450]
[3.4494]
[3.4494]
```
 "memchr",
```
Insertion in Cargo.lock at line 429 [8.450]
[3.4529]
[8.10877]
```
 "proc-macro-hack",
 "proc-macro-nested",
 "slab",
```
Insertion in Cargo.lock at line 733 [8.450]
[7.4944]
[2.6491]
```
 "futures",
```

Insertion in Cargo.lock at line 1080 [8.450]

[8.24314]

name = "proc-macro-nested"
version = "0.1.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "eba180dafb9038b050a4c280019bbedf9f2467b61e5d892dcad585bb57aadc5a"
[[package]]

attempt at brotli phase...

Dependencies

In channels

Change contents

Replacement in src/main.rs at line 4 [4.10]

Insertion in src/main.rs at line 11 [4.10]

Replacement in src/main.rs at line 17 [4.10]

Insertion in src/main.rs at line 114 [4.10]

Replacement in src/main.rs at line 125 [4.10]

Replacement in src/main.rs at line 147 [4.10]

Replacement in src/main.rs at line 154 [4.10]

Replacement in src/main.rs at line 158 [4.10]

Deletion in src/main.rs at line 183 [4.10]

Deletion in src/main.rs at line 185 [4.10]

Insertion in src/main.rs at line 196 [4.10]

Replacement in src/main.rs at line 266 [4.10]

Deletion in src/main.rs at line 268 [4.10]

Deletion in src/main.rs at line 271 [4.10]

Replacement in src/main.rs at line 272 [4.10]

Replacement in src/main.rs at line 291 [4.10]

Replacement in src/main.rs at line 297 [4.10]

Insertion in Cargo.toml at line 27 [8.20]

Insertion in Cargo.lock at line 337 [8.450]

Insertion in Cargo.lock at line 361 [8.450]

Insertion in Cargo.lock at line 369 [8.450]

Insertion in Cargo.lock at line 382 [8.450]

Insertion in Cargo.lock at line 410 [8.450]

Insertion in Cargo.lock at line 420 [8.450]

Insertion in Cargo.lock at line 422 [8.450]

Insertion in Cargo.lock at line 426 [8.450]

Insertion in Cargo.lock at line 429 [8.450]

Insertion in Cargo.lock at line 733 [8.450]

Insertion in Cargo.lock at line 1080 [8.450]