Skip to content

Z
Zope
Commit 866320f5 authored by Jim Fulton's avatar Jim Fulton
Browse files
Options

Added and tested Packless storage, which paves the way to 

new GC strategy.

Meaningless checkpoint of other files.
parent 1f8ea369
Hide whitespace changes
Inline Side-by-side
Showing with 755 additions and 156 deletions
lib/python/BDBStorage/BDBFullStorage.py
View file @ 866320f5
from base import Base
from bsddb3 import db
from struct import pack, unpack
import os, tempfile
from ZODB import POSException
def opendb(env, prefix):
d=db.Db(env)
d.open(self._prefix+'mini', db.DB_BTREE, db.DB_CREATE)
import os, tempfile, string, marshal
from ZODB import POSException, utils
from marshal import dump, load
class Full(Base):
def _setupDbs(self):
# Supports Base framework
self._index=self._setupDB('current')
for name in 'pickle', 'record', 'trans', 'vids', 'versions':
for name in 'pickle', 'record', 'transactions', 'vids', 'versions':
self._setupDB(name)
self._setupDB('currentVersions', flags=db.DB_DUP)
self._setupDB('transaction_oids', flags=db.DB_DUP)
c=self._vids.cursor()
v=c.get(db.DB_LAST)
if v: self._vid=utils.U64(v[0])
else: self._vid=0L
def _dbnames(self):
return ('current', 'pickle', 'record', 'trans',
# Supports Base framework
return ('current', 'pickle', 'record',
'transactions', 'transaction_oids',
'vids', 'versions', 'currentVersions')
def abortVersion(self, src, transaction):
......@@ -80,8 +87,6 @@ class Full(Base):
try:
newtid=self._serial
vid=self._vids[src]
try: dvid=self._vids[dest]
except KeyError: dvid=self._newvid(dest)
oids=[]; save_oid=oids.append
c=self._currentVersions.cursor()
......@@ -92,6 +97,12 @@ class Full(Base):
tmp=self._tmp
dump=marshal.dump
zero="\0\0\0\0\0\0\0\0"
try: dvid=self._vids[dest]
except KeyError:
dvid=self._newvid()
dump(('v',(vid, version)), tmp)
while i:
v, oid = i
......@@ -135,16 +146,20 @@ class Full(Base):
data = self._record[oid+serial][16:24]
return self._pickle[oid+data]
finally: self._lock_release()
def modifiedInVersion(self, oid):
self._lock_acquire()
try:
t=self._index[oid]
vid = self._record[oid+t][:8]
tid=self._current[oid]
vid=self._record[oid+tid][:8]
if vid == '\0\0\0\0\0\0\0\0': return ''
return self._versions[vid]
finally: self._lock_release()
def _newvid(self):
self._vid=self._vid+1
return utils.p64(self._vid)
def store(self, oid, serial, data, version, transaction):
if transaction is not self._transaction:
raise POSException.StorageTransactionError(self, transaction)
......@@ -155,6 +170,7 @@ class Full(Base):
try: vid=self._vids[version]
except:
vid=self._newvid()
dump(('v',(vid, version)), self._tmp)
else:
vid=nv='\0\0\0\0\0\0\0\0'
......@@ -181,7 +197,7 @@ class Full(Base):
def supportsVersions(self): return 1
def _finish(self, tid, u, d, e):
txn = self._env.txn_begin()
txn=self._env.txn_begin()
try:
tmp=self._tmp
ltmp=tmp.tell()
......@@ -191,9 +207,10 @@ class Full(Base):
records_put=self._records.put
pickles_put=self._pickle.put
current_put=self._current.put
transaction_oids_put=self._transaction_oids.put
currentVersions_put=self._currentVersions.put
l=pack(">HI",len(u), len(d))
self._trans.put(tid, l+u+d+e, txn)
self._transactions.put(tid, ' '+l+u+d+e, txn)
while ltmp:
try: op, arg = load(tmp)
except EOFError:
......@@ -201,6 +218,7 @@ class Full(Base):
else: raise
else:
if op=='s':
# store data
oid, vid, nv, back, data, pre = arg
key=oid+tid
if data:
......@@ -212,70 +230,185 @@ class Full(Base):
if vid != '/0/0/0/0/0/0/0/0':
versions_put(vid, oid, txn)
current_put(oid, tid, txn)
transaction_oids_put(tid, oid, txn)
elif op='d':
# discard a version (part of version commit and abort)
self._currentVersions.delete(arg, txn)
elif op='v':
# save a version definition
vid, version = arg
self._versions.put(vid, version, txn)
self._vids.put(version, vid, txn)
except:
txn.abort()
raise
else:
txn.commit()
def _undoable(self, txn):
txn.abort()
raise POSException.UndoError, 'Undoable transaction'
def undo(self, tid):
self._lock_acquire()
try:
status = self._transactions[tid][:1]
if status == 'p':
raise POSException.UndoError, 'Undoable transaction'
txn=self._env.txn_begin()
current=self._current
record=self._record
pickle=self._pickle
currentVersions=self._currentVersions
unpack=struct.unpack
try:
for oid in dups(self._transaction_oids, tid, txn):
if current.get(oid, txn) != tid: self._undoable(txn)
key=oid+tid
vid, nv, data, pre = unpack("8s8s8s8s",
record.get(key, txn))
record.delete(key, txn)
if data==tid: pickle.delete(key, txn)
if pre == '\0\0\0\0\0\0\0\0':
current.delete(oid, txn)
else:
current.put(oid, pre, txn)
try: pvid=record.get(oid+pre, txn)
except KeyError: self._undoable(txn)
if pvid != vid:
if vid != '\0\0\0\0\0\0\0\0':
del_dup(currentVersions, vid, oid, txn)
if pvid != '\0\0\0\0\0\0\0\0':
currentVersions.put(pvid, oid, txn)
self._transactions.delete(tid, txn)
self._transaction_oids.delete(tid, txn)
except:
txn.abort()
raise
else:
txn.commit()
finally: self._lock_release()
def undoLog(self, first, last, filter=None):
self._lock_acquire()
try:
c=self._transactions.cursor()
try:
i=0; r=[]; a=r.append
data=c.get(db.DB_LAST)
while data and i < last:
tid, data = data
status = data[:1]
if status == 'p': break
luser, ldesc = unpack("HI", data[1:17])
user=data[17:luser+17]
desc=data[luser+17:luser+17+ldesc]
ext=data[luser+17+ldesc:]
data={'id': tid,
'time': TimeStamp(tid).timeTime(),
'user_name': user or '',
'description': desc or '',
}
if ext:
try:
ext=loads(ext)
data.update(ext)
except: pass
if filter is None or filter(data):
if i >= first: a(data)
i=i+1
data=c.get(db.DB_PREV)
return r
finally: c.close()
finally: self._lock_release()
def versionEmpty(self, version):
self._lock_acquire()
try:
try: self._currentVersions[self._vids[version]]
except KeyError: return 1
else: return 0
finally: self._lock_release()
def tpc_vote(self, transaction):
if transaction is not self._transaction:
raise POSException.StorageTransactionError(self, transaction)
def versions(self, max=None):
self._lock_acquire()
try:
c=self._currentVersions.cursor()
try:
try: data=c.get(db.DB_NEXT_NODUP)
except: return ()
r=[]; a=r.append
while data:
a(data[0])
data=c.get(db.DB_NEXT_NODUP)
return r
finally: c.close()
finally: self._lock_release()
def history(self, oid, version=None, length=1, filter=None):
self._lock_acquire()
try:
txn = self._txn = self._env.txn_begin()
put=self._index.put
serial=self._serial
tmp=self._tmp
s=tmp.tell()
tmp.seek(0)
read=tmp.read
l=0
while l < s:
oid, ldata = unpack(">8sI", read(12))
data=read(ldata)
l=l+ldata+12
if ldata > s:
raise 'Temporary file corrupted'
put(oid, serial+data, txn)
tmp.seek(0)
if s > 999999: tmp.truncate()
tid=self._current[oid]
finally: self._lock_release()
def pack(self, t, referencesf):
self._lock_acquire()
try:
# Build an index of *only* those objects reachable
# from the root.
index=self._index
rootl=['\0\0\0\0\0\0\0\0']
pop=rootl.pop
pindex={}
referenced=pindex.has_key
while rootl:
oid=pop()
if referenced(oid): continue
# Scan non-version pickle for references
r=index[oid]
pindex[oid]=r
p=r[8:]
referencesf(p, rootl)
# Now delete any unreferenced entries:
for oid in index.keys():
if not referenced(oid): del index[oid]
finally: self._lock_release()
class dups:
"""Iterator for duplicate-record databases"
def __init__(self, db, key, txn=0):
if txn==0:
c=db.cursor()
else:
c=db.cursor(txn)
self._v=c.set(key)
self._g=c.get
self._i=0
def __getitem__(self, index):
i=self._i
if index==i: return self._v
if index < i or i < 0: raise IndexError, index
while i < index:
v=self._g(db.DB_NEXT_DUP)
if v:
i=i+1
else:
self._i=-1
raise IndexError, index
self._i=i
self._v=v
return v
def del_dup(database, key, value, txn):
c=database.cursor(txn)
try:
c.getBoth(key, value)
c.delete()
finally:
c.close()
lib/python/BDBStorage/BDBMinimalStorage.py
View file @ 866320f5
......@@ -5,13 +5,16 @@ from struct import pack, unpack
class Minimal(Base):
def _setupDbs(self):
self._index=self._setupDB('mini')
# Supports Base framework
self._index=self._setupDB('current')
self._setupDB('pickle')
def load(self, oid, version):
self._lock_acquire()
try:
p=self._index[oid]
return p[8:], p[:8] # pickle, serial
s=self._index[oid]
p=self._pickle[oid]
return p, s # pickle, serial
finally: self._lock_release()
def store(self, oid, serial, data, version, transaction):
......@@ -24,8 +27,7 @@ class Minimal(Base):
self._lock_acquire()
try:
if self._index.has_key(oid):
old=self._index[oid]
oserial=old[:8]
oserial=self._index[oid]
if serial != oserial: raise POSException.ConflictError
serial=self._serial
......@@ -35,14 +37,11 @@ class Minimal(Base):
return serial
def tpc_vote(self, transaction):
if transaction is not self._transaction:
raise POSException.StorageTransactionError(self, transaction)
self._lock_acquire()
def _finish(self, tid, u, d, e):
txn = self._env.txn_begin()
try:
txn = self._txn = self._env.txn_begin()
put=self._index.put
serial_put=self._index.put
pickle_put=self._pickle.put
serial=self._serial
tmp=self._tmp
s=tmp.tell()
......@@ -55,20 +54,25 @@ class Minimal(Base):
l=l+ldata+12
if ldata > s:
raise 'Temporary file corrupted'
put(oid, serial+data, txn)
serial_put(oid, serial, txn)
pickle_put(oid, data, txn)
tmp.seek(0)
if s > 999999: tmp.truncate()
finally: self._lock_release()
except:
txn.abort()
raise
else:
txn.commit()
def pack(self, t, referencesf):
self._lock_acquire()
try:
try:
# Build an index of *only* those objects reachable
# from the root.
index=self._index
index=self._pickle
rootl=['\0\0\0\0\0\0\0\0']
pop=rootl.pop
pindex={}
......@@ -78,9 +82,8 @@ class Minimal(Base):
if referenced(oid): continue
# Scan non-version pickle for references
r=index[oid]
pindex[oid]=r
p=r[8:]
p=index[oid]
pindex[oid]=1
referencesf(p, rootl)
# Now delete any unreferenced entries:
......@@ -88,3 +91,12 @@ class Minimal(Base):
if not referenced(oid): del index[oid]
finally: self._lock_release()
lib/python/BDBStorage/Full.py
View file @ 866320f5
from base import Base
from bsddb3 import db
from struct import pack, unpack
import os, tempfile
from ZODB import POSException
def opendb(env, prefix):
d=db.Db(env)
d.open(self._prefix+'mini', db.DB_BTREE, db.DB_CREATE)
import os, tempfile, string, marshal
from ZODB import POSException, utils
from marshal import dump, load
class Full(Base):
def _setupDbs(self):
# Supports Base framework
self._index=self._setupDB('current')
for name in 'pickle', 'record', 'trans', 'vids', 'versions':
for name in 'pickle', 'record', 'transactions', 'vids', 'versions':
self._setupDB(name)
self._setupDB('currentVersions', flags=db.DB_DUP)
self._setupDB('transaction_oids', flags=db.DB_DUP)
c=self._vids.cursor()
v=c.get(db.DB_LAST)
if v: self._vid=utils.U64(v[0])
else: self._vid=0L
def _dbnames(self):
return ('current', 'pickle', 'record', 'trans',
# Supports Base framework
return ('current', 'pickle', 'record',
'transactions', 'transaction_oids',
'vids', 'versions', 'currentVersions')
def abortVersion(self, src, transaction):
......@@ -80,8 +87,6 @@ class Full(Base):
try:
newtid=self._serial
vid=self._vids[src]
try: dvid=self._vids[dest]
except KeyError: dvid=self._newvid(dest)
oids=[]; save_oid=oids.append
c=self._currentVersions.cursor()
......@@ -92,6 +97,12 @@ class Full(Base):
tmp=self._tmp
dump=marshal.dump
zero="\0\0\0\0\0\0\0\0"
try: dvid=self._vids[dest]
except KeyError:
dvid=self._newvid()
dump(('v',(vid, version)), tmp)
while i:
v, oid = i
......@@ -135,16 +146,20 @@ class Full(Base):
data = self._record[oid+serial][16:24]
return self._pickle[oid+data]
finally: self._lock_release()
def modifiedInVersion(self, oid):
self._lock_acquire()
try:
t=self._index[oid]
vid = self._record[oid+t][:8]
tid=self._current[oid]
vid=self._record[oid+tid][:8]
if vid == '\0\0\0\0\0\0\0\0': return ''
return self._versions[vid]
finally: self._lock_release()
def _newvid(self):
self._vid=self._vid+1
return utils.p64(self._vid)
def store(self, oid, serial, data, version, transaction):
if transaction is not self._transaction:
raise POSException.StorageTransactionError(self, transaction)
......@@ -155,6 +170,7 @@ class Full(Base):
try: vid=self._vids[version]
except:
vid=self._newvid()
dump(('v',(vid, version)), self._tmp)
else:
vid=nv='\0\0\0\0\0\0\0\0'
......@@ -181,7 +197,7 @@ class Full(Base):
def supportsVersions(self): return 1
def _finish(self, tid, u, d, e):
txn = self._env.txn_begin()
txn=self._env.txn_begin()
try:
tmp=self._tmp
ltmp=tmp.tell()
......@@ -191,9 +207,10 @@ class Full(Base):
records_put=self._records.put
pickles_put=self._pickle.put
current_put=self._current.put
transaction_oids_put=self._transaction_oids.put
currentVersions_put=self._currentVersions.put
l=pack(">HI",len(u), len(d))
self._trans.put(tid, l+u+d+e, txn)
self._transactions.put(tid, ' '+l+u+d+e, txn)
while ltmp:
try: op, arg = load(tmp)
except EOFError:
......@@ -201,6 +218,7 @@ class Full(Base):
else: raise
else:
if op=='s':
# store data
oid, vid, nv, back, data, pre = arg
key=oid+tid
if data:
......@@ -212,70 +230,185 @@ class Full(Base):
if vid != '/0/0/0/0/0/0/0/0':
versions_put(vid, oid, txn)
current_put(oid, tid, txn)
transaction_oids_put(tid, oid, txn)
elif op='d':
# discard a version (part of version commit and abort)
self._currentVersions.delete(arg, txn)
elif op='v':
# save a version definition
vid, version = arg
self._versions.put(vid, version, txn)
self._vids.put(version, vid, txn)
except:
txn.abort()
raise
else:
txn.commit()
def _undoable(self, txn):
txn.abort()
raise POSException.UndoError, 'Undoable transaction'
def undo(self, tid):
self._lock_acquire()
try:
status = self._transactions[tid][:1]
if status == 'p':
raise POSException.UndoError, 'Undoable transaction'
txn=self._env.txn_begin()
current=self._current
record=self._record
pickle=self._pickle
currentVersions=self._currentVersions
unpack=struct.unpack
try:
for oid in dups(self._transaction_oids, tid, txn):
if current.get(oid, txn) != tid: self._undoable(txn)
key=oid+tid
vid, nv, data, pre = unpack("8s8s8s8s",
record.get(key, txn))
record.delete(key, txn)
if data==tid: pickle.delete(key, txn)
if pre == '\0\0\0\0\0\0\0\0':
current.delete(oid, txn)
else:
current.put(oid, pre, txn)
try: pvid=record.get(oid+pre, txn)
except KeyError: self._undoable(txn)
if pvid != vid:
if vid != '\0\0\0\0\0\0\0\0':
del_dup(currentVersions, vid, oid, txn)
if pvid != '\0\0\0\0\0\0\0\0':
currentVersions.put(pvid, oid, txn)
self._transactions.delete(tid, txn)
self._transaction_oids.delete(tid, txn)
except:
txn.abort()
raise
else:
txn.commit()
finally: self._lock_release()
def undoLog(self, first, last, filter=None):
self._lock_acquire()
try:
c=self._transactions.cursor()
try:
i=0; r=[]; a=r.append
data=c.get(db.DB_LAST)
while data and i < last:
tid, data = data
status = data[:1]
if status == 'p': break
luser, ldesc = unpack("HI", data[1:17])
user=data[17:luser+17]
desc=data[luser+17:luser+17+ldesc]
ext=data[luser+17+ldesc:]
data={'id': tid,
'time': TimeStamp(tid).timeTime(),
'user_name': user or '',
'description': desc or '',
}
if ext:
try:
ext=loads(ext)
data.update(ext)
except: pass
if filter is None or filter(data):
if i >= first: a(data)
i=i+1
data=c.get(db.DB_PREV)
return r
finally: c.close()
finally: self._lock_release()
def versionEmpty(self, version):
self._lock_acquire()
try:
try: self._currentVersions[self._vids[version]]
except KeyError: return 1
else: return 0
finally: self._lock_release()
def tpc_vote(self, transaction):
if transaction is not self._transaction:
raise POSException.StorageTransactionError(self, transaction)
def versions(self, max=None):
self._lock_acquire()
try:
c=self._currentVersions.cursor()
try:
try: data=c.get(db.DB_NEXT_NODUP)
except: return ()
r=[]; a=r.append
while data:
a(data[0])
data=c.get(db.DB_NEXT_NODUP)
return r
finally: c.close()
finally: self._lock_release()
def history(self, oid, version=None, length=1, filter=None):
self._lock_acquire()
try:
txn = self._txn = self._env.txn_begin()
put=self._index.put
serial=self._serial
tmp=self._tmp
s=tmp.tell()
tmp.seek(0)
read=tmp.read
l=0
while l < s:
oid, ldata = unpack(">8sI", read(12))
data=read(ldata)
l=l+ldata+12
if ldata > s:
raise 'Temporary file corrupted'
put(oid, serial+data, txn)
tmp.seek(0)
if s > 999999: tmp.truncate()
tid=self._current[oid]
finally: self._lock_release()
def pack(self, t, referencesf):
self._lock_acquire()
try:
# Build an index of *only* those objects reachable
# from the root.
index=self._index
rootl=['\0\0\0\0\0\0\0\0']
pop=rootl.pop
pindex={}
referenced=pindex.has_key
while rootl:
oid=pop()
if referenced(oid): continue
# Scan non-version pickle for references
r=index[oid]
pindex[oid]=r
p=r[8:]
referencesf(p, rootl)
# Now delete any unreferenced entries:
for oid in index.keys():
if not referenced(oid): del index[oid]
finally: self._lock_release()
class dups:
"""Iterator for duplicate-record databases"
def __init__(self, db, key, txn=0):
if txn==0:
c=db.cursor()
else:
c=db.cursor(txn)
self._v=c.set(key)
self._g=c.get
self._i=0
def __getitem__(self, index):
i=self._i
if index==i: return self._v
if index < i or i < 0: raise IndexError, index
while i < index:
v=self._g(db.DB_NEXT_DUP)
if v:
i=i+1
else:
self._i=-1
raise IndexError, index
self._i=i
self._v=v
return v
def del_dup(database, key, value, txn):
c=database.cursor(txn)
try:
c.getBoth(key, value)
c.delete()
finally:
c.close()
lib/python/BDBStorage/Minimal.py
View file @ 866320f5
......@@ -5,13 +5,16 @@ from struct import pack, unpack
class Minimal(Base):
def _setupDbs(self):
self._index=self._setupDB('mini')
# Supports Base framework
self._index=self._setupDB('current')
self._setupDB('pickle')
def load(self, oid, version):
self._lock_acquire()
try:
p=self._index[oid]
return p[8:], p[:8] # pickle, serial
s=self._index[oid]
p=self._pickle[oid]
return p, s # pickle, serial
finally: self._lock_release()
def store(self, oid, serial, data, version, transaction):
......@@ -24,8 +27,7 @@ class Minimal(Base):
self._lock_acquire()
try:
if self._index.has_key(oid):
old=self._index[oid]
oserial=old[:8]
oserial=self._index[oid]
if serial != oserial: raise POSException.ConflictError
serial=self._serial
......@@ -35,14 +37,11 @@ class Minimal(Base):
return serial
def tpc_vote(self, transaction):
if transaction is not self._transaction:
raise POSException.StorageTransactionError(self, transaction)
self._lock_acquire()
def _finish(self, tid, u, d, e):
txn = self._env.txn_begin()
try:
txn = self._txn = self._env.txn_begin()
put=self._index.put
serial_put=self._index.put
pickle_put=self._pickle.put
serial=self._serial
tmp=self._tmp
s=tmp.tell()
......@@ -55,20 +54,25 @@ class Minimal(Base):
l=l+ldata+12
if ldata > s:
raise 'Temporary file corrupted'
put(oid, serial+data, txn)
serial_put(oid, serial, txn)
pickle_put(oid, data, txn)
tmp.seek(0)
if s > 999999: tmp.truncate()
finally: self._lock_release()
except:
txn.abort()
raise
else:
txn.commit()
def pack(self, t, referencesf):
self._lock_acquire()
try:
try:
# Build an index of *only* those objects reachable
# from the root.
index=self._index
index=self._pickle
rootl=['\0\0\0\0\0\0\0\0']
pop=rootl.pop
pindex={}
......@@ -78,9 +82,8 @@ class Minimal(Base):
if referenced(oid): continue
# Scan non-version pickle for references
r=index[oid]
pindex[oid]=r
p=r[8:]
p=index[oid]
pindex[oid]=1
referencesf(p, rootl)
# Now delete any unreferenced entries:
......@@ -88,3 +91,12 @@ class Minimal(Base):
if not referenced(oid): del index[oid]
finally: self._lock_release()
lib/python/BDBStorage/MinimalReplicated.py 0 → 100644
View file @ 866320f5
from base import Base
from bsddb3 import db
from struct import pack, unpack
class Minimal(Base):
def _setupDbs(self):
# Supports Base framework
self._index=self._setupDB('current')
self._setupDB('pickle')
def load(self, oid, version):
self._lock_acquire()
try:
s=self._index[oid]
p=self._pickle[oid]
return p, s # pickle, serial
finally: self._lock_release()
def store(self, oid, serial, data, version, transaction):
if transaction is not self._transaction:
raise POSException.StorageTransactionError(self, transaction)
if version:
raise POSException.Unsupported, "Versions aren't supported"
self._lock_acquire()
try:
if self._index.has_key(oid):
oserial=self._index[oid]
if serial != oserial: raise POSException.ConflictError
serial=self._serial
self._tmp.write(oid+pack(">I", len(data)))
self._tmp.write(data)
finally: self._lock_release()
return serial
def _finish(self, tid, u, d, e):
txn = self._env.txn_begin()
try:
serial_put=self._index.put
pickle_put=self._pickle.put
serial=self._serial
tmp=self._tmp
s=tmp.tell()
tmp.seek(0)
read=tmp.read
l=0
while l < s:
oid, ldata = unpack(">8sI", read(12))
data=read(ldata)
l=l+ldata+12
if ldata > s:
raise 'Temporary file corrupted'
serial_put(oid, serial, txn)
pickle_put(oid, data, txn)
tmp.seek(0)
if s > 999999: tmp.truncate()
except:
txn.abort()
raise
else:
txn.commit()
def pack(self, t, referencesf):
self._lock_acquire()
try:
# Build an index of *only* those objects reachable
# from the root.
index=self._pickle
rootl=['\0\0\0\0\0\0\0\0']
pop=rootl.pop
pindex={}
referenced=pindex.has_key
while rootl:
oid=pop()
if referenced(oid): continue
# Scan non-version pickle for references
p=index[oid]
pindex[oid]=1
referencesf(p, rootl)
# Now delete any unreferenced entries:
for oid in index.keys():
if not referenced(oid): del index[oid]
finally: self._lock_release()
lib/python/BDBStorage/Packless.py 0 → 100644
View file @ 866320f5
from base import Base
from bsddb3 import db
from struct import pack, unpack
from ZODB.referencesf import referencesf
class Packless(Base):
def _setupDbs(self):
# Supports Base framework
self._index=self._setupDB('current')
self._setupDB('referenceCount')
self._setupDB('oreferences', flags=db.DB_DUP)
self._setupDB('opickle')
def _dbnames(self):
return 'current', 'referenceCount', 'oreferences', 'opickle'
def load(self, oid, version):
self._lock_acquire()
try:
s=self._index[oid]
p=self._opickle[oid]
return p, s # pickle, serial
finally: self._lock_release()
def store(self, oid, serial, data, version, transaction):
if transaction is not self._transaction:
raise POSException.StorageTransactionError(self, transaction)
if version:
raise POSException.Unsupported, "Versions aren't supported"
self._lock_acquire()
try:
if self._index.has_key(oid):
oserial=self._index[oid]
if serial != oserial: raise POSException.ConflictError
serial=self._serial
self._tmp.write(oid+pack(">I", len(data)))
self._tmp.write(data)
finally: self._lock_release()
return serial
def _finish(self, tid, u, d, e):
txn = self._env.txn_begin()
try:
zeros={}
referenceCount=self._referenceCount
referenceCount_get=referenceCount.get
referenceCount_put=referenceCount.put
oreferences=self._oreferences
oreferences_put=oreferences.put
serial_put=self._index.put
opickle_put=self._opickle.put
serial=self._serial
tmp=self._tmp
s=tmp.tell()
tmp.seek(0)
read=tmp.read
l=0
while l < s:
oid, ldata = unpack(">8sI", read(12))
data=read(ldata)
# get references
referencesl=[]
referencesf(data, referencesl)
references={}
for roid in referencesl: references[roid]=1
referenced=references.has_key
# Create refcnt
if not referenceCount_get(oid, txn):
referenceCount_put(oid, '\0\0\0\0', txn)
zeros[oid]=1
# update stored references
c=oreferences.cursor(txn)
try:
try: roid = c.set(oid)
except:
pass
else:
while roid:
roid=roid[1]
if referenced(roid):
# still referenced, so no need to update
del references[roid]
else:
# Delete the stored ref, since we no longer
# have it
c.delete()
# decrement refcnt:
rc=unpack(">i",
referenceCount_get(roid, txn))[0]
rc=rc-1
if rc < 0:
raise "Bad reference count, %s" % (rc+1)
referenceCount_put(roid, pack(">i", rc), txn)
if rc==0: zeros[roid]=1
roid=c.get(db.DB_NEXT_DUP)
finally: c.close()
# Now add any references that weren't already stored:
for roid in references.keys():
oreferences_put(oid, roid, txn)
# Create/update refcnt
rc=referenceCount_get(roid, txn)
if rc:
rc=unpack(">i", rc)[0]
if rc=='\0\0\0\0': del zeros[roid]
referenceCount_put(roid, pack(">i", rc+1), txn)
else:
referenceCount_put(roid, '\0\0\0\1', txn)
l=l+ldata+12
if ldata > s:
raise 'Temporary file corrupted'
serial_put(oid, serial, txn)
opickle_put(oid, data, txn)
if zeros:
for oid in zeros.keys():
if oid == '\0\0\0\0\0\0\0\0': continue
self._takeOutGarbage(oid, txn)
tmp.seek(0)
if s > 999999: tmp.truncate()
except:
txn.abort()
raise
else:
txn.commit()
def _takeOutGarbage(self, oid, txn):
# take out the garbage.
referenceCount=self._referenceCount
referenceCount.delete(oid, txn)
self._opickle.delete(oid, txn)
self._current.delete(oid, txn)
# Remove/decref references
referenceCount_get=referenceCount.get
referenceCount_put=referenceCount.put
c=self._oreferences.cursor(txn)
try:
try: roid = c.set(oid)
except:
pass
else:
while roid:
c.delete()
roid=roid[1]
# decrement refcnt:
rc=referenceCount_get(roid, txn)
if rc:
rc=unpack(">i", rc)[0]-1
if rc < 0:
raise "Bad reference count, %s" % (rc+1)
if rc==0: self._takeOutGarbage(roid, txn)
else: referenceCount_put(roid, pack(">i", rc), txn)
roid=c.get(db.DB_NEXT_DUP)
finally: c.close()
if self._len > 0: self._len=self._len-1
def pack(self, t, referencesf):
self._lock_acquire()
try:
pass
# TBD
finally: self._lock_release()
lib/python/BDBStorage/base.py
View file @ 866320f5
......@@ -19,21 +19,38 @@ class Base(BaseStorage):
self._init_oid()
def _setupDB(self, name, flags=0):
"""Open an individual database and assign to an "_" attribute.
"""
d=db.Db(self._env)
if flags: db.set_flags(flags)
if flags: d.set_flags(flags)
d.open(self._prefix+name, db.DB_BTREE, db.DB_CREATE)
setattr(self, '_'+name, d)
return d
def _setupDbs(self):
"""Set up the storages databases, typically using '_setupDB'.
"""
def _init_oid(self):
c=self._index.cursor()
v=c.get(db.DB_LAST)
if v: self._oid=v[0]
else: self._oid='\0\0\0\0\0\0\0\0'
_len=-1
def __len__(self):
# TBD
return 0
l=self._len
if l < 0:
l=self._len=len(self._index)
return l
def new_oid(self, last=None):
# increment the cached length:
l=self._len
if l >= 0: self._len=l+1
return BaseStorage.new_oid(self, last)
def getSize(self):
# TBD
......@@ -49,13 +66,20 @@ class Base(BaseStorage):
self._tmp.seek(0)
def close(self):
for name in self._dbnames:
"""Close the storage
by closing the databases it uses and closing it's environment.
"""
for name in self._dbnames():
getattr(self, '_'+name).close()
delattr(self, '_'+name)
self._env.close()
del self._env
def _dbnames(self):
"""Return a list of the names of the databases used by the storage.
"""
return ("index",)
def envFromString(name):
if not os.path.exists(name): os.mkdir(name)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7