diff --git a/rpcs3/Emu/Cell/Modules/sceNp.cpp b/rpcs3/Emu/Cell/Modules/sceNp.cpp index e82491eac1..1e09b8f493 100644 --- a/rpcs3/Emu/Cell/Modules/sceNp.cpp +++ b/rpcs3/Emu/Cell/Modules/sceNp.cpp @@ -6800,6 +6800,11 @@ error_code sceNpSignalingActivateConnection(u32 ctx_id, vm::ptr npId, v return SCE_NP_SIGNALING_ERROR_NOT_INITIALIZED; } + if (!get_signaling_context(ctx_id)) + { + return SCE_NP_SIGNALING_ERROR_CTX_NOT_FOUND; + } + if (!npId || !conn_id) { return SCE_NP_SIGNALING_ERROR_INVALID_ARGUMENT; diff --git a/rpcs3/Emu/Cell/Modules/sys_net_.h b/rpcs3/Emu/Cell/Modules/sys_net_.h index 9ee344b977..e8cab78835 100644 --- a/rpcs3/Emu/Cell/Modules/sys_net_.h +++ b/rpcs3/Emu/Cell/Modules/sys_net_.h @@ -2,19 +2,6 @@ #include "Emu/Cell/lv2/sys_net.h" -struct sys_net_sockinfo_t -{ - be_t s; - be_t proto; - be_t recv_queue_length; - be_t send_queue_length; - sys_net_in_addr local_adr; - be_t local_port; - sys_net_in_addr remote_adr; - be_t remote_port; - be_t state; -}; - struct sys_net_sockinfo_ex_t { be_t s; diff --git a/rpcs3/Emu/Cell/lv2/sys_net.cpp b/rpcs3/Emu/Cell/lv2/sys_net.cpp index 31a3c6ea07..709d888aeb 100644 --- a/rpcs3/Emu/Cell/lv2/sys_net.cpp +++ b/rpcs3/Emu/Cell/lv2/sys_net.cpp @@ -236,6 +236,18 @@ void fmt_class_string::format(std::string& out, u64 arg) fmt::append(out, "%u.%u.%u.%u", data[0], data[1], data[2], data[3]); } +template <> +void fmt_class_string::format(std::string& out, u64 arg) +{ + const auto& info = get_object(arg); + + fmt::append(out, "{ s=%d, proto=%s, recv_queue_length=%d, send_queue_length=%d, local=%s:%d, remote=%s:%d, state=%d }", + info.s, static_cast(info.proto.value()), info.recv_queue_length, info.send_queue_length, + np::ip_to_string(std::bit_cast(info.local_adr._s_addr)), info.local_port, + np::ip_to_string(std::bit_cast(info.remote_adr._s_addr)), info.remote_port, + info.state); +} + lv2_socket::lv2_socket(utils::serial& ar, lv2_socket_type _type) : family(ar) , type(_type) @@ -1242,8 +1254,8 @@ error_code sys_net_bnet_close(ppu_thread& ppu, s32 s) sock->close(); { - // Ensures the socket has no lingering copy from the network thread - std::lock_guard nw_lock(g_fxo->get().mutex_thread_loop); + // Ensures the socket has no lingering copy from the network threads + std::scoped_lock threads_lock(g_fxo->get().mutex_thread_loop, g_fxo->get().mutex_thread_loop); sock.reset(); } @@ -1262,7 +1274,7 @@ error_code sys_net_bnet_poll(ppu_thread& ppu, vm::ptr fds, s32 n } atomic_t signaled{0}; - + bool has_sockets = false; u64 timeout = ms < 0 ? 0 : ms * 1000ull; std::vector fds_buf; @@ -1273,6 +1285,7 @@ error_code sys_net_bnet_poll(ppu_thread& ppu, vm::ptr fds, s32 n lv2_obj::prepare_for_sleep(ppu); std::unique_lock nw_lock(g_fxo->get().mutex_thread_loop); + std::unique_lock p2p_lock(g_fxo->get().mutex_thread_loop); std::shared_lock lock(id_manager::g_mutex); std::vector<::pollfd> _fds(nfds); @@ -1292,6 +1305,7 @@ error_code sys_net_bnet_poll(ppu_thread& ppu, vm::ptr fds, s32 n if (auto sock = idm::check_unlocked(fds_buf[i].fd)) { + has_sockets = true; sock->poll(fds_buf[i], _fds[i]); #ifdef _WIN32 connecting[i] = sock->is_connecting(); @@ -1326,6 +1340,7 @@ error_code sys_net_bnet_poll(ppu_thread& ppu, vm::ptr fds, s32 n if (ms == 0 || signaled) { lock.unlock(); + p2p_lock.unlock(); nw_lock.unlock(); std::memcpy(fds.get_ptr(), fds_buf.data(), nfds * sizeof(sys_net_pollfd)); return not_an_error(signaled); @@ -1404,7 +1419,7 @@ error_code sys_net_bnet_poll(ppu_thread& ppu, vm::ptr fds, s32 n return {}; } - has_timedout = network_clear_queue(ppu); + has_timedout = network_clear_queue(ppu) || !has_sockets; clear_ppu_to_awake(ppu); ppu.state -= cpu_flag::signal; break; @@ -1451,6 +1466,7 @@ error_code sys_net_bnet_select(ppu_thread& ppu, s32 nfds, vm::ptrtv_sec * 1000000ull + _timeout->tv_usec; + bool has_sockets = false; if (nfds > 0 && nfds <= 1024) { @@ -1462,6 +1478,7 @@ error_code sys_net_bnet_select(ppu_thread& ppu, s32 nfds, vm::ptrget().mutex_thread_loop); + std::lock_guard p2p_lock(g_fxo->get().mutex_thread_loop); reader_lock lock(id_manager::g_mutex); std::vector<::pollfd> _fds(nfds); @@ -1484,6 +1501,7 @@ error_code sys_net_bnet_select(ppu_thread& ppu, s32 nfds, vm::ptrget().mutex_thread_loop); + std::scoped_lock threads_lock(g_fxo->get().mutex_thread_loop, g_fxo->get().mutex_thread_loop); const auto sock = idm::get_unlocked(static_cast(arg)); @@ -1805,8 +1823,9 @@ error_code sys_net_abort(ppu_thread& ppu, s32 type, u64 arg, s32 flags) sys_net.success("lv2_socket::handle_abort(): Closed socket %d", id); } - // Ensures the socket has no lingering copy from the network thread + // Ensures the socket has no lingering copy from the network threads g_fxo->get().mutex_thread_loop.lock_unlock(); + g_fxo->get().mutex_thread_loop.lock_unlock(); return not_an_error(::narrow(sockets.size()) - failed); } @@ -1822,6 +1841,16 @@ error_code sys_net_abort(ppu_thread& ppu, s32 type, u64 arg, s32 flags) return CELL_OK; } +struct net_infoctl_cmd_6_t +{ + be_t sock_id; + be_t zero_0; + be_t zero_1; + vm::bptr sock_info; + be_t n; + be_t zero_2; +}; + struct net_infoctl_cmd_9_t { be_t zero; @@ -1829,29 +1858,117 @@ struct net_infoctl_cmd_9_t // More (TODO) }; +static void net_write_sockinfo(s32 s, lv2_socket& sock, sys_net_sockinfo_t& info) +{ + info = {}; + info.s = s; + + switch (sock.get_type()) + { + case SYS_NET_SOCK_DGRAM: + case SYS_NET_SOCK_DGRAM_P2P: + info.proto = SYS_NET_IPPROTO_UDP; + break; + case SYS_NET_SOCK_STREAM: + case SYS_NET_SOCK_STREAM_P2P: + info.proto = SYS_NET_IPPROTO_TCP; + break; + default: + info.proto = static_cast(sock.get_protocol()); + break; + } + + if (const auto [res, sn_addr] = sock.getsockname(); res == CELL_OK) + { + const auto* addr_in = reinterpret_cast(&sn_addr); + info.local_adr._s_addr = addr_in->sin_addr; + info.local_port = addr_in->sin_port; + } + + if (const auto [res, sn_addr] = sock.getpeername(); res == CELL_OK) + { + const auto* addr_in = reinterpret_cast(&sn_addr); + info.remote_adr._s_addr = addr_in->sin_addr; + info.remote_port = addr_in->sin_port; + } + + sock.get_sockinfo(info); + + sys_net.trace("sys_net_infoctl(cmd=6): %s", info); +} + error_code sys_net_infoctl(ppu_thread& ppu, s32 cmd, vm::ptr arg) { ppu.state += cpu_flag::wait; - sys_net.todo("sys_net_infoctl(cmd=%d, arg=*0x%x)", cmd, arg); + if (cmd == 6 || cmd == 9) + sys_net.notice("sys_net_infoctl(cmd=%d, arg=*0x%x)", cmd, arg); + else + sys_net.todo("sys_net_infoctl(cmd=%d, arg=*0x%x)", cmd, arg); // TODO switch (cmd) { + case 6: + { + vm::ptr cmd_arg = vm::static_ptr_cast(arg); + + const vm::bptr sock_info = cmd_arg->sock_info; + const s32 sock_id = cmd_arg->sock_id; + const s32 max_infos = cmd_arg->n; + + sys_net.trace("cmd 6: sock_id: %d, sock_info: *0x%x, max_infos: %d", sock_id, sock_info, max_infos); + + std::scoped_lock threads_lock(g_fxo->get().mutex_thread_loop, g_fxo->get().mutex_thread_loop); + + if (sock_id == -1) + { + if (!sock_info) + { + return not_an_error(static_cast(idm::select([](u32, lv2_socket&) {}))); + } + + s32 num_infos = 0; + + idm::select([&](u32 id, lv2_socket& sock) + { + if (num_infos >= max_infos) + { + return; + } + + net_write_sockinfo(static_cast(id), sock, sock_info[num_infos]); + num_infos++; + }); + + return not_an_error(num_infos); + } + + if (!sock_info || max_infos < 1) + { + return -SYS_NET_EINVAL; + } + + const auto sock = idm::check(sock_id, [&](lv2_socket& sock) + { + net_write_sockinfo(sock_id, sock, sock_info[0]); + }); + + if (!sock) + { + return -SYS_NET_EBADF; + } + + break; + } case 9: { - constexpr auto nameserver = "nameserver \0"sv; + auto& nph = g_fxo->get>(); + std::string nameserver = "nameserver " + np::ip_to_string(nph.get_dns_ip()); - char buffer[nameserver.size() + 80]{}; - std::memcpy(buffer, nameserver.data(), nameserver.size()); - - auto& nph = g_fxo->get>(); - const auto dns_str = np::ip_to_string(nph.get_dns_ip()); - std::memcpy(buffer + nameserver.size() - 1, dns_str.data(), dns_str.size()); - - std::string_view name{buffer}; - vm::static_ptr_cast(arg)->zero = 0; - std::memcpy(vm::static_ptr_cast(arg)->server_name.get_ptr(), name.data(), name.size()); + vm::ptr cmd_arg = vm::static_ptr_cast(arg); + cmd_arg->zero = 0; + std::memcpy(cmd_arg->server_name.get_ptr(), nameserver.c_str(), nameserver.size() + 1); break; } default: break; diff --git a/rpcs3/Emu/Cell/lv2/sys_net.h b/rpcs3/Emu/Cell/lv2/sys_net.h index 68e7041df0..9297df0ab5 100644 --- a/rpcs3/Emu/Cell/lv2/sys_net.h +++ b/rpcs3/Emu/Cell/lv2/sys_net.h @@ -181,6 +181,24 @@ enum SYS_NET_POLLWRBAND = 0x0100, }; +enum +{ + SYS_NET_STATE_UNKNOWN = 0, + SYS_NET_STATE_CLOSED = 1, + SYS_NET_STATE_CREATED = 2, + SYS_NET_STATE_OPENED = 3, + SYS_NET_STATE_LISTEN = 4, + SYS_NET_STATE_SYN_SENT = 5, + SYS_NET_STATE_SYN_RECEIVED = 6, + SYS_NET_STATE_ESTABLISHED = 7, + SYS_NET_STATE_FIN_WAIT_1 = 8, + SYS_NET_STATE_FIN_WAIT_2 = 9, + SYS_NET_STATE_CLOSE_WAIT = 10, + SYS_NET_STATE_CLOSING = 11, + SYS_NET_STATE_LAST_ACK = 12, + SYS_NET_STATE_TIME_WAIT = 13, +}; + enum lv2_socket_abort_flags : s32 { SYS_NET_ABORT_STRICT_CHECK = 1, @@ -335,6 +353,19 @@ struct sys_net_linger be_t l_linger; }; +struct sys_net_sockinfo_t +{ + be_t s; + be_t proto; + be_t recv_queue_length; + be_t send_queue_length; + sys_net_in_addr local_adr; + be_t local_port; + sys_net_in_addr remote_adr; + be_t remote_port; + be_t state; +}; + class ppu_thread; // Syscalls diff --git a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket.h b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket.h index 1deb6efd2c..4cf9d02392 100644 --- a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket.h +++ b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket.h @@ -107,6 +107,8 @@ public: virtual void poll(sys_net_pollfd& sn_pfd, pollfd& native_pfd) = 0; virtual std::tuple select(bs_t selected, pollfd& native_pfd) = 0; + virtual void get_sockinfo(sys_net_sockinfo_t& info) = 0; + error_code abort_socket(s32 flags); public: diff --git a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_native.cpp b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_native.cpp index b422db814b..8bfcda66a5 100644 --- a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_native.cpp +++ b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_native.cpp @@ -1211,6 +1211,36 @@ void lv2_socket_native::set_non_blocking() np::set_socket_non_blocking(native_socket); } +void lv2_socket_native::get_sockinfo(sys_net_sockinfo_t& info) +{ + if (type != SYS_NET_SOCK_STREAM) + { + info.state = info.local_port ? SYS_NET_STATE_OPENED : SYS_NET_STATE_CREATED; + return; + } + + if (info.remote_port) + { + info.state = SYS_NET_STATE_ESTABLISHED; + return; + } + + { + std::lock_guard lock(mutex); + + int listening = 0; + socklen_t len = sizeof(listening); + + if (::getsockopt(native_socket, SOL_SOCKET, SO_ACCEPTCONN, reinterpret_cast(&listening), &len) == 0 && listening) + { + info.state = SYS_NET_STATE_LISTEN; + return; + } + } + + info.state = info.local_port ? SYS_NET_STATE_OPENED : SYS_NET_STATE_CREATED; +} + bool lv2_socket_native::is_socket_connected() { if (type != SYS_NET_SOCK_STREAM) diff --git a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_native.h b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_native.h index af9e6a57b7..e864dc40b5 100644 --- a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_native.h +++ b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_native.h @@ -52,6 +52,7 @@ public: void poll(sys_net_pollfd& sn_pfd, pollfd& native_pfd) override; std::tuple select(bs_t selected, pollfd& native_pfd) override; + void get_sockinfo(sys_net_sockinfo_t& info) override; bool is_socket_connected(); 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 903752085b..5f0addd656 100644 --- a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2p.cpp +++ b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2p.cpp @@ -364,20 +364,40 @@ s32 lv2_socket_p2p::shutdown([[maybe_unused]] s32 how) return CELL_OK; } +void lv2_socket_p2p::get_sockinfo(sys_net_sockinfo_t& info) +{ + std::lock_guard lock(mutex); + info.state = vport ? SYS_NET_STATE_OPENED : SYS_NET_STATE_CREATED; +} + +bs_t lv2_socket_p2p::get_pending_events() const +{ + bs_t pending{}; + + if (vport && !data.empty()) + { + sys_net.trace("[P2P] p2p_data for vport %d contains %d elements", vport, data.size()); + pending += lv2_socket::poll_t::read; + } + + pending += lv2_socket::poll_t::write; + + return pending; +} + void lv2_socket_p2p::poll(sys_net_pollfd& sn_pfd, [[maybe_unused]] pollfd& native_pfd) { std::lock_guard lock(mutex); ensure(vport); - // Check if it's a bound P2P socket - if ((sn_pfd.events & SYS_NET_POLLIN) && !data.empty()) + const bs_t pending = get_pending_events(); + + if ((sn_pfd.events & SYS_NET_POLLIN) && (pending & lv2_socket::poll_t::read)) { - sys_net.trace("[P2P] p2p_data for vport %d contains %d elements", vport, data.size()); sn_pfd.revents |= SYS_NET_POLLIN; } - // Data can always be written on a dgram socket - if (sn_pfd.events & SYS_NET_POLLOUT) + if ((sn_pfd.events & SYS_NET_POLLOUT) && (pending & lv2_socket::poll_t::write)) { sn_pfd.revents |= SYS_NET_POLLOUT; } @@ -387,20 +407,10 @@ std::tuple lv2_socket_p2p::select(bs_t sel { std::lock_guard lock(mutex); - bool read_set = false; - bool write_set = false; + const bs_t pending = get_pending_events() & selected; - // Check if it's a bound P2P socket - if ((selected & lv2_socket::poll_t::read) && vport && !data.empty()) - { - sys_net.trace("[P2P] p2p_data for vport %d contains %d elements", vport, data.size()); - read_set = true; - } - - if (selected & lv2_socket::poll_t::write) - { - write_set = true; - } + const bool read_set = !!(pending & lv2_socket::poll_t::read); + const bool write_set = !!(pending & lv2_socket::poll_t::write); return {read_set, write_set, false}; } diff --git a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2p.h b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2p.h index ec6c1d8b31..0556b2269e 100644 --- a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2p.h +++ b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2p.h @@ -32,10 +32,13 @@ public: void poll(sys_net_pollfd& sn_pfd, pollfd& native_pfd) override; std::tuple select(bs_t selected, pollfd& native_pfd) override; + void get_sockinfo(sys_net_sockinfo_t& info) override; void handle_new_data(sys_net_sockaddr_in_p2p p2p_addr, std::vector p2p_data); protected: + virtual bs_t get_pending_events() const; + // Port(actual bound port) and Virtual Port(indicated by u16 at the start of the packet) u16 port = 3658, vport = 0; u32 bound_addr = 0; 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 cb59557458..ce1dd56579 100644 --- a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2ps.cpp +++ b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2ps.cpp @@ -300,6 +300,38 @@ void lv2_socket_p2ps::save(utils::serial& ar) ar(status, max_backlog, backlog, op_port, op_vport, op_addr, data_beg_seq, received_data, cur_seq); } +void lv2_socket_p2ps::signal_pending_events() +{ + const bs_t pending = get_pending_events(); + bs_t events_happening{}; + + if ((pending & lv2_socket::poll_t::read) && events.test_and_reset(lv2_socket::poll_t::read)) + events_happening += lv2_socket::poll_t::read; + if ((pending & lv2_socket::poll_t::write) && events.test_and_reset(lv2_socket::poll_t::write)) + events_happening += lv2_socket::poll_t::write; + + if (!events_happening) + { + return; + } + + for (auto it = queue.begin(); it != queue.end();) + { + if (it->second(events_happening)) + { + it = queue.erase(it); + continue; + } + + it++; + } + + if (queue.empty()) + { + events.store({}); + } +} + bool lv2_socket_p2ps::handle_connected(p2ps_encapsulated_tcp* tcp_header, u8* data, ::sockaddr_storage* op_addr, nt_p2p_port* p2p_port) { std::lock_guard lock(mutex); @@ -334,25 +366,8 @@ bool lv2_socket_p2ps::handle_connected(p2ps_encapsulated_tcp* tcp_header, u8* da sys_net.trace("[P2PS] Sent ack %d", final_ack); send_u2s_packet(std::move(packet), reinterpret_cast<::sockaddr_in*>(op_addr), 0, false); - // check if polling is happening - if (data_available && events.test_and_reset(lv2_socket::poll_t::read)) - { - bs_t read_event = lv2_socket::poll_t::read; - for (auto it = queue.begin(); it != queue.end();) - { - if (it->second(read_event)) - { - it = queue.erase(it); - continue; - } - it++; - } - - if (queue.empty()) - { - events.store({}); - } - } + // Report that it is writeable if any poll/select are waiting on it + signal_pending_events(); }; if (status == p2ps_stream_status::stream_handshaking) @@ -483,24 +498,7 @@ bool lv2_socket_p2ps::handle_listening(p2ps_encapsulated_tcp* tcp_header, [[mayb } backlog.push_back(new_sock_id); - if (events.test_and_reset(lv2_socket::poll_t::read)) - { - bs_t read_event = lv2_socket::poll_t::read; - for (auto it = queue.begin(); it != queue.end();) - { - if (it->second(read_event)) - { - it = queue.erase(it); - continue; - } - it++; - } - - if (queue.empty()) - { - events.store({}); - } - } + signal_pending_events(); } else { @@ -541,7 +539,7 @@ void lv2_socket_p2ps::send_u2s_packet(std::vector data, const ::sockaddr_in* void lv2_socket_p2ps::close_stream_nl(nt_p2p_port* p2p_port) { - status = p2ps_stream_status::stream_closed; + status = p2ps_stream_status::stream_disconnected; for (auto it = p2p_port->bound_p2p_streams.begin(); it != p2p_port->bound_p2p_streams.end();) { @@ -555,6 +553,9 @@ void lv2_socket_p2ps::close_stream_nl(nt_p2p_port* p2p_port) auto& tcpm = g_fxo->get>(); tcpm.clear_all_messages(lv2_id); + + // Notify poll/select waiting on read, now that the stream is closed, it will read as readable with 0 bytes + signal_pending_events(); } void lv2_socket_p2ps::close_stream() @@ -740,7 +741,7 @@ std::optional lv2_socket_p2ps::connect(const sys_net_sockaddr& addr) { std::lock_guard lock(mutex); - if (status != p2ps_stream_status::stream_closed) + 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; @@ -752,7 +753,19 @@ std::optional lv2_socket_p2ps::connect(const sys_net_sockaddr& addr) // This is purposefully inverted, not a bug const u16 dst_vport = psa_in_p2p->sin_port; - const u16 dst_port = psa_in_p2p->sin_vport; + u16 dst_port = psa_in_p2p->sin_vport; + + sys_net.trace("[P2PS] Connecting to %s:%d:%d", name.sin_addr, dst_port, dst_vport); + + if (!dst_vport) + { + return -SYS_NET_EADDRNOTAVAIL; + } + + if (!dst_port) + { + dst_port = SCE_NP_PORT; + } socket_type real_socket{}; @@ -772,10 +785,13 @@ std::optional lv2_socket_p2ps::connect(const sys_net_sockaddr& addr) sys_net.warning("[P2PS] vport was unassigned before connect!"); vport = pport.get_port(); - while (pport.bound_p2p_vports.count(vport) || pport.bound_p2p_streams.count(static_cast(vport) << 32)) + 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); @@ -826,10 +842,16 @@ std::optional, sys_net_sockaddr>> lv2_socket_p2p if (!data_available) { + if (status == p2ps_stream_status::stream_disconnected) + { + sys_net.trace("[P2PS] Called recvfrom on a disconnected socket"); + return {{0, {}, {}}}; + } + if (status == p2ps_stream_status::stream_closed) { - sys_net.error("[P2PS] Called recvfrom on closed socket!"); - return {{0, {}, {}}}; + sys_net.error("[P2PS] Called recvfrom on an unconnected socket!"); + return {{-SYS_NET_ENOTCONN, {}, {}}}; } if (so_nbio || (flags & SYS_NET_MSG_DONTWAIT)) @@ -888,10 +910,16 @@ std::optional lv2_socket_p2ps::sendto([[maybe_unused]] s32 flags, const std lock.lock(); } + if (status == p2ps_stream_status::stream_disconnected) + { + sys_net.error("[P2PS] Called sendto on a disconnected socket!"); + return -SYS_NET_ECONNRESET; + } + if (status == p2ps_stream_status::stream_closed) { - sys_net.error("[P2PS] Called sendto on a closed socket!"); - return -SYS_NET_ECONNRESET; + sys_net.error("[P2PS] Called sendto on an unconnected socket!"); + return -SYS_NET_ENOTCONN; } constexpr u32 max_data_len = (65535 - (VPORT_P2P_HEADER_SIZE + sizeof(p2ps_encapsulated_tcp))); @@ -985,23 +1013,71 @@ s32 lv2_socket_p2ps::shutdown([[maybe_unused]] s32 how) return CELL_OK; } +void lv2_socket_p2ps::get_sockinfo(sys_net_sockinfo_t& info) +{ + std::lock_guard lock(mutex); + + switch (status) + { + case p2ps_stream_status::stream_closed: + info.state = vport ? SYS_NET_STATE_OPENED : SYS_NET_STATE_CREATED; + break; + case p2ps_stream_status::stream_listening: info.state = SYS_NET_STATE_LISTEN; break; + case p2ps_stream_status::stream_handshaking: info.state = SYS_NET_STATE_SYN_SENT; break; + case p2ps_stream_status::stream_connected: info.state = SYS_NET_STATE_ESTABLISHED; break; + case p2ps_stream_status::stream_disconnected: info.state = SYS_NET_STATE_CLOSED; break; + } + + info.recv_queue_length = static_cast(data_available); +} + +bs_t lv2_socket_p2ps::get_pending_events() const +{ + bs_t pending{}; + + if (status == p2ps_stream_status::stream_connected) + { + if (data_available) + { + sys_net.trace("[P2PS] socket has %d bytes available", data_available); + pending += lv2_socket::poll_t::read; + } + + pending += lv2_socket::poll_t::write; + } + else if (status == p2ps_stream_status::stream_listening) + { + if (const auto bsize = backlog.size()) + { + sys_net.trace("[P2PS] socket has %d clients available", bsize); + pending += lv2_socket::poll_t::read; + } + } + else if (status == p2ps_stream_status::stream_disconnected) + { + // Reads return EOF(0 bytes) and writes fail with ECONNRESET, report both so that + // waiters blocked on either one wake up instead of hanging until their timeout + pending += lv2_socket::poll_t::read; + pending += lv2_socket::poll_t::write; + } + + return pending; +} + void lv2_socket_p2ps::poll(sys_net_pollfd& sn_pfd, [[maybe_unused]] pollfd& native_pfd) { std::lock_guard lock(mutex); - sys_net.trace("[P2PS] poll checking for 0x%X", sn_pfd.events); - if (status == p2ps_stream_status::stream_connected) - { - if ((sn_pfd.events & SYS_NET_POLLIN) && data_available) - { - sys_net.trace("[P2PS] p2ps has %u bytes available", data_available); - sn_pfd.revents |= SYS_NET_POLLIN; - } - // Data can only be written if the socket is connected - if (sn_pfd.events & SYS_NET_POLLOUT && status == p2ps_stream_status::stream_connected) - { - sn_pfd.revents |= SYS_NET_POLLOUT; - } + const bs_t pending = get_pending_events(); + + if ((sn_pfd.events & SYS_NET_POLLIN) && (pending & lv2_socket::poll_t::read)) + { + sn_pfd.revents |= SYS_NET_POLLIN; + } + + if ((sn_pfd.events & SYS_NET_POLLOUT) && (pending & lv2_socket::poll_t::write)) + { + sn_pfd.revents |= SYS_NET_POLLOUT; } } @@ -1009,32 +1085,10 @@ std::tuple lv2_socket_p2ps::select(bs_t se { std::lock_guard lock(mutex); - bool read_set = false; - bool write_set = false; + const bs_t pending = get_pending_events() & selected; - if (status == p2ps_stream_status::stream_connected) - { - if ((selected & lv2_socket::poll_t::read) && data_available) - { - sys_net.trace("[P2PS] socket has %d bytes available", data_available); - read_set = true; - } - - if (selected & lv2_socket::poll_t::write) - { - sys_net.trace("[P2PS] socket is writeable"); - write_set = true; - } - } - else if (status == p2ps_stream_status::stream_listening) - { - const auto bsize = backlog.size(); - if ((selected & lv2_socket::poll_t::read) && bsize) - { - sys_net.trace("[P2PS] socket has %d clients available", bsize); - read_set = true; - } - } + const bool read_set = !!(pending & lv2_socket::poll_t::read); + const bool write_set = !!(pending & lv2_socket::poll_t::write); return {read_set, write_set, false}; } diff --git a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2ps.h b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2ps.h index ac23528d57..6fccdcd290 100644 --- a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2ps.h +++ b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_p2ps.h @@ -34,10 +34,11 @@ struct p2ps_encapsulated_tcp enum p2ps_stream_status { - stream_closed, // Default when port is not listening nor connected - stream_listening, // Stream is listening, accepting SYN packets - stream_handshaking, // Currently handshaking - stream_connected, // This is an established connection(after tcp handshake) + stream_closed, // Default when the stream was never listening nor connected + stream_listening, // Stream is listening, accepting SYN packets + stream_handshaking, // Currently handshaking + stream_connected, // This is an established connection(after tcp handshake) + stream_disconnected, // Was connected but is now closed }; enum p2ps_tcp_flags : u8 @@ -91,8 +92,11 @@ public: void poll(sys_net_pollfd& sn_pfd, pollfd& native_pfd) override; std::tuple select(bs_t selected, pollfd& native_pfd) override; + void get_sockinfo(sys_net_sockinfo_t& info) override; private: + bs_t get_pending_events() const override; + void signal_pending_events(); void close_stream_nl(nt_p2p_port* p2p_port); private: diff --git a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_raw.cpp b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_raw.cpp index 39ae39e5b5..2b68a06170 100644 --- a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_raw.cpp +++ b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_raw.cpp @@ -139,6 +139,11 @@ void lv2_socket_raw::poll([[maybe_unused]] sys_net_pollfd& sn_pfd, [[maybe_unuse LOG_ONCE(raw_poll, "lv2_socket_raw::poll"); } +void lv2_socket_raw::get_sockinfo(sys_net_sockinfo_t& info) +{ + info.state = SYS_NET_STATE_UNKNOWN; +} + std::tuple lv2_socket_raw::select([[maybe_unused]] bs_t selected, [[maybe_unused]] pollfd& native_pfd) { LOG_ONCE(raw_select, "lv2_socket_raw::select"); diff --git a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_raw.h b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_raw.h index a03339354f..3d1c64e37d 100644 --- a/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_raw.h +++ b/rpcs3/Emu/Cell/lv2/sys_net/lv2_socket_raw.h @@ -34,4 +34,5 @@ public: void poll(sys_net_pollfd& sn_pfd, pollfd& native_pfd) override; std::tuple select(bs_t selected, pollfd& native_pfd) override; + void get_sockinfo(sys_net_sockinfo_t& info) override; }; diff --git a/rpcs3/Emu/Cell/lv2/sys_net/network_context.cpp b/rpcs3/Emu/Cell/lv2/sys_net/network_context.cpp index 18a1c2a935..37715cb7e6 100644 --- a/rpcs3/Emu/Cell/lv2/sys_net/network_context.cpp +++ b/rpcs3/Emu/Cell/lv2/sys_net/network_context.cpp @@ -302,7 +302,7 @@ void p2p_thread::operator()() #endif if (ret_p2p > 0) { - std::lock_guard lock(list_p2p_ports_mutex); + std::scoped_lock lock(mutex_thread_loop, list_p2p_ports_mutex); auto fd_index = 0; auto process_fd = [&](nt_p2p_port& p2p_port) diff --git a/rpcs3/Emu/Cell/lv2/sys_net/network_context.h b/rpcs3/Emu/Cell/lv2/sys_net/network_context.h index 658196c4b4..aa60962a92 100644 --- a/rpcs3/Emu/Cell/lv2/sys_net/network_context.h +++ b/rpcs3/Emu/Cell/lv2/sys_net/network_context.h @@ -14,13 +14,13 @@ struct base_network_thread shared_mutex mutex_ppu_to_awake; std::vector ppu_to_awake; + shared_mutex mutex_thread_loop; void wake_threads(); }; struct network_thread : base_network_thread { - shared_mutex mutex_thread_loop; atomic_t num_polls = 0; static constexpr auto thread_name = "Network Thread";