blob: 97cc945c918e0d132b363753066b31fa461363bd (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
|
package pool
import (
"context"
"runtime"
"time"
"github.com/spiral/roadrunner/v2/interfaces/worker"
"github.com/spiral/roadrunner/v2/pkg/payload"
)
// Pool managed set of inner worker processes.
type Pool interface {
// GetConfig returns pool configuration.
GetConfig() interface{}
// Exec executes task with payload
Exec(rqs payload.Payload) (payload.Payload, error)
// ExecWithContext executes task with context which is used with timeout
ExecWithContext(ctx context.Context, rqs payload.Payload) (payload.Payload, error)
// Workers returns worker list associated with the pool.
Workers() (workers []worker.BaseProcess)
// Remove worker from the pool.
RemoveWorker(worker worker.BaseProcess) error
// Destroy all underlying stack (but let them to complete the task).
Destroy(ctx context.Context)
}
// Configures the pool behaviour.
type Config struct {
// Debug flag creates new fresh worker before every request.
Debug bool
// NumWorkers defines how many sub-processes can be run at once. This value
// might be doubled by Swapper while hot-swap. Defaults to number of CPU cores.
NumWorkers int64
// MaxJobs defines how many executions is allowed for the worker until
// it's destruction. set 1 to create new process for each new task, 0 to let
// worker handle as many tasks as it can.
MaxJobs int64
// AllocateTimeout defines for how long pool will be waiting for a worker to
// be freed to handle the task. Defaults to 60s.
AllocateTimeout time.Duration
// DestroyTimeout defines for how long pool should be waiting for worker to
// properly destroy, if timeout reached worker will be killed. Defaults to 60s.
DestroyTimeout time.Duration
// Supervision config to limit worker and pool memory usage.
Supervisor *SupervisorConfig
}
// InitDefaults enables default config values.
func (cfg *Config) InitDefaults() {
if cfg.NumWorkers == 0 {
cfg.NumWorkers = int64(runtime.NumCPU())
}
if cfg.AllocateTimeout == 0 {
cfg.AllocateTimeout = time.Minute
}
if cfg.DestroyTimeout == 0 {
cfg.DestroyTimeout = time.Minute
}
if cfg.Supervisor == nil {
return
}
cfg.Supervisor.InitDefaults()
}
type SupervisorConfig struct {
// WatchTick defines how often to check the state of worker.
WatchTick uint64
// TTL defines maximum time worker is allowed to live.
TTL uint64
// IdleTTL defines maximum duration worker can spend in idle mode. Disabled when 0.
IdleTTL uint64
// ExecTTL defines maximum lifetime per job.
ExecTTL uint64
// MaxWorkerMemory limits memory per worker.
MaxWorkerMemory uint64
}
// InitDefaults enables default config values.
func (cfg *SupervisorConfig) InitDefaults() {
if cfg.WatchTick == 0 {
cfg.WatchTick = 1
}
}
|
