Added stop committee runner queue consumption trigger #1436
Conversation
…rk-add-consume-queue-exit
…abs/ssv into alan/no-fork-add-consume-queue-exit
| var q queueContainer | ||
|
|
||
| // in case of any error try to call the ctx.cancel to prevent the ctx leak | ||
| defer func() { | ||
| var ok bool | ||
|
|
||
| v.mtx.Lock() | ||
| q, ok = v.Queues[slot] | ||
| v.mtx.Unlock() | ||
|
|
There was a problem hiding this comment.
StartDuty is already locking, so this is potential "deadlock" or just double locking, although this happens in a goroutine and the lock is released in start duty. Still I think its better to just pass the Queue q from StartDuty as we're already using the mutex to get it there.
by passing it to this function you don't need to use the mutex here at all.
operator/validator/controller.go
Outdated
| for _, slotQueue := range vc.Queues { | ||
| slotQueue.StopQueueF() | ||
| } |
There was a problem hiding this comment.
you can't do this here, accessing Queues requires mutex acquiring. best is to have a function inside the committee struct that handles this.
protocol/v2/ssv/validator/events.go
Outdated
|
|
||
| if err := dutyRunner.GetBaseRunner().QBFTController.OnTimeout(logger, *eventMsg); err != nil { | ||
| return fmt.Errorf("timeout event: %w", err) | ||
| } | ||
|
|
||
| c.Queues[slot].StopQueueF() |
There was a problem hiding this comment.
you need to lock the mutex to access Queues. the best will be to get it same time you're getting the runner at the beginning of function.
…rk-add-consume-queue-exit
Co-authored-by: Matus Kysel <[email protected]>
…consume-queue-exit
…abs/ssv into alan/no-fork-add-consume-queue-exit
| err := c.ConsumeQueue(logger, duty.Slot, c.ProcessMessage, r) | ||
| // Setting the cancel function separately due the queue could be created in HandleMessage | ||
|
|
||
| logger = c.logger.With(fields.DutyID(fields.FormatCommitteeDutyID(c.Operator.Committee, c.BeaconNetwork.EstimatedEpochAtSlot(duty.Slot), duty.Slot)), fields.Slot(duty.Slot)) |
There was a problem hiding this comment.
this should be in the other PR for committee ID no?
| if err := dutyRunner.GetBaseRunner().QBFTController.OnTimeout(logger, *eventMsg); err != nil { | ||
| return fmt.Errorf("timeout event: %w", err) | ||
| } | ||
|
|
||
| q.StopQueueF() |
There was a problem hiding this comment.
why are we stopping the queue on timeout message?
|
Replaced with PR #1556 based on the |
No description provided.