sys_net: fixes and improvements

Add thread lock for p2p sockets(used for poll/select to avoid event gap)
Fix poll/select returning EINTR if no sockets were polled
Implement sys_net_infoctl cmd 6(sys_net_get_sockinfo)
Cleanup sys_net_infoctl cmd 9 code
Add SYS_NET_STATE_* values to header
Rewrite P2P and P2PS poll/select implementations
Add extra P2PS state for disconnected and report it as readable with 0 bytes(EOF)
Fix P2PS sockets connecting without being bound first missing hashmap insert
Add missing error check in sceNpSignalingActivateConnection

rpcs3/Emu/Cell/Modules/sceNp.cpp

@@ -6800,6 +6800,11 @@ error_code sceNpSignalingActivateConnection(u32 ctx_id, vm::ptr<SceNpId> 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;

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<s32> s;
-	be_t<s32> proto;
-	be_t<s32> recv_queue_length;
-	be_t<s32> send_queue_length;
-	sys_net_in_addr local_adr;
-	be_t<s32> local_port;
-	sys_net_in_addr remote_adr;
-	be_t<s32> remote_port;
-	be_t<s32> state;
-};
-
 struct sys_net_sockinfo_ex_t
 {
 	be_t<s32> s;

rpcs3/Emu/Cell/lv2/sys_net.cpp

@@ -236,6 +236,18 @@ void fmt_class_string<struct in_addr>::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<sys_net_sockinfo_t>::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<lv2_ip_protocol>(info.proto.value()), info.recv_queue_length, info.send_queue_length,
+		np::ip_to_string(std::bit_cast<u32>(info.local_adr._s_addr)), info.local_port,
+		np::ip_to_string(std::bit_cast<u32>(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<network_context>().mutex_thread_loop);
+		// Ensures the socket has no lingering copy from the network threads
+		std::scoped_lock threads_lock(g_fxo->get<network_context>().mutex_thread_loop, g_fxo->get<p2p_context>().mutex_thread_loop);
 		sock.reset();
 	}
 
@@ -1262,7 +1274,7 @@ error_code sys_net_bnet_poll(ppu_thread& ppu, vm::ptr<sys_net_pollfd> fds, s32 n
 	}
 
 	atomic_t<s32> signaled{0};
-
+	bool has_sockets = false;
 	u64 timeout = ms < 0 ? 0 : ms * 1000ull;
 
 	std::vector<sys_net_pollfd> fds_buf;
@@ -1273,6 +1285,7 @@ error_code sys_net_bnet_poll(ppu_thread& ppu, vm::ptr<sys_net_pollfd> fds, s32 n
 		lv2_obj::prepare_for_sleep(ppu);
 
 		std::unique_lock nw_lock(g_fxo->get<network_context>().mutex_thread_loop);
+		std::unique_lock p2p_lock(g_fxo->get<p2p_context>().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<sys_net_pollfd> fds, s32 n
 
 			if (auto sock = idm::check_unlocked<lv2_socket>(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<sys_net_pollfd> 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<sys_net_pollfd> 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::ptr<sys_net_fd_set
 	sys_net_fd_set rwrite{}, _writefds{};
 	sys_net_fd_set rexcept{}, _exceptfds{};
 	u64 timeout = !_timeout ? 0 : _timeout->tv_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::ptr<sys_net_fd_set
 			_exceptfds = *exceptfds;
 
 		std::lock_guard nw_lock(g_fxo->get<network_context>().mutex_thread_loop);
+		std::lock_guard p2p_lock(g_fxo->get<p2p_context>().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::ptr<sys_net_fd_set
 			if (selected)
 			{
 				selected += lv2_socket::poll_t::error;
+				has_sockets = true;
 			}
 			else
 			{
@@ -1638,7 +1656,7 @@ error_code sys_net_bnet_select(ppu_thread& ppu, s32 nfds, vm::ptr<sys_net_fd_set
 					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;
@@ -1765,7 +1783,7 @@ error_code sys_net_abort(ppu_thread& ppu, s32 type, u64 arg, s32 flags)
 	{
 	case _socket:
 	{
-		std::lock_guard nw_lock(g_fxo->get<network_context>().mutex_thread_loop);
+		std::scoped_lock threads_lock(g_fxo->get<network_context>().mutex_thread_loop, g_fxo->get<p2p_context>().mutex_thread_loop);
 
 		const auto sock = idm::get_unlocked<lv2_socket>(static_cast<u32>(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<network_context>().mutex_thread_loop.lock_unlock();
+		g_fxo->get<p2p_context>().mutex_thread_loop.lock_unlock();
 
 		return not_an_error(::narrow<s32>(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<s32> sock_id;
+	be_t<u32> zero_0;
+	be_t<u32> zero_1;
+	vm::bptr<sys_net_sockinfo_t> sock_info;
+	be_t<s32> n;
+	be_t<u32> zero_2;
+};
+
 struct net_infoctl_cmd_9_t
 {
 	be_t<u32> 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<s32>(sock.get_protocol());
+		break;
+	}
+
+	if (const auto [res, sn_addr] = sock.getsockname(); res == CELL_OK)
+	{
+		const auto* addr_in = reinterpret_cast<const sys_net_sockaddr_in*>(&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<const sys_net_sockaddr_in*>(&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<void> 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<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;
+		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<network_context>().mutex_thread_loop, g_fxo->get<p2p_context>().mutex_thread_loop);
+
+		if (sock_id == -1)
+		{
+			if (!sock_info)
+			{
+				return not_an_error(static_cast<s32>(idm::select<lv2_socket>([](u32, lv2_socket&) {})));
+			}
+
+			s32 num_infos = 0;
+
+			idm::select<lv2_socket>([&](u32 id, lv2_socket& sock)
+				{
+					if (num_infos >= max_infos)
+					{
+						return;
+					}
+
+					net_write_sockinfo(static_cast<s32>(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<lv2_socket>(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<named_thread<np::np_handler>>();
+		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<named_thread<np::np_handler>>();
-		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<net_infoctl_cmd_9_t>(arg)->zero = 0;
-		std::memcpy(vm::static_ptr_cast<net_infoctl_cmd_9_t>(arg)->server_name.get_ptr(), name.data(), name.size());
+		vm::ptr<net_infoctl_cmd_9_t> cmd_arg = vm::static_ptr_cast<net_infoctl_cmd_9_t>(arg);
+		cmd_arg->zero = 0;
+		std::memcpy(cmd_arg->server_name.get_ptr(), nameserver.c_str(), nameserver.size() + 1);
 		break;
 	}
 	default: break;

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<s32> l_linger;
 };
 
+struct sys_net_sockinfo_t
+{
+	be_t<s32> s;
+	be_t<s32> proto;
+	be_t<s32> recv_queue_length;
+	be_t<s32> send_queue_length;
+	sys_net_in_addr local_adr;
+	be_t<s32> local_port;
+	sys_net_in_addr remote_adr;
+	be_t<s32> remote_port;
+	be_t<s32> state;
+};
+
 class ppu_thread;
 
 // Syscalls

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<bool, bool, bool> select(bs_t<poll_t> selected, pollfd& native_pfd) = 0;
 
+	virtual void get_sockinfo(sys_net_sockinfo_t& info) = 0;
+
 	error_code abort_socket(s32 flags);
 
 public:

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<char*>(&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)

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<bool, bool, bool> select(bs_t<poll_t> selected, pollfd& native_pfd) override;
+	void get_sockinfo(sys_net_sockinfo_t& info) override;
 
 	bool is_socket_connected();
 

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::poll_t> lv2_socket_p2p::get_pending_events() const
+{
+	bs_t<lv2_socket::poll_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<lv2_socket::poll_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<bool, bool, bool> lv2_socket_p2p::select(bs_t<lv2_socket::poll_t> sel
 {
 	std::lock_guard lock(mutex);
 
-	bool read_set  = false;
-	bool write_set = false;
+	const bs_t<lv2_socket::poll_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};
 }

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<bool, bool, bool> select(bs_t<poll_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<u8> p2p_data);
 
 protected:
+	virtual bs_t<poll_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;

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<lv2_socket::poll_t> pending = get_pending_events();
+	bs_t<lv2_socket::poll_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<lv2_socket::poll_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<lv2_socket::poll_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<u8> 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<named_thread<tcp_timeout_monitor>>();
 	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<s32> 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<s32> 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<s32> 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<u64>(vport) << 32))
+				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);
@@ -826,10 +842,16 @@ std::optional<std::tuple<s32, std::vector<u8>, 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<s32> 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<s32>(data_available);
+}
+
+bs_t<lv2_socket::poll_t> lv2_socket_p2ps::get_pending_events() const
+{
+	bs_t<lv2_socket::poll_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<lv2_socket::poll_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<bool, bool, bool> lv2_socket_p2ps::select(bs_t<lv2_socket::poll_t> se
 {
 	std::lock_guard lock(mutex);
 
-	bool read_set  = false;
-	bool write_set = false;
+	const bs_t<lv2_socket::poll_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};
 }

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<bool, bool, bool> select(bs_t<poll_t> selected, pollfd& native_pfd) override;
+	void get_sockinfo(sys_net_sockinfo_t& info) override;
 
 private:
+	bs_t<poll_t> get_pending_events() const override;
+	void signal_pending_events();
 	void close_stream_nl(nt_p2p_port* p2p_port);
 
 private:

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<bool, bool, bool> lv2_socket_raw::select([[maybe_unused]] bs_t<lv2_socket::poll_t> selected, [[maybe_unused]] pollfd& native_pfd)
 {
 	LOG_ONCE(raw_select, "lv2_socket_raw::select");

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<bool, bool, bool> select(bs_t<poll_t> selected, pollfd& native_pfd) override;
+	void get_sockinfo(sys_net_sockinfo_t& info) override;
 };

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)

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_thread*> ppu_to_awake;
+	shared_mutex mutex_thread_loop;
 
 	void wake_threads();
 };
 
 struct network_thread : base_network_thread
 {
-	shared_mutex mutex_thread_loop;
 	atomic_t<u32> num_polls = 0;
 
 	static constexpr auto thread_name = "Network Thread";