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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
// GoToSocial
// Copyright (C) GoToSocial Authors admin@gotosocial.org
// SPDX-License-Identifier: AGPL-3.0-or-later
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package workers
import (
"runtime"
"github.com/superseriousbusiness/gotosocial/internal/config"
"github.com/superseriousbusiness/gotosocial/internal/log"
"github.com/superseriousbusiness/gotosocial/internal/messages"
"github.com/superseriousbusiness/gotosocial/internal/scheduler"
"github.com/superseriousbusiness/gotosocial/internal/transport/delivery"
)
type Workers struct {
// Main task scheduler instance.
Scheduler scheduler.Scheduler
// Delivery provides a worker pool that
// handles outgoing ActivityPub deliveries.
// It contains an embedded (but accessible)
// indexed queue of Delivery{} objects.
Delivery delivery.WorkerPool
// Client provides a worker pool that handles
// incoming processing jobs from the client API.
Client MsgWorkerPool[*messages.FromClientAPI]
// Federator provides a worker pool that handles
// incoming processing jobs from the fedi API.
Federator MsgWorkerPool[*messages.FromFediAPI]
// Dereference provides a worker pool
// for asynchronous dereferencer jobs.
Dereference FnWorkerPool
// Processing provides a worker pool
// for asynchronous processing jobs,
// eg., import tasks, admin tasks.
Processing FnWorkerPool
// WebPush provides a worker pool for
// delivering Web Push notifications.
WebPush FnWorkerPool
// prevent pass-by-value.
_ nocopy
}
// StartScheduler starts the job scheduler.
func (w *Workers) StartScheduler() {
_ = w.Scheduler.Start()
// false = already running
log.Info(nil, "started scheduler")
}
// Start will start contained worker pools.
func (w *Workers) Start() {
var n int
maxprocs := runtime.GOMAXPROCS(0)
n = deliveryWorkers(maxprocs)
w.Delivery.Start(n)
log.Infof(nil, "started %d delivery workers", n)
n = 4 * maxprocs
w.Client.Start(n)
log.Infof(nil, "started %d client workers", n)
n = 4 * maxprocs
w.Federator.Start(n)
log.Infof(nil, "started %d federator workers", n)
n = 4 * maxprocs
w.Dereference.Start(n)
log.Infof(nil, "started %d dereference workers", n)
n = maxprocs
w.Processing.Start(n)
log.Infof(nil, "started %d processing workers", n)
n = maxprocs
w.WebPush.Start(n)
log.Infof(nil, "started %d Web Push workers", n)
}
// Stop will stop all of the contained
// worker pools (and global scheduler).
func (w *Workers) Stop() {
_ = w.Scheduler.Stop()
// false = not running
log.Info(nil, "stopped scheduler")
w.Delivery.Stop()
log.Info(nil, "stopped delivery workers")
w.Client.Stop()
log.Info(nil, "stopped client workers")
w.Federator.Stop()
log.Info(nil, "stopped federator workers")
w.Dereference.Stop()
log.Info(nil, "stopped dereference workers")
w.Processing.Stop()
log.Info(nil, "stopped processing workers")
w.WebPush.Stop()
log.Info(nil, "stopped WebPush workers")
}
// nocopy when embedded will signal linter to
// error on pass-by-value of parent struct.
type nocopy struct{}
func (*nocopy) Lock() {}
func (*nocopy) Unlock() {}
func deliveryWorkers(maxprocs int) int {
n := config.GetAdvancedSenderMultiplier()
if n < 1 {
// clamp to 1
return 1
}
return n * maxprocs
}
|
