Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fixes to executor controller logic and timing of active status #351

Merged
merged 2 commits into from
Apr 20, 2016
Merged

Fixes to executor controller logic and timing of active status #351

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9555631
controllers: fixing long delay when using a slow status rate
jbohren Apr 19, 2016
52c58ab
controllers: fixing bug in status rate logic exposed by 95556318fe9b8…
jbohren Apr 20, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
104 changes: 62 additions & 42 deletions catkin_tools/execution/controllers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -448,6 +448,13 @@ def run(self):
active_stages = dict()

start_time = self.pre_start_time or time.time()
last_update_time = time.time()

# If the status rate is too low, just disable it
if self.active_status_rate < 1E-3:
self.show_active_status = False
else:
update_duration = 1.0 / self.active_status_rate

# Disable the wide log padding if the status is disabled
if not self.show_active_status:
Expand All @@ -458,55 +465,68 @@ def run(self):
if not self.keep_running:
wide_log(clr('[{}] An internal error occurred!').format(self.label))
return

# Write a continuously-updated status line
if self.show_active_status:

# Try to get an event from the queue (non-blocking)
try:
event = self.event_queue.get(False)
except Empty:
# Print live status (overwrites last line)
status_line = clr('[{} {} s] [{}/{} complete] [{}/{} jobs] [{} queued]').format(
self.label,
format_time_delta_short(time.time() - start_time),
len(completed_jobs),
len(self.jobs),
job_server.running_jobs(),
job_server.max_jobs(),
len(queued_jobs) + len(active_jobs) - len(active_stages)
)

# Show failed jobs
if len(failed_jobs) > 0:
status_line += clr(' [@!@{rf}{}@| @{rf}failed@|]').format(len(failed_jobs))

# Check load / mem
if not job_server.load_ok():
status_line += clr(' [@!@{rf}High Load@|]')
if not job_server.mem_ok():
status_line += clr(' [@!@{rf}Low Memory@|]')

# Add active jobs
if len(active_jobs) == 0:
status_line += clr(' @/@!@{kf}Waiting for jobs...@|')
# Determine if the status should be shown based on the desired
# status rate
elapsed_time = time.time() - last_update_time
show_status_now = elapsed_time > update_duration

if show_status_now:
# Print live status (overwrites last line)
status_line = clr('[{} {} s] [{}/{} complete] [{}/{} jobs] [{} queued]').format(
self.label,
format_time_delta_short(time.time() - start_time),
len(completed_jobs),
len(self.jobs),
job_server.running_jobs(),
job_server.max_jobs(),
len(queued_jobs) + len(active_jobs) - len(active_stages)
)

# Show failed jobs
if len(failed_jobs) > 0:
status_line += clr(' [@!@{rf}{}@| @{rf}failed@|]').format(len(failed_jobs))

# Check load / mem
if not job_server.load_ok():
status_line += clr(' [@!@{rf}High Load@|]')
if not job_server.mem_ok():
status_line += clr(' [@!@{rf}Low Memory@|]')

# Add active jobs
if len(active_jobs) == 0:
status_line += clr(' @/@!@{kf}Waiting for jobs...@|')
else:
active_labels = []

for j, (s, t, p) in active_stages.items():
d = format_time_delta_short(cumulative_times[j] + time.time() - t)
if p == '':
active_labels.append(clr('[{}:{} - {}]').format(j, s, d))
else:
active_labels.append(clr('[{}:{} ({}%) - {}]').format(j, s, p, d))

status_line += ' ' + ' '.join(active_labels)

# Print the status line
# wide_log(status_line)
wide_log(status_line, rhs='', end='\r')
sys.stdout.flush()

# Store this update time
last_update_time = time.time()
else:
active_labels = []

for j, (s, t, p) in active_stages.items():
d = format_time_delta_short(cumulative_times[j] + time.time() - t)
if p == '':
active_labels.append(clr('[{}:{} - {}]').format(j, s, d))
else:
active_labels.append(clr('[{}:{} ({}%) - {}]').format(j, s, p, d))

status_line += ' ' + ' '.join(active_labels)

# Print the status line
# wide_log(status_line)
wide_log(status_line, rhs='', end='\r')
sys.stdout.flush()
if self.active_status_rate > 1E-5:
time.sleep(1.0 / self.active_status_rate)
continue
time.sleep(max(0.0, min(update_duration - elapsed_time, 0.01)))

# Only continue when no event was received
continue
else:
# Try to get an event from the queue (blocking)
try:
Expand Down