diff --git a/pgqueue/manager/manager.go b/pgqueue/manager/manager.go index 60dfc89..78604c7 100644 --- a/pgqueue/manager/manager.go +++ b/pgqueue/manager/manager.go @@ -79,7 +79,7 @@ func New(ctx context.Context, pool pg.PoolConn, opts ...Opt) (*Manager, error) { // Register a maintenance ticker if _, err := self.RegisterTicker(bootstrapCtx, schema.DefaultMaintenanceTickerName, schema.TickerMeta{ - Interval: types.Ptr(schema.DefaultMaintenancePeriod), + Interval: types.Ptr(self.maintenancePeriod), }, self.maintenance); err != nil { endBootstrapSpan(err) return nil, err diff --git a/pgqueue/manager/opt.go b/pgqueue/manager/opt.go index 24b1755..0d4327f 100644 --- a/pgqueue/manager/opt.go +++ b/pgqueue/manager/opt.go @@ -2,6 +2,7 @@ package manager import ( "os" + "time" // Packages schema "github.com/mutablelogic/go-pg/pgqueue/schema" @@ -17,10 +18,14 @@ type Opt func(*opt) error // opt combines all configuration options for Manager. type opt struct { - worker string - schema string - tracer trace.Tracer - metrics metric.Meter + worker string + schema string + tracer trace.Tracer + metrics metric.Meter + partitionSize uint64 + partitionThreshold float64 + partitionAhead uint64 + maintenancePeriod time.Duration } /////////////////////////////////////////////////////////////////////////////// @@ -40,6 +45,10 @@ func (o *opt) apply(opts ...Opt) error { func (o *opt) defaults() error { o.schema = schema.DefaultSchema + o.partitionSize = schema.DefaultPartitionSize + o.partitionThreshold = schema.DefaultPartitionThreshold + o.partitionAhead = schema.DefaultPartitionAhead + o.maintenancePeriod = schema.DefaultMaintenancePeriod if hostname, err := os.Hostname(); err != nil { return err } else { @@ -86,3 +95,43 @@ func WithMeter(meter metric.Meter) Opt { return nil } } + +// WithPartitionSize sets the task partition size. Values less than 1 are ignored. +func WithPartitionSize(size uint64) Opt { + return func(o *opt) error { + if size > 0 { + o.partitionSize = size + } + return nil + } +} + +// WithPartitionThreshold sets the partition creation threshold in the range (0,1]. +func WithPartitionThreshold(threshold float64) Opt { + return func(o *opt) error { + if threshold > 0 && threshold <= 1 { + o.partitionThreshold = threshold + } + return nil + } +} + +// WithPartitionAhead sets how many partitions to create when threshold is reached. +func WithPartitionAhead(ahead uint64) Opt { + return func(o *opt) error { + if ahead > 0 { + o.partitionAhead = ahead + } + return nil + } +} + +// WithMaintenancePeriod sets how often the maintenance ticker runs. Values less than 1 second are ignored. +func WithMaintenancePeriod(period time.Duration) Opt { + return func(o *opt) error { + if period >= time.Second { + o.maintenancePeriod = period + } + return nil + } +} diff --git a/pgqueue/manager/partition.go b/pgqueue/manager/partition.go index 3c84f6f..cd01c8b 100644 --- a/pgqueue/manager/partition.go +++ b/pgqueue/manager/partition.go @@ -38,26 +38,41 @@ func (manager *Manager) CreateNextPartition(ctx context.Context) (result string, } } - // Create next partition if no partitions exist, or seq is within 80% of upper bound - if maxEnd == 0 || seq >= uint64(float64(maxEnd)*0.8) { - start := maxEnd - if start == 0 { - start = 1 + // Create partition(s) if no partitions exist, or sequence has crossed threshold. + if maxEnd == 0 || seq >= uint64(float64(maxEnd)*manager.partitionThreshold) { + count := manager.partitionAhead + if count == 0 { + count = 1 } - end := start + schema.DefaultPartitionSize - name := fmt.Sprintf("task_%08d_%08d", start, end) - if err := manager.CreatePartition(ctx, schema.PartitionMeta{ - Partition: name, - Start: start, - End: end, - }); err != nil { - return "", err + + created := make([]string, 0, count) + for i := uint64(0); i < count; i++ { + start := maxEnd + if start == 0 { + start = 1 + } + end := start + manager.partitionSize + name := fmt.Sprintf("task_%08d_%08d", start, end) + if err := manager.CreatePartition(ctx, schema.PartitionMeta{ + Partition: name, + Start: start, + End: end, + }); err != nil { + return "", err + } + + created = append(created, name) + maxEnd = end } - // Reset the connection + // Reset the connection to refresh partition metadata. manager.PoolConn.Reset() - return name, nil + if len(created) == 1 { + return created[0], nil + } + + return fmt.Sprintf("%s (+%d more)", created[0], len(created)-1), nil } return "", nil diff --git a/pgqueue/schema/schema.go b/pgqueue/schema/schema.go index 29791ac..b9ea22f 100644 --- a/pgqueue/schema/schema.go +++ b/pgqueue/schema/schema.go @@ -20,10 +20,12 @@ const ( QueueListLimit = 100 TickerListLimit = 100 DefaultPartitionSize = 100_000 // tasks per partition + DefaultPartitionThreshold = 0.5 // create partition(s) when sequence reaches this fraction of the highest partition end + DefaultPartitionAhead = 1 // number of partition(s) to create when threshold is reached DefaultMaintenanceTickerName = "$maintenance$" DefaultCleanupTickerName = "$cleanup$" DefaultTickerPeriod = 5 * time.Second // how often to look for matured tickers DefaultQueuePeriod = 10 * time.Second // how often to poll queues for retries and missed notifications DefaultCleanupPeriod = 15 * time.Minute // how often to delete expired tasks - DefaultMaintenancePeriod = time.Hour // create and drop partitions + DefaultMaintenancePeriod = 10 * time.Minute // create and drop partitions )