.

wcfs/wcfs.go

@@ -608,7 +608,8 @@ type WatchLink struct {
 	//
 	// both already established, and watches being initialized in-progress are registered here.
 	// (see setupWatch)
-	byfileMu sync.Mutex
+	// XXX byfile -> wlink-global watchMu ?
+	byfileMu sync.Mutex          // zheadMu.W  |  zheadMu.R + byfileMu	(XXX recheck)
 	byfile   map[zodb.Oid]*Watch // {} foid -> Watch
 
 	// IO
@@ -623,12 +624,9 @@ type Watch struct {
 	link *WatchLink // link to client
 	file *BigFile	// XXX needed?
 
-	// XXX locking
-
-	at     zodb.Tid	// requested to be watched @at
-
-	// {} blk -> rev
-	pinned map[int64]zodb.Tid // blocks that are already pinned to be ≤ at
+	mu     sync.Mutex	// XXX ok ?
+	at     zodb.Tid           // requested to be watched @at
+	pinned map[int64]zodb.Tid // {} blk -> rev    blocks that are already pinned to be ≤ at
 }
 
 // -------- 3) Cache invariant --------
@@ -1206,6 +1204,7 @@ func (f *BigFile) readBlk(ctx context.Context, blk int64, dest []byte) (err erro
 // XXX do we really need to use/propagate caller context here? ideally update
 // watchers should be synchronous, and in practice we just use 30s timeout.
 // Should a READ interrupt cause watch update failure?
+// XXX -> pinWatchers? pinOnRead?
 func (f *BigFile) updateWatchers(ctx context.Context, blk int64, treepath []btree.LONode, zblk zBlk, blkrevMax zodb.Tid) {
 	// only head/ is being watched for
 	if f.head.rev != 0 {
@@ -1502,11 +1501,14 @@ func (wlink *WatchLink) setupWatch(ctx context.Context, foid zodb.Oid, at zodb.T
 	//               ↑            ↑
 	//              w.at         head
 	//
-	// XXX locking
+	// - also: there won't be simultaneous READs that would need to be
+	//   unpinned, because we update w.at to requested at early.
+	//
 	// XXX register only if watch was created anew, not updated?
-	w.at = at
-	f.watchTab[w] = struct{}{}
-	wlink.byfile[foid] = w
+	w.at = at			// XXX locking
+	// NOTE registering f.watchTab[w] and wlink.byfile[foid] = w must come together.
+	f.watchTab[w] = struct{}{}	// XXX locking
+	wlink.byfile[foid] = w		// XXX locking
 
 	// XXX defer -> unregister watch if error?
 
@@ -1553,9 +1555,6 @@ func (wlink *WatchLink) setupWatch(ctx context.Context, foid zodb.Oid, at zodb.T
 		return err
 	}
 
-	// XXX or register w to f & wlink here?
-	// NOTE registering f.watchTab[w] and wlink.byfile[foid] = w must come together.
-
 	return nil
 }
 
@@ -1618,9 +1617,8 @@ func (wlink *WatchLink) _serve() (err error) {
 		}
 
 		// unregister all watches created on this wlink
-		// XXX locking
-		for _, w := range wlink.byfile {
-			delete(w.file.watchTab, w)
+		for _, w := range wlink.byfile {	// XXX locking
+			delete(w.file.watchTab, w)	// XXX locking
 		}
 		wlink.byfile = nil