P1.1 groundwork: per-queue mutex в Queue + анализ блокеров
Добавлены поле mu (не сериализуется) и LockMu/RLockMu-аксессоры в models.Queue; исправлено копирование Queue by value (numberOfHiddenMessagesInQueue). Глобальный лок пока используется везде — поведение не менялось. Задокументированы блокеры полного перехода (DLQ-указатель, мутирующий marshalQueueMeta, релинк DLQ после загрузки, атомарность перевода).
This commit is contained in:
@@ -91,7 +91,7 @@ func GetQueueAttributesV1(req *http.Request) (int, interfaces.AbstractResponseBo
|
||||
queueAttributes = append(queueAttributes, models.Attribute{Name: "ApproximateNumberOfMessages", Value: strconv.Itoa(len(queue.Messages))})
|
||||
}
|
||||
if shouldInclude("ApproximateNumberOfMessagesNotVisible") {
|
||||
queueAttributes = append(queueAttributes, models.Attribute{Name: "ApproximateNumberOfMessagesNotVisible", Value: strconv.Itoa(numberOfHiddenMessagesInQueue(*queue))})
|
||||
queueAttributes = append(queueAttributes, models.Attribute{Name: "ApproximateNumberOfMessagesNotVisible", Value: strconv.Itoa(numberOfHiddenMessagesInQueue(queue))})
|
||||
}
|
||||
if shouldInclude("CreatedTimestamp") {
|
||||
queueAttributes = append(queueAttributes, models.Attribute{Name: "CreatedTimestamp", Value: "0000000000"})
|
||||
|
||||
+1
-1
@@ -136,7 +136,7 @@ func processQueueTick(key string, queue *models.Queue) {
|
||||
}
|
||||
}
|
||||
|
||||
func numberOfHiddenMessagesInQueue(queue models.Queue) int {
|
||||
func numberOfHiddenMessagesInQueue(queue *models.Queue) int {
|
||||
num := 0
|
||||
for _, m := range queue.Messages {
|
||||
if m.ReceiptHandle != "" || m.DelaySecs > 0 && time.Now().Before(m.SentTime.Add(time.Duration(m.DelaySecs)*time.Second)) {
|
||||
|
||||
+14
-1
@@ -2,6 +2,7 @@ package models
|
||||
|
||||
import (
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
@@ -46,6 +47,11 @@ func (m *SqsMessage) IsReadyForReceipt() bool {
|
||||
}
|
||||
|
||||
type Queue struct {
|
||||
// mu — per-queue блокировка содержимого очереди (Messages, Duplicates, FIFO-
|
||||
// состояние, LastPurgeTime). Заменяет удержание глобального SyncQueues.Lock
|
||||
// на время операций с одной очередью. sync.RWMutex сериализации не подлежит.
|
||||
mu sync.RWMutex `json:"-"`
|
||||
|
||||
Name string
|
||||
URL string
|
||||
Arn string
|
||||
@@ -53,7 +59,7 @@ type Queue struct {
|
||||
ReceiveMessageWaitTimeSeconds int
|
||||
DelaySeconds int
|
||||
MaximumMessageSize int
|
||||
MessageRetentionPeriod int // seconds // TODO - not used in the code yet
|
||||
MessageRetentionPeriod int // seconds (реализовано в processQueueTick)
|
||||
Messages []SqsMessage
|
||||
DeadLetterQueue *Queue
|
||||
MaxReceiveCount int
|
||||
@@ -69,6 +75,13 @@ type Queue struct {
|
||||
LastPurgeTime time.Time
|
||||
}
|
||||
|
||||
// LockMu/UnlockMu/RLockMu/RUnlockMu — доступ к per-queue блокировке извне пакета models
|
||||
// (поле mu неэкспортируемо).
|
||||
func (q *Queue) LockMu() { q.mu.Lock() }
|
||||
func (q *Queue) UnlockMu() { q.mu.Unlock() }
|
||||
func (q *Queue) RLockMu() { q.mu.RLock() }
|
||||
func (q *Queue) RUnlockMu() { q.mu.RUnlock() }
|
||||
|
||||
// NextSequenceNumber — возвращает следующий порядковый номер сообщения FIFO-группы.
|
||||
//
|
||||
// Логическая схема: каждый вызов инкрементирует счётчик группы и возвращает
|
||||
|
||||
Reference in New Issue
Block a user