[vlc-devel] [PATCH 1/4] chromecast: use an interrupt to notify the receiving thread of pending work
Steve Lhomme
robux4 at videolabs.io
Mon May 9 22:08:46 CEST 2016
--
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.
replaces https://patches.videolan.org/patch/13129/ and https://patches.videolan.org/patch/13241/
by NOT using a socket
replaces https://patches.videolan.org/patch/13299/ by removing the thread cancelation
---
modules/stream_out/chromecast/cast.cpp | 9 +-
modules/stream_out/chromecast/chromecast.h | 10 +-
modules/stream_out/chromecast/chromecast_ctrl.cpp | 129 ++++++++++++----------
3 files changed, 88 insertions(+), 60 deletions(-)
diff --git a/modules/stream_out/chromecast/cast.cpp b/modules/stream_out/chromecast/cast.cpp
index 99455ed..94808f5 100644
--- a/modules/stream_out/chromecast/cast.cpp
+++ b/modules/stream_out/chromecast/cast.cpp
@@ -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);
diff --git a/modules/stream_out/chromecast/chromecast.h b/modules/stream_out/chromecast/chromecast.h
index fc4e081..fc1f524 100644
--- a/modules/stream_out/chromecast/chromecast.h
+++ b/modules/stream_out/chromecast/chromecast.h
@@ -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 */
diff --git a/modules/stream_out/chromecast/chromecast_ctrl.cpp b/modules/stream_out/chromecast/chromecast_ctrl.cpp
index a86c3e5..b6d7917 100644
--- a/modules/stream_out/chromecast/chromecast_ctrl.cpp
+++ b/modules/stream_out/chromecast/chromecast_ctrl.cpp
@@ -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 );
+}
--
2.7.0
More information about the vlc-devel
mailing list