Commit b5f79391 authored by Olivier Crête's avatar Olivier Crête

outputstream: Don't wake up on every input buffer

So instead of actually blocking on the FD, block on a GCancellable
which is triggered when the writable callback is called. Also set the
application's GCancellable as a child of this source.
parent 83bcf954
...@@ -987,13 +987,17 @@ nice_agent_set_property ( ...@@ -987,13 +987,17 @@ nice_agent_set_property (
static void priv_pseudo_tcp_error (NiceAgent *agent, Stream *stream, static void priv_pseudo_tcp_error (NiceAgent *agent, Stream *stream,
Component *component) Component *component)
{ {
if (component->tcp_writable_cancellable) {
g_cancellable_reset (component->tcp_writable_cancellable);
g_clear_object (&component->tcp_writable_cancellable);
}
if (component->tcp) { if (component->tcp) {
agent_signal_component_state_change (agent, stream->id, agent_signal_component_state_change (agent, stream->id,
component->id, NICE_COMPONENT_STATE_FAILED); component->id, NICE_COMPONENT_STATE_FAILED);
component_detach_all_sockets (component); component_detach_all_sockets (component);
pseudo_tcp_socket_close (component->tcp, TRUE); pseudo_tcp_socket_close (component->tcp, TRUE);
g_object_unref (component->tcp); g_clear_object (&component->tcp);
component->tcp = NULL;
} }
if (component->tcp_clock) { if (component->tcp_clock) {
...@@ -1016,6 +1020,7 @@ pseudo_tcp_socket_opened (PseudoTcpSocket *sock, gpointer user_data) ...@@ -1016,6 +1020,7 @@ pseudo_tcp_socket_opened (PseudoTcpSocket *sock, gpointer user_data)
nice_debug ("Agent %p: s%d:%d pseudo Tcp socket Opened", agent, nice_debug ("Agent %p: s%d:%d pseudo Tcp socket Opened", agent,
stream->id, component->id); stream->id, component->id);
g_cancellable_cancel (component->tcp_writable_cancellable);
g_signal_emit (agent, signals[SIGNAL_RELIABLE_TRANSPORT_WRITABLE], 0, g_signal_emit (agent, signals[SIGNAL_RELIABLE_TRANSPORT_WRITABLE], 0,
stream->id, component->id); stream->id, component->id);
} }
...@@ -1113,6 +1118,7 @@ pseudo_tcp_socket_writable (PseudoTcpSocket *sock, gpointer user_data) ...@@ -1113,6 +1118,7 @@ pseudo_tcp_socket_writable (PseudoTcpSocket *sock, gpointer user_data)
nice_debug ("Agent %p: s%d:%d pseudo Tcp socket writable", agent, nice_debug ("Agent %p: s%d:%d pseudo Tcp socket writable", agent,
stream->id, component->id); stream->id, component->id);
g_cancellable_cancel (component->tcp_writable_cancellable);
g_signal_emit (agent, signals[SIGNAL_RELIABLE_TRANSPORT_WRITABLE], 0, g_signal_emit (agent, signals[SIGNAL_RELIABLE_TRANSPORT_WRITABLE], 0,
stream->id, component->id); stream->id, component->id);
} }
...@@ -1627,6 +1633,7 @@ nice_agent_add_stream ( ...@@ -1627,6 +1633,7 @@ nice_agent_add_stream (
pseudo_tcp_socket_closed, pseudo_tcp_socket_closed,
pseudo_tcp_socket_write_packet}; pseudo_tcp_socket_write_packet};
component->tcp = pseudo_tcp_socket_new (0, &tcp_callbacks); component->tcp = pseudo_tcp_socket_new (0, &tcp_callbacks);
component->tcp_writable_cancellable = g_cancellable_new ();
adjust_tcp_clock (agent, stream, component); adjust_tcp_clock (agent, stream, component);
nice_debug ("Agent %p: Create Pseudo Tcp Socket for component %d", nice_debug ("Agent %p: Create Pseudo Tcp Socket for component %d",
agent, i+1); agent, i+1);
...@@ -2904,6 +2911,9 @@ nice_agent_send_full ( ...@@ -2904,6 +2911,9 @@ nice_agent_send_full (
ret = pseudo_tcp_socket_send (component->tcp, (const gchar *) buf, buf_len); ret = pseudo_tcp_socket_send (component->tcp, (const gchar *) buf, buf_len);
adjust_tcp_clock (agent, stream, component); adjust_tcp_clock (agent, stream, component);
if (!pseudo_tcp_socket_can_send (component->tcp))
g_cancellable_reset (component->tcp_writable_cancellable);
/* In case of -1, the error is either EWOULDBLOCK or ENOTCONN, which both /* In case of -1, the error is either EWOULDBLOCK or ENOTCONN, which both
need the user to wait for the reliable-transport-writable signal */ need the user to wait for the reliable-transport-writable signal */
if (ret < 0 && if (ret < 0 &&
......
...@@ -188,10 +188,10 @@ component_free (Component *cmp) ...@@ -188,10 +188,10 @@ component_free (Component *cmp)
g_source_unref (cmp->tcp_clock); g_source_unref (cmp->tcp_clock);
cmp->tcp_clock = NULL; cmp->tcp_clock = NULL;
} }
g_clear_object (&cmp->tcp_writable_cancellable);
if (cmp->tcp) { if (cmp->tcp) {
pseudo_tcp_socket_close (cmp->tcp, TRUE); pseudo_tcp_socket_close (cmp->tcp, TRUE);
g_object_unref (cmp->tcp); g_clear_object(&cmp->tcp);
cmp->tcp = NULL;
} }
while ((data = g_queue_pop_head (&cmp->pending_io_messages)) != NULL) while ((data = g_queue_pop_head (&cmp->pending_io_messages)) != NULL)
......
...@@ -186,6 +186,7 @@ struct _Component ...@@ -186,6 +186,7 @@ struct _Component
GSource* tcp_clock; GSource* tcp_clock;
long last_clock_timeout; long last_clock_timeout;
gboolean tcp_readable; gboolean tcp_readable;
GCancellable *tcp_writable_cancellable;
guint min_port; guint min_port;
guint max_port; guint max_port;
......
...@@ -558,14 +558,24 @@ nice_output_stream_create_source (GPollableOutputStream *stream, ...@@ -558,14 +558,24 @@ nice_output_stream_create_source (GPollableOutputStream *stream,
Stream *_stream = NULL; Stream *_stream = NULL;
NiceAgent *agent; /* owned */ NiceAgent *agent; /* owned */
component_source = g_pollable_source_new (G_OBJECT (stream));
if (cancellable) {
GSource *cancellable_source = g_cancellable_source_new (cancellable);
g_source_set_dummy_callback (cancellable_source);
g_source_add_child_source (component_source, cancellable_source);
g_source_unref (cancellable_source);
}
/* Closed streams cannot have sources. */ /* Closed streams cannot have sources. */
if (g_output_stream_is_closed (G_OUTPUT_STREAM (stream))) if (g_output_stream_is_closed (G_OUTPUT_STREAM (stream)))
return g_pollable_source_new (G_OBJECT (stream)); /* dummy */ return component_source;
/* Has the agent disappeared? */ /* Has the agent disappeared? */
agent = g_weak_ref_get (&priv->agent_ref); agent = g_weak_ref_get (&priv->agent_ref);
if (agent == NULL) if (agent == NULL)
return g_pollable_source_new (G_OBJECT (stream)); /* dummy */ return component_source;
agent_lock (); agent_lock ();
...@@ -574,17 +584,17 @@ nice_output_stream_create_source (GPollableOutputStream *stream, ...@@ -574,17 +584,17 @@ nice_output_stream_create_source (GPollableOutputStream *stream,
&_stream, &component)) { &_stream, &component)) {
g_warning ("Could not find component %u in stream %u", priv->component_id, g_warning ("Could not find component %u in stream %u", priv->component_id,
priv->stream_id); priv->stream_id);
component_source = g_pollable_source_new (G_OBJECT (stream)); /* dummy */
goto done; goto done;
} }
/* Note: We need G_IO_IN here to handle pseudo-TCP streams. If our TCP if (component->tcp_writable_cancellable) {
* transmit buffer is full, but the kernel's receive buffer has pending ACKs GSource *cancellable_source =
* sitting in it, we need to receive those ACKs so we can transmit the head g_cancellable_source_new (component->tcp_writable_cancellable);
* bytes in the transmit buffer, and hence free up space in the tail of the
* buffer so the stream is writeable again. */ g_source_set_dummy_callback (cancellable_source);
component_source = component_source_new (component, G_OBJECT (stream), g_source_add_child_source (component_source, cancellable_source);
G_IO_IN | G_IO_OUT, cancellable); g_source_unref (cancellable_source);
}
done: done:
agent_unlock (); agent_unlock ();
......
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