becca.monster
Assemble a crew of worker processes to tackle your computational tasks.
:fire: remove load balancing for now
author
yoshie
date
(Sep 17, 2025, 8:41 PM +0200)
commit
c9063f47
c9063f47a4cfa0d676a88ea2daa85fc7518c89c8
parent
19805b50
19805b508b66b3ec4def942c408f1a3cb31da5a7
+32
-138
+1
-1
examples/src/backpressure.gleam
+1
-1
examples/src/backpressure.gleam
+31
-115
src/crew.gleam
+31
-115
src/crew.gleam
···
60
60
name: Name(PoolMsg(work, result)),
61
61
size: Int,
62
62
max_queue_length: Option(Int),
63
-
max_overflow_reductions: Option(Int),
64
63
init: fn() -> state,
65
64
init_timeout: Int,
66
65
work: fn(state, work) -> result,
···
116
115
run work: fn(state, work) -> result,
117
116
) -> Builder(state, work, result) {
118
117
let size = scheduler_count()
119
-
Builder(
120
-
name:,
121
-
size:,
122
-
max_queue_length: None,
123
-
max_overflow_reductions: None,
124
-
init_timeout:,
125
-
init:,
126
-
work:,
127
-
)
118
+
Builder(name:, size:, max_queue_length: None, init_timeout:, init:, work:)
128
119
}
129
120
130
121
/// Set the number of worker processes in the pool to a fixed number.
···
162
153
max_queue_length: Int,
163
154
) -> Builder(state, work, result) {
164
155
Builder(..builder, max_queue_length: Some(max_queue_length))
165
-
}
166
-
167
-
/// The `max_overflow_reductions` setting controls how many work items a single
168
-
/// request can submit before being deprioritised and being moved back to the
169
-
/// end of the queue.
170
-
///
171
-
/// This allows other requests to have a chance at getting completed,
172
-
/// distributing the load and making the entire system better behaved.
173
-
/// By default, `max_overflow_reductions` is equal to the queue length.
174
-
///
175
-
/// ## Example
176
-
///
177
-
/// ```gleam
178
-
/// crew.new(pool_name, worker)
179
-
/// |> crew.max_overflow_reductions(100)
180
-
/// ```
181
-
pub fn max_overflow_reductions(
182
-
builder: Builder(state, work, result),
183
-
max_overflow_reductions: Int,
184
-
) -> Builder(state, work, result) {
185
-
Builder(..builder, max_overflow_reductions: Some(max_overflow_reductions))
186
156
}
187
157
188
158
// -- START -------------------------------------------------------------------
···
492
462
channels: Dict(Receiver(result), Channel(work, result)),
493
463
//
494
464
queue: Deque(Work(work, result)),
495
-
overflow_channel_queue: Deque(Receiver(result)),
496
-
max_overflow_reductions: Int,
465
+
overflow_queue: Deque(Request(work, result)),
497
466
)
498
467
}
499
468
···
510
479
}
511
480
512
481
type Channel(work, result) {
513
-
Channel(
514
-
from: Pid,
515
-
workers: Set(Pid),
482
+
Channel(from: Pid, workers: Set(Pid), receive: Receiver(result))
483
+
}
484
+
485
+
type Request(work, result) {
486
+
Request(
487
+
caller: Pid,
516
488
receive: Receiver(result),
517
-
overflow_queue: Deque(Request(work)),
518
-
reductions: Int,
489
+
work: List(work),
490
+
enqueued: Option(Subject(Nil)),
519
491
)
520
-
}
521
-
522
-
type Request(work) {
523
-
Request(work: List(work), enqueued: Option(Subject(Nil)))
524
492
}
525
493
526
494
fn init_pool(
···
537
505
builder.max_queue_length
538
506
|> option.unwrap(builder.size * 100)
539
507
540
-
let max_overflow_reductions =
541
-
builder.max_overflow_reductions
542
-
|> option.unwrap(max_queue_length)
543
-
544
508
set_counter(builder.name, max_queue_length)
545
509
546
510
let state =
···
552
516
callers: dict.new(),
553
517
channels: dict.new(),
554
518
queue: deque.new(),
555
-
overflow_channel_queue: deque.new(),
556
-
max_overflow_reductions:,
519
+
overflow_queue: deque.new(),
557
520
)
558
521
559
522
actor.initialised(state)
···
662
625
// get the channel data or construct a new entry if needed.
663
626
let channel = case dict.get(state.channels, receive) {
664
627
Ok(channel) -> channel
665
-
Error(_) ->
666
-
Channel(
667
-
from: pid,
668
-
workers: set.new(),
669
-
receive:,
670
-
overflow_queue: deque.new(),
671
-
reductions: 0,
672
-
)
628
+
Error(_) -> Channel(from: pid, workers: set.new(), receive:)
673
629
}
674
630
675
631
// get the caller data and insert the new channel, starting to monitor if needed.
···
729
685
work: List(work),
730
686
enqueued: Option(Subject(Nil)),
731
687
) -> State(work, result) {
732
-
let Channel(from: caller, receive:, workers:, ..) = channel
688
+
let Channel(from: caller, receive:, workers:) = channel
733
689
734
690
case work, state.idle_workers {
735
691
[work, ..rest], [worker, ..idle_workers] -> {
···
763
719
Some(enqueued) -> process.send(enqueued, Nil)
764
720
None -> Nil
765
721
}
766
-
channel.overflow_queue
767
-
}
768
722
769
-
work ->
770
-
deque.push_back(channel.overflow_queue, Request(work:, enqueued:))
771
-
}
723
+
state.overflow_queue
724
+
}
772
725
773
-
// if we previously had no overflow, but now we do, we need to push
774
-
// this channel onto the overflow queue
775
-
let overflow_channel_queue = case
776
-
deque.is_empty(channel.overflow_queue),
777
-
deque.is_empty(overflow_queue)
778
-
{
779
-
True, False -> deque.push_back(state.overflow_channel_queue, receive)
780
-
_, _ -> state.overflow_channel_queue
726
+
work -> {
727
+
let request = Request(caller:, receive:, work:, enqueued:)
728
+
deque.push_back(state.overflow_queue, request)
729
+
}
781
730
}
782
731
783
-
let channel = Channel(..channel, overflow_queue:)
784
732
let channels = dict.insert(state.channels, receive, channel)
785
733
786
-
State(..state, queue:, channels:, overflow_channel_queue:)
734
+
State(..state, queue:, channels:, overflow_queue:)
787
735
}
788
736
789
737
[], _ -> {
···
847
795
) -> Result(#(Work(work, result), State(work, result)), Nil) {
848
796
use #(work, queue) <- result.try(deque.pop_front(state.queue))
849
797
850
-
use #(overflow_channel, overflow_channel_queue) <- try(
851
-
deque.pop_front(state.overflow_channel_queue),
798
+
use #(Request(caller:, receive:, work: request, enqueued:), overflow_queue) <- try(
799
+
deque.pop_front(state.overflow_queue),
852
800
fn(_) {
853
801
// overflow queue is empty, the queue got smaller and we can increase the capacity
854
802
increment_counter(state.name, 1)
···
856
804
},
857
805
)
858
806
859
-
use overflow_channel <- try(dict.get(state.channels, overflow_channel), fn(_) {
860
-
// that channel got closed, try again
861
-
pop_work(State(..state, overflow_channel_queue:))
862
-
})
863
-
864
-
let Channel(from: caller, receive:, reductions:, overflow_queue:, ..) =
865
-
overflow_channel
866
-
867
-
case deque.pop_front(overflow_queue) {
868
-
Ok(#(Request(work: [first, ..rest], enqueued:), overflow_queue)) -> {
807
+
case request {
808
+
[first, ..rest] -> {
869
809
// we can push one overflow item onto the queue as a replacement for the
870
810
// item that just got popped off.
871
811
let queue = deque.push_back(queue, Work(work: first, caller:, receive:))
872
-
let reductions = reductions + 1
873
812
874
813
// push the request back if there is still work to do.
875
814
// if there is none, send the enqueued signal.
876
815
let overflow_queue = case rest {
877
-
[_, ..] ->
878
-
deque.push_front(overflow_queue, Request(work: rest, enqueued:))
816
+
[_, ..] -> {
817
+
let request = Request(caller:, receive:, work: rest, enqueued:)
818
+
deque.push_front(overflow_queue, request)
819
+
}
879
820
880
821
[] -> {
881
822
case enqueued {
···
887
828
}
888
829
}
889
830
890
-
// update the channel with the updated queue and reductions count.
891
-
let overflow_channel =
892
-
Channel(..overflow_channel, overflow_queue:, reductions:)
893
-
let channels = dict.insert(state.channels, receive, overflow_channel)
894
-
895
-
// push the channel back if there still work to do, either as the first
896
-
// or last depending on reductions count.
897
-
let overflow_channel_queue = case
898
-
deque.is_empty(overflow_queue),
899
-
reductions % state.max_overflow_reductions == 0
900
-
{
901
-
True, _ -> overflow_channel_queue
902
-
False, False -> deque.push_front(overflow_channel_queue, receive)
903
-
False, True -> deque.push_back(overflow_channel_queue, receive)
904
-
}
905
-
906
-
Ok(#(work, State(..state, queue:, channels:, overflow_channel_queue:)))
831
+
Ok(#(work, State(..state, queue:, overflow_queue:)))
907
832
}
908
833
909
-
Ok(#(Request(work: [], enqueued:), overflow_queue)) -> {
834
+
[] -> {
910
835
// got an empty work request - this should not happen but I guess we can handle it
911
836
case enqueued {
912
837
Some(enqueued) -> process.send(enqueued, Nil)
913
838
None -> Nil
914
839
}
915
840
916
-
// remove this leftover request from the channel and then try again
917
-
let overflow_channel = Channel(..overflow_channel, overflow_queue:)
918
-
let channels = dict.insert(state.channels, receive, overflow_channel)
919
-
pop_work(State(..state, channels:))
920
-
}
921
-
922
-
Error(_) -> {
923
-
// no work left in this overflow channel, drop it from the overflow queue
924
-
// and try with the next one
925
-
pop_work(State(..state, overflow_channel_queue:))
841
+
pop_work(State(..state, overflow_queue:))
926
842
}
927
843
}
928
844
}
-22
src/crew/task_pool.gleam
-22
src/crew/task_pool.gleam
···
72
72
Builder(crew.max_queue_length(builder, max_queue_length))
73
73
}
74
74
75
-
/// The `max_overflow_reductions` setting controls how many work items a single
76
-
/// request can submit before being deprioritised and being moved back to the
77
-
/// end of the queue.
78
-
///
79
-
/// This allows other requests to have a chance at getting completed,
80
-
/// distributing the load and making the entire system better behaved.
81
-
/// By default, `max_overflow_reductions` is equal to the queue length.
82
-
///
83
-
/// ## Example
84
-
///
85
-
/// ```gleam
86
-
/// task_pool.new(pool_name)
87
-
/// |> task_pool.max_overflow_reductions(100)
88
-
/// ```
89
-
pub fn max_overflow_reductions(
90
-
builder: Builder,
91
-
max_overflow_reductions: Int,
92
-
) -> Builder {
93
-
let Builder(builder) = builder
94
-
Builder(crew.max_overflow_reductions(builder, max_overflow_reductions))
95
-
}
96
-
97
75
/// Start an unsupervised worker pool from the given builder.
98
76
///
99
77
/// Returns a supervisor that manages the pool and its workers. In most cases,