value.go
func (vlog *valueLog) write(reqs []*request) error {
	vlog.filesLock.RLock()
	maxFid := atomic.LoadUint32(&vlog.maxFid)
	curlf := vlog.filesMap[maxFid]
	vlog.filesLock.RUnlock()

	var buf bytes.Buffer
	toDisk := func() error {
		if buf.Len() == 0 {
			return nil
		}
		vlog.elog.Printf("Flushing %d blocks of total size: %d", len(reqs), buf.Len())
		n, err := curlf.fd.Write(buf.Bytes())
		if err != nil {
			return errors.Wrapf(err, "Unable to write to value log file: %q", curlf.path)
		}
		buf.Reset()
		y.NumWrites.Add(1)
		y.NumBytesWritten.Add(int64(n))
		vlog.elog.Printf("Done")
		atomic.AddUint32(&vlog.writableLogOffset, uint32(n))

		if vlog.woffset() > uint32(vlog.opt.ValueLogFileSize) ||
			vlog.numEntriesWritten > vlog.opt.ValueLogMaxEntries {
			var err error
			if err = curlf.doneWriting(vlog.woffset()); err != nil {
				return err
			}

			newid := atomic.AddUint32(&vlog.maxFid, 1)
			y.AssertTruef(newid > 0, "newid has overflown uint32: %v", newid)
			newlf, err := vlog.createVlogFile(newid)
			if err != nil {
				return err
			}
			curlf = newlf
		}
		return nil
	}

	for i := range reqs {
		b := reqs[i]
		b.Ptrs = b.Ptrs[:0]
		for j := range b.Entries {
			e := b.Entries[j]
			var p valuePointer

			p.Fid = curlf.fid
			// Use the offset including buffer length so far.
			p.Offset = vlog.woffset() + uint32(buf.Len())
			plen, err := encodeEntry(e, &buf) // Now encode the entry into buffer.
			if err != nil {
				return err
			}
			p.Len = uint32(plen)
			b.Ptrs = append(b.Ptrs, p)
		}
		vlog.numEntriesWritten += uint32(len(b.Entries))
		// We write to disk here so that all entries that are part of the same transaction are
		// written to the same vlog file.
		writeNow :=
			vlog.woffset()+uint32(buf.Len()) > uint32(vlog.opt.ValueLogFileSize) ||
				vlog.numEntriesWritten > uint32(vlog.opt.ValueLogMaxEntries)
		if writeNow {
			if err := toDisk(); err != nil {
				return err
			}
		}
	}
	return toDisk()
}
func (vlog *valueLog) write(reqs []*request) error {
	vlog.filesLock.RLock()
	maxFid := atomic.LoadUint32(&vlog.maxFid)
	curlf := vlog.filesMap[maxFid]
	vlog.filesLock.RUnlock()

	var buf bytes.Buffer
	toDisk := func() error {
		if buf.Len() == 0 {
			return nil
		}
		vlog.elog.Printf("Flushing %d blocks of total size: %d", len(reqs), buf.Len())
		n, err := curlf.fd.Write(buf.Bytes())
		if err != nil {
			return errors.Wrapf(err, "Unable to write to value log file: %q", curlf.path)
		}
		buf.Reset()
		y.NumWrites.Add(1)
		y.NumBytesWritten.Add(int64(n))
		vlog.elog.Printf("Done")
		atomic.AddUint32(&vlog.writableLogOffset, uint32(n))

		if vlog.woffset() > uint32(vlog.opt.ValueLogFileSize) ||
			vlog.numEntriesWritten > vlog.opt.ValueLogMaxEntries {
			var err error
			if err = curlf.doneWriting(vlog.woffset()); err != nil {
				return err
			}

			newid := atomic.AddUint32(&vlog.maxFid, 1)
			y.AssertTruef(newid > 0, "newid has overflown uint32: %v", newid)
			newlf, err := vlog.createVlogFile(newid)
			if err != nil {
				return err
			}
			curlf = newlf
		}
		return nil
	}

	for i := range reqs {
		b := reqs[i]
		b.Ptrs = b.Ptrs[:0]
		for j := range b.Entries {
			e := b.Entries[j]
			var p valuePointer

			p.Fid = curlf.fid
			// Use the offset including buffer length so far.
			p.Offset = vlog.woffset() + uint32(buf.Len())
			plen, err := encodeEntry(e, &buf) // Now encode the entry into buffer.
			if err != nil {
				return err
			}
			p.Len = uint32(plen)
			b.Ptrs = append(b.Ptrs, p)
		}
		vlog.numEntriesWritten += uint32(len(b.Entries))
		// We write to disk here so that all entries that are part of the same transaction are
		// written to the same vlog file.
		writeNow :=
			vlog.woffset()+uint32(buf.Len()) > uint32(vlog.opt.ValueLogFileSize) ||
				vlog.numEntriesWritten > uint32(vlog.opt.ValueLogMaxEntries)
		if writeNow {
			if err := toDisk(); err != nil {
				return err
			}
		}
	}
	return toDisk()
}
func ( vlog * valueLog ) write ( reqs [ ] * request ) error { vlog . filesLock . RLock ( ) maxFid := atomic . LoadUint32 ( & vlog . maxFid ) curlf := vlog . filesMap [ maxFid ] vlog . filesLock . RUnlock ( ) var buf bytes . Buffer toDisk := func ( ) error { if buf . Len ( ) == 0 { return nil } vlog . elog . Printf ( " " , len ( reqs ) , buf . Len ( ) ) n , err := curlf . fd . Write ( buf . Bytes ( ) ) if err != nil { return errors . Wrapf ( err , " " , curlf . path ) } buf . Reset ( ) y . NumWrites . Add ( 1 ) y . NumBytesWritten . Add ( int64 ( n ) ) vlog . elog . Printf ( " " ) atomic . AddUint32 ( & vlog . writableLogOffset , uint32 ( n ) ) if vlog . woffset ( ) > uint32 ( vlog . opt . ValueLogFileSize ) || vlog . numEntriesWritten > vlog . opt . ValueLogMaxEntries { var err error if err = curlf . doneWriting ( vlog . woffset ( ) ) ; err != nil { return err } newid := atomic . AddUint32 ( & vlog . maxFid , 1 ) y . AssertTruef ( newid > 0 , " " , newid ) newlf , err := vlog . createVlogFile ( newid ) if err != nil { return err } curlf = newlf } return nil } for i := range reqs { b := reqs [ i ] b . Ptrs = b . Ptrs [ : 0 ] for j := range b . Entries { e := b . Entries [ j ] var p valuePointer p . Fid = curlf . fid // Use the offset including buffer length so far. p . Offset = vlog . woffset ( ) + uint32 ( buf . Len ( ) ) plen , err := encodeEntry ( e , & buf ) // Now encode the entry into buffer. if err != nil { return err } p . Len = uint32 ( plen ) b . Ptrs = append ( b . Ptrs , p ) } vlog . numEntriesWritten += uint32 ( len ( b . Entries ) ) // We write to disk here so that all entries that are part of the same transaction are // written to the same vlog file. writeNow := vlog . woffset ( ) + uint32 ( buf . Len ( ) ) > uint32 ( vlog . opt . ValueLogFileSize ) || vlog . numEntriesWritten > uint32 ( vlog . opt . ValueLogMaxEntries ) if writeNow { if err := toDisk ( ) ; err != nil { return err } } } return toDisk ( ) }
// write is thread-unsafe by design and should not be called concurrently.
write is thread - unsafe by design and should not be called concurrently .