cubefs

1
// Copyright 2022 The CubeFS Authors.
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//     http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
12
// implied. See the License for the specific language governing
13
// permissions and limitations under the License.
14

15
package ec
16

17
import (
18
	"bytes"
19
	"crypto/rand"
20
	mrand "math/rand"
21
	"reflect"
22
	"testing"
23

24
	"github.com/stretchr/testify/require"
25

26
	"github.com/cubefs/cubefs/blobstore/common/codemode"
27
	"github.com/cubefs/cubefs/blobstore/util/bytespool"
28
)
29

30
var srcData = []byte("Hello world")
31

32
func copyShards(a [][]byte) [][]byte {
33
	b := make([][]byte, len(a))
34
	for i := range a {
35
		b[i] = append(b[i], a[i]...)
36
	}
37
	return b
38
}
39

40
func TestEncoderNew(t *testing.T) {
41
	{
42
		_, err := NewEncoder(Config{CodeMode: codemode.Tactic{}})
43
		require.ErrorIs(t, err, ErrInvalidCodeMode)
44
	}
45
	{
46
		_, err := NewEncoder(Config{CodeMode: codemode.EC15P12.Tactic()})
47
		require.NoError(t, err)
48
		_, err = NewEncoder(Config{CodeMode: codemode.EC16P20L2.Tactic()})
49
		require.NoError(t, err)
50
	}
51
}
52

53
func TestEncoder(t *testing.T) {
54
	cfg := Config{
55
		CodeMode:     codemode.EC15P12.Tactic(),
56
		EnableVerify: true,
57
		Concurrency:  10,
58
	}
59
	encoder, err := NewEncoder(cfg)
60
	require.NoError(t, err)
61

62
	// source data split
63
	shards, err := encoder.Split(srcData)
64
	require.NoError(t, err)
65

66
	// encode data
67
	err = encoder.Encode(shards)
68
	require.NoError(t, err)
69
	wbuff := bytes.NewBuffer(make([]byte, 0))
70
	err = encoder.Join(wbuff, shards, len(srcData))
71
	require.NoError(t, err)
72
	require.Equal(t, srcData, wbuff.Bytes())
73

74
	dataShards := encoder.GetDataShards(shards)
75
	// set one data shards broken
76
	for i := range dataShards[0] {
77
		dataShards[0][i] = 222
78
	}
79
	// reconstruct data and check
80
	err = encoder.ReconstructData(shards, []int{0})
81
	require.NoError(t, err)
82
	wbuff = bytes.NewBuffer(make([]byte, 0))
83
	err = encoder.Join(wbuff, shards, len(srcData))
84
	require.NoError(t, err)
85
	require.Equal(t, srcData, wbuff.Bytes())
86

87
	// reconstruct shard and check
88
	parityShards := encoder.GetParityShards(shards)
89
	for i := range parityShards[1] {
90
		parityShards[1][i] = 11
91
	}
92
	err = encoder.Reconstruct(shards, []int{cfg.CodeMode.N + 1})
93
	require.NoError(t, err)
94
	ok, err := encoder.Verify(shards)
95
	require.NoError(t, err)
96
	require.True(t, ok)
97
	wbuff = bytes.NewBuffer(make([]byte, 0))
98
	err = encoder.Join(wbuff, shards, len(srcData))
99
	require.NoError(t, err)
100
	require.Equal(t, srcData, wbuff.Bytes())
101

102
	ls := encoder.GetLocalShards(shards)
103
	require.Equal(t, 0, len(ls))
104
	si := encoder.GetShardsInIdc(shards, 0)
105
	require.Equal(t, (cfg.CodeMode.N+cfg.CodeMode.M)/3, len(si))
106
}
107

108
func TestLrcEncoder(t *testing.T) {
109
	cfg := Config{
110
		CodeMode:     codemode.EC6P10L2.Tactic(),
111
		EnableVerify: true,
112
	}
113
	encoder, err := NewEncoder(cfg)
114
	require.NoError(t, err)
115

116
	_, err = encoder.Split([]byte{})
117
	require.Error(t, err)
118

119
	// source data split
120
	shards, err := encoder.Split(srcData)
121
	require.NoError(t, err)
122
	{
123
		enoughBuff := make([]byte, 1<<10)
124
		copy(enoughBuff, srcData)
125
		enoughBuff = enoughBuff[:len(srcData)]
126
		_, err := encoder.Split(enoughBuff)
127
		require.NoError(t, err)
128
	}
129

130
	invalidShards := shards[:len(shards)-1]
131
	require.ErrorIs(t, encoder.Encode(invalidShards), ErrInvalidShards)
132
	require.ErrorIs(t, encoder.Encode(nil), ErrInvalidShards)
133

134
	// encode data
135
	err = encoder.Encode(shards)
136
	require.NoError(t, err)
137
	wbuff := bytes.NewBuffer(make([]byte, 0))
138
	err = encoder.Join(wbuff, shards, len(srcData))
139
	require.NoError(t, err)
140
	require.Equal(t, srcData, wbuff.Bytes())
141

142
	dataShards := encoder.GetDataShards(shards)
143
	// set one data shard broken
144
	for i := range dataShards[0] {
145
		dataShards[0][i] = 222
146
	}
147

148
	// test verify failed
149
	ok, err := encoder.Verify(shards)
150
	require.NoError(t, err)
151
	require.False(t, ok)
152

153
	// reconstruct data and check
154
	err = encoder.ReconstructData(shards, []int{0})
155
	require.NoError(t, err)
156
	wbuff = bytes.NewBuffer(make([]byte, 0))
157
	err = encoder.Join(wbuff, shards, len(srcData))
158
	require.NoError(t, err)
159
	require.Equal(t, srcData, wbuff.Bytes())
160

161
	// Local reconstruct shard and check
162
	localShardsInIdc := encoder.GetShardsInIdc(shards, 0)
163
	for idx := 0; idx < len(localShardsInIdc); idx++ {
164
		// set wrong data
165
		for i := range localShardsInIdc[idx] {
166
			localShardsInIdc[idx][i] = 11
167
		}
168
		// check must be false when a shard broken
169
		ok, err := encoder.Verify(shards)
170
		require.NoError(t, err)
171
		require.False(t, ok)
172

173
		err = encoder.Reconstruct(localShardsInIdc, []int{idx})
174
		require.NoError(t, err)
175
		ok, err = encoder.Verify(shards)
176
		require.NoError(t, err)
177
		require.True(t, ok)
178
	}
179

180
	badIdxs := make([]int, 0)
181

182
	// add a local broken
183
	for j := range shards[cfg.CodeMode.N+cfg.CodeMode.M+1] {
184
		shards[cfg.CodeMode.N+cfg.CodeMode.M+1][j] = 222
185
	}
186
	badIdxs = append(badIdxs, cfg.CodeMode.N+cfg.CodeMode.M+1)
187

188
	// test local verify failed
189
	ok, err = encoder.Verify(shards)
190
	require.NoError(t, err)
191
	require.False(t, ok)
192

193
	// global reconstruct shard and check
194
	dataShards = encoder.GetDataShards(shards)
195
	parityShards := encoder.GetParityShards(shards)
196
	for i := 0; i < cfg.CodeMode.M; i++ {
197
		if i%2 == 0 {
198
			badIdxs = append(badIdxs, i)
199
			// set wrong data
200
			if i < len(dataShards) {
201
				for j := range dataShards[i] {
202
					dataShards[i][j] = 222
203
				}
204
			}
205
		} else {
206
			badIdxs = append(badIdxs, cfg.CodeMode.N+i)
207
			// set wrong data
208
			for j := range parityShards[i] {
209
				parityShards[i][j] = 222
210
			}
211
		}
212
	}
213

214
	// test verify failed
215
	ok, err = encoder.Verify(shards)
216
	require.NoError(t, err)
217
	require.False(t, ok)
218

219
	err = encoder.Reconstruct(shards, badIdxs)
220
	require.NoError(t, err)
221
	ok, err = encoder.Verify(shards)
222
	require.NoError(t, err)
223
	require.True(t, ok)
224
	wbuff = bytes.NewBuffer(make([]byte, 0))
225
	err = encoder.Join(wbuff, shards, len(srcData))
226
	require.NoError(t, err)
227
	require.Equal(t, srcData, wbuff.Bytes())
228

229
	ls := encoder.GetLocalShards(shards)
230
	require.Equal(t, cfg.CodeMode.L, len(ls))
231
	si := encoder.GetShardsInIdc(shards, 0)
232
	require.Equal(t, (cfg.CodeMode.N+cfg.CodeMode.M+cfg.CodeMode.L)/cfg.CodeMode.AZCount, len(si))
233

234
	// test data len
235
	shards[badIdxs[0]] = shards[badIdxs[0]][:0]
236
	ok, err = encoder.Verify(shards)
237
	require.Error(t, err)
238
	require.False(t, ok)
239

240
	err = encoder.Reconstruct(shards, badIdxs)
241
	require.NoError(t, err)
242

243
	shards[badIdxs[len(badIdxs)-1]] = shards[len(badIdxs)-1][:0]
244
	ok, err = encoder.Verify(shards)
245
	require.Error(t, err)
246
	require.False(t, ok)
247
}
248

249
func TestLrcReconstruct(t *testing.T) {
250
	for _, cm := range codemode.GetAllCodeModes() {
251
		testLrcReconstruct(t, cm)
252
	}
253
}
254

255
func testLrcReconstruct(t *testing.T, cm codemode.CodeMode) {
256
	tactic := cm.Tactic()
257
	cfg := Config{CodeMode: tactic, EnableVerify: true}
258
	encoder, _ := NewEncoder(cfg)
259

260
	data := make([]byte, (1<<16)+mrand.Intn(1<<16))
261
	rand.Read(data)
262

263
	shards, err := encoder.Split(data)
264
	require.NoError(t, err)
265
	require.NoError(t, encoder.Encode(shards))
266

267
	origin := copyShards(shards)
268
	bads := make([]int, 0)
269
	for badIdx := tactic.N + tactic.M; badIdx < cm.GetShardNum(); badIdx++ {
270
		bads = append(bads, badIdx)
271
		for _, idx := range bads {
272
			bytespool.Zero(shards[idx])
273
			shards[idx] = shards[idx][:0]
274
		}
275
		require.NoError(t, encoder.Reconstruct(shards, bads))
276
		require.True(t, reflect.DeepEqual(origin, shards))
277
	}
278
	for badIdx := 0; badIdx < tactic.N+tactic.M; badIdx++ {
279
		bads = append(bads, badIdx)
280
	}
281
	require.Error(t, encoder.Reconstruct(copyShards(shards), bads))
282

283
	// use local ec reconstruct
284
	for azIdx := 0; azIdx < tactic.AZCount; azIdx++ {
285
		locals, n, m := tactic.LocalStripeInAZ(azIdx)
286
		var localShards [][]byte
287
		for _, idx := range locals {
288
			localShards = append(localShards, shards[idx])
289
		}
290
		localOrigin := copyShards(localShards)
291

292
		bads := make([]int, 0)
293
		for badIdx := n; badIdx < n+m; badIdx++ {
294
			bads = append(bads, badIdx)
295
			for _, idx := range bads {
296
				bytespool.Zero(localShards[idx])
297
				localShards[idx] = localShards[idx][:0]
298
			}
299
			require.NoError(t, encoder.Reconstruct(localShards, bads))
300
			require.True(t, reflect.DeepEqual(localOrigin, localShards))
301
		}
302
		if n > 0 {
303
			bads = append(bads, n-1)
304
			require.Error(t, encoder.Reconstruct(localShards, bads))
305
		}
306
	}
307
}
308