Default delay improvements

  * Fixed memory leak when coroutine was cancelled before delay() invocation
  * synchronized(this) is replaced with @synchronized to reduce TSH footprint

Author: qwwdfsad

core/kotlinx-coroutines-core/src/EventLoop.kt

@@ -72,9 +72,9 @@ public fun EventLoop(thread: Thread = Thread.currentThread(), parentJob: Job? =
 public fun EventLoop_Deprecated(thread: Thread = Thread.currentThread(), parentJob: Job? = null): CoroutineDispatcher =
     EventLoop(thread, parentJob) as CoroutineDispatcher
 
-private const val DELAYED = 0
-private const val REMOVED = 1
-private const val RESCHEDULED = 2
+internal const val DELAYED = 0
+internal const val REMOVED = 1
+internal const val RESCHEDULED = 2
 
 private const val MS_TO_NS = 1_000_000L
 private const val MAX_MS = Long.MAX_VALUE / MS_TO_NS
@@ -243,10 +243,10 @@ internal abstract class EventLoopBase: CoroutineDispatcher(), Delay, EventLoop {
 
     internal fun schedule(delayedTask: DelayedTask) {
         if (scheduleImpl(delayedTask)) {
-            // todo: we should unpark only when this delayed task became first in the queue
             unpark()
-        } else
+        } else {
             DefaultExecutor.schedule(delayedTask)
+        }
     }
 
     private fun scheduleImpl(delayedTask: DelayedTask): Boolean {

core/kotlinx-coroutines-core/src/internal/ThreadSafeHeap.kt

@@ -4,6 +4,7 @@
 
 package kotlinx.coroutines.experimental.internal
 
+import kotlinx.coroutines.experimental.*
 import java.util.*
 
 /**
@@ -26,42 +27,48 @@ public class ThreadSafeHeap<T> : SynchronizedObject() where T: ThreadSafeHeapNod
 
     public val isEmpty: Boolean get() = size == 0
 
-    public fun clear() = synchronized(this) {
+    @Synchronized
+    public fun clear() {
         Arrays.fill(a, 0, size, null)
         size = 0
     }
 
-    public fun peek(): T? = synchronized(this) { firstImpl() }
+    @Synchronized
+    public fun peek(): T? = firstImpl()
 
-    public fun removeFirstOrNull(): T? = synchronized(this) {
+    @Synchronized
+    public fun removeFirstOrNull(): T? =
         if (size > 0) {
             removeAtImpl(0)
-        } else
+        } else {
             null
-    }
+        }
 
-    public inline fun removeFirstIf(predicate: (T) -> Boolean): T? = synchronized(this) {
-        val first = firstImpl() ?: return@synchronized null
-        if (predicate(first)) {
+    @Synchronized
+    public inline fun removeFirstIf(predicate: (T) -> Boolean): T?  {
+        val first = firstImpl() ?: return null
+        return if (predicate(first)) {
             removeAtImpl(0)
-        } else
+        } else {
             null
+        }
     }
 
-    public fun addLast(node: T) = synchronized(this) {
-        addImpl(node)
-    }
+    @Synchronized
+    public fun addLast(node: T) = addImpl(node)
 
-    public fun addLastIf(node: T, cond: () -> Boolean): Boolean = synchronized(this) {
+    @Synchronized
+    public fun addLastIf(node: T, cond: () -> Boolean): Boolean =
         if (cond()) {
             addImpl(node)
             true
-        } else
+        } else {
             false
-    }
+        }
 
-    public fun remove(node: T): Boolean = synchronized(this) {
-        if (node.index < 0) {
+    @Synchronized
+    public fun remove(node: T): Boolean {
+        return if (node.index < 0) {
             false
         } else {
             removeAtImpl(node.index)
@@ -95,6 +102,11 @@ public class ThreadSafeHeap<T> : SynchronizedObject() where T: ThreadSafeHeapNod
 
     @PublishedApi
     internal fun addImpl(node: T) {
+        // TODO remove this after #541 when ThreadSafeHeapNode is gone
+        if (node is EventLoopBase.DelayedTask && node.state == REMOVED) {
+            return
+        }
+
         val a = realloc()
         val i = size++
         a[i] = node
@@ -140,4 +152,4 @@ public class ThreadSafeHeap<T> : SynchronizedObject() where T: ThreadSafeHeapNod
         ni.index = i
         nj.index = j
     }
-}
+}