func (jp *CJournalProvider) waitForBufferWritten(ctx context.Context, equalCP uint64) bte.BTE {
	for {
		jp.writtencpmu.Lock()
		ch := jp.writtencpchange
		cp := jp.writtencp
		jp.writtencpmu.Unlock()
		if cp >= equalCP {
			return nil
		}
		select {
		case <-ch:
			continue
		case <-ctx.Done():
			return bte.ErrW(bte.ContextError, "context error", ctx.Err())
		}
	}
}
Called without the lock held, waits for the persisted checkpoint number to
exceed or equal the given checkpoint
package cephprovider
import (
	"container/heap"
	"context"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/BTrDB/btrdb-server/bte"
	"github.com/BTrDB/btrdb-server/internal/configprovider"
	"github.com/BTrDB/btrdb-server/internal/jprovider"
	"github.com/ceph/go-ceph/rados"
	opentracing "github.com/opentracing/opentracing-go"
	"github.com/prometheus/client_golang/prometheus"
)
var writehook = func() {}
const CJournalProviderNamespace = "journalprovider"
const MaxObjectSize = 16 * 1024 * 1024
var pmCacheHit prometheus.Counter
var pmCacheMiss prometheus.Counter
var pmCacheInvalidate prometheus.Counter
var pmCanFreeCP prometheus.Gauge
var pmBeenFreedCP prometheus.Gauge
var pmMaxFreeCP prometheus.Gauge
var pmFreeListLength prometheus.Gauge
func init() {
	pmCacheHit = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "btrdb",
		Subsystem: "cephstore",
		Name:      "cachehit",
		Help:      "The number of hits off the ceph cache",
	})
	prometheus.MustRegister(pmCacheHit)
	pmCacheMiss = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "btrdb",
		Subsystem: "cephstore",
		Name:      "cachemiss",
		Help:      "The number of misses off the ceph cache",
	})
	prometheus.MustRegister(pmCacheMiss)
	pmCacheInvalidate = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "btrdb",
		Subsystem: "cephstore",
		Name:      "cacheinv",
		Help:      "The number of cache invalidations that have occurred",
	})
	prometheus.MustRegister(pmCacheInvalidate)
	pmCanFreeCP = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "btrdb",
		Subsystem: "journal",
		Name:      "canfree",
		Help:      "The highest checkpoint that can be freed",
	})
	prometheus.MustRegister(pmCanFreeCP)
	pmBeenFreedCP = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "btrdb",
		Subsystem: "journal",
		Name:      "beenfreed",
		Help:      "The highest checkpoint that has been freed",
	})
	prometheus.MustRegister(pmBeenFreedCP)
	pmMaxFreeCP = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "btrdb",
		Subsystem: "journal",
		Name:      "maxfree",
		Help:      "The highest checkpoint that is written and waiting to be freed",
	})
	prometheus.MustRegister(pmMaxFreeCP)
	pmFreeListLength = prometheus.NewGauge(prometheus.GaugeOpts{
		Namespace: "btrdb",
		Subsystem: "journal",
		Name:      "fll",
		Help:      "The length of the journal free list",
	})
	prometheus.MustRegister(pmFreeListLength)
}
func NewJournalProvider(cfg configprovider.Configuration, ccfg configprovider.ClusterConfiguration) (jprovider.JournalProvider, bte.BTE) {
	conn, err := rados.NewConn()
	if err != nil {
		lg.Panicf("Could not initialize ceph storage: %v", err)
	}
	err = conn.ReadConfigFile(cfg.StorageCephConf())
	if err != nil {
		lg.Panicf("Could not read ceph config: %v", err)
	}
	err = conn.Connect()
	if err != nil {
		lg.Panicf("Could not initialize ceph storage: %v", err)
	}
	return newJournalProvider(ccfg.NodeName(), conn, cfg.StorageCephJournalPool())
}
func newJournalProvider(ournodename string, conn *rados.Conn, pool string) (jprovider.JournalProvider, bte.BTE) {
	wbioctx, err := conn.OpenIOContext(pool)
	if err != nil {
		return nil, bte.ErrW(bte.CephError, "could not open ioctx", err)
	}
	wbioctx.SetNamespace(CJournalProviderNamespace)
	rbioctx, err := conn.OpenIOContext(pool)
	if err != nil {
		return nil, bte.ErrW(bte.CephError, "could not open ioctx", err)
	}
	rbioctx.SetNamespace(CJournalProviderNamespace)
	data := make([]byte, 8)
	nread, err := wbioctx.Read("node/"+ournodename, data, 0)
	if nread != 0 || err != rados.RadosErrorNotFound {
		return nil, bte.ErrF(bte.NodeExisted, "Node %s has existed before", ournodename)
	}
	err = wbioctx.WriteFull("node/"+ournodename, data)
	writehook()
	if err != nil {
		return nil, bte.ErrW(bte.CephError, "could not write node cookie", err)
	}
	rv := &CJournalProvider{
		conn:            conn,
		pool:            pool,
		wbioctx:         wbioctx,
		rbioctx:         rbioctx,
		nodename:        ournodename,
		writebacks:      make(chan writeBackRequest, 10),
		bufferpriq:      make(chan *bufferHandleReq, 10),
		bufferq:         make(chan *bufferHandleReq, 10),
		writtencpchange: make(chan struct{}),
		currentBuffer:   nil,
		canFreeCP:       1,  
		beenFreedCP:     1,  
		freelist:        CheckpointHeap{},
		pendingcp:       make(map[cprange]struct{}),
	}
	go rv.freeCheckpoints()
	go rv.printFreeSegmentList()
	go rv.startBufferQAdmission(context.Background())
	for i := 0; i < 10; i++ {
		go rv.startWritebackWorker(context.Background())
	}
	return rv, nil
}
func (jp *CJournalProvider) freeCheckpoints() {
	for {
		time.Sleep(5 * time.Second)
		jp.freelistmu.Lock()
		canFreeCP := jp.canFreeCP
		jp.freelistmu.Unlock()
		if canFreeCP > jp.beenFreedCP {
			lg.Infof("[JRN] performing actual free up to CP=%d", canFreeCP)
			jp.rbmu.Lock()
			jp.releaseJournalEntriesLockHeld(context.Background(), jp.nodename, jprovider.Checkpoint(canFreeCP), &configprovider.FullMashRange)
			jp.rbmu.Unlock()
			jp.beenFreedCP = canFreeCP
			pmBeenFreedCP.Set(float64(jp.beenFreedCP))
		}
	}
}
func (jp *CJournalProvider) printFreeSegmentList() {
	for {
		time.Sleep(5 * time.Second)
		lg.Infof("[JRN] CanFreeCP=%d BeenFreedCP=%d MaxFreeCP=%d FLL=%d", jp.canFreeCP, jp.beenFreedCP, jp.maxFreeCP, jp.freelist.Len())
	}
}
func (jp *CJournalProvider) ReleaseDisjointCheckpoint(ctx context.Context, cp jprovider.Checkpoint) bte.BTE {
	jp.freelistmu.Lock()
	jp.addFreedCheckpointToList(uint64(cp))
	for jp.freelist.Len() > 0 && jp.freelist.Peek() == jp.canFreeCP {
		heap.Pop(&jp.freelist)
		jp.canFreeCP++
	}
	pmCanFreeCP.Set(float64(jp.canFreeCP))
	jp.freelistmu.Unlock()
	return nil
}
type CJournalProvider struct {
	conn       *rados.Conn
	pool       string
	wbioctx    *rados.IOContext
	rbioctx    *rados.IOContext
	rbmu       sync.Mutex
	freelistmu sync.Mutex
	writtencp       uint64
	writtencpmu     sync.Mutex
	writtencpchange chan struct{}
	pendingcp map[cprange]struct{}
	nodename string
	writebacks chan writeBackRequest
	bufferq chan *bufferHandleReq
	bufferpriq    chan *bufferHandleReq
	currentBuffer *bufferEntry
	canFreeCP   uint64
	beenFreedCP uint64
	maxFreeCP   uint64
	freelist CheckpointHeap
}
func (jp *CJournalProvider) addFreedCheckpointToList(cp uint64) bte.BTE {
	if cp > jp.maxFreeCP {
		jp.maxFreeCP = cp
		pmMaxFreeCP.Set(float64(cp))
	}
	heap.Push(&jp.freelist, cp)
	return nil
}
type writeBackRequest struct {
	ObjName string
	Data         []byte
	Offset       uint64
	StartCP      uint64
	EndCPNonIncl uint64
}
type bufferEntry struct {
	data    []byte
	objname string
	bytesWritten uint64
	rng *configprovider.MashRange
	startCP uint64
	nextCP uint64
	nextWriteCP uint64
}
type bufferHandle struct {
	Err  error
	Buf  *bufferEntry
	Done chan struct{}
}
type bufferHandleReq struct {
	ctx    context.Context
	rng    *configprovider.MashRange
	size   int
	cancel func()
	ch     chan *bufferHandle
}
type cprange struct {
	Start uint64
	End   uint64
}
type objName struct {
	NodeName           string
	StartingCheckpoint uint64
}
type jiterator struct {
	jp         *CJournalProvider
	objectlist []string
	on         *objName
	nn         string
	value      *CJrecord
	buffer     []byte
}
func (jp *CJournalProvider) markCPRangeWritten(startCP, endCP uint64) {
	jp.writtencpmu.Lock()
	jp.pendingcp[cprange{Start: startCP, End: endCP}] = struct{}{}
outer:
	for {
		for cprange, _ := range jp.pendingcp {
			if jp.writtencp == cprange.Start-1 {
				jp.writtencp = cprange.End - 1
				ch := jp.writtencpchange
				jp.writtencpchange = make(chan struct{})
				close(ch)
				delete(jp.pendingcp, cprange)
				continue outer
			}
		}
		break
	}
	jp.writtencpmu.Unlock()
}
func (jp *CJournalProvider) waitForBufferHandle(ctx context.Context, rng *configprovider.MashRange, priority bool, size int) chan *bufferHandle {
	subctx, cancel := context.WithCancel(ctx)
	rv := make(chan *bufferHandle, 2)
	bhreq := &bufferHandleReq{
		ctx:    subctx,
		rng:    rng,
		size:   size,
		cancel: cancel,
		ch:     rv,
	}
	target := jp.bufferq
	if priority {
		target = jp.bufferpriq
	}
	select {
	case target <- bhreq:
		return rv
	case <-ctx.Done():
		cancel()
		rv <- &bufferHandle{
			Err:  ctx.Err(),
			Done: nil,
		}
		return rv
	}
}
func (jp *CJournalProvider) startBufferQAdmission(ctx context.Context) {
	for {
		if ctx.Err() != nil {
			return
		}
		var req *bufferHandleReq
		select {
		case req = <-jp.bufferpriq:
		default:
			select {
			case req = <-jp.bufferpriq:
			case req = <-jp.bufferq:
			}
		}
		if req.ctx.Err() != nil {
			req.ch <- &bufferHandle{
				Err:  req.ctx.Err(),
				Done: nil,
			}
			continue
		}
		if req.size != 0 {
			if jp.currentBuffer == nil ||
				!jp.currentBuffer.rng.Equal(req.rng) ||
				len(jp.currentBuffer.data)+req.size >= MaxObjectSize {
				if jp.currentBuffer != nil && !jp.currentBuffer.rng.Equal(req.rng) {
				}
				if jp.currentBuffer != nil && len(jp.currentBuffer.data)+req.size >= MaxObjectSize {
				}
				err := jp.writeBackBufferWithFlip(ctx, req.rng)
				if err != nil {
					panic(err)
				}
			}
		}
		done := make(chan struct{})
		span := opentracing.StartSpan("PQBufferHeldA")
		req.ch <- &bufferHandle{
			Err:  nil,
			Buf:  jp.currentBuffer,
			Done: done,
		}
		<-done
		span.Finish()
	}
}
func (jp *CJournalProvider) startWritebackWorker(ctx context.Context) {
	ioctx, err := jp.conn.OpenIOContext(jp.pool)
	if err != nil {
		panic(err)
	}
	ioctx.SetNamespace(CJournalProviderNamespace)
	for {
		if ctx.Err() != nil {
			ioctx.Destroy()
			return
		}
		args := <-jp.writebacks
		if len(args.Data) == 0 {
			panic("data len is zero")
		}
		err := ioctx.Write(args.ObjName, args.Data, args.Offset)
		writehook()
		if err != nil {
			panic(err)
		}
		jp.markCPRangeWritten(args.StartCP, args.EndCPNonIncl)
	}
}
func (jp *CJournalProvider) writeBackBufferNoFlip(ctx context.Context) bte.BTE {
	if jp.currentBuffer != nil {
		err := jp.writeExistingBuffer(ctx)
		if err != nil {
			return err
		}
	}
	return nil
}
func (jp *CJournalProvider) writeBackBufferWithFlip(ctx context.Context, newRange *configprovider.MashRange) bte.BTE {
	if jp.currentBuffer != nil {
		err := jp.writeExistingBuffer(ctx)
		if err != nil {
			return err
		}
	}
	err := jp.beginNewObject(newRange)
	return err
}
func (jp *CJournalProvider) Barrier(ctx context.Context, cp jprovider.Checkpoint) bte.BTE {
	if ctx.Err() != nil {
		return bte.ErrW(bte.ContextError, "context error", ctx.Err())
	}
	err := jp.WaitForCheckpoint(ctx, cp)
	if err != nil {
		return err
	}
	buf := jp.waitForBufferHandle(ctx, nil, true, 0)
	hnd := <-buf
	err = jp.writeBackBufferWithFlip(ctx, jp.currentBuffer.rng)
	if err != nil {
		panic(err)
	}
	hnd.Done <- struct{}{}
	return nil
}
func (jp *CJournalProvider) writeExistingBuffer(ctx context.Context) bte.BTE {
	if int(jp.currentBuffer.bytesWritten) == len(jp.currentBuffer.data) {
		return nil
	}
	span := opentracing.StartSpan("PQEnqWriteback")
	jp.writebacks <- writeBackRequest{
		ObjName:      jp.currentBuffer.objname,
		Data:         jp.currentBuffer.data[jp.currentBuffer.bytesWritten:],
		Offset:       jp.currentBuffer.bytesWritten,
		StartCP:      jp.currentBuffer.nextWriteCP,
		EndCPNonIncl: jp.currentBuffer.nextCP,
	}
	span.Finish()
	jp.currentBuffer.bytesWritten = uint64(len(jp.currentBuffer.data))
	jp.currentBuffer.nextWriteCP = jp.currentBuffer.nextCP
	return nil
}