File: store_footer.go

package info (click to toggle)
golang-github-couchbase-moss 0.0~git20170914.0.07c86e8-4
  • links: PTS, VCS
  • area: main
  • in suites: bullseye, buster, sid
  • size: 664 kB
  • sloc: python: 230; makefile: 37
file content (493 lines) | stat: -rw-r--r-- 12,802 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
//  Copyright (c) 2016 Couchbase, Inc.
//  Licensed under the Apache License, Version 2.0 (the "License");
//  you may not use this file except in compliance with the
//  License. You may obtain a copy of the License at
//    http://www.apache.org/licenses/LICENSE-2.0
//  Unless required by applicable law or agreed to in writing,
//  software distributed under the License is distributed on an "AS
//  IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
//  express or implied. See the License for the specific language
//  governing permissions and limitations under the License.

package moss

import (
	"bytes"
	"encoding/binary"
	"encoding/json"
	"fmt"
	"time"

	"github.com/edsrzf/mmap-go"
)

func (s *Store) persistFooter(file File, footer *Footer,
	options StorePersistOptions) error {
	startTime := time.Now()

	if !options.NoSync {
		err := file.Sync()
		if err != nil {
			return err
		}
	}

	err := s.persistFooterUnsynced(file, footer)
	if err != nil {
		return err
	}

	if !options.NoSync {
		err = file.Sync()
	}

	if err == nil {
		s.histograms["PersistFooterUsecs"].Add(
			uint64(time.Since(startTime).Nanoseconds()/1000), 1)
	}

	return err
}

func (s *Store) persistFooterUnsynced(file File, footer *Footer) error {
	jBuf, err := json.Marshal(footer)
	if err != nil {
		return err
	}

	finfo, err := file.Stat()
	if err != nil {
		return err
	}

	footerPos := pageAlignCeil(finfo.Size())
	footerLen := footerBegLen + len(jBuf) + footerEndLen

	footerBuf := bytes.NewBuffer(make([]byte, 0, footerLen))
	footerBuf.Write(StoreMagicBeg)
	footerBuf.Write(StoreMagicBeg)
	binary.Write(footerBuf, StoreEndian, uint32(StoreVersion))
	binary.Write(footerBuf, StoreEndian, uint32(footerLen))
	footerBuf.Write(jBuf)
	binary.Write(footerBuf, StoreEndian, footerPos)
	binary.Write(footerBuf, StoreEndian, uint32(footerLen))
	footerBuf.Write(StoreMagicEnd)
	footerBuf.Write(StoreMagicEnd)

	footerWritten, err := file.WriteAt(footerBuf.Bytes(), footerPos)
	if err != nil {
		return err
	}
	if footerWritten != len(footerBuf.Bytes()) {
		return fmt.Errorf("store: persistFooter error writing all footerBuf")
	}

	if AllocationGranularity != StorePageSize {
		// Some platforms (windows) only support mmap()'ing at an
		// allocation granularity that's != to a page size.
		//
		// However if on such platforms there are empty segments, then
		// due to the extra space imposed by the above granularity
		// requirement, mmap() can fail complaining about insufficient
		// file space.
		//
		// To avoid this error, simply pad up the file up to a page
		// boundary.  This pad of zeroes will not interfere with file
		// recovery.
		padding := make([]byte, int(pageAlignCeil(int64(footerWritten)))-footerWritten)
		_, err = file.WriteAt(padding, footerPos+int64(footerWritten))
		if err != nil {
			return err
		}
	}

	footer.fileName = finfo.Name()
	footer.filePos = footerPos

	return nil
}

// --------------------------------------------------------

// ReadFooter reads the last valid Footer from a file.
func ReadFooter(options *StoreOptions, file File) (*Footer, error) {
	finfo, err := file.Stat()
	if err != nil {
		return nil, err
	}

	fref := &FileRef{file: file, refs: 1}

	// To avoid an EOF while reading, start scanning the footer from
	// the last byte. This is under the assumption that the footer is
	// at least 2 bytes long.
	f, err := ScanFooter(options, fref, finfo.Name(), finfo.Size()-1)
	if err != nil {
		return nil, err
	}

	fref.DecRef() // ScanFooter added its own ref-counts on success.

	return f, err
}

// --------------------------------------------------------

// ScanFooter scans a file backwards from the given pos for a valid
// Footer, adding ref-counts to fref on success.
func ScanFooter(options *StoreOptions, fref *FileRef, fileName string,
	pos int64) (*Footer, error) {
	footerBeg := make([]byte, footerBegLen)

	// Align pos to the start of a page (floor).
	pos = pageAlignFloor(pos)

	for {
		for { // Scan for StoreMagicBeg, which may be a potential footer.
			if pos <= 0 {
				return nil, ErrNoValidFooter
			}

			n, err := fref.file.ReadAt(footerBeg, pos)
			if err != nil {
				return nil, err
			}

			if n == footerBegLen &&
				bytes.Equal(StoreMagicBeg, footerBeg[:lenMagicBeg]) &&
				bytes.Equal(StoreMagicBeg, footerBeg[lenMagicBeg:2*lenMagicBeg]) {
				break
			}

			// Move pos back by page size.
			pos -= int64(StorePageSize)
		}

		// Read and check the potential footer.
		footerBegBuf := bytes.NewBuffer(footerBeg[2*lenMagicBeg:])

		var version uint32
		if err := binary.Read(footerBegBuf, StoreEndian, &version); err != nil {
			return nil, err
		}
		if version != StoreVersion {
			return nil, fmt.Errorf("store: version mismatch, "+
				"current: %v != found: %v", StoreVersion, version)
		}

		var length uint32
		if err := binary.Read(footerBegBuf, StoreEndian, &length); err != nil {
			return nil, err
		}

		data := make([]byte, int64(length)-int64(footerBegLen))

		n, err := fref.file.ReadAt(data, pos+int64(footerBegLen))
		if err != nil {
			return nil, err
		}

		if n == len(data) &&
			bytes.Equal(StoreMagicEnd, data[n-lenMagicEnd*2:n-lenMagicEnd]) &&
			bytes.Equal(StoreMagicEnd, data[n-lenMagicEnd:]) {

			content := int(length) - footerBegLen - footerEndLen
			b := bytes.NewBuffer(data[content:])

			var offset int64
			if err = binary.Read(b, StoreEndian, &offset); err != nil {
				return nil, err
			}
			if offset != pos {
				return nil, fmt.Errorf("store: offset mismatch, "+
					"wanted: %v != found: %v", offset, pos)
			}

			var length1 uint32
			if err = binary.Read(b, StoreEndian, &length1); err != nil {
				return nil, err
			}
			if length1 != length {
				return nil, fmt.Errorf("store: length mismatch, "+
					"wanted: %v != found: %v", length1, length)
			}

			f := &Footer{refs: 1, fileName: fileName, filePos: offset}

			err = json.Unmarshal(data[:content], f)
			if err != nil {
				return nil, err
			}

			// json.Unmarshal would have just loaded the map.
			// We now need to load each segment into the map.
			// Also recursively load child footer segment stacks.
			err = f.loadSegments(options, fref)
			if err != nil {
				return nil, err
			}

			return f, nil
		}
		// Else, invalid footer - StoreMagicEnd missing and/or file
		// pos out of bounds.

		// Footer was invalid, so keep scanning.
		pos -= int64(StorePageSize)
	}
}

// --------------------------------------------------------

// loadSegments() loads the segments of a footer.  Adds new ref-counts
// to the fref on success.  The footer will be in an already closed
// state on error.
func (f *Footer) loadSegments(options *StoreOptions, fref *FileRef) (err error) {
	// Track mrefs that we need to DecRef() if there's an error.
	mrefs := make([]*mmapRef, 0, len(f.SegmentLocs))
	mrefs, err = f.doLoadSegments(options, fref, mrefs)
	if err != nil {
		for _, mref := range mrefs {
			mref.DecRef()
		}
		return err
	}

	return nil
}

func (f *Footer) doLoadSegments(options *StoreOptions, fref *FileRef,
	mrefs []*mmapRef) (mrefsSoFar []*mmapRef, err error) {
	// Recursively load the childFooters first.
	for _, childFooter := range f.ChildFooters {
		mrefs, err = childFooter.doLoadSegments(options, fref, mrefs)
		if err != nil {
			return mrefs, err
		}
	}

	if f.ss != nil && f.ss.a != nil {
		return mrefs, nil
	}

	osFile := ToOsFile(fref.file)
	if osFile == nil {
		return mrefs, fmt.Errorf("store: doLoadSegments convert to os.File error")
	}

	a := make([]Segment, len(f.SegmentLocs))

	for i := range f.SegmentLocs {
		sloc := &f.SegmentLocs[i]

		mref := sloc.mref
		if mref != nil {
			if mref.fref != fref {
				return mrefs, fmt.Errorf("store: doLoadSegments fref mismatch")
			}

			mref.AddRef()
			mrefs = append(mrefs, mref)
		} else {
			// We persist kvs before buf, so KvsOffset < BufOffset.
			begOffset := int64(sloc.KvsOffset)
			endOffset := int64(sloc.BufOffset + sloc.BufBytes)

			nbytes := int(endOffset - begOffset)

			// Some platforms (windows) only support mmap()'ing at an
			// allocation granularity that's != to a page size, so
			// calculate the actual offset/nbytes to use.
			begOffsetActual := pageOffset(begOffset, int64(AllocationGranularity))
			begOffsetDelta := int(begOffset - begOffsetActual)
			nbytesActual := nbytes + begOffsetDelta

			mm, err := mmap.MapRegion(osFile, nbytesActual, mmap.RDONLY, 0, begOffsetActual)
			if err != nil {
				return mrefs,
					fmt.Errorf("store: doLoadSegments mmap.Map(), "+
						"begOffsetActual = %v,"+
						"nbytesActual = %v, sloc = %v, err: %v",
						begOffsetActual, nbytesActual, sloc, err)
			}

			fref.AddRef() // New mref owns 1 fref ref-count.

			buf := mm[begOffsetDelta : begOffsetDelta+nbytes]

			sloc.mref = &mmapRef{fref: fref, mm: mm, buf: buf, refs: 1}

			mref = sloc.mref
			mrefs = append(mrefs, mref)

			segmentLoader, exists := SegmentLoaders[sloc.Kind]
			if !exists || segmentLoader == nil {
				return mrefs, fmt.Errorf("store: unknown SegmentLoc kind, sloc: %+v", sloc)
			}

			seg, err := segmentLoader(sloc)
			if err != nil {
				return mrefs, fmt.Errorf("store: segmentLoader failed, footer: %+v,"+
					" f.SegmentLocs: %+v, i: %d, options: %v err: %+v",
					f, f.SegmentLocs, i, options, err)
			}

			if options.SegmentKeysIndexMaxBytes > 0 {
				if a, ok := seg.(*segment); ok {
					a.buildIndex(options.SegmentKeysIndexMaxBytes,
						options.SegmentKeysIndexMinKeyBytes)
				}
			}

			mref.SetExt(seg)
		}

		a[i] = mref.GetExt().(Segment)
	}

	f.ss = &segmentStack{
		options: &options.CollectionOptions,
		a:       a,
		refs:    1,
	}

	return mrefs, nil
}

// --------------------------------------------------------

// ChildCollectionNames returns an array of child collection name strings.
func (f *Footer) ChildCollectionNames() ([]string, error) {
	var childNames = make([]string, len(f.ChildFooters))
	idx := 0
	for name := range f.ChildFooters {
		childNames[idx] = name
		idx++
	}
	return childNames, nil
}

// ChildCollectionSnapshot returns a Snapshot on a given child
// collection by its name.
func (f *Footer) ChildCollectionSnapshot(childCollectionName string) (
	Snapshot, error) {
	childFooter, exists := f.ChildFooters[childCollectionName]
	if !exists {
		return nil, nil
	}
	childFooter.AddRef()
	return childFooter, nil
}

// Close decrements the ref count on this footer
func (f *Footer) Close() error {
	f.DecRef()
	return nil
}

// AddRef increases the ref count on this footer
func (f *Footer) AddRef() {
	f.m.Lock()
	f.refs++
	f.m.Unlock()
}

// DecRef decreases the ref count on this footer
func (f *Footer) DecRef() {
	f.m.Lock()
	f.refs--
	if f.refs <= 0 {
		f.SegmentLocs.DecRef()
		f.SegmentLocs = nil
		f.ss = nil
	}
	f.m.Unlock()
}

// Length returns the length of this footer
func (f *Footer) Length() uint64 {
	jBuf, err := json.Marshal(f)
	if err != nil {
		return 0
	}

	footerLen := footerBegLen + len(jBuf) + footerEndLen
	return uint64(footerLen)
}

// --------------------------------------------------------

// segmentLocs returns the current SegmentLocs and segmentStack for
// a footer, while also incrementing the ref-count on the footer.  The
// caller must DecRef() the footer when done.
func (f *Footer) segmentLocs() (SegmentLocs, *segmentStack) {
	f.m.Lock()

	f.refs++

	slocs, ss := f.SegmentLocs, f.ss

	f.m.Unlock()

	return slocs, ss
}

// --------------------------------------------------------

// Get retrieves a val from the footer, and will return nil val
// if the entry does not exist in the footer.
func (f *Footer) Get(key []byte, readOptions ReadOptions) ([]byte, error) {
	_, ss := f.segmentLocs()
	if ss == nil {
		f.DecRef()
		return nil, nil
	}

	rv, err := ss.Get(key, readOptions)
	if err == nil && rv != nil && !readOptions.NoCopyValue {
		rv = append(make([]byte, 0, len(rv)), rv...) // Copy.
	}

	f.DecRef()

	return rv, err
}

// StartIterator returns a new Iterator instance on this footer.
//
// On success, the returned Iterator will be positioned so that
// Iterator.Current() will either provide the first entry in the
// range or ErrIteratorDone.
//
// A startKeyIncl of nil means the logical "bottom-most" possible key
// and an endKeyExcl of nil means the logical "top-most" possible key.
func (f *Footer) StartIterator(startKeyIncl, endKeyExcl []byte,
	iteratorOptions IteratorOptions) (Iterator, error) {
	_, ss := f.segmentLocs()
	if ss == nil {
		f.DecRef()
		return nil, nil
	}

	iter, err := ss.StartIterator(startKeyIncl, endKeyExcl, iteratorOptions)
	if err != nil || iter == nil {
		f.DecRef()
		return nil, err
	}

	initCloser, ok := iter.(InitCloser)
	if !ok || initCloser == nil {
		iter.Close()
		f.DecRef()
		return nil, ErrUnexpected
	}

	err = initCloser.InitCloser(f)
	if err != nil {
		iter.Close()
		f.DecRef()
		return nil, err
	}

	return iter, nil
}