[Disk Manager]: Snapshot: Check chunk contents for zeroes

cloud/disk_manager/internal/pkg/dataplane/common/chunk.go

@@ -1,11 +1,14 @@
 package common
 
 import (
+	"bytes"
 	"hash/crc32"
 )
 
 ////////////////////////////////////////////////////////////////////////////////
 
+var zeroes = make([]byte, 1024*1024)
+
 type Chunk struct {
 	ID          string
 	Index       uint32
@@ -18,3 +21,20 @@ type Chunk struct {
 func (chunk Chunk) Checksum() uint32 {
 	return crc32.ChecksumIEEE(chunk.Data)
 }
+
+func (chunk Chunk) CheckDataIsAllZeroes() bool {
+	for i := 0; i < len(chunk.Data); i += len(zeroes) {
+		endOffset := i + len(zeroes)
+
+		if endOffset > len(chunk.Data) {
+			endOffset = len(chunk.Data)
+		}
+
+		dataToCheck := chunk.Data[i:endOffset]
+
+		if !bytes.Equal(dataToCheck, zeroes[:len(dataToCheck)]) {
+			return false
+		}
+	}
+	return true
+}

cloud/disk_manager/internal/pkg/dataplane/common/chunk_test.go

@@ -0,0 +1,22 @@
+package common
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+////////////////////////////////////////////////////////////////////////////////
+
+func TestCheckDataIsAllZeroes(t *testing.T) {
+	chunk := Chunk{
+		Data: []byte{1, 2, 3, 4, 5},
+	}
+	require.False(t, chunk.CheckDataIsAllZeroes())
+
+	chunk = Chunk{Data: []byte{0, 0, 0}}
+	require.True(t, chunk.CheckDataIsAllZeroes())
+
+	chunk = Chunk{Data: make([]byte, 1024*1024*5)}
+	require.True(t, chunk.CheckDataIsAllZeroes())
+}

cloud/disk_manager/internal/pkg/dataplane/common/transfer.go

@@ -342,6 +342,10 @@ func (t Transferer) goRead(
 						return
 					}
 
+					if !chunk.Zero {
+						chunk.Zero = chunk.CheckDataIsAllZeroes()
+					}
+
 					select {
 					case chunks <- chunk:
 					case <-ctx.Done():

cloud/disk_manager/internal/pkg/dataplane/common/ya.make

@@ -7,6 +7,7 @@ SRCS(
 
 GO_TEST_SRCS(
     transfer_test.go
+    chunk_test.go
 )
 
 END()