Python3 sqlalchemy pymysql gevent sqlalchemy.util.queue.Empty gevent.hub.LoopExit: эта операция будет заблокирована навсегда

Question

Я использую python 3.4 и sqlalchemy и pymsql и gevent и celery вместе в своем приложении. Иногда во время вызова базы данных sql возникает следующая ошибка, не знаю, почему это произойдет. Есть идеи?

[2015-09-24 15:19:32,790: ERROR/Worker-1] Exception during reset or similar 
Traceback (most recent call last): 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/pool.py", line 950, in _do_get 
    return self._pool.get(wait, self._timeout) 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/util/queue.py", line 145, in get 
    raise Empty 
sqlalchemy.util.queue.Empty 

During handling of the above exception, another exception occurred: 

Traceback (most recent call last): 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/pool.py", line 567, in _finalize_fairy 
    fairy._reset(pool) 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/pool.py", line 701, in _reset 
    pool._dialect.do_rollback(self) 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/dialects/mysql/base.py", line 2354, in do_rollback 
    dbapi_connection.rollback() 
  File "/usr/local/lib/python3.4/dist-packages/pymysql/connections.py", line 711, in rollback 
    self._read_ok_packet() 
  File "/usr/local/lib/python3.4/dist-packages/pymysql/connections.py", line 685, in _read_ok_packet 
    pkt = self._read_packet() 
  File "/usr/local/lib/python3.4/dist-packages/pymysql/connections.py", line 882, in _read_packet 
    packet_header = self._read_bytes(4) 
  File "/usr/local/lib/python3.4/dist-packages/pymysql/connections.py", line 899, in _read_bytes 
    data = self._rfile.read(num_bytes) 
  File "/usr/lib/python3.4/socket.py", line 371, in readinto 
    return self._sock.recv_into(b) 
  File "/usr/local/lib/python3.4/dist-packages/gevent/_socket3.py", line 282, in recv_into 
    self._wait(self._read_event) 
  File "/usr/local/lib/python3.4/dist-packages/gevent/_socket3.py", line 97, in _wait 
    self.hub.wait(watcher) 
  File "/usr/local/lib/python3.4/dist-packages/gevent/hub.py", line 360, in wait 
    assert result is unique, 'Invalid switch into %s: %r (expected %r)' % (getcurrent(), result, unique) 
AssertionError: Invalid switch into <greenlet.greenlet object at 0x7fb774422800>: <gevent.event.AsyncResult object at 0x7fb76d323198> (expected <object object at 0x7fb777ff82d0>) 
[2015-09-24 15:19:32,911: ERROR/MainProcess] Task ppp.start_job[b574f2cd-4e9d-4f0f-97cc-79dd1ddee666] raised unexpected: LoopExit('This operation would block forever',) 
Traceback (most recent call last): 
  File "/usr/local/lib/python3.4/dist-packages/celery/app/trace.py", line 240, in trace_task 
    R = retval = fun(*args, **kwargs) 
  File "/usr/local/lib/python3.4/dist-packages/celery/app/trace.py", line 438, in __protected_call__ 
    return self.run(*args, **kwargs) 
  File "/home/aaa/ppp/tasks/start_job.py", line 45, in start_job 
    add_all_to_db(data_to_insert) 
  File "/home/aaa/ppp/tasks/utils/sqldb.py", line 284, in add_all_to_db 
    session.close() 
  File "/home/aaa/ppp/tasks/utils/sqldb.py", line 270, in get_session 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/sql/schema.py", line 3404, in create_all 
    tables=tables) 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/engine/base.py", line 1615, in _run_visitor 
    with self._optional_conn_ctx_manager(connection) as conn: 
  File "/usr/lib/python3.4/contextlib.py", line 59, in __enter__ 
    return next(self.gen) 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/engine/base.py", line 1608, in _optional_conn_ctx_manager 
    with self.contextual_connect() as conn: 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/engine/base.py", line 1798, in contextual_connect 
    self.pool.connect(), 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/pool.py", line 338, in connect 
    return _ConnectionFairy._checkout(self) 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/pool.py", line 644, in _checkout 
    fairy = _ConnectionRecord.checkout(pool) 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/pool.py", line 440, in checkout 
    rec = pool._do_get() 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/pool.py", line 963, in _do_get 
    return self._create_connection() 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/pool.py", line 285, in _create_connection 
    return _ConnectionRecord(self) 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/pool.py", line 411, in __init__ 
    self.connection = self.__connect() 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/pool.py", line 538, in __connect 
    connection = self.__pool._creator() 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/engine/strategies.py", line 90, in connect 
    return dialect.connect(*cargs, **cparams) 
  File "/usr/local/lib/python3.4/dist-packages/sqlalchemy/engine/default.py", line 377, in connect 
    return self.dbapi.connect(*cargs, **cparams) 
  File "/usr/local/lib/python3.4/dist-packages/pymysql/__init__.py", line 88, in Connect 
    return Connection(*args, **kwargs) 
  File "/usr/local/lib/python3.4/dist-packages/pymysql/connections.py", line 644, in __init__ 
    self._connect() 
  File "/usr/local/lib/python3.4/dist-packages/pymysql/connections.py", line 823, in _connect 
    (self.host, self.port), self.connect_timeout) 
  File "/usr/local/lib/python3.4/dist-packages/gevent/socket.py", line 51, in create_connection 
    for res in getaddrinfo(host, port, 0 if has_ipv6 else AF_INET, SOCK_STREAM): 
  File "/usr/local/lib/python3.4/dist-packages/gevent/_socketcommon.py", line 193, in getaddrinfo 
    return get_hub().resolver.getaddrinfo(host, port, family, socktype, proto, flags) 
  File "/usr/local/lib/python3.4/dist-packages/gevent/resolver_thread.py", line 34, in getaddrinfo 
    return self.pool.apply_e(self.expected_errors, _socket.getaddrinfo, args, kwargs) 
  File "/usr/local/lib/python3.4/dist-packages/gevent/threadpool.py", line 222, in apply_e 
    success, result = self.spawn(wrap_errors, expected_errors, function, args, kwargs).get() 
  File "/usr/local/lib/python3.4/dist-packages/gevent/event.py", line 233, in get 
    result = self.hub.switch() 
  File "/usr/local/lib/python3.4/dist-packages/gevent/hub.py", line 349, in switch 
    return greenlet.switch(self) 
gevent.hub.LoopExit: This operation would block forever 

Моя get_session функция:

def get_session(): 
    engine = create_engine('mysql+pymysql://{0}:{1}@{2}/{3}'.format(SQL_USR, SQL_PASS, SQL_HOST, DB_NAME), 
          echo=True, pool_size=200, pool_recycle=360) 
    Base.metadata.create_all(engine) 
    Session = sessionmaker(bind=engine) 
    session = Session() 
    return session 

Answer 1

На этой неделе я испытывал некоторые странные проблемы, несколько похожа на вашу, с python3.4, sqlalchemy и oracle. Я публикую решение, которое я нашел, если оно вам поможет.

У меня есть программы, которые отлично работают для python2.7, но время от времени я потратил некоторое время, пытаясь перенести их на 3.4. Получено несколько сбоев, некоторые из которых связаны с oracle доступом через sqlalchemy и после своего рода шаблона. Самый простой сценарий, в котором он был сбой, - загрузка данных в базу данных; в основном программа остановится навсегда. У меня не было такого исключения, как у вас, но я предполагаю, что происхождение проблемы может быть похоже на ваше.

Я думал о размещении вопроса здесь, но я ожидал, что это была просто проблема установки на моей машине, и в любом случае я не знаю, как правильно сформулировать вопрос. Я сделал немало испытаний и получил, что простой сценарий работает нормально для

• postgresql как для 2,7 и 3,4
• oracle для 2.7, как я упомянул выше

Это привело меня в более испытаний и в конце концов я сделал изменение, немного слепое, что я должен сказать, и это исправило это. Я делал нагрузку, как это,

for user_id in users_set: 
    reg = UserRank(user_id=int(user_id), rank_exec_id=rank_exec_id, rank_step_id=rank_step_id, 
        rank=rank_df.loc[user_id]) 
    session.add(reg) 
session.commit() 

где rank_df.loc[user_id] был типа numpy.float64, и я изменил его на

for user_id in users_set: 
    reg = UserRank(user_id=int(user_id), rank_exec_id=rank_exec_id, rank_step_id=rank_step_id, 
        rank=float(rank_df.loc[user_id])) 
    session.add(reg) 
session.commit() 

, таким образом, превращая его в сырой питона float. И он начал работать, хотя я еще не понял эту проблему.

Просто для полноты картины, это класс для таблицы,

class UserRank(Base): 
    __tablename__ = 'user_rank' 
    user_id = Column(BigInteger, primary_key=True) 
    rank_exec_id = Column(String(256), primary_key=True) 
    rank_step_id = Column(String(256), primary_key=True) 
    rank = Column(Float) 

Честно говоря, я с трудом верю свое объяснение, но я надеюсь, что это поможет вам в некотором роде.

Answer 2

Вы используете gevent. Вы применили для нее патч обезьяны? Пожалуйста, обратитесь к http://www.gevent.org/gevent.monkey.html за тем, как применить этот патч.