Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __locked_change_state(self, state, services):
"""the critical section of __change_state"""
if self._should_display_state(state):
pgctl_print(
state.strings.changing,
commafy(_services_to_names(services)),
)
services = [state(service) for service in services]
failed = []
start_time = now()
while services:
for service in services:
try:
service.change()
except Unsupervised:
pass # handled in state assertion, below
for service in tuple(services):
state_change_result = self.__locked_handle_service_change_state(
state,
service,
start_time,
def __change_state(self, state, services):
"""Changes the state of a supervised service using the svc command"""
with self.playground_locked():
for service in services:
try:
state(service).assert_()
except PgctlUserMessage:
break
else:
# Short-circuit, everything is in the correct state.
if self._should_display_state(state):
pgctl_print('Already {} {}'.format(
state.strings.changed.lower(),
commafy(_services_to_names(services))),
)
return []
# If we're starting a service, run the playground-wide "pre-start" hook (if it exists).
# This is intentionally done without holding a lock, since this might be very slow.
if state is Start:
self._run_playground_wide_hook('pre-start')
run_post_stop_hook = False
with self.playground_locked():
failures = self.__locked_change_state(state, services)
if state is Stop:
run_post_stop_hook = all(
service.state['state'] == 'down'
for service in self.all_services
)
def debug(self):
"""Allow a service to run in the foreground"""
try:
# start supervise in the foreground with the service up
service, = self.services
except ValueError:
raise PgctlUserMessage(
'Must debug exactly one service, not: ' + commafy(self.service_names),
)
if service.state['state'] != 'down':
self.stop()
self._run_playground_wide_hook('pre-start')
service.foreground() # never returns