dask_example.py

import numpy as np
import tensorflow as tf
import keras
from keras import backend as K
from distributed import Client, client


def new_session():
    sess = tf.Session()
    return sess


def test_simulation(size, sess):
    K.set_session(sess)

    X = np.random.random((10, 5))
    y = np.random.random((10, 1))
    model_input = keras.layers.Input((5,))
    hidden = keras.layers.Dense(size)(model_input)
    model_output = keras.layers.Dense(1)(hidden)

    model = keras.models.Model(model_input, model_output)
    model.compile(optimizer='sgd', loss='mse')
    model.fit(X, y, nb_epoch=10)


if __name__ == '__main__':
    SCHEDULER_IP = 'tcp://127.0.0.1:8786'
#    SCHEDULER_IP = '169.229.219.178:8786'
    client = Client(SCHEDULER_IP)
    
    sess = client.submit(new_session)
    res_list = [client.submit(test_simulation, i, sess) for i in range(5, 10)]
#    client.wait(res_list)

You can even throw a lock around the entire contents of test_simulation and the system fails.

Any updated news on this? I am having the same issue.