value.go
func (vlog *valueLog) iterate(lf *logFile, offset uint32, fn logEntry) (uint32, error) {
	fi, err := lf.fd.Stat()
	if err != nil {
		return 0, err
	}
	if int64(offset) == fi.Size() {
		// We're at the end of the file already. No need to do anything.
		return offset, nil
	}
	if vlog.opt.ReadOnly {
		// We're not at the end of the file. We'd need to replay the entries, or
		// possibly truncate the file.
		return 0, ErrReplayNeeded
	}

	// We're not at the end of the file. Let's Seek to the offset and start reading.
	if _, err := lf.fd.Seek(int64(offset), io.SeekStart); err != nil {
		return 0, errFile(err, lf.path, "Unable to seek")
	}

	reader := bufio.NewReader(lf.fd)
	read := &safeRead{
		k:            make([]byte, 10),
		v:            make([]byte, 10),
		recordOffset: offset,
	}

	var lastCommit uint64
	var validEndOffset uint32
	for {
		e, err := read.Entry(reader)
		if err == io.EOF {
			break
		} else if err == io.ErrUnexpectedEOF || err == errTruncate {
			break
		} else if err != nil {
			return 0, err
		} else if e == nil {
			continue
		}

		var vp valuePointer
		vp.Len = uint32(headerBufSize + len(e.Key) + len(e.Value) + crc32.Size)
		read.recordOffset += vp.Len

		vp.Offset = e.offset
		vp.Fid = lf.fid

		if e.meta&bitTxn > 0 {
			txnTs := y.ParseTs(e.Key)
			if lastCommit == 0 {
				lastCommit = txnTs
			}
			if lastCommit != txnTs {
				break
			}

		} else if e.meta&bitFinTxn > 0 {
			txnTs, err := strconv.ParseUint(string(e.Value), 10, 64)
			if err != nil || lastCommit != txnTs {
				break
			}
			// Got the end of txn. Now we can store them.
			lastCommit = 0
			validEndOffset = read.recordOffset

		} else {
			if lastCommit != 0 {
				// This is most likely an entry which was moved as part of GC.
				// We shouldn't get this entry in the middle of a transaction.
				break
			}
			validEndOffset = read.recordOffset
		}

		if err := fn(*e, vp); err != nil {
			if err == errStop {
				break
			}
			return 0, errFile(err, lf.path, "Iteration function")
		}
	}
	return validEndOffset, nil
}
func (vlog *valueLog) iterate(lf *logFile, offset uint32, fn logEntry) (uint32, error) {
	fi, err := lf.fd.Stat()
	if err != nil {
		return 0, err
	}
	if int64(offset) == fi.Size() {
		// We're at the end of the file already. No need to do anything.
		return offset, nil
	}
	if vlog.opt.ReadOnly {
		// We're not at the end of the file. We'd need to replay the entries, or
		// possibly truncate the file.
		return 0, ErrReplayNeeded
	}

	// We're not at the end of the file. Let's Seek to the offset and start reading.
	if _, err := lf.fd.Seek(int64(offset), io.SeekStart); err != nil {
		return 0, errFile(err, lf.path, "Unable to seek")
	}

	reader := bufio.NewReader(lf.fd)
	read := &safeRead{
		k:            make([]byte, 10),
		v:            make([]byte, 10),
		recordOffset: offset,
	}

	var lastCommit uint64
	var validEndOffset uint32
	for {
		e, err := read.Entry(reader)
		if err == io.EOF {
			break
		} else if err == io.ErrUnexpectedEOF || err == errTruncate {
			break
		} else if err != nil {
			return 0, err
		} else if e == nil {
			continue
		}

		var vp valuePointer
		vp.Len = uint32(headerBufSize + len(e.Key) + len(e.Value) + crc32.Size)
		read.recordOffset += vp.Len

		vp.Offset = e.offset
		vp.Fid = lf.fid

		if e.meta&bitTxn > 0 {
			txnTs := y.ParseTs(e.Key)
			if lastCommit == 0 {
				lastCommit = txnTs
			}
			if lastCommit != txnTs {
				break
			}

		} else if e.meta&bitFinTxn > 0 {
			txnTs, err := strconv.ParseUint(string(e.Value), 10, 64)
			if err != nil || lastCommit != txnTs {
				break
			}
			// Got the end of txn. Now we can store them.
			lastCommit = 0
			validEndOffset = read.recordOffset

		} else {
			if lastCommit != 0 {
				// This is most likely an entry which was moved as part of GC.
				// We shouldn't get this entry in the middle of a transaction.
				break
			}
			validEndOffset = read.recordOffset
		}

		if err := fn(*e, vp); err != nil {
			if err == errStop {
				break
			}
			return 0, errFile(err, lf.path, "Iteration function")
		}
	}
	return validEndOffset, nil
}
func ( vlog * valueLog ) iterate ( lf * logFile , offset uint32 , fn logEntry ) ( uint32 , error ) { fi , err := lf . fd . Stat ( ) if err != nil { return 0 , err } if int64 ( offset ) == fi . Size ( ) { // We're at the end of the file already. No need to do anything. return offset , nil } if vlog . opt . ReadOnly { // We're not at the end of the file. We'd need to replay the entries, or // possibly truncate the file. return 0 , ErrReplayNeeded } // We're not at the end of the file. Let's Seek to the offset and start reading. if _ , err := lf . fd . Seek ( int64 ( offset ) , io . SeekStart ) ; err != nil { return 0 , errFile ( err , lf . path , " " ) } reader := bufio . NewReader ( lf . fd ) read := & safeRead { k : make ( [ ] byte , 10 ) , v : make ( [ ] byte , 10 ) , recordOffset : offset , } var lastCommit uint64 var validEndOffset uint32 for { e , err := read . Entry ( reader ) if err == io . EOF { break } else if err == io . ErrUnexpectedEOF || err == errTruncate { break } else if err != nil { return 0 , err } else if e == nil { continue } var vp valuePointer vp . Len = uint32 ( headerBufSize + len ( e . Key ) + len ( e . Value ) + crc32 . Size ) read . recordOffset += vp . Len vp . Offset = e . offset vp . Fid = lf . fid if e . meta & bitTxn > 0 { txnTs := y . ParseTs ( e . Key ) if lastCommit == 0 { lastCommit = txnTs } if lastCommit != txnTs { break } } else if e . meta & bitFinTxn > 0 { txnTs , err := strconv . ParseUint ( string ( e . Value ) , 10 , 64 ) if err != nil || lastCommit != txnTs { break } // Got the end of txn. Now we can store them. lastCommit = 0 validEndOffset = read . recordOffset } else { if lastCommit != 0 { // This is most likely an entry which was moved as part of GC. // We shouldn't get this entry in the middle of a transaction. break } validEndOffset = read . recordOffset } if err := fn ( * e , vp ) ; err != nil { if err == errStop { break } return 0 , errFile ( err , lf . path , " " ) } } return validEndOffset , nil }
// iterate iterates over log file. It doesn't not allocate new memory for every kv pair.
// Therefore, the kv pair is only valid for the duration of fn call.
iterate iterates over log file . It doesn t not allocate new memory for every kv pair . Therefore the kv pair is only valid for the duration of fn call .