MetzNet
/
graphvent
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Changed WaitInfo struct to include reason

gql_cataclysm
noah metz 2023-11-07 20:51:34 -07:00
parent 8bb1dacf23
commit 8b91d0af0c
2 changed files with 36 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
lockable.go
Unescape Escape View File

@ -73,18 +73,18 @@ func (ext *LockableExt) HandleErrorSignal(ctx *Context, node *Node, source NodeI
info, info_found := node.ProcessResponse(ext.WaitInfos, signal) info, info_found := node.ProcessResponse(ext.WaitInfos, signal)
if info_found { if info_found {
state, found := ext.Requirements[info.NodeID] state, found := ext.Requirements[info.Destination]
if found == true { if found == true {
ctx.Log.Logf("lockable", "got mapped response %+v for %+v in state %s", signal, info, ReqStateStrings[state]) ctx.Log.Logf("lockable", "got mapped response %+v for %+v in state %s", signal, info, ReqStateStrings[state])
switch state { switch state {
case Locking: case Locking:
ext.State = AbortingLock ext.State = AbortingLock
ext.Requirements[info.NodeID] = Unlocked ext.Requirements[info.Destination] = Unlocked
for id, state := range(ext.Requirements) { for id, state := range(ext.Requirements) {
if state == Locked { if state == Locked {
ext.Requirements[id] = Unlocking ext.Requirements[id] = Unlocking
lock_signal := NewLockSignal("unlock") lock_signal := NewLockSignal("unlock")
ext.WaitInfos[lock_signal.Id] = node.QueueTimeout(id, lock_signal, 100*time.Millisecond) ext.WaitInfos[lock_signal.Id] = node.QueueTimeout("unlock", id, lock_signal, 100*time.Millisecond)
messages = messages.Add(ctx, id, node, nil, lock_signal) messages = messages.Add(ctx, id, node, nil, lock_signal)
ctx.Log.Logf("lockable", "sent abort unlock to %s from %s", id, node.ID) ctx.Log.Logf("lockable", "sent abort unlock to %s from %s", id, node.ID)
} }
@ -92,7 +92,7 @@ func (ext *LockableExt) HandleErrorSignal(ctx *Context, node *Node, source NodeI
case Unlocking: case Unlocking:
} }
} else { } else {
ctx.Log.Logf("lockable", "Got mapped error %s, but %s isn't a requirement", signal, info.NodeID) ctx.Log.Logf("lockable", "Got mapped error %s, but %s isn't a requirement", signal, info.Destination)
} }
} }
@ -143,16 +143,16 @@ func (ext *LockableExt) HandleSuccessSignal(ctx *Context, node *Node, source Nod
info, info_found := node.ProcessResponse(ext.WaitInfos, signal) info, info_found := node.ProcessResponse(ext.WaitInfos, signal)
if info_found == true { if info_found == true {
state, found := ext.Requirements[info.NodeID] state, found := ext.Requirements[info.Destination]
if found == false { if found == false {
ctx.Log.Logf("lockable", "Got success signal for requirement that is no longer in the map(%s), ignoring...", info.NodeID) ctx.Log.Logf("lockable", "Got success signal for requirement that is no longer in the map(%s), ignoring...", info.Destination)
} else { } else {
ctx.Log.Logf("lockable", "got mapped response %+v for %+v in state %s", signal, info, ReqStateStrings[state]) ctx.Log.Logf("lockable", "got mapped response %+v for %+v in state %s", signal, info, ReqStateStrings[state])
switch state { switch state {
case Locking: case Locking:
switch ext.State { switch ext.State {
case Locking: case Locking:
ext.Requirements[info.NodeID] = Locked ext.Requirements[info.Destination] = Locked
locked := 0 locked := 0
for _, s := range(ext.Requirements) { for _, s := range(ext.Requirements) {
if s == Locked { if s == Locked {
@ -170,13 +170,13 @@ func (ext *LockableExt) HandleSuccessSignal(ctx *Context, node *Node, source Nod
ctx.Log.Logf("lockable", "PARTIAL LOCK: %s - %d/%d", node.ID, locked, len(ext.Requirements)) ctx.Log.Logf("lockable", "PARTIAL LOCK: %s - %d/%d", node.ID, locked, len(ext.Requirements))
} }
case AbortingLock: case AbortingLock:
ext.Requirements[info.NodeID] = Unlocking ext.Requirements[info.Destination] = Unlocking
lock_signal := NewLockSignal("unlock") lock_signal := NewLockSignal("unlock")
ext.WaitInfos[lock_signal.Id] = node.QueueTimeout(info.NodeID, lock_signal, 100*time.Millisecond) ext.WaitInfos[lock_signal.Id] = node.QueueTimeout("unlock", info.Destination, lock_signal, 100*time.Millisecond)
messages = messages.Add(ctx, info.NodeID, node, nil, lock_signal) messages = messages.Add(ctx, info.Destination, node, nil, lock_signal)
ctx.Log.Logf("lockable", "sending abort_lock to %s for %s", info.NodeID, node.ID) ctx.Log.Logf("lockable", "sending abort_lock to %s for %s", info.Destination, node.ID)
} }
case AbortingLock: case AbortingLock:
ctx.Log.Logf("lockable", "Got success signal in AbortingLock %s", node.ID) ctx.Log.Logf("lockable", "Got success signal in AbortingLock %s", node.ID)
@ -247,7 +247,7 @@ func (ext *LockableExt) HandleLockSignal(ctx *Context, node *Node, source NodeID
} }
lock_signal := NewLockSignal("lock") lock_signal := NewLockSignal("lock")
ext.WaitInfos[lock_signal.Id] = node.QueueTimeout(id, lock_signal, 5000*time.Millisecond) ext.WaitInfos[lock_signal.Id] = node.QueueTimeout("lock", id, lock_signal, 5000*time.Millisecond)
ext.Requirements[id] = Locking ext.Requirements[id] = Locking
messages = messages.Add(ctx, id, node, nil, lock_signal) messages = messages.Add(ctx, id, node, nil, lock_signal)
@ -279,7 +279,7 @@ func (ext *LockableExt) HandleLockSignal(ctx *Context, node *Node, source NodeID
} }
lock_signal := NewLockSignal("unlock") lock_signal := NewLockSignal("unlock")
ext.WaitInfos[lock_signal.Id] = node.QueueTimeout(id, lock_signal, 100*time.Millisecond) ext.WaitInfos[lock_signal.Id] = node.QueueTimeout("unlock", id, lock_signal, 100*time.Millisecond)
ext.Requirements[id] = Unlocking ext.Requirements[id] = Unlocking
messages = messages.Add(ctx, id, node, nil, lock_signal) messages = messages.Add(ctx, id, node, nil, lock_signal)
@ -300,18 +300,18 @@ func (ext *LockableExt) HandleTimeoutSignal(ctx *Context, node *Node, source Nod
wait_info, found := node.ProcessResponse(ext.WaitInfos, signal) wait_info, found := node.ProcessResponse(ext.WaitInfos, signal)
if found == true { if found == true {
state, found := ext.Requirements[wait_info.NodeID] state, found := ext.Requirements[wait_info.Destination]
if found == true { if found == true {
ctx.Log.Logf("lockable", "%s timed out %s", wait_info.NodeID, ReqStateStrings[state]) ctx.Log.Logf("lockable", "%s timed out %s", wait_info.Destination, ReqStateStrings[state])
switch state { switch state {
case Locking: case Locking:
ext.State = AbortingLock ext.State = AbortingLock
ext.Requirements[wait_info.NodeID] = Unlocked ext.Requirements[wait_info.Destination] = Unlocked
for id, state := range(ext.Requirements) { for id, state := range(ext.Requirements) {
if state == Locked { if state == Locked {
ext.Requirements[id] = Unlocking ext.Requirements[id] = Unlocking
lock_signal := NewLockSignal("unlock") lock_signal := NewLockSignal("unlock")
ext.WaitInfos[lock_signal.Id] = node.QueueTimeout(id, lock_signal, 100*time.Millisecond) ext.WaitInfos[lock_signal.Id] = node.QueueTimeout("unlock", id, lock_signal, 100*time.Millisecond)
messages = messages.Add(ctx, id, node, nil, lock_signal) messages = messages.Add(ctx, id, node, nil, lock_signal)
ctx.Log.Logf("lockable", "sent abort unlock to %s from %s", id, node.ID) ctx.Log.Logf("lockable", "sent abort unlock to %s from %s", id, node.ID)
} }
@ -319,7 +319,7 @@ func (ext *LockableExt) HandleTimeoutSignal(ctx *Context, node *Node, source Nod
case Unlocking: case Unlocking:
} }
} else { } else {
ctx.Log.Logf("lockable", "%s timed out", wait_info.NodeID) ctx.Log.Logf("lockable", "%s timed out", wait_info.Destination)
} }
} }

21
node.go
Unescape Escape View File

@ -158,8 +158,9 @@ func (node *Node) Allows(ctx *Context, principal_id NodeID, action Tree)(map[uui
} }
type WaitInfo struct { type WaitInfo struct {
NodeID NodeID `gv:"node"` Destination NodeID `gv:"destination"`
Timeout uuid.UUID `gv:"timeout"` Timeout uuid.UUID `gv:"timeout"`
Reason string `gv:"reason"`
} }
type WaitMap map[uuid.UUID]WaitInfo type WaitMap map[uuid.UUID]WaitInfo
@ -178,14 +179,28 @@ func (node *Node) ProcessResponse(wait_map WaitMap, response ResponseSignal) (Wa
return WaitInfo{}, false return WaitInfo{}, false
} }
func (node *Node) NewTimeout(reason string, dest NodeID, timeout time.Duration) (WaitInfo, uuid.UUID) {
id := uuid.New()
timeout_signal := NewTimeoutSignal(id)
node.QueueSignal(time.Now().Add(timeout), timeout_signal)
return WaitInfo{
Destination: dest,
Timeout: timeout_signal.Id,
Reason: reason,
}, id
}
// Creates a timeout signal for signal, queues it for the node at the timeout, and returns the WaitInfo // Creates a timeout signal for signal, queues it for the node at the timeout, and returns the WaitInfo
func (node *Node) QueueTimeout(dest NodeID, signal Signal, timeout time.Duration) WaitInfo { func (node *Node) QueueTimeout(reason string, dest NodeID, signal Signal, timeout time.Duration) WaitInfo {
timeout_signal := NewTimeoutSignal(signal.ID()) timeout_signal := NewTimeoutSignal(signal.ID())
node.QueueSignal(time.Now().Add(timeout), timeout_signal) node.QueueSignal(time.Now().Add(timeout), timeout_signal)
return WaitInfo{ return WaitInfo{
NodeID: dest, Destination: dest,
Timeout: timeout_signal.Id, Timeout: timeout_signal.Id,
Reason: reason,
} }
} }