1 import time, threading, struct
3 from bsddb3 import db as bd
5 deadlock = bd.DBLockDeadlockError
7 class environment(lib.closable):
8 def __init__(self, path, *, create=True, recover=False, mode=0o666):
10 self.env.set_lk_detect(bd.DB_LOCK_RANDOM)
11 fl = bd.DB_THREAD | bd.DB_INIT_MPOOL | bd.DB_INIT_LOCK | bd.DB_INIT_LOG | bd.DB_INIT_TXN
16 self.env.open(path, fl, mode)
17 self.lastckp = self.lastarch = time.time()
18 self.lock = threading.Lock()
31 if now - self.lastckp > 60:
32 self.env.txn_checkpoint(1024)
34 if now - self.lastarch > 3600:
35 self.env.log_archive(bd.DB_ARCH_REMOVE)
40 def db(self, name, create=True, mode=0o666):
42 if name not in self.dbs:
43 self.dbs[name] = database(self, name, create, mode)
50 def __exit__(self, *excinfo):
54 def opendb(env, fnm, dnm, typ, fl, mode):
58 self.main.open(fnm, dnm, typ, fl, mode)
64 def __init__(self, env, flags=bd.DB_TXN_WRITE_NOSYNC):
65 self.tx = env.txn_begin(None, flags)
79 def __exit__(self, etype, exc, tb):
86 def wrapper(self, *args, tx=None, **kwargs):
90 with txn(envfun(self)) as ltx:
91 ret = fun(self, *args, tx=ltx, **kwargs)
97 return fun(self, *args, tx=tx, **kwargs)
101 class database(object):
102 def __init__(self, env, name, create, mode):
106 fl = bd.DB_THREAD | bd.DB_AUTO_COMMIT
109 self.cf = self._opendb("cf", bd.DB_HASH, fl)
110 self.ob = self._opendb("ob", bd.DB_HASH, fl)
112 def _opendb(self, dnm, typ, fl, init=None):
113 ret = bd.DB(self.env.env)
117 ret.open(self.fnm, dnm, typ, fl, self.mode)
122 @txnfun(lambda self: self.env.env)
123 def _nextseq(self, *, tx):
124 if self.cf.has_key(b"seq", txn=tx.tx):
125 seq = struct.unpack(">Q", self.cf.get(b"seq", txn=tx.tx))[0]
128 self.cf.put(b"seq", struct.pack(">Q", seq + 1), txn=tx.tx)
131 @txnfun(lambda self: self.env.env)
132 def add(self, ob, *, tx):
133 seq = self._nextseq(tx=tx)
134 self.ob.put(struct.pack(">Q", seq), ob, txn=tx.tx, flags=bd.DB_NOOVERWRITE)
137 @txnfun(lambda self: self.env.env)
138 def replace(self, id, ob, *, tx):
139 key = struct.pack(">Q", id)
140 if not self.ob.has_key(key, txn=tx.tx):
142 self.ob.put(key, ob, txn=tx.tx)
144 @txnfun(lambda self: self.env.env)
145 def get(self, id, *, tx):
146 ret = self.ob.get(struct.pack(">Q", id), None)
151 @txnfun(lambda self: self.env.env)
152 def remove(self, id, *, tx):
153 key = struct.pack(">Q", id)
154 if not self.ob.has_key(key, txn=tx.tx):
156 self.ob.delete(key, txn=tx.tx)