alex / alchimia Goto Github PK
View Code? Open in Web Editor NEWLicense: MIT License
License: MIT License
Pretty much every object in alchimia.engine
manages some state:
TwistedEngine and TwistedConnection are particularly problematic because the worker thread created behind the scenes is its own strong-reference root and will never be cleaned up during the lifetime of the process if these objects do not shut them down.
As seen in #30 , leaking these types of resources can give one the impression that the whole system is just freezing for no reason.
The build is now failing:
Picking up where #28 left off…
Currently, autoloading tables doesn't really work. If I try to do:
table = Table(table_name, metadata, autoload=True)
The call fails because Table._autoload()
tries to use Engine.bind_callable()
, which isn't defined on TwistedEngine
. However, you can do the following:
table = Table(table_name, metadata, autoload=True, autoload_with=engine._engine)
However, it's a bit messy to try to access engine._engine
. Would it make sense to either override Engine.bind_callable()
or to expose a more formal way of accessing the underlying engine with a property or something?
I believe it's fine to make the blocking call for this use case, since you should only be reflecting on the tables at startup time.
I'm writing app to consume event feed and store received events into mysql. Events should be routed to different tables based on a key in them and stored as soon as possible. But whatever i do, i can't get it to spawn more threads and performance is really bad for sequential inserts (~20 inserts/s).
multiple calls (params) => single thread, slow:
for item in data:
d = engine.execute(table.insert().values(**item))
d.addCallback(row_handler)
single call (multiparams) => single thread, fast:
d = engine.execute(table.insert(), data)
d.addCallback(all_handler)
I could possibly queue each event type and call multiparams style execute at regular intervals, but i'd rather use separate execute call for each event as soon as it arrives
I'm using version 0.7.0, mysql server 5.7, tested with pymysql, cymysql, mysqlconnector and mysqlclient, all the same.
When using adbapi, threads are created correctly
Am i missing something? Any ideas?
#27 is a pretty substantial fix, allowing the correct use of transactions for the first time. Someone (me) should do a release.
I've got some code that creates a SQLite DB, populates a schema, then writes some data.
If the SQLite DB is a file, this seems to work as expected:
2018-06-08T16:10:40-0700 [__main__.Index#info] Importing transmissions...
2018-06-08T16:10:40-0700 [__main__.Index#info] Initializing SQLite DB...
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,028 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,029 INFO sqlalchemy.engine.base.Engine ()
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,029 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,029 INFO sqlalchemy.engine.base.Engine ()
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,030 INFO sqlalchemy.engine.base.Engine PRAGMA table_info("TRANSMISSION")
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,030 INFO sqlalchemy.engine.base.Engine ()
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,031 INFO sqlalchemy.engine.base.Engine INSERT INTO "TRANSMISSION" ("STATION", "SYSTEM", "CHANNEL") VALUES (?, ?, ?)
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,031 INFO sqlalchemy.engine.base.Engine (('Radio 1', 'A', 'Ops'), ('Radio 4', 'B', 'Talk'))
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,032 INFO sqlalchemy.engine.base.Engine COMMIT
2018-06-08T16:10:40-0700 [-] Main loop terminated.
However, if the SQLite DB is an in-memory DB, it fails, complaining that the table being written to doesn't exist:
2018-06-08T16:12:34-0700 [__main__.Index#info] Importing transmissions...
2018-06-08T16:12:34-0700 [__main__.Index#info] Initializing SQLite DB...
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,912 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,912 INFO sqlalchemy.engine.base.Engine ()
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,913 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,913 INFO sqlalchemy.engine.base.Engine ()
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,914 INFO sqlalchemy.engine.base.Engine PRAGMA table_info("TRANSMISSION")
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,914 INFO sqlalchemy.engine.base.Engine ()
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,914 INFO sqlalchemy.engine.base.Engine
2018-06-08T16:12:34-0700 [stdout#info] CREATE TABLE "TRANSMISSION" (
2018-06-08T16:12:34-0700 [stdout#info] "STATION" VARCHAR NOT NULL,
2018-06-08T16:12:34-0700 [stdout#info] "SYSTEM" VARCHAR NOT NULL,
2018-06-08T16:12:34-0700 [stdout#info] "CHANNEL" VARCHAR NOT NULL
2018-06-08T16:12:34-0700 [stdout#info] )
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,915 INFO sqlalchemy.engine.base.Engine ()
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,915 INFO sqlalchemy.engine.base.Engine COMMIT
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,916 INFO sqlalchemy.engine.base.Engine INSERT INTO "TRANSMISSION" ("STATION", "SYSTEM", "CHANNEL") VALUES (?, ?, ?)
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,916 INFO sqlalchemy.engine.base.Engine (('Radio 1', 'A', 'Ops'), ('Radio 4', 'B', 'Talk'))
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,916 INFO sqlalchemy.engine.base.Engine ROLLBACK
2018-06-08T16:12:34-0700 [-] main function encountered error
Traceback (most recent call last):
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/twisted/internet/defer.py", line 500, in errback
self._startRunCallbacks(fail)
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/twisted/internet/defer.py", line 567, in _startRunCallbacks
self._runCallbacks()
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/twisted/internet/defer.py", line 653, in _runCallbacks
current.result = callback(current.result, *args, **kw)
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/twisted/internet/defer.py", line 1442, in gotResult
_inlineCallbacks(r, g, deferred)
--- <exception caught here> ---
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/twisted/internet/defer.py", line 1384, in _inlineCallbacks
result = result.throwExceptionIntoGenerator(g)
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/twisted/python/failure.py", line 422, in throwExceptionIntoGenerator
return g.throw(self.type, self.value, self.tb)
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/mixer/alch_test.py", line 94, in addTransmissions
for transmission in transmissions
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/alchimia/engine.py", line 33, in container
result = work(*args, **kwargs)
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 2075, in execute
return connection.execute(statement, *multiparams, **params)
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 948, in execute
return meth(self, multiparams, params)
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/sql/elements.py", line 269, in _execute_on_connection
return connection._execute_clauseelement(self, multiparams, params)
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1060, in _execute_clauseelement
compiled_sql, distilled_params
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1200, in _execute_context
context)
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1413, in _handle_dbapi_exception
exc_info
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 203, in raise_from_cause
reraise(type(exception), exception, tb=exc_tb, cause=cause)
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 186, in reraise
raise value.with_traceback(tb)
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1170, in _execute_context
context)
File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 505, in do_executemany
cursor.executemany(statement, parameters)
sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: TRANSMISSION [SQL: 'INSERT INTO "TRANSMISSION" ("STATION", "SYSTEM", "CHANNEL") VALUES (?, ?, ?)'] [parameters: (('Radio 1', 'A', 'Ops'), ('Radio 4', 'B', 'Talk'))] (Background on this error at: http://sqlalche.me/e/e3q8)
2018-06-08T16:12:34-0700 [-] Main loop terminated.
The difference between these two runs is whether line 71 or line 72 is commented out.
Hi, great work!
I'm kind of new to both SQLAlchemy and Twisted, but to what I got used to is called Declarative
http://docs.sqlalchemy.org/en/rel_0_9/orm/extensions/declarative.html and as far I know the way to manage this is by using Sessions.
Working with them gives me
'TwistedEngine' object has no attribute 'contextual_connect'
Hello!
It would be quite helpful to explicitly confirm whether alchimia.tests.doubles is public or not. If it isn't, both things in there right now are pretty critical for testing apps using alchimia.
I've already filed https://twistedmatrix.com/trac/ticket/8891 upstream.
I would love to see the motivation for this project somewhere? what does it enable me to do?
I could guess, but that's just wrong.
Trying to use Alchimia with sqlite3
, as documented on the front page, I get:
sqlalchemy.exc.ProgrammingError: (sqlite3.ProgrammingError) SQLite objects created in a thread can only be used in that same thread.The object was created in thread id 123145310715904 and this is thread id 123145306509312
As documented here https://www.python.org/dev/peps/pep-0249/#threadsafety in the DB-API PEP, the threadsafety
attribute of the sqlite3
module is 1
, which means you aren't allowed to move connections between threads.
I'm starting this as a place to have the conversation about a txpostgres implementation of this. I'm willing to work on it as I'm starting a new project that would benefit from it.
Having checked the source so far, I think the majority of the work would be abstracting the deferToThread stuff into a less thread-y naming convention and then hooking up the txpostgres connection objects.
the txpostgres pool is pretty limited compared to the sqla ones, so there may be a little bit of integration worth doing there too.
Anyone have any other thoughts?
I keep getting this error when I make many simultaneous queries (klein -> alchimia -> PostgreSQL):
QueuePool limit of size 5 overflow 10 reached, connection timed out, timeout 30
I'm working around it right now by setting max_overflow=-1
(http://docs.sqlalchemy.org/en/rel_0_8/core/pooling.html#sqlalchemy.pool.QueuePool.params.max_overflow) but I think that will cause problems if I actually open more connections than the Postgres database allows. So I think I (or someone) could implement a Deferred
-returning pool that returns Connections
asynchronously and has requests for connections wait in a queue.
I'm filing this here, because I think alchimia should come with such a pool built in. I'll take a look at what an implementation might look like on Monday.
The result of inserting a row doesn't have an inserted_primary_key
attribute. I'm thinking about how to do this and will submit a PR if I come up with a solution (or do you already have one?)
Using code substantially similar to @hawkowl (https://atleastfornow.net/blog/alchimia/),
Pypy PyPy 5.4.1
Twisted 16.5.0
alchimia 0.70
psycopg2cffi 2.7.5
QueuePool is exhausted immediately. As far as I can tell connections are never garbage collected, rather they timeout after 30 seconds. The problem is not present when using alchimia 0.6.1 under the same Pypy or running 0.7.0 in under CPython 2.7.12 .
I was unable to distill the failure case to a simpler one. The workload is update heavy. I uploaded the failing code here: https://github.com/nocko/c3next
I am at a loss for how to troubleshoot this. Please advise.
Possibly related to #30?
Currently, if you do something like statement.execute()
instead of engine.execute(statement)
, it doesn't work, because Engine._execute_clauseelement
isn't overridden.
What do you think of overriding it to just do the same thing as Engine.execute
? It looks like it's pretty similar under the hood:
Python 2.7.5
Twisted==16.4.1
alchimia==0.7.0
psycopg2==2.6.2
SQLAlchemy==1.0.15
Linux ahost-02 3.10.0-229.7.2.el7.x86_64 #1 SMP Tue Jun 23 05:02:16 PDT 2015 x86_64 x86_64 x86_64 GNU/Linux
In my application calling engine.execute() results in hang - Deferred is never called. I injected verbose logging in alchimia/engine.py and it appears the inner function "container" in _defer_to_worker is not being called:
2016-10-05T05:58:43Z [DEBUG] a.e.TwistedEngine - execute called. args=(<sqlalchemy.sql.selectable.Select at 0x4b267d0; Select object>,) kwargs={}
2016-10-05T05:58:43Z [DEBUG] a.e.TwistedEngine - _defer_to_engine called. a=(<sqlalchemy.sql.selectable.Select at 0x4b267d0; Select object>,) f=<bound method Engine.execute of Engine(postgresql+psycopg2://zzz:***@localhost/zzz)> k={}
2016-10-05T05:58:43Z [DEBUG] a.engine - _defer_to_worker called. args=(<sqlalchemy.sql.selectable.Select at 0x4b267d0; Select object>,) worker=<twisted._threads._threadworker.ThreadWorker object at 0x499e290> work=<bound method Engine.execute of Engine(postgresql+psycopg2://zzz:***@localhost/zzz)> deliver=<bound method EPollReactor.callFromThread of <twisted.internet.epollreactor.EPollReactor object at 0x3224fd0>> kwargs={}
If I change related code to call to engine.connect() the call succeeds as usual. Injected logging:
2016-10-05T06:16:30Z [DEBUG] d.c.FooController - listing foo
2016-10-05T06:16:30Z [DEBUG] a.e.TwistedEngine - connect: called.
2016-10-05T06:16:30Z [DEBUG] a.engine - _threaded_worker#_start_worker: starting thread. target=<function work at 0x50edb18>
2016-10-05T06:16:30Z [DEBUG] a.engine - _threaded_worker#_start_worker: daemonized thead ... starting. thread=<Thread(Thread-3, initial daemon)>
2016-10-05T06:16:30Z [DEBUG] a.e.TwistedEngine - connect: deferring to worker. worker=<twisted._threads._threadworker.ThreadWorker object at 0x50e4f10>
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker called. args=() worker=<twisted._threads._threadworker.ThreadWorker object at 0x50e4f10> work=<bound method Engine.connect of Engine(postgresql+psycopg2://zzz:***@localhost/zzz)> deliver=<bound method EPollReactor.callFromThread of <twisted.internet.epollreactor.EPollReactor object at 0x33293d0>> kwargs={}
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: running.
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: got result. result=<sqlalchemy.engine.base.Connection object at 0x50f67d0>
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: delivered result.
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker called. args=() worker=<twisted._threads._threadworker.ThreadWorker object at 0x50e4f10> work=<bound method Connection.begin of <sqlalchemy.engine.base.Connection object at 0x50f67d0>> deliver=<bound method EPollReactor.callFromThread of <twisted.internet.epollreactor.EPollReactor object at 0x33293d0>> kwargs={}
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: running.
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: got result. result=<sqlalchemy.engine.base.RootTransaction object at 0x50e4b50>
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: delivered result.
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker called. args=(<sqlalchemy.sql.selectable.Select at 0x50f6810; Select object>,) worker=<twisted._threads._threadworker.ThreadWorker object at 0x50e4f10> work=<bound method Connection.execute of <sqlalchemy.engine.base.Connection object at 0x50f67d0>> deliver=<bound method EPollReactor.callFromThread of <twisted.internet.epollreactor.EPollReactor object at 0x33293d0>> kwargs={}
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: running.
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: got result. result=<sqlalchemy.engine.result.ResultProxy object at 0x50f64d0>
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: delivered result.
2016-10-05T06:16:30Z [DEBUG] d.datastore - got result. result=<alchimia.engine.TwistedResultProxy object at 0x50f68d0>
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker called. args=() worker=<twisted._threads._threadworker.ThreadWorker object at 0x50e4f10> work=<bound method ResultProxy.fetchall of <sqlalchemy.engine.result.ResultProxy object at 0x50f64d0>> deliver=<bound method EPollReactor.callFromThread of <twisted.internet.epollreactor.EPollReactor object at 0x33293d0>> kwargs={}
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: running.
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: got result. result=[…<snip>…]
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: delivered result.
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker called. args=() worker=<twisted._threads._threadworker.ThreadWorker object at 0x50e4f10> work=<bound method RootTransaction.commit of <sqlalchemy.engine.base.RootTransaction object at 0x50e4b50>> deliver=<bound method EPollReactor.callFromThread of <twisted.internet.epollreactor.EPollReactor object at 0x33293d0>> kwargs={}
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: running.
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: got result. result=None
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: delivered result.
2016-10-05T06:16:30Z [INFO] t.p.log - "127.0.0.1" - - [05/Oct/2016:06:16:29 +0000] "GET /foo HTTP/1.1" 200 1478 "-" "-".
The code in question that hangs:
def get_foo(self):
foo = tables.foo
stmt = select([foo.c.foo_id, foo.c.modified])
stmt = stmt.order_by(foo.c.foo_id.asc())
d = self.engine.execute(stmt)
return d.addCallback(blah) # blah just prints result ... but this is never called back
Do you have a table of sqlalchemy.core features vs features supported by alchimia?
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google ❤️ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.