Cassandra Db queries not working inside child process

[vartika singh]

HI, 

I have created a subprocess using gipc inside the python/flask app using gevent, and trying to write data in Cassandra Db. getting some unexpected error  - ERROR - operation on destroyed loop. 

I even faced this issue for Cassandra when I was trying to run DB queries inside the event loop. Do Cassandra doesn't work inside a child process? Please provide a fix if this is a known issue.

Thanks.

[Bret McGuire]

   Thanks for the question Vartika!

   The Python driver does support use cases involving child processes, including cases using multiprocessing.  See the driver documentation on this point as well as the blog post linked there.  I would also very definitely highlight the restrictions around sharing Cluster, Session or ResponseFuture instances across processes mentioned in the documentation.

   In your specific case something is trying to access the active count for an event loop after it's been closed (the relevant source for your error appears to be here).  The actual root cause of this error is difficult to identify with just this error message.  Do you have any additional information which might help us narrow down the cause?  A stack would be ideal if you have access to it.

   Thanks again!

   - Bret -

[vartika singh]

Thanks, Bret, please find the error stack trace.

2021-12-08 13:28:13,650 - models.Table1 - ERROR - operation on destroyed loop

Process _GProcess-1:

Traceback (most recent call last):

  File "/usr/local/Cellar/python@3.9/3.9.7/Frameworks/Python.framework/Versions/3.9/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap

    self.run()

  File "/usr/local/Cellar/python@3.9/3.9.7/Frameworks/Python.framework/Versions/3.9/lib/python3.9/multiprocessing/process.py", line 108, in run

    self._target(*self._args, **self._kwargs)

  File "/Users/v0s01ti/env/lib/python3.9/site-packages/gipc/gipc.py", line 396, in _child

    target(*args, **kwargs)

  File "/Users/project/src/main.py", line 82, in child_process

    self.run_solver( _tasks, _workers, version)

  File "/Users/project/src/main.py", line 144, in run_rand_func

  File "/Users/project/src/main.py", line 161, in save_entry_to_db

    self.run_rand_funct.save_results(val1, val2)

  File "/Users/project/src/main.py", line 551, in save_result

    raise DatabaseError(DATABASE_ERROR)

[Bret McGuire]

   It's hard to tell too much from just this stack but it appears that the access of the destroyed loop is occurring within the function that's executed by the child process.  Note the call to "target()" on line 396 of gipc.py in the gipc package.  That call runs in the child process and executes the function supplied by the user.  But before it does so it goes to considerable lengths to shutdown and restart any existing gevent event loops.  It looks like this is done to avoid using any gevent state which might have been inherited from the parent but which may be in a non-good state in the child process.  So what may be happening is something like the following:

1. The parent process creates a Cluster object backed by a gevent event loop
2. A function (let's call it F) is defined which uses this Cluster object to execute some CQL calls
3. The parent passes F to gipc, which uses multiprocessing to execute it in a child process
4. The gipc code running in the child process resets the gevent state, including the underlying event loop (as discussed above)
5. F then tries to execute CQL calls using the Cluster object created in the parent, but since the underlying event loop has been closed the "operation on destroyed loop" error is thrown instead

   It's for precisely this reason that the Python driver docs state that child processes shouldn't re-use driver-related state they inherit from their parent.  Each child process should open and use their own Cluster, Session and ResponseFuture objects.  Which leads me to ask: is the function you're passing to gipc creating it's own Cluster and/or Session?  Or is it trying to use a Cluster and/or Session which has been created somewhere else in your program?

[vartika singh]

Thanks Bret, understood the issue. Could you please help me in defining the Cassandra configuration for the child process.

--
To unsubscribe from this group and stop receiving emails from it, send an email to python-driver-u...@lists.datastax.com.

[vartika singh]

I have created a separate connections for the child process as well but still facing the same issue. Cassandra connection object for both parent and child process.

[vartika singh]

please find the stack trace

Process _GProcess-1:

Traceback (most recent call last):

  File "/usr/local/Cellar/python@3.9/3.9.7/Frameworks/Python.framework/Versions/3.9/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap

    self.run()

  File "/usr/local/Cellar/python@3.9/3.9.7/Frameworks/Python.framework/Versions/3.9/lib/python3.9/multiprocessing/process.py", line 108, in run

    self._target(*self._args, **self._kwargs)

  File "/Users/env/lib/python3.9/site-packages/gipc/gipc.py", line 396, in _child

    target(*args, **kwargs)

  File "Users/<path>", line 92, in child_process

    self.some_func( a, b)

  File "Users/<path>", line 158, in some_func

    save_entry_database()

  File "Users/<path>", line 175, in save_entry_database

    dd = session.execute("use database")

  File "cassandra/cluster.py", line 2618, in cassandra.cluster.Session.execute

  File "cassandra/cluster.py", line 2661, in cassandra.cluster.Session.execute_async

  File "cassandra/cluster.py", line 2955, in cassandra.cluster.Session._create_response_future

  File "cassandra/cluster.py", line 4310, in cassandra.cluster.ResponseFuture.__init__

  File "cassandra/cluster.py", line 4327, in cassandra.cluster.ResponseFuture._start_timer

  File "/Users/v0s01ti/env/lib/python3.9/site-packages/cassandra/io/geventreactor.py", line 59, in create_timer

    cls._new_timer.set()

  File "src/gevent/event.py", line 115, in gevent._gevent_cevent.Event.set

  File "src/gevent/_abstract_linkable.py", line 226, in gevent._gevent_c_abstract_linkable.AbstractLinkable._check_and_notify

  File "src/gevent/libev/corecext.pyx", line 722, in gevent.libev.corecext.loop.run_callback

  File "src/gevent/libev/corecext.pyx", line 272, in gevent.libev.corecext._check_loop

ValueError: operation on destroyed loop