Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

wal: do not race reader and writer

Browse Source
This commit is contained in:
Xiang Li 2015-03-05 20:19:17 -08:00
parent 6b9b695167
commit ab72c3ec88
2 changed files with 16 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
wal/decoder.go
View File

@ -19,6 +19,7 @@ import (
"encoding/binary" "encoding/binary"
"hash" "hash"
"io" "io"
"sync"
"github.com/coreos/etcd/pkg/crc" "github.com/coreos/etcd/pkg/crc"
"github.com/coreos/etcd/pkg/pbutil" "github.com/coreos/etcd/pkg/pbutil"
@ -27,7 +28,9 @@ import (
) )
type decoder struct { type decoder struct {
mu sync.Mutex
br *bufio.Reader br *bufio.Reader
c io.Closer c io.Closer
crc hash.Hash32 crc hash.Hash32
} }
@ -41,6 +44,9 @@ func newDecoder(rc io.ReadCloser) *decoder {
} }
func (d *decoder) decode(rec *walpb.Record) error { func (d *decoder) decode(rec *walpb.Record) error {
d.mu.Lock()
defer d.mu.Unlock()
rec.Reset() rec.Reset()
l, err := readInt64(d.br) l, err := readInt64(d.br)
if err != nil { if err != nil {

8
wal/encoder.go
View File

@ -19,13 +19,16 @@ import (
"encoding/binary" "encoding/binary"
"hash" "hash"
"io" "io"
"sync"
"github.com/coreos/etcd/pkg/crc" "github.com/coreos/etcd/pkg/crc"
"github.com/coreos/etcd/wal/walpb" "github.com/coreos/etcd/wal/walpb"
) )
type encoder struct { type encoder struct {
mu sync.Mutex
bw *bufio.Writer bw *bufio.Writer
crc hash.Hash32 crc hash.Hash32
} }
@ -37,6 +40,9 @@ func newEncoder(w io.Writer, prevCrc uint32) *encoder {
} }
func (e *encoder) encode(rec *walpb.Record) error { func (e *encoder) encode(rec *walpb.Record) error {
e.mu.Lock()
defer e.mu.Unlock()
e.crc.Write(rec.Data) e.crc.Write(rec.Data)
rec.Crc = e.crc.Sum32() rec.Crc = e.crc.Sum32()
data, err := rec.Marshal() data, err := rec.Marshal()
@ -51,6 +57,8 @@ func (e *encoder) encode(rec *walpb.Record) error {
} }
func (e *encoder) flush() error { func (e *encoder) flush() error {
e.mu.Lock()
defer e.mu.Unlock()
return e.bw.Flush() return e.bw.Flush()
} }