yindo/gbe_fork
1
0
Fork 0
mirror of https://github.com/Detanup01/gbe_fork.git synced 2026-02-04 05:21:16 +01:00
Code Issues Packages Projects Releases Wiki Activity
Files
e072a14625aafa5ee1b28f2c97b5594e5d42836b
gbe_fork/dll/p2p_manager.cpp

735 lines
24 KiB
C++
Raw Blame History

/* Copyright (C) 2019 Mr Goldberg
This file is part of the Goldberg Emulator
The Goldberg Emulator is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation; either
version 3 of the License, or (at your option) any later version.
The Goldberg Emulator is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public
License along with the Goldberg Emulator; if not, see
<http://www.gnu.org/licenses/>. */
#include "dll/p2p_manager.hpp"
#include "dll/dll.h"
//kingdom 2 crowns doesn't work with a 0.3 delay or lower
// appid 353090 becomes unstable when joining a lobby if the time is too low,
// it takes between ~5-12 seconds to get past the message "Waiting for our local client to connect..." !!!
constexpr static double SESSION_REQUEST_DELAY = 2.0;
// https://partner.steamgames.com/doc/api/ISteamNetworking#SendP2PPacket
// "if we can't get through to the user after a timeout of 20 seconds, then an error will be posted"
constexpr static double SESSION_REQUEST_TIMEOUT = 20.0;
constexpr static double PACKET_MAX_TIME_TO_LIVE = 20.0;
const std::chrono::high_resolution_clock::time_point& P2p_Manager::Packet_t::get_time_created() const
{
return time_created;
}
void P2p_Manager::steam_networking_callback(void *object, Common_Message *msg)
{
// PRINT_DEBUG_ENTRY();
auto *obj = (P2p_Manager *)object;
obj->network_callback(msg);
}
void P2p_Manager::steam_run_every_runcb(void *object)
{
// PRINT_DEBUG_ENTRY();
auto *obj = (P2p_Manager *)object;
obj->periodic_callback();
}
SteamCallBacks* P2p_Manager::get_my_callbacks(const CSteamID &my_id) const
{
const CSteamID our_id_client = settings_client->get_local_steam_id();
const CSteamID our_id_server = settings_server->get_local_steam_id();
SteamCallBacks *callbacks = nullptr;
if (my_id == our_id_client) {
PRINT_DEBUG(" using our client callbacks");
return callbacks_client;
} else if (my_id == our_id_server) {
PRINT_DEBUG(" using our server callbacks");
return callbacks_server;
}
PRINT_DEBUG("[X] Id=[%llu] is not ours!", my_id.ConvertToUint64());
return nullptr;
}
bool P2p_Manager::is_same_peer(const CSteamID &id, const CSteamID &peer_id) const
{
if (id == peer_id) {
return true;
}
return false;
}
void P2p_Manager::send_peer_session_failure(const Peer_Src_t &peer_conn)
{
Common_Message update_msg{};
update_msg.set_source_id(peer_conn.my_dest_id.ConvertToUint64());
update_msg.set_dest_id(peer_conn.remote_id.ConvertToUint64());
update_msg.set_allocated_network(new Network_pb);
update_msg.mutable_network()->set_type(Network_pb::FAILED_CONNECT);
PRINT_DEBUG(
"sent a connection failue packet, src id (our client/gameserver)=[%llu], dest id (peer)=[%llu]",
(uint64)update_msg.source_id(), (uint64)update_msg.dest_id()
);
network->sendTo(&update_msg, true);
}
void P2p_Manager::trigger_session_request(const CSteamID &remote_id, const CSteamID &my_id)
{
PRINT_DEBUG(
"triggering session request callback for source steamid=[%llu], I am=[%llu]",
remote_id.ConvertToUint64(), my_id.ConvertToUint64()
);
P2PSessionRequest_t data{};
data.m_steamIDRemote = remote_id;
get_my_callbacks(my_id)->addCBResult(data.k_iCallback, &data, sizeof(data), SESSION_REQUEST_DELAY);
}
bool P2p_Manager::remove_connection(const CSteamID &remote_id, const CSteamID &my_id)
{
auto rem_it = std::remove_if(connections.begin(), connections.end(), [&remote_id, &my_id, this](const Connection_t &item){
return is_same_peer(remote_id, item.peer_conn.remote_id)
&& is_same_peer(my_id, item.peer_conn.my_dest_id)
;
});
if (connections.end() != rem_it) {
connections.erase(rem_it, connections.end());
return true;
}
return false;
}
P2p_Manager::Connection_t* P2p_Manager::create_connection(const CSteamID &remote_id, const CSteamID &my_id)
{
auto conn = get_connection(remote_id, my_id);
if (conn) {
return conn;
}
Connection_t connection{};
connection.peer_conn.remote_id = remote_id;
connection.peer_conn.my_dest_id = my_id;
auto &conn_ref = connections.emplace_back(std::move(connection));
PRINT_DEBUG(
"created for/them=[%llu], from/me=[%llu]",
conn_ref.peer_conn.remote_id.ConvertToUint64(), conn_ref.peer_conn.my_dest_id.ConvertToUint64()
);
return &conn_ref;
}
P2p_Manager::Connection_t* P2p_Manager::get_connection(const CSteamID &remote_id, const CSteamID &my_id)
{
auto conn = std::find_if(connections.begin(), connections.end(), [&remote_id, &my_id, this](const Connection_t &item) {
return is_same_peer(remote_id, item.peer_conn.remote_id)
&& is_same_peer(my_id, item.peer_conn.my_dest_id)
;
});
if (connections.end() == conn) {
return nullptr;
}
return &(*conn);
}
P2p_Manager::P2p_Manager(
class Settings *settings_client, class Settings *settings_server,
class SteamCallBacks *callbacks_client, class SteamCallBacks *callbacks_server,
class Networking *network, class RunEveryRunCB *run_every_runcb
)
{
this->settings_client = settings_client;
this->settings_server = settings_server;
this->callbacks_client = callbacks_client;
this->callbacks_server = callbacks_server;
this->network = network;
this->run_every_runcb = run_every_runcb;
for (auto settings : { settings_client, settings_server }) {
network->setCallback(CALLBACK_ID_NETWORKING, settings->get_local_steam_id(), &P2p_Manager::steam_networking_callback, this);
network->setCallback(CALLBACK_ID_USER_STATUS, settings->get_local_steam_id(), &P2p_Manager::steam_networking_callback, this);
}
run_every_runcb->add(&P2p_Manager::steam_run_every_runcb, this);
}
P2p_Manager::~P2p_Manager()
{
for (auto settings : { settings_client, settings_server }) {
network->rmCallback(CALLBACK_ID_NETWORKING, settings->get_local_steam_id(), &P2p_Manager::steam_networking_callback, this);
network->rmCallback(CALLBACK_ID_USER_STATUS, settings->get_local_steam_id(), &P2p_Manager::steam_networking_callback, this);
}
run_every_runcb->remove(&P2p_Manager::steam_run_every_runcb, this);
}
bool P2p_Manager::store_packet(
CSteamID my_id, CSteamID steamIDRemote,
const void *pubData, uint32 cubData, int nChannel,
EP2PSend send_type
)
{
std::lock_guard lock(p2p_mtx);
auto conn = create_connection(steamIDRemote, my_id);
{
Packet_t channel_msg{};
channel_msg.is_processed = conn->is_accepted;
channel_msg.send_type = send_type;
if (pubData && cubData > 0) {
channel_msg.data.assign((const char *)pubData, (const char *)pubData + cubData);
}
auto &channel = conn->channels[nChannel];
channel.packets.emplace_back(std::move(channel_msg));
}
PRINT_DEBUG(
"stored msg, size=[%u], from=[%llu] (connection accepted=%i), channel=[%i]",
cubData, steamIDRemote.ConvertToUint64(), (int)conn->is_accepted, nChannel
);
return conn->is_accepted;
}
std::optional<std::tuple<
decltype(P2p_Manager::connections)::iterator,
decltype(P2p_Manager::Connection_t::channels)::iterator,
decltype(P2p_Manager::Channel_t::packets)::iterator
>> P2p_Manager::get_next_packet(CSteamID my_id, int nChannel)
{
const auto my_settings = is_same_peer(my_id, settings_client->get_local_steam_id())
? settings_client
: settings_server;
for (auto conn_it = connections.begin(); connections.end() != conn_it; ++conn_it) {
if (!conn_it->is_accepted) {
continue;
}
auto ch_it = conn_it->channels.find(nChannel);
// channel doesn't exist for this connection
if (conn_it->channels.end() == ch_it) {
continue;
}
auto &packets = ch_it->second.packets;
auto packet_it = packets.begin();
// channel has no packets
if (packets.end() == packet_it) {
continue;
}
// no need to check next packets in this channel if the first one isn't processed yet
// once the connection is accepted, all messages in all channels will be marked as processed
if (!packet_it->is_processed) {
continue;
}
const bool is_packet_for_me = is_same_peer(my_id, conn_it->peer_conn.my_dest_id);
if (!is_packet_for_me) {
bool can_share_packet = false;
switch (my_settings->old_p2p_behavior.mode) {
default:
case OldP2pBehavior::EPacketShareMode::DEFAULT: {
// appids (353090, 301300) do this:
// - send packet from client >>> to gameserver
// - use the **client** to check for these packets
// it should use the gameserver instead
// but the client is expected to return these packets to the game
// this seems to be an old behavior
//
// on the contrary appids (701160, 248390) use k_EP2PSendReliable
// and they do not need the gameserver to share its packets with the client
// even multiplayer in appid 248390 won't work if packets were shared
can_share_packet =
packet_it->send_type == EP2PSend::k_EP2PSendUnreliable ||
packet_it->send_type == EP2PSend::k_EP2PSendUnreliableNoDelay;
}
break;
case OldP2pBehavior::EPacketShareMode::ALWAYS_SHARE: {
can_share_packet = true;
}
break;
case OldP2pBehavior::EPacketShareMode::NEVER_SHARE: {
can_share_packet = false;
}
break;
}
// avoid sharing packets between client and gameserver
if (!can_share_packet) {
continue;
}
}
return std::make_tuple(conn_it, ch_it, packet_it);
}
return {};
}
bool P2p_Manager::send_packet(CSteamID my_id, CSteamID steamIDRemote, const void *pubData, uint32 cubData, EP2PSend eP2PSendType, int nChannel)
{
bool reliable = false;
if (eP2PSendType == EP2PSend::k_EP2PSendReliable || eP2PSendType == EP2PSend::k_EP2PSendReliableWithBuffering) {
reliable = true;
}
// !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
// don't lock 2 or more mutexes at the same time
// to avoid the problem of lock ordering deadlock
// !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
{
std::lock_guard lock(p2p_mtx);
auto conn = create_connection(steamIDRemote, my_id);
if (!conn->is_accepted) {
// https://partner.steamgames.com/doc/api/ISteamNetworking#AcceptP2PSessionWithUser
// "If you've called SendP2PPacket on the other user, this implicitly accepts the session request"
conn->is_accepted = true;
PRINT_DEBUG(
"auto-accepting remote connections from=[%llu], I am=[%llu]",
conn->peer_conn.remote_id.ConvertToUint64(), conn->peer_conn.my_dest_id.ConvertToUint64()
);
}
}
Common_Message msg{};
msg.set_source_id(my_id.ConvertToUint64());
msg.set_dest_id(steamIDRemote.ConvertToUint64());
msg.set_allocated_network(new Network_pb);
msg.mutable_network()->set_type(Network_pb::DATA);
msg.mutable_network()->set_channel(nChannel);
msg.mutable_network()->set_data(pubData, cubData);
msg.mutable_network()->set_send_type(eP2PSendType);
bool ret = false;
{
std::lock_guard lock(global_mutex);
ret = network->sendTo(&msg, reliable);
PRINT_DEBUG(
"Sent remote message with size=[%zu] from=[%llu] to=[%llu], is_ok=%u",
msg.network().data().size(), (uint64)msg.source_id(), (uint64)msg.dest_id(), ret
);
}
return ret;
}
bool P2p_Manager::is_packet_available(CSteamID my_id, uint32 *pcubMsgSize, int nChannel)
{
std::lock_guard lock(p2p_mtx);
const CSteamID our_id_client = settings_client->get_local_steam_id();
const CSteamID our_id_server = settings_server->get_local_steam_id();
if (pcubMsgSize) *pcubMsgSize = 0;
auto packet_opt = get_next_packet(my_id, nChannel);
if (!packet_opt) {
return false;
}
auto &[conn_it, ch_it, packet_it] = packet_opt.value();
uint32 size = static_cast<uint32>(packet_it->data.size());
if (pcubMsgSize) {
*pcubMsgSize = size;
}
PRINT_DEBUG(
" available message from=[%llu], size=[%u]",
conn_it->peer_conn.remote_id.ConvertToUint64(), size
);
return true;
}
bool P2p_Manager::read_packet(CSteamID my_id, void *pubDest, uint32 cubDest, uint32 *pcubMsgSize, CSteamID *psteamIDRemote, int nChannel)
{
std::lock_guard lock(p2p_mtx);
if (pcubMsgSize) *pcubMsgSize = 0;
if (psteamIDRemote) *psteamIDRemote = k_steamIDNil;
auto packet_opt = get_next_packet(my_id, nChannel);
if (!packet_opt) {
return false;
}
auto &[conn_it, ch_it, packet_it] = packet_opt.value();
if (psteamIDRemote) {
*psteamIDRemote = conn_it->peer_conn.remote_id;
}
uint32 size = static_cast<uint32>(packet_it->data.size());
if (cubDest < size) {
// https://partner.steamgames.com/doc/api/ISteamNetworking#ReadP2PPacket
// "If the cubDest buffer is too small for the packet, then the message will be truncated"
size = cubDest;
}
if (pcubMsgSize) {
*pcubMsgSize = size;
}
if (pubDest) {
memcpy(pubDest, packet_it->data.data(), size);
}
ch_it->second.packets.erase(packet_it);
PRINT_DEBUG(
" copied message from=[%llu], size=[%u]",
conn_it->peer_conn.remote_id.ConvertToUint64(), size
);
return true;
}
bool P2p_Manager::close_channel(CSteamID my_id, CSteamID steamIDRemote, int nChannel)
{
std::lock_guard lock(p2p_mtx);
auto conn = get_connection(steamIDRemote, my_id);
if (!conn) {
PRINT_DEBUG(
"[X] no connection to remote user [%llu] was found, I am [%llu]",
steamIDRemote.ConvertToUint64(), my_id.ConvertToUint64()
);
return false;
}
conn->channels.erase(nChannel);
PRINT_DEBUG(
"closed channel [%i] with remote user [%llu], I am [%llu]",
nChannel, steamIDRemote.ConvertToUint64(), my_id.ConvertToUint64()
);
if (conn->channels.empty()) {
// https://partner.steamgames.com/doc/api/ISteamNetworking#CloseP2PChannelWithUser
// "Once all channels to a user have been closed,"
// "the open session to the user will be closed and new data from this user will trigger a new P2PSessionRequest_t callback."
PRINT_DEBUG(
"[?] all channels with remote user [%llu] are closed, removing connection, I am [%llu]",
steamIDRemote.ConvertToUint64(), my_id.ConvertToUint64()
);
remove_connection(steamIDRemote, my_id);
remove_connection(my_id, steamIDRemote);
}
return true;
}
bool P2p_Manager::close_session(CSteamID my_id, CSteamID steamIDRemote)
{
std::lock_guard lock(p2p_mtx);
const bool res_1 = remove_connection(steamIDRemote, my_id);
const bool res_2 = remove_connection(my_id, steamIDRemote);
return res_1 || res_2;
}
bool P2p_Manager::get_session_state(CSteamID my_id, CSteamID steamIDRemote, P2PSessionState_t *pConnectionState)
{
// !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
// don't lock 2 or more mutexes at the same time
// to avoid the problem of lock ordering deadlock
// !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
{
std::lock_guard lock(p2p_mtx);
auto conn = get_connection(steamIDRemote, my_id);
if (!conn) {
if (pConnectionState) {
pConnectionState->m_bConnectionActive = false;
pConnectionState->m_bConnecting = false;
pConnectionState->m_eP2PSessionError = EP2PSessionError::k_EP2PSessionErrorTimeout;
pConnectionState->m_bUsingRelay = false;
pConnectionState->m_nBytesQueuedForSend = 0;
pConnectionState->m_nPacketsQueuedForSend = 0;
pConnectionState->m_nRemoteIP = 0;
pConnectionState->m_nRemotePort = 0;
}
PRINT_DEBUG(
"no connection to remote user [%llu], I am [%llu]",
steamIDRemote.ConvertToUint64(), my_id.ConvertToUint64()
);
return false;
}
if (pConnectionState) {
int32 pending_packets = 0;
int32 pending_bytes = 0;
for (const auto& [ch_idx, channel] : conn->channels) {
pending_packets += (int32)channel.packets.size();
for (const auto &msg : channel.packets) {
pending_bytes += (int32)msg.data.size();
}
}
pConnectionState->m_bConnectionActive = conn->is_accepted;
pConnectionState->m_bConnecting = !conn->is_accepted;
pConnectionState->m_eP2PSessionError = EP2PSessionError::k_EP2PSessionErrorNone;
pConnectionState->m_bUsingRelay = false; // TODO
pConnectionState->m_nPacketsQueuedForSend = pending_packets;
pConnectionState->m_nBytesQueuedForSend = pending_bytes;
}
}
if (pConnectionState) {
std::lock_guard lock(global_mutex);
pConnectionState->m_nRemoteIP = network->getIP(steamIDRemote);
pConnectionState->m_nRemotePort = network->getPort(steamIDRemote);
}
PRINT_DEBUG(
"remote user [%llu] has a session/connection, I am [%llu]",
steamIDRemote.ConvertToUint64(), my_id.ConvertToUint64()
);
return true;
}
bool P2p_Manager::accept_session(CSteamID my_id, CSteamID steamIDRemote)
{
std::lock_guard lock(p2p_mtx);
auto conn = get_connection(steamIDRemote, my_id);
if (!conn) {
PRINT_DEBUG("[X] no connection from=[%llu], I am=[%llu]", steamIDRemote.ConvertToUint64(), my_id.ConvertToUint64());
return false;
}
if (!conn->is_accepted) {
conn->is_accepted = true;
PRINT_DEBUG("accepted new session from=[%llu], I am=[%llu]", steamIDRemote.ConvertToUint64(), my_id.ConvertToUint64());
}
return true;
}
void P2p_Manager::periodic_handle_connections(const std::chrono::high_resolution_clock::time_point &now)
{
auto conn_it = connections.begin();
while (connections.end() != conn_it) {
bool is_remove = false;
if (!conn_it->is_accepted) {
if (check_timedout(conn_it->time_added, SESSION_REQUEST_TIMEOUT, now)) {
is_remove = true;
send_peer_session_failure(conn_it->peer_conn);
}
}
if (is_remove) {
conn_it = connections.erase(conn_it);
} else {
++conn_it;
}
}
}
void P2p_Manager::periodic_handle_channels(const std::chrono::high_resolution_clock::time_point &now)
{
for (auto &conn : connections) {
if (!conn.is_accepted) {
continue;
}
// remove channels with no packets
auto ch_it = conn.channels.begin();
while (conn.channels.end() != ch_it) {
auto &channel = ch_it->second;
if (channel.packets.empty()) {
ch_it = conn.channels.erase(ch_it);
} else {
++ch_it;
}
}
}
}
void P2p_Manager::periodic_handle_packets(const std::chrono::high_resolution_clock::time_point &now)
{
for (auto &conn : connections) {
if (!conn.is_accepted) {
continue;
}
for (auto &[ch_num, channel] : conn.channels) {
// remove outadated packets, and mark the rest as processed
auto packet_it = channel.packets.begin();
while (channel.packets.end() != packet_it) {
bool is_remove = false;
if (check_timedout(packet_it->get_time_created(), PACKET_MAX_TIME_TO_LIVE, now)) {
is_remove = true;
} else {
packet_it->is_processed = true;
}
if (is_remove) {
packet_it = channel.packets.erase(packet_it);
} else {
++packet_it;
}
}
}
}
}
void P2p_Manager::periodic_callback()
{
std::lock_guard lock(p2p_mtx);
auto now = std::chrono::high_resolution_clock::now();
periodic_handle_connections(now);
periodic_handle_channels(now);
periodic_handle_packets(now);
}
void P2p_Manager::network_data_packets(Common_Message *msg)
{
const CSteamID src_id = (uint64)msg->source_id();
const CSteamID my_dest_id = (uint64)msg->dest_id(); // this is us
PRINT_DEBUG("got network msg from [%llu], I am [%llu], type <%u>",
src_id.ConvertToUint64(), my_dest_id.ConvertToUint64(), msg->network().type()
);
switch (msg->network().type()) {
case Network_pb::DATA: {
PRINT_DEBUG("got network data message");
const bool conn_is_accepted = store_packet(
my_dest_id, src_id,
msg->network().data().c_str(), (uint32)msg->network().data().size(),
(int)msg->network().channel(),
(EP2PSend)msg->network().send_type()
);
if (!conn_is_accepted) {
trigger_session_request(src_id, my_dest_id);
}
}
break;
case Network_pb::FAILED_CONNECT: {
PRINT_DEBUG("[X] got connection failure packet");
P2PSessionConnectFail_t data{};
data.m_steamIDRemote = src_id;
data.m_eP2PSessionError = EP2PSessionError::k_EP2PSessionErrorTimeout;
get_my_callbacks(my_dest_id)->addCBResult(data.k_iCallback, &data, sizeof(data));
{
std::lock_guard lock(p2p_mtx);
remove_connection(src_id, my_dest_id);
remove_connection(my_dest_id, src_id);
}
}
break;
}
}
void P2p_Manager::network_low_level(Common_Message *msg)
{
const CSteamID src_id = (uint64)msg->source_id();
const CSteamID my_dest_id = (uint64)msg->dest_id(); // this is us
switch (msg->low_level().type()) {
case Low_Level::CONNECT: {
}
break;
case Low_Level::DISCONNECT: {
bool any_conn_removed = false;
{
std::lock_guard lock(p2p_mtx);
any_conn_removed |= remove_connection(src_id, my_dest_id);
any_conn_removed |= remove_connection(my_dest_id, src_id);
}
if (any_conn_removed) {
PRINT_DEBUG(
"[X] remote user [%llu] disconnected, sending P2PSessionConnectFail_t, I am [%llu]",
src_id.ConvertToUint64(), my_dest_id.ConvertToUint64()
);
P2PSessionConnectFail_t data{};
data.m_steamIDRemote = src_id;
data.m_eP2PSessionError = k_EP2PSessionErrorDestinationNotLoggedIn;
get_my_callbacks(my_dest_id)->addCBResult(data.k_iCallback, &data, sizeof(data));
}
}
break;
}
}
void P2p_Manager::network_callback(Common_Message *msg)
{
if (msg->has_network()) {
network_data_packets(msg);
}
if (msg->has_low_level()) {
network_low_level(msg);
}
}
Reference in New Issue View Git Blame Copy Permalink