Improve concurrency

- Update Python to use async/await
- Use asyncpg
- Use connection pool
eyusupov/concurrency
Eldar Yusupov 4 years ago
parent 66d17625d8
commit a92bf6288c

@ -1,4 +1,4 @@
FROM python:3.4
FROM python:3.7
RUN mkdir -p /usr/src/app
WORKDIR /usr/src/app

@ -1,6 +1,6 @@
tornado==4.4.2
sqlalchemy==1.1.5
mercantile==0.9.0
pyproj==1.9.5.1
pyyaml==3.12
psycopg2==2.6.2
asyncpg==0.19.0
Click==7.0
mercantile==1.1.2
pyproj==2.4.0
PyYAML==5.1.2
tornado==6.0.3

@ -1,47 +1,38 @@
import tornado.ioloop
import tornado.web
import asyncpg
import io
import os
from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy import create_engine
from sqlalchemy import inspect
from sqlalchemy import text
from sqlalchemy.orm import sessionmaker
import logging
import tornado
import tornado.web
from tornado.log import enable_pretty_logging
import mercantile
import pyproj
import yaml
import sys
import itertools
log = logging.getLogger('tornado.application')
def GetTM2Source(file):
with open(file,'r') as stream:
tm2source = yaml.load(stream)
return tm2source
def GeneratePrepared(layers):
def GenerateFunction(layers):
queries = []
prepared = "PREPARE gettile(geometry, numeric, numeric, numeric) AS "
function = "CREATE OR REPLACE FUNCTION gettile(geometry, numeric, numeric, numeric) RETURNS SETOF bytea AS $$"
for layer in layers['Layer']:
layer_query = layer['Datasource']['table'].lstrip().rstrip() # Remove lead and trailing whitespace
layer_query = layer['Datasource']['table'].strip()
layer_query = layer_query[1:len(layer_query)-6] # Remove enough characters to remove first and last () and "AS t"
layer_query = layer_query.replace("geometry", "ST_AsMVTGeom(geometry,!bbox!,4096,0,true) AS mvtgeometry")
base_query = "SELECT ST_ASMVT('"+layer['id']+"', 4096, 'mvtgeometry', tile) FROM ("+layer_query+" WHERE ST_AsMVTGeom(geometry, !bbox!,4096,0,true) IS NOT NULL) AS tile"
queries.append(base_query.replace("!bbox!","$1").replace("!scale_denominator!","$2").replace("!pixel_width!","$3").replace("!pixel_height!","$4"))
prepared = prepared + " UNION ALL ".join(queries) + ";"
print(prepared)
return(prepared)
function = function + " UNION ALL ".join(queries) + ";$$ LANGUAGE SQL"
print(function)
return(function)
layers = GetTM2Source("/mapping/data.yml")
prepared = GeneratePrepared(layers)
engine = create_engine('postgresql://'+os.getenv('POSTGRES_USER','openmaptiles')+':'+os.getenv('POSTGRES_PASSWORD','openmaptiles')+'@'+os.getenv('POSTGRES_HOST','postgres')+':'+os.getenv('POSTGRES_PORT','5432')+'/'+os.getenv('POSTGRES_DB','openmaptiles'))
inspector = inspect(engine)
DBSession = sessionmaker(bind=engine)
session = DBSession()
session.execute(prepared)
dsn = 'postgresql://'+os.getenv('POSTGRES_USER','openmaptiles')+':'+os.getenv('POSTGRES_PASSWORD','openmaptiles')+'@'+os.getenv('POSTGRES_HOST','postgres')+':'+os.getenv('POSTGRES_PORT','5432')+'/'+os.getenv('POSTGRES_DB','openmaptiles')
def bounds(zoom,x,y):
inProj = pyproj.Proj(init='epsg:4326')
@ -62,7 +53,7 @@ def zoom_to_scale_denom(zoom): # For !scale_denominator!
def replace_tokens(query,s,w,n,e,scale_denom):
return query.replace("!bbox!","ST_MakeBox2D(ST_Point("+w+", "+s+"), ST_Point("+e+", "+n+"))").replace("!scale_denominator!",scale_denom).replace("!pixel_width!","256").replace("!pixel_height!","256")
def get_mvt(zoom,x,y):
async def get_mvt(connection,zoom,x,y):
try: # Sanitize the inputs
sani_zoom,sani_x,sani_y = float(zoom),float(x),float(y)
del zoom,x,y
@ -73,30 +64,45 @@ def get_mvt(zoom,x,y):
scale_denom = zoom_to_scale_denom(sani_zoom)
tilebounds = bounds(sani_zoom,sani_x,sani_y)
s,w,n,e = str(tilebounds['s']),str(tilebounds['w']),str(tilebounds['n']),str(tilebounds['e'])
final_query = "EXECUTE gettile(!bbox!, !scale_denominator!, !pixel_width!, !pixel_height!);"
final_query = "SELECT gettile(!bbox!, !scale_denominator!, !pixel_width!, !pixel_height!);"
sent_query = replace_tokens(final_query,s,w,n,e,scale_denom)
response = list(session.execute(sent_query))
print(sent_query)
log.info(sent_query)
response = await connection.fetch(sent_query)
layers = filter(None,list(itertools.chain.from_iterable(response)))
final_tile = b''
for layer in layers:
final_tile = final_tile + io.BytesIO(layer).getvalue()
final_tile = final_tile + io.BytesIO(layer).getvalue()
return final_tile
class GetTile(tornado.web.RequestHandler):
def get(self, zoom,x,y):
def initialize(self, pool):
self.pool = pool
async def get(self, zoom,x,y):
self.set_header("Content-Type", "application/x-protobuf")
self.set_header("Content-Disposition", "attachment")
self.set_header("Access-Control-Allow-Origin", "*")
response = get_mvt(zoom,x,y)
self.write(response)
async with self.pool.acquire() as connection:
response = await get_mvt(connection, zoom,x,y)
self.write(response)
async def get_pool():
pool = await asyncpg.create_pool(dsn = dsn)
layers = GetTM2Source(os.getenv("MAPPING_FILE", "/mapping/data.yml"))
# Make this prepared statement from the tm2source
create_function = GenerateFunction(layers)
async with pool.acquire() as connection:
await connection.execute(create_function)
return pool
def m():
if __name__ == "__main__":
# Make this prepared statement from the tm2source
application = tornado.web.Application([(r"/tiles/([0-9]+)/([0-9]+)/([0-9]+).pbf", GetTile)])
enable_pretty_logging()
io_loop = tornado.ioloop.IOLoop.current()
pool = io_loop.run_sync(get_pool)
application = tornado.web.Application([(r"/tiles/([0-9]+)/([0-9]+)/([0-9]+).pbf", GetTile, dict(pool=pool))])
print("Postserve started..")
application.listen(8080)
tornado.ioloop.IOLoop.instance().start()
application.listen(int(os.getenv("LISTEN_PORT", "8080")))
io_loop.start()
m()
if __name__ == "__main__":
m()

Loading…
Cancel
Save