Some final fixes for the v4lsrc elements. remove software sync thread (use GST_ELEMENT_THREAD_SUGGESTED instead) make...

Original commit message from CVS:
Some final fixes for the v4lsrc elements.

* remove software sync thread (use GST_ELEMENT_THREAD_SUGGESTED instead)
* make all src elements threadsafe
* fix num_buffer argument setting in v4l2src (VIDIOC_S_PARM)
* re-add bufsize (RO) for v4lmjpegsrc
* fix the A/V sync calculation in all elements (spvf=GST_SECOND/fps, not GST_SECOND*fps)
* probably some more crap....

With all this, it actually works quite well. The TODO files describes the
next steps in order to make a full-featured video recorder based on these
elements and GStreamer (bottom). Making a simple recorder should be fairly
easy now, btw.

sys/v4l2/gstv4l2src.c

@@ -195,6 +195,8 @@ gst_v4l2src_class_init (GstV4l2SrcClass *klass)
 static void
 gst_v4l2src_init (GstV4l2Src *v4l2src)
 {
+	GST_FLAG_SET(GST_ELEMENT(v4l2src), GST_ELEMENT_THREAD_SUGGESTED);
+
 	v4l2src->srcpad = gst_pad_new_from_template(src_template, "src");
 	gst_element_add_pad(GST_ELEMENT(v4l2src), v4l2src->srcpad);
 
@@ -257,14 +259,14 @@ gst_v4l2src_get_fps (GstV4l2Src *v4l2src)
 	/* if that failed ... */
  
 	if (!GST_V4L2_IS_OPEN(GST_V4L2ELEMENT(v4l2src)))
-		return FALSE;
+		return 0.;
 
 	if (!gst_v4l2_get_norm(GST_V4L2ELEMENT(v4l2src), &norm))
-		return FALSE;
+		return 0.;
 
 	std = ((struct v4l2_standard *) g_list_nth_data(GST_V4L2ELEMENT(v4l2src)->norms, norm));
-	fps = std->frameperiod.numerator / std->frameperiod.denominator;
+	fps = (1. * std->frameperiod.denominator) / std->frameperiod.numerator;
- 
+
 	return fps;
 }
 
@@ -643,7 +645,7 @@ gst_v4l2src_srcconnect (GstPad  *pad,
 				gst_caps_get_int(caps, "width", &w);
 			} else {
 				int max;
-				gst_caps_get_int_range(caps, "width", &w, &max);
+				gst_caps_get_int_range(caps, "width", &max, &w);
 			}
 		}
 		if (gst_caps_has_property(caps, "height")) {
@@ -651,7 +653,7 @@ gst_v4l2src_srcconnect (GstPad  *pad,
 				gst_caps_get_int(caps, "height", &h);
 			} else {
 				int max;
-				gst_caps_get_int_range(caps, "height", &h, &max);
+				gst_caps_get_int_range(caps, "height", &max, &h);
 			}
 		}
 
@@ -790,14 +792,14 @@ gst_v4l2src_get (GstPad *pad)
 			 * timeframe. This means that if time - begin_time = X sec,
 			 * we want to have written X*fps frames. If we've written
 			 * more - drop, if we've written less - dup... */
-			if (v4l2src->handled * fps * GST_SECOND - time >
+			if (v4l2src->handled * (GST_SECOND/fps) - time >
-			    1.5 * fps * GST_SECOND) {
+			    1.5 * (GST_SECOND/fps)) {
 				/* yo dude, we've got too many frames here! Drop! DROP! */
 				v4l2src->need_writes--; /* -= (v4l2src->handled - (time / fps)); */
 				g_signal_emit(G_OBJECT(v4l2src),
 				              gst_v4l2src_signals[SIGNAL_FRAME_DROP], 0);
-			} else if (v4l2src->handled * fps * GST_SECOND - time <
+			} else if (v4l2src->handled * (GST_SECOND/fps) - time <
-			             -1.5 * fps * GST_SECOND) {
+			             -1.5 * (GST_SECOND/fps)) {
 				/* this means we're lagging far behind */
 				v4l2src->need_writes++; /* += ((time / fps) - v4l2src->handled); */
 				g_signal_emit(G_OBJECT(v4l2src),
@@ -820,7 +822,7 @@ gst_v4l2src_get (GstPad *pad)
 		v4l2src->use_num_times[num] = 1;
 	}
 
-	GST_BUFFER_DATA(buf) = GST_V4L2ELEMENT(v4l2src)->buffer[num];
+	GST_BUFFER_DATA(buf) = gst_v4l2src_get_buffer(v4l2src, num);
 	GST_BUFFER_SIZE(buf) = v4l2src->bufsettings.bytesused;
 	if (v4l2src->use_fixed_fps)
 		GST_BUFFER_TIMESTAMP(buf) = v4l2src->handled * GST_SECOND / fps;
@@ -831,6 +833,7 @@ gst_v4l2src_get (GstPad *pad)
 	v4l2src->handled++;
 	g_signal_emit(G_OBJECT(v4l2src),
 		      gst_v4l2src_signals[SIGNAL_FRAME_CAPTURE], 0);
+
 	return buf;
 }
 
@@ -1001,7 +1004,7 @@ gst_v4l2src_buffer_free (GstBufferPool *pool,
 		return; /* we've already cleaned up ourselves */
 
 	for (n=0;n<v4l2src->breq.count;n++)
-		if (GST_BUFFER_DATA(buf) == GST_V4L2ELEMENT(v4l2src)->buffer[n]) {
+		if (GST_BUFFER_DATA(buf) == gst_v4l2src_get_buffer(v4l2src, n)) {
 			v4l2src->use_num_times[n]--;
 			if (v4l2src->use_num_times[n] <= 0) {
 				gst_v4l2src_requeue_frame(v4l2src, n);

sys/v4l2/gstv4l2src.h

@@ -52,6 +52,17 @@ struct _GstV4l2Src {
 	struct v4l2_requestbuffers breq;
 	struct v4l2_format format;
 
+	/* num of queued frames and some GThread stuff
+	 * to wait if there's not enough */
+	gint8 *frame_queue_state;
+	GMutex *mutex_queue_state;
+	GCond *cond_queue_state;
+	gint num_queued;
+	gint queue_frame;
+
+	/* True if we want to stop */
+	gboolean quit;
+
 	/* A/V sync... frame counter and internal cache */
 	gulong handled;
 	gint last_frame;

sys/v4l2/v4l2src_calls.c

@@ -40,6 +40,12 @@
 #define MAP_FAILED ( (caddr_t) -1 )
 #endif
 
+enum {
+  QUEUE_STATE_ERROR = -1,
+  QUEUE_STATE_READY_FOR_QUEUE,
+  QUEUE_STATE_QUEUED,
+  QUEUE_STATE_SYNCED,
+};
 
 /******************************************************
  * gst_v4l2src_fill_format_list():
@@ -113,20 +119,28 @@ gst_v4l2src_queue_frame (GstV4l2Src *v4l2src,
 {
 	DEBUG("queueing frame %d", num);
 
+	if (v4l2src->frame_queue_state[num] != QUEUE_STATE_READY_FOR_QUEUE) {
+		return FALSE;
+	}
+
 	v4l2src->bufsettings.index = num;
-	if (ioctl(GST_V4L2ELEMENT(v4l2src)->video_fd, VIDIOC_QBUF, &v4l2src->bufsettings) < 0) {
+	if (ioctl(GST_V4L2ELEMENT(v4l2src)->video_fd,
+		  VIDIOC_QBUF, &v4l2src->bufsettings) < 0) {
 		gst_element_error(GST_ELEMENT(v4l2src),
 			"Error queueing buffer %d on device %s: %s",
 			num, GST_V4L2ELEMENT(v4l2src)->device, g_strerror(errno));
 		return FALSE;
 	}
 
+	v4l2src->frame_queue_state[num] = QUEUE_STATE_QUEUED;
+	v4l2src->num_queued++;
+
 	return TRUE;
 }
 
 
 /******************************************************
- * gst_v4lsrc_sync_frame():
+ * gst_v4l2src_sync_next_frame():
  *   sync on a frame for capturing
  * return value: TRUE on success, FALSE on error
  ******************************************************/
@@ -135,15 +149,28 @@ static gboolean
 gst_v4l2src_sync_next_frame (GstV4l2Src *v4l2src,
                              gint       *num)
 {
-	if (ioctl(GST_V4L2ELEMENT(v4l2src)->video_fd, VIDIOC_DQBUF, &v4l2src->bufsettings) < 0) {
+	if (v4l2src->num_queued <= 0) {
-		gst_element_error(GST_ELEMENT(v4l2src),
-			"Error syncing on a buffer on device %s: %s",
-			GST_V4L2ELEMENT(v4l2src)->device, g_strerror(errno));
 		return FALSE;
 	}
+
+	while (ioctl(GST_V4L2ELEMENT(v4l2src)->video_fd,
+		     VIDIOC_DQBUF, &v4l2src->bufsettings) < 0) {
+		/* if the sync() got interrupted, we can retry */
+		if (errno != EINTR) {
+			gst_element_error(GST_ELEMENT(v4l2src),
+				"Error syncing on a buffer on device %s: %s",
+				GST_V4L2ELEMENT(v4l2src)->device, g_strerror(errno));
+			return FALSE;
+		}
+		DEBUG("Sync got interrupted");
+	}
+
 	DEBUG("synced on frame %d", v4l2src->bufsettings.index);
 	*num = v4l2src->bufsettings.index;
 
+	v4l2src->frame_queue_state[*num] = QUEUE_STATE_SYNCED;
+	v4l2src->num_queued--;
+
 	return TRUE;
 }
 
@@ -205,7 +232,8 @@ gst_v4l2src_set_capture (GstV4l2Src          *v4l2src,
 		return FALSE;
 	}
 
-	return TRUE;
+	/* update internal info */
+	return gst_v4l2src_get_capture(v4l2src);;
 }
 
 
@@ -220,15 +248,28 @@ gst_v4l2src_capture_init (GstV4l2Src *v4l2src)
 {
 	gint n;
 	gchar *desc = NULL;
+	struct v4l2_buffer buf;
 
 	DEBUG("initting the capture system");
 
 	GST_V4L2_CHECK_OPEN(GST_V4L2ELEMENT(v4l2src));
 	GST_V4L2_CHECK_NOT_ACTIVE(GST_V4L2ELEMENT(v4l2src));
 
+	/* set num of buffers */
+	if (v4l2src->breq.count > MIN_BUFFERS_QUEUED) {
+		struct v4l2_streamparm p;
+		p.type = v4l2src->format.type;
+
+		/* only if supported */
+		if (ioctl(GST_V4L2ELEMENT(v4l2src)->video_fd,
+			  VIDIOC_G_PARM, &p) == 0) {
+			p.parm.capture.readbuffers = v4l2src->breq.count;
+			ioctl(GST_V4L2ELEMENT(v4l2src)->video_fd,
+			      VIDIOC_S_PARM, &p);
+		}
+	}
+
 	/* request buffer info */
-	if (v4l2src->breq.count < MIN_BUFFERS_QUEUED)
-		v4l2src->breq.count = MIN_BUFFERS_QUEUED;
 	v4l2src->breq.type = v4l2src->format.type;
 	if (ioctl(GST_V4L2ELEMENT(v4l2src)->video_fd, VIDIOC_REQBUFS, &v4l2src->breq) < 0) {
 		gst_element_error(GST_ELEMENT(v4l2src),
@@ -255,24 +296,42 @@ gst_v4l2src_capture_init (GstV4l2Src *v4l2src)
 	gst_info("Got %d buffers (%s) of size %d KB\n",
 		v4l2src->breq.count, desc, v4l2src->format.fmt.pix.sizeimage/1024);
 
-	v4l2src->use_num_times = (gint *) malloc(sizeof(gint) * v4l2src->breq.count);
+	/* keep track of queued buffers */
-	if (!v4l2src->use_num_times) {
+	v4l2src->frame_queue_state = (gint8 *)
-		gst_element_error(GST_ELEMENT(v4l2src),
+		g_malloc(sizeof(gint8) * v4l2src->breq.count);
-			"Error creating sync-use-time tracker: %s",
+
-			g_strerror(errno));
+	/* track how often to use each frame */
-		return FALSE;
+	v4l2src->use_num_times = (gint *)
-	}
+		g_malloc(sizeof(gint) * v4l2src->breq.count);
+
+	/* lock for the frame_state */
+	v4l2src->mutex_queue_state = g_mutex_new();
+	v4l2src->cond_queue_state = g_cond_new();
 
 	/* Map the buffers */
-	GST_V4L2ELEMENT(v4l2src)->buffer = (guint8 **) g_malloc(sizeof(guint8*) * v4l2src->breq.count);
+	GST_V4L2ELEMENT(v4l2src)->buffer = (guint8 **)
+		g_malloc(sizeof(guint8 *) * v4l2src->breq.count);
 	for (n=0;n<v4l2src->breq.count;n++) {
-		GST_V4L2ELEMENT(v4l2src)->buffer[n] = mmap(0, v4l2src->format.fmt.pix.sizeimage, 
+		buf.index = n;
-			PROT_READ|PROT_WRITE, MAP_SHARED, GST_V4L2ELEMENT(v4l2src)->video_fd, v4l2src->format.fmt.pix.sizeimage*n);
+		buf.type = V4L2_BUF_TYPE_VIDEO_CAPTURE;
+		if (ioctl(GST_V4L2ELEMENT(v4l2src)->video_fd,
+			  VIDIOC_QUERYBUF, &buf) < 0) {
+			gst_element_error(GST_ELEMENT(v4l2src),
+					  "Failed to get buffer (%d) properties: %s",
+					  n, g_strerror(errno));
+			gst_v4l2src_capture_deinit(v4l2src);
+			return FALSE;
+		}
+		GST_V4L2ELEMENT(v4l2src)->buffer[n] = mmap(0,
+			buf.length, PROT_READ|PROT_WRITE, MAP_SHARED,
+			GST_V4L2ELEMENT(v4l2src)->video_fd, buf.m.offset);
 		if (GST_V4L2ELEMENT(v4l2src)->buffer[n] == MAP_FAILED) {
 			gst_element_error(GST_ELEMENT(v4l2src),
-				"Error mapping video buffer %d on device %s: %s",
+				"Error mapping video buffer (%d) on device %s: %s",
-				n, GST_V4L2ELEMENT(v4l2src)->device, g_strerror(errno));
+				n, GST_V4L2ELEMENT(v4l2src)->device,
+				g_strerror(errno));
 			GST_V4L2ELEMENT(v4l2src)->buffer[n] = NULL;
+			gst_v4l2src_capture_deinit(v4l2src);
 			return FALSE;
 		}
 	}
@@ -296,10 +355,22 @@ gst_v4l2src_capture_start (GstV4l2Src *v4l2src)
 	GST_V4L2_CHECK_OPEN(GST_V4L2ELEMENT(v4l2src));
 	GST_V4L2_CHECK_ACTIVE(GST_V4L2ELEMENT(v4l2src));
 
-	/* queue all buffers, this starts streaming capture */
+	g_mutex_lock(v4l2src->mutex_queue_state);
-	for (n=0;n<v4l2src->breq.count;n++)
+
-		if (!gst_v4l2src_queue_frame(v4l2src, n))
+	v4l2src->quit = FALSE;
+	v4l2src->num_queued = 0;
+	v4l2src->queue_frame = 0;
+
+	/* set all buffers ready to queue , this starts streaming capture */
+	for (n=0;n<v4l2src->breq.count;n++) {
+		v4l2src->frame_queue_state[n] = QUEUE_STATE_READY_FOR_QUEUE;
+		if (!gst_v4l2src_queue_frame(v4l2src, n)) {
+			g_mutex_unlock(v4l2src->mutex_queue_state);
+			gst_v4l2src_capture_stop(v4l2src);
 			return FALSE;
+		}
+	}
+
 	n = 1;
 	if (ioctl(GST_V4L2ELEMENT(v4l2src)->video_fd, VIDIOC_STREAMON, &n) < 0) {
 		gst_element_error(GST_ELEMENT(v4l2src),
@@ -308,6 +379,8 @@ gst_v4l2src_capture_start (GstV4l2Src *v4l2src)
 		return FALSE;
 	}
 
+	g_mutex_unlock(v4l2src->mutex_queue_state);
+
 	return TRUE;
 }
 
@@ -327,14 +400,63 @@ gst_v4l2src_grab_frame (GstV4l2Src *v4l2src,
 	GST_V4L2_CHECK_OPEN(GST_V4L2ELEMENT(v4l2src));
 	GST_V4L2_CHECK_ACTIVE(GST_V4L2ELEMENT(v4l2src));
 
+	g_mutex_lock(v4l2src->mutex_queue_state);
+
+	/* do we have enough frames? */
+	while (v4l2src->num_queued < MIN_BUFFERS_QUEUED ||
+	       v4l2src->frame_queue_state[v4l2src->queue_frame] ==
+			QUEUE_STATE_READY_FOR_QUEUE) {
+		while (v4l2src->frame_queue_state[v4l2src->queue_frame] !=
+				QUEUE_STATE_READY_FOR_QUEUE &&
+		       !v4l2src->quit) {
+			GST_DEBUG(GST_CAT_PLUGIN_INFO,
+				  "Waiting for frames to become available (%d < %d)",
+				  v4l2src->num_queued, MIN_BUFFERS_QUEUED);
+			g_cond_wait(v4l2src->cond_queue_state,
+				    v4l2src->mutex_queue_state);
+		}
+		if (v4l2src->quit) {
+			g_mutex_unlock(v4l2src->mutex_queue_state);
+			return TRUE; /* it won't get through anyway */
+		}
+		if (!gst_v4l2src_queue_frame(v4l2src, v4l2src->queue_frame)) {
+			g_mutex_unlock(v4l2src->mutex_queue_state);
+			return FALSE;
+		}
+		v4l2src->queue_frame = (v4l2src->queue_frame + 1) % v4l2src->breq.count;
+	}
+
 	/* syncing on the buffer grabs it */
-	if (!gst_v4l2src_sync_next_frame(v4l2src, num))
+	if (!gst_v4l2src_sync_next_frame(v4l2src, num)) {
+		g_mutex_unlock(v4l2src->mutex_queue_state);
 		return FALSE;
+	}
+
+	g_mutex_unlock(v4l2src->mutex_queue_state);
 
 	return TRUE;
 }
 
 
+/******************************************************
+ *
+ ******************************************************/
+
+guint8 *
+gst_v4l2src_get_buffer (GstV4l2Src *v4l2src,
+		        gint        num)
+{
+	if (!GST_V4L2_IS_ACTIVE(GST_V4L2ELEMENT(v4l2src)) ||
+	    !GST_V4L2_IS_OPEN(GST_V4L2ELEMENT(v4l2src)))
+		return NULL;
+
+	if (num < 0 || num >= v4l2src->breq.count)
+		return NULL;
+
+	return GST_V4L2ELEMENT(v4l2src)->buffer[num];
+}
+
+
 /******************************************************
  * gst_v4l2src_requeue_frame():
  *   re-queue a frame after we're done with the buffer
@@ -349,9 +471,23 @@ gst_v4l2src_requeue_frame (GstV4l2Src *v4l2src,
 	GST_V4L2_CHECK_OPEN(GST_V4L2ELEMENT(v4l2src));
 	GST_V4L2_CHECK_ACTIVE(GST_V4L2ELEMENT(v4l2src));
 
-	/* and let's queue the buffer */
+	/* mark frame as 'ready to requeue' */
-	if (!gst_v4l2src_queue_frame(v4l2src, num))
+	g_mutex_lock(v4l2src->mutex_queue_state);
+
+	if (v4l2src->frame_queue_state[num] != QUEUE_STATE_SYNCED) {
+		gst_element_error(GST_ELEMENT(v4l2src),
+				  "Invalid state %d (expected %d), can't requeue",
+				  v4l2src->frame_queue_state[num],
+				  QUEUE_STATE_SYNCED);
 		return FALSE;
+	}
+
+	v4l2src->frame_queue_state[num] = QUEUE_STATE_READY_FOR_QUEUE;
+
+	/* let an optional wait know */
+	g_cond_broadcast(v4l2src->cond_queue_state);
+
+	g_mutex_unlock(v4l2src->mutex_queue_state);
 
 	return TRUE;
 }
@@ -372,7 +508,10 @@ gst_v4l2src_capture_stop (GstV4l2Src *v4l2src)
 	GST_V4L2_CHECK_OPEN(GST_V4L2ELEMENT(v4l2src));
 	GST_V4L2_CHECK_ACTIVE(GST_V4L2ELEMENT(v4l2src));
 
-	/* we actually need to sync on all queued buffers but not on the non-queued ones */
+	g_mutex_lock(v4l2src->mutex_queue_state);
+
+	/* we actually need to sync on all queued buffers but not
+	 * on the non-queued ones */
 	if (ioctl(GST_V4L2ELEMENT(v4l2src)->video_fd, VIDIOC_STREAMOFF, &n) < 0) {
 		gst_element_error(GST_ELEMENT(v4l2src),
 			"Error stopping streaming capture for %s: %s",
@@ -380,6 +519,17 @@ gst_v4l2src_capture_stop (GstV4l2Src *v4l2src)
 		return FALSE;
 	}
 
+	/* make an optional pending wait stop */
+	v4l2src->quit = TRUE;
+	g_cond_broadcast(v4l2src->cond_queue_state);
+                                                                                
+	/* sync on remaining frames */
+	while (v4l2src->num_queued > 0) {
+		gst_v4l2src_sync_next_frame(v4l2src, &n);
+	}
+
+	g_mutex_unlock(v4l2src->mutex_queue_state);
+
 	return TRUE;
 }
 
@@ -393,23 +543,29 @@ gst_v4l2src_capture_stop (GstV4l2Src *v4l2src)
 gboolean
 gst_v4l2src_capture_deinit (GstV4l2Src *v4l2src)
 {
-	gint n;
+	int n;
-
+	
 	DEBUG("deinitting capture system");
 	GST_V4L2_CHECK_OPEN(GST_V4L2ELEMENT(v4l2src));
 	GST_V4L2_CHECK_ACTIVE(GST_V4L2ELEMENT(v4l2src));
 
 	/* unmap the buffer */
 	for (n=0;n<v4l2src->breq.count;n++) {
-		if (!GST_V4L2ELEMENT(v4l2src)->buffer[n])
+		if (!GST_V4L2ELEMENT(v4l2src)->buffer[n]) {
 			break;
-		munmap(GST_V4L2ELEMENT(v4l2src)->buffer[n], v4l2src->format.fmt.pix.sizeimage);
+		}
+		munmap(GST_V4L2ELEMENT(v4l2src)->buffer[n],
+		       v4l2src->format.fmt.pix.sizeimage);
 		GST_V4L2ELEMENT(v4l2src)->buffer[n] = NULL;
 	}
+
+	/* free buffer tracker */
 	g_free(GST_V4L2ELEMENT(v4l2src)->buffer);
 	GST_V4L2ELEMENT(v4l2src)->buffer = NULL;
-
+	g_mutex_free(v4l2src->mutex_queue_state);
-	free(v4l2src->use_num_times);
+	g_cond_free(v4l2src->cond_queue_state);
+	g_free(v4l2src->frame_queue_state);
+	g_free(v4l2src->use_num_times);
 
 	return TRUE;
 }

sys/v4l2/v4l2src_calls.h

@@ -33,6 +33,8 @@ gboolean	gst_v4l2src_capture_init	(GstV4l2Src *v4l2src);
 gboolean	gst_v4l2src_capture_start	(GstV4l2Src *v4l2src);
 gboolean	gst_v4l2src_grab_frame		(GstV4l2Src *v4l2src,
 						 gint       *num);
+guint8 *	gst_v4l2src_get_buffer		(GstV4l2Src *v4l2src,
+						 gint        num);
 gboolean	gst_v4l2src_requeue_frame	(GstV4l2Src *v4l2src,
 						 gint        num);
 gboolean	gst_v4l2src_capture_stop	(GstV4l2Src *v4l2src);