Bind our UDP sockets to the same local address used by RTSP handshake

This means we can ensure a consistent local address for our outgoing PING
traffic to keep the UDP flows alive without having to call connect() which breaks
with multi-homed hosts on GFE and Sunshine v0.20 and earlier.
This commit is contained in:
Cameron Gutman 2023-10-26 01:09:26 -05:00
commit 05c3f9c754
10 changed files with 66 additions and 40 deletions

View file

@ -75,13 +75,6 @@ int initializeAudioStream(void) {
memcpy(&avRiKeyId, StreamConfig.remoteInputAesIv, sizeof(avRiKeyId)); memcpy(&avRiKeyId, StreamConfig.remoteInputAesIv, sizeof(avRiKeyId));
avRiKeyId = BE32(avRiKeyId); avRiKeyId = BE32(avRiKeyId);
// For GFE 3.22 compatibility, we must start the audio ping thread before the RTSP handshake.
// It will not reply to our RTSP PLAY request until the audio ping has been received.
rtpSocket = bindUdpSocket(RemoteAddr.ss_family, 0);
if (rtpSocket == INVALID_SOCKET) {
return LastSocketFail();
}
return 0; return 0;
} }
@ -92,8 +85,15 @@ int notifyAudioPortNegotiationComplete(void) {
LC_ASSERT(!pingThreadStarted); LC_ASSERT(!pingThreadStarted);
LC_ASSERT(AudioPortNumber != 0); LC_ASSERT(AudioPortNumber != 0);
// For GFE 3.22 compatibility, we must start the audio ping thread before the RTSP handshake.
// It will not reply to our RTSP PLAY request until the audio ping has been received.
rtpSocket = bindUdpSocket(RemoteAddr.ss_family, &LocalAddr, AddrLen, 0);
if (rtpSocket == INVALID_SOCKET) {
return LastSocketFail();
}
// Connect our audio socket to the target address and port // Connect our audio socket to the target address and port
int err = connectUdpSocket(rtpSocket, &RemoteAddr, RemoteAddrLen, AudioPortNumber); int err = connectUdpSocket(rtpSocket, &RemoteAddr, AddrLen, AudioPortNumber);
if (err != 0) { if (err != 0) {
return err; return err;
} }

View file

@ -9,7 +9,8 @@ static int terminationCallbackErrorCode;
// Common globals // Common globals
char* RemoteAddrString; char* RemoteAddrString;
struct sockaddr_storage RemoteAddr; struct sockaddr_storage RemoteAddr;
SOCKADDR_LEN RemoteAddrLen; struct sockaddr_storage LocalAddr;
SOCKADDR_LEN AddrLen;
int AppVersionQuad[4]; int AppVersionQuad[4];
STREAM_CONFIGURATION StreamConfig; STREAM_CONFIGURATION StreamConfig;
CONNECTION_LISTENER_CALLBACKS ListenerCallbacks; CONNECTION_LISTENER_CALLBACKS ListenerCallbacks;
@ -254,6 +255,7 @@ int LiStartConnection(PSERVER_INFORMATION serverInfo, PSTREAM_CONFIGURATION stre
memcpy(&ListenerCallbacks, clCallbacks, sizeof(ListenerCallbacks)); memcpy(&ListenerCallbacks, clCallbacks, sizeof(ListenerCallbacks));
ListenerCallbacks.connectionTerminated = ClInternalConnectionTerminated; ListenerCallbacks.connectionTerminated = ClInternalConnectionTerminated;
memset(&LocalAddr, 0, sizeof(LocalAddr));
NegotiatedVideoFormat = 0; NegotiatedVideoFormat = 0;
memcpy(&StreamConfig, streamConfig, sizeof(StreamConfig)); memcpy(&StreamConfig, streamConfig, sizeof(StreamConfig));
OriginalVideoBitrate = streamConfig->bitrate; OriginalVideoBitrate = streamConfig->bitrate;
@ -344,13 +346,13 @@ int LiStartConnection(PSERVER_INFORMATION serverInfo, PSTREAM_CONFIGURATION stre
if (RtspPortNumber != 48010) { if (RtspPortNumber != 48010) {
// If we have an alternate RTSP port, use that as our test port. The host probably // If we have an alternate RTSP port, use that as our test port. The host probably
// isn't listening on 47989 or 47984 anyway, since they're using alternate ports. // isn't listening on 47989 or 47984 anyway, since they're using alternate ports.
err = resolveHostName(serverInfo->address, AF_UNSPEC, RtspPortNumber, &RemoteAddr, &RemoteAddrLen); err = resolveHostName(serverInfo->address, AF_UNSPEC, RtspPortNumber, &RemoteAddr, &AddrLen);
if (err != 0) { if (err != 0) {
// Sleep for a second and try again. It's possible that we've attempt to connect // Sleep for a second and try again. It's possible that we've attempt to connect
// before the host has gotten around to listening on the RTSP port. Give it some // before the host has gotten around to listening on the RTSP port. Give it some
// time before retrying. // time before retrying.
PltSleepMs(1000); PltSleepMs(1000);
err = resolveHostName(serverInfo->address, AF_UNSPEC, RtspPortNumber, &RemoteAddr, &RemoteAddrLen); err = resolveHostName(serverInfo->address, AF_UNSPEC, RtspPortNumber, &RemoteAddr, &AddrLen);
} }
} }
else { else {
@ -360,12 +362,12 @@ int LiStartConnection(PSERVER_INFORMATION serverInfo, PSTREAM_CONFIGURATION stre
// TCP 48010 is a last resort because: // TCP 48010 is a last resort because:
// a) it's not always listening and there's a race between listen() on the host and our connect() // a) it's not always listening and there's a race between listen() on the host and our connect()
// b) it's not used at all by certain host versions which perform RTSP over ENet // b) it's not used at all by certain host versions which perform RTSP over ENet
err = resolveHostName(serverInfo->address, AF_UNSPEC, 47984, &RemoteAddr, &RemoteAddrLen); err = resolveHostName(serverInfo->address, AF_UNSPEC, 47984, &RemoteAddr, &AddrLen);
if (err != 0) { if (err != 0) {
err = resolveHostName(serverInfo->address, AF_UNSPEC, 47989, &RemoteAddr, &RemoteAddrLen); err = resolveHostName(serverInfo->address, AF_UNSPEC, 47989, &RemoteAddr, &AddrLen);
} }
if (err != 0) { if (err != 0) {
err = resolveHostName(serverInfo->address, AF_UNSPEC, 48010, &RemoteAddr, &RemoteAddrLen); err = resolveHostName(serverInfo->address, AF_UNSPEC, 48010, &RemoteAddr, &AddrLen);
} }
} }
if (err != 0) { if (err != 0) {

View file

@ -1597,7 +1597,7 @@ int startControlStream(void) {
LC_ASSERT(ControlPortNumber != 0); LC_ASSERT(ControlPortNumber != 0);
enet_address_set_address(&address, (struct sockaddr *)&RemoteAddr, RemoteAddrLen); enet_address_set_address(&address, (struct sockaddr *)&RemoteAddr, AddrLen);
enet_address_set_port(&address, ControlPortNumber); enet_address_set_port(&address, ControlPortNumber);
// Create a client // Create a client
@ -1665,7 +1665,7 @@ int startControlStream(void) {
else { else {
// NB: Do NOT use ControlPortNumber here. 47995 is correct for these old versions. // NB: Do NOT use ControlPortNumber here. 47995 is correct for these old versions.
LC_ASSERT(ControlPortNumber == 0); LC_ASSERT(ControlPortNumber == 0);
ctlSock = connectTcpSocket(&RemoteAddr, RemoteAddrLen, ctlSock = connectTcpSocket(&RemoteAddr, AddrLen,
47995, CONTROL_STREAM_TIMEOUT_SEC); 47995, CONTROL_STREAM_TIMEOUT_SEC);
if (ctlSock == INVALID_SOCKET) { if (ctlSock == INVALID_SOCKET) {
stopping = true; stopping = true;

View file

@ -699,7 +699,7 @@ int startInputStream(void) {
// After Gen 5, we send input on the control stream // After Gen 5, we send input on the control stream
if (AppVersionQuad[0] < 5) { if (AppVersionQuad[0] < 5) {
inputSock = connectTcpSocket(&RemoteAddr, RemoteAddrLen, inputSock = connectTcpSocket(&RemoteAddr, AddrLen,
35043, INPUT_STREAM_TIMEOUT_SEC); 35043, INPUT_STREAM_TIMEOUT_SEC);
if (inputSock == INVALID_SOCKET) { if (inputSock == INVALID_SOCKET) {
return LastSocketFail(); return LastSocketFail();

View file

@ -16,7 +16,8 @@
// Common globals // Common globals
extern char* RemoteAddrString; extern char* RemoteAddrString;
extern struct sockaddr_storage RemoteAddr; extern struct sockaddr_storage RemoteAddr;
extern SOCKADDR_LEN RemoteAddrLen; extern struct sockaddr_storage LocalAddr;
extern SOCKADDR_LEN AddrLen;
extern int AppVersionQuad[4]; extern int AppVersionQuad[4];
extern STREAM_CONFIGURATION StreamConfig; extern STREAM_CONFIGURATION StreamConfig;
extern CONNECTION_LISTENER_CALLBACKS ListenerCallbacks; extern CONNECTION_LISTENER_CALLBACKS ListenerCallbacks;

View file

@ -220,28 +220,36 @@ void closeSocket(SOCKET s) {
#endif #endif
} }
SOCKET bindUdpSocket(int addrfamily, int bufferSize) { SOCKET bindUdpSocket(int addressFamily, struct sockaddr_storage* localAddr, SOCKADDR_LEN addrLen, int bufferSize) {
SOCKET s; SOCKET s;
struct sockaddr_storage addr; LC_SOCKADDR bindAddr;
int err; int err;
SOCKADDR_LEN addrLen;
#ifdef AF_INET6 s = createSocket(addressFamily, SOCK_DGRAM, IPPROTO_UDP, false);
LC_ASSERT(addrfamily == AF_INET || addrfamily == AF_INET6);
addrLen = (addrfamily == AF_INET ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6));
#else
LC_ASSERT(addrfamily == AF_INET);
addrLen = sizeof(struct sockaddr_in);
#endif
s = createSocket(addrfamily, SOCK_DGRAM, IPPROTO_UDP, false);
if (s == INVALID_SOCKET) { if (s == INVALID_SOCKET) {
return INVALID_SOCKET; return INVALID_SOCKET;
} }
memset(&addr, 0, sizeof(addr)); // Use localAddr to bind if it was provided
addr.ss_family = addrfamily; if (localAddr && localAddr->ss_family != 0) {
if (bind(s, (struct sockaddr*) &addr, addrLen) == SOCKET_ERROR) { memcpy(&bindAddr, localAddr, addrLen);
SET_PORT(&bindAddr, 0);
}
else {
// Otherwise wildcard bind to the specified address family
memset(&bindAddr, 0, sizeof(bindAddr));
SET_FAMILY(&bindAddr, addressFamily);
#ifdef AF_INET6
LC_ASSERT(addressFamily == AF_INET || addressFamily == AF_INET6);
addrLen = (addressFamily == AF_INET ? sizeof(struct sockaddr_in) : sizeof(struct sockaddr_in6));
#else
LC_ASSERT(addressFamily == AF_INET);
addrLen = sizeof(struct sockaddr_in);
#endif
}
if (bind(s, (struct sockaddr*) &bindAddr, addrLen) == SOCKET_ERROR) {
err = LastSocketError(); err = LastSocketError();
Limelog("bind() failed: %d\n", err); Limelog("bind() failed: %d\n", err);
closeSocket(s); closeSocket(s);

View file

@ -68,9 +68,11 @@ typedef socklen_t SOCKADDR_LEN;
#ifdef AF_INET6 #ifdef AF_INET6
typedef struct sockaddr_in6 LC_SOCKADDR; typedef struct sockaddr_in6 LC_SOCKADDR;
#define SET_FAMILY(addr, family) ((addr)->sin6_family = (family))
#define SET_PORT(addr, port) ((addr)->sin6_port = htons(port)) #define SET_PORT(addr, port) ((addr)->sin6_port = htons(port))
#else #else
typedef struct sockaddr_in LC_SOCKADDR; typedef struct sockaddr_in LC_SOCKADDR;
#define SET_FAMILY(addr, family) ((addr)->sin_family = (family))
#define SET_PORT(addr, port) ((addr)->sin_port = htons(port)) #define SET_PORT(addr, port) ((addr)->sin_port = htons(port))
#endif #endif
@ -88,7 +90,7 @@ SOCKET createSocket(int addressFamily, int socketType, int protocol, bool nonBlo
SOCKET connectTcpSocket(struct sockaddr_storage* dstaddr, SOCKADDR_LEN addrlen, unsigned short port, int timeoutSec); SOCKET connectTcpSocket(struct sockaddr_storage* dstaddr, SOCKADDR_LEN addrlen, unsigned short port, int timeoutSec);
int connectUdpSocket(SOCKET s, struct sockaddr_storage* dstaddr, SOCKADDR_LEN addrlen, unsigned short port); int connectUdpSocket(SOCKET s, struct sockaddr_storage* dstaddr, SOCKADDR_LEN addrlen, unsigned short port);
int sendMtuSafe(SOCKET s, char* buffer, int size); int sendMtuSafe(SOCKET s, char* buffer, int size);
SOCKET bindUdpSocket(int addrfamily, int bufferSize); SOCKET bindUdpSocket(int addressFamily, struct sockaddr_storage* localAddr, SOCKADDR_LEN addrLen, int bufferSize);
int enableNoDelay(SOCKET s); int enableNoDelay(SOCKET s);
int setSocketNonBlocking(SOCKET s, bool enabled); int setSocketNonBlocking(SOCKET s, bool enabled);
int recvUdpSocket(SOCKET s, char* buffer, int size, bool useSelect); int recvUdpSocket(SOCKET s, char* buffer, int size, bool useSelect);

View file

@ -229,7 +229,7 @@ static bool transactRtspMessageTcp(PRTSP_MESSAGE request, PRTSP_MESSAGE response
// returns HTTP 200 OK for the /launch request before the RTSP handshake port // returns HTTP 200 OK for the /launch request before the RTSP handshake port
// is listening. // is listening.
do { do {
sock = connectTcpSocket(&RemoteAddr, RemoteAddrLen, RtspPortNumber, RTSP_CONNECT_TIMEOUT_SEC); sock = connectTcpSocket(&RemoteAddr, AddrLen, RtspPortNumber, RTSP_CONNECT_TIMEOUT_SEC);
if (sock == INVALID_SOCKET) { if (sock == INVALID_SOCKET) {
*error = LastSocketError(); *error = LastSocketError();
if (*error == ECONNREFUSED) { if (*error == ECONNREFUSED) {
@ -320,6 +320,18 @@ static bool transactRtspMessageTcp(PRTSP_MESSAGE request, PRTSP_MESSAGE response
Limelog("Failed to parse RTSP response\n"); Limelog("Failed to parse RTSP response\n");
} }
// Fetch the local address for this socket if it's not populated yet
if (LocalAddr.ss_family == 0) {
SOCKADDR_LEN addrLen = (SOCKADDR_LEN)sizeof(LocalAddr);
if (getsockname(sock, (struct sockaddr*)&LocalAddr, &addrLen) < 0) {
Limelog("Failed to get local address: %d\n", LastSocketError());
memset(&LocalAddr, 0, sizeof(LocalAddr));
}
else {
LC_ASSERT(addrLen == AddrLen);
}
}
Exit: Exit:
if (serializedMessage != NULL) { if (serializedMessage != NULL) {
free(serializedMessage); free(serializedMessage);
@ -819,7 +831,7 @@ int performRtspHandshake(PSERVER_INFORMATION serverInfo) {
ENetAddress address; ENetAddress address;
ENetEvent event; ENetEvent event;
enet_address_set_address(&address, (struct sockaddr *)&RemoteAddr, RemoteAddrLen); enet_address_set_address(&address, (struct sockaddr *)&RemoteAddr, AddrLen);
enet_address_set_port(&address, RtspPortNumber); enet_address_set_port(&address, RtspPortNumber);
// Create a client that can use 1 outgoing connection and 1 channel // Create a client that can use 1 outgoing connection and 1 channel

View file

@ -73,7 +73,7 @@ int LiFindExternalAddressIP4(const char* stunServer, unsigned short stunPort, un
goto Exit; goto Exit;
} }
sock = bindUdpSocket(hints.ai_family, 0); sock = bindUdpSocket(hints.ai_family, NULL, 0, 0);
if (sock == INVALID_SOCKET) { if (sock == INVALID_SOCKET) {
err = LastSocketFail(); err = LastSocketFail();
Limelog("Failed to connect to STUN server: %d\n", err); Limelog("Failed to connect to STUN server: %d\n", err);

View file

@ -261,7 +261,8 @@ int startVideoStream(void* rendererContext, int drFlags) {
return err; return err;
} }
rtpSocket = bindUdpSocket(RemoteAddr.ss_family, RTP_RECV_PACKETS_BUFFERED * (StreamConfig.packetSize + MAX_RTP_HEADER_SIZE)); rtpSocket = bindUdpSocket(RemoteAddr.ss_family, &LocalAddr, AddrLen,
RTP_RECV_PACKETS_BUFFERED * (StreamConfig.packetSize + MAX_RTP_HEADER_SIZE));
if (rtpSocket == INVALID_SOCKET) { if (rtpSocket == INVALID_SOCKET) {
VideoCallbacks.cleanup(); VideoCallbacks.cleanup();
return LastSocketError(); return LastSocketError();
@ -269,7 +270,7 @@ int startVideoStream(void* rendererContext, int drFlags) {
// Connect our video socket to the target address and port // Connect our video socket to the target address and port
LC_ASSERT(VideoPortNumber != 0); LC_ASSERT(VideoPortNumber != 0);
err = connectUdpSocket(rtpSocket, &RemoteAddr, RemoteAddrLen, VideoPortNumber); err = connectUdpSocket(rtpSocket, &RemoteAddr, AddrLen, VideoPortNumber);
if (err != 0) { if (err != 0) {
VideoCallbacks.cleanup(); VideoCallbacks.cleanup();
closeSocket(rtpSocket); closeSocket(rtpSocket);
@ -301,7 +302,7 @@ int startVideoStream(void* rendererContext, int drFlags) {
if (AppVersionQuad[0] == 3) { if (AppVersionQuad[0] == 3) {
// Connect this socket to open port 47998 for our ping thread // Connect this socket to open port 47998 for our ping thread
firstFrameSocket = connectTcpSocket(&RemoteAddr, RemoteAddrLen, firstFrameSocket = connectTcpSocket(&RemoteAddr, AddrLen,
FIRST_FRAME_PORT, FIRST_FRAME_TIMEOUT_SEC); FIRST_FRAME_PORT, FIRST_FRAME_TIMEOUT_SEC);
if (firstFrameSocket == INVALID_SOCKET) { if (firstFrameSocket == INVALID_SOCKET) {
VideoCallbacks.stop(); VideoCallbacks.stop();