This is an archive of the discontinued Mercurial Phabricator instance.

Differential D1564

worker: make windows workers daemons
ClosedPublic

Authored by wlis on Nov 30 2017, 2:26 PM.

Details

Reviewers
ikostia
durin42
Group Reviewers
hg-reviewers
Commits
rHG86b8cc1f244e: worker: make windows workers daemons
rHG487e7e6e89a0: worker: make windows workers daemons
Summary

The windows workers weren't daemons and were not correctly killed when ctrl-c'd from the terminal. Withi this change when the main thread is killed, all daemons get killed as well.
I also reduced the time we give to workers to cleanup nicely to not have people ctrl-c'ing when they get inpatient.

The output when threads clened up nicely:

PS C:\<dir>> hg.exe sparse --disable-profile SparseProfiles/<profile>.sparse
interrupted!

The output when threads don't clenup in 1 sec:

PS C:\<dir> hg.exe sparse --enable-profile SparseProfiles/<profile>.sparse
failed to kill worker threads while handling an exception
interrupted!
Exception in thread Thread-4 (most likely raised during interpreter shutdown):
PS C:\<dir>>
Test Plan

Run hg command on windows (pull/update/sparse). Ctrl-C'd sparse --enable-profile command that was using threads and observed in proces explorer that all threads got killed.
ran tests on CentOS

Diff Detail

Repository
rHG Mercurial
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Event Timeline

wlis created this revision.Nov 30 2017, 2:26 PM
ikostia accepted this revision.Nov 30 2017, 4:10 PM
ikostia added a subscriber: ikostia.

This *looks* reasonable to me, but I am afraid I could be missing something important. Anyway, since we can turn it off via a config, I'd say it's good to be in the hotfixes.

durham added a subscriber: durham.Nov 30 2017, 4:29 PM
durham added inline comments.
mercurial/worker.py
286

Why only do it on keyboard interrupt? What if there's another exception? If you did it for all exceptions, you could drop the trykillworkers() inside the loop, and just throw the exception up to here.

wlis added inline comments.Nov 30 2017, 4:39 PM
mercurial/worker.py
286

this is a very good point. will do.

wlis updated this revision to Diff 4024.Nov 30 2017, 7:02 PM
ikostia accepted this revision.Dec 1 2017, 11:11 AM
durin42 added a subscriber: durin42.Dec 5 2017, 5:50 PM

Is this missing a dependent revision annotation?

durin42 added a comment.Dec 5 2017, 5:53 PM

(specifically on D1460?)

durin42 added a parent revision: D1460: workers: add config to enable/diable workers.Dec 11 2017, 12:23 PM
durin42 accepted this revision.

This looks good to me, but I managed to confirm that it does depend on D1460 by looking at patch metadata via hg phabread, so I've added the necessary annotation.

This revision is now accepted and ready to land.Dec 11 2017, 12:24 PM
Closed by commit rHG487e7e6e89a0: worker: make windows workers daemons (authored by wlis). · Explain WhyDec 15 2017, 12:13 PM
This revision was automatically updated to reflect the committed changes.
wlis added inline comments.Dec 18 2017, 2:04 PM
mercurial/worker.py
285

I forgot to update this diff with KeyboardInterrupt again which doesn't derive from Exception, but BaseException instead. I will try to get the 1 line patch up asap.
Currently the ctrl-c wouldn't be handled gracefully on windows

Revision Contents
Changeset List

PathPackages
Mmercurial/worker.py (52 lines)

Diff 4488

mercurial/worker.py

# worker.py - master-slave parallelism support # worker.py - master-slave parallelism support
# #
# Copyright 2013 Facebook, Inc. # Copyright 2013 Facebook, Inc.
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import errno import errno
import os import os
import signal import signal
import sys import sys
import threading import threading
import time
from .i18n import _ from .i18n import _
from . import ( from . import (
encoding, encoding,
error, error,
pycompat, pycompat,
scmutil, scmutil,
util, util,
group=None, target=None, name=None, verbose=None): group=None, target=None, name=None, verbose=None):
threading.Thread.__init__(self, group=group, target=target, threading.Thread.__init__(self, group=group, target=target,
name=name, verbose=verbose) name=name, verbose=verbose)
self._taskqueue = taskqueue self._taskqueue = taskqueue
self._resultqueue = resultqueue self._resultqueue = resultqueue
self._func = func self._func = func
self._staticargs = staticargs self._staticargs = staticargs
self._interrupted = False self._interrupted = False
self.daemon = True
self.exception = None self.exception = None
def interrupt(self): def interrupt(self):
self._interrupted = True self._interrupted = True
def run(self): def run(self):
try: try:
while not self._taskqueue.empty(): while not self._taskqueue.empty():
break break
except Exception as e: except Exception as e:
# store the exception such that the main thread can resurface # store the exception such that the main thread can resurface
# it as if the func was running without workers. # it as if the func was running without workers.
self.exception = e self.exception = e
raise raise
threads = [] threads = []
def killworkers(): def trykillworkers():
# Allow up to 1 second to clean worker threads nicely
cleanupend = time.time() + 1
for t in threads: for t in threads:
t.interrupt() t.interrupt()
for t in threads: for t in threads:
# try to let the threads handle interruption, but don't wait remainingtime = cleanupend - time.time()
# indefintely. the thread could be in infinite loop, handling t.join(remainingtime)
# a very long task or in a deadlock situation
t.join(5)
if t.is_alive(): if t.is_alive():
raise error.Abort(_('failed to join worker thread')) # pass over the workers joining failure. it is more
# important to surface the inital exception than the
# fact that one of workers may be processing a large
# task and does not get to handle the interruption.
ui.warn(_("failed to kill worker threads while "
"handling an exception\n"))
return
workers = _numworkers(ui) workers = _numworkers(ui)
resultqueue = util.queue() resultqueue = util.queue()
taskqueue = util.queue() taskqueue = util.queue()
# partition work to more pieces than workers to minimize the chance # partition work to more pieces than workers to minimize the chance
# of uneven distribution of large tasks between the workers # of uneven distribution of large tasks between the workers
for pargs in partition(args, workers * 20): for pargs in partition(args, workers * 20):
taskqueue.put(pargs) taskqueue.put(pargs)
for _i in range(workers): for _i in range(workers):
t = Worker(taskqueue, resultqueue, func, staticargs) t = Worker(taskqueue, resultqueue, func, staticargs)
threads.append(t) threads.append(t)
t.start() t.start()
try:
while len(threads) > 0: while len(threads) > 0:
while not resultqueue.empty(): while not resultqueue.empty():
yield resultqueue.get() yield resultqueue.get()
threads[0].join(0.05) threads[0].join(0.05)
finishedthreads = [_t for _t in threads if not _t.is_alive()] finishedthreads = [_t for _t in threads if not _t.is_alive()]
for t in finishedthreads: for t in finishedthreads:
if t.exception is not None: if t.exception is not None:
try:
killworkers()
except Exception:
# pass over the workers joining failure. it is more
# important to surface the inital exception than the
# fact that one of workers may be processing a large
# task and does not get to handle the interruption.
ui.warn(_("failed to kill worker threads while handling "
"an exception"))
raise t.exception raise t.exception
threads.remove(t) threads.remove(t)
except Exception: # re-raises
wlisAuthorUnsubmitted
Not Done

I forgot to update this diff with KeyboardInterrupt again which doesn't derive from Exception, but BaseException instead. I will try to get the 1 line patch up asap.
Currently the ctrl-c wouldn't be handled gracefully on windows

wlis: I forgot to update this diff with KeyboardInterrupt again which doesn't derive from Exception…
trykillworkers()
durhamUnsubmitted
Not Done

Why only do it on keyboard interrupt? What if there's another exception? If you did it for all exceptions, you could drop the trykillworkers() inside the loop, and just throw the exception up to here.

durham: Why only do it on keyboard interrupt? What if there's another exception? If you did it for…
wlisAuthorUnsubmitted
Not Done

this is a very good point. will do.

wlis: this is a very good point. will do.
raise
while not resultqueue.empty(): while not resultqueue.empty():
yield resultqueue.get() yield resultqueue.get()
if pycompat.iswindows: if pycompat.iswindows:
_platformworker = _windowsworker _platformworker = _windowsworker
else: else:
_platformworker = _posixworker _platformworker = _posixworker
_exitstatus = _posixexitstatus _exitstatus = _posixexitstatus