SunnerLi
commented
7 years ago I got some hint from the stackoverflow.
The following split the code as the two part.
The first part is the code of kera models, and the other one is main program.
I named the model code as model.py.
The contain of model.py is shown below:
from keras.layers import Conv2D, Dense, Flatten, Input, MaxPool2D
from keras.models import Sequential
from keras import backend as K
import tensorflow as tf
import numpy as np
import threading
import os
class Has1ConvLayerCNN(object):
graph = None
model = None
model_name = None
def __init__(self):
pass
def createModelAndOp(self, model_name='small_model.h5'):
self.graph = tf.Graph()
self.model_name = model_name
with self.graph.as_default():
self.model = Sequential()
self.model.add(Conv2D(5, (2, 2), activation='relu', input_shape=(27, 27, 1)))
self.model.add(MaxPool2D())
self.model.add(Conv2D(10, (2, 2), activation='relu'))
self.model.add(MaxPool2D())
self.model.add(Conv2D(15, (2, 2), activation='relu'))
self.model.add(Conv2D(20, (2, 2), activation='relu'))
self.model.add(Flatten())
self.model.add(Dense(100, activation='relu'))
self.model.add(Dense(10, activation='sigmoid'))
self.model.compile(optimizer='sgd', loss='mse')
def train(x, y):
with tf.Session(graph=self.graph) as sess:
sess.run(tf.global_variables_initializer())
self.model.fit(x, y, verbose=0)
self.model.save_weights(self.model_name)
def test(x):
with tf.Session(graph=self.graph) as sess:
sess.run(tf.global_variables_initializer())
if os.path.exists(model_name):
self.model.load_weights(model_name)
return self.model.predict(x)
# Use closure to return the function
return train, test
class Has2ConvLayerCNN(object):
graph = None
model = None
model_name = None
def __init__(self):
pass
def createModelAndOp(self, model_name='big_model.h5'):
self.graph = tf.Graph()
self.model_name = model_name
with self.graph.as_default():
self.model = Sequential()
self.model.add(Conv2D(8, (2, 2), activation='relu', input_shape=(27, 27, 1)))
self.model.add(MaxPool2D())
self.model.add(Conv2D(16, (2, 2), activation='relu'))
self.model.add(MaxPool2D())
self.model.add(Conv2D(32, (2, 2), activation='relu'))
self.model.add(Conv2D(64, (2, 2), activation='relu'))
self.model.add(Flatten())
self.model.add(Dense(128, activation='relu'))
self.model.add(Dense(10, activation='sigmoid'))
self.model.compile(optimizer='sgd', loss='mse')
def train(x, y):
with tf.Session(graph=self.graph) as sess:
sess.run(tf.global_variables_initializer())
self.model.fit(x, y, verbose=0)
self.model.save_weights(self.model_name)
def test(x):
with tf.Session(graph=self.graph) as sess:
sess.run(tf.global_variables_initializer())
if os.path.exists(model_name):
self.model.load_weights(model_name)
return self.model.predict(x)
# Use closure to return the function
return train, testAs you can see, the function should use closure style to program. Or the error will occure. I don't really know the reason, but I haven't encounter the problem while I use closure.
Next, the main program is this:
from keras.layers import Conv2D, Dense, Flatten, Input, MaxPool2D
from model import Has1ConvLayerCNN, Has2ConvLayerCNN
from keras.models import Sequential
from keras import backend as K
import tensorflow as tf
import numpy as np
import threading
import time
import os
if __name__ == '__main__':
# Create data and model instance
x, y = np.random.random([4, 27, 27, 1]), np.random.random([4, 10])
small_cnn = Has1ConvLayerCNN()
big_cnn = Has2ConvLayerCNN()
# Form operations list
small_cnn_train, small_cnn_test = small_cnn.createModelAndOp()
big_cnn_train, big_cnn_test = big_cnn.createModelAndOp()
train_ops = [small_cnn_train, big_cnn_train]
test_ops = [small_cnn_test, big_cnn_test]
# Create the list of threads (train operations)
threads = []
for i in range(len(train_ops)):
threads.append(threading.Thread(target=train_ops[i], args=(x, y)))
print "------------ Train --------------------"
# Show the time of multithreading
_time = time.time()
for i in range(len(threads)):
threads[i].start()
print "<< Start whole thread >> time spend: ", time.time() - _time
for i in range(len(threads)):
threads[i].join(5)
print "<< Wait whole threads join >> time spend: ", time.time() - _time
# Show the time of working with order
_time = time.time()
train_ops[0](x, y)
train_ops[1](x, y)
print "<< Main thread work in order >> time spend: ", time.time() - _time
# Create the list of threads (test operations)
threads = []
for i in range(len(train_ops)):
threads.append(threading.Thread(target=test_ops[i], args=(x,)))
print "------------ Test --------------------"
# Show the time of multithreading
_time = time.time()
for i in range(len(threads)):
threads[i].start()
print "<< Start whole thread >> time spend: ", time.time() - _time
for i in range(len(threads)):
threads[i].join(5)
print "<< Wait whole threads join >> time spend: ", time.time() - _time
# Show the time of working with order
_time = time.time()
for i in range(len(test_ops)):
_x = np.expand_dims(x[0], 0)
test_ops[i](_x)
print "<< Main thread work in order >> time spend: ", time.time() - _timeAn the following is the result:
Using TensorFlow backend.
------------ Train --------------------
2017-06-18 17:23:28.364516: W tensorflow/core/platform/cpu_feature_guard.cc:45] The TensorFlow library wasn't compiled to use SSE4.1 instructions, but these are available on your machine and could speed up CPU computations.
2017-06-18 17:23:28.364537: W tensorflow/core/platform/cpu_feature_guard.cc:45] The TensorFlow library wasn't compiled to use SSE4.2 instructions, but these are available on your machine and could speed up CPU computations.
2017-06-18 17:23:28.364542: W tensorflow/core/platform/cpu_feature_guard.cc:45] The TensorFlow library wasn't compiled to use AVX instructions, but these are available on your machine and could speed up CPU computations.
2017-06-18 17:23:28.364546: W tensorflow/core/platform/cpu_feature_guard.cc:45] The TensorFlow library wasn't compiled to use AVX2 instructions, but these are available on your machine and could speed up CPU computations.
2017-06-18 17:23:28.364550: W tensorflow/core/platform/cpu_feature_guard.cc:45] The TensorFlow library wasn't compiled to use FMA instructions, but these are available on your machine and could speed up CPU computations.
<< Start whole thread >> time spend: 0.000718832015991
<< Wait whole threads join >> time spend: 0.377538919449
<< Main thread work in order >> time spend: 0.152100801468
------------ Test --------------------
<< Start whole thread >> time spend: 0.00042986869812
<< Wait whole threads join >> time spend: 0.079735994339
<< Main thread work in order >> time spend: 0.0590391159058After this testing, I didn't suggest to use multithreading technique to implement the different model in individual thread. As you can see, the program will spend more time in training or testing.
If there's some error or misunderstanding, welcome to raise your opinion.
I want to run the different model in the same time. However, I cannot get the correct answer for each running time. The following is my code. I generate the MNIST-like data, and build the 2 CNN network to do the training and testing.
I create individual session and graph object. Sometime I got the correct answer as the following:
However, I got the error sometimes...
Is there any wrong about my implementation? or some thing I should notice while doing multi-thread work?