0_blep_backtrace.txt

(lldb) run
Process 15007 launched: '/Users/benl/src/blep/target/debug/blep' (x86_64)
starting thread ThreadId(2)
starting thread ThreadId(4)
starting thread ThreadId(3)
starting thread ThreadId(5)
200 202ms
200 205ms
200 201ms
200 207ms
200 202ms
200 210ms
200 218ms
200 223ms
200 221ms
200 218ms
200 233ms
200 235ms
200 243ms
200 239ms
200 241ms
200 240ms
200 247ms
200 234ms
200 242ms
200 244ms
200 250ms
blep(15007,0x7000103a8000) malloc: *** error for object 0x100807840: pointer being freed was not allocated
blep(15007,0x7000103a8000) malloc: *** set a breakpoint in malloc_error_break to debug
Process 15007 stopped
* thread #2, name = 'curl agent', stop reason = signal SIGABRT
    frame #0: 0x00007fff6c3662c2 libsystem_kernel.dylib`__pthread_kill + 10
libsystem_kernel.dylib`__pthread_kill:
->  0x7fff6c3662c2 <+10>: jae    0x7fff6c3662cc            ; <+20>
    0x7fff6c3662c4 <+12>: movq   %rax, %rdi
    0x7fff6c3662c7 <+15>: jmp    0x7fff6c360453            ; cerror_nocancel
    0x7fff6c3662cc <+20>: retq
Target 0: (blep) stopped.
(lldb) thread backtrace
error: need to add support for DW_TAG_base_type '()' encoded with DW_ATE = 0x7, bit_size = 0
error: need to add support for DW_TAG_base_type '()' encoded with DW_ATE = 0x7, bit_size = 0
error: need to add support for DW_TAG_base_type '()' encoded with DW_ATE = 0x7, bit_size = 0
* thread #2, name = 'curl agent', stop reason = signal SIGABRT
  * frame #0: 0x00007fff6c3662c2 libsystem_kernel.dylib`__pthread_kill + 10
    frame #1: 0x00007fff6c421bf1 libsystem_pthread.dylib`pthread_kill + 284
    frame #2: 0x00007fff6c2d06a6 libsystem_c.dylib`abort + 127
    frame #3: 0x00007fff6c3df077 libsystem_malloc.dylib`malloc_vreport + 545
    frame #4: 0x00007fff6c3dee38 libsystem_malloc.dylib`malloc_report + 151
    frame #5: 0x00007fff69b9b7c9 libcrypto.42.dylib`lh_free + 82
    frame #6: 0x00007fff69b82766 libcrypto.42.dylib`int_thread_del_item + 145
    frame #7: 0x00007fff69b81d60 libcrypto.42.dylib`ERR_remove_thread_state + 84
    frame #8: 0x00007fff69cb0ce7 libcurl.4.dylib`Curl_close + 183
    frame #9: 0x00007fff69cc3e71 libcurl.4.dylib`curl_easy_cleanup + 39
    frame #10: 0x00000001000a038b blep`_$LT$curl..easy..handler..Easy2$LT$H$GT$$u20$as$u20$core..ops..drop..Drop$GT$::drop::h3b06bce3a2e69e59(self=0x000070001
03a6680) at handler.rs:2932:12
    frame #11: 0x0000000100091ab5 blep`core::ptr::drop_in_place::h73919fbecdae0114((null)=0x00007000103a6680) at mod.rs:174
    frame #12: 0x0000000100030022 blep`isahc::agent::AgentContext::complete_request::hbc2792097c05c3d8(self=0x00007000103a74e8, token=0, result=Result<(), cur
l::error::Error> @ 0x00007000103a6818) at agent.rs:315:4
    frame #13: 0x000000010003199a blep`isahc::agent::AgentContext::dispatch::h91925077f8575c3a(self=0x00007000103a74e8) at agent.rs:433:39
    frame #14: 0x0000000100032301 blep`isahc::agent::AgentContext::run::hcc7025c1931654db(self=AgentContext @ 0x00007000103a74e8) at agent.rs:460:12
    frame #15: 0x000000010002e469 blep`isahc::agent::AgentBuilder::spawn::_$u7b$$u7b$closure$u7d$$u7d$::he74ea49d9fca5b25 at agent.rs:111:37
    frame #16: 0x000000010006cafb blep`std::sys_common::backtrace::__rust_begin_short_backtrace::hd8648a93e71531b7(f=<unavailable>) at backtrace.rs:129:4
    frame #17: 0x000000010003e2bb blep`std::thread::Builder::spawn_unchecked::_$u7b$$u7b$closure$u7d$$u7d$::_$u7b$$u7b$closure$u7d$$u7d$::hdb0d6c3c406ead41 at
 mod.rs:475:16
    frame #18: 0x000000010007199b blep`_$LT$std..panic..AssertUnwindSafe$LT$F$GT$$u20$as$u20$core..ops..function..FnOnce$LT$$LP$$RP$$GT$$GT$::call_once::h1d9e
7d2c39cbb620(self=<unavailable>, _args=<unavailable>) at panic.rs:318:8
    frame #19: 0x000000010004792c blep`std::panicking::try::do_call::h8b690700a0556827(data="") at panicking.rs:305:39
    frame #20: 0x000000010021925b blep`__rust_maybe_catch_panic at lib.rs:86:7 [opt]
    frame #21: 0x0000000100046a2a blep`std::panicking::try::h0749fe8c4c73bba4(f=<unavailable>) at panicking.rs:281:12
    frame #22: 0x0000000100075d7b blep`std::panic::catch_unwind::h67f2564bea693dca(f=<unavailable>) at panic.rs:394:13
    frame #23: 0x000000010003e0db blep`std::thread::Builder::spawn_unchecked::_$u7b$$u7b$closure$u7d$$u7d$::hb5e00aa6fcef2b38 at mod.rs:474:29
    frame #24: 0x000000010008f555 blep`core::ops::function::FnOnce::call_once$u7b$$u7b$vtable.shim$u7d$$u7d$::he526458753de5c22((null)=0x00007000103a7e30, (nu
ll)=<unavailable>) at function.rs:232:4
    frame #25: 0x000000010020b0de blep`_$LT$alloc..boxed..Box$LT$F$GT$$u20$as$u20$core..ops..function..FnOnce$LT$A$GT$$GT$::call_once::h606f2dced48ff3c4 at bo
        frame #26: 0x0000000100218bde blep`std::sys::unix::thread::Thread::new::thread_start::h7199626a1bd56873 [inlined] _$LT$alloc..boxed..Box$LT$F$GT$$u20$as$u
20$core..ops..function..FnOnce$LT$A$GT$$GT$::call_once::h0ece8cce658c93ae at boxed.rs:1015:8 [opt]
    frame #27: 0x0000000100218bd2 blep`std::sys::unix::thread::Thread::new::thread_start::h7199626a1bd56873 [inlined] std::sys_common::thread::start_thread::h
30bbd9f3fb79e88c at thread.rs:13 [opt]
    frame #28: 0x0000000100218b5e blep`std::sys::unix::thread::Thread::new::thread_start::h7199626a1bd56873 at thread.rs:80 [opt]
    frame #29: 0x00007fff6c41f2eb libsystem_pthread.dylib`_pthread_body + 126
    frame #30: 0x00007fff6c422249 libsystem_pthread.dylib`_pthread_start + 66
    frame #31: 0x00007fff6c41e40d libsystem_pthread.dylib`thread_start + 13
(lldb)
* thread #2, name = 'curl agent', stop reason = signal SIGABRT
  * frame #0: 0x00007fff6c3662c2 libsystem_kernel.dylib`__pthread_kill + 10
    frame #1: 0x00007fff6c421bf1 libsystem_pthread.dylib`pthread_kill + 284
    frame #2: 0x00007fff6c2d06a6 libsystem_c.dylib`abort + 127
    frame #3: 0x00007fff6c3df077 libsystem_malloc.dylib`malloc_vreport + 545
    frame #4: 0x00007fff6c3dee38 libsystem_malloc.dylib`malloc_report + 151
    frame #5: 0x00007fff69b9b7c9 libcrypto.42.dylib`lh_free + 82
    frame #6: 0x00007fff69b82766 libcrypto.42.dylib`int_thread_del_item + 145
    frame #7: 0x00007fff69b81d60 libcrypto.42.dylib`ERR_remove_thread_state + 84
    frame #8: 0x00007fff69cb0ce7 libcurl.4.dylib`Curl_close + 183
    frame #9: 0x00007fff69cc3e71 libcurl.4.dylib`curl_easy_cleanup + 39
    frame #10: 0x00000001000a038b blep`_$LT$curl..easy..handler..Easy2$LT$H$GT$$u20$as$u20$core..ops..drop..Drop$GT$::drop::h3b06bce3a2e69e59(self=0x000070001
03a6680) at handler.rs:2932:12
    frame #11: 0x0000000100091ab5 blep`core::ptr::drop_in_place::h73919fbecdae0114((null)=0x00007000103a6680) at mod.rs:174
    frame #12: 0x0000000100030022 blep`isahc::agent::AgentContext::complete_request::hbc2792097c05c3d8(self=0x00007000103a74e8, token=0, result=Result<(), cur
l::error::Error> @ 0x00007000103a6818) at agent.rs:315:4
    frame #13: 0x000000010003199a blep`isahc::agent::AgentContext::dispatch::h91925077f8575c3a(self=0x00007000103a74e8) at agent.rs:433:39
    frame #14: 0x0000000100032301 blep`isahc::agent::AgentContext::run::hcc7025c1931654db(self=AgentContext @ 0x00007000103a74e8) at agent.rs:460:12
    frame #15: 0x000000010002e469 blep`isahc::agent::AgentBuilder::spawn::_$u7b$$u7b$closure$u7d$$u7d$::he74ea49d9fca5b25 at agent.rs:111:37
    frame #16: 0x000000010006cafb blep`std::sys_common::backtrace::__rust_begin_short_backtrace::hd8648a93e71531b7(f=<unavailable>) at backtrace.rs:129:4
    frame #17: 0x000000010003e2bb blep`std::thread::Builder::spawn_unchecked::_$u7b$$u7b$closure$u7d$$u7d$::_$u7b$$u7b$closure$u7d$$u7d$::hdb0d6c3c406ead41 at
 mod.rs:475:16
    frame #18: 0x000000010007199b blep`_$LT$std..panic..AssertUnwindSafe$LT$F$GT$$u20$as$u20$core..ops..function..FnOnce$LT$$LP$$RP$$GT$$GT$::call_once::h1d9e
7d2c39cbb620(self=<unavailable>, _args=<unavailable>) at panic.rs:318:8
    frame #19: 0x000000010004792c blep`std::panicking::try::do_call::h8b690700a0556827(data="") at panicking.rs:305:39
    frame #20: 0x000000010021925b blep`__rust_maybe_catch_panic at lib.rs:86:7 [opt]
    frame #21: 0x0000000100046a2a blep`std::panicking::try::h0749fe8c4c73bba4(f=<unavailable>) at panicking.rs:281:12
    frame #22: 0x0000000100075d7b blep`std::panic::catch_unwind::h67f2564bea693dca(f=<unavailable>) at panic.rs:394:13
    frame #23: 0x000000010003e0db blep`std::thread::Builder::spawn_unchecked::_$u7b$$u7b$closure$u7d$$u7d$::hb5e00aa6fcef2b38 at mod.rs:474:29
    frame #24: 0x000000010008f555 blep`core::ops::function::FnOnce::call_once$u7b$$u7b$vtable.shim$u7d$$u7d$::he526458753de5c22((null)=0x00007000103a7e30, (nu
ll)=<unavailable>) at function.rs:232:4
    frame #25: 0x000000010020b0de blep`_$LT$alloc..boxed..Box$LT$F$GT$$u20$as$u20$core..ops..function..FnOnce$LT$A$GT$$GT$::call_once::h606f2dced48ff3c4 at bo
xed.rs:1015:8 [opt]
    frame #26: 0x0000000100218bde blep`std::sys::unix::thread::Thread::new::thread_start::h7199626a1bd56873 [inlined] _$LT$alloc..boxed..Box$LT$F$GT$$u20$as$u
20$core..ops..function..FnOnce$LT$A$GT$$GT$::call_once::h0ece8cce658c93ae at boxed.rs:1015:8 [opt]
    frame #27: 0x0000000100218bd2 blep`std::sys::unix::thread::Thread::new::thread_start::h7199626a1bd56873 [inlined] std::sys_common::thread::start_thread::h
30bbd9f3fb79e88c at thread.rs:13 [opt]
    frame #28: 0x0000000100218b5e blep`std::sys::unix::thread::Thread::new::thread_start::h7199626a1bd56873 at thread.rs:80 [opt]
        frame #29: 0x00007fff6c41f2eb libsystem_pthread.dylib`_pthread_body + 126
    frame #30: 0x00007fff6c422249 libsystem_pthread.dylib`_pthread_start + 66
    frame #31: 0x00007fff6c41e40d libsystem_pthread.dylib`thread_start + 13
    

blep.rs

mod interval;

use smol::Task;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant};

use futures::StreamExt;
use http_types::StatusCode;

use interval::Interval;

fn main() {
    for _ in 0..4 {
        thread::spawn(|| {
            println!("starting thread {:?}", thread::current().id());
            smol::run(futures::future::pending::<()>())
        });
    }

    let n = 30;
    let counter = Arc::new(AtomicUsize::new(n));
    let url = "http://localhost:8080/drip?duration=0.1&numbytes=16";

    let c2 = counter.clone();
    let mut ticker = Interval::every(Duration::from_millis(1)).take(n);
    smol::run(async move {

        while let Some(tick) = ticker.next().await {
            let url = url.clone();
            let counter = c2.clone();

            Task::spawn(async move {
                match benchmark(tick, url).await {
                    BenchmarkResult::HttpResponse {
                        status,
                        start_at,
                        body_at: Some(body_at),
                        ..
                    } => {
                        println!(
                            "{} {}ms",
                            status,
                            body_at.duration_since(start_at).as_millis()
                        );
                    }
                    _ => {}
                };
                counter.fetch_sub(1, Ordering::Relaxed);
            }).detach();
        }
    });

    while counter.load(Ordering::Relaxed) != 0 {}
}

enum BenchmarkResult {
    Error, // FIXME
    HttpResponse {
        start_at: Instant,
        status: StatusCode,
        #[allow(unused)]
        headers_at: Instant,
        body_at: Option<Instant>,
        #[allow(unused)]
        body_len: Option<usize>,
    },
}

async fn benchmark(start_at: Instant, url: &str) -> BenchmarkResult {
    let mut resp = match surf::get(url).await {
        Err(_) => return BenchmarkResult::Error,
        Ok(resp) => resp,
    };
    let headers_at = Instant::now();

    let body_len = match resp.body_bytes().await {
        Err(_) => return BenchmarkResult::Error,
        Ok(bs) => Some(bs.len()),
    };
    let body_at = Some(Instant::now());

    BenchmarkResult::HttpResponse { status: resp.status(), start_at, headers_at, body_at, body_len }
}

interval.rs

use smol::Timer;
use std::time::{Duration, Instant};
use futures::ready;
use futures::stream::Stream;

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};


pub struct Interval {
    next: Timer,
    interval: Duration,
}

impl Interval {
    pub fn every(interval: Duration) -> Interval {
        let next = Timer::after(interval);
        Interval { next, interval }
    }
}

impl Stream for Interval {
    type Item = Instant;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
        let t = ready!(Pin::new(&mut self.next).poll(cx));
        self.next = Timer::at(t + self.interval);

        Poll::Ready(Some(t))
    }
}