Sporadic concurrency issue leading to zero task progress

[adamklein]

Versions:

- dask=0.14.3=py35_1
- distributed=1.16.3=py35_0
- tornado=4.5.1=py35_0

I'll try to create a repro but until I can do that, I'll just describe the setup here as a placeholder.

I am running a LocalCluster with 6 processes (# cpus) and 1 thread per worker.

I am submitting tasks from tasks using worker_client and using an AsCompleted(loop=client.loop) in the worker_client context to await their completion. Occasionally I will see a secondary task complete (ie, the task-of-task has finished what it is supposed to do), but the original task (that launched it) continues to wait on its queue for notification of completion, halting progress of the program. When I interrupt the program, my stack traces are as follows:

In worker_client context / original task:

^CTraceback (most recent call last):                                   
  File "<string>", line 1, in <module>                                 
  File "lib/python3.5/multiprocessing/forkserver.py", line 164, in main
	rfds = [key.fileobj for (key, events) in selector.select()]                                                                                                                     
  File "lib/python3.5/selectors.py", line 441, in select
	fd_event_list = self._epoll.poll(timeout, max_ev)                                                           
KeyboardInterrupt                                                      
  File "...", line 248, in run
	for f in queue:                                                                                                 
  File "lib/python3.5/site-packages/distributed/client.py", line 2634, in __next__
	return self.queue.get()                                                                         
  File "lib/python3.5/queue.py", line 164, in get
	self.not_empty.wait()                                                                                    
  File "lib/python3.5/threading.py", line 293, in wait
	waiter.acquire()    

Where the secondary task was running:

tornado.application - ERROR - Exception in callback functools.partial(<function wrap.<locals>.null_wrapper at 0x7f3fbd31b730>, <tornado.concurrent.Future object at 0x7f3fbd2c9d68>)
Traceback (most recent call last):
  File "lib/python3.5/site-packages/tornado/ioloop.py", line 605, in _run_callback
	ret = callback()
  File "lib/python3.5/site-packages/tornado/stack_context.py", line 277, in null_wrapper
	return fn(*args, **kwargs)
  File "lib/python3.5/site-packages/tornado/ioloop.py", line 626, in _discard_future_result
	future.result()
  File "lib/python3.5/site-packages/tornado/concurrent.py", line 238, in result
	raise_exc_info(self._exc_info)
  File "<string>", line 4, in raise_exc_info
  File "lib/python3.5/site-packages/tornado/gen.py", line 1063, in run
	yielded = self.gen.throw(*exc_info)
  File "lib/python3.5/site-packages/distributed/client.py", line 2600, in track_future
	yield _wait(future)
  File "lib/python3.5/site-packages/tornado/gen.py", line 1055, in run
	value = future.result()
  File "lib/python3.5/site-packages/tornado/concurrent.py", line 238, in result
	raise_exc_info(self._exc_info)
  File "<string>", line 4, in raise_exc_info
  File "lib/python3.5/site-packages/tornado/gen.py", line 1069, in run
	yielded = self.gen.send(value)
  File "lib/python3.5/site-packages/distributed/client.py", line 2482, in _wait
	raise CancelledError(cancelled)
concurrent.futures._base.CancelledError: ['mytask-349a99bf-9ffa-4fc6-988e-d85345a654d6']

[adamklein]

EDIT: AsCompleted is a red herring. Locks up the same just client.gather().

This inevitably locks for me, whether after the first one or two iterations, or after several:

"""
Repro of https://github.com/dask/distributed/issues/1176
"""

from distributed import LocalCluster, worker_client
from distributed.client import Client


def task(x):
    return x


def inner_runner(x):
    with worker_client() as client:
        q = []
        for i in range(1, x+1):
            f = client.submit(task, i, pure=False)
            q.append(f)

        tot = sum(client.gather(q))

        return tot


def outer_runner(x):
    with worker_client() as client:
        q = []
        for i in range(1, x+1):
            f = client.submit(inner_runner, i, pure=False)
            q.append(f)

        tot = sum(client.gather(q))

        return tot


def run_test():
    with LocalCluster() as cluster:
        with Client(cluster) as client:
            input('Run now')
            for _ in range(100):
                futs = [client.submit(outer_runner, i, pure=False) for i in range(1, 11)]
                # this will hang eventually
                results = client.gather(futs)
                print('Results = {}'.format(results))


if __name__ == '__main__':
    run_test()
    input('done')

Output

Results = [1, 4, 10, 20, 35, 56, 84, 120, 165, 220]
Results = [1, 4, 10, 20, 35, 56, 84, 120, 165, 220]
<hang after a few times>

[adamklein]

Seems not to hang if LocalCluster(n_workers=1), whereas n_workers=2 gets it to hang for me.

[adamklein]

Some more random info. If I set inner_runner to submit without pure=False, I still get hanging but with some scary log output (many of these per outermost iteration):

distributed.scheduler - FATAL - Tried writing to closed comm: {'op': 'key-in-memory', 'key': 'task-644a381c188046862f8e700da0776916'}

[mrocklin]

First, thank you for going through the effort of the excellent reproducible example. This is perfect.

Second, when I try this locally I'm actually unable to reproduce with either pure=True or pure=False. Can I ask you to update to the newest version (just released a day ago) and see if the problem persists?

[adamklein]

I just upgraded to the latest versions below, but unfortunately the problem persists. Do you also not get the FATAL log message if you toggle pure on the inner_runner?

- dask=0.15.0=py35_0
- distributed=1.17.0=py35_0
- tornado=4.5.1=py35_0

(For me, sometimes it iterates as much as 30-35 times before locking.)

EDIT: I just tried this example at home on my (4 core) macbook air, and it does not lock up. Whereas at work I'm on a linux machine...

> cat /proc/version
Linux version 4.1.35-pv-ts2 (gcc version 4.7.2 (Debian 4.7.2-5) ) #1 SMP Mon Dec 5 21:15:47 GMT 2016

vs on macbook air

> uname -r
16.6.0

[adamklein]

Possibly related:
https://bugs.python.org/issue25680

I'm seeing hangs here:

rfds = [key.fileobj for (key, events) in selector.select()]
  File "lib/python3.5/selectors.py", line 441, in select
    fd_event_list = self._epoll.poll(timeout, max_ev)
KeyboardInterrupt

[adamklein]

I'm not sure whether this is related, but I hit this error in the same test that originally produced the deadlock I reported above:

Traceback (most recent call last):                                                                                                                                                                        
  File "mycode", line 34, in dask_client                                                                                                                  
	with worker_client(timeout=60) as client:                                                                                                                                                             
  File "lib/python3.5/contextlib.py", line 59, in __enter__                                                                                                                
	return next(self.gen)                                                                                                                                                                                 
  File "lib/python3.5/site-packages/distributed/worker_client.py", line 58, in worker_client                                                                               
	sync(wc.loop, gen.with_timeout, timedelta(seconds=timeout), wc._started)                                                                                                                              
  File "lib/python3.5/site-packages/distributed/utils.py", line 234, in sync                                                                                               
	six.reraise(*error[0])                                                                                                                                                                                
  File "lib/python3.5/site-packages/six.py", line 686, in reraise                                                                                                          
	raise value                                                                                                                                                                                           
  File "lib/python3.5/site-packages/distributed/utils.py", line 223, in f                                                                                                  
	result[0] = yield make_coro()                                                                                                                                                                         
  File "lib/python3.5/site-packages/tornado/gen.py", line 1055, in run                                                                                                     
	value = future.result()                                                                                                                                                                                                                                 
  File "lib/python3.5/site-packages/tornado/concurrent.py", line 238, in result                                                                                                                                                              
	raise_exc_info(self._exc_info)                                                                                                                                                                                                                          
  File "<string>", line 4, in raise_exc_info                                                                                                                                                                                                                
  File "lib/python3.5/site-packages/tornado/gen.py", line 1063, in run                                                                                                                                                       
	yielded = self.gen.throw(*exc_info)                                                                                                                                                                                                                     
  File "lib/python3.5/site-packages/distributed/client.py", line 636, in _start                                                                                                                                                              
	yield self._ensure_connected(timeout=timeout)                                                                                                                                                                                                           
  File "lib/python3.5/site-packages/tornado/gen.py", line 1055, in run                                                                                                                                                       
	value = future.result()                                                                                                                                                                                                                                 
  File "lib/python3.5/site-packages/tornado/concurrent.py", line 238, in result                                                                                                                                              
	raise_exc_info(self._exc_info)                                                                                                                                                                                                                          
  File "<string>", line 4, in raise_exc_info                                                                                                                                                                                                                
  File "lib/python3.5/site-packages/tornado/gen.py", line 1069, in run                                                                                                                                                       
	yielded = self.gen.send(value)                                                                                                                                                                                                                          
  File "lib/python3.5/site-packages/distributed/client.py", line 680, in _ensure_connected                                                                                                                                   
	assert len(msg) == 1                                                                                                                                                                                  
AssertionError        

During handling of the above exception, another exception occurred:                                                                
																													  
Traceback (most recent call last):                                                                                    
  File "lib/python3.5/site-packages/tornado/gen.py", line 307, in wrapper      
	yielded = next(result)                                                                                               
  File "lib/python3.5/site-packages/distributed/client.py", line 832, in _shutdown     
	self._send_to_scheduler({'op': 'close-stream'})                                                           
  File "lib/python3.5/site-packages/distributed/client.py", line 580, in _send_to_scheduler
	self.loop.add_callback(self.scheduler_comm.send, msg)                        
AttributeError: 'NoneType' object has no attribute 'send'

tornado.application - ERROR - Future <tornado.concurrent.Future object at 0x7fbfbf7aae48> exception was never retrieved: Traceback (most recent call last):                                               
    <... repeat of above Traceback ...>

Any idea what this might be?

[adamklein]

Some encouraging news on this. Trying out #1201 (which I've rebased on master to pick up latest changes as well), I edit with worker_client() as client: to instead be:

client = get_client()
# time.sleep(0.01)  # necessary?
secede()

I do not get lockups anymore in this configuration. If I don't stick in the time.sleep, I will occasionally get this and a resulting python exception (killedworker):

tornado.application - ERROR - Exception in callback functools.partial(<function wrap.<locals>.null_wrapper at 0x7fc925c68d08>, 'inner_runner-7082c80f-4cd1-4279-aced-7a15f55f467a', 'long-running', compute_duration=1.3828277587890625e-05)
Traceback (most recent call last):
  File "/lib/python3.5/site-packages/tornado/ioloop.py", line 605, in _run_callback
	ret = callback()
  File "/lib/python3.5/site-packages/tornado/stack_context.py", line 277, in null_wrapper
	return fn(*args, **kwargs)
  File "/home/aklein/src/distributed/distributed/worker.py", line 1287, in transition
	state = func(key, **kwargs)
  File "/home/aklein/src/distributed/distributed/worker.py", line 1416, in transition_executing_long_running
	'compute_duration': compute_duration})
  File "/home/aklein/src/distributed/distributed/batched.py", line 101, in send
	raise CommClosedError
distributed.comm.core.CommClosedError

Also, if I leave worker_client() as in the original example, it doesn't proceed at all. If I stick a time.sleep after that line, I get this:

tornado.application - ERROR - Exception in callback functools.partial(<function wrap.<locals>.null_wrapper at 0x7f6727f5cea0>, <tornado.concurrent.Future object at 0x7f6727f5aa90>)
Traceback (most recent call last):
  File "/lib/python3.5/site-packages/tornado/ioloop.py", line 605, in _run_callback
	ret = callback()
  File "/lib/python3.5/site-packages/tornado/stack_context.py", line 277, in null_wrapper
	return fn(*args, **kwargs)
  File "/lib/python3.5/site-packages/tornado/ioloop.py", line 626, in _discard_future_result
	future.result()
  File "/lib/python3.5/site-packages/tornado/concurrent.py", line 238, in result
	raise_exc_info(self._exc_info)
  File "<string>", line 4, in raise_exc_info
  File "/lib/python3.5/site-packages/tornado/gen.py", line 1069, in run
	yielded = self.gen.send(value)
  File "/home/aklein/src/distributed/distributed/scheduler.py", line 3161, in reevaluate_occupancy
	self._reevaluate_occupancy_worker(w)
  File "/home/aklein/src/distributed/distributed/scheduler.py", line 3181, in _reevaluate_occupancy_worker
	processing[key] = duration + comm
TypeError: unsupported operand type(s) for +: 'NoneType' and 'float'

[mrocklin]

Can you print out or return the value of get_client()._asynchronous ?

[mrocklin]

also get_client()._started.done() and get_client().status if convenient

[mrocklin]

Actually in this case

  File "/home/aklein/src/distributed/distributed/scheduler.py", line 3161, in reevaluate_occupancy
	self._reevaluate_occupancy_worker(w)
  File "/home/aklein/src/distributed/distributed/scheduler.py", line 3181, in _reevaluate_occupancy_worker
	processing[key] = duration + comm

Can you verify that you're using the most recent version of that branch? I fixed an issue very similar to that recently.

[mrocklin]

I remain concerned about the need for a sleep though

[mrocklin]

Generally the approach in #1201 helps the problem here in two ways:

It avoids creating tons of clients, and instead reuses the same client in many threads. To achieve this we made the client threadsafe (for common operations). There is no reason to setup and teardown a client each time, you just use a client that is sitting around on the worker process.

It subtly tweaks scheduling of tasks submitted by tasks to try to prefer subtasks over super-tasks of the same generation. This should stop a super-task-blowup in the common case.

[adamklein]

Using worker_client, this test no longer locks up at all. Do you still want me to dig in to any of the above? I think this is solved. (As of commit 007cbf7)

[mrocklin]

No, I think we're good