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()
}
Mark that the given range of checkpoints is now durable on disk
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
}