func (w *WAL) CloseIncomplete() (int64, error) {
	w.wg.Wait()
	return atomic.LoadInt64(&w.atomicUnfinishedTxns), w.logFile.Close()
}
CloseIncomplete closes the WAL and reports the number of transactions that
are still uncomitted.
package writeaheadlog
import (
	"bytes"
	"encoding/binary"
	"fmt"
	"os"
	"sort"
	"sync"
	"sync/atomic"
	"unsafe"
	"github.com/NebulousLabs/errors"
)
type WAL struct {
	atomicNextTxnNum uint64
	atomicUnfinishedTxns int64
	atomicSyncStatus uint32          
	atomicSyncState  unsafe.Pointer  
	availablePages []uint64
	filePageCount uint64
	wg sync.WaitGroup
	deps    dependencies
	logFile file
	mu      sync.Mutex
	path    string  
}
func (w *WAL) allocatePages(numPages uint64) {
	start := w.filePageCount + 1
	for i := start; i < start+numPages; i++ {
		w.availablePages = append(w.availablePages, uint64(i)*pageSize)
	}
	w.filePageCount += numPages
}
func newWal(path string, deps dependencies) (txns []*Transaction, w *WAL, err error) {
	newWal := &WAL{
		deps: deps,
		path: path,
	}
	ss := new(syncState)
	ss.rwMu.Lock()
	atomic.StorePointer(&newWal.atomicSyncState, unsafe.Pointer(ss))
	data, err := deps.readFile(path)
	if err == nil {
		newWal.logFile, err = deps.openFile(path, os.O_RDWR, 0600)
		if err != nil {
			return nil, nil, errors.Extend(errors.New("unable to open wal logFile"), err)
		}
		txns, err = newWal.recoverWAL(data)
		if err != nil {
			err = errors.Compose(err, newWal.logFile.Close())
			return nil, nil, errors.Extend(err, errors.New("unable to perform wal recovery"))
		}
		return txns, newWal, nil
	} else if !os.IsNotExist(err) {
		return nil, nil, errors.Extend(err, errors.New("walFile was not opened successfully"))
	}
	newWal.logFile, err = deps.create(path)
	if err != nil {
		return nil, nil, errors.Extend(err, errors.New("walFile could not be created"))
	}
	if err = writeWALMetadata(newWal.logFile); err != nil {
		return nil, nil, errors.Extend(err, errors.New("Failed to write metadata to file"))
	}
	return nil, newWal, nil
}
func readWALMetadata(data []byte) (uint16, error) {
	if len(data) < len(metadataHeader)+len(metadataVersion)+metadataStatusSize {
		return 0, errors.New("unable to read wal metadata")
	}
	if !bytes.Equal(data[:len(metadataHeader)], metadataHeader[:]) {
		return 0, errors.New("file header is incorrect")
	}
	if !bytes.Equal(data[len(metadataHeader):len(metadataHeader)+len(metadataVersion)], metadataVersion[:]) {
		return 0, errors.New("file version is unrecognized - maybe you need to upgrade")
	}
	fileState := uint16(data[len(metadataHeader)+len(metadataVersion)])
	if fileState <= 0 || fileState > 3 {
		fileState = recoveryStateUnclean
	}
	return fileState, nil
}
func (w *WAL) recoverWAL(data []byte) ([]*Transaction, error) {
	recoveryState, err := readWALMetadata(data[0:])
	if err != nil {
		return nil, errors.Extend(err, errors.New("unable to read wal metadata"))
	}
	if recoveryState == recoveryStateClean {
		if err := w.writeRecoveryState(recoveryStateUnclean); err != nil {
			return nil, errors.Extend(err, errors.New("unable to write WAL recovery state"))
		}
		return nil, nil
	}
	type diskPage struct {
		page
		nextPageOffset uint64
	}
	pageSet := make(map[uint64]*diskPage)  
	for i := uint64(pageSize); i+pageSize <= uint64(len(data)); i += pageSize {
		nextOffset := binary.LittleEndian.Uint64(data[i:])
		if nextOffset < pageSize {
			continue
		}
		pageSet[i] = &diskPage{
			page: page{
				offset:  i,
				payload: data[i+pageMetaSize : i+pageSize],
			},
			nextPageOffset: nextOffset,
		}
	}
	for _, p := range pageSet {
		if nextDiskPage, ok := pageSet[p.nextPageOffset]; ok {
			p.nextPage = &nextDiskPage.page
		}
	}
	var txns []*Transaction
nextTxn:
	for i := pageSize; i+pageSize <= len(data); i += pageSize {
		status := binary.LittleEndian.Uint64(data[i:])
		if status != txnStatusComitted {
			continue
		}
		seq := binary.LittleEndian.Uint64(data[i+8:])
		var diskChecksum checksum
		n := copy(diskChecksum[:], data[i+16:])
		nextPageOffset := binary.LittleEndian.Uint64(data[i+16+n:])
		firstPage := &page{
			offset:  uint64(i),
			payload: data[i+firstPageMetaSize : i+pageSize],
		}
		if nextDiskPage, ok := pageSet[nextPageOffset]; ok {
			firstPage.nextPage = &nextDiskPage.page
		}
		visited := make(map[uint64]struct{})
		for page := firstPage; page != nil; page = page.nextPage {
			if _, exists := visited[page.offset]; exists {
				continue nextTxn
			}
			visited[page.offset] = struct{}{}
		}
		txn := &Transaction{
			status:         status,
			setupComplete:  true,
			commitComplete: true,
			sequenceNumber: seq,
			firstPage:      firstPage,
			wal:            w,
		}
		if txn.checksum() != diskChecksum {
			continue
		}
		var updateBytes []byte
		for page := txn.firstPage; page != nil; page = page.nextPage {
			updateBytes = append(updateBytes, page.payload...)
		}
		updates, err := unmarshalUpdates(updateBytes)
		if err != nil {
			continue
		}
		txn.Updates = updates
		txns = append(txns, txn)
	}
	sort.Slice(txns, func(i, j int) bool {
		return txns[i].sequenceNumber < txns[j].sequenceNumber
	})
	w.filePageCount = uint64(len(data)) / pageSize
	if len(data)%pageSize != 0 {
		w.filePageCount++
	}
	if w.filePageCount > 0 {
		w.filePageCount--
	}
	usedPages := make(map[uint64]struct{})
	for _, txn := range txns {
		for page := txn.firstPage; page != nil; page = page.nextPage {
			usedPages[page.offset] = struct{}{}
		}
	}
	for offset := uint64(pageSize); offset < w.filePageCount*pageSize; offset += pageSize {
		if _, exists := usedPages[offset]; !exists {
			w.availablePages = append(w.availablePages, offset)
		}
	}
	w.atomicUnfinishedTxns = int64(len(txns))
	return txns, nil
}
func (w *WAL) writeRecoveryState(state uint16) error {
	_, err := w.logFile.WriteAt([]byte{byte(state)}, int64(len(metadataHeader)+len(metadataVersion)))
	if err != nil {
		return err
	}
	return w.logFile.Sync()
}
func (w *WAL) managedReservePages(data []byte) *page {
	numPages := uint64(len(data) / MaxPayloadSize)
	if len(data)%MaxPayloadSize != 0 {
		numPages++
	}
	w.mu.Lock()
	defer w.mu.Unlock()
	if pagesNeeded := int64(numPages) - int64(len(w.availablePages)); pagesNeeded > 0 {
		w.allocatePages(uint64(pagesNeeded))
		if int64(len(w.availablePages)) != int64(numPages) {
			panic(fmt.Errorf("sanity check failed: num of available pages (%v) != num of required pages (%v)", len(w.availablePages), numPages))
		}
	}
	reservedPages := w.availablePages[uint64(len(w.availablePages))-numPages:]
	w.availablePages = w.availablePages[:uint64(len(w.availablePages))-numPages]
	buf := bytes.NewBuffer(data)
	pages := make([]page, numPages)
	for i := range pages {
		if uint64(i+1) < numPages {
			pages[i].nextPage = &pages[i+1]
		}
		pages[i].offset = reservedPages[i]
		pages[i].payload = buf.Next(MaxPayloadSize)
	}
	return &pages[0]
}
func writeWALMetadata(f file) error {
	data := make([]byte, 0, len(metadataHeader)+len(metadataVersion)+metadataStatusSize)
	data = append(data, metadataHeader[:]...)
	data = append(data, metadataVersion[:]...)
	data = append(data, byte(recoveryStateUnclean))
	data = append(data, byte('\n'))
	_, err := f.WriteAt(data, 0)
	return err
}
func (w *WAL) Close() error {
	var err1 error
	if atomic.LoadInt64(&w.atomicUnfinishedTxns) != 0 {
		err1 = errors.New("There are still non-released transactions left")
	}
	if err1 == nil && !w.deps.disrupt("UncleanShutdown") {
		err1 = w.writeRecoveryState(recoveryStateClean)
	}
	w.wg.Wait()
	err2 := w.logFile.Close()
	return errors.Compose(err1, err2)
}