diff --git a/rpcs3/Emu/Cell/lv2/sys_net.cpp b/rpcs3/Emu/Cell/lv2/sys_net.cpp index 709d888aeb..188aed54ca 100644 --- a/rpcs3/Emu/Cell/lv2/sys_net.cpp +++ b/rpcs3/Emu/Cell/lv2/sys_net.cpp @@ -1911,6 +1911,11 @@ error_code sys_net_infoctl(ppu_thread& ppu, s32 cmd, vm::ptr arg) { case 6: { + if (!arg) + { + return -SYS_NET_EINVAL; + } + vm::ptr cmd_arg = vm::static_ptr_cast(arg); const vm::bptr sock_info = cmd_arg->sock_info; @@ -1928,6 +1933,11 @@ error_code sys_net_infoctl(ppu_thread& ppu, s32 cmd, vm::ptr arg) return not_an_error(static_cast(idm::select([](u32, lv2_socket&) {}))); } + if (max_infos < 0) + { + return -SYS_NET_EINVAL; + } + s32 num_infos = 0; idm::select([&](u32 id, lv2_socket& sock) @@ -1963,6 +1973,11 @@ error_code sys_net_infoctl(ppu_thread& ppu, s32 cmd, vm::ptr arg) } case 9: { + if (!arg) + { + return -SYS_NET_EINVAL; + } + auto& nph = g_fxo->get>(); std::string nameserver = "nameserver " + np::ip_to_string(nph.get_dns_ip()); diff --git a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2p.cpp b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2p.cpp index 5f0addd656..d6f98d218a 100644 --- a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2p.cpp +++ b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2p.cpp @@ -92,8 +92,7 @@ s32 lv2_socket_p2p::connect_followup() std::pair 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_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 pending = get_pending_events(); diff --git a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2ps.cpp b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2ps.cpp index ce1dd56579..512f6dd62c 100644 --- a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2ps.cpp +++ b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2ps.cpp @@ -100,93 +100,102 @@ public: if (thread_ctrl::state() == thread_state::aborting) return; - std::lock_guard lock(data_mutex); + std::vector streams_to_close; - const auto now = steady_clock::now(); - // Check for messages that haven't been acked - std::set 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 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(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(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(sock); - sock_p2ps.close_stream(); - }); - it = msgs.erase(it); - continue; - } - // resend the message - const auto res = idm::check(msg.sock_id, [&](lv2_socket& sock) -> bool - { - ensure(sock.get_type() == SYS_NET_SOCK_STREAM_P2P); - auto& sock_p2ps = reinterpret_cast(sock); - - while (np::sendto_possibly_ipv6(sock_p2ps.get_socket(), reinterpret_cast(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(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(std::chrono::duration_cast(expected_timepoint - current_timepoint).count()); + } } else { - timeout = static_cast(std::chrono::duration_cast(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(sock_id, [](lv2_socket& sock) + { + ensure(sock.get_type() == SYS_NET_SOCK_STREAM_P2P); + auto& sock_p2ps = reinterpret_cast(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(status)); + sys_net.warning("[P2PS] lv2_socket_p2ps::handle_listening() called on a non listening socket(%d)!", static_cast(status)); return false; } @@ -562,11 +576,19 @@ void lv2_socket_p2ps::close_stream() { auto& nc = g_fxo->get(); - 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 lv2_socket_p2ps::getsockname() std::optional 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(&addr); auto name = sys_net_addr_to_native_addr(addr); @@ -767,38 +781,40 @@ std::optional lv2_socket_p2ps::connect(const sys_net_sockaddr& addr) dst_port = SCE_NP_PORT; } - socket_type real_socket{}; - auto& nc = g_fxo->get(); + 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 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(vport) << 32) | (static_cast(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 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(vport) << 32) | (static_cast(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; diff --git a/rpcs3/Emu/Cell/lv2/sys_net/nt_p2p_port.cpp b/rpcs3/Emu/Cell/lv2/sys_net/nt_p2p_port.cpp index 67be0df63b..b410e7036f 100644 --- a/rpcs3/Emu/Cell/lv2/sys_net/nt_p2p_port.cpp +++ b/rpcs3/Emu/Cell/lv2/sys_net/nt_p2p_port.cpp @@ -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) diff --git a/rpcs3/Emu/savestate_utils.cpp b/rpcs3/Emu/savestate_utils.cpp index a1d8680c31..f26e806045 100644 --- a/rpcs3/Emu/savestate_utils.cpp +++ b/rpcs3/Emu/savestate_utils.cpp @@ -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)