wal/wal.go
func (w *WAL) ReadAll() (metadata []byte, state raftpb.HardState, ents []raftpb.Entry, err error) {
	w.mu.Lock()
	defer w.mu.Unlock()

	rec := &walpb.Record{}
	decoder := w.decoder

	var match bool
	for err = decoder.decode(rec); err == nil; err = decoder.decode(rec) {
		switch rec.Type {
		case entryType:
			e := mustUnmarshalEntry(rec.Data)
			if e.Index > w.start.Index {
				ents = append(ents[:e.Index-w.start.Index-1], e)
			}
			w.enti = e.Index

		case stateType:
			state = mustUnmarshalState(rec.Data)

		case metadataType:
			if metadata != nil && !bytes.Equal(metadata, rec.Data) {
				state.Reset()
				return nil, state, nil, ErrMetadataConflict
			}
			metadata = rec.Data

		case crcType:
			crc := decoder.crc.Sum32()
			// current crc of decoder must match the crc of the record.
			// do no need to match 0 crc, since the decoder is a new one at this case.
			if crc != 0 && rec.Validate(crc) != nil {
				state.Reset()
				return nil, state, nil, ErrCRCMismatch
			}
			decoder.updateCRC(rec.Crc)

		case snapshotType:
			var snap walpb.Snapshot
			pbutil.MustUnmarshal(&snap, rec.Data)
			if snap.Index == w.start.Index {
				if snap.Term != w.start.Term {
					state.Reset()
					return nil, state, nil, ErrSnapshotMismatch
				}
				match = true
			}

		default:
			state.Reset()
			return nil, state, nil, fmt.Errorf("unexpected block type %d", rec.Type)
		}
	}

	switch w.tail() {
	case nil:
		// We do not have to read out all entries in read mode.
		// The last record maybe a partial written one, so
		// ErrunexpectedEOF might be returned.
		if err != io.EOF && err != io.ErrUnexpectedEOF {
			state.Reset()
			return nil, state, nil, err
		}
	default:
		// We must read all of the entries if WAL is opened in write mode.
		if err != io.EOF {
			state.Reset()
			return nil, state, nil, err
		}
		// decodeRecord() will return io.EOF if it detects a zero record,
		// but this zero record may be followed by non-zero records from
		// a torn write. Overwriting some of these non-zero records, but
		// not all, will cause CRC errors on WAL open. Since the records
		// were never fully synced to disk in the first place, it's safe
		// to zero them out to avoid any CRC errors from new writes.
		if _, err = w.tail().Seek(w.decoder.lastOffset(), io.SeekStart); err != nil {
			return nil, state, nil, err
		}
		if err = fileutil.ZeroToEnd(w.tail().File); err != nil {
			return nil, state, nil, err
		}
	}

	err = nil
	if !match {
		err = ErrSnapshotNotFound
	}

	// close decoder, disable reading
	if w.readClose != nil {
		w.readClose()
		w.readClose = nil
	}
	w.start = walpb.Snapshot{}

	w.metadata = metadata

	if w.tail() != nil {
		// create encoder (chain crc with the decoder), enable appending
		w.encoder, err = newFileEncoder(w.tail().File, w.decoder.lastCRC())
		if err != nil {
			return
		}
	}
	w.decoder = nil

	return metadata, state, ents, err
}
func (w *WAL) ReadAll() (metadata []byte, state raftpb.HardState, ents []raftpb.Entry, err error) {
	w.mu.Lock()
	defer w.mu.Unlock()

	rec := &walpb.Record{}
	decoder := w.decoder

	var match bool
	for err = decoder.decode(rec); err == nil; err = decoder.decode(rec) {
		switch rec.Type {
		case entryType:
			e := mustUnmarshalEntry(rec.Data)
			if e.Index > w.start.Index {
				ents = append(ents[:e.Index-w.start.Index-1], e)
			}
			w.enti = e.Index

		case stateType:
			state = mustUnmarshalState(rec.Data)

		case metadataType:
			if metadata != nil && !bytes.Equal(metadata, rec.Data) {
				state.Reset()
				return nil, state, nil, ErrMetadataConflict
			}
			metadata = rec.Data

		case crcType:
			crc := decoder.crc.Sum32()
			// current crc of decoder must match the crc of the record.
			// do no need to match 0 crc, since the decoder is a new one at this case.
			if crc != 0 && rec.Validate(crc) != nil {
				state.Reset()
				return nil, state, nil, ErrCRCMismatch
			}
			decoder.updateCRC(rec.Crc)

		case snapshotType:
			var snap walpb.Snapshot
			pbutil.MustUnmarshal(&snap, rec.Data)
			if snap.Index == w.start.Index {
				if snap.Term != w.start.Term {
					state.Reset()
					return nil, state, nil, ErrSnapshotMismatch
				}
				match = true
			}

		default:
			state.Reset()
			return nil, state, nil, fmt.Errorf("unexpected block type %d", rec.Type)
		}
	}

	switch w.tail() {
	case nil:
		// We do not have to read out all entries in read mode.
		// The last record maybe a partial written one, so
		// ErrunexpectedEOF might be returned.
		if err != io.EOF && err != io.ErrUnexpectedEOF {
			state.Reset()
			return nil, state, nil, err
		}
	default:
		// We must read all of the entries if WAL is opened in write mode.
		if err != io.EOF {
			state.Reset()
			return nil, state, nil, err
		}
		// decodeRecord() will return io.EOF if it detects a zero record,
		// but this zero record may be followed by non-zero records from
		// a torn write. Overwriting some of these non-zero records, but
		// not all, will cause CRC errors on WAL open. Since the records
		// were never fully synced to disk in the first place, it's safe
		// to zero them out to avoid any CRC errors from new writes.
		if _, err = w.tail().Seek(w.decoder.lastOffset(), io.SeekStart); err != nil {
			return nil, state, nil, err
		}
		if err = fileutil.ZeroToEnd(w.tail().File); err != nil {
			return nil, state, nil, err
		}
	}

	err = nil
	if !match {
		err = ErrSnapshotNotFound
	}

	// close decoder, disable reading
	if w.readClose != nil {
		w.readClose()
		w.readClose = nil
	}
	w.start = walpb.Snapshot{}

	w.metadata = metadata

	if w.tail() != nil {
		// create encoder (chain crc with the decoder), enable appending
		w.encoder, err = newFileEncoder(w.tail().File, w.decoder.lastCRC())
		if err != nil {
			return
		}
	}
	w.decoder = nil

	return metadata, state, ents, err
}
func ( w * WAL ) ReadAll ( ) ( metadata [ ] byte , state raftpb . HardState , ents [ ] raftpb . Entry , err error ) { w . mu . Lock ( ) defer w . mu . Unlock ( ) rec := & walpb . Record { } decoder := w . decoder var match bool for err = decoder . decode ( rec ) ; err == nil ; err = decoder . decode ( rec ) { switch rec . Type { case entryType : e := mustUnmarshalEntry ( rec . Data ) if e . Index > w . start . Index { ents = append ( ents [ : e . Index - w . start . Index - 1 ] , e ) } w . enti = e . Index case stateType : state = mustUnmarshalState ( rec . Data ) case metadataType : if metadata != nil && ! bytes . Equal ( metadata , rec . Data ) { state . Reset ( ) return nil , state , nil , ErrMetadataConflict } metadata = rec . Data case crcType : crc := decoder . crc . Sum32 ( ) // current crc of decoder must match the crc of the record. // do no need to match 0 crc, since the decoder is a new one at this case. if crc != 0 && rec . Validate ( crc ) != nil { state . Reset ( ) return nil , state , nil , ErrCRCMismatch } decoder . updateCRC ( rec . Crc ) case snapshotType : var snap walpb . Snapshot pbutil . MustUnmarshal ( & snap , rec . Data ) if snap . Index == w . start . Index { if snap . Term != w . start . Term { state . Reset ( ) return nil , state , nil , ErrSnapshotMismatch } match = true } default : state . Reset ( ) return nil , state , nil , fmt . Errorf ( " " , rec . Type ) } } switch w . tail ( ) { case nil : // We do not have to read out all entries in read mode. // The last record maybe a partial written one, so // ErrunexpectedEOF might be returned. if err != io . EOF && err != io . ErrUnexpectedEOF { state . Reset ( ) return nil , state , nil , err } default : // We must read all of the entries if WAL is opened in write mode. if err != io . EOF { state . Reset ( ) return nil , state , nil , err } // decodeRecord() will return io.EOF if it detects a zero record, // but this zero record may be followed by non-zero records from // a torn write. Overwriting some of these non-zero records, but // not all, will cause CRC errors on WAL open. Since the records // were never fully synced to disk in the first place, it's safe // to zero them out to avoid any CRC errors from new writes. if _ , err = w . tail ( ) . Seek ( w . decoder . lastOffset ( ) , io . SeekStart ) ; err != nil { return nil , state , nil , err } if err = fileutil . ZeroToEnd ( w . tail ( ) . File ) ; err != nil { return nil , state , nil , err } } err = nil if ! match { err = ErrSnapshotNotFound } // close decoder, disable reading if w . readClose != nil { w . readClose ( ) w . readClose = nil } w . start = walpb . Snapshot { } w . metadata = metadata if w . tail ( ) != nil { // create encoder (chain crc with the decoder), enable appending w . encoder , err = newFileEncoder ( w . tail ( ) . File , w . decoder . lastCRC ( ) ) if err != nil { return } } w . decoder = nil return metadata , state , ents , err }
// ReadAll reads out records of the current WAL.
// If opened in write mode, it must read out all records until EOF. Or an error
// will be returned.
// If opened in read mode, it will try to read all records if possible.
// If it cannot read out the expected snap, it will return ErrSnapshotNotFound.
// If loaded snap doesn't match with the expected one, it will return
// all the records and error ErrSnapshotMismatch.
// TODO: detect not-last-snap error.
// TODO: maybe loose the checking of match.
// After ReadAll, the WAL will be ready for appending new records.
ReadAll reads out records of the current WAL . If opened in write mode it must read out all records until EOF . Or an error will be returned . If opened in read mode it will try to read all records if possible . If it cannot read out the expected snap it will return ErrSnapshotNotFound . If loaded snap doesn t match with the expected one it will return all the records and error ErrSnapshotMismatch . TODO : detect not - last - snap error . TODO : maybe loose the checking of match . After ReadAll the WAL will be ready for appending new records .