<commit_before>package txnengine

import (
	"fmt"
	capn "github.com/glycerine/go-capnproto"
	mdb "github.com/msackman/gomdb"
	mdbs "github.com/msackman/gomdb/server"
	"goshawkdb.io/common"
	"goshawkdb.io/server"
	msgs "goshawkdb.io/server/capnp"
	"goshawkdb.io/server/configuration"
	"goshawkdb.io/server/db"
	"goshawkdb.io/server/dispatcher"
	"math/rand"
	"time"
)

type VarManager struct {
	LocalConnection
	disk       *mdbs.MDBServer
	active     map[common.VarUUId]*Var
	exe        *dispatcher.Executor
	onIdle     func()
	lc         LocalConnection
	callbacks  []func()
	beaterLive bool
}

func init() {
	db.DB.Vars = &mdbs.DBISettings{Flags: mdb.CREATE}
}

func NewVarManager(exe *dispatcher.Executor, server *mdbs.MDBServer, lc LocalConnection) *VarManager {
	return &VarManager{
		LocalConnection: lc,
		disk:            server,
		active:          make(map[common.VarUUId]*Var),
		exe:             exe,
		callbacks:       []func(){},
	}
}

func (vm *VarManager) ApplyToVar(fun func(*Var, error), createIfMissing bool, uuid *common.VarUUId) {
	v, err := vm.find(uuid)
	if err == mdb.NotFound && createIfMissing {
		v = NewVar(uuid, vm.exe, vm.disk, vm)
		vm.active[*v.UUId] = v
		server.Log(uuid, "New var")
	} else if err != nil {
		fun(nil, err)
		return
	}
	fun(v, nil)
	if _, found := vm.active[*uuid]; !found && !v.isIdle() {
		panic(fmt.Sprintf("Var is not active, yet is not idle! %v %v", uuid, fun))
	} else if found && vm.onIdle != nil {
		v.ForceToIdle()
		vm.checkAllIdle()
	} else if vm.onIdle != nil {
		vm.checkAllIdle()
	}
}

func (vm *VarManager) ForceToIdle(onIdle func()) {
	vm.onIdle = onIdle
	if !vm.checkAllIdle() {
		for uuid, v := range vm.active {
			if configuration.TopologyVarUUId.Compare(&uuid) == common.EQ {
				continue
			}
			v.ForceToIdle()
		}
		vm.checkAllIdle()
	}
}

func (vm *VarManager) checkAllIdle() bool {
	onIdle := vm.onIdle
	if onIdle == nil {
		return true
	} else if l := len(vm.active); l == 0 {
		vm.onIdle = nil
		onIdle()
		return true
	} else if _, found := vm.active[*configuration.TopologyVarUUId]; l == 1 && found {
		vm.onIdle = nil
		onIdle()
		return true
	}
	return false
}

// var.VarLifecycle interface
func (vm *VarManager) SetInactive(v *Var) {
	server.Log(v.UUId, "is now inactive")
	v1, found := vm.active[*v.UUId]
	switch {
	case !found:
		panic(fmt.Sprintf("%v inactive but doesn't exist!\n", v.UUId))
	case v1 != v:
		panic(fmt.Sprintf("%v inactive but different var! %p %p\n", v.UUId, v, v1))
	default:
		//fmt.Printf("%v is now inactive. ", v.UUId)
		delete(vm.active, *v.UUId)
	}
}

func (vm *VarManager) Immigrate(uuid *common.VarUUId, varCap *msgs.Var, txnId *common.TxnId, txnCap *msgs.Txn, cont func(error)) {
	if _, found := vm.active[*uuid]; found {
		server.Log("Immigration ignored as var is active. Assuming it arrived late.", uuid, txnId)
		cont(nil)
		return
	}

	txnBites := db.TxnToRootBytes(txnCap)
	varSeg := capn.NewBuffer(nil)
	varCapRoot := msgs.NewRootVar(varSeg)
	varCapRoot.SetId(varCap.Id())<commit_msg>Ok, so buffering of immigration in varmanagers until the var is idle is done. However, correct merging of var state is not done, and it also occurs to me that as the topology can complete before all such migration msgs have been received I'm going to need a way to filter these out if they arrive late. Ref T6.<commit_after>package txnengine

import (
	"fmt"
	capn "github.com/glycerine/go-capnproto"
	mdb "github.com/msackman/gomdb"
	mdbs "github.com/msackman/gomdb/server"
	"goshawkdb.io/common"
	"goshawkdb.io/server"
	msgs "goshawkdb.io/server/capnp"
	"goshawkdb.io/server/configuration"
	"goshawkdb.io/server/db"
	"goshawkdb.io/server/dispatcher"
	"math/rand"
	"time"
)

type VarManager struct {
	LocalConnection
	disk       *mdbs.MDBServer
	active     map[common.VarUUId]*Var
	onIdle     func()
	onVarIdle  map[common.VarUUId][]func()
	lc         LocalConnection
	callbacks  []func()
	beaterLive bool
	exe        *dispatcher.Executor
}

func init() {
	db.DB.Vars = &mdbs.DBISettings{Flags: mdb.CREATE}
}

func NewVarManager(exe *dispatcher.Executor, server *mdbs.MDBServer, lc LocalConnection) *VarManager {
	return &VarManager{
		LocalConnection: lc,
		disk:            server,
		active:          make(map[common.VarUUId]*Var),
		onVarIdle:       make(map[common.VarUUId][]func()),
		callbacks:       []func(){},
		exe:             exe,
	}
}

func (vm *VarManager) ApplyToVar(fun func(*Var, error), createIfMissing bool, uuid *common.VarUUId) {
	v, err := vm.find(uuid)
	if err == mdb.NotFound && createIfMissing {
		v = NewVar(uuid, vm.exe, vm.disk, vm)
		vm.active[*v.UUId] = v
		server.Log(uuid, "New var")
	} else if err != nil {
		fun(nil, err)
		return
	}
	fun(v, nil)
	if _, found := vm.active[*uuid]; !found && !v.isIdle() {
		panic(fmt.Sprintf("Var is not active, yet is not idle! %v %v", uuid, fun))
	} else if vm.onIdle != nil {
		if found {
			v.ForceToIdle()
		}
		vm.checkAllIdle()
	}
}

func (vm *VarManager) ForceToIdle(onIdle func()) {
	vm.onIdle = onIdle
	if !vm.checkAllIdle() {
		for uuid, v := range vm.active {
			if configuration.TopologyVarUUId.Compare(&uuid) == common.EQ {
				continue
			}
			v.ForceToIdle()
		}
		vm.checkAllIdle()
	}
}

func (vm *VarManager) checkAllIdle() bool {
	onIdle := vm.onIdle
	if onIdle == nil {
		return true
	} else if l := len(vm.active); l == 0 {
		vm.onIdle = nil
		onIdle()
		return true
	} else if _, found := vm.active[*configuration.TopologyVarUUId]; l == 1 && found {
		vm.onIdle = nil
		onIdle()
		return true
	}
	return false
}

// var.VarLifecycle interface
func (vm *VarManager) SetInactive(v *Var) {
	server.Log(v.UUId, "is now inactive")
	v1, found := vm.active[*v.UUId]
	switch {
	case !found:
		panic(fmt.Sprintf("%v inactive but doesn't exist!\n", v.UUId))
	case v1 != v:
		panic(fmt.Sprintf("%v inactive but different var! %p %p\n", v.UUId, v, v1))
	default:
		//fmt.Printf("%v is now inactive. ", v.UUId)
		delete(vm.active, *v.UUId)
		if funcs, found := vm.onVarIdle[*v.UUId]; found {
			delete(vm.onVarIdle, *v.UUId)
			for _, f := range funcs {
				f()
			}
		}
	}
}

func (vm *VarManager) Immigrate(uuid *common.VarUUId, varCap *msgs.Var, txnId *common.TxnId, txnCap *msgs.Txn, cont func(error)) {
	if _, found := vm.active[*uuid]; found {
		funcs := vm.onVarIdle[*uuid]
		vm.onVarIdle[*uuid] = append(funcs, func() { vm.Immigrate(uuid, varCap, txnId, txnCap, cont) })
		return
	}

	txnBites := db.TxnToRootBytes(txnCap)
	varSeg := capn.NewBuffer(nil)
	varCapRoot := msgs.NewRootVar(varSeg)
	varCapRoot.SetId(varCap.Id())
{'fraction_non_alphanumeric': 0.1494394102288435, 'fraction_numerical': 0.0016894486254031638, 'mean_word_length': 3.595624558927311, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7426150', 'n_tokens_mistral': 2725, 'n_tokens_neox': 2565, 'n_words': 598}