Skip to content

Bracket manager#

SHBracketManager(n_configs, budgets, bracket_id=None) #

Bases: object

Synchronous Successive Halving utilities

Source code in src/dehb/utils/bracket_manager.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
def __init__(self, n_configs, budgets, bracket_id=None):
    assert len(n_configs) == len(budgets)
    self.n_configs = n_configs
    self.budgets = budgets
    self.bracket_id = bracket_id
    self.sh_bracket = {}
    self._sh_bracket = {}
    self._config_map = {}
    for i, budget in enumerate(budgets):
        # sh_bracket keeps track of jobs/configs that are still to be scheduled/allocatted
        # _sh_bracket keeps track of jobs/configs that have been run and results retrieved for
        # (sh_bracket[i] + _sh_bracket[i]) == n_configs[i] is when no jobs have been scheduled
        #   or all jobs for that budget/rung are over
        # (sh_bracket[i] + _sh_bracket[i]) < n_configs[i] indicates a job has been scheduled
        #   and is queued/running and the bracket needs to be paused till results are retrieved
        self.sh_bracket[budget] = n_configs[i]  # each scheduled job does -= 1
        self._sh_bracket[budget] = 0  # each retrieved job does +=1
    self.n_rungs = len(budgets)
    self.current_rung = 0

get_budget(rung=None) #

Returns the exact budget that rung is pointing to.

Returns current rung's budget if no rung is passed.

Source code in src/dehb/utils/bracket_manager.py 
27
28
29
30
31
32
33
34
def get_budget(self, rung=None):
    """ Returns the exact budget that rung is pointing to.

    Returns current rung's budget if no rung is passed.
    """
    if rung is not None:
        return self.budgets[rung]
    return self.budgets[self.current_rung]

get_lower_budget_promotions(budget) #

Returns the immediate lower budget and the number of configs to be promoted from there

Source code in src/dehb/utils/bracket_manager.py 
36
37
38
39
40
41
42
43
44
def get_lower_budget_promotions(self, budget):
    """ Returns the immediate lower budget and the number of configs to be promoted from there
    """
    assert budget in self.budgets
    rung = np.where(budget == self.budgets)[0][0]
    prev_rung = np.clip(rung - 1, a_min=0, a_max=self.n_rungs-1)
    lower_budget = self.budgets[prev_rung]
    num_promote_configs = self.n_configs[rung]
    return lower_budget, num_promote_configs

get_next_job_budget() #

Returns the budget that will be selected if current_rung is incremented by 1

Source code in src/dehb/utils/bracket_manager.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def get_next_job_budget(self):
    """ Returns the budget that will be selected if current_rung is incremented by 1
    """
    if self.sh_bracket[self.get_budget()] > 0:
        # the current rung still has unallocated jobs (>0)
        return self.get_budget()
    else:
        # the current rung has no more jobs to allocate, increment it
        rung = (self.current_rung + 1) % self.n_rungs
        if self.sh_bracket[self.get_budget(rung)] > 0:
            # the incremented rung has unallocated jobs (>0)
            return self.get_budget(rung)
        else:
            # all jobs for this bracket has been allocated/bracket is complete
            # no more budgets to evaluate and can return None
            pass
        return None

register_job(budget) #

Registers the allocation of a configuration for the budget and updates current rung

This function must be called when scheduling a job in order to allow the bracket manager to continue job and budget allocation without waiting for jobs to finish and return results necessarily. This feature can be leveraged to run brackets asynchronously.

Source code in src/dehb/utils/bracket_manager.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
def register_job(self, budget):
    """ Registers the allocation of a configuration for the budget and updates current rung

    This function must be called when scheduling a job in order to allow the bracket manager
    to continue job and budget allocation without waiting for jobs to finish and return
    results necessarily. This feature can be leveraged to run brackets asynchronously.
    """
    assert budget in self.budgets
    assert self.sh_bracket[budget] > 0
    self.sh_bracket[budget] -= 1
    if not self._is_rung_pending(self.current_rung):
        # increment current rung if no jobs left in the rung
        self.current_rung = (self.current_rung + 1) % self.n_rungs

complete_job(budget) #

Notifies the bracket that a job for a budget has been completed

This function must be called when a config for a budget has finished evaluation to inform the Bracket Manager that no job needs to be waited for and the next rung can begin for the synchronous Successive Halving case.

Source code in src/dehb/utils/bracket_manager.py 
78
79
80
81
82
83
84
85
86
87
88
def complete_job(self, budget):
    """ Notifies the bracket that a job for a budget has been completed

    This function must be called when a config for a budget has finished evaluation to inform
    the Bracket Manager that no job needs to be waited for and the next rung can begin for the
    synchronous Successive Halving case.
    """
    assert budget in self.budgets
    _max_configs = self.n_configs[list(self.budgets).index(budget)]
    assert self._sh_bracket[budget] < _max_configs
    self._sh_bracket[budget] += 1

previous_rung_waits() #

Returns True if none of the rungs < current rung is waiting for results

Source code in src/dehb/utils/bracket_manager.py 
105
106
107
108
109
110
111
def previous_rung_waits(self):
    """ Returns True if none of the rungs < current rung is waiting for results
    """
    for rung in range(self.current_rung):
        if self._is_rung_waiting(rung) and not self._is_rung_pending(rung):
            return True
    return False

is_bracket_done() #

Returns True if all configs in all rungs in the bracket have been allocated

Source code in src/dehb/utils/bracket_manager.py 
113
114
115
116
def is_bracket_done(self):
    """ Returns True if all configs in all rungs in the bracket have been allocated
    """
    return ~self.is_pending() and ~self.is_waiting()

is_pending() #

Returns True if any of the rungs/budgets have still a configuration to submit

Source code in src/dehb/utils/bracket_manager.py 
118
119
120
121
def is_pending(self):
    """ Returns True if any of the rungs/budgets have still a configuration to submit
    """
    return np.any([self._is_rung_pending(i) > 0 for i, _ in enumerate(self.budgets)])

is_waiting() #

Returns True if any of the rungs/budgets have a configuration pending/running

Source code in src/dehb/utils/bracket_manager.py 
123
124
125
126
def is_waiting(self):
    """ Returns True if any of the rungs/budgets have a configuration pending/running
    """
    return np.any([self._is_rung_waiting(i) > 0 for i, _ in enumerate(self.budgets)])