Bitmessage/MiNode
3
0
Fork 1
Code Pull Requests 9 Actions Activity

WIP: Implementing sqlite objects storage #13

Draft
lee.miller wants to merge 11 commits from lee.miller/MiNode:sqlite into v0.3
pull from: lee.miller/MiNode:sqlite
merge into: Bitmessage:v0.3
Conversation 0 Commits 11 Files Changed 10 +25 -14
1 changed files with 25 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 60d08571b1 - Show all commits

39
minode/sql.py
View File

@ -1,5 +1,6 @@
"""Inventory implementation using sqlite""" """Inventory implementation using sqlite"""
import logging
import os import os
import sqlite3 import sqlite3
import threading import threading
@ -14,6 +15,7 @@ class Inventory():
"""sqlite inventory""" """sqlite inventory"""
def __init__(self): def __init__(self):
self._lock = threading.Lock() self._lock = threading.Lock()
self._deleted = 0
self._db = sqlite3.connect( self._db = sqlite3.connect(
os.path.join(shared.data_directory, 'objects.dat'), os.path.join(shared.data_directory, 'objects.dat'),
check_same_thread=False check_same_thread=False
@ -29,17 +31,16 @@ class Inventory():
COMMIT; COMMIT;
""") """)
self.rowid = len(self) or None self.rowid = len(self) or None
cur = self._db.cursor()
cur.execute("SELECT value FROM status WHERE key='lastvacuumtime'")
now = int(time.time())
try: try:
vacuumed = cur.fetchone()[0] self.lastvacuumtime = self._db.execute(
"SELECT value FROM status WHERE key='lastvacuumtime'"
).fetchone()[0]
except TypeError: except TypeError:
pass self.lastvacuumtime = int(time.time())
else: self._db.execute(
if vacuumed < now - 86400: # 24 hours "INSERT INTO status VALUES ('lastvacuumtime', ?)",
cur.execute('VACUUM') (self.lastvacuumtime,)
cur.execute("INSERT INTO status VALUES ('lastvacuumtime', ?)", (now,)) )
self._db.commit() self._db.commit()
self._db.row_factory = self.__object self._db.row_factory = self.__object
@ -55,12 +56,22 @@ class Inventory():
def cleanup(self): def cleanup(self):
"""Remove expired objects""" """Remove expired objects"""
with self._lock: with self._lock:
self._db.execute( now = int(time.time())
'DELETE FROM objects WHERE expires < ?', cur = self._db.execute(
(int(time.time()) - 3 * 3600,) 'DELETE FROM objects WHERE expires < ?', (now - 3 * 3600,))
)
self._db.commit() self._db.commit()
# conditional vacuum and validity check self._deleted += cur.rowcount
logging.debug('Deleted %s expired objects', cur.rowcount)
# conditional vacuum and validity check (TODO)
# every 24 hours or after deleting a lot of items
if self._deleted > 10000 or self.lastvacuumtime < now - 86400:
logging.info('Doing VACUUM for objects')
cur.execute('VACUUM')
cur.execute(
"INSERT INTO status VALUES ('lastvacuumtime', ?)", (now,))
self._db.commit()
self._deleted = 0
self.lastvacuumtime = now
def filter(self, stream=None, object_type=None, tag=None): def filter(self, stream=None, object_type=None, tag=None):
"""Generator of objects with the given parameters""" """Generator of objects with the given parameters"""