Delay() accuracy issues / Weird behavior of job sc

问题:

I'm currently trying to build a job scheduler as shown below. My goal is to be able to schedule the launch of arbitrary functions (here of (Long) -> Unit)) with as much accuracy as possible on their launch time (sub-millisecond would be ideal).

import java.util.*
import kotlinx.coroutines.*
import java.util.concurrent.PriorityBlockingQueue
import kotlin.math.max
import java.time.Instant

fun nowInMicrosSinceEpoch() : Long {
    val now = Instant.now()
    return now.toEpochMilli() * 1000L + (now.getNano().toLong() / 1000L)
}

open class TimeCallback(open var time : Long, open val callback : (Long) -> Unit) {
    open fun run(){
        callback(time)
    }

    override fun toString() : String {
        return "(TimeCallback - T:${time/1000L})"
    }
}

class PulseCallback(override var time : Long,
                    override val callback : (Long) -> Unit,
                    val pulsePeriod : Long,
                    val callbackQueue : AbstractQueue<TimeCallback>) : TimeCallback(time, callback) {
    override fun run(){
        callback(time)
        time += pulsePeriod
        callbackQueue.add(this)
    }

    override fun toString() : String {
        return "(PulseCallback - T:${time/1000L} - PP:${pulsePeriod/1000L})"
    }
}

abstract class Clock {
    protected abstract var currentTime: Long
    protected val comparator : Comparator<TimeCallback> = compareBy<TimeCallback> { x -> x.time }

    abstract fun start()
    abstract fun stop()
    abstract fun addCallback(time: Long, callback: (Long) -> Unit)
    abstract fun addPulseCallback(time: Long, pulsePeriod: Long, callback: (Long) -> Unit)
    abstract fun getTime() : Long
}

class LiveClock : Clock() {
    override var currentTime : Long = nowInMicrosSinceEpoch()
    private val callbacks : PriorityBlockingQueue<TimeCallback> = PriorityBlockingQueue<TimeCallback>(10000, comparator)

    private var clockCoroutine : Job? = null

    override fun start(){
        clockCoroutine = GlobalScope.launch {
            try{
                var waitTime : Long
                while(true) {
                    println(callbacks)
                    val callback: TimeCallback = callbacks.take()
                    currentTime = nowInMicrosSinceEpoch()
                    waitTime = max(callback.time - currentTime, 0L) / 1000L
                    println("Now is ${currentTime/1000L}, waiting $waitTime ms until ${callback.time/1000L}")
                    delay(waitTime)
                    callback.run()
                }
            } finally {
                println("Clock was stopped by CancellationException.")
            }
        }
    }

    override fun stop(){
        // Cannot stop before starting!
        clockCoroutine!!.cancel()
    }

    override fun addCallback(time: Long, callback: (Long) -> Unit){
        callbacks.add(TimeCallback(
            time = time,
            callback = callback
        ))
    }

    override fun addPulseCallback(firstPulse: Long, pulsePeriod: Long, callback: (Long) -> Unit){
        callbacks.add(PulseCallback(
            time = firstPulse,
            pulsePeriod = pulsePeriod,
            callback = callback,
            callbackQueue = callbacks
        ))
    }

    override fun getTime() : Long {
        return nowInMicrosSinceEpoch()
    }
}

fun printTest(t : Long){
    println("Time difference: ${nowInMicrosSinceEpoch()/1000L - (t/1000L)} ms")
}

fun main(args: Array<String>) {
    val clock = LiveClock()
    clock.addPulseCallback(nowInMicrosSinceEpoch(), 1000*1000L, ::printTest)
    clock.addPulseCallback(nowInMicrosSinceEpoch(), 500*1000L, ::printTest)
    clock.start()
    runBlocking {
        // Run for 100 seconds...
        delay(100000L)
    }
}

However, even with the very simple example above (in main()), I obtain significant time differences between the scheduled times and the times at which the scheduled functions are actually run. Some are even run before their scheduled time (see last line below, negative time difference), which remains a mystery to me. How is it possible that the callback is run before the time delay() was called?

Thanks!

[(PulseCallback - T:1547692545172 - PP:1000), (PulseCallback - T:1547692545184 - PP:500)]
Now is 1547692545262, waiting 0 ms until 1547692545172
1547692545264 - Time difference: 92 ms
[(PulseCallback - T:1547692545184 - PP:500), (PulseCallback - T:1547692546172 - PP:1000)]
Now is 1547692545264, waiting 0 ms until 1547692545184
1547692545264 - Time difference: 80 ms
[(PulseCallback - T:1547692545684 - PP:500), (PulseCallback - T:1547692546172 - PP:1000)]
Now is 1547692545264, waiting 420 ms until 1547692545684
1547692546110 - Time difference: 426 ms
[(PulseCallback - T:1547692546172 - PP:1000), (PulseCallback - T:1547692546184 - PP:500)]
Now is 1547692546110, waiting 62 ms until 1547692546172
1547692546234 - Time difference: 62 ms
[(PulseCallback - T:1547692546184 - PP:500), (PulseCallback - T:1547692547172 - PP:1000)]
Now is 1547692546234, waiting 0 ms until 1547692546184
1547692546234 - Time difference: 50 ms
[(PulseCallback - T:1547692546684 - PP:500), (PulseCallback - T:1547692547172 - PP:1000)]
Now is 1547692546234, waiting 450 ms until 1547692546684
1547692546136 - Time difference: -548 ms
[(PulseCallback - T:1547692547172 - PP:1000), (PulseCallback - T:1547692547184 - PP:500)]
Now is 1547692546136, waiting 1036 ms until 1547692547172

回答1:

The implementation of nowInMicrosSinceEpoch() is wrong. The millisecond value is applied twice.

To show this, here is Java code to print the values used in nowInMicrosSinceEpoch():

Instant now = Instant.now();
System.out.println(now);
System.out.printf("%23d        toEpochMilli()%n", now.toEpochMilli());
System.out.printf("%26d     toEpochMilli() * 1000 = a%n", now.toEpochMilli() * 1000L);
System.out.printf("%29d  getNano()%n", now.getNano());
System.out.printf("%26d     getNano() / 1000 = b%n", now.getNano() / 1000L);
System.out.printf("%26d     a + b%n", now.toEpochMilli() * 1000L + now.getNano() / 1000L);

Output

2019-02-02T00:16:58.999999999Z
          1549066618999        toEpochMilli()
          1549066618999000     toEpochMilli() * 1000 = a
                    999999999  getNano()
                    999999     getNano() / 1000 = b
          1549066619998999     a + b

So when the clock rolls over from x:58.999999999Z to x:59.000000000Z you get:

2019-02-02T00:16:59.000000000Z
          1549066619000        toEpochMilli()
          1549066619000000     toEpochMilli() * 1000 = a
                    000000000  getNano()
                    000000     getNano() / 1000 = b
          1549066619000000     a + b

The value 1 nanosecond later returns a value that is 998999 microsecond earlier.
The calculated value is running at double speed, and jumps back 1 second every second.

The correct formula is (in Java):

Instant now = Instant.now();
return now.getEpochSecond() * 1000000L + now.getNano() / 1000L;