Commit bf2f1278 authored by Steve Lhomme's avatar Steve Lhomme Committed by Jean-Baptiste Kempf

chromecast: use an interrupt to notify the receiving thread of pending work

In some cases we will need to send STOP or SEEK messages from some threads
that have been interrupted, resulting in TLS not sending the data. So now
we send messages from the receiving thread.
Signed-off-by: Jean-Baptiste Kempf's avatarJean-Baptiste Kempf <jb@videolan.org>
parent 6a2f6ea6
......@@ -410,6 +410,10 @@ static int Open(vlc_object_t *p_this)
int i_device_port;
std::stringstream ss;
vlc_interrupt_t *p_interrupt = vlc_interrupt_create();
if (unlikely(p_interrupt == NULL))
goto error;
config_ChainParse(p_stream, SOUT_CFG_PREFIX, ppsz_sout_options, p_stream->p_cfg);
psz_ip = var_GetNonEmptyString( p_stream, SOUT_CFG_PREFIX "ip");
......@@ -422,12 +426,13 @@ static int Open(vlc_object_t *p_this)
i_device_port = var_InheritInteger(p_stream, SOUT_CFG_PREFIX "port");
i_local_server_port = var_InheritInteger(p_stream, SOUT_CFG_PREFIX "http-port");
p_intf = new(std::nothrow) intf_sys_t( p_this, i_local_server_port, psz_ip, i_device_port );
p_intf = new(std::nothrow) intf_sys_t( p_this, i_local_server_port, psz_ip, i_device_port, p_interrupt );
if ( p_intf == NULL)
{
msg_Err( p_this, "cannot load the Chromecast controler" );
goto error;
}
p_interrupt = NULL;
psz_mux = var_GetNonEmptyString(p_stream, SOUT_CFG_PREFIX "mux");
if (psz_mux == NULL)
......@@ -471,6 +476,8 @@ static int Open(vlc_object_t *p_this)
return VLC_SUCCESS;
error:
if (p_interrupt)
vlc_interrupt_destroy(p_interrupt);
delete p_intf;
free(psz_ip);
free(psz_mux);
......
......@@ -32,6 +32,7 @@
#include <vlc_common.h>
#include <vlc_plugin.h>
#include <vlc_tls.h>
#include <vlc_interrupt.h>
#include <sstream>
......@@ -77,7 +78,7 @@ enum receiver_state {
*****************************************************************************/
struct intf_sys_t
{
intf_sys_t(vlc_object_t * const p_this, int local_port, std::string device_addr, int device_port = 0);
intf_sys_t(vlc_object_t * const p_this, int local_port, std::string device_addr, int device_port, vlc_interrupt_t *);
~intf_sys_t();
bool isFinishedPlaying() {
......@@ -145,6 +146,8 @@ private:
void processMessage(const castchannel::CastMessage &msg);
void notifySendRequest();
int sendMessage(const castchannel::CastMessage &msg);
void buildMessage(const std::string & namespace_,
......@@ -170,6 +173,11 @@ private:
bool has_input;
static void* ChromecastThread(void* p_data);
vlc_interrupt_t *p_ctl_thread_interrupt;
int recvPacket(bool &b_msgReceived, uint32_t &i_payloadSize,
unsigned *pi_received, uint8_t *p_data, bool *pb_pingTimeout,
int *pi_wait_delay, int *pi_wait_retries);
};
#endif /* VLC_CHROMECAST_H */
......@@ -91,7 +91,7 @@ void intf_sys_t::buildMessage(const std::string & namespace_,
/*****************************************************************************
* intf_sys_t: class definition
*****************************************************************************/
intf_sys_t::intf_sys_t(vlc_object_t * const p_this, int port, std::string device_addr, int device_port)
intf_sys_t::intf_sys_t(vlc_object_t * const p_this, int port, std::string device_addr, int device_port, vlc_interrupt_t *p_interrupt)
: p_module(p_this)
, i_port(port)
, i_target_port(device_port)
......@@ -105,6 +105,7 @@ intf_sys_t::intf_sys_t(vlc_object_t * const p_this, int port, std::string device
, i_receiver_requestId(0)
, i_requestId(0)
, has_input(false)
, p_ctl_thread_interrupt(p_interrupt)
{
vlc_mutex_init(&lock);
vlc_cond_init(&loadCommandCond);
......@@ -137,11 +138,14 @@ intf_sys_t::~intf_sys_t()
break;
}
vlc_cancel(chromecastThread);
vlc_interrupt_kill( p_ctl_thread_interrupt );
vlc_join(chromecastThread, NULL);
disconnectChromecast();
vlc_interrupt_destroy( p_ctl_thread_interrupt );
vlc_cond_destroy(&loadCommandCond);
vlc_mutex_destroy(&lock);
}
......@@ -240,9 +244,8 @@ void intf_sys_t::disconnectChromecast()
* @param i_payloadSize returns the payload size of the message received
* @return the number of bytes received of -1 on error
*/
// Use here only C linkage and POD types as this function is a cancelation point.
extern "C" int recvPacket(vlc_object_t *p_module, bool &b_msgReceived,
uint32_t &i_payloadSize, int i_sock_fd, vlc_tls_t *p_tls,
int intf_sys_t::recvPacket(bool &b_msgReceived,
uint32_t &i_payloadSize,
unsigned *pi_received, uint8_t *p_data, bool *pb_pingTimeout,
int *pi_wait_delay, int *pi_wait_retries)
{
......@@ -254,7 +257,11 @@ extern "C" int recvPacket(vlc_object_t *p_module, bool &b_msgReceived,
* If we do not receive one after 6 seconds, we send a PING.
* If after this PING, we do not receive a PONG, then we consider the
* connection as dead. */
if (poll(ufd, 1, *pi_wait_delay) == 0)
ssize_t val = vlc_poll_i11e(ufd, 1, *pi_wait_delay);
if ( val == -1 && errno != EINTR )
return -1;
if (val == 0)
{
if (*pb_pingTimeout)
{
......@@ -282,37 +289,54 @@ extern "C" int recvPacket(vlc_object_t *p_module, bool &b_msgReceived,
*pi_wait_retries = PING_WAIT_RETRIES;
}
int i_ret;
/* Packet structure:
* +------------------------------------+------------------------------+
* | Payload size (uint32_t big endian) | Payload data |
* +------------------------------------+------------------------------+ */
while (*pi_received < PACKET_HEADER_LEN)
int i_ret = 0;
if ( ufd[0].revents & POLLIN )
{
// We receive the header.
i_ret = tls_Recv(p_tls, p_data + *pi_received, PACKET_HEADER_LEN - *pi_received);
if (i_ret <= 0)
return i_ret;
*pi_received += i_ret;
}
/* we have received stuff */
// We receive the payload.
/* Packet structure:
* +------------------------------------+------------------------------+
* | Payload size (uint32_t big endian) | Payload data |
* +------------------------------------+------------------------------+ */
while (*pi_received < PACKET_HEADER_LEN)
{
// We receive the header.
i_ret = tls_Recv(p_tls, p_data + *pi_received, PACKET_HEADER_LEN - *pi_received);
if (i_ret <= 0)
return i_ret;
*pi_received += i_ret;
}
// Get the size of the payload
i_payloadSize = U32_AT( p_data );
const uint32_t i_maxPayloadSize = PACKET_MAX_LEN - PACKET_HEADER_LEN;
// We receive the payload.
if (i_payloadSize > i_maxPayloadSize)
{
// Error case: the packet sent by the Chromecast is too long: we drop it.
msg_Err( p_module, "Packet too long: droping its data");
// Get the size of the payload
i_payloadSize = U32_AT( p_data );
const uint32_t i_maxPayloadSize = PACKET_MAX_LEN - PACKET_HEADER_LEN;
uint32_t i_size = i_payloadSize - (*pi_received - PACKET_HEADER_LEN);
if (i_size > i_maxPayloadSize)
i_size = i_maxPayloadSize;
if (i_payloadSize > i_maxPayloadSize)
{
// Error case: the packet sent by the Chromecast is too long: we drop it.
msg_Err( p_module, "Packet too long: droping its data");
uint32_t i_size = i_payloadSize - (*pi_received - PACKET_HEADER_LEN);
if (i_size > i_maxPayloadSize)
i_size = i_maxPayloadSize;
i_ret = tls_Recv(p_tls, p_data + PACKET_HEADER_LEN, i_size);
if (i_ret <= 0)
return i_ret;
*pi_received += i_ret;
if (*pi_received < i_payloadSize + PACKET_HEADER_LEN)
return i_ret;
*pi_received = 0;
return -1;
}
i_ret = tls_Recv(p_tls, p_data + PACKET_HEADER_LEN, i_size);
// Normal case
i_ret = tls_Recv(p_tls, p_data + *pi_received,
i_payloadSize - (*pi_received - PACKET_HEADER_LEN));
if (i_ret <= 0)
return i_ret;
*pi_received += i_ret;
......@@ -320,23 +344,15 @@ extern "C" int recvPacket(vlc_object_t *p_module, bool &b_msgReceived,
if (*pi_received < i_payloadSize + PACKET_HEADER_LEN)
return i_ret;
assert(*pi_received == i_payloadSize + PACKET_HEADER_LEN);
*pi_received = 0;
return -1;
b_msgReceived = true;
}
// Normal case
i_ret = tls_Recv(p_tls, p_data + *pi_received,
i_payloadSize - (*pi_received - PACKET_HEADER_LEN));
if (i_ret <= 0)
return i_ret;
*pi_received += i_ret;
if (*pi_received < i_payloadSize + PACKET_HEADER_LEN)
return i_ret;
if ( val == -1 && errno == EINTR )
/* we have stuff to send */
i_ret = 1;
assert(*pi_received == i_payloadSize + PACKET_HEADER_LEN);
*pi_received = 0;
b_msgReceived = true;
return i_ret;
}
......@@ -808,35 +824,30 @@ void intf_sys_t::pushMediaPlayerMessage(const std::stringstream & payload) {
*****************************************************************************/
void* intf_sys_t::ChromecastThread(void* p_data)
{
int canc;
intf_sys_t *p_sys = reinterpret_cast<intf_sys_t*>(p_data);
p_sys->setConnectionStatus( CHROMECAST_DISCONNECTED );
p_sys->i_sock_fd = p_sys->connectChromecast();
if (p_sys->i_sock_fd < 0)
{
canc = vlc_savecancel();
msg_Err( p_sys->p_module, "Could not connect the Chromecast" );
vlc_mutex_lock(&p_sys->lock);
p_sys->setConnectionStatus(CHROMECAST_CONNECTION_DEAD);
vlc_mutex_unlock(&p_sys->lock);
vlc_restorecancel(canc);
return NULL;
}
char psz_localIP[NI_MAXNUMERICHOST];
if (net_GetSockAddress(p_sys->i_sock_fd, psz_localIP, NULL))
{
canc = vlc_savecancel();
msg_Err( p_sys->p_module, "Cannot get local IP address" );
vlc_mutex_lock(&p_sys->lock);
p_sys->setConnectionStatus(CHROMECAST_CONNECTION_DEAD);
vlc_mutex_unlock(&p_sys->lock);
vlc_restorecancel(canc);
return NULL;
}
canc = vlc_savecancel();
vlc_interrupt_set( p_sys->p_ctl_thread_interrupt );
p_sys->serverIP = psz_localIP;
vlc_mutex_lock(&p_sys->lock);
......@@ -844,9 +855,8 @@ void* intf_sys_t::ChromecastThread(void* p_data)
vlc_mutex_unlock(&p_sys->lock);
p_sys->msgAuth();
vlc_restorecancel(canc);
while ( p_sys->handleMessages() );
while ( !vlc_killed() && p_sys->handleMessages() );
return NULL;
}
......@@ -862,12 +872,11 @@ bool intf_sys_t::handleMessages()
bool b_msgReceived = false;
uint32_t i_payloadSize = 0;
int i_ret = recvPacket( p_module, b_msgReceived, i_payloadSize, i_sock_fd,
p_tls, &i_received, p_packet, &b_pingTimeout,
&i_waitdelay, &i_retries);
int canc = vlc_savecancel();
// Not cancellation-safe part.
int i_ret = recvPacket( b_msgReceived, i_payloadSize,
&i_received, p_packet, &b_pingTimeout,
&i_waitdelay, &i_retries);
#if defined(_WIN32)
if ((i_ret < 0 && WSAGetLastError() != WSAEWOULDBLOCK) || (i_ret == 0))
......@@ -892,8 +901,12 @@ bool intf_sys_t::handleMessages()
msg.ParseFromArray(p_packet + PACKET_HEADER_LEN, i_payloadSize);
processMessage(msg);
}
vlc_restorecancel(canc);
vlc_mutex_locker locker(&lock);
return conn_status != CHROMECAST_CONNECTION_DEAD;
}
void intf_sys_t::notifySendRequest()
{
vlc_interrupt_raise( p_ctl_thread_interrupt );
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment