mtsatellite/cmd/mtwebmapper/tilesupdater.go

170 lines
3.6 KiB
Go
Raw Normal View History

// Copyright 2014 by Sascha L. Teichmann
// Use of this source code is governed by the MIT license
// that can be found in the LICENSE file.
package main
import (
"encoding/json"
"log"
"net/http"
"path/filepath"
"sync"
"bitbucket.org/s_l_teichmann/mtredisalize/common"
)
type tileUpdater struct {
changes map[xz]bool
mapDir string
redisAddress string
colors *common.Colors
workers int
cond *sync.Cond
mu sync.Mutex
}
type xz struct {
X int16
Z int16
}
type xzm struct {
P xz
Mask uint16
}
func (c xz) quantize() xz {
2014-09-20 21:57:01 +02:00
return xz{X: (c.X - -1933) / 16, Z: (c.Z - -1933) / 16}
}
func (c xz) dequantize() xz {
return xz{X: c.X*16 + -1933, Z: c.Z*16 + -1933}
}
func (c xz) parent() xzm {
xp, xr := c.X>>1, uint16(c.X&1)
zp, zr := c.Z>>1, uint16(c.Z&1)
return xzm{
P: xz{X: xp, Z: zp},
Mask: ((1 << xr) << 2) | (1 << zr)}
}
func newTileUpdater(mapDir, redisAddress string, colors *common.Colors, workers int) *tileUpdater {
tu := tileUpdater{
mapDir: mapDir,
redisAddress: redisAddress,
changes: map[xz]bool{},
colors: colors,
workers: workers}
tu.cond = sync.NewCond(&tu.mu)
return &tu
}
func (tu *tileUpdater) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
var err error
var newChanges []xz
decoder := json.NewDecoder(r.Body)
if err = decoder.Decode(&newChanges); err != nil {
log.Printf("WARN: JSON document broken: %s", err)
http.Error(rw, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
return
}
if len(newChanges) > 0 {
tu.cond.L.Lock()
for _, c := range newChanges {
tu.changes[c.quantize()] = true
}
tu.cond.L.Unlock()
tu.cond.Signal()
}
rw.WriteHeader(http.StatusOK)
}
func (tu *tileUpdater) doUpdates() {
for {
var changes map[xz]bool
tu.cond.L.Lock()
for len(tu.changes) == 0 {
tu.cond.Wait()
}
changes = tu.changes
tu.changes = map[xz]bool{}
tu.cond.L.Unlock()
baseDir := filepath.Join(tu.mapDir, "8")
jobs := make(chan xz)
var done sync.WaitGroup
for i, n := 0, min(tu.workers, len(changes)); i < n; i++ {
var client *common.RedisClient
var err error
if client, err = common.NewRedisClient("tcp", tu.redisAddress); err != nil {
log.Printf("WARN: Cannot connect to redis server: %s", err)
continue
}
btc := common.NewBaseTileCreator(client, tu.colors, baseDir, true)
done.Add(1)
go updateBaseTiles(jobs, btc, &done)
}
parentJobs := make(map[xz]uint16)
for c, _ := range changes {
//log.Printf("job: %+v", c)
jobs <- c
pxz := c.parent()
parentJobs[pxz.P] |= pxz.Mask
}
close(jobs)
done.Wait()
for level := 7; level >= 0; level-- {
pJobs := make(chan xzm)
for i, n := 0, min(len(parentJobs), tu.workers); i < n; i++ {
done.Add(1)
go updatePyramidTiles(level, tu.mapDir, pJobs, &done)
}
ppJobs := make(map[xz]uint16)
for c, mask := range parentJobs {
pJobs <- xzm{P: c, Mask: mask}
pxz := c.parent()
ppJobs[pxz.P] |= pxz.Mask
}
close(pJobs)
done.Wait()
parentJobs = ppJobs
}
}
}
func min(a, b int) int {
if a < b {
return a
}
return b
}
func updatePyramidTiles(level int, baseDir string, jobs chan xzm, done *sync.WaitGroup) {
defer done.Done()
for job := range jobs {
_ = job
}
}
func updateBaseTiles(jobs chan xz, btc *common.BaseTileCreator, done *sync.WaitGroup) {
defer btc.Close()
defer done.Done()
for job := range jobs {
xz := job.dequantize()
//log.Printf("%d/%d %d/%d", x, z, job.X, job.Z)
if err := btc.CreateTile(xz.X-1, xz.Z-1, int(job.X), int(job.Z)); err != nil {
log.Printf("WARN: create tile failed: %s", err)
}
}
}