daggerml.util

View source
 1import logging
 2import time
 3from dataclasses import dataclass
 4from random import randint
 5
 6import boto3 as boto3
 7
 8from daggerml._core import get_client as get_client
 9
10logger = logging.getLogger(__name__)
11
12
13def current_time_millis():
14    return round(time.time() * 1000)
15
16
17@dataclass
18class BackoffWithJitter:
19    min: int = 100
20    max: int = 10000
21    k: int = 2
22    state: int = 0
23
24    def __call__(self):
25        if self.state == 0:
26            self.state = randint(self.min, self.min * 2)
27            return self.state
28        self.state = min(self.max, randint(self.min, max(self.min, self.state) * self.k))
29        return self.state

logger

logger= <Logger daggerml.util (WARNING)>

current_time_millis

def current_time_millis():
View source
14def current_time_millis():
15    return round(time.time() * 1000)

BackoffWithJitter

@dataclass
class BackoffWithJitter:
View source
18@dataclass
19class BackoffWithJitter:
20    min: int = 100
21    max: int = 10000
22    k: int = 2
23    state: int = 0
24
25    def __call__(self):
26        if self.state == 0:
27            self.state = randint(self.min, self.min * 2)
28            return self.state
29        self.state = min(self.max, randint(self.min, max(self.min, self.state) * self.k))
30        return self.state

BackoffWithJitter.__init__

BackoffWithJitter(min: int = 100, max: int = 10000, k: int = 2, state: int = 0)

BackoffWithJitter.min

min: int= 100

BackoffWithJitter.max

max: int= 10000

BackoffWithJitter.k

k: int= 2

BackoffWithJitter.state

state: int= 0