Use std/monotimes in the stdlib

This commit is contained in:
Oscar Nihlgård 2019-07-15 21:28:41 +02:00
commit 1f32498a47
3 changed files with 25 additions and 20 deletions

View file

@ -169,7 +169,7 @@
include "system/inclrtl" include "system/inclrtl"
import os, tables, strutils, times, heapqueue, lists, options, asyncstreams import os, tables, strutils, times, heapqueue, lists, options, asyncstreams
import options, math import options, math, std/monotimes
import asyncfutures except callSoon import asyncfutures except callSoon
import nativesockets, net, deques import nativesockets, net, deques
@ -184,7 +184,7 @@ export asyncstreams
type type
PDispatcherBase = ref object of RootRef PDispatcherBase = ref object of RootRef
timers*: HeapQueue[tuple[finishAt: float, fut: Future[void]]] timers*: HeapQueue[tuple[finishAt: MonoTime, fut: Future[void]]]
callbacks*: Deque[proc () {.gcsafe.}] callbacks*: Deque[proc () {.gcsafe.}]
proc processTimers( proc processTimers(
@ -192,7 +192,7 @@ proc processTimers(
): Option[int] {.inline.} = ): Option[int] {.inline.} =
# Pop the timers in the order in which they will expire (smaller `finishAt`). # Pop the timers in the order in which they will expire (smaller `finishAt`).
var count = p.timers.len var count = p.timers.len
let t = epochTime() let t = getMonoTime()
while count > 0 and t >= p.timers[0].finishAt: while count > 0 and t >= p.timers[0].finishAt:
p.timers.pop().fut.complete() p.timers.pop().fut.complete()
dec count dec count
@ -201,8 +201,8 @@ proc processTimers(
# Return the number of miliseconds in which the next timer will expire. # Return the number of miliseconds in which the next timer will expire.
if p.timers.len == 0: return if p.timers.len == 0: return
let milisecs = (p.timers[0].finishAt - epochTime()) * 1000 let millisecs = (p.timers[0].finishAt - getMonoTime()).inMilliseconds
return some(ceil(milisecs).int) return some(millisecs.int + 1)
proc processPendingCallbacks(p: PDispatcherBase; didSomeWork: var bool) = proc processPendingCallbacks(p: PDispatcherBase; didSomeWork: var bool) =
while p.callbacks.len > 0: while p.callbacks.len > 0:
@ -1778,7 +1778,11 @@ proc sleepAsync*(ms: int | float): owned(Future[void]) =
## ``ms`` milliseconds. ## ``ms`` milliseconds.
var retFuture = newFuture[void]("sleepAsync") var retFuture = newFuture[void]("sleepAsync")
let p = getGlobalDispatcher() let p = getGlobalDispatcher()
p.timers.push((epochTime() + (ms / 1000), retFuture)) when ms is int:
p.timers.push((getMonoTime() + initDuration(milliseconds = ms), retFuture))
elif ms is float:
let ns = (ms * 1_000_000).int64
p.timers.push((getMonoTime() + initDuration(nanoseconds = ns), retFuture))
return retFuture return retFuture
proc withTimeout*[T](fut: Future[T], timeout: int): owned(Future[bool]) = proc withTimeout*[T](fut: Future[T], timeout: int): owned(Future[bool]) =

View file

@ -176,7 +176,7 @@
## ##
import net, strutils, uri, parseutils, strtabs, base64, os, mimetypes, import net, strutils, uri, parseutils, strtabs, base64, os, mimetypes,
math, random, httpcore, times, tables, streams math, random, httpcore, times, tables, streams, std/monotimes
import asyncnet, asyncdispatch, asyncfile import asyncnet, asyncdispatch, asyncfile
import nativesockets import nativesockets
@ -610,7 +610,7 @@ type
contentTotal: BiggestInt contentTotal: BiggestInt
contentProgress: BiggestInt contentProgress: BiggestInt
oneSecondProgress: BiggestInt oneSecondProgress: BiggestInt
lastProgressReport: float lastProgressReport: MonoTime
when SocketType is AsyncSocket: when SocketType is AsyncSocket:
bodyStream: FutureStream[string] bodyStream: FutureStream[string]
parseBodyFut: Future[void] parseBodyFut: Future[void]
@ -706,13 +706,13 @@ proc reportProgress(client: HttpClient | AsyncHttpClient,
progress: BiggestInt) {.multisync.} = progress: BiggestInt) {.multisync.} =
client.contentProgress += progress client.contentProgress += progress
client.oneSecondProgress += progress client.oneSecondProgress += progress
if epochTime() - client.lastProgressReport >= 1.0: if (getMonoTime() - client.lastProgressReport).inSeconds > 1:
if not client.onProgressChanged.isNil: if not client.onProgressChanged.isNil:
await client.onProgressChanged(client.contentTotal, await client.onProgressChanged(client.contentTotal,
client.contentProgress, client.contentProgress,
client.oneSecondProgress) client.oneSecondProgress)
client.oneSecondProgress = 0 client.oneSecondProgress = 0
client.lastProgressReport = epochTime() client.lastProgressReport = getMonoTime()
proc recvFull(client: HttpClient | AsyncHttpClient, size: int, timeout: int, proc recvFull(client: HttpClient | AsyncHttpClient, size: int, timeout: int,
keep: bool): Future[int] {.multisync.} = keep: bool): Future[int] {.multisync.} =
@ -784,7 +784,7 @@ proc parseBody(client: HttpClient | AsyncHttpClient,
client.contentTotal = 0 client.contentTotal = 0
client.contentProgress = 0 client.contentProgress = 0
client.oneSecondProgress = 0 client.oneSecondProgress = 0
client.lastProgressReport = 0 client.lastProgressReport = MonoTime()
when client is AsyncHttpClient: when client is AsyncHttpClient:
assert(not client.bodyStream.finished) assert(not client.bodyStream.finished)

View file

@ -65,7 +65,8 @@
## echo("Client connected from: ", address) ## echo("Client connected from: ", address)
{.deadCodeElim: on.} # dce option deprecated {.deadCodeElim: on.} # dce option deprecated
import nativesockets, os, strutils, parseutils, times, sets, options import nativesockets, os, strutils, parseutils, times, sets, options,
std/monotimes
export Port, `$`, `==` export Port, `$`, `==`
export Domain, SockType, Protocol export Domain, SockType, Protocol
@ -1077,7 +1078,7 @@ proc recv*(socket: Socket, data: pointer, size: int): int {.tags: [ReadIOEffect]
# Save the error in case it gets reset. # Save the error in case it gets reset.
socket.lastError = osLastError() socket.lastError = osLastError()
proc waitFor(socket: Socket, waited: var float, timeout, size: int, proc waitFor(socket: Socket, waited: var Duration, timeout, size: int,
funcName: string): int {.tags: [TimeEffect].} = funcName: string): int {.tags: [TimeEffect].} =
## determines the amount of characters that can be read. Result will never ## determines the amount of characters that can be read. Result will never
## be larger than ``size``. For unbuffered sockets this will be ``1``. ## be larger than ``size``. For unbuffered sockets this will be ``1``.
@ -1092,7 +1093,7 @@ proc waitFor(socket: Socket, waited: var float, timeout, size: int,
result = socket.bufLen - socket.currPos result = socket.bufLen - socket.currPos
result = min(result, size) result = min(result, size)
else: else:
if timeout - int(waited * 1000.0) < 1: if timeout - waited.inMilliseconds < 1:
raise newException(TimeoutError, "Call to '" & funcName & "' timed out.") raise newException(TimeoutError, "Call to '" & funcName & "' timed out.")
when defineSsl: when defineSsl:
@ -1104,17 +1105,17 @@ proc waitFor(socket: Socket, waited: var float, timeout, size: int,
if sslPending != 0: if sslPending != 0:
return min(sslPending, size) return min(sslPending, size)
var startTime = epochTime() var startTime = getMonoTime()
let selRet = select(socket, timeout - int(waited * 1000.0)) let selRet = select(socket, (timeout - waited.inMilliseconds).int)
if selRet < 0: raiseOSError(osLastError()) if selRet < 0: raiseOSError(osLastError())
if selRet != 1: if selRet != 1:
raise newException(TimeoutError, "Call to '" & funcName & "' timed out.") raise newException(TimeoutError, "Call to '" & funcName & "' timed out.")
waited += (epochTime() - startTime) waited += (getMonoTIme() - startTime)
proc recv*(socket: Socket, data: pointer, size: int, timeout: int): int {. proc recv*(socket: Socket, data: pointer, size: int, timeout: int): int {.
tags: [ReadIOEffect, TimeEffect].} = tags: [ReadIOEffect, TimeEffect].} =
## overload with a ``timeout`` parameter in milliseconds. ## overload with a ``timeout`` parameter in milliseconds.
var waited = 0.0 # number of seconds already waited var waited: Duration # duration already waited
var read = 0 var read = 0
while read < size: while read < size:
@ -1223,7 +1224,7 @@ proc readLine*(socket: Socket, line: var TaintedString, timeout = -1,
if flags.isDisconnectionError(lastError): setLen(line.string, 0); return if flags.isDisconnectionError(lastError): setLen(line.string, 0); return
socket.socketError(n, lastError = lastError) socket.socketError(n, lastError = lastError)
var waited = 0.0 var waited: Duration
setLen(line.string, 0) setLen(line.string, 0)
while true: while true:
@ -1307,7 +1308,7 @@ proc skip*(socket: Socket, size: int, timeout = -1) =
## bytes takes longer than specified a TimeoutError exception will be raised. ## bytes takes longer than specified a TimeoutError exception will be raised.
## ##
## Returns the number of skipped bytes. ## Returns the number of skipped bytes.
var waited = 0.0 var waited: Duration
var dummy = alloc(size) var dummy = alloc(size)
var bytesSkipped = 0 var bytesSkipped = 0
while bytesSkipped != size: while bytesSkipped != size: