renderd.py

#!/usr/bin/python
#
# mod_tile rendering daemon example written in Python.
# The code is mostly a direct port of the C implementation.
#
# This is currently experimental and not intended as a replacement
# of the C implementation, but works surpringly well. It should be
# easier to produce custom variations of the rendering pipeline,
# e.g. such as compositing tiles from multiple layers.
#
# It needs more work to make it more Pythonic, split it
# into more appropriate classes, add documentation, fix bugs etc.
#
# I'm not yet convinced this is the best approach to integrating
# Python with the render daemon. Two other options I'm considering:
#
# - Use the C renderd code with python binding allowing the replacement
# of just the core tile rendering code (this is the bit that people
# may want to tweak)
#
# - Split the functionality into a seperate queue handler daemon and
# render daemon. This would remove a lot of the complexity around the
# request handling which most people probably won't want to touch.
# The queue handler might stay in C with a smaller python rendering daemon

import sys, os
import SocketServer
import struct
import thread
import threading
import socket
import ConfigParser
import mapnik
import time
import errno
from math import pi,cos,sin,log,exp,atan
from StringIO import StringIO

import cairo
import cStringIO

MAX_ZOOM = 18
METATILE = 8
META_MAGIC = "META"

class protocol:
    # ENUM values for commandStatus field in protocol packet
    Ignore = 0
    Render = 1
    Dirty = 2
    Done = 3
    NotDone = 4

class ProtocolPacket:
    def __init__(self, version, fields = ""):
        self.version = version
        self.xmlname = ""
        self.x = 0
        self.y = 0
        self.z = 0
        self.mx = 0
        self.my = 0
        self.commandStatus = protocol.Ignore
        self.fields = fields

    def len(self):
        return struct.calcsize(self.fields)

    def bad_request(self):
        # Check that the requested (x,y,z) is invalid
        x = self.x
        y = self.y
        z = self.z

        if (z < 0) or (z > MAX_ZOOM):
            return True
        limit = (1 << z) -1
        if (x < 0) or (x > limit):
            return True
        if (y < 0) or (y > limit):
            return True
        return False

    def meta_tuple(self):
        # This metatile tuple is used to identify duplicate request in the rendering queue
        return (self.xmlname, self.mx, self.my, self.z)

class ProtocolPacketV1(ProtocolPacket):
    def __init__(self):
        ProtocolPacket(1)
        self.fields = "5i"

    def receive(self, data, dest):
        version, request, x, y, z = struct.unpack(self.fields, data)

        if version != 1:
            print "Received V1 packet with incorect version %d" % version
        else:
            #print "Got V1 request, command(%d), x(%d), y(%d), z(%d)" \
            #    % (request, x, y, z)
            self.commandStatus = request
            self.x = x
            self.y = y
            self.z = z
            self.xmlname = "default"
            # Calculate Meta-tile value for this x/y
            self.mx = x & ~(METATILE-1)
            self.my = y & ~(METATILE-1)
            self.dest = dest


    def send(self, status):
        x = self.x
        y = self.y
        z = self.z
        data = struct.pack(self.fields, (1, status, x, y, z))
        try: 
            self.dest.send(data)
        except socket.error, e:
               if e[0] != errno.EBADF:
                   raise


class ProtocolPacketV2(ProtocolPacket):
    def __init__(self):
        ProtocolPacket(2)
        self.fields = "5i41sxxx"

    def receive(self, data, dest):
        version, request, x, y, z, xmlname = struct.unpack(self.fields, data)

        if version != 2:
            print "Received V2 packet with incorect version %d" % version
        else:
            #print "Got V2 request, command(%d), xmlname(%s), x(%d), y(%d), z(%d)" \
            #    % (request, xmlname, x, y, z)
            self.commandStatus = request
            self.x = x
            self.y = y
            self.z = z
            self.xmlname = xmlname.rstrip('\000') # Remove trailing NULs
            # Calculate Meta-tile value for this x/y
            self.mx = x & ~(METATILE-1)
            self.my = y & ~(METATILE-1)
            self.dest = dest

    def send(self, status):
        x = self.x
        y = self.y
        z = self.z
        xmlname = self.xmlname
        data = struct.pack(self.fields, 2, status, x, y, z, xmlname)
        try:
            self.dest.send(data)
        except socket.error, e:
               if e[0] != errno.EBADF:
                   raise

DEG_TO_RAD = pi/180
RAD_TO_DEG = 180/pi


class SphericalProjection:
    def __init__(self,levels=18):
        self.Bc = []
        self.Cc = []
        self.zc = []
        self.Ac = []
        c = 256
        for d in range(0,levels+1):
            e = c/2;
            self.Bc.append(c/360.0)
            self.Cc.append(c/(2 * pi))
            self.zc.append((e,e))
            self.Ac.append(c)
            c *= 2

    def minmax(self, a,b,c):
        a = max(a,b)
        a = min(a,c)
        return a

    def fromLLtoPixel(self,ll,zoom):
         d = self.zc[zoom]
         e = round(d[0] + ll[0] * self.Bc[zoom])
         f = self.minmax(sin(DEG_TO_RAD * ll[1]),-0.9999,0.9999)
         g = round(d[1] + 0.5*log((1+f)/(1-f))*-self.Cc[zoom])
         return (e,g)

    def fromPixelToLL(self,px,zoom):
         e = self.zc[zoom]
         f = (px[0] - e[0])/self.Bc[zoom]
         g = (px[1] - e[1])/-self.Cc[zoom]
         h = RAD_TO_DEG * ( 2 * atan(exp(g)) - 0.5 * pi)
         return (f,h)


class RenderThread:
    def __init__(self, tile_path, styles, queue_handler):
        self.tile_path = tile_path
        self.queue_handler = queue_handler
        self.maps = {}
        self.prj = {}
        for xmlname in styles:
            #print "Creating Mapnik map object for %s with %s" % (xmlname, styles[xmlname])
            m = mapnik.Map(256, 256)
            self.maps[xmlname] = m
            # Load XML style
            mapnik.load_map(m, styles[xmlname], True)
            # Obtain <Map> projection
            self.prj[xmlname] = mapnik.Projection(m.srs)

        # Projects between tile pixel co-ordinates and LatLong (EPSG:4326)
        self.tileproj = SphericalProjection(MAX_ZOOM)

    def render_with_agg(self, m, size):
        # Render image with default Agg renderer
        im = mapnik.Image(size, size)
        mapnik.render(m, im)
        return im

    def render_with_cairo(self, m, size):
        surface = cairo.ImageSurface(cairo.FORMAT_ARGB32, size, size)
        mapnik.render(m, surface)
        return mapnik.Image.from_cairo(surface)

    def split_meta_image(self, im, sz, format = 'png256'):
        # Split image up into NxN grid of tile images
        tiles = {}
        for yy in range(0,sz):
            for xx in range(0,sz):
                view = im.view(xx * 256 , yy * 256, 256, 256)
                tile = view.tostring(format)
                tiles[(xx, yy)] = tile

        return tiles


    def render_meta(self, m, style, x, y, z, sz):
        # Calculate pixel positions of bottom-left & top-right
        p0 = (x * 256, (y + sz) * 256)
        p1 = ((x + sz) * 256, y * 256)

        # Convert to LatLong (EPSG:4326)
        l0 = self.tileproj.fromPixelToLL(p0, z);
        l1 = self.tileproj.fromPixelToLL(p1, z);

        # Convert to map projection (e.g. mercator co-ords EPSG:900913)
        c0 = self.prj[style].forward(mapnik.Coord(l0[0],l0[1]))
        c1 = self.prj[style].forward(mapnik.Coord(l1[0],l1[1]))

        # Bounding box for the meta-tile
        bbox = mapnik.Envelope(c0.x,c0.y, c1.x,c1.y)
        render_size = 256 * sz
        m.resize(render_size, render_size)
        m.zoom_to_box(bbox)
        m.buffer_size = 128

        im = self.render_with_agg(m, render_size)
        #im = self.render_with_cairo(m, render_size)
        return self.split_meta_image(im, sz)

    def render_request(self, t):
        (xmlname, x, y, z) = t
        # Calculate the meta tile size to use for this zoom level
        size = min(METATILE, 1 << z)
        try:
            m = self.maps[xmlname]
        except KeyError:
            print "No map for: '%s'" % xmlname
            return False
        tiles = self.render_meta(m, xmlname, x, y, z, size)
        self.meta_save(xmlname, x, y, z, size, tiles)

        print "Done xmlname(%s) z(%d) x(%d-%d) y(%d-%d)" % \
            (xmlname, z, x, x+size-1, y, y+size-1)

        return True;

    def xyz_to_meta(self, xmlname, x,y, z):
        mask = METATILE -1
        x &= ~mask
        y &= ~mask
        hashes = {}

        for i in range(0,5):
            hashes[i] = ((x & 0x0f) << 4) | (y & 0x0f)
            x >>= 4
            y >>= 4

        meta = "%s/%s/%d/%u/%u/%u/%u/%u.meta" % (self.tile_path, xmlname, z, hashes[4], hashes[3], hashes[2], hashes[1], hashes[0])
        return meta

    def xyz_to_meta_offset(self, xmlname, x,y, z):
        mask = METATILE -1
        offset = (x & mask) * METATILE + (y & mask)
        return offset


    def meta_save(self, xmlname, x, y, z, size, tiles):
        #print "Saving %d tiles" % (size * size)
        meta_path = self.xyz_to_meta(xmlname, x, y, z)
        d = os.path.dirname(meta_path)
        if not os.path.exists(d):
            try:
                os.makedirs(d)
            except OSError:
                # Multiple threads can race when creating directories,
                # ignore exception if the directory now exists
                if not os.path.exists(d):
                    raise

        tmp = "%s.tmp.%d" % (meta_path, thread.get_ident())
        f = open(tmp, "w")

        f.write(struct.pack("4s4i", META_MAGIC, METATILE * METATILE, x, y, z))
        offset = len(META_MAGIC) + 4 * 4
        # Need to pre-compensate the offsets for the size of the offset/size table we are about to write
        offset += (2 * 4) * (METATILE * METATILE)
        # Collect all the tile sizes
        sizes = {}
        offsets = {}
        for xx in range(0, size):
            for yy in range(0, size):
                mt = self.xyz_to_meta_offset(xmlname, x+xx, y+yy, z)
                sizes[mt] = len(tiles[(xx, yy)])
                offsets[mt] = offset
                offset += sizes[mt]
        # Write out the offset/size table
        for mt in range(0, METATILE * METATILE):
            if mt in sizes:
                f.write(struct.pack("2i", offsets[mt], sizes[mt]))
            else:
                f.write(struct.pack("2i", 0, 0))
        # Write out the tiles
        for xx in range(0, size):
            for yy in range(0, size):
                f.write(tiles[(xx, yy)])

        f.close()
        os.rename(tmp, meta_path)
        #print "Wrote: %s" % meta_path

    def loop(self):
        while True:
            #Fetch a meta-tile to render
            r = self.queue_handler.fetch()
            rendered = self.render_request(r)
            # Retrieve all requests for this meta-tile
            requests = self.queue_handler.pop_requests(r)
            for request in requests:
                if (request.commandStatus == protocol.Render):
                    if rendered == True:
                        request.send(protocol.Done)
                    else:
                        request.send(protocol.NotDone)


def start_renderers(num_threads, tile_path, styles, queue_handler):
    for i in range(num_threads):
        renderer = RenderThread(tile_path, styles, queue_handler)
        render_thread = threading.Thread(target=renderer.loop)
        render_thread.setDaemon(True)
        render_thread.start()
        print "Started render thread %s" % render_thread.getName()

class RequestQueues:
    def __init__(self, request_limit = 32, dirty_limit = 1000):
        # We store requests in several lists
        # - Incoming render requests are initally put into the request queue
        # If the request queue is full then the new request is demoted to the dirty queue
        # - Incoming 'dirty' requests are put into the dirty queue, or dropped if this is full
        # - The render queue holds the requests which are in progress by the render threads
        self.requests = {}
        self.dirties = {}
        self.rendering = {}

        self.request_limit = request_limit
        self.dirty_limit = dirty_limit
        self.not_empty = threading.Condition()


    def add(self, request):
        self.not_empty.acquire()
        try:
            # Before adding this new request we first look if this tile is already pending
            # If so, the new request is tacked on to the existing one
            # FIXME: Add short-circuit for overload condition?
            t = request.meta_tuple()
            if t in self.rendering:
                self.rendering[t].append(request)
                return "rendering"
            if t in self.requests:
                self.requests[t].append(request)
                return "requested"
            if t in self.dirties:
                self.dirties[t].append(request)
                return "dirty"
            # If we've reached here then there are no existing requests for this tile
            if (request.commandStatus == protocol.Render) and (len(self.requests) < self.request_limit):
                self.requests[t] = [request]
                self.not_empty.notify()
                return "requested"
            if len(self.dirties) < self.dirty_limit:
                self.dirties[t] = [request]
                self.not_empty.notify()
                return "dirty"
            return "dropped"
        finally:
            self.not_empty.release()


    def fetch(self):
        # Fetches a request tuple from the request or dirty queue
        # The requests are moved to the rendering queue while they are being rendered
        self.not_empty.acquire()
        try:
            while (len(self.requests) == 0) and (len(self.dirties) == 0):
                self.not_empty.wait()
            # Pull request from one of the incoming queues
            try:
                item = self.requests.popitem()
            except KeyError:
                try:
                    item = self.dirties.popitem()
                except KeyError:
                    print "Odd, queues empty"
                    return

            t = item[0]
            self.rendering[t] = item[1]
            return t
        finally:
            self.not_empty.release()

    def pop_requests(self, t):
        # Removes this tuple from the rendering queue
        # and returns the list of request for the tuple
        self.not_empty.acquire()
        try:
            return self.rendering.pop(t)
        except KeyError:
            # Should never happen. It implies the requests queues are broken
            print "WARNING: Failed to locate request in rendering list!"
        finally:
            self.not_empty.release()


class ThreadedUnixStreamHandler(SocketServer.BaseRequestHandler):

    def rx_request(self, request):
        if (request.commandStatus != protocol.Render) \
           and (request.commandStatus != protocol.Dirty):
               return

        if request.bad_request():
            if (request.commandStatus == protocol.Render):
                request.send(protocol.NotDone)
            return

        #cur_thread = threading.currentThread()
        #print "%s: xml(%s) z(%d) x(%d) y(%d)" % \
        #    (cur_thread.getName(), request.xmlname, request.z, request.x, request.y)

        status = self.server.queue_handler.add(request)
        if status in ("rendering", "requested"):
            # Request queued, response will be sent on completion
            return

        # The tile won't be rendered soon, tell the requestor straight away
        if (request.commandStatus == protocol.Render):
            request.send(protocol.NotDone)

    def handle(self):
        cur_thread = threading.currentThread()
        #print "%s: New connection" % cur_thread.getName()
        len_v1 = ProtocolPacketV1().len()
        len_v2 = ProtocolPacketV2().len()
        max_len = max(len_v1, len_v2)

        while True:
            try:
                data = self.request.recv(max_len)
            except socket.error, e:
                if e[0] == errno.ECONNRESET:
                    #print "Connection reset by peer"
                    break
                else:
                    raise

            if len(data) == len_v1:
                req_v1 = ProtocolPacketV1()
                req_v1.receive(data, self.request)
                self.rx_request(req_v1)
            if len(data) == len_v2:
                req_v2 = ProtocolPacketV2()
                req_v2.receive(data, self.request)
                self.rx_request(req_v2)
            elif len(data) == 0:
                #print "%s: Connection closed" % cur_thread.getName()
                break
            else:
                print "Invalid request length %d" % len(data)
                break

class ThreadedUnixStreamServer(SocketServer.ThreadingMixIn, SocketServer.UnixStreamServer):
    def __init__(self, address, queue_handler, handler):
        if(os.path.exists(address)):
           os.unlink(address)
        self.address = address
        self.queue_handler = queue_handler
        SocketServer.UnixStreamServer.__init__(self, address, handler)
        self.daemon_threads = True

def listener(address, queue_handler):
    # Create the server
    server = ThreadedUnixStreamServer(address, queue_handler, ThreadedUnixStreamHandler)
    # The socket needs to be writeable by Apache
    os.chmod(address, 0666)
    # Loop forever servicing requests
    server.serve_forever()

def display_config(config):
    for xmlname in config.sections():
        if xmlname != "renderd" and xmlname != "mapnik":
            print "Layer name: %s" % xmlname
            uri = config.get(xmlname, "uri")
            xml = config.get(xmlname, "xml")
            print "    URI(%s) = XML(%s)" % (uri, xml)

def read_styles(config):
    styles = {}
    for xmlname in config.sections():
        if xmlname != "renderd" and xmlname != "mapnik":
            styles[xmlname] = config.get(xmlname, "xml")
    return styles

if __name__ == "__main__":
    try:
        cfg_file = os.environ['RENDERD_CFG']
    except KeyError:
        cfg_file = "/etc/renderd.conf"

    # Unifont has a better character coverage and is used as a fallback for DejaVu
    # if you use a style based on the osm-template-fonset.xml
    mapnik.FontEngine.instance().register_font("/home/jburgess/osm/fonts/unifont-5.1.20080706.ttf")

    default_cfg = StringIO("""
[renderd]
socketname=/tmp/osm-renderd
num_threads=4
tile_dir=/var/lib/mod_tile
""")

    config = ConfigParser.ConfigParser()
    config.readfp(default_cfg)
    config.read(cfg_file)
    display_config(config)
    styles = read_styles(config)

    num_threads    = config.getint("renderd", "num_threads")
    renderd_socket = config.get("renderd", "socketname")
    tile_dir       = config.get("renderd", "tile_dir")

    queue_handler = RequestQueues()
    start_renderers(num_threads, tile_dir, styles, queue_handler)
    listener(renderd_socket, queue_handler)