.

go/zodb/connection.go

@@ -20,6 +20,7 @@ import (
 	"sync"
 
 	"lab.nexedi.com/kirr/go123/mem"
+	"lab.nexedi.com/kirr/neo/go/transaction"
 	"lab.nexedi.com/kirr/neo/go/zodb/internal/weak"
 )
 
@@ -40,6 +41,8 @@ import (
 // inside transaction where Connection was opened.
 type Connection struct {
 	stor	IStorage	// underlying storage
+	db      *DB		// Connection is part of this DB
+	txn	transaction.Transaction	// opened under this txn; nil if idle in DB pool.
 
 	// current view of database; stable inside a transaction.
 	at	Tid
@@ -123,19 +126,13 @@ func (e *wrongClassError) Error() string {
 }
 
 
-// get returns in-RAM object corresponding to specified ZODB object according to current database view.
-//
-// If there is already in-RAM object that corresponds to oid, that in-RAM object is returned.
-// Otherwise new in-RAM object is created according to specified class.
-//
-// The object's data is not necessarily loaded after get returns. Use
-// PActivate to make sure the object is fully loaded.
-//
-// XXX object scope.
+// get is like Get, but used when we already know object class.
 //
 // Use-case: in ZODB references are (pyclass, oid), so new ghost is created
 // without further loading anything.
 func (conn *Connection) get(class string, oid Oid) (IPersistent, error) {
+	// XXX txn check
+
 	conn.objmu.Lock()		// XXX -> rlock
 	wobj := conn.objtab[oid]
 	var obj IPersistent
@@ -178,6 +175,8 @@ func (conn *Connection) get(class string, oid Oid) (IPersistent, error) {
 // The object's data is not necessarily loaded after Get returns. Use
 // PActivate to make sure the object is fully loaded.
 func (conn *Connection) Get(ctx context.Context, oid Oid) (IPersistent, error) {
+	// XXX txn check
+
 	conn.objmu.Lock()		// XXX -> rlock
 	wobj := conn.objtab[oid]
 	var xobj interface{}
@@ -217,5 +216,7 @@ func (conn *Connection) Get(ctx context.Context, oid Oid) (IPersistent, error) {
 //
 // XXX must be called ... (XXX e.g. outside transaction boundary) so that there is no race on .at .
 func (conn *Connection) load(ctx context.Context, oid Oid) (_ *mem.Buf, serial Tid, _ error) {
+	// XXX txn check
+
 	return conn.stor.Load(ctx, Xid{Oid: oid, At: conn.at})
 }

go/zodb/db.go

@@ -14,6 +14,13 @@
 package zodb
 // application-level database connection.
 
+import (
+	"context"
+	"sync"
+
+	"lab.nexedi.com/kirr/neo/go/transaction"
+)
+
 // DB represents a handle to database at application level and contains pool
 // of connections. If application opens connection via DB.Open, the connection
 // will be automatically put back into DB pool for future reuse after
@@ -28,4 +35,87 @@ package zodb
 //
 // DB is safe to access from multiple goroutines simultaneously.
 type DB struct {
+	stor	IStorage
+
+	mu	sync.Mutex
+	connv	[]*Connection
+}
+
+
+// NewDB creates new database handle.
+//
+// XXX text, options.
+func NewDB(stor IStorage) *DB {
+	return &DB{stor: stor}
+}
+
+// Open opens new connection to the database.
+//
+// XXX must be called under transaction.
+//
+// XXX text
+func (db *DB) Open(ctx context.Context) *Connection {
+	txn := transaction.Current(ctx)
+
+	conn := db.get()
+
+	// XXX sync storage for lastTid -> conn.at
+
+	conn.txn = txn
+	txn.RegisterSync((*connTxnSync)(conn))
+
+	return conn
+}
+
+
+// get returns connection from db pool, or creates new one if pool was empty.
+func (db *DB) get() *Connection {
+	db.mu.Lock()
+	defer db.mu.Unlock()
+
+	var conn *Connection
+	if l := len(db.connv); l > 0 {
+		// pool is !empty - use latest closed conn.
+		// XXX zodb/py orders conn ↑ by conn._cache.cache_non_ghost_count.
+		// XXX this way top of the stack is the "most live".
+		conn = db.connv[l - 1]
+		db.connv[l - 1] = nil
+		db.connv = db.connv[:l-1]
+
+		// XXX assert conn.txn == nil
+	} else {
+		conn = &Connection{stor: db.stor, db: db}
+	}
+
+	// XXX assert conn.db == db
+	return conn
+}
+
+// put puts connection back into db pool.
+func (db *DB) put(conn *Connection) {
+	// XXX assert conn.db == db
+	conn.txn = nil
+	//XXX want to do this; but also need to preserve info at..last_tid to process invalidations
+	//conn.at  = 0
+
+	db.mu.Lock()
+	defer db.mu.Unlock()
+
+	// XXX check if len(connv) > X, and drop conn if yes
+	db.connv = append(db.connv, conn)
+}
+
+// ---- txn sync ----
+
+type connTxnSync Connection // hide from public API
+
+func (conn *connTxnSync) BeforeCompletion(txn transaction.Transaction) {
+	// XXX check txn
+	// nothing
+}
+
+func (conn *connTxnSync) AfterCompletion(txn transaction.Transaction) {
+	// XXX check txn
+
+	conn.db.put((*Connection)(conn))
 }