rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-01-15 01:12:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make network random unstable

Browse Source
This commit is contained in:
Arthur Petukhovsky
2023-08-29 13:55:36 +00:00
parent 2fd351fd63
commit 1effb586ba
8 changed files with 69 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
libs/walproposer/bindgen_deps.h
View File

@@ -22,6 +22,8 @@ void WalProposerRust();
void WalProposerCleanup();
extern bool debug_enabled;
// Initialize global variables before calling any Postgres C code.
void MyContextInit();

1
libs/walproposer/build.rs
View File

@@ -118,6 +118,7 @@ fn main() -> anyhow::Result<()> {
.allowlist_var("neon_tenant_walproposer")
.allowlist_var("syncSafekeepers")
.allowlist_var("sim_redo_start_lsn")
.allowlist_var("debug_enabled")
.clang_arg(format!("-I{inc_server_path}"))
.clang_arg(format!("-I{inc_pgxn_path}"))
.clang_arg(format!("-DSIMLIB"))

24
libs/walproposer/libpqwalproposer.c
View File

@@ -18,7 +18,7 @@ struct WalProposerConn
static bool
ensure_nonblocking_status(WalProposerConn *conn, bool is_nonblocking)
{
walprop_log(LOG, "not implemented");
// walprop_log(LOG, "not implemented");
return false;
}
@@ -26,14 +26,14 @@ ensure_nonblocking_status(WalProposerConn *conn, bool is_nonblocking)
char *
walprop_error_message(WalProposerConn *conn)
{
walprop_log(LOG, "not implemented");
// walprop_log(LOG, "not implemented");
return NULL;
}
WalProposerConnStatusType
walprop_status(WalProposerConn *conn)
{
walprop_log(LOG, "not implemented: walprop_status");
// walprop_log(LOG, "not implemented: walprop_status");
return WP_CONNECTION_OK;
}
@@ -57,21 +57,21 @@ walprop_connect_start(char *conninfo)
WalProposerConnectPollStatusType
walprop_connect_poll(WalProposerConn *conn)
{
walprop_log(LOG, "not implemented: walprop_connect_poll");
// walprop_log(LOG, "not implemented: walprop_connect_poll");
return WP_CONN_POLLING_OK;
}
bool
walprop_send_query(WalProposerConn *conn, char *query)
{
walprop_log(LOG, "not implemented: walprop_send_query");
// walprop_log(LOG, "not implemented: walprop_send_query");
return true;
}
WalProposerExecStatusType
walprop_get_query_result(WalProposerConn *conn)
{
walprop_log(LOG, "not implemented: walprop_get_query_result");
// walprop_log(LOG, "not implemented: walprop_get_query_result");
return WP_EXEC_SUCCESS_COPYBOTH;
}
@@ -84,14 +84,14 @@ walprop_socket(WalProposerConn *conn)
int
walprop_flush(WalProposerConn *conn)
{
walprop_log(LOG, "not implemented");
// walprop_log(LOG, "not implemented");
return 0;
}
void
walprop_finish(WalProposerConn *conn)
{
walprop_log(LOG, "walprop_finish not implemented");
// walprop_log(LOG, "walprop_finish not implemented");
}
/*
@@ -113,7 +113,7 @@ walprop_async_read(WalProposerConn *conn, char **buf, int *amount)
event = sim_epoll_rcv(0);
walprop_log(LOG, "walprop_async_read, T: %d, tcp: %d, tag: %d", (int) event.tag, (int) event.tcp, (int) event.any_message);
// walprop_log(LOG, "walprop_async_read, T: %d, tcp: %d, tag: %d", (int) event.tag, (int) event.tcp, (int) event.any_message);
Assert(event.tcp == conn->tcp);
Assert(event.tag == Message);
Assert(event.any_message == Bytes);
@@ -121,7 +121,7 @@ walprop_async_read(WalProposerConn *conn, char **buf, int *amount)
msg = (char*) sim_msg_get_bytes(&len);
*buf = msg;
*amount = len;
walprop_log(LOG, "walprop_async_read: %d", (int) len);
// walprop_log(LOG, "walprop_async_read: %d", (int) len);
return PG_ASYNC_READ_SUCCESS;
}
@@ -129,7 +129,7 @@ walprop_async_read(WalProposerConn *conn, char **buf, int *amount)
PGAsyncWriteResult
walprop_async_write(WalProposerConn *conn, void const *buf, size_t size)
{
walprop_log(LOG, "walprop_async_write");
// walprop_log(LOG, "walprop_async_write");
sim_msg_set_bytes(buf, size);
sim_tcp_send(conn->tcp);
return PG_ASYNC_WRITE_SUCCESS;
@@ -142,7 +142,7 @@ walprop_async_write(WalProposerConn *conn, void const *buf, size_t size)
bool
walprop_blocking_write(WalProposerConn *conn, void const *buf, size_t size)
{
walprop_log(LOG, "walprop_blocking_write");
// walprop_log(LOG, "walprop_blocking_write");
sim_msg_set_bytes(buf, size);
sim_tcp_send(conn->tcp);
return true;

4
libs/walproposer/src/lib.rs
View File

@@ -30,3 +30,7 @@ pub fn c_context() -> Option<Box<dyn Fn(NodeOs) + Send + Sync>> {
unsafe { bindings::MyContextInit(); }
}))
}
pub fn enable_debug() {
unsafe { bindings::debug_enabled = true; }
}

9
libs/walproposer/src/simtest/log.rs
View File

@@ -4,6 +4,8 @@ use safekeeper::simlib::{world::World, sync::Mutex};
use tracing_subscriber::fmt::{time::FormatTime, format::Writer};
use utils::logging;
use crate::bindings;
#[derive(Clone)]
pub struct SimClock {
@@ -38,12 +40,17 @@ impl FormatTime for SimClock {
}
pub fn init_logger() -> SimClock {
let debug_enabled = unsafe { bindings::debug_enabled };
let clock = SimClock::default();
let base_logger = tracing_subscriber::fmt()
.with_target(false)
.with_timer(clock.clone())
.with_ansi(true)
// .with_max_level(tracing::Level::DEBUG)
.with_max_level(match debug_enabled {
true => tracing::Level::DEBUG,
false => tracing::Level::INFO,
})
.with_writer(std::io::stdout);
base_logger.init();

28
libs/walproposer/src/simtest/util.rs
View File

@@ -452,7 +452,7 @@ pub fn generate_schedule(seed: u64) -> Schedule {
let cnt = rng.gen_range(1..100);
for _ in 0..cnt {
time += rng.gen_range(0..100);
time += rng.gen_range(0..500);
let action = match rng.gen_range(0..3) {
0 => TestAction::WriteTx(rng.gen_range(1..10)),
1 => TestAction::RestartSafekeeper(rng.gen_range(0..3)),
@@ -464,3 +464,29 @@ pub fn generate_schedule(seed: u64) -> Schedule {
schedule
}
pub fn generate_network_opts(seed: u64) -> NetworkOptions {
let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
let timeout = rng.gen_range(100..2000);
let max_delay = rng.gen_range(1..2*timeout);
let min_delay = rng.gen_range(1..=max_delay);
let max_fail_prob = rng.gen_range(0.0..0.9);
let connect_fail_prob = rng.gen_range(0.0..max_fail_prob);
let send_fail_prob = rng.gen_range(0.0..connect_fail_prob);
NetworkOptions {
keepalive_timeout: Some(timeout),
connect_delay: Delay {
min: min_delay,
max: max_delay,
fail_prob: connect_fail_prob,
},
send_delay: Delay {
min: min_delay,
max: max_delay,
fail_prob: send_fail_prob,
},
}
}

12
libs/walproposer/src/simtest/wp_sk.rs
View File

@@ -20,8 +20,8 @@ use crate::{
simtest::{
log::{init_logger, SimClock},
safekeeper::run_server,
util::{generate_schedule, TestConfig},
},
util::{generate_schedule, TestConfig, generate_network_opts},
}, enable_debug,
};
use super::{
@@ -162,8 +162,10 @@ fn test_random_schedules() -> anyhow::Result<()> {
let mut config = TestConfig::new(Some(clock));
config.network.keepalive_timeout = Some(100);
for i in 0..1000 {
for i in 0..30000 {
let seed: u64 = rand::thread_rng().gen();
config.network = generate_network_opts(seed);
let test = config.start(seed);
warn!("Running test with seed {}", seed);
@@ -178,6 +180,7 @@ fn test_random_schedules() -> anyhow::Result<()> {
#[test]
fn test_one_schedule() -> anyhow::Result<()> {
enable_debug();
let clock = init_logger();
let mut config = TestConfig::new(Some(clock));
config.network.keepalive_timeout = Some(100);
@@ -191,7 +194,8 @@ fn test_one_schedule() -> anyhow::Result<()> {
// test.run_schedule(&schedule)?;
// test.world.stop_all();
let seed = 9982078935230494862;
let seed = 11245530003696902397;
config.network = generate_network_opts(seed);
let test = config.start(seed);
warn!("Running test with seed {}", seed);

9
libs/walproposer/test.c
View File

@@ -57,6 +57,8 @@ void RunClientC(uint32_t serverId) {
}
}
bool debug_enabled = false;
bool initializedMemoryContext = false;
// pthread_mutex_init(&lock, NULL)?
pthread_mutex_t lock;
@@ -80,8 +82,11 @@ void MyContextInit() {
if (!SelectConfigFiles(NULL, progname))
exit(1);
log_min_messages = FATAL;
// log_min_messages = LOG;
if (debug_enabled) {
log_min_messages = LOG;
} else {
log_min_messages = FATAL;
}
Log_line_prefix = "[%p] ";
InitializeMaxBackends();