mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2024-07-05 14:16:06 +00:00
Code Issues 1 Projects Releases Wiki Activity
gstreamer/gst/tcp/gstmultifdsink.c
Tim-Philipp Müller 84a3add409 multifdsink, multisocketsink: fix broken action signal setup 
We can't just make a vfunc that takes a union of int
and pointer as argument, and then set up subclass-specific
action signals and signals that take int (in multifdsink's
case) or a GSocket * (in multisocketsink's case), and then
expect everything to Just Work. This blows up spectacularly
on PPC G4 for some reason.

Fixes multifdsink unit test on PPC, and fixes aborts in
multisocketunit test (now hangs in gst_pad_push - progress).
2012-09-09 01:57:59 +01:00

1146 lines
38 KiB
C
Raw Blame History

/* GStreamer
* Copyright (C) <1999> Erik Walthinsen <omega@cse.ogi.edu>
* Copyright (C) <2004> Thomas Vander Stichele <thomas at apestaart dot org>
* Copyright (C) 2006 Wim Taymans <wim at fluendo dot com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library 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
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 59 Temple Place - Suite 330,
* Boston, MA 02111-1307, USA.
*/
/**
* SECTION:element-multifdsink
* @see_also: tcpserversink
*
* This plugin writes incoming data to a set of file descriptors. The
* file descriptors can be added to multifdsink by emitting the #GstMultiFdSink::add signal.
* For each descriptor added, the #GstMultiFdSink::client-added signal will be called.
*
* As of version 0.10.8, a client can also be added with the #GstMultiFdSink::add-full signal
* that allows for more control over what and how much data a client
* initially receives.
*
* Clients can be removed from multifdsink by emitting the #GstMultiFdSink::remove signal. For
* each descriptor removed, the #GstMultiFdSink::client-removed signal will be called. The
* #GstMultiFdSink::client-removed signal can also be fired when multifdsink decides that a
* client is not active anymore or, depending on the value of the
* #GstMultiFdSink:recover-policy property, if the client is reading too slowly.
* In all cases, multifdsink will never close a file descriptor itself.
* The user of multifdsink is responsible for closing all file descriptors.
* This can for example be done in response to the #GstMultiFdSink::client-fd-removed signal.
* Note that multifdsink still has a reference to the file descriptor when the
* #GstMultiFdSink::client-removed signal is emitted, so that "get-stats" can be performed on
* the descriptor; it is therefore not safe to close the file descriptor in
* the #GstMultiFdSink::client-removed signal handler, and you should use the
* #GstMultiFdSink::client-fd-removed signal to safely close the fd.
*
* Multifdsink internally keeps a queue of the incoming buffers and uses a
* separate thread to send the buffers to the clients. This ensures that no
* client write can block the pipeline and that clients can read with different
* speeds.
*
* When adding a client to multifdsink, the #GstMultiFdSink:sync-method property will define
* which buffer in the queued buffers will be sent first to the client. Clients
* can be sent the most recent buffer (which might not be decodable by the
* client if it is not a keyframe), the next keyframe received in
* multifdsink (which can take some time depending on the keyframe rate), or the
* last received keyframe (which will cause a simple burst-on-connect).
* Multifdsink will always keep at least one keyframe in its internal buffers
* when the sync-mode is set to latest-keyframe.
*
* As of version 0.10.8, there are additional values for the #GstMultiFdSink:sync-method
* property to allow finer control over burst-on-connect behaviour. By selecting
* the 'burst' method a minimum burst size can be chosen, 'burst-keyframe'
* additionally requires that the burst begin with a keyframe, and
* 'burst-with-keyframe' attempts to burst beginning with a keyframe, but will
* prefer a minimum burst size even if it requires not starting with a keyframe.
*
* Multifdsink can be instructed to keep at least a minimum amount of data
* expressed in time or byte units in its internal queues with the
* #GstMultiFdSink:time-min and #GstMultiFdSink:bytes-min properties respectively.
* These properties are useful if the application adds clients with the
* #GstMultiFdSink::add-full signal to make sure that a burst connect can
* actually be honored.
*
* When streaming data, clients are allowed to read at a different rate than
* the rate at which multifdsink receives data. If the client is reading too
* fast, no data will be send to the client until multifdsink receives more
* data. If the client, however, reads too slowly, data for that client will be
* queued up in multifdsink. Two properties control the amount of data
* (buffers) that is queued in multifdsink: #GstMultiFdSink:buffers-max and
* #GstMultiFdSink:buffers-soft-max. A client that falls behind by
* #GstMultiFdSink:buffers-max is removed from multifdsink forcibly.
*
* A client with a lag of at least #GstMultiFdSink:buffers-soft-max enters the recovery
* procedure which is controlled with the #GstMultiFdSink:recover-policy property.
* A recover policy of NONE will do nothing, RESYNC_LATEST will send the most recently
* received buffer as the next buffer for the client, RESYNC_SOFT_LIMIT
* positions the client to the soft limit in the buffer queue and
* RESYNC_KEYFRAME positions the client at the most recent keyframe in the
* buffer queue.
*
* multifdsink will by default synchronize on the clock before serving the
* buffers to the clients. This behaviour can be disabled by setting the sync
* property to FALSE. Multifdsink will by default not do QoS and will never
* drop late buffers.
*
* Last reviewed on 2006-09-12 (0.10.10)
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include <gst/gst-i18n-plugin.h>
#include <sys/ioctl.h>
#ifdef HAVE_UNISTD_H
#include <unistd.h>
#endif
#include <string.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <netinet/in.h>
#ifdef HAVE_FIONREAD_IN_SYS_FILIO
#include <sys/filio.h>
#endif
#include "gstmultifdsink.h"
#include "gsttcp-marshal.h"
#define NOT_IMPLEMENTED 0
GST_DEBUG_CATEGORY_STATIC (multifdsink_debug);
#define GST_CAT_DEFAULT (multifdsink_debug)
/* MultiFdSink signals and args */
enum
{
/* methods */
SIGNAL_ADD,
SIGNAL_ADD_BURST,
SIGNAL_REMOVE,
SIGNAL_REMOVE_FLUSH,
SIGNAL_GET_STATS,
/* signals */
SIGNAL_CLIENT_ADDED,
SIGNAL_CLIENT_REMOVED,
SIGNAL_CLIENT_HANDLE_REMOVED,
LAST_SIGNAL
};
/* this is really arbitrarily chosen */
#define DEFAULT_HANDLE_READ TRUE
enum
{
PROP_0,
PROP_HANDLE_READ,
PROP_LAST
};
static void gst_multi_fd_sink_stop_pre (GstMultiHandleSink * mhsink);
static void gst_multi_fd_sink_stop_post (GstMultiHandleSink * mhsink);
static gboolean gst_multi_fd_sink_start_pre (GstMultiHandleSink * mhsink);
static gpointer gst_multi_fd_sink_thread (GstMultiHandleSink * mhsink);
static void gst_multi_fd_sink_add (GstMultiFdSink * sink, int fd);
static void gst_multi_fd_sink_add_full (GstMultiFdSink * sink, int fd,
GstSyncMethod sync, GstFormat min_format, guint64 min_value,
GstFormat max_format, guint64 max_value);
static void gst_multi_fd_sink_remove (GstMultiFdSink * sink, int fd);
static void gst_multi_fd_sink_remove_flush (GstMultiFdSink * sink, int fd);
static GstStructure *gst_multi_fd_sink_get_stats (GstMultiFdSink * sink,
int fd);
static void gst_multi_fd_sink_emit_client_added (GstMultiHandleSink * mhsink,
GstMultiSinkHandle handle);
static void gst_multi_fd_sink_emit_client_removed (GstMultiHandleSink * mhsink,
GstMultiSinkHandle handle, GstClientStatus status);
static GstMultiHandleClient *gst_multi_fd_sink_new_client (GstMultiHandleSink *
mhsink, GstMultiSinkHandle handle, GstSyncMethod sync_method);
static void gst_multi_fd_sink_client_free (GstMultiHandleSink * m,
GstMultiHandleClient * client);
static int gst_multi_fd_sink_client_get_fd (GstMultiHandleClient * client);
static void gst_multi_fd_sink_handle_debug (GstMultiSinkHandle handle,
gchar debug[30]);
static gpointer gst_multi_fd_sink_handle_hash_key (GstMultiSinkHandle handle);
static void gst_multi_fd_sink_hash_adding (GstMultiHandleSink * mhsink,
GstMultiHandleClient * mhclient);
static void gst_multi_fd_sink_hash_removing (GstMultiHandleSink * mhsink,
GstMultiHandleClient * mhclient);
static void gst_multi_fd_sink_hash_changed (GstMultiHandleSink * mhsink);
static void gst_multi_fd_sink_set_property (GObject * object, guint prop_id,
const GValue * value, GParamSpec * pspec);
static void gst_multi_fd_sink_get_property (GObject * object, guint prop_id,
GValue * value, GParamSpec * pspec);
#define gst_multi_fd_sink_parent_class parent_class
G_DEFINE_TYPE (GstMultiFdSink, gst_multi_fd_sink, GST_TYPE_MULTI_HANDLE_SINK);
static guint gst_multi_fd_sink_signals[LAST_SIGNAL] = { 0 };
static void
gst_multi_fd_sink_class_init (GstMultiFdSinkClass * klass)
{
GObjectClass *gobject_class;
GstElementClass *gstelement_class;
GstMultiHandleSinkClass *gstmultihandlesink_class;
gobject_class = (GObjectClass *) klass;
gstelement_class = (GstElementClass *) klass;
gstmultihandlesink_class = (GstMultiHandleSinkClass *) klass;
gobject_class->set_property = gst_multi_fd_sink_set_property;
gobject_class->get_property = gst_multi_fd_sink_get_property;
/**
* GstMultiFdSink::handle-read
*
* Handle read requests from clients and discard the data.
*
* Since: 0.10.23
*/
g_object_class_install_property (gobject_class, PROP_HANDLE_READ,
g_param_spec_boolean ("handle-read", "Handle Read",
"Handle client reads and discard the data",
DEFAULT_HANDLE_READ, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
/**
* GstMultiFdSink::add:
* @gstmultifdsink: the multifdsink element to emit this signal on
* @fd: the file descriptor to add to multifdsink
*
* Hand the given open file descriptor to multifdsink to write to.
*/
gst_multi_fd_sink_signals[SIGNAL_ADD] =
g_signal_new ("add", G_TYPE_FROM_CLASS (klass),
G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
G_STRUCT_OFFSET (GstMultiFdSinkClass, add), NULL, NULL,
g_cclosure_marshal_VOID__INT, G_TYPE_NONE, 1, G_TYPE_INT);
/**
* GstMultiFdSink::add-full:
* @gstmultifdsink: the multifdsink element to emit this signal on
* @fd: the file descriptor to add to multifdsink
* @sync: the sync method to use
* @format_min: the format of @value_min
* @value_min: the minimum amount of data to burst expressed in
* @format_min units.
* @format_max: the format of @value_max
* @value_max: the maximum amount of data to burst expressed in
* @format_max units.
*
* Hand the given open file descriptor to multifdsink to write to and
* specify the burst parameters for the new connection.
*/
gst_multi_fd_sink_signals[SIGNAL_ADD_BURST] =
g_signal_new ("add-full", G_TYPE_FROM_CLASS (klass),
G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
G_STRUCT_OFFSET (GstMultiFdSinkClass, add_full), NULL, NULL,
gst_tcp_marshal_VOID__INT_ENUM_INT_UINT64_INT_UINT64, G_TYPE_NONE, 6,
G_TYPE_INT, GST_TYPE_SYNC_METHOD, GST_TYPE_FORMAT, G_TYPE_UINT64,
GST_TYPE_FORMAT, G_TYPE_UINT64);
/**
* GstMultiFdSink::remove:
* @gstmultifdsink: the multifdsink element to emit this signal on
* @fd: the file descriptor to remove from multifdsink
*
* Remove the given open file descriptor from multifdsink.
*/
gst_multi_fd_sink_signals[SIGNAL_REMOVE] =
g_signal_new ("remove", G_TYPE_FROM_CLASS (klass),
G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
G_STRUCT_OFFSET (GstMultiFdSinkClass, remove), NULL, NULL,
gst_tcp_marshal_VOID__INT, G_TYPE_NONE, 1, G_TYPE_INT);
/**
* GstMultiFdSink::remove-flush:
* @gstmultifdsink: the multifdsink element to emit this signal on
* @fd: the file descriptor to remove from multifdsink
*
* Remove the given open file descriptor from multifdsink after flushing all
* the pending data to the fd.
*/
gst_multi_fd_sink_signals[SIGNAL_REMOVE_FLUSH] =
g_signal_new ("remove-flush", G_TYPE_FROM_CLASS (klass),
G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
G_STRUCT_OFFSET (GstMultiFdSinkClass, remove_flush), NULL, NULL,
gst_tcp_marshal_VOID__INT, G_TYPE_NONE, 1, G_TYPE_INT);
/**
* GstMultiFdSink::get-stats:
* @gstmultifdsink: the multifdsink element to emit this signal on
* @fd: the file descriptor to get stats of from multifdsink
*
* Get statistics about @fd. This function returns a GValueArray to ease
* automatic wrapping for bindings.
*
* Returns: a GValueArray with the statistics. The array contains guint64
* values that represent respectively: total number of bytes sent, time
* when the client was added, time when the client was
* disconnected/removed, time the client is/was active, last activity
* time (in epoch seconds), number of buffers dropped.
* All times are expressed in nanoseconds (GstClockTime).
* The array can be 0-length if the client was not found.
*/
gst_multi_fd_sink_signals[SIGNAL_GET_STATS] =
g_signal_new ("get-stats", G_TYPE_FROM_CLASS (klass),
G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
G_STRUCT_OFFSET (GstMultiFdSinkClass, get_stats), NULL, NULL,
gst_tcp_marshal_BOXED__INT, GST_TYPE_STRUCTURE, 1, G_TYPE_INT);
/**
* GstMultiFdSink::client-added:
* @gstmultifdsink: the multifdsink element that emitted this signal
* @fd: the file descriptor that was added to multifdsink
*
* The given file descriptor was added to multifdsink. This signal will
* be emitted from the streaming thread so application should be prepared
* for that.
*/
gst_multi_fd_sink_signals[SIGNAL_CLIENT_ADDED] =
g_signal_new ("client-added", G_TYPE_FROM_CLASS (klass),
G_SIGNAL_RUN_LAST, 0, NULL, NULL, gst_tcp_marshal_VOID__INT, G_TYPE_NONE,
1, G_TYPE_INT);
/**
* GstMultiFdSink::client-removed:
* @gstmultifdsink: the multifdsink element that emitted this signal
* @fd: the file descriptor that is to be removed from multifdsink
* @status: the reason why the client was removed
*
* The given file descriptor is about to be removed from multifdsink. This
* signal will be emitted from the streaming thread so applications should
* be prepared for that.
*
* @gstmultifdsink still holds a handle to @fd so it is possible to call
* the get-stats signal from this callback. For the same reason it is
* not safe to close() and reuse @fd in this callback.
*/
gst_multi_fd_sink_signals[SIGNAL_CLIENT_REMOVED] =
g_signal_new ("client-removed", G_TYPE_FROM_CLASS (klass),
G_SIGNAL_RUN_LAST, 0, NULL, NULL, gst_tcp_marshal_VOID__INT_ENUM,
G_TYPE_NONE, 2, G_TYPE_INT, GST_TYPE_CLIENT_STATUS);
/**
* GstMultiFdSink::client-fd-removed:
* @gstmultifdsink: the multifdsink element that emitted this signal
* @fd: the file descriptor that was removed from multifdsink
*
* The given file descriptor was removed from multifdsink. This signal will
* be emitted from the streaming thread so applications should be prepared
* for that.
*
* In this callback, @gstmultifdsink has removed all the information
* associated with @fd and it is therefore not possible to call get-stats
* with @fd. It is however safe to close() and reuse @fd in the callback.
*
* Since: 0.10.7
*/
/* FIXME: rename to client-fd-removed */
gst_multi_fd_sink_signals[SIGNAL_CLIENT_HANDLE_REMOVED] =
g_signal_new ("client-handle-removed", G_TYPE_FROM_CLASS (klass),
G_SIGNAL_RUN_LAST, 0, NULL, NULL, gst_tcp_marshal_VOID__INT,
G_TYPE_NONE, 1, G_TYPE_INT);
gst_element_class_set_static_metadata (gstelement_class,
"Multi filedescriptor sink", "Sink/Network",
"Send data to multiple filedescriptors",
"Thomas Vander Stichele <thomas at apestaart dot org>, "
"Wim Taymans <wim@fluendo.com>");
klass->add = GST_DEBUG_FUNCPTR (gst_multi_fd_sink_add);
klass->add_full = GST_DEBUG_FUNCPTR (gst_multi_fd_sink_add_full);
klass->remove = GST_DEBUG_FUNCPTR (gst_multi_fd_sink_remove);
klass->remove_flush = GST_DEBUG_FUNCPTR (gst_multi_fd_sink_remove_flush);
klass->get_stats = GST_DEBUG_FUNCPTR (gst_multi_fd_sink_get_stats);
gstmultihandlesink_class->emit_client_added =
gst_multi_fd_sink_emit_client_added;
gstmultihandlesink_class->emit_client_removed =
gst_multi_fd_sink_emit_client_removed;
gstmultihandlesink_class->stop_pre =
GST_DEBUG_FUNCPTR (gst_multi_fd_sink_stop_pre);
gstmultihandlesink_class->stop_post =
GST_DEBUG_FUNCPTR (gst_multi_fd_sink_stop_post);
gstmultihandlesink_class->start_pre =
GST_DEBUG_FUNCPTR (gst_multi_fd_sink_start_pre);
gstmultihandlesink_class->thread =
GST_DEBUG_FUNCPTR (gst_multi_fd_sink_thread);
gstmultihandlesink_class->new_client =
GST_DEBUG_FUNCPTR (gst_multi_fd_sink_new_client);
gstmultihandlesink_class->client_free = gst_multi_fd_sink_client_free;
gstmultihandlesink_class->client_get_fd =
GST_DEBUG_FUNCPTR (gst_multi_fd_sink_client_get_fd);
gstmultihandlesink_class->handle_debug =
GST_DEBUG_FUNCPTR (gst_multi_fd_sink_handle_debug);
gstmultihandlesink_class->handle_hash_key =
GST_DEBUG_FUNCPTR (gst_multi_fd_sink_handle_hash_key);
gstmultihandlesink_class->hash_changed =
GST_DEBUG_FUNCPTR (gst_multi_fd_sink_hash_changed);
gstmultihandlesink_class->hash_adding =
GST_DEBUG_FUNCPTR (gst_multi_fd_sink_hash_adding);
gstmultihandlesink_class->hash_removing =
GST_DEBUG_FUNCPTR (gst_multi_fd_sink_hash_removing);
GST_DEBUG_CATEGORY_INIT (multifdsink_debug, "multifdsink", 0, "FD sink");
}
static void
gst_multi_fd_sink_init (GstMultiFdSink * this)
{
GstMultiHandleSink *mhsink = GST_MULTI_HANDLE_SINK (this);
mhsink->handle_hash = g_hash_table_new (g_direct_hash, g_direct_equal);
this->handle_read = DEFAULT_HANDLE_READ;
}
/* methods to emit signals */
static void
gst_multi_fd_sink_emit_client_added (GstMultiHandleSink * mhsink,
GstMultiSinkHandle handle)
{
g_signal_emit (mhsink, gst_multi_fd_sink_signals[SIGNAL_CLIENT_ADDED], 0,
handle.fd);
}
static void
gst_multi_fd_sink_emit_client_removed (GstMultiHandleSink * mhsink,
GstMultiSinkHandle handle, GstClientStatus status)
{
g_signal_emit (mhsink, gst_multi_fd_sink_signals[SIGNAL_CLIENT_REMOVED], 0,
handle.fd, status);
}
static void
gst_multi_fd_sink_client_free (GstMultiHandleSink * mhsink,
GstMultiHandleClient * client)
{
g_signal_emit (mhsink,
gst_multi_fd_sink_signals[SIGNAL_CLIENT_HANDLE_REMOVED], 0,
client->handle.fd);
}
/* action signals */
static void
gst_multi_fd_sink_add (GstMultiFdSink * sink, int fd)
{
GstMultiSinkHandle handle;
handle.fd = fd;
gst_multi_handle_sink_add (GST_MULTI_HANDLE_SINK_CAST (sink), handle);
}
static void
gst_multi_fd_sink_add_full (GstMultiFdSink * sink, int fd,
GstSyncMethod sync, GstFormat min_format, guint64 min_value,
GstFormat max_format, guint64 max_value)
{
GstMultiSinkHandle handle;
handle.fd = fd;
gst_multi_handle_sink_add_full (GST_MULTI_HANDLE_SINK_CAST (sink), handle,
sync, min_format, min_value, max_format, max_value);
}
static void
gst_multi_fd_sink_remove (GstMultiFdSink * sink, int fd)
{
GstMultiSinkHandle handle;
handle.fd = fd;
gst_multi_handle_sink_remove (GST_MULTI_HANDLE_SINK_CAST (sink), handle);
}
static void
gst_multi_fd_sink_remove_flush (GstMultiFdSink * sink, int fd)
{
GstMultiSinkHandle handle;
handle.fd = fd;
gst_multi_handle_sink_remove_flush (GST_MULTI_HANDLE_SINK_CAST (sink),
handle);
}
static GstStructure *
gst_multi_fd_sink_get_stats (GstMultiFdSink * sink, int fd)
{
GstMultiSinkHandle handle;
handle.fd = fd;
return gst_multi_handle_sink_get_stats (GST_MULTI_HANDLE_SINK_CAST (sink),
handle);
}
/* vfuncs */
static GstMultiHandleClient *
gst_multi_fd_sink_new_client (GstMultiHandleSink * mhsink,
GstMultiSinkHandle handle, GstSyncMethod sync_method)
{
struct stat statbuf;
GstTCPClient *client;
GstMultiHandleClient *mhclient;
GstMultiFdSink *sink = GST_MULTI_FD_SINK (mhsink);
GstMultiHandleSinkClass *mhsinkclass =
GST_MULTI_HANDLE_SINK_GET_CLASS (mhsink);
/* create client datastructure */
client = g_new0 (GstTCPClient, 1);
mhclient = (GstMultiHandleClient *) client;
mhclient->handle = handle;
gst_poll_fd_init (&client->gfd);
client->gfd.fd = mhclient->handle.fd;
gst_multi_handle_sink_client_init (mhclient, sync_method);
mhsinkclass->handle_debug (handle, mhclient->debug);
/* set the socket to non blocking */
if (fcntl (handle.fd, F_SETFL, O_NONBLOCK) < 0) {
GST_ERROR_OBJECT (mhsink, "failed to make socket %s non-blocking: %s",
mhclient->debug, g_strerror (errno));
}
/* we always read from a client */
gst_poll_add_fd (sink->fdset, &client->gfd);
/* we don't try to read from write only fds */
if (sink->handle_read) {
gint flags;
flags = fcntl (handle.fd, F_GETFL, 0);
if ((flags & O_ACCMODE) != O_WRONLY) {
gst_poll_fd_ctl_read (sink->fdset, &client->gfd, TRUE);
}
}
/* figure out the mode, can't use send() for non sockets */
if (fstat (handle.fd, &statbuf) == 0 && S_ISSOCK (statbuf.st_mode)) {
client->is_socket = TRUE;
gst_multi_handle_sink_setup_dscp_client (mhsink, mhclient);
}
return mhclient;
}
static int
gst_multi_fd_sink_client_get_fd (GstMultiHandleClient * client)
{
GstTCPClient *tclient = (GstTCPClient *) client;
return tclient->gfd.fd;
}
static void
gst_multi_fd_sink_handle_debug (GstMultiSinkHandle handle, gchar debug[30])
{
g_snprintf (debug, 30, "[fd %5d]", handle.fd);
}
static gpointer
gst_multi_fd_sink_handle_hash_key (GstMultiSinkHandle handle)
{
return GINT_TO_POINTER (handle.fd);
}
static void
gst_multi_fd_sink_hash_changed (GstMultiHandleSink * mhsink)
{
GstMultiFdSink *sink = GST_MULTI_FD_SINK (mhsink);
gst_poll_restart (sink->fdset);
}
/* handle a read on a client fd,
* which either indicates a close or should be ignored
* returns FALSE if some error occured or the client closed. */
static gboolean
gst_multi_fd_sink_handle_client_read (GstMultiFdSink * sink,
GstTCPClient * client)
{
int avail, fd;
gboolean ret;
GstMultiHandleClient *mhclient = (GstMultiHandleClient *) client;
fd = client->gfd.fd;
if (ioctl (fd, FIONREAD, &avail) < 0)
goto ioctl_failed;
GST_DEBUG_OBJECT (sink, "%s select reports client read of %d bytes",
mhclient->debug, avail);
ret = TRUE;
if (avail == 0) {
/* client sent close, so remove it */
GST_DEBUG_OBJECT (sink, "%s client asked for close, removing",
mhclient->debug);
mhclient->status = GST_CLIENT_STATUS_CLOSED;
ret = FALSE;
} else if (avail < 0) {
GST_WARNING_OBJECT (sink, "%s avail < 0, removing", mhclient->debug);
mhclient->status = GST_CLIENT_STATUS_ERROR;
ret = FALSE;
} else {
guint8 dummy[512];
gint nread;
/* just Read 'n' Drop, could also just drop the client as it's not supposed
* to write to us except for closing the socket, I guess it's because we
* like to listen to our customers. */
do {
/* this is the maximum we can read */
gint to_read = MIN (avail, 512);
GST_DEBUG_OBJECT (sink, "%s client wants us to read %d bytes",
mhclient->debug, to_read);
nread = read (fd, dummy, to_read);
if (nread < -1) {
GST_WARNING_OBJECT (sink, "%s could not read %d bytes: %s (%d)",
mhclient->debug, to_read, g_strerror (errno), errno);
mhclient->status = GST_CLIENT_STATUS_ERROR;
ret = FALSE;
break;
} else if (nread == 0) {
GST_WARNING_OBJECT (sink, "%s 0 bytes in read, removing",
mhclient->debug);
mhclient->status = GST_CLIENT_STATUS_ERROR;
ret = FALSE;
break;
}
avail -= nread;
}
while (avail > 0);
}
return ret;
/* ERRORS */
ioctl_failed:
{
GST_WARNING_OBJECT (sink, "%s ioctl failed: %s (%d)",
mhclient->debug, g_strerror (errno), errno);
mhclient->status = GST_CLIENT_STATUS_ERROR;
return FALSE;
}
}
/* Handle a write on a client,
* which indicates a read request from a client.
*
* For each client we maintain a queue of GstBuffers that contain the raw bytes
* we need to send to the client.
*
* We first check to see if we need to send streamheaders. If so, we queue them.
*
* Then we run into the main loop that tries to send as many buffers as
* possible. It will first exhaust the mhclient->sending queue and if the queue
* is empty, it will pick a buffer from the global queue.
*
* Sending the buffers from the mhclient->sending queue is basically writing
* the bytes to the socket and maintaining a count of the bytes that were
* sent. When the buffer is completely sent, it is removed from the
* mhclient->sending queue and we try to pick a new buffer for sending.
*
* When the sending returns a partial buffer we stop sending more data as
* the next send operation could block.
*
* This functions returns FALSE if some error occured.
*/
static gboolean
gst_multi_fd_sink_handle_client_write (GstMultiFdSink * sink,
GstTCPClient * client)
{
gboolean more;
gboolean flushing;
GstClockTime now;
GTimeVal nowtv;
GstMultiHandleSink *mhsink = GST_MULTI_HANDLE_SINK (sink);
GstMultiHandleSinkClass *mhsinkclass =
GST_MULTI_HANDLE_SINK_GET_CLASS (mhsink);
GstMultiHandleClient *mhclient = (GstMultiHandleClient *) client;
int fd = mhclient->handle.fd;
g_get_current_time (&nowtv);
now = GST_TIMEVAL_TO_TIME (nowtv);
flushing = mhclient->status == GST_CLIENT_STATUS_FLUSHING;
more = TRUE;
do {
gint maxsize;
if (!mhclient->sending) {
/* client is not working on a buffer */
if (mhclient->bufpos == -1) {
/* client is too fast, remove from write queue until new buffer is
* available */
// FIXME: specific
gst_poll_fd_ctl_write (sink->fdset, &client->gfd, FALSE);
//
/* if we flushed out all of the client buffers, we can stop */
if (mhclient->flushcount == 0)
goto flushed;
return TRUE;
} else {
/* client can pick a buffer from the global queue */
GstBuffer *buf;
GstClockTime timestamp;
/* for new connections, we need to find a good spot in the
* bufqueue to start streaming from */
if (mhclient->new_connection && !flushing) {
gint position =
gst_multi_handle_sink_new_client_position (mhsink, mhclient);
if (position >= 0) {
/* we got a valid spot in the queue */
mhclient->new_connection = FALSE;
mhclient->bufpos = position;
} else {
/* cannot send data to this client yet */
// FIXME: specific
gst_poll_fd_ctl_write (sink->fdset, &client->gfd, FALSE);
return TRUE;
}
}
/* we flushed all remaining buffers, no need to get a new one */
if (mhclient->flushcount == 0)
goto flushed;
/* grab buffer */
buf = g_array_index (mhsink->bufqueue, GstBuffer *, mhclient->bufpos);
mhclient->bufpos--;
/* update stats */
timestamp = GST_BUFFER_TIMESTAMP (buf);
if (mhclient->first_buffer_ts == GST_CLOCK_TIME_NONE)
mhclient->first_buffer_ts = timestamp;
if (timestamp != -1)
mhclient->last_buffer_ts = timestamp;
/* decrease flushcount */
if (mhclient->flushcount != -1)
mhclient->flushcount--;
GST_LOG_OBJECT (sink, "%s client %p at position %d",
mhclient->debug, client, mhclient->bufpos);
/* queueing a buffer will ref it */
mhsinkclass->client_queue_buffer (mhsink, mhclient, buf);
/* need to start from the first byte for this new buffer */
mhclient->bufoffset = 0;
}
}
/* see if we need to send something */
if (mhclient->sending) {
ssize_t wrote;
GstBuffer *head;
GstMapInfo info;
guint8 *data;
/* pick first buffer from list */
head = GST_BUFFER (mhclient->sending->data);
if (!gst_buffer_map (head, &info, GST_MAP_READ))
g_return_val_if_reached (FALSE);
data = info.data;
maxsize = info.size - mhclient->bufoffset;
// FIXME: specific
/* try to write the complete buffer */
#ifdef MSG_NOSIGNAL
#define FLAGS MSG_NOSIGNAL
#else
#define FLAGS 0
#endif
if (client->is_socket) {
wrote = send (fd, data + mhclient->bufoffset, maxsize, FLAGS);
} else {
wrote = write (fd, data + mhclient->bufoffset, maxsize);
}
gst_buffer_unmap (head, &info);
if (wrote < 0) {
/* hmm error.. */
if (errno == EAGAIN) {
/* nothing serious, resource was unavailable, try again later */
more = FALSE;
} else if (errno == ECONNRESET) {
goto connection_reset;
} else {
goto write_error;
}
} else {
if (wrote < maxsize) {
/* partial write means that the client cannot read more and we should
* stop sending more */
GST_LOG_OBJECT (sink,
"partial write on %s of %" G_GSSIZE_FORMAT " bytes",
mhclient->debug, wrote);
mhclient->bufoffset += wrote;
more = FALSE;
} else {
/* complete buffer was written, we can proceed to the next one */
mhclient->sending = g_slist_remove (mhclient->sending, head);
gst_buffer_unref (head);
/* make sure we start from byte 0 for the next buffer */
mhclient->bufoffset = 0;
}
/* update stats */
mhclient->bytes_sent += wrote;
mhclient->last_activity_time = now;
mhsink->bytes_served += wrote;
}
}
} while (more);
return TRUE;
/* ERRORS */
flushed:
{
GST_DEBUG_OBJECT (sink, "%s flushed, removing", mhclient->debug);
mhclient->status = GST_CLIENT_STATUS_REMOVED;
return FALSE;
}
connection_reset:
{
GST_DEBUG_OBJECT (sink, "%s connection reset by peer, removing",
mhclient->debug);
mhclient->status = GST_CLIENT_STATUS_CLOSED;
return FALSE;
}
write_error:
{
GST_WARNING_OBJECT (sink,
"%s could not write, removing client: %s (%d)", mhclient->debug,
g_strerror (errno), errno);
mhclient->status = GST_CLIENT_STATUS_ERROR;
return FALSE;
}
}
static void
gst_multi_fd_sink_hash_adding (GstMultiHandleSink * mhsink,
GstMultiHandleClient * mhclient)
{
GstMultiFdSink *sink = GST_MULTI_FD_SINK (mhsink);
GstTCPClient *client = (GstTCPClient *) mhclient;
gst_poll_fd_ctl_write (sink->fdset, &client->gfd, TRUE);
}
static void
gst_multi_fd_sink_hash_removing (GstMultiHandleSink * mhsink,
GstMultiHandleClient * mhclient)
{
GstMultiFdSink *sink = GST_MULTI_FD_SINK (mhsink);
GstTCPClient *client = (GstTCPClient *) mhclient;
gst_poll_remove_fd (sink->fdset, &client->gfd);
}
/* Handle the clients. Basically does a blocking select for one
* of the client fds to become read or writable. We also have a
* filedescriptor to receive commands on that we need to check.
*
* After going out of the select call, we read and write to all
* clients that can do so. Badly behaving clients are put on a
* garbage list and removed.
*/
static void
gst_multi_fd_sink_handle_clients (GstMultiFdSink * sink)
{
int result;
GList *clients, *next;
gboolean try_again;
GstMultiFdSinkClass *fclass;
guint cookie;
GstMultiHandleSink *mhsink = GST_MULTI_HANDLE_SINK (sink);
int fd;
fclass = GST_MULTI_FD_SINK_GET_CLASS (sink);
do {
try_again = FALSE;
/* check for:
* - server socket input (ie, new client connections)
* - client socket input (ie, clients saying goodbye)
* - client socket output (ie, client reads) */
GST_LOG_OBJECT (sink, "waiting on action on fdset");
result =
gst_poll_wait (sink->fdset,
mhsink->timeout != 0 ? mhsink->timeout : GST_CLOCK_TIME_NONE);
/* Handle the special case in which the sink is not receiving more buffers
* and will not disconnect inactive client in the streaming thread. */
if (G_UNLIKELY (result == 0)) {
GstClockTime now;
GTimeVal nowtv;
g_get_current_time (&nowtv);
now = GST_TIMEVAL_TO_TIME (nowtv);
CLIENTS_LOCK (mhsink);
for (clients = mhsink->clients; clients; clients = next) {
GstTCPClient *client;
GstMultiHandleClient *mhclient;
client = (GstTCPClient *) clients->data;
mhclient = (GstMultiHandleClient *) client;
next = g_list_next (clients);
if (mhsink->timeout > 0
&& now - mhclient->last_activity_time > mhsink->timeout) {
mhclient->status = GST_CLIENT_STATUS_SLOW;
gst_multi_handle_sink_remove_client_link (mhsink, clients);
}
}
CLIENTS_UNLOCK (mhsink);
return;
} else if (result < 0) {
GST_WARNING_OBJECT (sink, "wait failed: %s (%d)", g_strerror (errno),
errno);
if (errno == EBADF) {
/* ok, so one or more of the fds is invalid. We loop over them to find
* the ones that give an error to the F_GETFL fcntl. */
CLIENTS_LOCK (mhsink);
restart:
cookie = mhsink->clients_cookie;
for (clients = mhsink->clients; clients; clients = next) {
GstTCPClient *client;
GstMultiHandleClient *mhclient;
long flags;
int res;
if (cookie != mhsink->clients_cookie) {
GST_DEBUG_OBJECT (sink, "Cookie changed finding bad fd");
goto restart;
}
client = (GstTCPClient *) clients->data;
mhclient = (GstMultiHandleClient *) client;
next = g_list_next (clients);
fd = client->gfd.fd;
res = fcntl (fd, F_GETFL, &flags);
if (res == -1) {
GST_WARNING_OBJECT (sink, "fnctl failed for %d, removing: %s (%d)",
fd, g_strerror (errno), errno);
if (errno == EBADF) {
mhclient->status = GST_CLIENT_STATUS_ERROR;
/* releases the CLIENTS lock */
gst_multi_handle_sink_remove_client_link (mhsink, clients);
}
}
}
CLIENTS_UNLOCK (mhsink);
/* after this, go back in the select loop as the read/writefds
* are not valid */
try_again = TRUE;
} else if (errno == EINTR) {
/* interrupted system call, just redo the wait */
try_again = TRUE;
} else if (errno == EBUSY) {
/* the call to gst_poll_wait() was flushed */
return;
} else {
/* this is quite bad... */
GST_ELEMENT_ERROR (sink, RESOURCE, READ, (NULL),
("select failed: %s (%d)", g_strerror (errno), errno));
return;
}
} else {
GST_LOG_OBJECT (sink, "wait done: %d sockets with events", result);
}
} while (try_again);
/* subclasses can check fdset with this virtual function */
if (fclass->wait)
fclass->wait (sink, sink->fdset);
/* Check the clients */
CLIENTS_LOCK (mhsink);
restart2:
cookie = mhsink->clients_cookie;
for (clients = mhsink->clients; clients; clients = next) {
GstTCPClient *client;
GstMultiHandleClient *mhclient;
if (mhsink->clients_cookie != cookie) {
GST_DEBUG_OBJECT (sink, "Restarting loop, cookie out of date");
goto restart2;
}
client = (GstTCPClient *) clients->data;
mhclient = (GstMultiHandleClient *) client;
next = g_list_next (clients);
if (mhclient->status != GST_CLIENT_STATUS_FLUSHING
&& mhclient->status != GST_CLIENT_STATUS_OK) {
gst_multi_handle_sink_remove_client_link (mhsink, clients);
continue;
}
if (gst_poll_fd_has_closed (sink->fdset, &client->gfd)) {
mhclient->status = GST_CLIENT_STATUS_CLOSED;
gst_multi_handle_sink_remove_client_link (mhsink, clients);
continue;
}
if (gst_poll_fd_has_error (sink->fdset, &client->gfd)) {
GST_WARNING_OBJECT (sink, "gst_poll_fd_has_error for %d", client->gfd.fd);
mhclient->status = GST_CLIENT_STATUS_ERROR;
gst_multi_handle_sink_remove_client_link (mhsink, clients);
continue;
}
if (gst_poll_fd_can_read (sink->fdset, &client->gfd)) {
/* handle client read */
if (!gst_multi_fd_sink_handle_client_read (sink, client)) {
gst_multi_handle_sink_remove_client_link (mhsink, clients);
continue;
}
}
if (gst_poll_fd_can_write (sink->fdset, &client->gfd)) {
/* handle client write */
if (!gst_multi_fd_sink_handle_client_write (sink, client)) {
gst_multi_handle_sink_remove_client_link (mhsink, clients);
continue;
}
}
}
CLIENTS_UNLOCK (mhsink);
}
/* we handle the client communication in another thread so that we do not block
* the gstreamer thread while we select() on the client fds */
static gpointer
gst_multi_fd_sink_thread (GstMultiHandleSink * mhsink)
{
GstMultiFdSink *sink = GST_MULTI_FD_SINK (mhsink);
while (mhsink->running) {
gst_multi_fd_sink_handle_clients (sink);
}
return NULL;
}
static void
gst_multi_fd_sink_set_property (GObject * object, guint prop_id,
const GValue * value, GParamSpec * pspec)
{
GstMultiFdSink *multifdsink;
multifdsink = GST_MULTI_FD_SINK (object);
switch (prop_id) {
case PROP_HANDLE_READ:
multifdsink->handle_read = g_value_get_boolean (value);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
}
}
static void
gst_multi_fd_sink_get_property (GObject * object, guint prop_id, GValue * value,
GParamSpec * pspec)
{
GstMultiFdSink *multifdsink;
multifdsink = GST_MULTI_FD_SINK (object);
switch (prop_id) {
case PROP_HANDLE_READ:
g_value_set_boolean (value, multifdsink->handle_read);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
}
}
static gboolean
gst_multi_fd_sink_start_pre (GstMultiHandleSink * mhsink)
{
GstMultiFdSink *mfsink = GST_MULTI_FD_SINK (mhsink);
GST_INFO_OBJECT (mfsink, "starting");
if ((mfsink->fdset = gst_poll_new (TRUE)) == NULL)
goto socket_pair;
return TRUE;
/* ERRORS */
socket_pair:
{
GST_ELEMENT_ERROR (mfsink, RESOURCE, OPEN_READ_WRITE, (NULL),
GST_ERROR_SYSTEM);
return FALSE;
}
}
static gboolean
multifdsink_hash_remove (gpointer key, gpointer value, gpointer data)
{
return TRUE;
}
static void
gst_multi_fd_sink_stop_pre (GstMultiHandleSink * mhsink)
{
GstMultiFdSink *mfsink = GST_MULTI_FD_SINK (mhsink);
gst_poll_set_flushing (mfsink->fdset, TRUE);
}
static void
gst_multi_fd_sink_stop_post (GstMultiHandleSink * mhsink)
{
GstMultiFdSink *mfsink = GST_MULTI_FD_SINK (mhsink);
if (mfsink->fdset) {
gst_poll_free (mfsink->fdset);
mfsink->fdset = NULL;
}
g_hash_table_foreach_remove (mhsink->handle_hash, multifdsink_hash_remove,
mfsink);
}
Reference in a new issue View git blame Copy permalink