Index: gst/gstclock.c
===================================================================
RCS file: /cvs/gstreamer/gstreamer/gst/gstclock.c,v
retrieving revision 1.110
diff -u -r1.110 gstclock.c
--- gst/gstclock.c	25 Aug 2008 11:06:33 -0000	1.110
+++ gst/gstclock.c	18 Dec 2008 14:53:22 -0000
@@ -166,7 +166,7 @@
   entry->type = type;
   entry->time = time;
   entry->interval = interval;
-  entry->status = GST_CLOCK_BUSY;
+  entry->status = GST_CLOCK_OK;
   entry->func = NULL;
   entry->user_data = NULL;
 
Index: gst/gstsystemclock.c
===================================================================
RCS file: /cvs/gstreamer/gstreamer/gst/gstsystemclock.c,v
retrieving revision 1.49
diff -u -r1.49 gstsystemclock.c
--- gst/gstsystemclock.c	24 Dec 2007 14:35:24 -0000	1.49
+++ gst/gstsystemclock.c	18 Dec 2008 14:53:22 -0000
@@ -39,6 +39,18 @@
  */
 
 #include "gst_private.h"
+#ifdef HAVE_UNISTD_H
+#include <unistd.h>
+#endif
+
+#ifdef G_OS_WIN32
+#include <winsock2.h>
+#define EINPROGRESS WSAEINPROGRESS
+#else
+#include <sys/socket.h>
+#endif
+
+#include <gst/gstpoll.h>
 #include "gstinfo.h"
 
 #include "gstsystemclock.h"
@@ -46,12 +58,25 @@
 /* Define this to get some extra debug about jitter from each clock_wait */
 #undef WAIT_DEBUGGING
 
+#define GST_TYPE_CLOCK_TYPE (gst_clock_type_get_type())
+
+enum
+{
+  PROP_0,
+  PROP_CLOCK_TYPE
+  /* FILL ME */
+};
+
 /* the one instance of the systemclock */
 static GstClock *_the_system_clock = NULL;
 
 static void gst_system_clock_class_init (GstSystemClockClass * klass);
 static void gst_system_clock_init (GstSystemClock * clock);
 static void gst_system_clock_dispose (GObject * object);
+static void gst_system_clock_set_property (GObject * object, guint prop_id,
+    const GValue * value, GParamSpec * pspec);
+static void gst_system_clock_get_property (GObject * object, guint prop_id,
+    GValue * value, GParamSpec * pspec);
 
 static GstClockTime gst_system_clock_get_internal_time (GstClock * clock);
 static guint64 gst_system_clock_get_resolution (GstClock * clock);
@@ -98,6 +123,22 @@
   return clock_type;
 }
 
+static GType
+gst_clock_type_get_type (void)
+{
+  static GType clock_type_type = 0;
+  static const GEnumValue clock_types[] = {
+    {GST_CLOCK_TYPE_REALTIME, "GST_CLOCK_TYPE_REALTIME", "realtime"},
+    {GST_CLOCK_TYPE_MONOTONIC, "GST_CLOCK_TYPE_MONOTONIC", "monotonic"},
+    {0, NULL, NULL},
+  };
+
+  if (G_UNLIKELY (!clock_type_type)) {
+    clock_type_type = g_enum_register_static ("GstClockType", clock_types);
+  }
+  return clock_type_type;
+}
+
 static void
 gst_system_clock_class_init (GstSystemClockClass * klass)
 {
@@ -112,6 +153,13 @@
   parent_class = g_type_class_peek_parent (klass);
 
   gobject_class->dispose = gst_system_clock_dispose;
+  gobject_class->set_property = gst_system_clock_set_property;
+  gobject_class->get_property = gst_system_clock_get_property;
+
+  g_object_class_install_property (gobject_class, PROP_CLOCK_TYPE,
+      g_param_spec_enum ("clock-type", "Clock type",
+          "The type of underlying clock implementation used",
+          GST_TYPE_CLOCK_TYPE, GST_CLOCK_TYPE_REALTIME, G_PARAM_READWRITE));
 
   gstclock_class->get_internal_time = gst_system_clock_get_internal_time;
   gstclock_class->get_resolution = gst_system_clock_get_resolution;
@@ -129,11 +177,21 @@
       GST_CLOCK_FLAG_CAN_DO_PERIODIC_SYNC |
       GST_CLOCK_FLAG_CAN_DO_PERIODIC_ASYNC);
 
+  clock->ABI.abidata.clock_type = GST_CLOCK_TYPE_REALTIME;
+#ifdef G_OS_WIN32
+  if (_pipe (clock->ABI.abidata.fds, 4096, _O_BINARY) < 0) {
+    GST_CAT_ERROR (GST_CAT_CLOCK, "could not create pipe");
+  }
+#else
+  if (socketpair (PF_UNIX, SOCK_STREAM, 0, clock->ABI.abidata.fds) < 0) {
+    GST_CAT_ERROR (GST_CAT_CLOCK, "could not create socketpair");
+  }
+#endif
 #if 0
   /* Uncomment this to start the async clock thread straight away */
-  GST_CLOCK_LOCK (clock);
+  GST_OBJECT_LOCK (clock);
   gst_system_clock_start_async (clock);
-  GST_CLOCK_UNLOCK (clock);
+  GST_OBJECT_UNLOCK (clock);
 #endif
 }
 
@@ -171,6 +229,43 @@
     _the_system_clock = NULL;
     GST_CAT_DEBUG (GST_CAT_CLOCK, "disposed system clock");
   }
+
+  close (GST_SYSTEM_CLOCK (clock)->ABI.abidata.fds[0]);
+  close (GST_SYSTEM_CLOCK (clock)->ABI.abidata.fds[1]);
+}
+
+static void
+gst_system_clock_set_property (GObject * object, guint prop_id,
+    const GValue * value, GParamSpec * pspec)
+{
+  GstSystemClock *src = GST_SYSTEM_CLOCK (object);
+
+  switch (prop_id) {
+    case PROP_CLOCK_TYPE:
+      src->ABI.abidata.clock_type = g_value_get_enum (value);
+      GST_CAT_DEBUG (GST_CAT_CLOCK, "clock-type set to %d",
+          src->ABI.abidata.clock_type);
+      break;
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+gst_system_clock_get_property (GObject * object, guint prop_id, GValue * value,
+    GParamSpec * pspec)
+{
+  GstSystemClock *src = GST_SYSTEM_CLOCK (object);
+
+  switch (prop_id) {
+    case PROP_CLOCK_TYPE:
+      g_value_set_enum (value, src->ABI.abidata.clock_type);
+      break;
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
 }
 
 /**
@@ -214,6 +309,33 @@
   return clock;
 }
 
+static void
+gst_system_clock_clear_async_wakeups_unlocked (GstClock * clock)
+{
+  gchar c;
+
+  while (GST_SYSTEM_CLOCK (clock)->ABI.abidata.async_wakeup_count > 0) {
+    GST_CAT_DEBUG (GST_CAT_CLOCK, "reading wakeup byte");
+    if (read (GST_SYSTEM_CLOCK (clock)->ABI.abidata.fds[0], &c, 1) == -1) {
+      GST_CAT_ERROR (GST_CAT_CLOCK, "read failed");
+      g_assert_not_reached ();
+    }
+    GST_SYSTEM_CLOCK (clock)->ABI.abidata.async_wakeup_count--;
+  }
+  GST_CLOCK_BROADCAST (clock);
+}
+
+static void
+gst_system_clock_wakeup_async_unlocked (GstClock * clock)
+{
+  GST_CAT_DEBUG (GST_CAT_CLOCK, "writing wakeup byte");
+  if (write (GST_SYSTEM_CLOCK (clock)->ABI.abidata.fds[1], "A", 1) == -1) {
+    GST_CAT_ERROR (GST_CAT_CLOCK, "write failed");
+    g_assert_not_reached ();
+  }
+  GST_SYSTEM_CLOCK (clock)->ABI.abidata.async_wakeup_count++;
+}
+
 /* this thread reads the sorted clock entries from the queue.
  *
  * It waits on each of them and fires the callback when the timeout occurs.
@@ -285,6 +407,10 @@
               entry->user_data);
           GST_OBJECT_LOCK (clock);
         }
+        if (clock->entries->data != entry) {
+          /* new entries have been added, clear async wakeups */
+          gst_system_clock_clear_async_wakeups_unlocked (clock);
+        }
         if (entry->type == GST_CLOCK_ENTRY_PERIODIC) {
           /* adjust time now */
           entry->time = requested + entry->interval;
@@ -303,6 +429,8 @@
          * was canceled. Whatever it is, pick the head entry of the list and
          * continue waiting. */
         GST_CAT_DEBUG (GST_CAT_CLOCK, "async entry %p needs restart", entry);
+        /* clear async wakeups, if any */
+        gst_system_clock_clear_async_wakeups_unlocked (clock);
         continue;
       default:
         GST_CAT_DEBUG (GST_CAT_CLOCK,
@@ -323,21 +451,60 @@
   GST_CAT_DEBUG (GST_CAT_CLOCK, "exit system clock thread");
 }
 
+#ifdef HAVE_POSIX_TIMERS
+static inline clockid_t
+clock_type_to_posix_id (GstClockType clock_type)
+{
+#ifdef HAVE_MONOTONIC_CLOCK
+  if (clock_type == GST_CLOCK_TYPE_MONOTONIC)
+    return CLOCK_MONOTONIC;
+  else
+#endif
+  return CLOCK_REALTIME;
+}
+#endif
+
 /* MT safe */
 static GstClockTime
 gst_system_clock_get_internal_time (GstClock * clock)
 {
+#ifdef HAVE_POSIX_TIMERS
+  GstSystemClock *sysclock = GST_SYSTEM_CLOCK_CAST (clock);
+  clockid_t ptype;
+  struct timespec ts;
+
+  ptype = clock_type_to_posix_id (sysclock->ABI.abidata.clock_type);
+
+  if (G_UNLIKELY (clock_gettime (ptype, &ts)))
+    return GST_CLOCK_TIME_NONE;
+
+  return GST_TIMESPEC_TO_TIME (ts);
+#else
   GTimeVal timeval;
 
   g_get_current_time (&timeval);
 
   return GST_TIMEVAL_TO_TIME (timeval);
+#endif
 }
 
 static guint64
 gst_system_clock_get_resolution (GstClock * clock)
 {
+#ifdef HAVE_POSIX_TIMERS
+  GstSystemClock *sysclock = GST_SYSTEM_CLOCK_CAST (clock);
+  clockid_t ptype;
+  struct timespec ts;
+
+  ptype = clock_type_to_posix_id (sysclock->ABI.abidata.clock_type);
+
+  if (G_UNLIKELY (clock_getres (ptype, &ts)))
+    return GST_CLOCK_TIME_NONE;
+
+  return GST_TIMESPEC_TO_TIME (ts);
+#else
   return 1 * GST_USECOND;
+#endif
 }
 
 /* synchronously wait on the given GstClockEntry.
@@ -360,6 +527,7 @@
 gst_system_clock_id_wait_jitter_unlocked (GstClock * clock,
     GstClockEntry * entry, GstClockTimeDiff * jitter, gboolean restart)
 {
+  GstSystemClock *sysclock = GST_SYSTEM_CLOCK_CAST (clock);
   GstClockTime entryt, real, now, target;
   GstClockTimeDiff diff;
 
@@ -392,51 +560,100 @@
       GST_TIME_ARGS (entryt), GST_TIME_ARGS (now), GST_TIME_ARGS (real), diff);
 
   if (diff > 0) {
-    GTimeVal tv;
-
+    GstPoll *fdset;
+    GstPollFD fd = GST_POLL_FD_INIT;
 #ifdef WAIT_DEBUGGING
     GstClockTime result, final;
 #endif
 
-    GST_TIME_TO_TIMEVAL (target, tv);
+    fdset = gst_poll_new (FALSE);
+    fd.fd = sysclock->ABI.abidata.fds[0];
+    gst_poll_add_fd (fdset, &fd);
+    gst_poll_fd_ctl_read (fdset, &fd, TRUE);
+
+    while (entry->status != GST_CLOCK_UNSCHEDULED) {
+      gint pollret;
+
+      /* now wait on the entry, it either times out or the fd is written. */
+      entry->status = GST_CLOCK_BUSY;
+      GST_OBJECT_UNLOCK (clock);
+      pollret = gst_poll_wait (fdset, diff);
+      /* another thread can read the fd before we get the lock */
+      GST_OBJECT_LOCK (clock);
+      if (entry->status == GST_CLOCK_UNSCHEDULED) {
+        gchar c;
+
+        GST_CAT_DEBUG (GST_CAT_CLOCK, "entry %p unlocked with signal", entry);
+        if (read (sysclock->ABI.abidata.fds[0], &c, 1) == -1) {
+          GST_CAT_ERROR (GST_CAT_CLOCK, "read failed");
+          g_assert_not_reached ();
+        }
+        GST_CLOCK_BROADCAST (clock);
+      } else {
+        if (pollret != 0) {
+          /* some other id got unlocked */
 
-    while (TRUE) {
-      /* now wait on the entry, it either times out or the cond is signaled. */
-      if (!GST_CLOCK_TIMED_WAIT (clock, &tv)) {
-        /* timeout, this is fine, we can report success now */
-        GST_CAT_DEBUG (GST_CAT_CLOCK, "entry %p unlocked after timeout", entry);
-        entry->status = GST_CLOCK_OK;
+          if (!restart) {
+            /* this can happen if the entry got unlocked because of an async
+             * entry was added to the head of the async queue. */
+            GST_CAT_DEBUG (GST_CAT_CLOCK, "wakeup waiting for entry %p", entry);
+            break;
+          }
+
+          /* mark ourselves as EARLY, we release the lock and we could be
+           * unscheduled ourselves but we don't want the unscheduling thread
+           * to write on the fd */
+          entry->status = GST_CLOCK_EARLY;
+
+          /* before waiting on the cond, check if another thread read the fd
+           * before we got the lock */
+          while (gst_poll_wait (fdset, 0) > 0) {
+            GST_CLOCK_WAIT (clock);
+          }
+
+          if (entry->status == GST_CLOCK_UNSCHEDULED) {
+            break;
+          }
+
+          /* recalc diff for next gst_poll_wait */
+          real = GST_CLOCK_GET_CLASS (clock)->get_internal_time (clock);
+          now = gst_clock_adjust_unlocked (clock, real);
+          diff = entryt - now;
+          if (diff <= 0) {
+            entry->status = GST_CLOCK_OK;
+            break;
+          }
+        } else {
+          /* sleep if gst_poll_wait returned early */
+          real = GST_CLOCK_GET_CLASS (clock)->get_internal_time (clock);
+          now = gst_clock_adjust_unlocked (clock, real);
+          diff = entryt - now;
+          if (diff > 0) {
+            g_usleep (diff / 1000);
+          }
+
+          /* timeout, this is fine, we can report success now */
+          GST_CAT_DEBUG (GST_CAT_CLOCK, "entry %p unlocked after timeout",
+              entry);
+          entry->status = GST_CLOCK_OK;
 
 #ifdef WAIT_DEBUGGING
-        real = GST_CLOCK_GET_CLASS (clock)->get_internal_time (clock);
-        result = gst_clock_adjust_unlocked (clock, real);
-        final = gst_system_clock_get_internal_time (clock);
-        GST_CAT_DEBUG (GST_CAT_CLOCK, "Waited for %" G_GINT64_FORMAT
-            " got %" G_GINT64_FORMAT " diff %" G_GINT64_FORMAT
-            " %g target-offset %" G_GINT64_FORMAT " %g", entryt, result,
-            result - entryt,
-            (double) (GstClockTimeDiff) (result - entryt) / GST_SECOND,
-            (final - target),
-            ((double) (GstClockTimeDiff) (final - target)) / GST_SECOND);
+          real = GST_CLOCK_GET_CLASS (clock)->get_internal_time (clock);
+          result = gst_clock_adjust_unlocked (clock, real);
+          final = gst_system_clock_get_internal_time (clock);
+          GST_CAT_DEBUG (GST_CAT_CLOCK, "Waited for %" G_GINT64_FORMAT
+              " got %" G_GINT64_FORMAT " diff %" G_GINT64_FORMAT
+              " %g target-offset %" G_GINT64_FORMAT " %g", entryt, result,
+              result - entryt,
+              (double) (GstClockTimeDiff) (result - entryt) / GST_SECOND,
+              (final - target),
+              ((double) (GstClockTimeDiff) (final - target)) / GST_SECOND);
 #endif
-        break;
-      } else {
-        /* the waiting is interrupted because the GCond was signaled. This can
-         * be because this or some other entry was unscheduled. */
-        GST_CAT_DEBUG (GST_CAT_CLOCK, "entry %p unlocked with signal", entry);
-        /* if the entry is unscheduled, we can stop waiting for it, else we
-         * continue our while loop. */
-        if (entry->status == GST_CLOCK_UNSCHEDULED)
           break;
-        /* else restart if we must */
-        if (!restart)
-          break;
-
-        /* this can happen if the entry got unlocked because of an async entry
-         * was added to the head of the async queue. */
-        GST_CAT_DEBUG (GST_CAT_CLOCK, "continue waiting for entry %p", entry);
+        }
       }
     }
+    gst_poll_free (fdset);
   } else if (diff == 0) {
     entry->status = GST_CLOCK_OK;
   } else {
@@ -496,6 +713,8 @@
 static GstClockReturn
 gst_system_clock_id_wait_async (GstClock * clock, GstClockEntry * entry)
 {
+  gboolean empty;
+
   GST_CAT_DEBUG (GST_CAT_CLOCK, "adding async entry %p", entry);
 
   GST_OBJECT_LOCK (clock);
@@ -504,6 +723,8 @@
   if (!gst_system_clock_start_async (GST_SYSTEM_CLOCK (clock)))
     goto thread_error;
 
+  empty = (clock->entries == NULL);
+
   /* need to take a ref */
   gst_clock_id_ref ((GstClockID) entry);
   /* insert the entry in sorted order */
@@ -514,13 +735,18 @@
    * front, else the thread is just waiting for another entry and
    * will get to this entry automatically. */
   if (clock->entries->data == entry) {
-    GST_CAT_DEBUG (GST_CAT_CLOCK, "async entry added to head, sending signal");
+    GST_CAT_DEBUG (GST_CAT_CLOCK, "async entry added to head");
     /* this will wake up _all_ entries waiting for the clock because we have
      * only one cond for all entries (makes allocation faster). Entries that
      * have not timed out will have their status set to BUSY and should continue
      * to wait. In the case of the async ones, the new head entry should be
      * taken and waited for. */
-    GST_CLOCK_BROADCAST (clock);
+    if (empty) {
+      GST_CAT_DEBUG (GST_CAT_CLOCK, "sending signal");
+      GST_CLOCK_BROADCAST (clock);
+    } else {
+      gst_system_clock_wakeup_async_unlocked (clock);
+    }
   }
   GST_OBJECT_UNLOCK (clock);
 
@@ -543,12 +769,18 @@
 {
   GST_CAT_DEBUG (GST_CAT_CLOCK, "unscheduling entry %p", entry);
 
-  GST_OBJECT_LOCK (clock);
   /* mark entry as unscheduled, then wake up all entries. The entries that did
    * not timeout will be woken up but immediatly go to sleep again because their
    * status would still be busy. */
+  GST_OBJECT_LOCK (clock);
+  if (entry->status == GST_CLOCK_BUSY) {
+    GST_CAT_DEBUG (GST_CAT_CLOCK, "writing unschedule byte 'U'");
+    if (write (GST_SYSTEM_CLOCK (clock)->ABI.abidata.fds[1], "U", 1) == -1) {
+      GST_CAT_ERROR (GST_CAT_CLOCK, "write failed");
+      g_assert_not_reached ();
+    }
+  }
+  /* when it leaves the poll, it'll detect the unscheduled */
   entry->status = GST_CLOCK_UNSCHEDULED;
-  GST_CAT_DEBUG (GST_CAT_CLOCK, "sending signal");
-  GST_CLOCK_BROADCAST (clock);
   GST_OBJECT_UNLOCK (clock);
 }
Index: gst/gstsystemclock.h
===================================================================
RCS file: /cvs/gstreamer/gstreamer/gst/gstsystemclock.h,v
retrieving revision 1.15
diff -u -r1.15 gstsystemclock.h
--- gst/gstsystemclock.h	28 Oct 2005 18:14:24 -0000	1.15
+++ gst/gstsystemclock.h	18 Dec 2008 14:53:22 -0000
@@ -31,6 +31,7 @@
 
 #define GST_TYPE_SYSTEM_CLOCK 			(gst_system_clock_get_type ())
 #define GST_SYSTEM_CLOCK(obj) 			(G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_SYSTEM_CLOCK, GstSystemClock))
+#define GST_SYSTEM_CLOCK_CAST(obj)		((GstSystemClock *)(obj))
 #define GST_IS_SYSTEM_CLOCK(obj) 		(G_TYPE_CHECK_INSTANCE_TYPE ((obj), GST_TYPE_SYSTEM_CLOCK))
 #define GST_SYSTEM_CLOCK_CLASS(klass) 		(G_TYPE_CHECK_CLASS_CAST ((klass), GST_TYPE_SYSTEM_CLOCK, GstSystemClockClass))
 #define GST_IS_SYSTEM_CLOCK_CLASS(klass) 	(G_TYPE_CHECK_CLASS_TYPE ((klass), GST_TYPE_SYSTEM_CLOCK))
@@ -41,6 +42,17 @@
 typedef struct _GstSystemClockClass GstSystemClockClass;
 
 /**
+ * GstClockType:
+ * @GST_CLOCK_TYPE_REALTIME: time since Epoch
+ * @GST_CLOCK_TYPE_MONOTONIC: monotonic time since some unspecified starting
+ *                            point
+ */
+typedef enum {
+  GST_CLOCK_TYPE_REALTIME	= 0,
+  GST_CLOCK_TYPE_MONOTONIC	= 1
+} GstClockType;
+
+/**
  * GstSystemClock:
  * @clock: The parent clock
  *
@@ -53,7 +65,15 @@
   GThread	*thread;	/* thread for async notify */
   gboolean 	 stopping;
 
-  gpointer _gst_reserved[GST_PADDING];
+  /* ABI added */
+  union {
+    struct {
+      GstClockType	 clock_type;
+      gint		 fds[2];
+      gint		 async_wakeup_count;
+    } abidata;
+    gpointer _gst_reserved[GST_PADDING-0];
+  } ABI;
 };
 
 struct _GstSystemClockClass {
