mirror of
https://github.com/etcd-io/etcd.git
synced 2024-09-27 06:25:44 +00:00
bfe3a3d08e
On ReadAll, WAL seeks to the end of the last record in the tail. If the tail did not end with preallocated space, the decoder would report 0 as the last offset and begin writing at offset 0 of the tail. Fixes #4903
128 lines
2.8 KiB
Go
128 lines
2.8 KiB
Go
// Copyright 2015 CoreOS, 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 wal
|
|
|
|
import (
|
|
"bufio"
|
|
"encoding/binary"
|
|
"hash"
|
|
"io"
|
|
"sync"
|
|
|
|
"github.com/coreos/etcd/pkg/crc"
|
|
"github.com/coreos/etcd/pkg/pbutil"
|
|
"github.com/coreos/etcd/raft/raftpb"
|
|
"github.com/coreos/etcd/wal/walpb"
|
|
)
|
|
|
|
type decoder struct {
|
|
mu sync.Mutex
|
|
brs []*bufio.Reader
|
|
|
|
// lastValidOff file offset following the last valid decoded record
|
|
lastValidOff int64
|
|
crc hash.Hash32
|
|
}
|
|
|
|
func newDecoder(r ...io.Reader) *decoder {
|
|
readers := make([]*bufio.Reader, len(r))
|
|
for i := range r {
|
|
readers[i] = bufio.NewReader(r[i])
|
|
}
|
|
return &decoder{
|
|
brs: readers,
|
|
crc: crc.New(0, crcTable),
|
|
}
|
|
}
|
|
|
|
func (d *decoder) decode(rec *walpb.Record) error {
|
|
rec.Reset()
|
|
d.mu.Lock()
|
|
defer d.mu.Unlock()
|
|
return d.decodeRecord(rec)
|
|
}
|
|
|
|
func (d *decoder) decodeRecord(rec *walpb.Record) error {
|
|
if len(d.brs) == 0 {
|
|
return io.EOF
|
|
}
|
|
|
|
l, err := readInt64(d.brs[0])
|
|
if err == io.EOF || (err == nil && l == 0) {
|
|
// hit end of file or preallocated space
|
|
d.brs = d.brs[1:]
|
|
if len(d.brs) == 0 {
|
|
return io.EOF
|
|
}
|
|
d.lastValidOff = 0
|
|
return d.decodeRecord(rec)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
data := make([]byte, l)
|
|
if _, err = io.ReadFull(d.brs[0], data); err != nil {
|
|
// ReadFull returns io.EOF only if no bytes were read
|
|
// the decoder should treat this as an ErrUnexpectedEOF instead.
|
|
if err == io.EOF {
|
|
err = io.ErrUnexpectedEOF
|
|
}
|
|
return err
|
|
}
|
|
if err := rec.Unmarshal(data); err != nil {
|
|
return err
|
|
}
|
|
|
|
// skip crc checking if the record type is crcType
|
|
if rec.Type != crcType {
|
|
d.crc.Write(rec.Data)
|
|
if err := rec.Validate(d.crc.Sum32()); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
// record decoded as valid; point last valid offset to end of record
|
|
d.lastValidOff += l + 8
|
|
return nil
|
|
}
|
|
|
|
func (d *decoder) updateCRC(prevCrc uint32) {
|
|
d.crc = crc.New(prevCrc, crcTable)
|
|
}
|
|
|
|
func (d *decoder) lastCRC() uint32 {
|
|
return d.crc.Sum32()
|
|
}
|
|
|
|
func (d *decoder) lastOffset() int64 { return d.lastValidOff }
|
|
|
|
func mustUnmarshalEntry(d []byte) raftpb.Entry {
|
|
var e raftpb.Entry
|
|
pbutil.MustUnmarshal(&e, d)
|
|
return e
|
|
}
|
|
|
|
func mustUnmarshalState(d []byte) raftpb.HardState {
|
|
var s raftpb.HardState
|
|
pbutil.MustUnmarshal(&s, d)
|
|
return s
|
|
}
|
|
|
|
func readInt64(r io.Reader) (int64, error) {
|
|
var n int64
|
|
err := binary.Read(r, binary.LittleEndian, &n)
|
|
return n, err
|
|
}
|