Initial commit

Author: invpt

.gitignore

@@ -0,0 +1 @@
+drp-delta

Makefile

@@ -0,0 +1,8 @@
+build: *.go src/*.go
+	go build
+
+run: build
+	./drp-delta
+
+clean:
+	rm drp-delta

go.mod

@@ -0,0 +1,5 @@
+module drp-delta
+
+go 1.20
+
+require github.com/google/uuid v1.3.0

go.sum

@@ -0,0 +1,2 @@
+github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
+github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=

main.go

@@ -0,0 +1,7 @@
+package main
+
+import "fmt"
+
+func main() {
+	fmt.Println("Hello, World")
+}

src/defserver.go

@@ -0,0 +1,201 @@
+package drpdelta
+
+func (d *definition) server(inbox <-chan InboundMessage, outbox chan<- OutboundMessage) {
+	for inMsg := range inbox {
+		switch inContent := inMsg.Content.(type) {
+		case ChangeMessage:
+			inContentRef := &inContent
+
+			d.changeLists[inMsg.Sender] = append(d.changeLists[inMsg.Sender], inContentRef)
+			for _, tx := range inContent.txs {
+				txid := tx.id
+				if _, found := d.changeGraph[txid]; !found {
+					d.changeGraph[txid] = map[Address]*ChangeMessage{}
+				}
+
+				d.changeGraph[txid][inMsg.Sender] = inContentRef
+			}
+
+			for _, pred := range inContent.preds {
+				if _, found := d.changeGraph[pred.id]; !found {
+					d.changeGraph[pred.id] = map[Address]*ChangeMessage{}
+					for _, name := range pred.writes {
+						d.changeGraph[pred.id][name] = nil
+					}
+				}
+			}
+		}
+
+		d.tick(inbox, outbox)
+	}
+}
+
+func (d *definition) tick(inbox <-chan InboundMessage, outbox chan<- OutboundMessage) {
+	var smallestTransitivePreds map[*ChangeMessage]struct{}
+	for _, changeList := range d.changeLists {
+		if len(changeList) == 0 {
+			continue
+		}
+
+		change := changeList[0]
+
+		if transitivePreds, ok := d.findTransitivePreds(change); ok {
+			if smallestTransitivePreds == nil || len(transitivePreds) < len(smallestTransitivePreds) {
+				smallestTransitivePreds = transitivePreds
+			}
+		}
+	}
+
+	if smallestTransitivePreds != nil {
+		allTxs, allPreds := unionTxsPreds(smallestTransitivePreds)
+
+		for inputName, changeList := range d.changeLists {
+			for i, change := range changeList {
+				if _, found := smallestTransitivePreds[change]; found {
+					// execute the change
+					d.replicas[inputName] = change.value
+					delete(smallestTransitivePreds, change)
+				} else {
+					d.changeLists[inputName] = d.changeLists[inputName][i:]
+					break
+				}
+			}
+		}
+
+		if len(smallestTransitivePreds) > 0 {
+			panic("invariant broken: didn't exhaust the transitive preds set!")
+		}
+
+		newValue := d.f(d.replicas)
+
+		for _, sub := range d.subscribers {
+			outbox <- OutboundMessage{Target: sub, Content: ChangeMessage{txs: allTxs, preds: allPreds, value: newValue}}
+		}
+	}
+}
+
+func (d *definition) findTransitivePreds(change *ChangeMessage) (map[*ChangeMessage]struct{}, bool) {
+	preds := map[*ChangeMessage]struct{}{}
+	ok := d._findTransitivePreds(change, preds)
+	return preds, ok
+}
+
+func (d *definition) _findTransitivePreds(change *ChangeMessage, preds map[*ChangeMessage]struct{}) bool {
+	if _, found := preds[change]; found {
+		return true
+	}
+	preds[change] = struct{}{}
+
+	for _, tx := range change.txs {
+		var overlappingInputs []Address
+		for _, name := range tx.writes {
+			for inputName, inputTransitiveVariableDeps := range d.inputTransitiveVariableDeps {
+				if _, found := inputTransitiveVariableDeps[name]; found {
+					overlappingInputs = append(overlappingInputs, inputName)
+				}
+			}
+		}
+
+		for _, overlappingInput := range overlappingInputs {
+			thisInputChange := d.changeGraph[tx.id][overlappingInput]
+			if thisInputChange == nil {
+				// this input has not yet received this pred
+				return false
+			} else if !d._findTransitivePreds(thisInputChange, preds) {
+				// the recursive findTransitivePreds call failed
+				return false
+			}
+		}
+	}
+
+	for _, changePred := range change.preds {
+		if namesToChanges, found := d.changeGraph[changePred.id]; found {
+			if namesToChanges == nil {
+				panic("invariant 1 broken in _findTransitivePreds")
+			}
+
+			var overlappingInputs []Address
+			for name := range namesToChanges {
+				for inputName, inputTransitiveVariableDeps := range d.inputTransitiveVariableDeps {
+					if _, found := inputTransitiveVariableDeps[name]; found {
+						overlappingInputs = append(overlappingInputs, inputName)
+					}
+				}
+			}
+
+			for _, overlappingInput := range overlappingInputs {
+				thisInputChange := d.changeGraph[changePred.id][overlappingInput]
+				if thisInputChange == nil {
+					// this input has not yet received this pred
+					return false
+				} else if !d._findTransitivePreds(thisInputChange, preds) {
+					// the recursive findTransitivePreds call failed
+					return false
+				}
+			}
+		} else {
+			panic("invariant 2 broken in _findTransitivePreds")
+		}
+	}
+
+	return true
+}
+
+func unionTxsPreds(changes map[*ChangeMessage]struct{}) (txs []Tx, preds []Tx) {
+	txsMap := map[Txid][]Address{}
+	predsMap := map[Txid][]Address{}
+
+	for change, _ := range changes {
+		for _, tx := range change.txs {
+			txsMap[tx.id] = tx.writes
+		}
+		for _, tx := range change.preds {
+			predsMap[tx.id] = tx.writes
+		}
+	}
+
+	txs = make([]Tx, len(txsMap))
+	preds = make([]Tx, len(predsMap))
+	for id, writes := range txsMap {
+		txs = append(txs, Tx{id, writes})
+	}
+	for id, writes := range predsMap {
+		preds = append(txs, Tx{id, writes})
+	}
+
+	return
+}
+
+func NewDefinition(address Address, deps []ReactiveNode, f func(map[Address]any) any) definition {
+	replicas := map[Address]any{}
+	changeLists := map[Address][]*ChangeMessage{}
+	for _, dep := range deps {
+		dep.Subscribe(address)
+		replicas[dep.Address()] = nil
+		changeLists[dep.Address()] = []*ChangeMessage{}
+	}
+
+	return definition{
+		Address:                     address,
+		replicas:                    replicas,
+		changeGraph:                 map[Txid]map[Address]*ChangeMessage{},
+		changeLists:                 changeLists,
+		f:                           f,
+		subscribers:                 []Address{},
+		inputTransitiveVariableDeps: map[Address]map[Address]struct{}{}, // TODO: properly impl this
+	}
+}
+
+func (d *definition) Subscribe(who Address) {
+	d.subscribers = append(d.subscribers, who)
+}
+
+type definition struct {
+	Address                     Address
+	replicas                    map[Address]any
+	changeGraph                 map[Txid]map[Address]*ChangeMessage
+	changeLists                 map[Address][]*ChangeMessage
+	f                           func(map[Address]any) any
+	subscribers                 []Address
+	inputTransitiveVariableDeps map[Address]map[Address]struct{}
+}

src/doaction.go

@@ -0,0 +1,81 @@
+package drpdelta
+
+import "fmt"
+
+func DoAction(action Action, inbox <-chan InboundMessage, outbox chan<- OutboundMessage) {
+	txid := NewTxid()
+
+	// make a list of all the things we write to for later use
+	writes := []Address{}
+	for name, kind := range action.targets {
+		if kind == write {
+			writes = append(writes, name)
+		}
+	}
+
+	// request all queues
+	for name := range action.targets {
+		outbox <- OutboundMessage{Target: name, Content: QueueMessage{txid: txid}}
+	}
+	// wait to be granted all queues
+	queueResponseCount := 0
+	for inMsg := range inbox {
+		if _, ok := inMsg.Content.(QueueGrantedMessage); ok {
+			queueResponseCount += 1
+
+			if queueResponseCount == len(action.targets) {
+				break
+			}
+		} else {
+			fmt.Println("Unexpected message:", inMsg)
+			panic("")
+		}
+	}
+
+	// request all locks
+	for name, kind := range action.targets {
+		outbox <- OutboundMessage{Target: name, Content: LockMessage{txid: txid, kind: kind}}
+	}
+	// wait to be granted all locks
+	lockResponseCount := 0
+	for inMsg := range inbox {
+		if _, ok := inMsg.Content.(LockGrantedMessage); ok {
+			lockResponseCount += 1
+
+			if lockResponseCount == len(action.targets) {
+				break
+			}
+		} else {
+			fmt.Println("Unexpected message:", inMsg)
+			panic("")
+		}
+	}
+
+	// request all reads
+	for name := range action.targets {
+		outbox <- OutboundMessage{Target: name, Content: ReadMessage{tx: Tx{id: txid, writes: writes}}}
+	}
+	// wait to be granted all reads
+	values := map[Address]any{}
+	preds := []Tx{}
+	for inMsg := range inbox {
+		if inContent, ok := inMsg.Content.(ReadResultMessage); ok {
+			values[inMsg.Sender] = inContent.value
+			preds = append(preds, inContent.valueTx)
+		}
+	}
+
+	// do all writes
+	for name, value := range action.f(values) {
+		if kind, ok := action.targets[name]; !ok || kind != write {
+			panic("invariant broken: action tries to write to variable it doesn't have a write lock for")
+		}
+
+		outbox <- OutboundMessage{Target: name, Content: WriteMessage{tx: Tx{id: txid, writes: writes}, preds: preds, value: value}}
+	}
+}
+
+type Action struct {
+	targets map[Address]lockKind
+	f       func(map[Address]any) map[Address]any
+}

src/messages.go

@@ -0,0 +1,69 @@
+package drpdelta
+
+import "github.com/google/uuid"
+
+type InboundMessage struct {
+	Sender  Address
+	Content any
+}
+
+type OutboundMessage struct {
+	Target  Address
+	Content any
+}
+
+type Address string
+
+type QueueMessage struct {
+	txid Txid
+}
+
+type LockMessage struct {
+	txid Txid
+	kind lockKind
+}
+
+type ReadMessage struct {
+	tx Tx
+}
+
+type WriteMessage struct {
+	tx    Tx
+	preds []Tx
+	value any
+}
+
+type QueueGrantedMessage struct {
+	txid Txid
+}
+
+type LockGrantedMessage struct {
+	txid Txid
+	kind lockKind
+}
+
+type ReadResultMessage struct {
+	txid    Txid
+	value   any
+	valueTx Tx
+}
+
+type ChangeMessage struct {
+	txs   []Tx
+	preds []Tx
+	value any
+}
+
+// A transaction ID plus the set of names written to by the transaction.
+type Tx struct {
+	// the transaction's ID
+	id Txid
+	// the list of values written to by the transaction
+	writes []Address
+}
+
+type Txid string
+
+func NewTxid() Txid {
+	return Txid(uuid.New().String())
+}

src/orchestrator.go

@@ -0,0 +1,6 @@
+package drpdelta
+
+type ReactiveNode interface {
+	Subscribe(Address)
+	Address() Address
+}