1515
1616import io
1717import logging
18+ import random
1819import threading
19-
20+ import time
2021from collections import defaultdict
2122
2223from horovod .runner .common .util import safe_shell_exec
2324from horovod .runner .elastic .worker import HostUpdateResult
2425
26+ # The default lower bound for cooldown period. If a range is provided,
27+ # the provided lower limit must be at or above this lower bound
28+ DEFAULT_COOLDOWN_LOWER_LIMIT_SECONDS = 1
29+ # The default upper bound for cooldown period. If a range is provided,
30+ # the provided upper limit must be at or below this upper bound
31+ DEFAULT_COOLDOWN_UPPER_LIMIT_SECONDS = 1 * 60 * 60
2532
2633class HostState (object ):
27- def __init__ (self ):
34+
35+ def __init__ (self , cooldown_range = None ):
2836 self ._event = threading .Event ()
2937
30- # TODO(travis): blacklisted hosts should have a timeout period that increases with each failure
3138 self ._blacklisted = False
39+ self ._blacklist_count = 0
40+ if cooldown_range :
41+ HostState ._validate_cooldown_range (cooldown_range )
42+ self ._cooldown_lower_limit , self ._cooldown_upper_limit = cooldown_range
43+ else :
44+ self ._cooldown_lower_limit = - 1
45+ self ._cooldown_upper_limit = - 1
46+ self ._cooldown_period_end_ts = 0
47+
48+ @staticmethod
49+ def _validate_cooldown_range (cooldown_range ):
50+ cooldown_lower_limit , cooldown_upper_limit = cooldown_range
51+
52+ if (cooldown_lower_limit < DEFAULT_COOLDOWN_LOWER_LIMIT_SECONDS ):
53+ raise ValueError (f"Provided cooldown lower limit: { cooldown_lower_limit } \
54+ { DEFAULT_COOLDOWN_LOWER_LIMIT_SECONDS }
55+
56+
57+ if (cooldown_upper_limit > DEFAULT_COOLDOWN_UPPER_LIMIT_SECONDS ):
58+ raise ValueError (f"Provided cooldown upper limit: { cooldown_upper_limit } \
59+ { DEFAULT_COOLDOWN_UPPER_LIMIT_SECONDS }
3260
3361 def get_event (self ):
3462 if self ._event .is_set ():
@@ -39,13 +67,48 @@ def get_event(self):
3967 def set_event (self ):
4068 self ._event .set ()
4169
70+ def _in_cooldown_period (self , current_time ):
71+ return self ._cooldown_period_end_ts > current_time
72+
73+
74+ def _set_cooldown_period (self , current_time ):
75+ if self ._cooldown_lower_limit == - 1 or self ._cooldown_upper_limit == - 1 :
76+ return
77+ self ._blacklist_count += 1
78+
79+ cooldown_delay = self ._cooldown_lower_limit * (1 << self ._blacklist_count ) + (random .uniform (0 ,1 ) * self ._cooldown_lower_limit )
80+ logging .debug (f"{ self ._blacklist_count } { self ._cooldown_period_end_ts } { cooldown_delay } )
81+ # We need to ensure that the cooldown upper limit is the upper bound of the delay
82+ cooldown_delta_seconds = max (self ._cooldown_lower_limit , min (self ._cooldown_upper_limit , cooldown_delay ))
83+
84+ self ._cooldown_period_end_ts = current_time + cooldown_delta_seconds
85+ logging .debug (f"cooldown delta seconds: { cooldown_delta_seconds } )
86+
4287 def blacklist (self ):
88+ """Moves this host to a blacklist, and starts the cooldown period."""
4389 self ._blacklisted = True
90+ now = time .time ()
91+ if self ._in_cooldown_period (now ):
92+ return
93+ self ._set_cooldown_period (now )
4494 self .set_event ()
4595
96+ def whitelist (self ):
97+ """Ends the cooldown period and moves this host out of blacklist."""
98+ self ._cooldown_period_end_ts = 0
99+ self ._blacklisted = False
100+
46101 def is_blacklisted (self ):
102+ """Checks if the host is in the blacklist."""
47103 return self ._blacklisted
48104
105+ def is_resurrected (self ):
106+ """Checks if host is in an expired cooldown period."""
107+ if self ._cooldown_period_end_ts > 0 :
108+ return not self ._in_cooldown_period (time .time ())
109+ return False
110+
111+
49112
50113class DiscoveredHosts (object ):
51114 def __init__ (self , host_slots , host_assignment_order ):
@@ -76,15 +139,17 @@ def update(self, hosts_state):
76139 if not hosts_state [host ].is_blacklisted ()]
77140 return self
78141
142+ def __str__ (self ):
143+ return f"slots: { self ._host_slots } { self ._host_assignment_order }
144+
79145
80146class HostManager (object ):
81- def __init__ (self , discovery ):
147+ def __init__ (self , discovery , cooldown_range = None ):
82148 self ._current_hosts = DiscoveredHosts (host_slots = {}, host_assignment_order = [])
83- self ._hosts_state = defaultdict (HostState )
149+ self ._hosts_state = defaultdict (lambda : HostState ( cooldown_range ) )
84150 self ._discovery = discovery
85151
86152 def update_available_hosts (self ):
87- # TODO(travis): also check for hosts removed from the blacklist in the future
88153 def check_update (cur_host_slots , prev_host_slots ):
89154 res = HostUpdateResult .no_update
90155
@@ -103,17 +168,32 @@ def check_update(cur_host_slots, prev_host_slots):
103168 elif cur_host_slots [h ] < prev_host_slots [h ]:
104169 # h has removed some slots
105170 res |= HostUpdateResult .removed
171+ elif self ._hosts_state [h ].is_resurrected ():
172+ res |= HostUpdateResult .added
106173 return res
107174
108175 prev_host_slots = self ._current_hosts .host_slots
109176 prev_host_assignment_order = self ._current_hosts .host_assignment_order
110177 host_slots = self ._discovery .find_available_hosts_and_slots ()
111- if prev_host_slots != host_slots :
112- available_hosts = set ([host for host in host_slots .keys () if not self ._hosts_state [host ].is_blacklisted ()])
178+
179+ def whitelist_all_hosts ():
180+ for host in host_slots .keys ():
181+ if self ._hosts_state [host ].is_resurrected ():
182+ self ._hosts_state [host ].whitelist ()
183+
184+ def has_resurrected_hosts ():
185+ resurrected_hosts = [host for host in host_slots .keys () if self ._hosts_state [host ].is_resurrected ()]
186+ return len (resurrected_hosts ) > 0
187+
188+ if prev_host_slots != host_slots or has_resurrected_hosts ():
189+ available_hosts = set ([host for host in host_slots .keys () \
190+ if not (self ._hosts_state [host ].is_blacklisted () and not self ._hosts_state [host ].is_resurrected ())])
113191 host_assignment_order = HostManager .order_available_hosts (available_hosts , prev_host_assignment_order )
114192 self ._current_hosts = DiscoveredHosts (host_slots = host_slots ,
115193 host_assignment_order = host_assignment_order )
116- return check_update (self ._current_hosts .host_slots , prev_host_slots )
194+ host_update_state = check_update (self ._current_hosts .host_slots , prev_host_slots )
195+ whitelist_all_hosts ()
196+ return host_update_state
117197 else :
118198 return HostUpdateResult .no_update
119199
@@ -123,7 +203,7 @@ def current_hosts(self):
123203
124204 def blacklist (self , host ):
125205 if not self ._hosts_state [host ].is_blacklisted ():
126- logging .warning ('blacklist failing host: {}' .format (host ))
206+ logging .info ('blacklist failing host: {}' .format (host ))
127207 self ._hosts_state [host ].blacklist ()
128208
129209 def is_blacklisted (self , host ):
0 commit comments