modes_kml: fix case where KML generation thread dies

SQLite does not deal well with concurrency.  Avoid database locked
errors by synchronizing database access using a threading.Lock().
This commit is contained in:
Johnathan Corgan 2012-10-06 08:44:33 -07:00
parent d88d21f672
commit c0d24f12c9
2 changed files with 19 additions and 12 deletions

View File

@ -93,14 +93,14 @@ class modes_kml(threading.Thread, modes_output_sql):
#read the database and add KML #read the database and add KML
q = "select distinct icao from positions where seen > datetime('now', '-5 minute')" q = "select distinct icao from positions where seen > datetime('now', '-5 minute')"
c = self._db.cursor() c = self._db.cursor()
c.execute(q) self.locked_execute(c, q)
icaolist = c.fetchall() icaolist = c.fetchall()
#now we have a list icaolist of all ICAOs seen in the last 5 minutes #now we have a list icaolist of all ICAOs seen in the last 5 minutes
for icao in icaolist: for icao in icaolist:
#print "ICAO: %x" % icao #print "ICAO: %x" % icao
q = "select * from positions where icao=%i and seen > datetime('now', '-2 hour') ORDER BY seen DESC" % icao q = "select * from positions where icao=%i and seen > datetime('now', '-2 hour') ORDER BY seen DESC" % icao
c.execute(q) self.locked_execute(c, q)
track = c.fetchall() track = c.fetchall()
#print "Track length: %i" % len(track) #print "Track length: %i" % len(track)
if len(track) != 0: if len(track) != 0:
@ -128,7 +128,7 @@ class modes_kml(threading.Thread, modes_output_sql):
#now get metadata #now get metadata
q = "select ident from ident where icao=%i" % icao q = "select ident from ident where icao=%i" % icao
c.execute(q) self.locked_execute(c, q)
r = c.fetchall() r = c.fetchall()
if len(r) != 0: if len(r) != 0:
ident = r[0][0] ident = r[0][0]
@ -136,7 +136,7 @@ class modes_kml(threading.Thread, modes_output_sql):
#if ident is None: ident = "" #if ident is None: ident = ""
#get most recent speed/heading/vertical #get most recent speed/heading/vertical
q = "select seen, speed, heading, vertical from vectors where icao=%i order by seen desc limit 1" % icao q = "select seen, speed, heading, vertical from vectors where icao=%i order by seen desc limit 1" % icao
c.execute(q) self.locked_execute(c, q)
r = c.fetchall() r = c.fetchall()
if len(r) != 0: if len(r) != 0:
seen = r[0][0] seen = r[0][0]

View File

@ -19,7 +19,7 @@
# Boston, MA 02110-1301, USA. # Boston, MA 02110-1301, USA.
# #
import time, os, sys import time, os, sys, threading
from string import split, join from string import split, join
import modes_parse import modes_parse
import sqlite3 import sqlite3
@ -28,6 +28,8 @@ from modes_exceptions import *
class modes_output_sql(modes_parse.modes_parse): class modes_output_sql(modes_parse.modes_parse):
def __init__(self, mypos, filename): def __init__(self, mypos, filename):
modes_parse.modes_parse.__init__(self, mypos) modes_parse.modes_parse.__init__(self, mypos)
self._lock = threading.Lock()
#create the database #create the database
self.db = sqlite3.connect(filename) self.db = sqlite3.connect(filename)
#now execute a schema to create the tables you need #now execute a schema to create the tables you need
@ -39,7 +41,7 @@ class modes_output_sql(modes_parse.modes_parse):
"lat" REAL, "lat" REAL,
"lon" REAL "lon" REAL
);""" );"""
c.execute(query) self.locked_execute(c, query)
query = """CREATE TABLE IF NOT EXISTS "vectors" ( query = """CREATE TABLE IF NOT EXISTS "vectors" (
"icao" INTEGER KEY NOT NULL, "icao" INTEGER KEY NOT NULL,
"seen" TEXT NOT NULL, "seen" TEXT NOT NULL,
@ -47,26 +49,31 @@ class modes_output_sql(modes_parse.modes_parse):
"heading" REAL, "heading" REAL,
"vertical" REAL "vertical" REAL
);""" );"""
c.execute(query) self.locked_execute(c, query)
query = """CREATE TABLE IF NOT EXISTS "ident" ( query = """CREATE TABLE IF NOT EXISTS "ident" (
"icao" INTEGER PRIMARY KEY NOT NULL, "icao" INTEGER PRIMARY KEY NOT NULL,
"ident" TEXT NOT NULL "ident" TEXT NOT NULL
);""" );"""
c.execute(query) self.locked_execute(c, query)
c.close() c.close()
self.db.commit() self.db.commit()
def __del__(self): def __del__(self):
self.db.close() self.db.close()
def locked_execute(self, c, query):
with self._lock:
c.execute(query)
def output(self, message): def output(self, message):
try: try:
query = self.make_insert_query(message) query = self.make_insert_query(message)
if query is not None: if query is not None:
c = self.db.cursor() with self._lock:
c.execute(query) c = self.db.cursor()
c.close() c.execute(query)
self.db.commit() c.close()
self.db.commit()
except ADSBError: except ADSBError:
pass pass