cubefs

1
// Copyright 2018 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 metanode
16

17
import (
18
	"bytes"
19
	"encoding/binary"
20
	"fmt"
21
	"math"
22

23
	"github.com/cubefs/cubefs/proto"
24
	"github.com/cubefs/cubefs/util/log"
25
)
26

27
// Dentry wraps necessary properties of the `dentry` information in file system.
28
// Marshal exporterKey:
29
//  +-------+----------+------+
30
//  | item  | ParentId | Name |
31
//  +-------+----------+------+
32
//  | bytes |    8     | rest |
33
//  +-------+----------+------+
34
// Marshal value:
35
//  +-------+-------+------+
36
//  | item  | Inode | Type |
37
//  +-------+-------+------+
38
//  | bytes |   8   |   4  |
39
//  +-------+-------+------+
40
// Marshal entity:
41
//  +-------+-----------+--------------+-----------+--------------+
42
//  | item  | KeyLength | MarshaledKey | ValLength | MarshaledVal |
43
//  +-------+-----------+--------------+-----------+--------------+
44
//  | bytes |     4     |   KeyLength  |     4     |   ValLength  |
45
//  +-------+-----------+--------------+-----------+--------------+
46

47
type DentryMultiSnap struct {
48
	VerSeq     uint64
49
	dentryList DentryBatch
50
}
51

52
type Dentry struct {
53
	ParentId uint64 // FileID value of the parent inode.
54
	Name     string // Name of the current dentry.
55
	Inode    uint64 // FileID value of the current inode.
56
	Type     uint32
57
	// snapshot
58
	multiSnap *DentryMultiSnap
59
}
60

61
func NewDentrySnap(seq uint64) *DentryMultiSnap {
62
	return &DentryMultiSnap{
63
		VerSeq: seq,
64
	}
65
}
66

67
func (d *Dentry) getSnapListLen() int {
68
	if d.multiSnap == nil {
69
		return 0
70
	}
71
	return len(d.multiSnap.dentryList)
72
}
73

74
func (d *Dentry) addVersion(ver uint64) {
75
	dn := d.CopyDirectly().(*Dentry)
76
	dn.setVerSeq(d.getSeqFiled())
77
	d.setVerSeq(ver)
78
	d.multiSnap.dentryList = append([]*Dentry{dn}, d.multiSnap.dentryList...)
79
}
80

81
func (d *Dentry) setVerSeq(verSeq uint64) {
82
	if verSeq == 0 {
83
		return
84
	}
85
	if d.multiSnap == nil {
86
		d.multiSnap = NewDentrySnap(verSeq)
87
	} else {
88
		d.multiSnap.VerSeq = verSeq
89
	}
90
}
91

92
func (d *Dentry) getSeqFiled() (verSeq uint64) {
93
	if d.multiSnap == nil {
94
		return 0
95
	}
96
	return d.multiSnap.VerSeq
97
}
98

99
func isSeqEqual(ver_1 uint64, ver_2 uint64) bool {
100
	if isInitSnapVer(ver_1) {
101
		ver_1 = 0
102
	}
103
	if isInitSnapVer(ver_2) {
104
		ver_2 = 0
105
	}
106
	return (ver_1 & math.MaxInt64) == (ver_2 & math.MaxInt64)
107
}
108

109
func (d *Dentry) getVerSeq() (verSeq uint64) {
110
	if d.multiSnap == nil {
111
		return 0
112
	}
113
	return d.multiSnap.VerSeq & math.MaxInt64
114
}
115

116
func (d *Dentry) isDeleted() bool {
117
	if d.multiSnap == nil {
118
		return false
119
	}
120
	return (d.multiSnap.VerSeq >> 63) != 0
121
}
122

123
func (d *Dentry) setDeleted() {
124
	if d.multiSnap == nil {
125
		log.LogErrorf("action[setDeleted] d %v be set deleted not found multiSnap", d)
126
		return
127
	}
128
	log.LogDebugf("action[setDeleted] d %v be set deleted", d)
129
	d.multiSnap.VerSeq |= uint64(1) << 63
130
}
131

132
func (d *Dentry) minimizeSeq() (verSeq uint64) {
133
	cnt := d.getSnapListLen()
134
	if cnt == 0 {
135
		return d.getVerSeq()
136
	}
137
	return d.multiSnap.dentryList[cnt-1].getVerSeq()
138
}
139

140
func (d *Dentry) isEffective(verSeq uint64) bool {
141
	if verSeq == 0 {
142
		return false
143
	}
144
	if isInitSnapVer(verSeq) {
145
		verSeq = 0
146
	}
147
	return verSeq >= d.minimizeSeq()
148
}
149

150
// isHit return the right version or else return the version can be seen
151
func (d *Dentry) getDentryFromVerList(verSeq uint64, isHit bool) (den *Dentry, idx int) {
152
	if verSeq == 0 || (verSeq >= d.getVerSeq() && !isInitSnapVer(verSeq)) {
153
		if d.isDeleted() {
154
			log.LogDebugf("action[getDentryFromVerList] tmp dentry %v, is deleted, seq [%v]", d, d.getVerSeq())
155
			return
156
		}
157
		return d, 0
158
	}
159

160
	// read the oldest version snapshot,the oldest version is 0 should make a different with the lastest uncommit version read(with seq 0)
161
	if isInitSnapVer(verSeq) {
162
		if d.getVerSeq() == 0 {
163
			return d, 0
164
		}
165
		denListLen := d.getSnapListLen()
166
		if denListLen == 0 {
167
			return
168
		}
169
		den = d.multiSnap.dentryList[denListLen-1]
170
		if d.multiSnap.dentryList[denListLen-1].getVerSeq() != 0 || d.multiSnap.dentryList[denListLen-1].isDeleted() {
171
			return nil, 0
172
		}
173
		return den, denListLen
174
	}
175
	if d.multiSnap == nil {
176
		return
177
	}
178
	for id, lDen := range d.multiSnap.dentryList {
179
		if verSeq < lDen.getVerSeq() {
180
			log.LogDebugf("action[getDentryFromVerList] den in ver list %v, return nil, request seq [%v], history ver seq [%v]", lDen, verSeq, lDen.getVerSeq())
181
		} else {
182
			if lDen.isDeleted() {
183
				log.LogDebugf("action[getDentryFromVerList] den in ver list %v, return nil due to latest is deleted", lDen)
184
				return
185
			}
186
			if isHit && lDen.getVerSeq() != verSeq {
187
				log.LogDebugf("action[getDentryFromVerList] den in ver list %v, return nil due to ver not equal %v vs %v", lDen, lDen.getVerSeq(), verSeq)
188
				return
189
			}
190
			return lDen, id + 1
191
		}
192
	}
193
	log.LogDebugf("action[getDentryFromVerList] den in ver list not found right dentry with seq [%v]", verSeq)
194
	return
195
}
196

197
func (d *Dentry) getLastestVer(reqVerSeq uint64, commit bool, verlist []*proto.VolVersionInfo) (uint64, bool) {
198
	if len(verlist) == 0 {
199
		return 0, false
200
	}
201
	for id, info := range verlist {
202
		if commit && id == len(verlist)-1 {
203
			break
204
		}
205
		if info.Ver >= reqVerSeq { // include reqSeq itself
206
			return info.Ver, true
207
		}
208
	}
209

210
	log.LogDebugf("action[getLastestVer] inode[%v] reqVerseq [%v] not found, the largetst one %v",
211
		d.Inode, reqVerSeq, verlist[len(verlist)-1].Ver)
212
	return 0, false
213
}
214

215
func (d *Dentry) deleteTopLayer(mpVerSeq uint64) (rd *Dentry, dmore bool, clean bool) {
216
	if d.isDeleted() {
217
		log.LogDebugf("action[deleteTopLayer.delSeq_0] do noting dentry %v seq 0 be deleted before", d)
218
		return nil, false, false
219
	}
220

221
	// if there's no snapshot itself, nor have snapshot after dentry's ver then need unlink directly and make no snapshot
222
	// just move to upper layer,the request snapshot be dropped
223
	if d.getSnapListLen() == 0 {
224
		if d.getVerSeq() == mpVerSeq {
225
			// operate dentry directly
226
			log.LogDebugf("action[deleteTopLayer.delSeq_0] no snapshot depend on this dentry,could drop seq 0 dentry %v", d)
227
			return d, true, true
228
		}
229
	}
230

231
	if d.getVerSeq() < mpVerSeq {
232
		dn := d.CopyDirectly()
233
		dn.(*Dentry).setVerSeq(d.getVerSeq())
234
		d.setVerSeq(mpVerSeq)
235
		d.multiSnap.dentryList = append([]*Dentry{dn.(*Dentry)}, d.multiSnap.dentryList...)
236
		log.LogDebugf("action[deleteTopLayer.delSeq_0] create version and push to dentry list. dentry %v", dn.(*Dentry))
237
	} else {
238
		d.setVerSeq(mpVerSeq)
239
	}
240
	d.setVerSeq(mpVerSeq)
241
	d.setDeleted() // denParm create at the same version.no need to push to history list
242
	log.LogDebugf("action[deleteTopLayer.delSeq_0] den %v be set deleted at version seq [%v]", d, mpVerSeq)
243

244
	return d, true, false
245
}
246

247
func (d *Dentry) updateTopLayerSeq(delVerSeq uint64, verlist []*proto.VolVersionInfo) (rd *Dentry, dmore bool, clean bool) {
248
	if !isSeqEqual(delVerSeq, d.getVerSeq()) {
249
		// header layer do nothing and be depends on should not be dropped
250
		log.LogDebugf("action[updateTopLayerSeq.inSnapList_del_%v] den %v first layer do nothing", delVerSeq, d)
251
		return d, false, false
252
	}
253
	for _, info := range verlist {
254
		if info.Ver > d.getVerSeq() {
255
			d.setVerSeq(info.Ver)
256
			return d, false, false
257
		}
258
	}
259
	return d, true, true
260
}
261

262
func (d *Dentry) cleanDeletedVersion(index int) (bDrop bool) {
263
	if index == 0 {
264
		if len(d.multiSnap.dentryList) == 0 && d.isDeleted() {
265
			bDrop = true
266
		}
267
		return
268
	}
269
	delIdx := index - 1
270
	if !d.multiSnap.dentryList[delIdx].isDeleted() {
271
		return
272
	}
273

274
	// del the dentry before
275
	log.LogDebugf("ction[cleanDeleteVersion] dentry (%v) delete the last seq [%v] which set deleted before",
276
		d, d.multiSnap.dentryList[delIdx].getVerSeq())
277
	d.multiSnap.dentryList = append(d.multiSnap.dentryList[:delIdx], d.multiSnap.dentryList[:delIdx+1]...)
278

279
	if len(d.multiSnap.dentryList) == 0 && d.isDeleted() {
280
		log.LogDebugf("ction[cleanDeleteVersion] dentry (%v) require to be deleted", d)
281
		bDrop = true
282
	}
283
	return
284
}
285

286
// the lastest dentry may be deleted before and set status DentryDeleted,
287
// the scope of  deleted happened from the DentryDeleted flag owner(include in) to the file with the same name be created is invisible,
288
// if create anther dentry with larger verSeq, put the deleted dentry to the history list.
289
// return doMore bool.True means need do next step on caller such as unlink parentIO
290
func (d *Dentry) deleteVerSnapshot(delVerSeq uint64, mpVerSeq uint64, verlist []*proto.VolVersionInfo) (rd *Dentry, dmore bool, clean bool) { // bool is doMore
291
	log.LogDebugf("action[deleteVerSnapshot] enter.dentry %v delVerseq [%v] mpver [%v] verList %v", d, delVerSeq, mpVerSeq, verlist)
292
	// create denParm version
293
	if !isInitSnapVer(delVerSeq) && delVerSeq > mpVerSeq {
294
		panic(fmt.Sprintf("Dentry version %v large than mp[%v]", delVerSeq, mpVerSeq))
295
	}
296

297
	if delVerSeq == 0 {
298
		return d.deleteTopLayer(mpVerSeq)
299
	} else {
300
		var (
301
			idx    int
302
			den    *Dentry
303
			endSeq uint64
304
		)
305
		if den, idx = d.getDentryFromVerList(delVerSeq, true); den == nil {
306
			log.LogDebugf("action[deleteVerSnapshot.inSnapList_del_%v] den %v not found", delVerSeq, d)
307
			return nil, false, false
308
		}
309
		if idx == 0 { // top layer
310
			return d.updateTopLayerSeq(delVerSeq, verlist)
311
		}
312
		// if any alive snapshot in mp dimension exist in seq scope from den to next ascend neighbor, dio snapshot be keep or else drop
313
		startSeq := den.getVerSeq()
314
		realIdx := idx - 1 // index in history list layer
315
		if realIdx == 0 {
316
			endSeq = d.getVerSeq()
317
		} else {
318
			endSeq = d.multiSnap.dentryList[realIdx-1].getVerSeq()
319
			if d.multiSnap.dentryList[realIdx-1].isDeleted() {
320
				log.LogInfof("action[deleteVerSnapshot.inSnapList_del_%v] inode[%v] layer %v name %v be deleted already!",
321
					delVerSeq, d.Inode, realIdx, d.multiSnap.dentryList[realIdx-1].Name)
322
			}
323
		}
324

325
		log.LogDebugf("action[deleteVerSnapshot.inSnapList_del_%v] inode[%v] try drop multiVersion idx %v effective seq scope [%v,%v) ", delVerSeq,
326
			d.Inode, realIdx, den.getVerSeq(), endSeq)
327

328
		for _, info := range verlist {
329
			if info.Ver >= startSeq && info.Ver < endSeq { // the version itself not include in
330
				log.LogDebugf("action[deleteVerSnapshotInList.inSnapList_del_%v] inode[%v] dir layer idx %v include snapshot %v.don't drop", delVerSeq, den.Inode, realIdx, info.Ver)
331
				// there's some snapshot depends on the version trying to be deleted,
332
				// keep it,all the snapshots which depends on this version will reach here when make snapshot delete, and found the scope is minimized
333
				// other versions depends upon this version will be found zero finally after deletions and do clean
334
				den.setVerSeq(info.Ver)
335
				return den, false, false
336
			}
337
			if info.Ver >= endSeq {
338
				break
339
			}
340
			log.LogDebugf("action[deleteVerSnapshotInList.inSnapList_del_%v] inode[%v] try drop scope [%v, %v), mp ver [%v] not suitable",
341
				delVerSeq, den.Inode, den.getVerSeq(), endSeq, info.Ver)
342
		}
343

344
		log.LogDebugf("action[deleteVerSnapshotInList.inSnapList_del_%v] inode[%v] try drop multiVersion idx %v", delVerSeq, den.Inode, realIdx)
345
		d.multiSnap.dentryList = append(d.multiSnap.dentryList[:realIdx], d.multiSnap.dentryList[realIdx+1:]...)
346
		if d.cleanDeletedVersion(realIdx) {
347
			return den, true, true
348
		}
349
		return den, false, false
350
	}
351
}
352

353
func (d *Dentry) String() string {
354
	str := fmt.Sprintf("dentry(name:[%v],parentId:[%v],inode:[%v],type:[%v],seq:[%v],isDeleted:[%v],dentryList_len[%v])",
355
		d.Name, d.ParentId, d.Inode, d.Type, d.getVerSeq(), d.isDeleted(), d.getSnapListLen())
356
	if d.getSnapListLen() > 0 {
357
		for idx, den := range d.multiSnap.dentryList {
358
			str += fmt.Sprintf("idx:%v,content(%v))", idx, den)
359
		}
360
	}
361
	return str
362
}
363

364
type TxDentry struct {
365
	// ParInode *Inode
366
	Dentry *Dentry
367
	TxInfo *proto.TransactionInfo
368
}
369

370
func NewTxDentry(parentID uint64, name string, ino uint64, mode uint32, parInode *Inode, txInfo *proto.TransactionInfo) *TxDentry {
371
	dentry := &Dentry{
372
		ParentId: parentID,
373
		Name:     name,
374
		Inode:    ino,
375
		Type:     mode,
376
	}
377

378
	txDentry := &TxDentry{
379
		// ParInode: parInode,
380
		Dentry: dentry,
381
		TxInfo: txInfo,
382
	}
383
	return txDentry
384
}
385

386
func (td *TxDentry) Marshal() (result []byte, err error) {
387
	buff := bytes.NewBuffer(make([]byte, 0))
388

389
	//bs, err := td.ParInode.Marshal()
390
	//if err != nil {
391
	//	return nil, err
392
	//}
393
	//if err = binary.Write(buff, binary.BigEndian, uint32(len(bs))); err != nil {
394
	//	return nil, err
395
	//}
396
	//if _, err := buff.Write(bs); err != nil {
397
	//	return nil, err
398
	//}
399

400
	bs, err := td.Dentry.Marshal()
401
	if err != nil {
402
		return nil, err
403
	}
404
	if err = binary.Write(buff, binary.BigEndian, uint32(len(bs))); err != nil {
405
		return nil, err
406
	}
407
	if _, err := buff.Write(bs); err != nil {
408
		return nil, err
409
	}
410

411
	bs, err = td.TxInfo.Marshal()
412
	if err != nil {
413
		return nil, err
414
	}
415
	if err = binary.Write(buff, binary.BigEndian, uint32(len(bs))); err != nil {
416
		return nil, err
417
	}
418
	if _, err := buff.Write(bs); err != nil {
419
		return nil, err
420
	}
421
	result = buff.Bytes()
422
	return
423
}
424

425
func (td *TxDentry) Unmarshal(raw []byte) (err error) {
426
	buff := bytes.NewBuffer(raw)
427
	var dataLen uint32
428
	if err = binary.Read(buff, binary.BigEndian, &dataLen); err != nil {
429
		return
430
	}
431
	data := make([]byte, int(dataLen))
432
	if _, err = buff.Read(data); err != nil {
433
		return
434
	}
435

436
	dentry := &Dentry{}
437
	if err = dentry.Unmarshal(data); err != nil {
438
		return
439
	}
440
	td.Dentry = dentry
441

442
	if err = binary.Read(buff, binary.BigEndian, &dataLen); err != nil {
443
		return
444
	}
445
	data = make([]byte, int(dataLen))
446
	if _, err = buff.Read(data); err != nil {
447
		return
448
	}
449
	txInfo := proto.NewTransactionInfo(0, proto.TxTypeUndefined)
450
	if err = txInfo.Unmarshal(data); err != nil {
451
		return
452
	}
453
	td.TxInfo = txInfo
454
	return
455
}
456

457
type TxUpdateDentry struct {
458
	OldDentry *Dentry
459
	NewDentry *Dentry
460
	TxInfo    *proto.TransactionInfo
461
}
462

463
func NewTxUpdateDentry(oldDentry *Dentry, newDentry *Dentry, txInfo *proto.TransactionInfo) *TxUpdateDentry {
464
	txUpdateDentry := &TxUpdateDentry{
465
		OldDentry: oldDentry,
466
		NewDentry: newDentry,
467
		TxInfo:    txInfo,
468
	}
469
	return txUpdateDentry
470
}
471

472
func (td *TxUpdateDentry) Marshal() (result []byte, err error) {
473
	buff := bytes.NewBuffer(make([]byte, 0))
474
	bs, err := td.OldDentry.Marshal()
475
	if err != nil {
476
		return nil, err
477
	}
478
	if err = binary.Write(buff, binary.BigEndian, uint32(len(bs))); err != nil {
479
		return nil, err
480
	}
481
	if _, err := buff.Write(bs); err != nil {
482
		return nil, err
483
	}
484

485
	bs, err = td.NewDentry.Marshal()
486
	if err != nil {
487
		return nil, err
488
	}
489
	if err = binary.Write(buff, binary.BigEndian, uint32(len(bs))); err != nil {
490
		return nil, err
491
	}
492
	if _, err := buff.Write(bs); err != nil {
493
		return nil, err
494
	}
495

496
	bs, err = td.TxInfo.Marshal()
497
	if err != nil {
498
		return nil, err
499
	}
500
	if err = binary.Write(buff, binary.BigEndian, uint32(len(bs))); err != nil {
501
		return nil, err
502
	}
503
	if _, err := buff.Write(bs); err != nil {
504
		return nil, err
505
	}
506
	result = buff.Bytes()
507
	return
508
}
509

510
func (td *TxUpdateDentry) Unmarshal(raw []byte) (err error) {
511
	buff := bytes.NewBuffer(raw)
512
	var dataLen uint32
513
	if err = binary.Read(buff, binary.BigEndian, &dataLen); err != nil {
514
		return
515
	}
516
	data := make([]byte, int(dataLen))
517
	if _, err = buff.Read(data); err != nil {
518
		return
519
	}
520

521
	oldDentry := &Dentry{}
522
	if err = oldDentry.Unmarshal(data); err != nil {
523
		return
524
	}
525
	td.OldDentry = oldDentry
526

527
	if err = binary.Read(buff, binary.BigEndian, &dataLen); err != nil {
528
		return
529
	}
530
	data = make([]byte, int(dataLen))
531
	if _, err = buff.Read(data); err != nil {
532
		return
533
	}
534

535
	newDentry := &Dentry{}
536
	if err = newDentry.Unmarshal(data); err != nil {
537
		return
538
	}
539
	td.NewDentry = newDentry
540

541
	if err = binary.Read(buff, binary.BigEndian, &dataLen); err != nil {
542
		return
543
	}
544
	data = make([]byte, int(dataLen))
545
	if _, err = buff.Read(data); err != nil {
546
		return
547
	}
548
	txInfo := proto.NewTransactionInfo(0, proto.TxTypeUndefined)
549
	if err = txInfo.Unmarshal(data); err != nil {
550
		return
551
	}
552
	td.TxInfo = txInfo
553
	return
554
}
555

556
type DentryBatch []*Dentry
557

558
// todo(leon chang), buffer need alloc first before and write directly consider the space and performance
559

560
// Marshal marshals a dentry into a byte array.
561
func (d *Dentry) Marshal() (result []byte, err error) {
562
	keyBytes := d.MarshalKey()
563
	valBytes := d.MarshalValue()
564
	keyLen := uint32(len(keyBytes))
565
	valLen := uint32(len(valBytes))
566
	buff := bytes.NewBuffer(make([]byte, 0))
567
	buff.Grow(int(keyLen + valLen + 8))
568

569
	if err = binary.Write(buff, binary.BigEndian, keyLen); err != nil {
570
		return
571
	}
572
	if _, err = buff.Write(keyBytes); err != nil {
573
		return
574
	}
575
	if err = binary.Write(buff, binary.BigEndian, valLen); err != nil {
576
		return nil, err
577
	}
578
	if _, err = buff.Write(valBytes); err != nil {
579
		return
580
	}
581
	result = buff.Bytes()
582
	return
583
}
584

585
// Unmarshal unmarshals the dentry from a byte array.
586
func (d *Dentry) Unmarshal(raw []byte) (err error) {
587
	var (
588
		keyLen uint32
589
		valLen uint32
590
	)
591
	buff := bytes.NewBuffer(raw)
592
	if err = binary.Read(buff, binary.BigEndian, &keyLen); err != nil {
593
		return
594
	}
595
	keyBytes := make([]byte, keyLen)
596
	if _, err = buff.Read(keyBytes); err != nil {
597
		return
598
	}
599
	if err = d.UnmarshalKey(keyBytes); err != nil {
600
		return
601
	}
602
	if err = binary.Read(buff, binary.BigEndian, &valLen); err != nil {
603
		return
604
	}
605
	valBytes := make([]byte, valLen)
606
	if _, err = buff.Read(valBytes); err != nil {
607
		return
608
	}
609
	err = d.UnmarshalValue(valBytes)
610
	return
611
}
612

613
// Marshal marshals the dentryBatch into a byte array.
614
func (d DentryBatch) Marshal() ([]byte, error) {
615
	buff := bytes.NewBuffer(make([]byte, 0))
616
	if err := binary.Write(buff, binary.BigEndian, uint32(len(d))); err != nil {
617
		return nil, err
618
	}
619
	for _, dentry := range d {
620
		bs, err := dentry.Marshal()
621
		if err != nil {
622
			return nil, err
623
		}
624
		if err = binary.Write(buff, binary.BigEndian, uint32(len(bs))); err != nil {
625
			return nil, err
626
		}
627
		if _, err := buff.Write(bs); err != nil {
628
			return nil, err
629
		}
630
	}
631
	return buff.Bytes(), nil
632
}
633

634
// Unmarshal unmarshals the dentryBatch.
635
func DentryBatchUnmarshal(raw []byte) (DentryBatch, error) {
636
	buff := bytes.NewBuffer(raw)
637
	var batchLen uint32
638
	if err := binary.Read(buff, binary.BigEndian, &batchLen); err != nil {
639
		return nil, err
640
	}
641

642
	result := make(DentryBatch, 0, int(batchLen))
643

644
	var dataLen uint32
645
	for j := 0; j < int(batchLen); j++ {
646
		if err := binary.Read(buff, binary.BigEndian, &dataLen); err != nil {
647
			return nil, err
648
		}
649
		data := make([]byte, int(dataLen))
650
		if _, err := buff.Read(data); err != nil {
651
			return nil, err
652
		}
653
		den := &Dentry{}
654
		if err := den.Unmarshal(data); err != nil {
655
			return nil, err
656
		}
657
		result = append(result, den)
658
	}
659

660
	return result, nil
661
}
662

663
// Less tests whether the current dentry is less than the given one.
664
// This method is necessary fot B-Tree item implementation.
665
func (d *Dentry) Less(than BtreeItem) (less bool) {
666
	dentry, ok := than.(*Dentry)
667
	less = ok && ((d.ParentId < dentry.ParentId) || ((d.ParentId == dentry.ParentId) && (d.Name < dentry.Name)))
668
	return
669
}
670

671
func (d *Dentry) CopyDirectly() BtreeItem {
672
	newDentry := *d
673
	newDentry.multiSnap = nil
674
	return &newDentry
675
}
676

677
func (d *Dentry) Copy() BtreeItem {
678
	newDentry := *d
679
	if d.multiSnap != nil {
680
		newDentry.multiSnap = &DentryMultiSnap{
681
			VerSeq:     d.multiSnap.VerSeq,
682
			dentryList: d.multiSnap.dentryList,
683
		}
684
	}
685
	return &newDentry
686
}
687

688
// MarshalKey is the bytes version of the MarshalKey method which returns the byte slice result.
689
func (d *Dentry) MarshalKey() (k []byte) {
690
	buff := bytes.NewBuffer(make([]byte, 0))
691
	buff.Grow(32)
692
	if err := binary.Write(buff, binary.BigEndian, &d.ParentId); err != nil {
693
		panic(err)
694
	}
695
	buff.Write([]byte(d.Name))
696
	k = buff.Bytes()
697
	return
698
}
699

700
// UnmarshalKey unmarshals the exporterKey from bytes.
701
func (d *Dentry) UnmarshalKey(k []byte) (err error) {
702
	buff := bytes.NewBuffer(k)
703
	if err = binary.Read(buff, binary.BigEndian, &d.ParentId); err != nil {
704
		return
705
	}
706
	d.Name = string(buff.Bytes())
707
	return
708
}
709

710
func (d *Dentry) MarshalValue() []byte {
711
	buff := bytes.NewBuffer(nil)
712
	buff.Grow(24 + d.getSnapListLen()*20)
713

714
	writeBinary := func(data interface{}) {
715
		if err := binary.Write(buff, binary.BigEndian, data); err != nil {
716
			panic(err)
717
		}
718
	}
719

720
	writeBinary(&d.Inode)
721
	writeBinary(&d.Type)
722
	seq := d.getSeqFiled()
723
	if seq == 0 {
724
		return buff.Bytes()
725
	}
726
	writeBinary(&seq)
727

728
	verCnt := uint32(d.getSnapListLen())
729
	writeBinary(&verCnt)
730

731
	if d.getSnapListLen() > 0 {
732
		for _, dd := range d.multiSnap.dentryList {
733
			writeBinary(&dd.Inode)
734
			writeBinary(&dd.Type)
735
			seq = dd.getSeqFiled()
736
			writeBinary(&seq)
737
		}
738
	}
739

740
	return buff.Bytes()
741
}
742

743
func (d *Dentry) UnmarshalValue(val []byte) (err error) {
744
	buff := bytes.NewBuffer(val)
745
	if err = binary.Read(buff, binary.BigEndian, &d.Inode); err != nil {
746
		return
747
	}
748
	if err = binary.Read(buff, binary.BigEndian, &d.Type); err != nil {
749
		return
750
	}
751

752
	if len(val) >= 24 {
753
		var seq uint64
754
		if err = binary.Read(buff, binary.BigEndian, &seq); err != nil {
755
			return
756
		}
757

758
		d.multiSnap = NewDentrySnap(seq)
759

760
		verCnt := uint32(0)
761
		if err = binary.Read(buff, binary.BigEndian, &verCnt); err != nil {
762
			return
763
		}
764

765
		for i := 0; i < int(verCnt); i++ {
766
			// todo(leonchang) name and parentid should be removed to reduce space
767
			den := &Dentry{
768
				Name:     d.Name,
769
				ParentId: d.ParentId,
770
			}
771
			if err = binary.Read(buff, binary.BigEndian, &den.Inode); err != nil {
772
				return
773
			}
774
			if err = binary.Read(buff, binary.BigEndian, &den.Type); err != nil {
775
				return
776
			}
777
			if err = binary.Read(buff, binary.BigEndian, &seq); err != nil {
778
				return
779
			}
780
			if seq > 0 {
781
				den.multiSnap = NewDentrySnap(seq)
782
			}
783
			d.multiSnap.dentryList = append(d.multiSnap.dentryList, den)
784
		}
785
	}
786

787
	return
788
}
789