sys/shm/gstshmsrc.c

   1 /* GStreamer
   2  * Copyright (C) <2009> Collabora Ltd
   3  *  @author: Olivier Crete <olivier.crete@collabora.co.uk
   4  * Copyright (C) <2009> Nokia Inc
   5  *
   6  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Library General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2 of the License, or (at your option) any later version.
  10  *
  11  * This library is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Library General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Library General Public
  17  * License along with this library; if not, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19  * Boston, MA 02111-1307, USA.
  20  */
  21
  22 #ifdef HAVE_CONFIG_H
  23 #include "config.h"
  24 #endif
  25
  26 #include "gstshmsrc.h"
  27
  28 #include <gst/gst.h>
  29
  30 #include <string.h>
  31
  32 /* signals */
  33 enum
  34 {
  35   LAST_SIGNAL
  36 };
  37
  38 /* properties */
  39 enum
  40 {
  41   PROP_0,
  42   PROP_SOCKET_PATH,
  43   PROP_IS_LIVE
  44 };
  45
  46 struct GstShmBuffer
  47 {
  48   char *buf;
  49   GstShmPipe *pipe;
  50 };
  51
  52
  53 GST_DEBUG_CATEGORY_STATIC (shmsrc_debug);
  54 #define GST_CAT_DEFAULT shmsrc_debug
  55
  56 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
  57     GST_PAD_SRC,
  58     GST_PAD_ALWAYS,
  59     GST_STATIC_CAPS_ANY);
  60
  61 GST_BOILERPLATE (GstShmSrc, gst_shm_src, GstPushSrc, GST_TYPE_PUSH_SRC);
  62
  63 static void gst_shm_src_set_property (GObject * object, guint prop_id,
  64     const GValue * value, GParamSpec * pspec);
  65 static void gst_shm_src_get_property (GObject * object, guint prop_id,
  66     GValue * value, GParamSpec * pspec);
  67 static void gst_shm_src_finalize (GObject * object);
  68 static gboolean gst_shm_src_start (GstBaseSrc * bsrc);
  69 static gboolean gst_shm_src_stop (GstBaseSrc * bsrc);
  70 static GstFlowReturn gst_shm_src_create (GstPushSrc * psrc,
  71     GstBuffer ** outbuf);
  72 static gboolean gst_shm_src_unlock (GstBaseSrc * bsrc);
  73 static gboolean gst_shm_src_unlock_stop (GstBaseSrc * bsrc);
  74 static GstStateChangeReturn gst_shm_src_change_state (GstElement * element,
  75     GstStateChange transition);
  76
  77 static void gst_shm_pipe_inc (GstShmPipe * pipe);
  78 static void gst_shm_pipe_dec (GstShmPipe * pipe);
  79
  80 // static guint gst_shm_src_signals[LAST_SIGNAL] = { 0 };
  81
  82
  83 static void
  84 gst_shm_src_base_init (gpointer g_class)
  85 {
  86   GstElementClass *element_class = GST_ELEMENT_CLASS (g_class);
  87
  88   gst_element_class_add_static_pad_template (element_class, &srctemplate);
  89
  90   gst_element_class_set_details_simple (element_class,
  91       "Shared Memory Source",
  92       "Source",
  93       "Receive data from the sharem memory sink",
  94       "Olivier Crete <olivier.crete@collabora.co.uk>");
  95 }
  96
  97 static void
  98 gst_shm_src_class_init (GstShmSrcClass * klass)
  99 {
 100   GObjectClass *gobject_class;
 101   GstElementClass *gstelement_class;
 102   GstBaseSrcClass *gstbasesrc_class;
 103   GstPushSrcClass *gstpush_src_class;
 104
 105   gobject_class = (GObjectClass *) klass;
 106   gstelement_class = (GstElementClass *) klass;
 107   gstbasesrc_class = (GstBaseSrcClass *) klass;
 108   gstpush_src_class = (GstPushSrcClass *) klass;
 109
 110   gobject_class->set_property = gst_shm_src_set_property;
 111   gobject_class->get_property = gst_shm_src_get_property;
 112   gobject_class->finalize = gst_shm_src_finalize;
 113
 114   gstelement_class->change_state = gst_shm_src_change_state;
 115
 116   gstbasesrc_class->start = GST_DEBUG_FUNCPTR (gst_shm_src_start);
 117   gstbasesrc_class->stop = GST_DEBUG_FUNCPTR (gst_shm_src_stop);
 118   gstbasesrc_class->unlock = GST_DEBUG_FUNCPTR (gst_shm_src_unlock);
 119   gstbasesrc_class->unlock_stop = GST_DEBUG_FUNCPTR (gst_shm_src_unlock_stop);
 120
 121   gstpush_src_class->create = gst_shm_src_create;
 122
 123   g_object_class_install_property (gobject_class, PROP_SOCKET_PATH,
 124       g_param_spec_string ("socket-path",
 125           "Path to the control socket",
 126           "The path to the control socket used to control the shared memory"
 127           " transport", NULL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 128
 129   g_object_class_install_property (gobject_class, PROP_IS_LIVE,
 130       g_param_spec_boolean ("is-live", "Is this a live source",
 131           "True if the element cannot produce data in PAUSED", FALSE,
 132           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 133
 134   GST_DEBUG_CATEGORY_INIT (shmsrc_debug, "shmsrc", 0, "Shared Memory Source");
 135 }
 136
 137 static void
 138 gst_shm_src_init (GstShmSrc * self, GstShmSrcClass * g_class)
 139 {
 140   self->poll = gst_poll_new (TRUE);
 141   gst_poll_fd_init (&self->pollfd);
 142 }
 143
 144 static void
 145 gst_shm_src_finalize (GObject * object)
 146 {
 147   GstShmSrc *self = GST_SHM_SRC (object);
 148
 149   gst_poll_free (self->poll);
 150
 151   G_OBJECT_CLASS (parent_class)->finalize (object);
 152 }
 153
 154
 155 static void
 156 gst_shm_src_set_property (GObject * object, guint prop_id,
 157     const GValue * value, GParamSpec * pspec)
 158 {
 159   GstShmSrc *self = GST_SHM_SRC (object);
 160
 161   switch (prop_id) {
 162     case PROP_SOCKET_PATH:
 163       GST_OBJECT_LOCK (object);
 164       if (self->pipe) {
 165         GST_WARNING_OBJECT (object, "Can not modify socket path while the "
 166             "element is playing");
 167       } else {
 168         g_free (self->socket_path);
 169         self->socket_path = g_value_dup_string (value);
 170       }
 171       GST_OBJECT_UNLOCK (object);
 172       break;
 173     case PROP_IS_LIVE:
 174       gst_base_src_set_live (GST_BASE_SRC (object),
 175           g_value_get_boolean (value));
 176       break;
 177     default:
 178       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 179       break;
 180   }
 181 }
 182
 183 static void
 184 gst_shm_src_get_property (GObject * object, guint prop_id,
 185     GValue * value, GParamSpec * pspec)
 186 {
 187   GstShmSrc *self = GST_SHM_SRC (object);
 188
 189   switch (prop_id) {
 190     case PROP_SOCKET_PATH:
 191       GST_OBJECT_LOCK (object);
 192       g_value_set_string (value, self->socket_path);
 193       GST_OBJECT_UNLOCK (object);
 194       break;
 195     case PROP_IS_LIVE:
 196       g_value_set_boolean (value, gst_base_src_is_live (GST_BASE_SRC (object)));
 197       break;
 198     default:
 199       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 200       break;
 201   }
 202 }
 203
 204 static gboolean
 205 gst_shm_src_start_reading (GstShmSrc * self)
 206 {
 207   GstShmPipe *gstpipe = g_slice_new0 (GstShmPipe);
 208
 209   gstpipe->use_count = 1;
 210   gstpipe->src = gst_object_ref (self);
 211
 212   if (!self->socket_path) {
 213     GST_ELEMENT_ERROR (self, RESOURCE, NOT_FOUND,
 214         ("No path specified for socket."), (NULL));
 215     return FALSE;
 216   }
 217
 218   GST_DEBUG_OBJECT (self, "Opening socket %s", self->socket_path);
 219
 220   GST_OBJECT_LOCK (self);
 221   gstpipe->pipe = sp_client_open (self->socket_path);
 222   GST_OBJECT_UNLOCK (self);
 223
 224   if (!gstpipe->pipe) {
 225     GST_ELEMENT_ERROR (self, RESOURCE, OPEN_READ_WRITE,
 226         ("Could not open socket %s: %d %s", self->socket_path, errno,
 227             strerror (errno)), (NULL));
 228     gst_shm_pipe_dec (gstpipe);
 229     return FALSE;
 230   }
 231
 232   self->pipe = gstpipe;
 233
 234   gst_poll_set_flushing (self->poll, FALSE);
 235
 236   gst_poll_fd_init (&self->pollfd);
 237   self->pollfd.fd = sp_get_fd (self->pipe->pipe);
 238   gst_poll_add_fd (self->poll, &self->pollfd);
 239   gst_poll_fd_ctl_read (self->poll, &self->pollfd, TRUE);
 240
 241   return TRUE;
 242 }
 243
 244 static void
 245 gst_shm_src_stop_reading (GstShmSrc * self)
 246 {
 247   GST_DEBUG_OBJECT (self, "Stopping %p", self);
 248
 249   if (self->pipe) {
 250     gst_shm_pipe_dec (self->pipe);
 251     self->pipe = NULL;
 252   }
 253
 254   gst_poll_remove_fd (self->poll, &self->pollfd);
 255   gst_poll_fd_init (&self->pollfd);
 256
 257   gst_poll_set_flushing (self->poll, TRUE);
 258 }
 259
 260 static gboolean
 261 gst_shm_src_start (GstBaseSrc * bsrc)
 262 {
 263   if (gst_base_src_is_live (bsrc))
 264     return TRUE;
 265   else
 266     return gst_shm_src_start_reading (GST_SHM_SRC (bsrc));
 267 }
 268
 269 static gboolean
 270 gst_shm_src_stop (GstBaseSrc * bsrc)
 271 {
 272   if (!gst_base_src_is_live (bsrc))
 273     gst_shm_src_stop_reading (GST_SHM_SRC (bsrc));
 274
 275   return TRUE;
 276 }
 277
 278
 279 static void
 280 free_buffer (gpointer data)
 281 {
 282   struct GstShmBuffer *gsb = data;
 283   g_return_if_fail (gsb->pipe != NULL);
 284   g_return_if_fail (gsb->pipe->src != NULL);
 285
 286   GST_LOG ("Freeing buffer %p", gsb->buf);
 287
 288   GST_OBJECT_LOCK (gsb->pipe->src);
 289   sp_client_recv_finish (gsb->pipe->pipe, gsb->buf);
 290   GST_OBJECT_UNLOCK (gsb->pipe->src);
 291
 292   gst_shm_pipe_dec (gsb->pipe);
 293
 294   g_slice_free (struct GstShmBuffer, gsb);
 295 }
 296
 297 static GstFlowReturn
 298 gst_shm_src_create (GstPushSrc * psrc, GstBuffer ** outbuf)
 299 {
 300   GstShmSrc *self = GST_SHM_SRC (psrc);
 301   gchar *buf = NULL;
 302   int rv = 0;
 303   struct GstShmBuffer *gsb;
 304
 305   do {
 306     if (gst_poll_wait (self->poll, GST_CLOCK_TIME_NONE) < 0) {
 307       if (errno == EBUSY)
 308         return GST_FLOW_WRONG_STATE;
 309       GST_ELEMENT_ERROR (self, RESOURCE, READ, ("Failed to read from shmsrc"),
 310           ("Poll failed on fd: %s", strerror (errno)));
 311       return GST_FLOW_ERROR;
 312     }
 313
 314     if (self->unlocked)
 315       return GST_FLOW_WRONG_STATE;
 316
 317     if (gst_poll_fd_has_closed (self->poll, &self->pollfd)) {
 318       GST_ELEMENT_ERROR (self, RESOURCE, READ, ("Failed to read from shmsrc"),
 319           ("Control socket has closed"));
 320       return GST_FLOW_ERROR;
 321     }
 322
 323     if (gst_poll_fd_has_error (self->poll, &self->pollfd)) {
 324       GST_ELEMENT_ERROR (self, RESOURCE, READ, ("Failed to read from shmsrc"),
 325           ("Control socket has error"));
 326       return GST_FLOW_ERROR;
 327     }
 328
 329     if (gst_poll_fd_can_read (self->poll, &self->pollfd)) {
 330       buf = NULL;
 331       GST_LOG_OBJECT (self, "Reading from pipe");
 332       GST_OBJECT_LOCK (self);
 333       rv = sp_client_recv (self->pipe->pipe, &buf);
 334       GST_OBJECT_UNLOCK (self);
 335       if (rv < 0) {
 336         GST_ELEMENT_ERROR (self, RESOURCE, READ, ("Failed to read from shmsrc"),
 337             ("Error reading control data: %d", rv));
 338         return GST_FLOW_ERROR;
 339       }
 340     }
 341   } while (buf == NULL);
 342
 343   GST_LOG_OBJECT (self, "Got buffer %p of size %d", buf, rv);
 344
 345   gsb = g_slice_new0 (struct GstShmBuffer);
 346   gsb->buf = buf;
 347   gsb->pipe = self->pipe;
 348   gst_shm_pipe_inc (self->pipe);
 349
 350   *outbuf = gst_buffer_new ();
 351   GST_BUFFER_FLAG_SET (*outbuf, GST_BUFFER_FLAG_READONLY);
 352   GST_BUFFER_DATA (*outbuf) = (guint8 *) buf;
 353   GST_BUFFER_SIZE (*outbuf) = rv;
 354   GST_BUFFER_MALLOCDATA (*outbuf) = (guint8 *) gsb;
 355   GST_BUFFER_FREE_FUNC (*outbuf) = free_buffer;
 356
 357   return GST_FLOW_OK;
 358 }
 359
 360 static GstStateChangeReturn
 361 gst_shm_src_change_state (GstElement * element, GstStateChange transition)
 362 {
 363   GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
 364   GstShmSrc *self = GST_SHM_SRC (element);
 365
 366   switch (transition) {
 367     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
 368       if (gst_base_src_is_live (GST_BASE_SRC (element)))
 369         if (!gst_shm_src_start_reading (self))
 370           return GST_STATE_CHANGE_FAILURE;
 371     default:
 372       break;
 373   }
 374
 375   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
 376   if (ret == GST_STATE_CHANGE_FAILURE)
 377     return ret;
 378
 379   switch (transition) {
 380     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
 381       if (gst_base_src_is_live (GST_BASE_SRC (element)))
 382         gst_shm_src_stop_reading (self);
 383     default:
 384       break;
 385   }
 386
 387   return ret;
 388 }
 389
 390 static gboolean
 391 gst_shm_src_unlock (GstBaseSrc * bsrc)
 392 {
 393   GstShmSrc *self = GST_SHM_SRC (bsrc);
 394
 395   self->unlocked = TRUE;
 396   gst_poll_set_flushing (self->poll, TRUE);
 397
 398   return TRUE;
 399 }
 400
 401 static gboolean
 402 gst_shm_src_unlock_stop (GstBaseSrc * bsrc)
 403 {
 404   GstShmSrc *self = GST_SHM_SRC (bsrc);
 405
 406   self->unlocked = FALSE;
 407   gst_poll_set_flushing (self->poll, FALSE);
 408
 409   return TRUE;
 410 }
 411
 412 static void
 413 gst_shm_pipe_inc (GstShmPipe * pipe)
 414 {
 415   g_return_if_fail (pipe);
 416   g_return_if_fail (pipe->src);
 417   g_return_if_fail (pipe->use_count > 0);
 418
 419   GST_OBJECT_LOCK (pipe->src);
 420   pipe->use_count++;
 421   GST_OBJECT_UNLOCK (pipe->src);
 422 }
 423
 424 static void
 425 gst_shm_pipe_dec (GstShmPipe * pipe)
 426 {
 427   g_return_if_fail (pipe);
 428   g_return_if_fail (pipe->src);
 429   g_return_if_fail (pipe->use_count > 0);
 430
 431   GST_OBJECT_LOCK (pipe->src);
 432   pipe->use_count--;
 433
 434   if (pipe->use_count > 0) {
 435     GST_OBJECT_UNLOCK (pipe->src);
 436     return;
 437   }
 438
 439   if (pipe->pipe)
 440     sp_close (pipe->pipe);
 441   GST_OBJECT_UNLOCK (pipe->src);
 442
 443   gst_object_unref (pipe->src);
 444   g_slice_free (GstShmPipe, pipe);
 445 }