I'm trying to query a single model from multiple threads. The use case is
that a single query on the model doesn't utilize the full power of the GPU,
so the throughput could be improved by starting multiple queries in
parallel instead of serializing the requests. I don't want to load multiple
instances of the model onto the GPU due to memory constraints. Since I
think that theano is not fully thread-safe, the idea is to compile multiple
instances of the computational graph (theano.function) and have the workers
of the threadpool work on a single graph at a time. The shared variables of
the model are only loaded once. See the example code below (the model there
is just a dummy model for demonstrating the problem).
This works if the model is on CPU, but fails if it is on the GPU. I get the
following error: GpuArrayException: b'cuEventCreate:
CUDA_ERROR_INVALID_CONTEXT: invalid device context' Is there a way to
initialize the context for each individual thread or clone the context of
the parent thread? Thanks in advance!
Example code:
------------------------------------------------------------------------------------------
import theano
from theano import tensor as T
import numpy as np
from collections import OrderedDict, namedtuple
from concurrent.futures import ThreadPoolExecutor
from queue import Queue
#this huge matrix will be shared between computational graphs
W1 = theano.shared(np.random.rand(100,1000000).astype('float32'))
#function for defining the computational graphs
def create_graph():
H = theano.shared(np.zeros(100, dtype='float32'))
updatesH = OrderedDict()
updatesH[H] = T.zeros_like(H)
reset = theano.function([], updates=updatesH)
X = T.fvector()
updatesH[H] = H + X
upd = theano.function([X], updates=updatesH)
Y = T.dot(H, W1).mean()
calc = theano.function([], Y)
return reset, upd, calc
#queue of computational graphs; the workers will remove one computational
graph at a time, use it for computations and then put it back
class GraphPoolHandler:
def __init__(self, n_threads):
self.graph_queue = Queue()
for i in range(n_threads):
self.graph_queue.put(create_graph()) #fill the queue with
computational graphs
def run(self, n, x):
reset, upd, calc = self.graph_queue.get()
reset()
for i in range(n):
upd(x)
res = calc()
self.graph_queue.put((reset, upd, calc))
return res
graph_pool = GraphPoolHandler(4)
thread_pool = ThreadPoolExecutor(max_workers=4)
def print_callback(v):
print(v.result(), end='\n')
f1 = thread_pool.submit(graph_pool.run, 2, np.arange(100).astype('float32'))
f2 = thread_pool.submit(graph_pool.run, 1, np.arange(100).astype('float32'))
f3 = thread_pool.submit(graph_pool.run, 4, np.arange(100).astype('float32'))
f4 = thread_pool.submit(graph_pool.run, 3, np.arange(100).astype('float32'))
f5 = thread_pool.submit(graph_pool.run, 5, np.arange(100).astype('float32'))
f1.add_done_callback(print_callback)
f2.add_done_callback(print_callback)
f3.add_done_callback(print_callback)
f4.add_done_callback(print_callback)
f5.add_done_callback(print_callback)
------------------------------------------------------------------------------------------
--
---
You received this message because you are subscribed to the Google Groups
"theano-users" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to theano-users+unsubscr...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.
