.

wcfs/wcfs.go

@@ -420,6 +420,9 @@ package main
 // head.zheadMu		WLock by handleδZ
 //			RLock by read
 // ...
+//
+// WatchLink.byfileMu	> Watch.mu
+// BigFile.watchMu      > Watch.mu
 
 import (
 	"bufio"
@@ -570,6 +573,7 @@ type BigFile struct {
 	// progress of being established. XXX text
 	//
 	// XXX locking -> watchMu?
+	// XXX -> watches ?
 	watchTab map[*Watch]struct{}
 }
 
@@ -598,14 +602,12 @@ type WatchLink struct {
 	id   int32     // ID of this /head/watch handle (for debug log)
 	head *Head
 
-	// watches established over this watch link.
-	// XXX in-progress - where?	-> (XXX no - see vvv) nowhere; here only established watches are added
-	// XXX -> in-progress here - so that access to new blocks after δFtail
-	//        was queried also send pins.
+	// watches associated with this watch link.
 	//
-	// XXX locking?
-	// XXX -> watchByFile?
-	byfile map[zodb.Oid]*Watch // {} foid -> Watch
+	// both already established, and watches being initialized in-progress are registered here.
+	// (see setupWatch)
+	byfileMu sync.Mutex
+	byfile   map[zodb.Oid]*Watch // {} foid -> Watch
 
 	// IO
 	reqNext uint64 // stream ID for next wcfs-originated request; 0 is reserved for control messages
@@ -1395,14 +1397,31 @@ func (wlink *WatchLink) setupWatch(ctx context.Context, foid zodb.Oid, at zodb.T
 	head := wlink.head
 	bfdir := head.bfdir
 
-	// XXX locking
-	// XXX head.zheadMu.RLock() + defer unlock (see vvv for unpin vs pin and locked head)
+	// wait for zhead.At ≥ at
+	if at != zodb.InvalidTid {
+		err = head.zheadWait(ctx, at)
+		if err != nil {
+			return err
+		}
+	}
+
+	// make sure zhead.At stays unchanged while we are preparing the watch
+	// (see vvv e.g. about unpin to @head for why it is neeed)
+	head.zheadMu.RLock()
+	defer head.zheadMu.RUnlock()
+	headAt := head.zconn.At()
+
+	if at < bfdir.δFtail.Tail() {
+		return fmt.Errorf("too far away back from head/at (@%s); δt = %s",
+			headAt, headAt.Time().Sub(at.Time().Time))
+	}
 
 	// XXX if watch was already established - we need to update it
+	// XXX locking (.byfile)
 	w := wlink.byfile[foid]
 	if w == nil {
 		// watch was not previously established - set it up anew
-		// XXX locking
+		// XXX locking (.fileTab)
 		f := bfdir.fileTab[foid]
 		if f == nil {
 			// by "invalidation protocol" watch is setup after data file was opened
@@ -1432,23 +1451,9 @@ func (wlink *WatchLink) setupWatch(ctx context.Context, foid zodb.Oid, at zodb.T
 		return fmt.Errorf("going back in history is forbidden")
 	}
 
-	// XXX locking
-	err = head.zheadWait(ctx, at)
-	if err != nil {
-		return err
-	}
-
-	// XXX locking
-	headAt := head.zconn.At()
-
-	if at < bfdir.δFtail.Tail() {
-		return fmt.Errorf("too far away back from head/at (@%s); δt = %s",
-			headAt, headAt.Time().Sub(at.Time().Time))
-	}
-
 	f := w.file
 
-	// register w to f here early, so that READs going in parallel to us
+	// register w to f early, so that READs going in parallel to us
 	// preparing and processing initial pins, also send pins to w for read
 	// blocks. If we don't, we can miss to send pin to w for a freshly read
 	// block which could have revision > w.at:	XXX test
@@ -1467,7 +1472,7 @@ func (wlink *WatchLink) setupWatch(ctx context.Context, foid zodb.Oid, at zodb.T
 	// pin for #3.
 	//
 	// NOTE for `unpin blk` to -> @head we can be sure there won't be
-	// simultaneous `pin blk` request, because:		XXX recheck
+	// simultaneous `pin blk` request, because:
 	//
 	// - unpin means blk was previously pinned,
 	// - blk was pinned means it is tracked by δFtail,
@@ -1475,7 +1480,7 @@ func (wlink *WatchLink) setupWatch(ctx context.Context, foid zodb.Oid, at zodb.T
 	//   there is indeed no blk change in that region,
 	// - which means that δblk with rev > w.at might be only > head,
 	// - but such δblk are processed with zhead wlocked and we keep zhead
-	//   rlocked during pin setup.		XXX rlock zhead during setupWatch
+	//   rlocked during pin setup.
 	//
 	//          δ                      δ
 	//      ----x----.------------]----x----
@@ -1573,8 +1578,6 @@ func (wlink *WatchLink) serve() {
 		log.Error(err)
 	}
 
-	// XXX deactiwate all watches from wlink.byfile[*]
-
 	head := wlink.head
 	head.wlinkMu.Lock()
 	delete(head.wlinkTab, wlink)