func (seg *CephSegment) Write(uuid []byte, address uint64, data []byte) (uint64, error) {
	atomic.AddInt64(&totalbytes, int64(len(data)))
	if address != seg.naddr {
		lg.Panic("Non-sequential write")
	}
	if len(seg.wcache)+len(data)+2 > cap(seg.wcache) {
		seg.flushWrite()
	}
	base := len(seg.wcache)
	seg.wcache = seg.wcache[:base+2]
	seg.wcache[base] = byte(len(data))
	seg.wcache[base+1] = byte(len(data) >> 8)
	seg.wcache = append(seg.wcache, data...)
	naddr := address + uint64(len(data)+2)
	if ((naddr + MAX_EXPECTED_OBJECT_SIZE + 2) >> 24) != (address >> 24) {
		if seg.ishot {
			naddr = <-seg.sp.hot_alloc
		} else {
			naddr = <-seg.sp.cold_alloc
		}
		seg.naddr = naddr
		seg.flushWrite()
		return naddr, nil
	}
	seg.naddr = naddr
	return naddr, nil
}
Writes a slice to the segment, returns immediately
Returns nil if op is OK, otherwise ErrNoSpace or ErrInvalidArgument
It is up to the implementer to work out how to report no space immediately
The uint64 is the address to be used for the next write
package cephprovider
import (
	"context"
	"encoding/binary"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/BTrDB/btrdb-server/internal/bprovider"
	"github.com/BTrDB/btrdb-server/internal/configprovider"
	"github.com/BTrDB/btrdb-server/internal/rez"
	"github.com/ceph/go-ceph/rados"
	logging "github.com/op/go-logging"
)
var lg *logging.Logger
func init() {
	lg = logging.MustGetLogger("log")
}
const NUM_HOT_HANDLES = 128
const NUM_COLD_HANDLES = 128
const METADATA_BASE = 0xFF00000000000000
const INITIAL_COLD_BASE_ADDRESS = 0
const INITIAL_HOT_BASE_ADDRESS = 0x8000000000000000
const ADDR_LOCK_SIZE = 0x1000000000
const ADDR_OBJ_SIZE = 0x0001000000
const MAX_EXPECTED_OBJECT_SIZE = 20485
const RADOS_CACHE_SIZE = 512
const OFFSET_MASK = 0xFFFFFF
const WORTH_CACHING = OFFSET_MASK - MAX_EXPECTED_OBJECT_SIZE
const SEGCACHE_SIZE = 100 * 1024
const WCACHE_SIZE = 1 << 20
const SBLOCK_CHUNK_SHIFT = 20
const SBLOCK_CHUNK_MASK = 0xFFFFF
const SBLOCKS_PER_CHUNK = 1 << SBLOCK_CHUNK_SHIFT
const SBLOCK_SIZE = 16
var provided_rh int64
func IsAddressHot(addr uint64) bool {
	return addr >= INITIAL_HOT_BASE_ADDRESS
}
func UUIDSliceToArr(id []byte) [16]byte {
	rv := [16]byte{}
	copy(rv[:], id)
	return rv
}
type CephSegment struct {
	ishot       bool
	h           *rados.IOContext
	rez         *rez.Resource
	sp          *CephStorageProvider
	ptr         uint64
	naddr       uint64
	base        uint64  
	uid         [16]byte
	wcache      []byte
	wcache_base uint64
}
type chunkreqindex struct {
	UUID [16]byte
	Addr uint64
}
type CephStorageProvider struct {
	conn *rados.Conn
	hot_ptr    uint64
	hot_alloc  chan uint64
	cold_ptr   uint64
	cold_alloc chan uint64
	hot_segaddrcache  map[[16]byte]uint64
	hot_segcachelock  sync.Mutex
	cold_segaddrcache map[[16]byte]uint64
	cold_segcachelock sync.Mutex
	chunklock sync.Mutex
	chunkgate map[chunkreqindex][]chan []byte
	rcache *CephCache
	dataPool    string
	hotPool     string
	journalPool string
	hot_handle_q  chan *rados.IOContext
	cold_handle_q chan *rados.IOContext
	cfg configprovider.Configuration
	annotationMu sync.Mutex
	rm *rez.RezManager
}
func (sp *CephStorageProvider) DropCache() {
	sp.rcache.dropCache()
}
func (sp *CephStorageProvider) getHandle(ctx context.Context, ishot bool) (*rez.Resource, *rados.IOContext, error) {
	var h *rez.Resource
	var err error
	if ishot {
		h, err = sp.rm.Get(ctx, rez.CephHotHandle)
	} else {
		h, err = sp.rm.Get(ctx, rez.CephColdHandle)
	}
	if err != nil {
		return nil, nil, err
	}
	return h, h.Val().(*rados.IOContext), nil
}
func (sp *CephStorageProvider) initializeHotHandles() {
	sp.rm.CreateResourcePool(rez.CephHotHandle, func() interface{} {
		ioctx, err := sp.conn.OpenIOContext(sp.hotPool)
		if err != nil {
			panic(err)
		}
		return ioctx
	}, func(v interface{}) {
		v.(*rados.IOContext).Destroy()
	})
}
func (sp *CephStorageProvider) initializeColdHandles() {
	sp.rm.CreateResourcePool(rez.CephColdHandle, func() interface{} {
		ioctx, err := sp.conn.OpenIOContext(sp.dataPool)
		if err != nil {
			panic(err)
		}
		return ioctx
	}, func(v interface{}) {
		v.(*rados.IOContext).Destroy()
	})
}
func (seg *CephSegment) BaseAddress() uint64 {
	return seg.base
}
func (seg *CephSegment) Unlock() {
	seg.flushWrite()
	seg.rez.Release()
	if seg.ishot {
		if (seg.naddr & OFFSET_MASK) < WORTH_CACHING {
			seg.sp.hot_segcachelock.Lock()
			seg.sp.hotPruneSegCache()
			seg.sp.hot_segaddrcache[seg.uid] = seg.naddr
			seg.sp.hot_segcachelock.Unlock()
		}
	} else {
		if (seg.naddr & OFFSET_MASK) < WORTH_CACHING {
			seg.sp.cold_segcachelock.Lock()
			seg.sp.coldPruneSegCache()
			seg.sp.cold_segaddrcache[seg.uid] = seg.naddr
			seg.sp.cold_segcachelock.Unlock()
		}
	}
}
func (seg *CephSegment) flushWrite() {
	if len(seg.wcache) == 0 {
		return
	}
	address := seg.wcache_base
	aa := address >> 24
	oid := fmt.Sprintf("%032x%010x", seg.uid, aa)
	offset := address & OFFSET_MASK
	err := seg.h.Write(oid, seg.wcache, offset)
	if err != nil {
		panic(fmt.Errorf("ceph write error: %v", err))
	}
	for i := 0; i < len(seg.wcache); i += R_CHUNKSIZE {
		seg.sp.rcache.cacheInvalidate((uint64(i) + seg.wcache_base) & R_ADDRMASK)
	}
	seg.wcache = make([]byte, 0, WCACHE_SIZE)
	seg.wcache_base = seg.naddr
}
var totalbytes int64