MCPcopy Create free account
hub / github.com/MaterializeInc/materialize / test_create_sockets

Function test_create_sockets

src/cluster/src/communication.rs:540–665  ·  view source on GitHub ↗

too slow

()

Source from the content-addressed store, hash-verified

538 #[test] // allow(test-attribute)
539 #[cfg_attr(miri, ignore)] // too slow
540 fn test_create_sockets() {
541 const NUM_PROCESSES: usize = 10;
542 const NUM_CRASHES: usize = 3;
543
544 configure_tracing_for_turmoil();
545
546 let seed = std::env::var("SEED")
547 .ok()
548 .and_then(|x| x.parse().ok())
549 .unwrap_or_else(rand::random);
550
551 info!("initializing rng with seed {seed}");
552 let mut rng = SmallRng::seed_from_u64(seed);
553
554 let mut sim = turmoil::Builder::new()
555 .enable_random_order()
556 .rng_seed(rng.random())
557 .build();
558
559 let processes: Vec<_> = (0..NUM_PROCESSES).map(|i| format!("process-{i}")).collect();
560 let addresses: Vec<_> = processes
561 .iter()
562 .map(|n| format!("turmoil:{n}:7777"))
563 .collect();
564
565 // Channel for processes to report successful connection.
566 let (ready_tx, mut ready_rx) = mpsc::unbounded_channel();
567
568 // A watch for informing processes about the beginning of the stable phase.
569 // This is used to delay the processes' final connectivity checks until after we know that
570 // processes won't randomly crash anymore.
571 let (stable_tx, stable_rx) = watch::channel(false);
572
573 for (index, name) in processes.iter().enumerate() {
574 let addresses = addresses.clone();
575 let ready_tx = ready_tx.clone();
576 let stable_rx = stable_rx.clone();
577 sim.host(&name[..], move || {
578 let addresses = addresses.clone();
579 let ready_tx = ready_tx.clone();
580 let mut stable_rx = stable_rx.clone();
581 async move {
582 'protocol: loop {
583 let mut sockets = match create_sockets(index, &addresses).await {
584 Ok(sockets) => sockets,
585 Err(error) if error.is_fatal() => Err(error)?,
586 Err(error) => {
587 info!("creating sockets failed: {error}; retrying protocol");
588 continue 'protocol;
589 }
590 };
591
592 // We have a connection to each peer, but some of them might be broken, in
593 // which case we should restart the `create_sockets` protocol. In the real
594 // world we would notice the broken connections eventually after writing to
595 // them enough, but in the test we want something more deterministic, so we
596 // let processes ping each other instead.
597 //

Callers 1

fuzz_create_socketsFunction · 0.85

Calls 15

channelFunction · 0.85
create_socketsFunction · 0.85
enumerateMethod · 0.80
wait_forMethod · 0.80
unwrapMethod · 0.80
bounceMethod · 0.80
timeoutFunction · 0.70
okMethod · 0.45
parseMethod · 0.45
buildMethod · 0.45
collectMethod · 0.45

Tested by

no test coverage detected