.

wcfs/zodb.go

@@ -22,12 +22,14 @@ import (
 	pickle "github.com/kisielk/og-rek"
 )
 
+// Object is common base for in-process representation of ZODB object.
 type Object struct {
 	jar	*Connection
 	oid	zodb.Oid
 	serial	zodb.Tid
 }
 
+// PyObject is common base for in-process representation of ZODB Python objects.
 type PyObject struct {
 	Object
 
@@ -39,6 +41,8 @@ type PyObject struct {
 //
 // XXX Connection, and {Py}Object methods that relate to it, are not safe for
 // modifications from multiple goroutines simultaneously.
+//
+// XXX ^^^ better must be safe - use case: e.g. prefetch.
 type Connection struct {
 	stor	zodb.IStorage	// underlying storage
 	at	zodb.Tid	// current view of database
@@ -52,7 +56,7 @@ type Connection struct {
 	// -> we can use that {} when loading a persistent Ref twice to get to the same object.
 	//
 	// however: if Connection keeps strong link to pyobj, just
-	// pyobj.PDeactivate will not fully release pyobj if there are not
+	// pyobj.PDeactivate will not fully release pyobj if there are no
 	// references to it from other objects:
 	//
 	//	- deactivate will release pyobj state (ok)
@@ -90,7 +94,18 @@ type Connection struct {
 	//
 	// NOTE2 finalizers don't run on when they are attached to an object in cycle.
 	// Hopefully we don't have cycles with ZBtree/ZBucket	XXX verify this
-	objtab map[zodb.Oid]WeakRef
+	objmu  sync.Mutex
+	objtab map[zodb.Oid]interface{}	// oid -> WeakRef(PyObject) | loadInProgress
+}
+
+// loadInProgress entry in Conn.objtab tells users that try to get the entry
+// that another goroutine is already in progress of loading it.
+type loadInProgress struct {
+	ready chan struct{} // closed when loading finishes
+
+	// result of the load
+	pyobj *PyObject
+	err   error
 }
 
 // Gets loads and decodes object from the database according to its current view.
@@ -99,6 +114,60 @@ type Connection struct {
 // XXX   this is needed if there are several persistent references to the same object.
 //       however wendelin.core does not do this.
 func (conn *Connection) Get(ctx context.Context, oid zodb.Oid) (*PyObject, error) {
+	conn.objmu.Lock()		// XXX -> rlock
+	objentry := conn.objtab[oid]
+
+	// someone else is already loading the object.
+	// we have to wait for that someone to finish the load.
+	if load, ok := objentry.(*loadInProgress); ok {
+		conn.objmu.Unlock()
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()	// XXX err ctx
+		case <-xobj.ready:
+			return xobj.pyobj, err
+		}
+	}
+
+	// noone else is loading the object. Investigate whether it is simply not
+	// in objtab, or it is lost weakref, and if so we have to initiate the load.
+	var xobj interface{}
+	if objentry != nil {
+		wobj := objentry.(*WeakRef)
+		xobj = wobj.Get()
+	}
+	if xpyobj != nil {
+		// the object is already there
+		conn.objmu.Unlock()
+		return xpyobj, nil	// XXX + activate?
+	}
+
+	// the object is not there, we have to (re)load it.
+	load := &loadInProgress{ready: make(chan struct)}
+	conn.objTab[oid] = load
+	conn.objmu.Unlock()
+
+	pyobj, err = conn.loadpy(ctx, oid)
+	load.pyobj = pyobj
+	load.err   = err
+	close(load.ready)
+
+	// replace objtab entry with WeakRef
+	conn.objmu.Lock()
+	defer conn.objmu.Unlock()
+	if x := conn.objtab[oid]; x != load {
+		panic(fmt.Sprintf("zodb.Conn: objtab[%s]: load finished, but it is: %#v", oid, x)
+	}
+	if err != nil {
+		// it was loading error - remove the entry not to cause uncontrolled objtab growth.
+		// if this object is accessed next time - the load will be retried.
+		delete(conn.objtab[oid])
+	} else {
+		conn.objtab[oid] = NewWeakRef(pyobj)
+	}
+
+
+/*
 	// XXX -> loadpy
 	buf, serial, err := conn.stor.Load(ctx, zodb.Xid{Oid: oid, At: conn.at})
 	if err != nil {
@@ -117,6 +186,7 @@ func (conn *Connection) Get(ctx context.Context, oid zodb.Oid) (*PyObject, error
 		pyclass: pyclass,
 		pystate: pystate,
 	}, nil
+*/
 }
 
 func (conn *Connection) loadpy(ctx context.Context, oid zodb.Oid) (pyclass pickle.Class, pystate interface{}, serial zodb.Tid, _ error) {