Jamesmoore/arch 259 silo cow migration improvements

cmd/connect.go

@@ -134,6 +134,10 @@ func runConnect(_ *cobra.Command, _ []string) {
 
 	dg, err = devicegroup.NewFromProtocol(protoCtx, pro, tweak, nil, nil, log, siloMetrics)
 
+	if err != nil {
+		panic(err)
+	}
+
 	for _, d := range dg.GetDeviceSchema() {
 		expName := dg.GetExposedDeviceByName(d.Name)
 		if expName != nil {

cmd/serve.go

@@ -40,6 +40,7 @@ var serveContinuous bool
 
 var serveMetrics string
 var serveDebug bool
+var serveConcurrency int
 
 func init() {
 	rootCmd.AddCommand(cmdServe)
@@ -48,6 +49,7 @@ func init() {
 	cmdServe.Flags().BoolVarP(&serveDebug, "debug", "d", false, "Debug logging (trace)")
 	cmdServe.Flags().StringVarP(&serveMetrics, "metrics", "m", "", "Prom metrics address")
 	cmdServe.Flags().BoolVarP(&serveContinuous, "continuous", "C", false, "Continuous sync")
+	cmdServe.Flags().IntVarP(&serveConcurrency, "concurrency", "x", 100, "Max total concurrency")
 }
 
 func runServe(_ *cobra.Command, _ []string) {
@@ -144,7 +146,7 @@ func runServe(_ *cobra.Command, _ []string) {
 			panic(err)
 		}
 
-		err = dg.MigrateAll(1000, func(ps map[string]*migrator.MigrationProgress) {
+		err = dg.MigrateAll(serveConcurrency, func(ps map[string]*migrator.MigrationProgress) {
 			for name, p := range ps {
 				fmt.Printf("[%s] Progress Moved: %d/%d %.2f%% Clean: %d/%d %.2f%% InProgress: %d\n",
 					name, p.MigratedBlocks, p.TotalBlocks, p.MigratedBlocksPerc,

cmd/sync.go

@@ -311,7 +311,7 @@ func syncMigrateS3(_ uint32, name string,
 			mig.SetMigratedBlock(b)
 		}
 
-		sinfo.tracker.TrackAt(0, int64(sinfo.tracker.Size()))
+		sinfo.tracker.TrackAt(int64(sinfo.tracker.Size()), 0)
 	} else {
 		fmt.Printf("Doing migration...\n")
 

pkg/storage/devicegroup/device_group_from.go

@@ -104,13 +104,6 @@ func NewFromProtocol(ctx context.Context,
 		d.EventHandler = eventHandler
 
 		destStorageFactory := func(_ *packets.DevInfo) storage.Provider {
-			/*
-				d.WaitingCacheLocal, d.WaitingCacheRemote = waitingcache.NewWaitingCacheWithLogger(d.Prov, int(di.BlockSize), dg.log)
-
-				if d.Exp != nil {
-					d.Exp.SetProvider(d.WaitingCacheLocal)
-				}
-			*/
 			return d.WaitingCacheRemote
 		}
 

pkg/storage/devicegroup/device_group_test.go

@@ -3,9 +3,9 @@ package devicegroup
 import (
 	"context"
 	"crypto/rand"
+	"fmt"
 	"io"
 	"os"
-	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -29,7 +29,7 @@ var testDeviceSchema = []*config.DeviceSchema{
 		System:    "file",
 		BlockSize: "1m",
 		//	Expose:    true,
-		Location: "testdev_test1",
+		Location: "test_data/test1",
 	},
 
 	{
@@ -38,11 +38,21 @@ var testDeviceSchema = []*config.DeviceSchema{
 		System:    "file",
 		BlockSize: "1m",
 		//		Expose:    true,
-		Location: "testdev_test2",
+		Location: "test_data/test2",
+		ROSource: &config.DeviceSchema{
+			Name:      "test_data/test2state",
+			Size:      "16m",
+			System:    "file",
+			BlockSize: "1m",
+			Location:  "test_data/test2base",
+		},
+		ROSourceShared: true,
 	},
 }
 
 func setupDeviceGroup(t *testing.T) *DeviceGroup {
+	err := os.Mkdir("test_data", 0777)
+	assert.NoError(t, err)
 	/*
 		currentUser, err := user.Current()
 		if err != nil {
@@ -60,8 +70,7 @@ func setupDeviceGroup(t *testing.T) *DeviceGroup {
 		err = dg.CloseAll()
 		assert.NoError(t, err)
 
-		os.Remove("testdev_test1")
-		os.Remove("testdev_test2")
+		os.RemoveAll("test_data")
 	})
 
 	return dg
@@ -205,12 +214,6 @@ func TestDeviceGroupMigrate(t *testing.T) {
 		return
 	}
 
-	// Remove the receiving files
-	t.Cleanup(func() {
-		os.Remove("testrecv_test1")
-		os.Remove("testrecv_test2")
-	})
-
 	log := logging.New(logging.Zerolog, "silo", os.Stdout)
 	log.SetLevel(types.TraceLevel)
 
@@ -236,22 +239,36 @@ func TestDeviceGroupMigrate(t *testing.T) {
 	}()
 
 	// Lets write some data...
+	buff := make([]byte, 4096)
 	for _, s := range testDeviceSchema {
 		prov := dg.GetProviderByName(s.Name)
-		buff := make([]byte, prov.Size())
-		_, err := rand.Read(buff)
-		assert.NoError(t, err)
-		_, err = prov.WriteAt(buff, 0)
-		assert.NoError(t, err)
+		if s.Name == "test1" {
+			allBuff := make([]byte, prov.Size())
+			_, err := rand.Read(allBuff)
+			assert.NoError(t, err)
+			_, err = prov.WriteAt(allBuff, 0)
+			assert.NoError(t, err)
+		} else {
+			for _, offset := range []int64{10000, 300000, 700000} {
+				_, err := rand.Read(buff)
+				assert.NoError(t, err)
+				_, err = prov.WriteAt(buff, offset)
+				assert.NoError(t, err)
+			}
+		}
 	}
 
 	var dg2 *DeviceGroup
 	var wg sync.WaitGroup
 
 	// We will tweak schema in recv here so we have separate paths.
 	tweak := func(_ int, _ string, schema *config.DeviceSchema) *config.DeviceSchema {
-		schema.Location = strings.ReplaceAll(schema.Location, "testdev_test1", "testrecv_test1")
-		schema.Location = strings.ReplaceAll(schema.Location, "testdev_test2", "testrecv_test2")
+		schema.Location = fmt.Sprintf(schema.Location, ".recv")
+		// Tweak overlay if there is one as well...
+		if schema.ROSource != nil {
+			schema.ROSource.Location = fmt.Sprintf(schema.ROSource.Location, ".recv")
+			schema.ROSource.Name = fmt.Sprintf(schema.ROSource.Name, ".recv")
+		}
 		return schema
 	}
 
@@ -319,4 +336,10 @@ func TestDeviceGroupMigrate(t *testing.T) {
 	w1.Close()
 	r2.Close()
 	w2.Close()
+
+	// Show some metrics...
+	pMetrics := prSource.GetMetrics()
+	fmt.Printf("Protocol SENT (packets %d data %d urgentPackets %d urgentData %d) RECV (packets %d data %d)\n",
+		pMetrics.PacketsSent, pMetrics.DataSent, pMetrics.UrgentPacketsSent, pMetrics.UrgentDataSent,
+		pMetrics.PacketsRecv, pMetrics.DataRecv)
 }

pkg/storage/devicegroup/device_group_to.go

@@ -148,6 +148,22 @@ func (dg *DeviceGroup) MigrateAll(maxConcurrency int, progressHandler func(p map
 		}
 	}
 
+	// Check if the devices are actually all here?
+	for _, d := range dg.devices {
+		if d.WaitingCacheLocal != nil {
+			wcMetrics := d.WaitingCacheLocal.GetMetrics()
+			if wcMetrics.AvailableRemote < uint64(d.NumBlocks) {
+				if dg.log != nil {
+					dg.log.Warn().
+						Str("name", d.Schema.Name).
+						Uint64("availableRemoteBlocks", wcMetrics.AvailableRemote).
+						Uint64("numBlocks", uint64(d.NumBlocks)).
+						Msg("migrating away a possibly incomplete source")
+				}
+			}
+		}
+	}
+
 	ctime := time.Now()
 
 	if dg.log != nil {
@@ -280,8 +296,22 @@ func (dg *DeviceGroup) MigrateAll(maxConcurrency int, progressHandler func(p map
 	// Now start them all migrating, and collect err
 	for _, d := range dg.devices {
 		go func() {
-			err := d.Migrator.Migrate(d.NumBlocks)
-			errs <- err
+			migrateBlocks := d.NumBlocks
+			unrequired := d.DirtyRemote.GetUnrequiredBlocks()
+			alreadyBlocks := make([]uint32, 0)
+			for _, b := range unrequired {
+				d.Orderer.Remove(int(b))
+				migrateBlocks--
+				alreadyBlocks = append(alreadyBlocks, uint32(b))
+			}
+
+			err := d.To.SendYouAlreadyHave(d.BlockSize, alreadyBlocks)
+			if err != nil {
+				errs <- err
+			} else {
+				err = d.Migrator.Migrate(migrateBlocks)
+				errs <- err
+			}
 		}()
 	}
 
@@ -363,6 +393,7 @@ func (dg *DeviceGroup) MigrateDirty(hooks *MigrateDirtyHooks) error {
 					cont, err := hooks.PostGetDirty(d.Schema.Name, blocks)
 					if err != nil {
 						errs <- err
+						return
 					}
 					if !cont {
 						break

pkg/storage/dirtytracker/dirty_tracker.go

@@ -146,14 +146,48 @@ func (dt *DirtyTracker) trackArea(length int64, offset int64) {
 	dt.tracking.SetBits(bStart, bEnd)
 }
 
+/**
+ * Ask downstream if there are blocks that aren't required for a migration.
+ * Typically these would be a base overlay, but could be something else.
+ * As well as returning the blocks, this call will update the dirty tracker as if the blocks had been read (To start tracking dirty changes)
+ */
+func (dtr *Remote) GetUnrequiredBlocks() []uint {
+	// Make sure no writes get through...
+	dtr.dt.writeLock.Lock()
+	defer dtr.dt.writeLock.Unlock()
+
+	// Snapshot blocks from any CoW, and update the tracking
+	cowBlocks := storage.SendSiloEvent(dtr.dt.prov, storage.EventTypeCowGetBlocks, dtr)
+	if len(cowBlocks) == 1 {
+		blocks := cowBlocks[0].([]uint)
+		for _, b := range blocks {
+			offset := b * uint(dtr.dt.blockSize)
+			length := uint(dtr.dt.blockSize)
+
+			// NB overflow here shouldn't matter. TrackArea will cope and truncate it.
+			dtr.dt.trackArea(int64(length), int64(offset))
+		}
+		return blocks
+	}
+	return nil
+}
+
 /**
  * Start tracking at the given offset and length
  *
  */
-func (dtr *Remote) TrackAt(offset int64, length int64) {
+func (dtr *Remote) TrackAt(length int64, offset int64) {
 	dtr.dt.trackArea(length, offset)
 }
 
+/**
+ * Check which blocks are being tracked
+ *
+ */
+func (dtr *Remote) GetTrackedBlocks() []uint {
+	return dtr.dt.tracking.Collect(0, dtr.dt.tracking.Length())
+}
+
 /**
  * Get a quick measure of how many blocks are currently dirty
  *

pkg/storage/metrics/prometheus/prometheus.go

@@ -128,6 +128,8 @@ type Metrics struct {
 	toProtocolSentWriteAtBytes         *prometheus.GaugeVec
 	toProtocolSentWriteAtWithMap       *prometheus.GaugeVec
 	toProtocolSentRemoveFromMap        *prometheus.GaugeVec
+	toProtocolSentYouAlreadyHave       *prometheus.GaugeVec
+	toProtocolSentYouAlreadyHaveBytes  *prometheus.GaugeVec
 	toProtocolRecvNeedAt               *prometheus.GaugeVec
 	toProtocolRecvDontNeedAt           *prometheus.GaugeVec
 
@@ -281,6 +283,10 @@ func New(reg prometheus.Registerer, config *MetricsConfig) *Metrics {
 			Namespace: config.Namespace, Subsystem: config.SubToProtocol, Name: "sent_write_at_with_map", Help: "sentWriteAtWithMap"}, []string{"device"}),
 		toProtocolSentRemoveFromMap: prometheus.NewGaugeVec(prometheus.GaugeOpts{
 			Namespace: config.Namespace, Subsystem: config.SubToProtocol, Name: "sent_remove_from_map", Help: "sentRemoveFromMap"}, []string{"device"}),
+		toProtocolSentYouAlreadyHave: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Namespace: config.Namespace, Subsystem: config.SubToProtocol, Name: "sent_you_already_have", Help: "sentYouAlreadyHave"}, []string{"device"}),
+		toProtocolSentYouAlreadyHaveBytes: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Namespace: config.Namespace, Subsystem: config.SubToProtocol, Name: "sent_you_already_have_bytes", Help: "sentYouAlreadyHaveBytes"}, []string{"device"}),
 		toProtocolRecvNeedAt: prometheus.NewGaugeVec(prometheus.GaugeOpts{
 			Namespace: config.Namespace, Subsystem: config.SubToProtocol, Name: "recv_need_at", Help: "recvNeedAt"}, []string{"device"}),
 		toProtocolRecvDontNeedAt: prometheus.NewGaugeVec(prometheus.GaugeOpts{
@@ -434,7 +440,8 @@ func New(reg prometheus.Registerer, config *MetricsConfig) *Metrics {
 		met.toProtocolSentDirtyList, met.toProtocolSentReadAt, met.toProtocolSentWriteAtHash, met.toProtocolSentWriteAtHashBytes,
 		met.toProtocolSentWriteAtComp, met.toProtocolSentWriteAtCompBytes, met.toProtocolSentWriteAtCompDataBytes,
 		met.toProtocolSentWriteAt, met.toProtocolSentWriteAtBytes, met.toProtocolSentWriteAtWithMap,
-		met.toProtocolSentRemoveFromMap, met.toProtocolRecvNeedAt, met.toProtocolRecvDontNeedAt,
+		met.toProtocolSentRemoveFromMap, met.toProtocolSentYouAlreadyHave, met.toProtocolSentYouAlreadyHaveBytes,
+		met.toProtocolRecvNeedAt, met.toProtocolRecvDontNeedAt,
 	)
 
 	reg.MustRegister(met.fromProtocolRecvEvents, met.fromProtocolRecvHashes, met.fromProtocolRecvDevInfo,
@@ -591,6 +598,8 @@ func (m *Metrics) AddToProtocol(name string, proto *protocol.ToProtocol) {
 		m.toProtocolSentWriteAtBytes.WithLabelValues(name).Set(float64(met.SentWriteAtBytes))
 		m.toProtocolSentWriteAtWithMap.WithLabelValues(name).Set(float64(met.SentWriteAtWithMap))
 		m.toProtocolSentRemoveFromMap.WithLabelValues(name).Set(float64(met.SentRemoveFromMap))
+		m.toProtocolSentYouAlreadyHave.WithLabelValues(name).Set(float64(met.SentYouAlreadyHave))
+		m.toProtocolSentYouAlreadyHaveBytes.WithLabelValues(name).Set(float64(met.SentYouAlreadyHaveBytes))
 		m.toProtocolRecvNeedAt.WithLabelValues(name).Set(float64(met.RecvNeedAt))
 		m.toProtocolRecvDontNeedAt.WithLabelValues(name).Set(float64(met.RecvDontNeedAt))
 	})