Initial impl moved dirtyTracker and volatilityMonitor up

pkg/peer/make_migratable.go

@@ -10,9 +10,7 @@ import (
 	"github.com/loopholelabs/drafter/pkg/runner"
 	"github.com/loopholelabs/goroutine-manager/pkg/manager"
 	"github.com/loopholelabs/silo/pkg/storage/blocks"
-	"github.com/loopholelabs/silo/pkg/storage/dirtytracker"
 	"github.com/loopholelabs/silo/pkg/storage/modules"
-	"github.com/loopholelabs/silo/pkg/storage/volatilitymonitor"
 )
 
 type ResumedPeer[L ipc.AgentServerLocal, R ipc.AgentServerRemote[G], G any] struct {
@@ -79,10 +77,8 @@ func (resumedPeer *ResumedPeer[L, R, G]) MakeMigratable(
 		stage3Inputs,
 		func(index int, input makeMigratableFilterStage, output *makeMigratableDeviceStage, addDefer func(deferFunc func() error)) error {
 			output.prev = input
-
-			dirtyLocal, dirtyRemote := dirtytracker.NewDirtyTracker(input.prev.storage, int(input.prev.blockSize))
-			output.dirtyRemote = dirtyRemote
-			monitor := volatilitymonitor.NewVolatilityMonitor(dirtyLocal, int(input.prev.blockSize), input.makeMigratableDevice.Expiry)
+			output.dirtyRemote = input.prev.dirtyRemote
+			monitor := input.prev.volatilityMonitor
 
 			local := modules.NewLockable(monitor)
 			output.storage = local

pkg/peer/migrate_from.go

@@ -12,6 +12,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"syscall"
+	"time"
 
 	"github.com/loopholelabs/drafter/internal/utils"
 	"github.com/loopholelabs/drafter/pkg/ipc"
@@ -23,12 +24,16 @@ import (
 	"github.com/loopholelabs/silo/pkg/storage"
 	"github.com/loopholelabs/silo/pkg/storage/config"
 	"github.com/loopholelabs/silo/pkg/storage/device"
+	"github.com/loopholelabs/silo/pkg/storage/dirtytracker"
 	"github.com/loopholelabs/silo/pkg/storage/protocol"
 	"github.com/loopholelabs/silo/pkg/storage/protocol/packets"
+	"github.com/loopholelabs/silo/pkg/storage/volatilitymonitor"
 	"github.com/loopholelabs/silo/pkg/storage/waitingcache"
 	"golang.org/x/sys/unix"
 )
 
+const volatilityExpiry = time.Minute * 30
+
 type MigrateFromDevice[L ipc.AgentServerLocal, R ipc.AgentServerRemote[G], G any] struct {
 	Name string `json:"name"`
 
@@ -196,7 +201,11 @@ func (peer *Peer[L, R, G]) MigrateFrom(
 							}
 						}
 
-						dev.SetProvider(local)
+						dirtyLocal, dirtyRemote := dirtytracker.NewDirtyTracker(local, int(di.BlockSize))
+						vmonitor := volatilitymonitor.NewVolatilityMonitor(dirtyLocal, int(di.BlockSize), volatilityExpiry)
+						vmonitor.AddAll()
+
+						dev.SetProvider(vmonitor)
 
 						stage2InputsLock.Lock()
 						migratedPeer.stage2Inputs = append(migratedPeer.stage2Inputs, migrateFromStage{
@@ -207,8 +216,11 @@ func (peer *Peer[L, R, G]) MigrateFrom(
 							id:     index,
 							remote: true,
 
-							storage: local,
-							device:  dev,
+							storage:           local,
+							device:            dev,
+							dirtyLocal:        dirtyLocal,
+							dirtyRemote:       dirtyRemote,
+							volatilityMonitor: vmonitor,
 						})
 						stage2InputsLock.Unlock()
 
@@ -494,7 +506,11 @@ func (peer *Peer[L, R, G]) MigrateFrom(
 				addDefer(local.Close)
 				addDefer(dev.Shutdown)
 
-				dev.SetProvider(local)
+				dirtyLocal, dirtyRemote := dirtytracker.NewDirtyTracker(local, int(input.BlockSize))
+				vmonitor := volatilitymonitor.NewVolatilityMonitor(dirtyLocal, int(input.BlockSize), volatilityExpiry)
+				vmonitor.AddAll()
+
+				dev.SetProvider(vmonitor)
 
 				stage2InputsLock.Lock()
 				migratedPeer.stage2Inputs = append(migratedPeer.stage2Inputs, migrateFromStage{
@@ -505,8 +521,11 @@ func (peer *Peer[L, R, G]) MigrateFrom(
 					id:     uint32(index),
 					remote: false,
 
-					storage: local,
-					device:  dev,
+					storage:           local,
+					device:            dev,
+					dirtyLocal:        dirtyLocal,
+					dirtyRemote:       dirtyRemote,
+					volatilityMonitor: vmonitor,
 				})
 				stage2InputsLock.Unlock()
 

pkg/peer/stages.go

@@ -6,6 +6,7 @@ import (
 	"github.com/loopholelabs/silo/pkg/storage/blocks"
 	"github.com/loopholelabs/silo/pkg/storage/dirtytracker"
 	"github.com/loopholelabs/silo/pkg/storage/modules"
+	"github.com/loopholelabs/silo/pkg/storage/volatilitymonitor"
 )
 
 type migrateFromStage struct {
@@ -16,8 +17,11 @@ type migrateFromStage struct {
 	id     uint32
 	remote bool
 
-	storage storage.Provider
-	device  storage.ExposedStorage
+	storage           storage.Provider
+	device            storage.ExposedStorage
+	dirtyLocal        *dirtytracker.Local
+	dirtyRemote       *dirtytracker.Remote
+	volatilityMonitor *volatilitymonitor.VolatilityMonitor
 }
 
 type makeMigratableFilterStage struct {