defer.go

//  Copyright (c) 2013 Couchbase, Inc.
//  Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
//  except in compliance with the License. You may obtain a copy of the License at
//    http://www.apache.org/licenses/LICENSE-2.0
//  Unless required by applicable law or agreed to in writing, software distributed under the
//  License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
//  either express or implied. See the License for the specific language governing permissions
//  and limitations under the License.

// defer goroutine that handles mvcc snapshots. Functionalites of this process
// augments the functionalit of MVCC process.
package btree

import (
	"log"
	"sync/atomic"
	"time"
)

const (
	DEFER_ADD byte = iota
	DEFER_DELETE
)

type DEFER struct {
	deferReq chan []interface{}
}

// FIXME deprecated. nobody is this now.
func (wstore *WStore) pingCache(what byte, fpos int64, node Node) {
	wstore.deferReq <- []interface{}{WS_PINGCACHE, what, fpos, node}
}

// Add intermediate key into the KD's ping-cache. FIXME this logic is not
// integrated with the btree algorithm
func (wstore *WStore) pingKey(what byte, fpos int64, key []byte) {
	wstore.deferReq <- []interface{}{WS_PINGKD, what, fpos, key}
}

// Add intermediate docid into the KD's ping-cache. FIXME this logic is not
// integrated with the btree algorithm
func (wstore *WStore) pingDocid(what byte, fpos int64, docid []byte) {
	wstore.deferReq <- []interface{}{WS_PINGKD, what, fpos, docid}
}

// Post a multi-version snapshot, generated by index mutation, to deferr-
// process.
func (wstore *WStore) postMV(mv *MV) {
	wstore.deferReq <- []interface{}{WS_MV, mv}
}

// Synchronize disk snapshot with in-memory snapshot.
func (wstore *WStore) syncSnapshot(minAccess int64, force bool) {
	syncChan := make(chan []interface{})
	x := []interface{}{WS_SYNCSNAPSHOT, minAccess, syncChan, force}
	wstore.deferReq <- x
	<-syncChan
}

func doDefer(wstore *WStore) {
	var cmd []interface{}
	var oldmv *MV
	// Following collection objects are used for every cycle of MVCC snapshot
	// synchronization.
	addKDs := make(map[int64][]byte)
	delKDs := make(map[int64][]byte)
	for {
		cmd = <-wstore.deferReq
		if cmd != nil {
			switch cmd[0].(byte) {

			case WS_PINGCACHE: // FIXME not being used by anyone
				what, fpos, node := cmd[1].(byte), cmd[2].(int64), cmd[3].(Node)
				if what == DEFER_ADD {
					wstore._pingCache(fpos, node)
				}

			case WS_PINGKD: // FIXME not yet integrated with btree algorithm
				what, fpos, v := cmd[1].(byte), cmd[2].(int64), cmd[3].([]byte)
				kdping := (*map[int64][]byte)(atomic.LoadPointer(&wstore.kdping))
				if what == DEFER_ADD {
					addKDs[fpos] = v
					(*kdping)[fpos] = v
				} else if what == DEFER_DELETE {
					delKDs[fpos] = v
					delete(*kdping, fpos)
				}

			case WS_MV: // postMV()
				mv := cmd[1].(*MV)
				if oldmv != nil && wstore.Debug {
					if oldmv.root != mv.stales[0] {
						log.Panicln("snapshots are not chained", oldmv, mv)
					}
				}
				oldmv = mv

				for fpos, node := range mv.commits { // update commitQ & ping cache
					wstore._pingCache(fpos, node)
				}
				wstore.maxlenMVQ = max(wstore.maxlenMVQ, int64(len(wstore.mvQ)))
				if wstore.Debug {
					log.Println("MVComms", commitkeys(mv.commits))
					log.Println("MVStales", mv.stales)
				}

			case WS_SYNCSNAPSHOT: // syncSnapshot()
				var mvroot, mvts int64

				minAccess, syncChan := cmd[1].(int64), cmd[2].(chan []interface{})
				force := cmd[3].(bool)
				hdts := wstore.head.timestamp

				if throttleMVCC(wstore, minAccess, hdts) {
					syncChan <- nil
					continue
				}

				if wstore.Debug {
					log.Println("Minimum access", minAccess, hdts)
					log.Println("accessQ", wstore.accessQ)
				}

				commitQ, snapshot := snapshotToCommit(wstore, hdts)
				recycleQ := recycleSnapshot(wstore, minAccess, hdts, force)

				wstore.recycleCount += int64(len(recycleQ))
				if wstore.Debug {
					wstore.assertNotMemberCache(recycleQ)
				}

				if snapshot == nil {
					mvroot, mvts = wstore.head.root, hdts
				} else {
					mvroot, mvts = snapshot.root, snapshot.timestamp
				}

				wstore.flushSnapshot(commitQ, recycleQ, mvroot, mvts, force)
				wstore.setSnapShot(recycleQ, mvroot, mvts)

				// Update btree's ping cache
				for _, node := range commitQ {
					wstore._pingCache(node.getKnode().fpos, node)
				}
				for _, fpos := range recycleQ {
					wstore._pingCacheEvict(fpos)
				}
				if wstore.Debug {
					wstore.assertNotMemberCache(recycleQ)
				}

				// Reset and restart the cycle of snapshot synchronization
				addKDs = make(map[int64][]byte)
				delKDs = make(map[int64][]byte)
				wstore.commitQ = make(map[int64]Node)
				syncChan <- nil

			case WS_CLOSE: // Quit
				syncChan := cmd[1].(chan []interface{})
				syncChan <- nil
			}
		} else {
			break
		}
	}
}

func throttleMVCC(wstore *WStore, minAccess, hdts int64) bool {
	if minAccess == 0 {
		return false
	}
	if (hdts - minAccess) < int64(wstore.DrainRate*2) {
		return false
	}
	if wstore.Debug {
		log.Println(
			"Sleeping for ",
			wstore.MVCCThrottleRate*time.Millisecond,
			hdts,
			minAccess,
		)
	}
	time.Sleep(wstore.MVCCThrottleRate * time.Millisecond)
	return true
}

// Commit next batch of snapshots from head.timestamp
func snapshotToCommit(wstore *WStore, hdts int64) ([]Node, *MV) {
	var snapshot *MV
	commitQ := make([]Node, 0, len(wstore.commitQ))
	for _, mvp := range wstore.mvQ {
		if mvp.timestamp > hdts {
			for fpos, node := range mvp.commits {
				delete(wstore.commitQ, fpos)
				commitQ = append(commitQ, node)
			}
			snapshot = mvp
		}
	}
	return commitQ, snapshot
}

// Gather RecycleQ
func recycleSnapshot(wstore *WStore, minAccess, hdts int64, force bool) []int64 {
	recycleQ := make([]int64, 0, wstore.DrainRate*wstore.Maxlevel)
	if minAccess == 0 || hdts == 0 || minAccess > hdts {
		skip := 0
		for _, mvp := range wstore.mvQ {
			// If all of them are false break out of the loop
			if !(force || hdts == 0 || (mvp.timestamp < hdts)) {
				break
			}
			recycleQ = append(recycleQ, mvp.stales...)
			for _, fpos := range mvp.stales {
				delete(wstore.commitQ, fpos)
				wstore._pingCacheEvict(fpos)
			}
			skip++
		}
		wstore.mvQ = wstore.mvQ[skip:]
	}
	if wstore.Debug {
		log.Println("stales", recycleQ)
	}
	return recycleQ
}

func commitkeys(commits map[int64]Node) []int64 {
	ks := make([]int64, 0)
	for _, node := range commits {
		ks = append(ks, node.getKnode().fpos)
	}
	return ks
}