RPCS3/rpcs3
1
0
Fork 0
mirror of https://github.com/RPCS3/rpcs3.git synced 2026-10-11 22:49:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

sys_net: more fixes

Browse Source 
Add parameter checks to sys_net_infoctl
Fix P2P getpeername() fatal
Fix possible deadlock in tcp_timeout_monitor
Remove vport assert from poll
Add upgrade path for new P2PS state in savestates
Fix P2PS close_stream() not waking threads
Fix possible deadlock in P2PS connect()
Ensure RST is sent on unhandled packets
This commit is contained in:
RipleyTom
2026-08-10 20:03:56 +03:00
committed by Elad
parent 6df35891ad
commit 7b6a8cc2d6
5 changed files with 146 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rpcs3/Emu/Cell/lv2/sys_net.cpp
+15
View File
@@ -1911,6 +1911,11 @@ error_code sys_net_infoctl(ppu_thread& ppu, s32 cmd, vm::ptr<void> arg)
{
case 6:
{
if (!arg)
{
return -SYS_NET_EINVAL;
}
vm::ptr<net_infoctl_cmd_6_t> cmd_arg = vm::static_ptr_cast<net_infoctl_cmd_6_t>(arg);
const vm::bptr<sys_net_sockinfo_t> sock_info = cmd_arg->sock_info;
@@ -1928,6 +1933,11 @@ error_code sys_net_infoctl(ppu_thread& ppu, s32 cmd, vm::ptr<void> arg)
return not_an_error(static_cast<s32>(idm::select<lv2_socket>([](u32, lv2_socket&) {})));
}
if (max_infos < 0)
{
return -SYS_NET_EINVAL;
}
s32 num_infos = 0;
idm::select<lv2_socket>([&](u32 id, lv2_socket& sock)
@@ -1963,6 +1973,11 @@ error_code sys_net_infoctl(ppu_thread& ppu, s32 cmd, vm::ptr<void> arg)
}
case 9:
{
if (!arg)
{
return -SYS_NET_EINVAL;
}
auto& nph = g_fxo->get<named_thread<np::np_handler>>();
std::string nameserver = "nameserver " + np::ip_to_string(nph.get_dns_ip());
rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2p.cpp
+1 -3
View File
@@ -92,8 +92,7 @@ s32 lv2_socket_p2p::connect_followup()
std::pair<s32, sys_net_sockaddr> lv2_socket_p2p::getpeername()
{
sys_net.fatal("[P2P] getpeername() called on a P2P socket");
return {};
return {-SYS_NET_ENOTCONN, {}};
}
s32 lv2_socket_p2p::listen([[maybe_unused]] s32 backlog)
@@ -388,7 +387,6 @@ bs_t<lv2_socket::poll_t> lv2_socket_p2p::get_pending_events() const
void lv2_socket_p2p::poll(sys_net_pollfd& sn_pfd, [[maybe_unused]] pollfd& native_pfd)
{
std::lock_guard lock(mutex);
ensure(vport);
const bs_t<lv2_socket::poll_t> pending = get_pending_events();
rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2ps.cpp
+120 -104
View File
@@ -100,93 +100,102 @@ public:
if (thread_ctrl::state() == thread_state::aborting)
return;
std::lock_guard lock(data_mutex);
std::vector<s32> streams_to_close;
const auto now = steady_clock::now();
// Check for messages that haven't been acked
std::set<s32> rtt_increased;
for (auto it = msgs.begin(); it != msgs.end();)
{
if (it->first > now)
break;
std::lock_guard lock(data_mutex);
// reply is late, increases rtt
auto& msg = it->second;
rtt_info rtt = rtts[msg.sock_id];
// Only increases rtt once per loop(in case a big number of packets are sent at once)
if (!rtt_increased.count(msg.sock_id))
const auto now = steady_clock::now();
// Check for messages that haven't been acked
std::set<s32> rtt_increased;
for (auto it = msgs.begin(); it != msgs.end();)
{
rtt.num_retries += 1;
// Increases current rtt by 10%
rtt.rtt_time += (rtt.rtt_time / 10);
rtts[msg.sock_id] = rtt;
if (it->first > now)
break;
rtt_increased.emplace(msg.sock_id);
}
// reply is late, increases rtt
auto& msg = it->second;
rtt_info rtt = rtts[msg.sock_id];
// Only increases rtt once per loop(in case a big number of packets are sent at once)
if (!rtt_increased.count(msg.sock_id))
{
rtt.num_retries += 1;
// Increases current rtt by 10%
rtt.rtt_time += (rtt.rtt_time / 10);
rtts[msg.sock_id] = rtt;
if (rtt.num_retries >= 10)
{
// Too many retries, need to notify the socket that the connection is dead
idm::check<lv2_socket>(msg.sock_id, [&](lv2_socket& sock)
rtt_increased.emplace(msg.sock_id);
}
if (rtt.num_retries >= 10)
{
// Too many retries, need to notify the socket that the connection is dead
sys_net.error("[P2PS] Too many retries, closing the stream");
streams_to_close.push_back(msg.sock_id);
it = msgs.erase(it);
continue;
}
// resend the message
const auto res = idm::check<lv2_socket>(msg.sock_id, [&](lv2_socket& sock) -> bool
{
sys_net.error("[P2PS] Too many retries, closing the stream");
ensure(sock.get_type() == SYS_NET_SOCK_STREAM_P2P);
auto& sock_p2ps = reinterpret_cast<lv2_socket_p2ps&>(sock);
sock_p2ps.close_stream();
});
it = msgs.erase(it);
continue;
}
// resend the message
const auto res = idm::check<lv2_socket>(msg.sock_id, [&](lv2_socket& sock) -> bool
{
ensure(sock.get_type() == SYS_NET_SOCK_STREAM_P2P);
auto& sock_p2ps = reinterpret_cast<lv2_socket_p2ps&>(sock);
while (np::sendto_possibly_ipv6(sock_p2ps.get_socket(), reinterpret_cast<const char*>(msg.data.data()), ::size32(msg.data), &msg.dst_addr, 0) == -1)
{
const sys_net_error err = get_last_error(false);
// concurrency on the socket(from a sendto for example) can result in EAGAIN error in which case we try again
if (err == SYS_NET_EAGAIN)
while (np::sendto_possibly_ipv6(sock_p2ps.get_socket(), reinterpret_cast<const char*>(msg.data.data()), ::size32(msg.data), &msg.dst_addr, 0) == -1)
{
continue;
const sys_net_error err = get_last_error(false);
// concurrency on the socket(from a sendto for example) can result in EAGAIN error in which case we try again
if (err == SYS_NET_EAGAIN)
{
continue;
}
sys_net.error("[P2PS] Resending the packet failed(%s), closing the stream", err);
streams_to_close.push_back(msg.sock_id);
return false;
}
return true;
});
sys_net.error("[P2PS] Resending the packet failed(%s), closing the stream", err);
sock_p2ps.close_stream();
return false;
}
return true;
});
if (!res || !res.ret)
{
it = msgs.erase(it);
continue;
}
if (!res || !res.ret)
{
// Update key timeout
msgs.insert(std::make_pair(now + rtt.rtt_time, std::move(msg)));
it = msgs.erase(it);
continue;
}
// Update key timeout
msgs.insert(std::make_pair(now + rtt.rtt_time, std::move(msg)));
it = msgs.erase(it);
}
if (!msgs.empty())
{
const auto current_timepoint = steady_clock::now();
const auto expected_timepoint = msgs.begin()->first;
if (current_timepoint > expected_timepoint)
if (!msgs.empty())
{
wakey = 1;
const auto current_timepoint = steady_clock::now();
const auto expected_timepoint = msgs.begin()->first;
if (current_timepoint > expected_timepoint)
{
wakey = 1;
}
else
{
timeout = static_cast<atomic_wait_timeout>(std::chrono::duration_cast<std::chrono::nanoseconds>(expected_timepoint - current_timepoint).count());
}
}
else
{
timeout = static_cast<atomic_wait_timeout>(std::chrono::duration_cast<std::chrono::nanoseconds>(expected_timepoint - current_timepoint).count());
timeout = atomic_wait_timeout::inf;
}
}
else
for (const s32 sock_id : streams_to_close)
{
timeout = atomic_wait_timeout::inf;
idm::check<lv2_socket>(sock_id, [](lv2_socket& sock)
{
ensure(sock.get_type() == SYS_NET_SOCK_STREAM_P2P);
auto& sock_p2ps = reinterpret_cast<lv2_socket_p2ps&>(sock);
sock_p2ps.close_stream();
});
}
}
}
@@ -292,6 +301,11 @@ lv2_socket_p2ps::lv2_socket_p2ps(utils::serial& ar, lv2_socket_type type)
: lv2_socket_p2p(ar, type)
{
ar(status, max_backlog, backlog, op_port, op_vport, op_addr, data_beg_seq, received_data, cur_seq);
if (GET_SERIALIZATION_VERSION(lv2_net) < 3 && status == p2ps_stream_status::stream_closed && op_addr)
{
status = p2ps_stream_status::stream_disconnected;
}
}
void lv2_socket_p2ps::save(utils::serial& ar)
@@ -443,7 +457,7 @@ bool lv2_socket_p2ps::handle_listening(p2ps_encapsulated_tcp* tcp_header, [[mayb
if (status != p2ps_stream_status::stream_listening)
{
sys_net.error("[P2PS] lv2_socket_p2ps::handle_listening() called on a non listening socket(%d)!", static_cast<u8>(status));
sys_net.warning("[P2PS] lv2_socket_p2ps::handle_listening() called on a non listening socket(%d)!", static_cast<u8>(status));
return false;
}
@@ -562,11 +576,19 @@ void lv2_socket_p2ps::close_stream()
{
auto& nc = g_fxo->get<p2p_context>();
std::lock_guard lock(nc.list_p2p_ports_mutex);
auto& p2p_port = ::at32(nc.list_p2p_ports, port);
std::lock_guard threads_lock(nc.mutex_thread_loop);
std::scoped_lock more_lock(p2p_port.bound_p2p_vports_mutex, mutex);
close_stream_nl(&p2p_port);
{
std::lock_guard lock(nc.list_p2p_ports_mutex);
auto& p2p_port = ::at32(nc.list_p2p_ports, port);
std::scoped_lock more_lock(p2p_port.bound_p2p_vports_mutex, mutex);
close_stream_nl(&p2p_port);
}
// close_stream() is only called from tcp_timeout_monitor which is outside of p2p thread scope so
// we need to wake the threads here.
nc.wake_threads();
}
p2ps_stream_status lv2_socket_p2ps::get_status() const
@@ -739,14 +761,6 @@ std::pair<s32, sys_net_sockaddr> lv2_socket_p2ps::getsockname()
std::optional<s32> lv2_socket_p2ps::connect(const sys_net_sockaddr& addr)
{
std::lock_guard lock(mutex);
if (status != p2ps_stream_status::stream_closed && status != p2ps_stream_status::stream_disconnected)
{
sys_net.error("[P2PS] Called connect on a socket that is not closed!");
return -SYS_NET_EALREADY;
}
p2ps_encapsulated_tcp send_hdr;
const auto psa_in_p2p = reinterpret_cast<const sys_net_sockaddr_in_p2p*>(&addr);
auto name = sys_net_addr_to_native_addr(addr);
@@ -767,38 +781,40 @@ std::optional<s32> lv2_socket_p2ps::connect(const sys_net_sockaddr& addr)
dst_port = SCE_NP_PORT;
}
socket_type real_socket{};
auto& nc = g_fxo->get<p2p_context>();
std::lock_guard list_lock(nc.list_p2p_ports_mutex);
nc.create_p2p_port(port);
auto& pport = ::at32(nc.list_p2p_ports, port);
std::lock_guard vport_lock(pport.bound_p2p_vports_mutex);
std::lock_guard lock(mutex);
if (status != p2ps_stream_status::stream_closed && status != p2ps_stream_status::stream_disconnected)
{
std::lock_guard list_lock(nc.list_p2p_ports_mutex);
nc.create_p2p_port(port);
auto& pport = ::at32(nc.list_p2p_ports, port);
real_socket = pport.p2p_socket;
{
std::lock_guard lock(pport.bound_p2p_vports_mutex);
if (vport == 0)
{
// Unassigned vport, assigns one
sys_net.warning("[P2PS] vport was unassigned before connect!");
vport = pport.get_port();
while (pport.bound_p2ps_vports.count(vport))
{
vport = pport.get_port();
}
std::set<s32> bound_ports{lv2_id};
pport.bound_p2ps_vports.insert(std::make_pair(vport, std::move(bound_ports)));
}
const u64 key = name.sin_addr.s_addr | (static_cast<u64>(vport) << 32) | (static_cast<u64>(dst_vport) << 48);
pport.bound_p2p_streams.emplace(key, lv2_id);
}
sys_net.error("[P2PS] Called connect on a socket that is not closed!");
return -SYS_NET_EALREADY;
}
native_socket = real_socket;
if (vport == 0)
{
// Unassigned vport, assigns one
sys_net.warning("[P2PS] vport was unassigned before connect!");
vport = pport.get_port();
while (pport.bound_p2ps_vports.count(vport))
{
vport = pport.get_port();
}
std::set<s32> bound_ports{lv2_id};
pport.bound_p2ps_vports.insert(std::make_pair(vport, std::move(bound_ports)));
}
const u64 key = name.sin_addr.s_addr | (static_cast<u64>(vport) << 32) | (static_cast<u64>(dst_vport) << 48);
pport.bound_p2p_streams.emplace(key, lv2_id);
native_socket = pport.p2p_socket;
send_hdr.src_port = vport;
send_hdr.dst_port = dst_vport;
rpcs3/Emu/Cell/lv2/sys_net/nt_p2p_port.cpp
+9 -2
View File
@@ -337,13 +337,20 @@ bool nt_p2p_port::recv_data()
if (bound_p2ps_vports.contains(tcp_header->dst_port))
{
const auto& bound_sockets = ::at32(bound_p2ps_vports, tcp_header->dst_port);
bool handled = false;
for (const auto sock_id : bound_sockets)
{
sys_net.trace("Received packet for listening STREAM-P2P socket(s=%d)", sock_id);
handle_listening(sock_id, tcp_header, p2p_data.data() + sizeof(p2ps_encapsulated_tcp), &native_addr);
handled |= handle_listening(sock_id, tcp_header, p2p_data.data() + sizeof(p2ps_encapsulated_tcp), &native_addr);
}
return true;
if (handled)
{
return true;
}
// The vport is only reserved(e.g. by a connect()ed socket), no socket is listening on it, reply with RST as if it was unbound
}
if (tcp_header->flags == p2ps_tcp_flags::RST)
rpcs3/Emu/savestate_utils.cpp
+1 -1
View File
@@ -45,7 +45,7 @@ SERIALIZATION_VER(ppu, 1, 1, 2/*PPU sleep
SERIALIZATION_VER(spu, 2, 1)
SERIALIZATION_VER(lv2_sync, 3, 1)
SERIALIZATION_VER(lv2_vm, 4, 1)
SERIALIZATION_VER(lv2_net, 5, 1, 2/*TCP Feign conection loss*/)
SERIALIZATION_VER(lv2_net, 5, 1, 2/*TCP Feign conection loss*/, 3/*P2PS stream_disconnected status*/)
SERIALIZATION_VER(lv2_fs, 6, 1, 2/*NPDRM key saving*/)
SERIALIZATION_VER(lv2_prx_overlay, 7, 1)
SERIALIZATION_VER(lv2_memory, 8, 1)