loadscope and flake8 don't work together with one node

I'm not sure if this is a loadscope bug or a problem with the implementation of pytest-flake8 but when I try to run --flake8 --dist=loadscope -n 1 everything hangs:

$ pytest -v --dist=loadscope -n 1 --flake8 --fulltrace  tests/test_register.py 
============================= test session starts ==============================
platform darwin -- Python 3.6.0, pytest-3.2.0, py-1.4.34, pluggy-0.4.0 -- /Users/timj/work/lsstsw3/miniconda/bin/python
cachedir: .cache
rootdir: /Volumes/G-RAID with Thunderbolt/transient/lsstsw3/build/pipe_tasks, inifile: setup.cfg
plugins: session2file-0.1.9, forked-0.3.dev0+g1dd93f6.d20170815, xdist-1.19.2.dev0+g459d52e.d20170815, flake8-0.8.1
[gw0] darwin Python 3.6.0 cwd: /Volumes/G-RAID with Thunderbolt/transient/lsstsw3/build/pipe_tasks
[gw0] Python 3.6.0 |Continuum Analytics, Inc.| (default, Dec 23 2016, 13:19:00)  -- [GCC 4.2.1 Compatible Apple LLVM 6.0 (clang-600.0.57)]
gw0 [5]
scheduling tests via LoadScopeScheduling
^C
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! KeyboardInterrupt !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!

The following commands all work fine:

$ pytest -v --dist=loadscope -n 1 --fulltrace  tests/test_register.py
$ pytest -v --dist=loadscope -n 2 --flake8 --fulltrace  tests/test_register.py 
$ pytest -v  -n 1 --flake8 --fulltrace  tests/test_register.py
$ pytest -v  -n 2 --flake8 --fulltrace  tests/test_register.py

leading to the conclusion that everything hangs only when one subprocess is used and loadscope is enabled and flake8 testing is enabled.

$ pytest -v --dist=loadscope -n 2 --flake8 --fulltrace  tests/test_register.py 
============================= test session starts ==============================
platform darwin -- Python 3.6.0, pytest-3.2.0, py-1.4.34, pluggy-0.4.0 -- /Users/timj/work/lsstsw3/miniconda/bin/python
cachedir: .cache
rootdir: /Volumes/G-RAID with Thunderbolt/transient/lsstsw3/build/pipe_tasks, inifile: setup.cfg
plugins: session2file-0.1.9, forked-0.3.dev0+g1dd93f6.d20170815, xdist-1.19.2.dev0+g459d52e.d20170815, flake8-0.8.1
[gw0] darwin Python 3.6.0 cwd: /Volumes/G-RAID with Thunderbolt/transient/lsstsw3/build/pipe_tasks
[gw1] darwin Python 3.6.0 cwd: /Volumes/G-RAID with Thunderbolt/transient/lsstsw3/build/pipe_tasks
[gw0] Python 3.6.0 |Continuum Analytics, Inc.| (default, Dec 23 2016, 13:19:00)  -- [GCC 4.2.1 Compatible Apple LLVM 6.0 (clang-600.0.57)]
[gw1] Python 3.6.0 |Continuum Analytics, Inc.| (default, Dec 23 2016, 13:19:00)  -- [GCC 4.2.1 Compatible Apple LLVM 6.0 (clang-600.0.57)]
gw0 [5] / gw1 [5]
scheduling tests via LoadScopeScheduling

tests/test_register.py::RegisterTestCase::testRegister 
[gw1] PASSED tests/test_register.py::RegisterTestCase::testRegister 
tests/test_register.py::RegisterTestCase::testRejection 
[gw1] PASSED tests/test_register.py::RegisterTestCase::testRejection 
tests/test_register.py::MyMemoryTestCase::testFileDescriptorLeaks <- ../../../../../../Users/timj/work/lsstsw3/stack/DarwinX86/utils/13.0-9-gf29e843+2/python/lsst/utils/tests.py 
[gw1] PASSED tests/test_register.py::MyMemoryTestCase::testFileDescriptorLeaks <- ../../../../../../Users/timj/work/lsstsw3/stack/DarwinX86/utils/13.0-9-gf29e843+2/python/lsst/utils/tests.py 
tests/test_register.py::MyMemoryTestCase::testLeaks <- ../../../../../../Users/timj/work/lsstsw3/stack/DarwinX86/utils/13.0-9-gf29e843+2/python/lsst/utils/tests.py 
tests/test_register.py 
[gw1] PASSED tests/test_register.py::MyMemoryTestCase::testLeaks <- ../../../../../../Users/timj/work/lsstsw3/stack/DarwinX86/utils/13.0-9-gf29e843+2/python/lsst/utils/tests.py 
[gw0] FAILED tests/test_register.py 

(the failure is simply that this particular file has a flake8 issue).

I am wondering if the pytest-flake8 plugin is not correctly returning scoping information to the scheduler in a similar way to it not working properly with pytest-randomly (tholo/pytest-flake8#26), even so, how come -n 2 is fine?

When it hangs this is the stack trace:

scheduling tests via LoadScopeScheduling
^C
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! KeyboardInterrupt !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!

config = <_pytest.config.Config object at 0x10ae57b00>
doit = <function _main at 0x10ae2f378>

    def wrap_session(config, doit):
        """Skeleton command line program"""
        session = Session(config)
        session.exitstatus = EXIT_OK
        initstate = 0
        try:
            try:
                config._do_configure()
                initstate = 1
                config.hook.pytest_sessionstart(session=session)
                initstate = 2
>               session.exitstatus = doit(config, session) or 0

../../stack/DarwinX86/pytest/3.2.0/lib/python/pytest-3.2.0-py3.6.egg/_pytest/main.py:110: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

config = <_pytest.config.Config object at 0x10ae57b00>
session = <Session 'pipe_tasks'>

    def _main(config, session):
        """ default command line protocol for initialization, session,
        running tests and reporting. """
        config.hook.pytest_collection(session=session)
>       config.hook.pytest_runtestloop(session=session)

../../stack/DarwinX86/pytest/3.2.0/lib/python/pytest-3.2.0-py3.6.egg/_pytest/main.py:146: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <_HookCaller 'pytest_runtestloop'>
kwargs = {'__multicall__': <_MultiCall 0 results, 1 meths, kwargs={'session': <Session 'pipe_tasks'>, '__multicall__': <_MultiCall 0 results, 1 meths, kwargs={...}>}>, 'session': <Session 'pipe_tasks'>}

    def __call__(self, **kwargs):
        assert not self.is_historic()
>       return self._hookexec(self, self._nonwrappers + self._wrappers, kwargs)

../../stack/DarwinX86/pytest/3.2.0/lib/python/pytest-3.2.0-py3.6.egg/_pytest/vendored_packages/pluggy.py:745: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <_pytest.config.PytestPluginManager object at 0x10ac03d68>
hook = <_HookCaller 'pytest_runtestloop'>
methods = [<_pytest.vendored_packages.pluggy.HookImpl object at 0x10ae666d8>]
kwargs = {'__multicall__': <_MultiCall 0 results, 1 meths, kwargs={'session': <Session 'pipe_tasks'>, '__multicall__': <_MultiCall 0 results, 1 meths, kwargs={...}>}>, 'session': <Session 'pipe_tasks'>}

    def _hookexec(self, hook, methods, kwargs):
        # called from all hookcaller instances.
        # enable_tracing will set its own wrapping function at self._inner_hookexec
>       return self._inner_hookexec(hook, methods, kwargs)

../../stack/DarwinX86/pytest/3.2.0/lib/python/pytest-3.2.0-py3.6.egg/_pytest/vendored_packages/pluggy.py:339: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

hook = <_HookCaller 'pytest_runtestloop'>
methods = [<_pytest.vendored_packages.pluggy.HookImpl object at 0x10ae666d8>]
kwargs = {'__multicall__': <_MultiCall 0 results, 1 meths, kwargs={'session': <Session 'pipe_tasks'>, '__multicall__': <_MultiCall 0 results, 1 meths, kwargs={...}>}>, 'session': <Session 'pipe_tasks'>}

    self._inner_hookexec = lambda hook, methods, kwargs: \
>       _MultiCall(methods, kwargs, hook.spec_opts).execute()

../../stack/DarwinX86/pytest/3.2.0/lib/python/pytest-3.2.0-py3.6.egg/_pytest/vendored_packages/pluggy.py:334: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <_MultiCall 0 results, 1 meths, kwargs={'session': <Session 'pipe_tasks'>, '__multicall__': <_MultiCall 0 results, 1 meths, kwargs={...}>}>

    def execute(self):
        all_kwargs = self.kwargs
        self.results = results = []
        firstresult = self.specopts.get("firstresult")
    
        while self.hook_impls:
            hook_impl = self.hook_impls.pop()
            try:
                args = [all_kwargs[argname] for argname in hook_impl.argnames]
            except KeyError:
                for argname in hook_impl.argnames:
                    if argname not in all_kwargs:
                        raise HookCallError(
                            "hook call must provide argument %r" % (argname,))
            if hook_impl.hookwrapper:
                return _wrapped_call(hook_impl.function(*args), self.execute)
>           res = hook_impl.function(*args)

../../stack/DarwinX86/pytest/3.2.0/lib/python/pytest-3.2.0-py3.6.egg/_pytest/vendored_packages/pluggy.py:614: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <xdist.dsession.DSession object at 0x10b21a1d0>

    def pytest_runtestloop(self):
        self.sched = self.config.hook.pytest_xdist_make_scheduler(
            config=self.config,
            log=self.log
        )
        assert self.sched is not None
    
        self.shouldstop = False
        while not self.session_finished:
>           self.loop_once()

/Users/timj/work/lsstsw3/stack/DarwinX86/pytest_xdist/1.19.1/lib/python/pytest_xdist-1.19.2.dev0+g459d52e.d20170815-py3.6.egg/xdist/dsession.py:114: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <xdist.dsession.DSession object at 0x10b21a1d0>

    def loop_once(self):
        """Process one callback from one of the slaves."""
        while 1:
            try:
>               eventcall = self.queue.get(timeout=2.0)

/Users/timj/work/lsstsw3/stack/DarwinX86/pytest_xdist/1.19.1/lib/python/pytest_xdist-1.19.2.dev0+g459d52e.d20170815-py3.6.egg/xdist/dsession.py:124: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <queue.Queue object at 0x10b21a240>, block = True, timeout = 2.0

    def get(self, block=True, timeout=None):
        '''Remove and return an item from the queue.
    
            If optional args 'block' is true and 'timeout' is None (the default),
            block if necessary until an item is available. If 'timeout' is
            a non-negative number, it blocks at most 'timeout' seconds and raises
            the Empty exception if no item was available within that time.
            Otherwise ('block' is false), return an item if one is immediately
            available, else raise the Empty exception ('timeout' is ignored
            in that case).
            '''
        with self.not_empty:
            if not block:
                if not self._qsize():
                    raise Empty
            elif timeout is None:
                while not self._qsize():
                    self.not_empty.wait()
            elif timeout < 0:
                raise ValueError("'timeout' must be a non-negative number")
            else:
                endtime = time() + timeout
                while not self._qsize():
                    remaining = endtime - time()
                    if remaining <= 0.0:
                        raise Empty
>                   self.not_empty.wait(remaining)

/Users/timj/work/lsstsw3/miniconda/lib/python3.6/queue.py:173: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Condition(<unlocked _thread.lock object at 0x10b1ac4e0>, 0)>
timeout = 1.999992159951944

    def wait(self, timeout=None):
        """Wait until notified or until a timeout occurs.
    
            If the calling thread has not acquired the lock when this method is
            called, a RuntimeError is raised.
    
            This method releases the underlying lock, and then blocks until it is
            awakened by a notify() or notify_all() call for the same condition
            variable in another thread, or until the optional timeout occurs. Once
            awakened or timed out, it re-acquires the lock and returns.
    
            When the timeout argument is present and not None, it should be a
            floating point number specifying a timeout for the operation in seconds
            (or fractions thereof).
    
            When the underlying lock is an RLock, it is not released using its
            release() method, since this may not actually unlock the lock when it
            was acquired multiple times recursively. Instead, an internal interface
            of the RLock class is used, which really unlocks it even when it has
            been recursively acquired several times. Another internal interface is
            then used to restore the recursion level when the lock is reacquired.
    
            """
        if not self._is_owned():
            raise RuntimeError("cannot wait on un-acquired lock")
        waiter = _allocate_lock()
        waiter.acquire()
        self._waiters.append(waiter)
        saved_state = self._release_save()
        gotit = False
        try:    # restore state no matter what (e.g., KeyboardInterrupt)
            if timeout is None:
                waiter.acquire()
                gotit = True
            else:
                if timeout > 0:
>                   gotit = waiter.acquire(True, timeout)
E                   KeyboardInterrupt

/Users/timj/work/lsstsw3/miniconda/lib/python3.6/threading.py:299: KeyboardInterrupt
======================== no tests ran in 65.08 seconds =========================

Reproduce the hang with tests/test_register.py using --dist=loadscope -n 1 --flake8, then compare it with the working -n 2 and non-loadscope commands shown. Trace the loadscope scheduler and pytest-flake8 interaction; done means the one-node command completes without hanging while the existing flake8 failure remains reported.