Проблемы с задержкой () / странное поведение планировщика заданий

Question

В настоящее время я пытаюсь построить планировщик заданий, как показано ниже. Моя цель состоит в том, чтобы иметь возможность планировать запуск произвольных функций (здесь (Long) -> Unit)) с максимально возможной точностью во время их запуска (идеальной была бы миллисекунда).

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)
    }
}

Однако, даже с очень простым примером выше (в main ()) я получаю значительную разницу во времени между запланированным временем и временем, в которое запланированные функции фактически выполняются. Некоторые даже запускаются раньше запланированного времени (см. Последнюю строку ниже, отрицательная разница во времени), что для меня остается загадкой. Как это возможно, что обратный вызов выполняется до вызова time delay ()?

Спасибо!

[(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

Answer 1

Реализация nowInMicrosSinceEpoch() неверна. Значение в миллисекундах применяется дважды.

Чтобы показать это, вот код Java для печати значений, используемых в 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);

выход

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

Поэтому, когда часы переворачиваются с x:58.999999999Z на x:59.000000000Z, вы получаете:

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

Значение 1 наносекунда спустя возвращает значение, которое на 998999 микросекунд раньше.
Рассчитанное значение работает с удвоенной скоростью и каждую секунду возвращается назад на 1 секунду.

Правильная формула (на Java):

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