feat(distributed): learn our own upload speed by watching ourselves
The child limit is derived from the upload speed the server records for us, and the server pushes a user's stats to whoever watches that user — so a client that does not watch itself never learns its own figure and would carry the default cap forever. Nicotine+ reads the same number off the stats the server sends about the logged-in user. We now watch ourselves as each session starts, and the stats that come back — the initial reply and every later push — set the figure the limit is derived from. Client::own_average_speed exposes it.
This commit is contained in:
@@ -821,6 +821,12 @@ impl ClientContext {
|
||||
self.own_interests.clone()
|
||||
}
|
||||
|
||||
/// The upload speed the server records for us, if it has said.
|
||||
#[must_use]
|
||||
pub const fn own_average_speed(&self) -> Option<u32> {
|
||||
self.own_average_speed
|
||||
}
|
||||
|
||||
/// Record what the server says our own upload speed is, and re-derive the
|
||||
/// child limit from it.
|
||||
pub fn set_own_average_speed(&mut self, speed: u32) {
|
||||
|
||||
@@ -443,17 +443,35 @@ impl Client {
|
||||
average_speed,
|
||||
shared_files,
|
||||
shared_folders,
|
||||
} => match client_context.write_safe() {
|
||||
Ok(mut ctx) => ctx.apply_user_stats(
|
||||
} => {
|
||||
// Stats about ourselves carry the speed the server
|
||||
// records for us — pushed whenever it changes, since
|
||||
// we watch ourselves — and the distributed child
|
||||
// limit is derived from it.
|
||||
let own = username == own_username;
|
||||
match client_context.write_safe() {
|
||||
Ok(mut ctx) => {
|
||||
ctx.apply_user_stats(
|
||||
username,
|
||||
average_speed,
|
||||
shared_files,
|
||||
shared_folders,
|
||||
),
|
||||
);
|
||||
if own {
|
||||
ctx.set_own_average_speed(average_speed);
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
error!("[client] UserStatsReceived write: {}", e);
|
||||
error!(
|
||||
"[client] UserStatsReceived write: {}",
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
if own {
|
||||
Self::announce_child_capacity(&client_context);
|
||||
}
|
||||
}
|
||||
},
|
||||
ClientOperation::WatchedUserReceived {
|
||||
username,
|
||||
exists,
|
||||
@@ -461,19 +479,36 @@ impl Client {
|
||||
average_speed,
|
||||
shared_files,
|
||||
shared_folders,
|
||||
} => match client_context.write_safe() {
|
||||
Ok(mut ctx) => ctx.apply_watched_user(
|
||||
} => {
|
||||
let own = username == own_username;
|
||||
match client_context.write_safe() {
|
||||
Ok(mut ctx) => {
|
||||
ctx.apply_watched_user(
|
||||
username,
|
||||
exists,
|
||||
status,
|
||||
average_speed,
|
||||
shared_files,
|
||||
shared_folders,
|
||||
),
|
||||
);
|
||||
// Watching ourselves is how our own recorded
|
||||
// speed reaches us, and the child limit is
|
||||
// derived from it.
|
||||
if own && let Some(speed) = average_speed {
|
||||
ctx.set_own_average_speed(speed);
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
error!("[client] WatchedUserReceived write: {}", e);
|
||||
error!(
|
||||
"[client] WatchedUserReceived write: {}",
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
if own {
|
||||
Self::announce_child_capacity(&client_context);
|
||||
}
|
||||
}
|
||||
},
|
||||
ClientOperation::RoomEvent(event) => {
|
||||
match client_context.write_safe() {
|
||||
Ok(mut ctx) => ctx.apply_room_event(event),
|
||||
@@ -930,6 +965,15 @@ impl Client {
|
||||
Err(_) => continue,
|
||||
};
|
||||
let Some(sender) = sender else { continue };
|
||||
// Watch ourselves: the server pushes a user's stats
|
||||
// to whoever watches them, and our own recorded
|
||||
// upload speed is what the distributed child limit is
|
||||
// derived from. Nicotine+ reads the same figure from
|
||||
// the stats the server sends about the logged-in
|
||||
// user.
|
||||
let _ = sender.send(ServerMessage::SendMessage(
|
||||
MessageFactory::build_watch_user(&own_username),
|
||||
));
|
||||
for item in &interests.likes {
|
||||
let _ = sender.send(ServerMessage::SendMessage(
|
||||
MessageFactory::build_add_thing_i_like(item),
|
||||
|
||||
@@ -327,6 +327,17 @@ impl Client {
|
||||
self.send_server_message(MessageFactory::build_global_recommendations())
|
||||
}
|
||||
|
||||
/// The upload speed the server records for us, once it has reported it.
|
||||
/// It is what the distributed child limit is derived from, and it only
|
||||
/// becomes known after an upload has finished and been reported.
|
||||
#[must_use]
|
||||
pub fn own_average_speed(&self) -> Option<u32> {
|
||||
self.context
|
||||
.read_safe()
|
||||
.ok()
|
||||
.and_then(|ctx| ctx.own_average_speed())
|
||||
}
|
||||
|
||||
/// What we ourselves like and hate, as set through this client. The
|
||||
/// server forgets these when the session ends, so they are re-sent after
|
||||
/// each login.
|
||||
|
||||
@@ -5113,3 +5113,66 @@ fn our_interests_are_sent_again_after_a_new_login() {
|
||||
"a new session should carry the interests too, got {likes:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_finished_upload_teaches_us_our_own_recorded_speed() {
|
||||
// The server records a client's upload speed and pushes it to whoever
|
||||
// watches that user. We watch ourselves at login, which is how our own
|
||||
// figure — the one the distributed child limit is derived from — reaches
|
||||
// us without anyone asking on our behalf.
|
||||
let server = server_or_skip!();
|
||||
|
||||
let share_dir = unique_download_dir();
|
||||
let content: Vec<u8> = (0..8192u32).map(|i| (i % 251) as u8).collect();
|
||||
let filename = "e2e_ownspeed_probe.bin";
|
||||
std::fs::write(share_dir.join(filename), &content).unwrap();
|
||||
|
||||
let (sharer, leecher) = sharer_and_searcher(
|
||||
&server,
|
||||
&share_dir,
|
||||
"e2e_ownspeed_sharer",
|
||||
"e2e_ownspeed_leecher",
|
||||
);
|
||||
assert_eq!(
|
||||
sharer.own_average_speed(),
|
||||
Some(0),
|
||||
"a client that has never uploaded is recorded at zero"
|
||||
);
|
||||
|
||||
let query = "ownspeed";
|
||||
let _ = leecher.search(query, Duration::from_secs(3));
|
||||
let (path, size) = reply_from(&leecher, query, "e2e_ownspeed_sharer")
|
||||
.and_then(|reply| reply.files.into_iter().next())
|
||||
.map(|file| (file.name, file.size))
|
||||
.expect("the leecher should find the file");
|
||||
|
||||
let download_dir = unique_download_dir();
|
||||
let (_download, status_rx) = leecher
|
||||
.download(
|
||||
path,
|
||||
"e2e_ownspeed_sharer".to_string(),
|
||||
size,
|
||||
download_dir.display().to_string(),
|
||||
)
|
||||
.expect("start download");
|
||||
assert!(
|
||||
wait_for_completion(&leecher, &status_rx, Duration::from_secs(25)),
|
||||
"the download should complete"
|
||||
);
|
||||
|
||||
let deadline = Instant::now() + Duration::from_secs(15);
|
||||
while Instant::now() < deadline
|
||||
&& sharer.own_average_speed().unwrap_or(0) == 0
|
||||
{
|
||||
std::thread::sleep(Duration::from_millis(250));
|
||||
}
|
||||
assert!(
|
||||
sharer.own_average_speed().unwrap_or(0) > 0,
|
||||
"the speed the server recorded for us should reach us unasked, got {:?} (info: {:?})",
|
||||
sharer.own_average_speed(),
|
||||
sharer.user_info("e2e_ownspeed_sharer"),
|
||||
);
|
||||
|
||||
let _ = std::fs::remove_dir_all(share_dir);
|
||||
let _ = std::fs::remove_dir_all(download_dir);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user