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

Merge pull request #408 from otavepto/patch/rp2p

Browse Source 
Revert breaking changes to the P2P networking
This commit is contained in:
Detanup01
2026-01-18 11:37:57 +01:00
committed by GitHub
parent 3dcbaf8f21 63f803dc00
commit 41cd9b8843
13 changed files with 365 additions and 1031 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
dll/dll/p2p_manager.hpp
View File

@@ -1,121 +0,0 @@
/* 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/>. */
#pragma once
#include "base.h"
#include <tuple>
class P2p_Manager
{
private:
struct Peer_Src_t
{
CSteamID remote_id{};
CSteamID my_dest_id{};
};
struct Packet_t
{
private:
std::chrono::high_resolution_clock::time_point time_created = std::chrono::high_resolution_clock::now();
public:
bool is_processed = false;
std::vector<char> data{};
EP2PSend send_type = EP2PSend::k_EP2PSendUnreliable;
const std::chrono::high_resolution_clock::time_point& get_time_created() const;
};
struct Channel_t
{
std::list<Packet_t> packets{};
};
struct Connection_t
{
Peer_Src_t peer_conn{};
std::chrono::high_resolution_clock::time_point time_added = std::chrono::high_resolution_clock::now();
bool is_accepted = false;
std::unordered_map<int, Channel_t> channels{};
};
std::recursive_mutex p2p_mtx{};
std::list<Connection_t> connections{};
class Settings *settings_client{};
class Settings *settings_server{};
class SteamCallBacks *callbacks_client{};
class SteamCallBacks *callbacks_server{};
class Networking *network{};
class RunEveryRunCB *run_every_runcb{};
SteamCallBacks* get_my_callbacks(const CSteamID &my_id) const;
bool is_same_peer(const CSteamID &id, const CSteamID &peer_id) const;
void send_peer_session_failure(const Peer_Src_t &peer_conn);
void trigger_session_request(const CSteamID &remote_id, const CSteamID &my_id);
bool remove_connection(const CSteamID &remote_id, const CSteamID &my_id);
Connection_t* create_connection(const CSteamID &remote_id, const CSteamID &my_id);
Connection_t* get_connection(const CSteamID &remote_id, const CSteamID &my_id);
// true if the connection was already accepted,
// false otherwise.
bool store_packet(
CSteamID my_id, CSteamID steamIDRemote,
const void *pubData, uint32 cubData, int nChannel,
EP2PSend send_type
);
std::optional<std::tuple<
decltype(connections)::iterator,
decltype(Connection_t::channels)::iterator,
decltype(Channel_t::packets)::iterator
>> get_next_packet(CSteamID my_id, int nChannel);
void periodic_handle_connections(const std::chrono::high_resolution_clock::time_point &now);
void periodic_handle_channels(const std::chrono::high_resolution_clock::time_point &now);
void periodic_handle_packets(const std::chrono::high_resolution_clock::time_point &now);
void periodic_callback();
void network_data_packets(Common_Message *msg);
void network_low_level(Common_Message *msg);
void network_callback(Common_Message *msg);
static void steam_networking_callback(void *object, Common_Message *msg);
static void steam_run_every_runcb(void *object);
public:
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
);
~P2p_Manager();
bool send_packet(CSteamID my_id, CSteamID steamIDRemote, const void *pubData, uint32 cubData, EP2PSend eP2PSendType, int nChannel);
bool is_packet_available(CSteamID my_id, uint32 *pcubMsgSize, int nChannel);
bool read_packet(CSteamID my_id, void *pubDest, uint32 cubDest, uint32 *pcubMsgSize, CSteamID *psteamIDRemote, int nChannel);
bool close_channel(CSteamID my_id, CSteamID steamIDRemote, int nChannel);
bool close_session(CSteamID my_id, CSteamID steamIDRemote);
bool get_session_state(CSteamID my_id, CSteamID steamIDRemote, P2PSessionState_t *pConnectionState);
bool accept_session(CSteamID my_id, CSteamID steamIDRemote);
};

29
dll/dll/settings.h
View File

@@ -200,32 +200,6 @@ struct Branch_Info {
bool active = false;
};
struct OldP2pBehavior {
enum class EPacketShareMode {
// if the sending type is unreliable (UDP), share packets between gameserver and client
// otherwise, don't share packets
DEFAULT,
// always share packets between gameserver and client
ALWAYS_SHARE,
// never share packets between gameserver and client
NEVER_SHARE,
_LAST,
};
static EPacketShareMode to_share_mode(int val) {
if (val < 0 || val >= (unsigned)EPacketShareMode::_LAST) {
return EPacketShareMode::DEFAULT;
}
return (EPacketShareMode)val;
}
EPacketShareMode mode = EPacketShareMode::DEFAULT;
};
class Settings {
private:
CSteamID steam_id{}; // user id
@@ -399,9 +373,6 @@ public:
// free weekend
bool free_weekend = false;
// old P2P (ISteamNetworking) behavior
OldP2pBehavior old_p2p_behavior{};
// voice chat
bool enable_voice_chat = false;

1
dll/dll/steam_client.h
View File

@@ -173,7 +173,6 @@ public:
Steam_AppTicket *steam_app_ticket{};
Steam_Overlay* steam_overlay{};
P2p_Manager *p2p_manager{};
PlaytimeCounter* playtime_counter{};

23
dll/dll/steam_networking.h
View File

@@ -19,7 +19,11 @@
#define __INCLUDED_STEAM_NETWORKING_H__
#include "base.h"
#include "p2p_manager.hpp"
struct Steam_Networking_Connection {
CSteamID remote{};
std::set<int> open_channels{};
};
struct steam_listen_socket {
SNetListenSocket_t id{};
@@ -57,25 +61,36 @@ public ISteamNetworking
{
class Settings *settings{};
class Networking *network{};
class P2p_Manager *p2p_manager{};
class SteamCallBacks *callbacks{};
class RunEveryRunCB *run_every_runcb{};
std::recursive_mutex messages_mutex{};
std::list<Common_Message> messages{};
std::list<Common_Message> unprocessed_messages{};
std::recursive_mutex connections_edit_mutex{};
std::vector<struct Steam_Networking_Connection> connections{};
std::vector<struct steam_listen_socket> listen_sockets{};
std::vector<struct steam_connection_socket> connection_sockets{};
std::map<CSteamID, std::chrono::high_resolution_clock::time_point> new_connection_times{};
std::queue<CSteamID> new_connections_to_call_cb{};
SNetListenSocket_t socket_number = 0;
bool connection_exists(CSteamID id);
struct Steam_Networking_Connection *get_or_create_connection(CSteamID id);
void remove_connection(CSteamID id);
SNetSocket_t create_connection_socket(CSteamID target, int nVirtualPort, uint32 nIP, uint16 nPort, SNetListenSocket_t id=0, enum steam_socket_connection_status status=SOCKET_CONNECTING, SNetSocket_t other_id=0);
struct steam_connection_socket *get_connection_socket(SNetSocket_t id);
void remove_killed_connection_sockets();
static void steam_networking_callback(void *object, Common_Message *msg);
static void steam_run_every_runcb(void *object);
static void steam_networking_run_every_runcp(void *object);
public:
Steam_Networking(class Settings *settings, class Networking *network, class P2p_Manager *p2p_manager, class SteamCallBacks *callbacks, class RunEveryRunCB *run_every_runcb);
Steam_Networking(class Settings *settings, class Networking *network, class SteamCallBacks *callbacks, class RunEveryRunCB *run_every_runcb);
~Steam_Networking();
////////////////////////////////////////////////////////////////////////////////////////////

13
dll/net.proto
View File

@@ -81,15 +81,18 @@ message Low_Level {
}
message Network_pb {
uint32 channel = 1;
bytes data = 2;
enum Types {
DATA = 0;
FAILED_CONNECT = 1;
NEW_CONNECTION = 1;
}
Types type = 1;
int32 channel = 2;
bytes data = 3;
uint32 send_type = 4; // EP2PSend
Types type = 3;
bool processed = 128;
uint64 time_processed = 129;
}
message Network_Old {

63
dll/network.cpp
View File

@@ -715,33 +715,34 @@ bool Networking::handle_announce(Common_Message *msg, IP_PORT ip_port)
Common_Message msg_ = create_announce(true);
size_t size = msg_.ByteSizeLong();
std::vector<char> buffer(size);
msg_.SerializeToArray(buffer.data(), static_cast<int>(size));
char *buffer = new char[size];
msg_.SerializeToArray(buffer, static_cast<int>(size));
IP_PORT ipp;
ipp.ip = msg->announce().peers(i).ip();
ipp.port = htons(msg->announce().peers(i).udp_port());
send_packet_to(udp_socket, ipp, buffer.data(), static_cast<unsigned long>(size));
send_packet_to(udp_socket, ipp, buffer, static_cast<unsigned long>(size));
delete[] buffer;
}
}
conn->last_received = std::chrono::high_resolution_clock::now();
if (msg->announce().type() == Announce::PING) {
{
Common_Message msg = create_announce(false);
size_t size = msg.ByteSizeLong();
std::vector<char> buffer(size);
msg.SerializeToArray(buffer.data(), static_cast<int>(size));
send_packet_to(udp_socket, ip_port, buffer.data(), static_cast<unsigned long>(size));
}
char *buffer = new char[size];
msg.SerializeToArray(buffer, static_cast<int>(size));
send_packet_to(udp_socket, ip_port, buffer, static_cast<unsigned long>(size));
delete[] buffer;
//send ping packet if not pinged
if (!conn->udp_pinged) {
Common_Message msg = create_announce(true);
size_t size = msg.ByteSizeLong();
std::vector<char> buffer(size);
msg.SerializeToArray(buffer.data(), static_cast<int>(size));
send_packet_to(udp_socket, ip_port, buffer.data(), static_cast<unsigned long>(size));
char *buffer = new char[size];
msg.SerializeToArray(buffer, static_cast<int>(size));
send_packet_to(udp_socket, ip_port, buffer, static_cast<unsigned long>(size));
delete[] buffer;
}
} else if (msg->announce().type() == Announce::PONG) {
conn->udp_ip_port = ip_port;
@@ -1105,15 +1106,13 @@ void Networking::Run()
auto i = std::find(c.ids.begin(), c.ids.end(), steam_id);
if (i != c.ids.end()) {
c.ids.erase(i);
PRINT_DEBUG("REMOVE OLD USER CONNECTION ID [%llu]", steam_id.ConvertToUint64());
run_callback_user(steam_id, false, c.appid);
PRINT_DEBUG("REMOVE OLD CONNECTION ID");
}
}
}
for (auto &steam_id : conn.ids) {
run_callback_user(steam_id, true, conn.appid);
}
for (auto &steam_id : conn.ids) run_callback_user(steam_id, true, conn.appid);
}
conn.connected = true;
@@ -1150,15 +1149,11 @@ void Networking::Run()
auto conn = std::begin(connections);
while (conn != std::end(connections)) {
if (check_timedout(conn->last_received, USER_TIMEOUT + time_extra)) {
if (conn->connected) {
PRINT_DEBUG("USER TIMEOUT");
for (auto &steam_id : conn->ids) {
run_callback_user(steam_id, false, conn->appid);
}
}
if (conn->connected) for (auto &steam_id : conn->ids) run_callback_user(steam_id, false, conn->appid);
kill_tcp_socket(conn->tcp_socket_outgoing);
kill_tcp_socket(conn->tcp_socket_incoming);
conn = connections.erase(conn);
PRINT_DEBUG("USER TIMEOUT");
} else {
++conn;
}
@@ -1167,11 +1162,7 @@ void Networking::Run()
for (auto &conn: connections) {
if (!(conn.tcp_socket_incoming.received_data || conn.tcp_socket_outgoing.received_data)) {
if (conn.connected) {
for (auto &steam_id : conn.ids) {
run_callback_user(steam_id, false, conn.appid);
}
}
if (conn.connected) for (auto &steam_id : conn.ids) run_callback_user(steam_id, false, conn.appid);
conn.connected = false;
}
}
@@ -1321,29 +1312,15 @@ bool Networking::sendToAll(Common_Message *msg, bool reliable)
void Networking::run_callbacks(Callback_Ids id, Common_Message *msg)
{
const uint64 message_destination_steamid = msg->dest_id();
for (const auto &cb : callbacks[id].callbacks) {
const uint64 callback_allowed_steamid = cb.steam_id.ConvertToUint64();
for (auto &cb : callbacks[id].callbacks) {
uint64 callback_allowed_steamid = cb.steam_id.ConvertToUint64();
uint64 message_destination_steamid = msg->dest_id();
if (callback_allowed_steamid == 0 || // callback wants to receive all messages (callback for broadcast)
message_destination_steamid == 0 || // message was broadcasted to all (broadcast message)
callback_allowed_steamid == message_destination_steamid) { // callback destination is the same as the message destination
// change broadcast destination ID to a specific one
// this is required since otherwise the CSteamID of the destination would be 0 (invalid ID)
if (message_destination_steamid == 0) {
msg->set_dest_id(callback_allowed_steamid);
}
// invoke the callback
cb.message_callback(cb.object, msg);
}
}
// restore broadcast destination ID
if (message_destination_steamid == 0) {
msg->set_dest_id(0);
}
}
void Networking::run_callback_user(CSteamID steam_id, bool online, uint32 appid)

734
dll/p2p_manager.cpp
View File

@@ -1,734 +0,0 @@
/* 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);
}
}

7
dll/settings_parser.cpp
View File

@@ -1583,13 +1583,6 @@ static void parse_simple_features(class Settings *settings_client, class Setting
settings_client->download_steamhttp_requests = ini.GetBoolValue("main::connectivity", "download_steamhttp_requests", settings_client->download_steamhttp_requests);
settings_server->download_steamhttp_requests = ini.GetBoolValue("main::connectivity", "download_steamhttp_requests", settings_server->download_steamhttp_requests);
settings_client->old_p2p_behavior.mode = OldP2pBehavior::to_share_mode(
(int)ini.GetLongValue("main::connectivity", "old_p2p_packet_sharing_mode", (unsigned)settings_client->old_p2p_behavior.mode)
);
settings_server->old_p2p_behavior.mode = OldP2pBehavior::to_share_mode(
(int)ini.GetLongValue("main::connectivity", "old_p2p_packet_sharing_mode", (unsigned)settings_server->old_p2p_behavior.mode)
);
// [main::misc]
settings_client->achievement_bypass = ini.GetBoolValue("main::misc", "achievements_bypass", settings_client->achievement_bypass);

16
dll/steam_client.cpp
View File

@@ -105,17 +105,12 @@ Steam_Client::Steam_Client()
steam_utils = new Steam_Utils(settings_client, callback_results_client, callbacks_client, steam_overlay);
ugc_bridge = new Ugc_Remote_Storage_Bridge(settings_client);
p2p_manager = new P2p_Manager(
settings_client, settings_server,
callbacks_client, callbacks_server,
network, run_every_runcb
);
steam_matchmaking = new Steam_Matchmaking(settings_client, local_storage, network, callback_results_client, callbacks_client, run_every_runcb);
steam_matchmaking_servers = new Steam_Matchmaking_Servers(settings_client, local_storage, network);
steam_user_stats = new Steam_User_Stats(settings_client, network, local_storage, callback_results_client, callbacks_client, run_every_runcb, steam_overlay);
steam_apps = new Steam_Apps(settings_client, callback_results_client, callbacks_client);
steam_networking = new Steam_Networking(settings_client, network, p2p_manager, callbacks_client, run_every_runcb);
steam_networking = new Steam_Networking(settings_client, network, callbacks_client, run_every_runcb);
steam_remote_storage = new Steam_Remote_Storage(settings_client, ugc_bridge, local_storage, callback_results_client, callbacks_client, run_every_runcb);
steam_screenshots = new Steam_Screenshots(local_storage, callbacks_client);
steam_http = new Steam_HTTP(settings_client, network, callback_results_client, callbacks_client);
@@ -150,7 +145,7 @@ Steam_Client::Steam_Client()
steam_gameserver_user = new Steam_User(settings_server, local_storage, network, callback_results_server, callbacks_server);
steam_gameserver_utils = new Steam_Utils(settings_server, callback_results_server, callbacks_server, steam_overlay);
steam_gameserverstats = new Steam_GameServerStats(settings_server, network, callback_results_server, callbacks_server, run_every_runcb);
steam_gameserver_networking = new Steam_Networking(settings_server, network, p2p_manager, callbacks_server, run_every_runcb);
steam_gameserver_networking = new Steam_Networking(settings_server, network, callbacks_server, run_every_runcb);
steam_gameserver_http = new Steam_HTTP(settings_server, network, callback_results_server, callbacks_server);
steam_gameserver_inventory = new Steam_Inventory(settings_server, callback_results_server, callbacks_server, run_every_runcb, local_storage);
steam_gameserver_ugc = new Steam_UGC(settings_server, ugc_bridge, local_storage, callback_results_server, callbacks_server);
@@ -237,7 +232,6 @@ Steam_Client::~Steam_Client()
DEL_INST(steam_app_disable_update);
DEL_INST(steam_billing);
DEL_INST(p2p_manager);
DEL_INST(ugc_bridge);
DEL_INST(steam_utils);
@@ -960,15 +954,15 @@ void Steam_Client::RunCallbacks(bool runClientCB, bool runGameserverCB)
// PRINT_DEBUG("steam_matchmaking_servers *********");
steam_matchmaking_servers->RunCallbacks();
// PRINT_DEBUG("run_every_runcb *********");
run_every_runcb->run();
// PRINT_DEBUG("steam_gameserver *********");
steam_gameserver->RunCallbacks();
// PRINT_DEBUG("steam_user *********");
steam_gameserver_user->RunCallbacks();
// PRINT_DEBUG("run_every_runcb *********");
run_every_runcb->run();
if (runClientCB) {
// PRINT_DEBUG("callback_results_client *********");
callback_results_client->runCallResults();

7
dll/steam_gameserver.cpp
View File

@@ -20,9 +20,6 @@
#define SEND_SERVER_RATE 5.0
// appid 353090 takes ~1.2-1.4 sec to create the gameserver
#define LOGON_DELAY 1.3
void Steam_GameServer::set_version(const char *pchVersionString)
{
@@ -844,14 +841,14 @@ void Steam_GameServer::RunCallbacks()
if (temp_call_servers_connected) {
PRINT_DEBUG("SteamServersConnected_t");
SteamServersConnected_t data{};
callbacks->addCBResult(data.k_iCallback, &data, sizeof(data), LOGON_DELAY);
callbacks->addCBResult(data.k_iCallback, &data, sizeof(data), 0.1);
}
if (logged_in && !policy_response_called) {
PRINT_DEBUG("GSPolicyResponse_t");
GSPolicyResponse_t data{};
data.m_bSecure = !!(flags & k_unServerFlagSecure);
callbacks->addCBResult(data.k_iCallback, &data, sizeof(data), LOGON_DELAY);
callbacks->addCBResult(data.k_iCallback, &data, sizeof(data), 0.11);
policy_response_called = true;
}

7
dll/steam_matchmaking.cpp
View File

@@ -23,14 +23,11 @@
#define REQUEST_LOBBY_DATA_TIMEOUT 6.0
#define LOBBY_DELETED_TIMEOUT 2
// appid 353090 takes ~100ms to create the lobby
#define LOBBY_CREATE_DELAY 0.2
#define LOBBY_CREATE_DELAY 0.07 //artificial delay for lobby creation
#define FILTER_MAX_DEFAULT 4096
// https://partner.steamgames.com/doc/api/ISteamMatchmaking#RequestLobbyList
// "this call can take from 300ms to 5 seconds to complete, and has a timeout of 20 seconds."
#define LOBBY_SEARCH_TIMEOUT 1.0
#define LOBBY_SEARCH_TIMEOUT 0.2 //Tested on real steam
google::protobuf::Map<std::string,std::string>::const_iterator Steam_Matchmaking::caseinsensitive_find(const ::google::protobuf::Map< ::std::string, ::std::string >& map, std::string key)

354
dll/steam_networking.cpp
View File

@@ -16,13 +16,79 @@
<http://www.gnu.org/licenses/>. */
#include "dll/steam_networking.h"
#include "dll/dll.h"
//packet timeout in seconds for non connections
#define ORPHANED_PACKET_TIMEOUT (20)
#define NEW_CONNECTION_TIMEOUT (20.0)
//kingdom 2 crowns doesn't work with a 0.3 delay or lower
#define NEW_CONNECTION_DELAY (0.4)
#define OLD_CHANNEL_NUMBER 1
bool Steam_Networking::connection_exists(CSteamID id)
{
std::lock_guard<std::recursive_mutex> lock(connections_edit_mutex);
return std::find_if(connections.begin(), connections.end(), [&id](struct Steam_Networking_Connection const& conn) { return conn.remote == id;}) != connections.end();
}
struct Steam_Networking_Connection* Steam_Networking::get_or_create_connection(CSteamID id)
{
std::lock_guard<std::recursive_mutex> lock(connections_edit_mutex);
auto conn = std::find_if(connections.begin(), connections.end(), [&id](struct Steam_Networking_Connection const& conn) { return conn.remote == id;});
if (connections.end() == conn) {
struct Steam_Networking_Connection connection;
connection.remote = id;
connections.push_back(connection);
return &(connections[connections.size() - 1]);
} else {
return &(*conn);
}
}
void Steam_Networking::remove_connection(CSteamID id)
{
{
std::lock_guard<std::recursive_mutex> lock(connections_edit_mutex);
auto conn = std::begin(connections);
while (conn != std::end(connections)) {
if (conn->remote == id) {
conn = connections.erase(conn);
} else {
++conn;
}
}
}
//pretty sure steam also clears the entire queue of messages for that connection
{
std::lock_guard<std::recursive_mutex> lock(messages_mutex);
auto msg = std::begin(messages);
while (msg != std::end(messages)) {
if (msg->source_id() == id.ConvertToUint64()) {
msg = messages.erase(msg);
} else {
++msg;
}
}
}
{
auto msg = std::begin(unprocessed_messages);
while (msg != std::end(unprocessed_messages)) {
if (msg->source_id() == id.ConvertToUint64()) {
msg = unprocessed_messages.erase(msg);
} else {
++msg;
}
}
}
}
SNetSocket_t Steam_Networking::create_connection_socket(CSteamID target, int nVirtualPort, uint32 nIP, uint16 nPort, SNetListenSocket_t id, enum steam_socket_connection_status status, SNetSocket_t other_id)
{
static SNetSocket_t socket_number = 0;
@@ -107,7 +173,7 @@ void Steam_Networking::steam_networking_callback(void *object, Common_Message *m
steam_networking->Callback(msg);
}
void Steam_Networking::steam_run_every_runcb(void *object)
void Steam_Networking::steam_networking_run_every_runcp(void *object)
{
// PRINT_DEBUG_ENTRY();
@@ -115,26 +181,25 @@ void Steam_Networking::steam_run_every_runcb(void *object)
steam_networking->RunCallbacks();
}
Steam_Networking::Steam_Networking(class Settings *settings, class Networking *network, class P2p_Manager *p2p_manager, class SteamCallBacks *callbacks, class RunEveryRunCB *run_every_runcb)
Steam_Networking::Steam_Networking(class Settings *settings, class Networking *network, class SteamCallBacks *callbacks, class RunEveryRunCB *run_every_runcb)
{
this->settings = settings;
this->network = network;
this->p2p_manager = p2p_manager;
this->callbacks = callbacks;
this->run_every_runcb = run_every_runcb;
this->network->setCallback(CALLBACK_ID_NETWORKING, settings->get_local_steam_id(), &Steam_Networking::steam_networking_callback, this);
this->network->setCallback(CALLBACK_ID_USER_STATUS, settings->get_local_steam_id(), &Steam_Networking::steam_networking_callback, this);
this->run_every_runcb->add(&Steam_Networking::steam_run_every_runcb, this);
this->run_every_runcb->add(&Steam_Networking::steam_networking_run_every_runcp, this);
PRINT_DEBUG("user id %llu", settings->get_local_steam_id().ConvertToUint64());
PRINT_DEBUG("user id %llu messages: %p", settings->get_local_steam_id().ConvertToUint64(), &messages);
}
Steam_Networking::~Steam_Networking()
{
this->network->rmCallback(CALLBACK_ID_NETWORKING, settings->get_local_steam_id(), &Steam_Networking::steam_networking_callback, this);
this->network->rmCallback(CALLBACK_ID_USER_STATUS, settings->get_local_steam_id(), &Steam_Networking::steam_networking_callback, this);
this->run_every_runcb->remove(&Steam_Networking::steam_run_every_runcb, this);
this->run_every_runcb->remove(&Steam_Networking::steam_networking_run_every_runcp, this);
}
////////////////////////////////////////////////////////////////////////////////////////////
@@ -152,15 +217,31 @@ Steam_Networking::~Steam_Networking()
// using different channels to talk to the same user will still use the same underlying p2p connection, saving on resources
bool Steam_Networking::SendP2PPacket( CSteamID steamIDRemote, const void *pubData, uint32 cubData, EP2PSend eP2PSendType, int nChannel)
{
PRINT_DEBUG(
"size=[%u] sendtype: <%u> channel: [%u] from=[%llu] to=[%llu]",
cubData, eP2PSendType, nChannel, settings->get_local_steam_id().ConvertToUint64(), steamIDRemote.ConvertToUint64()
);
PRINT_DEBUG("len %u sendtype: %u channel: %u to: %llu", cubData, eP2PSendType, nChannel, steamIDRemote.ConvertToUint64());
std::lock_guard<std::recursive_mutex> lock(global_mutex);
bool reliable = false;
if (eP2PSendType == k_EP2PSendReliable || eP2PSendType == k_EP2PSendReliableWithBuffering) reliable = true;
Common_Message msg;
msg.set_source_id(settings->get_local_steam_id().ConvertToUint64());
msg.set_dest_id(steamIDRemote.ConvertToUint64());
msg.set_allocated_network(new Network_pb);
return p2p_manager->send_packet(
settings->get_local_steam_id(), steamIDRemote,
pubData, cubData, eP2PSendType, nChannel
);
if (!connection_exists(steamIDRemote)) {
msg.mutable_network()->set_type(Network_pb::NEW_CONNECTION);
network->sendTo(&msg, true);
}
msg.mutable_network()->set_channel(nChannel);
msg.mutable_network()->set_data(pubData, cubData);
msg.mutable_network()->set_type(Network_pb::DATA);
struct Steam_Networking_Connection *conn = get_or_create_connection(steamIDRemote);
new_connection_times.erase(steamIDRemote);
conn->open_channels.insert(nChannel);
bool ret = network->sendTo(&msg, reliable);
PRINT_DEBUG("Sent message with size: %zu %u", msg.network().data().size(), ret);
return ret;
}
bool Steam_Networking::SendP2PPacket( CSteamID steamIDRemote, const void *pubData, uint32 cubData, EP2PSend eP2PSendType )
@@ -172,20 +253,26 @@ bool Steam_Networking::SendP2PPacket( CSteamID steamIDRemote, const void *pubDat
// returns true if any data is available for read, and the amount of data that will need to be read
bool Steam_Networking::IsP2PPacketAvailable( uint32 *pcubMsgSize, int nChannel)
{
PRINT_DEBUG(
"channel=[%i], my steam id=%llu (is server=%u)",
nChannel, settings->get_local_steam_id().ConvertToUint64(), settings->get_local_steam_id().BGameServerAccount()
);
PRINT_DEBUG("channel: %i", nChannel);
std::lock_guard<std::recursive_mutex> lock(messages_mutex);
//Not sure if this should be here because it slightly screws up games that don't like such low "pings"
//Commenting it out for now because it looks like it causes a bug where 20xx gets stuck in an infinite receive packet loop
//this->network->Run();
//RunCallbacks();
return p2p_manager->is_packet_available(
settings->get_local_steam_id(),
pcubMsgSize, nChannel
);
PRINT_DEBUG("Messages %zu %p", messages.size(), &messages);
for (auto &msg : messages) {
if (connection_exists((uint64)msg.source_id()) && msg.mutable_network()->channel() == nChannel && msg.network().processed()) {
uint32 size = static_cast<uint32>(msg.mutable_network()->data().size());
if (pcubMsgSize) *pcubMsgSize = size;
PRINT_DEBUG("available with size: %u", size);
return true;
}
}
PRINT_DEBUG("(not available)");
if (pcubMsgSize) *pcubMsgSize = 0;
return false;
}
bool Steam_Networking::IsP2PPacketAvailable( uint32 *pcubMsgSize)
@@ -200,17 +287,38 @@ bool Steam_Networking::IsP2PPacketAvailable( uint32 *pcubMsgSize)
// this call is not blocking, and will return false if no data is available
bool Steam_Networking::ReadP2PPacket( void *pubDest, uint32 cubDest, uint32 *pcubMsgSize, CSteamID *psteamIDRemote, int nChannel)
{
PRINT_DEBUG("%u %i %p", cubDest, nChannel, pubDest);
PRINT_DEBUG("%u %i", cubDest, nChannel);
std::lock_guard<std::recursive_mutex> lock(messages_mutex);
//Not sure if this should be here because it slightly screws up games that don't like such low "pings"
//Commenting it out for now because it looks like it causes a bug where 20xx gets stuck in an infinite receive packet loop
//this->network->Run();
//RunCallbacks();
return p2p_manager->read_packet(
settings->get_local_steam_id(),
pubDest, cubDest, pcubMsgSize, psteamIDRemote, nChannel
);
bool read = false;
PRINT_DEBUG("Number messages %zu", messages.size());
auto msg = std::begin(messages);
while (msg != std::end(messages)) {
if (connection_exists((uint64)msg->source_id()) && msg->network().channel() == nChannel && msg->network().processed()) {
uint32 msg_size = static_cast<uint32>(msg->network().data().size());
if (msg_size > cubDest) msg_size = cubDest;
if (pcubMsgSize) *pcubMsgSize = msg_size;
memcpy(pubDest, msg->network().data().data(), msg_size);
PRINT_DEBUG("%s",
common_helpers::uint8_vector_to_hex_string(std::vector<uint8_t>((uint8_t*)pubDest, (uint8_t*)pubDest + msg_size)).c_str());
*psteamIDRemote = CSteamID((uint64)msg->source_id());
PRINT_DEBUG("len %u channel: %u from: " "%" PRIu64 "", msg_size, nChannel, msg->source_id());
msg = messages.erase(msg);
return true;
}
++msg;
}
if (pcubMsgSize) *pcubMsgSize = 0;
if (psteamIDRemote) *psteamIDRemote = k_steamIDNil;
return false;
}
bool Steam_Networking::ReadP2PPacket( void *pubDest, uint32 cubDest, uint32 *pcubMsgSize, CSteamID *psteamIDRemote)
@@ -227,15 +335,11 @@ bool Steam_Networking::ReadP2PPacket( void *pubDest, uint32 cubDest, uint32 *pcu
// (if you've called SendP2PPacket() on the other user, this implicitly accepts the session request)
bool Steam_Networking::AcceptP2PSessionWithUser( CSteamID steamIDRemote )
{
PRINT_DEBUG("from [%llu], I am=[%llu]", steamIDRemote.ConvertToUint64(), settings->get_local_steam_id().ConvertToUint64());
if (p2p_manager->accept_session(settings->get_local_steam_id().ConvertToUint64(), steamIDRemote)) {
PRINT_DEBUG(" connection accepted");
return true;
}
PRINT_DEBUG(" [X] connection NOT accepted");
return false;
PRINT_DEBUG("%llu", steamIDRemote.ConvertToUint64());
std::lock_guard<std::recursive_mutex> lock(global_mutex);
struct Steam_Networking_Connection *conn = get_or_create_connection(steamIDRemote);
if (conn) new_connection_times.erase(steamIDRemote);
return !!conn;
}
@@ -244,8 +348,15 @@ bool Steam_Networking::AcceptP2PSessionWithUser( CSteamID steamIDRemote )
bool Steam_Networking::CloseP2PSessionWithUser( CSteamID steamIDRemote )
{
PRINT_DEBUG("%llu", steamIDRemote.ConvertToUint64());
std::lock_guard<std::recursive_mutex> lock(global_mutex);
if (!connection_exists(steamIDRemote)) {
return p2p_manager->close_session(settings->get_local_steam_id().ConvertToUint64(), steamIDRemote);
return false;
}
remove_connection(steamIDRemote);
return true;
}
@@ -254,9 +365,20 @@ bool Steam_Networking::CloseP2PSessionWithUser( CSteamID steamIDRemote )
// user will trigger a P2PSessionRequest_t callback
bool Steam_Networking::CloseP2PChannelWithUser( CSteamID steamIDRemote, int nChannel )
{
PRINT_DEBUG("%llu", steamIDRemote.ConvertToUint64());
PRINT_DEBUG_ENTRY();
std::lock_guard<std::recursive_mutex> lock(global_mutex);
if (!connection_exists(steamIDRemote)) {
return false;
}
return p2p_manager->close_channel(settings->get_local_steam_id().ConvertToUint64(), steamIDRemote, nChannel);
struct Steam_Networking_Connection *conn = get_or_create_connection(steamIDRemote);
conn->open_channels.erase(nChannel);
if (conn->open_channels.size() == 0) {
remove_connection(steamIDRemote);
}
return true;
}
@@ -266,8 +388,36 @@ bool Steam_Networking::CloseP2PChannelWithUser( CSteamID steamIDRemote, int nCha
bool Steam_Networking::GetP2PSessionState( CSteamID steamIDRemote, P2PSessionState_t *pConnectionState )
{
PRINT_DEBUG("%llu", steamIDRemote.ConvertToUint64());
std::lock_guard<std::recursive_mutex> lock(global_mutex);
if (!connection_exists(steamIDRemote) && (steamIDRemote != settings->get_local_steam_id())) {
if (pConnectionState) {
pConnectionState->m_bConnectionActive = false;
pConnectionState->m_bConnecting = false;
pConnectionState->m_eP2PSessionError = 0;
pConnectionState->m_bUsingRelay = false;
pConnectionState->m_nBytesQueuedForSend = 0;
pConnectionState->m_nPacketsQueuedForSend = 0;
pConnectionState->m_nRemoteIP = 0;
pConnectionState->m_nRemotePort = 0;
}
return p2p_manager->get_session_state(settings->get_local_steam_id().ConvertToUint64(), steamIDRemote, pConnectionState);
PRINT_DEBUG("No Connection");
return false;
}
if (pConnectionState) {
pConnectionState->m_bConnectionActive = true;
pConnectionState->m_bConnecting = false;
pConnectionState->m_eP2PSessionError = 0;
pConnectionState->m_bUsingRelay = false;
pConnectionState->m_nBytesQueuedForSend = 0;
pConnectionState->m_nPacketsQueuedForSend = 0;
pConnectionState->m_nRemoteIP = network->getIP(steamIDRemote);
pConnectionState->m_nRemotePort = 12345;
}
PRINT_DEBUG("Connection");
return true;
}
@@ -278,7 +428,6 @@ bool Steam_Networking::GetP2PSessionState( CSteamID steamIDRemote, P2PSessionSta
// P2P packet relay is allowed by default
bool Steam_Networking::AllowP2PPacketRelay( bool bAllow )
{
PRINT_DEBUG_TODO();
PRINT_DEBUG("%u", bAllow);
return true;
}
@@ -620,18 +769,115 @@ int Steam_Networking::GetMaxPacketSize( SNetSocket_t hSocket )
return 1500;
}
void Steam_Networking::RunCallbacks()
{
uint64 current_time = std::chrono::duration_cast<std::chrono::duration<uint64>>(std::chrono::system_clock::now().time_since_epoch()).count();
{
std::lock_guard<std::recursive_mutex> lock(messages_mutex);
{
auto msg = std::begin(unprocessed_messages);
while (msg != std::end(unprocessed_messages)) {
CSteamID source_id((uint64)msg->source_id());
if (!connection_exists(source_id)) {
if (new_connection_times.find(source_id) == new_connection_times.end()) {
new_connections_to_call_cb.push(source_id);
new_connection_times[source_id] = std::chrono::high_resolution_clock::now();
}
} else {
struct Steam_Networking_Connection *conn = get_or_create_connection(source_id);
conn->open_channels.insert(msg->network().channel());
}
msg->mutable_network()->set_processed(true);
msg->mutable_network()->set_time_processed(current_time);
messages.push_back(*msg);
msg = unprocessed_messages.erase(msg);
}
}
auto msg = std::begin(messages);
while (msg != std::end(messages)) {
bool deleted = false;
if (msg->network().processed()) {
if (!connection_exists((uint64)msg->source_id())) {
if (msg->network().time_processed() + ORPHANED_PACKET_TIMEOUT < current_time) {
deleted = true;
}
}
}
if (deleted) {
msg = messages.erase(msg);
} else {
++msg;
}
}
}
while (!new_connections_to_call_cb.empty()) {
CSteamID source_id = new_connections_to_call_cb.front();
auto t = new_connection_times.find(source_id);
if (t == new_connection_times.end()) {
new_connections_to_call_cb.pop();
continue;
}
if (!check_timedout(t->second, NEW_CONNECTION_DELAY)) {
break;
}
P2PSessionRequest_t data;
memset(&data, 0, sizeof(data));
data.m_steamIDRemote = source_id;
callbacks->addCBResult(data.k_iCallback, &data, sizeof(data));
new_connections_to_call_cb.pop();
}
//TODO: not sure if sockets should be wiped right away
remove_killed_connection_sockets();
for(auto it = new_connection_times.begin(); it != new_connection_times.end(); ) {
if (std::chrono::duration_cast<std::chrono::duration<double>>(std::chrono::high_resolution_clock::now() - it->second).count() > NEW_CONNECTION_TIMEOUT) {
it = new_connection_times.erase(it);
//TODO send packet to other side to tell them connection has "failed".
} else {
++it;
}
}
}
void Steam_Networking::Callback(Common_Message *msg)
{
if (msg->has_network()) {
PRINT_DEBUG("got msg from: " "%" PRIu64 " to: " "%" PRIu64 " size %zu type %u | messages %p: %zu",
msg->source_id(), msg->dest_id(), msg->network().data().size(), msg->network().type(), &messages, messages.size()
);
PRINT_DEBUG("msg data: '%s'",
common_helpers::uint8_vector_to_hex_string(std::vector<uint8_t>(msg->network().data().begin(), msg->network().data().end())).c_str());
if (msg->network().type() == Network_pb::DATA) {
unprocessed_messages.push_back(Common_Message(*msg));
}
if (msg->network().type() == Network_pb::NEW_CONNECTION) {
std::lock_guard<std::recursive_mutex> lock(messages_mutex);
auto msg_temp = std::begin(messages);
while (msg_temp != std::end(messages)) {
//only delete processed to handle unreliable message arriving at the same time.
if (msg_temp->source_id() == msg->source_id() && msg_temp->network().processed()) {
msg_temp = messages.erase(msg_temp);
} else {
++msg_temp;
}
}
}
}
if (msg->has_network_old()) {
PRINT_DEBUG("got old network socket msg %u", msg->network_old().type());
PRINT_DEBUG("got network socket msg %u", msg->network_old().type());
if (msg->network_old().type() == Network_Old::CONNECTION_REQUEST_IP) {
for (auto & listen : listen_sockets) {
if (listen.nPort == msg->network_old().port()) {
@@ -698,9 +944,17 @@ void Steam_Networking::Callback(Common_Message *msg)
if (msg->has_low_level()) {
if (msg->low_level().type() == Low_Level::DISCONNECT) {
CSteamID source_id((uint64)msg->source_id());
if (connection_exists(source_id)) {
P2PSessionConnectFail_t data;
data.m_steamIDRemote = source_id;
data.m_eP2PSessionError = k_EP2PSessionErrorDestinationNotLoggedIn;
callbacks->addCBResult(data.k_iCallback, &data, sizeof(data));
}
for (auto & socket : connection_sockets) {
if (socket.target.ConvertToUint64() == msg->source_id()) {
struct SocketStatusCallback_t data{};
struct SocketStatusCallback_t data;
socket.status = SOCKET_DISCONNECTED;
data.m_hSocket = socket.id;
data.m_hListenSocket = socket.listen_id;
@@ -709,7 +963,9 @@ void Steam_Networking::Callback(Common_Message *msg)
callbacks->addCBResult(data.k_iCallback, &data, sizeof(data));
}
}
} else if (msg->low_level().type() == Low_Level::CONNECT) {
} else
if (msg->low_level().type() == Low_Level::CONNECT) {
}
}

13
post_build/steam_settings.EXAMPLE/configs.main.EXAMPLE.ini
View File

@@ -125,19 +125,6 @@ disable_lobby_creation=0
# this will **not** work if the app is using native/OS web APIs
# default=0
download_steamhttp_requests=0
# sharing mode of the old P2P networking interface (ISteamNetworking).
# this defines how the gameserver and the client intances can read P2P packets received from other players.
#
# some older games require sharing the received packets, i.e if the gameserver instance received a packet,
# then the client instance should be able to detect/read it as well.
# other games won't work properly if the packets were shared,
# and the original receiver (either gameserver or client) must handle that packet
#
# 0=share packets flagged as "unreliable" (k_EP2PSendUnreliable | k_EP2PSendUnreliableNoDelay), otherwise don't share packets
# 1=always share the packets
# 2=never share the packets
# default=0
old_p2p_packet_sharing_mode=0
# mostly workarounds for specific problems
[main::misc]