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
|
// 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 (
"context"
"log"
"runtime"
"codeberg.org/gruf/go-runners"
"github.com/superseriousbusiness/gotosocial/internal/config"
"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
// ClientAPI provides a worker pool that handles both
// incoming client actions, and our own side-effects.
ClientAPI runners.WorkerPool
// Federator provides a worker pool that handles both
// incoming federated actions, and our own side-effects.
Federator runners.WorkerPool
// Enqueue functions for clientAPI / federator worker pools,
// these are pointers to Processor{}.Enqueue___() msg functions.
// This prevents dependency cycling as Processor depends on Workers.
EnqueueClientAPI func(context.Context, ...messages.FromClientAPI)
EnqueueFediAPI func(context.Context, ...messages.FromFediAPI)
// Blocking processing functions for clientAPI / federator.
// These are pointers to Processor{}.Process___() msg functions.
// This prevents dependency cycling as Processor depends on Workers.
//
// Rather than queueing messages for asynchronous processing, these
// functions will process immediately and in a blocking manner, and
// will not use up a worker slot.
//
// As such, you should only call them in special cases where something
// synchronous needs to happen before you can do something else.
ProcessFromClientAPI func(context.Context, messages.FromClientAPI) error
ProcessFromFediAPI func(context.Context, messages.FromFediAPI) error
// Media manager worker pools.
Media runners.WorkerPool
// prevent pass-by-value.
_ nocopy
}
// Start will start all of the contained
// worker pools (and global scheduler).
func (w *Workers) Start() {
// Get currently set GOMAXPROCS.
maxprocs := runtime.GOMAXPROCS(0)
tryUntil("starting scheduler", 5, w.Scheduler.Start)
tryUntil("start delivery workerpool", 5, func() bool {
n := config.GetAdvancedSenderMultiplier()
if n < 1 {
// clamp min senders to 1.
return w.Delivery.Start(1)
}
return w.Delivery.Start(n * maxprocs)
})
tryUntil("starting client API workerpool", 5, func() bool {
return w.ClientAPI.Start(4*maxprocs, 400*maxprocs)
})
tryUntil("starting federator workerpool", 5, func() bool {
return w.Federator.Start(4*maxprocs, 400*maxprocs)
})
tryUntil("starting media workerpool", 5, func() bool {
return w.Media.Start(8*maxprocs, 80*maxprocs)
})
}
// Stop will stop all of the contained worker pools (and global scheduler).
func (w *Workers) Stop() {
tryUntil("stopping scheduler", 5, w.Scheduler.Stop)
tryUntil("stopping delivery workerpool", 5, w.Delivery.Stop)
tryUntil("stopping client API workerpool", 5, w.ClientAPI.Stop)
tryUntil("stopping federator workerpool", 5, w.Federator.Stop)
tryUntil("stopping media workerpool", 5, w.Media.Stop)
}
// nocopy when embedded will signal linter to
// error on pass-by-value of parent struct.
type nocopy struct{}
func (*nocopy) Lock() {}
func (*nocopy) Unlock() {}
// tryUntil will attempt to call 'do' for 'count' attempts, before panicking with 'msg'.
func tryUntil(msg string, count int, do func() bool) {
for i := 0; i < count; i++ {
if do() {
return
}
}
log.Panicf("failed %s after %d tries", msg, count)
}
|
