Errores de base de datos en Django cuando se utiliza el subprocesamiento

Question

Estoy trabajando en una aplicación web de Django que necesita consultar una base de datos PostgreSQL. Al implementar la concurrencia mediante la interfaz de Python threading, obtengo errores DoesNotExist para los elementos consultados. Por supuesto, estos errores no ocurren cuando se realizan las consultas secuencialmente.

Te voy a enseñar una prueba de unidad que escribí para demostrar el comportamiento inesperado:

class ThreadingTest(TestCase): 
    fixtures = ['demo_city',] 

    def test_sequential_requests(self): 
     """ 
     A very simple request to database, made sequentially. 

     A fixture for the cities has been loaded above. It is supposed to be 
     six cities in the testing database now. We will made a request for 
     each one of the cities sequentially. 
     """ 
     for number in range(1, 7): 
      c = City.objects.get(pk=number) 
      self.assertEqual(c.pk, number) 

    def test_threaded_requests(self): 
     """ 
     Now, to test the threaded behavior, we will spawn a thread for 
     retrieving each city from the database. 
     """ 

     threads = [] 
     cities = [] 

     def do_requests(number): 
      cities.append(City.objects.get(pk=number)) 

     [threads.append(threading.Thread(target=do_requests, args=(n,))) for n in range(1, 7)] 

     [t.start() for t in threads] 
     [t.join() for t in threads] 

     self.assertNotEqual(cities, []) 

Como se puede ver, la primera prueba realiza algunas peticiones de base de datos de forma secuencial, los cuales son de hecho trabajando sin ningún problema. La segunda prueba, sin embargo, realiza exactamente las mismas solicitudes, pero cada solicitud se genera en un hilo. En realidad, esta falla y devuelve una excepción DoesNotExist.

La salida de la ejecución de esta unidad de pruebas es la siguiente:

test_sequential_requests (cesta.core.tests.threadbase.ThreadingTest) ... ok 
test_threaded_requests (cesta.core.tests.threadbase.ThreadingTest) ... 

Exception in thread Thread-1: 
Traceback (most recent call last): 
    File "/usr/lib/python2.6/threading.py", line 532, in __bootstrap_inner 
    self.run() 
    File "/usr/lib/python2.6/threading.py", line 484, in run 
    self.__target(*self.__args, **self.__kwargs) 
    File "/home/jose/Work/cesta/trunk/src/cesta/core/tests/threadbase.py", line 45, in do_requests 
    cities.append(City.objects.get(pk=number)) 
    File "/home/jose/Work/cesta/trunk/parts/django/django/db/models/manager.py", line 132, in get 
    return self.get_query_set().get(*args, **kwargs) 
    File "/home/jose/Work/cesta/trunk/parts/django/django/db/models/query.py", line 349, in get 
    % self.model._meta.object_name) 
DoesNotExist: City matching query does not exist. 

... otros hilos devuelve una salida similar ...

Exception in thread Thread-6: 
Traceback (most recent call last): 
    File "/usr/lib/python2.6/threading.py", line 532, in __bootstrap_inner 
    self.run() 
    File "/usr/lib/python2.6/threading.py", line 484, in run 
    self.__target(*self.__args, **self.__kwargs) 
    File "/home/jose/Work/cesta/trunk/src/cesta/core/tests/threadbase.py", line 45, in do_requests 
    cities.append(City.objects.get(pk=number)) 
    File "/home/jose/Work/cesta/trunk/parts/django/django/db/models/manager.py", line 132, in get 
    return self.get_query_set().get(*args, **kwargs) 
    File "/home/jose/Work/cesta/trunk/parts/django/django/db/models/query.py", line 349, in get 
    % self.model._meta.object_name) 
DoesNotExist: City matching query does not exist. 


FAIL 

====================================================================== 
FAIL: test_threaded_requests (cesta.core.tests.threadbase.ThreadingTest) 
---------------------------------------------------------------------- 
Traceback (most recent call last): 
    File "/home/jose/Work/cesta/trunk/src/cesta/core/tests/threadbase.py", line 52, in test_threaded_requests 
    self.assertNotEqual(cities, []) 
AssertionError: [] == [] 

---------------------------------------------------------------------- 
Ran 2 tests in 0.278s 

FAILED (failures=1) 
Destroying test database for alias 'default' ('test_cesta')... 

Recuerde que todo esto es sucediendo en una base de datos PostgreSQL, que se supone que es segura para subprocesos, no con el SQLite o similares. La prueba se ejecutó usando PostgreSQL también.

En este punto, estoy totalmente perdido acerca de lo que puede estar fallando. Alguna idea o sugerencia?

Gracias!

EDIT: Escribí una pequeña vista para comprobar si funciona correctamente. Aquí está el código de la vista:.

def get_cities(request): 
    queue = Queue.Queue() 

    def get_async_cities(q, n): 
     city = City.objects.get(pk=n) 
     q.put(city) 

    threads = [threading.Thread(target=get_async_cities, args=(queue, number)) for number in range(1, 5)] 

    [t.start() for t in threads] 
    [t.join() for t in threads] 

    cities = list() 

    while not queue.empty(): 
     cities.append(queue.get()) 

    return render_to_response('async/cities.html', {'cities': cities}, 
     context_instance=RequestContext(request)) 

(Por favor, no tome en cuenta la locura de escribir la lógica de la aplicación dentro del código de la vista Recuerde que esto es sólo una prueba de concepto y no sería nunca se en la aplicación real)

El resultado es que el código funciona bien, las solicitudes se realizan con éxito en los hilos y la vista finalmente muestra las ciudades después de llamar a su URL.

Por lo tanto, creo que realizar consultas utilizando hilos solo será un problema cuando deba probar el código. En producción, funcionará sin ningún problema.

¿Alguna sugerencia útil para probar este tipo de código con éxito?

Answer 1

Parece que es un problema con las transacciones. Si está creando elementos dentro de la solicitud (o prueba) actual, es casi seguro que se encuentren en una transacción no comprometida a la que no se puede acceder desde la conexión separada en la otra cadena. Probablemente necesite manage your transctions manually para que esto funcione.

Answer 2

Intente utilizar TransactionTestCase:

class ThreadingTest(TransactionTestCase): 

TestCase mantiene los datos en la memoria y no emite un COMMIT a la base de datos. Probablemente los hilos intentan conectarse directamente a DB, mientras que los datos no están todavía allí.Seedescription aquí: https://docs.djangoproject.com/en/dev/topics/testing/?from=olddocs#django.test.TransactionTestCase

TransactionTestCase y TestCase son idénticos excepto por la forma en el que la base de datos se restablece a un estado conocido y la capacidad para código de prueba para probar los efectos de cometer y reversión. A TransactionTestCase restablece la base de datos antes de que la prueba se ejecute en truncando todas las tablas y volviendo a cargar los datos iniciales. Un TransactionTestCase puede invocar commit y rollback y observar los efectos de estas llamadas en la base de datos.

Answer 3

se vuelve más claro a partir de esta parte de la documentación

class LiveServerTestCase(TransactionTestCase): 
    """ 
    ... 
    Note that it inherits from TransactionTestCase instead of TestCase because 
    the threads do not share the same transactions (unless if using in-memory 
    sqlite) and each thread needs to commit all their transactions so that the 
    other thread can see the changes. 
    """ 

Ahora, la transacción no se ha cometido dentro de un TestCase, por lo tanto, los cambios no son visibles para el otro hilo.