Keras Tensorflow - 从多线程预测的异常

Question

我使用的是张量流1.3.0后端的keras 2.0.8。

我在类init中加载模型，然后用它来预测多线程。

import tensorflow as tf 
from keras import backend as K 
from keras.models import load_model 


class CNN: 
    def __init__(self, model_path): 
     self.cnn_model = load_model(model_path) 
     self.session = K.get_session() 
     self.graph = tf.get_default_graph() 

    def query_cnn(self, data): 
     X = self.preproccesing(data) 
     with self.session.as_default(): 
      with self.graph.as_default(): 
       return self.cnn_model.predict(X) 

我初始化CNN一次和query_cnn方法从多个线程会发生。

的例外，我在我的日志得到的是：

File "/home/*/Similarity/CNN.py", line 43, in query_cnn 
    return self.cnn_model.predict(X) 
    File "/usr/local/lib/python3.5/dist-packages/keras/models.py", line 913, in predict 
    return self.model.predict(x, batch_size=batch_size, verbose=verbose) 
    File "/usr/local/lib/python3.5/dist-packages/keras/engine/training.py", line 1713, in predict 
    verbose=verbose, steps=steps) 
    File "/usr/local/lib/python3.5/dist-packages/keras/engine/training.py", line 1269, in _predict_loop 
    batch_outs = f(ins_batch) 
    File "/usr/local/lib/python3.5/dist-packages/keras/backend/tensorflow_backend.py", line 2273, in __call__ 
    **self.session_kwargs) 
    File "/usr/local/lib/python3.5/dist-packages/tensorflow/python/client/session.py", line 895, in run 
    run_metadata_ptr) 
    File "/usr/local/lib/python3.5/dist-packages/tensorflow/python/client/session.py", line 1124, in _run 
    feed_dict_tensor, options, run_metadata) 
    File "/usr/local/lib/python3.5/dist-packages/tensorflow/python/client/session.py", line 1321, in _do_run 
    options, run_metadata) 
    File "/usr/local/lib/python3.5/dist-packages/tensorflow/python/client/session.py", line 1340, in _do_call 
    raise type(e)(node_def, op, message) 
tensorflow.python.framework.errors_impl.NotFoundError: PruneForTargets: Some target nodes not found: group_deps 

的代码工作正常，大多数时候，它可能是一些问题与多线程。

我该如何解决？

Answer 1

确保在创建其他线程之前完成图形创建。

在图上调用finalize()可能会对您有所帮助。

def __init__(self, model_path): 
     self.cnn_model = load_model(model_path) 
     self.session = K.get_session() 
     self.graph = tf.get_default_graph() 
     self.graph.finalize() 

更新1：finalize()会让你的曲线只读的，因此它可以在多个线程安全地使用。作为一个副作用，它会帮助你找到无意的行为，有时候会发生内存泄漏，因为当你试图修改图时它会抛出异常。

想象一下，您有一个线程可以处理输入的一个热门编码。 （坏榜样:)

def preprocessing(self, data): 
    one_hot_data = tf.one_hot(data, depth=self.num_classes) 
    return self.session.run(one_hot_data) 

如果您打印的图形对象的数量，你会发现，它会随着时间而增加

# amount of nodes in tf graph 
print(len(list(tf.get_default_graph().as_graph_def().node))) 

但如果你先定义图表，将不会的情况下（稍好代码）：

def preprocessing(self, data): 
    # run pre-created operation with self.input as placeholder 
    return self.session.run(self.one_hot_data, feed_dict={self.input: data}) 

更新2：根据这一thread你需要调用model._make_predict_function()在做多线程之前在keras模型上。

Keras builds the GPU function the first time you call predict(). That way, if you never call predict, you save some time and resources. However, the first time you call predict is slightly slower than every other time.

更新的代码：

def __init__(self, model_path): 
    self.cnn_model = load_model(model_path) 
    self.cnn_model._make_predict_function() # have to initialize before threading 
    self.session = K.get_session() 
    self.graph = tf.get_default_graph() 
    self.graph.finalize() # make graph read-only 

更新3：我做的热身概念的证明，因为_make_predict_function()似乎并不如预期的工作。 首先，我创建了一个假人模型：

import tensorflow as tf 
from keras.layers import * 
from keras.models import * 

model = Sequential() 
model.add(Dense(256, input_shape=(2,))) 
model.add(Dense(1, activation='softmax')) 

model.compile(loss='mean_squared_error', optimizer='adam') 

model.save("dummymodel") 

然后在另一个脚本中，我加载的模型，并使其在多线程

import tensorflow as tf 
from keras import backend as K 
from keras.models import load_model 
import threading as t 
import numpy as np 

K.clear_session() 

class CNN: 
    def __init__(self, model_path): 

     self.cnn_model = load_model(model_path) 
     self.cnn_model.predict(np.array([[0,0]])) # warmup 
     self.session = K.get_session() 
     self.graph = tf.get_default_graph() 
     self.graph.finalize() # finalize 

    def preproccesing(self, data): 
     # dummy 
     return data 

    def query_cnn(self, data): 
     X = self.preproccesing(data) 
     with self.session.as_default(): 
      with self.graph.as_default(): 
       prediction = self.cnn_model.predict(X) 
     print(prediction) 
     return prediction 


cnn = CNN("dummymodel") 

th = t.Thread(target=cnn.query_cnn, kwargs={"data": np.random.random((500, 2))}) 
th2 = t.Thread(target=cnn.query_cnn, kwargs={"data": np.random.random((500, 2))}) 
th3 = t.Thread(target=cnn.query_cnn, kwargs={"data": np.random.random((500, 2))}) 
th4 = t.Thread(target=cnn.query_cnn, kwargs={"data": np.random.random((500, 2))}) 
th5 = t.Thread(target=cnn.query_cnn, kwargs={"data": np.random.random((500, 2))}) 
th.start() 
th2.start() 
th3.start() 
th4.start() 
th5.start() 

th2.join() 
th.join() 
th3.join() 
th5.join() 
th4.join() 

谈到该行的warmingup运行，并最终确定我能够重现你的第一个问题