Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
spawn blocking instead of block_in_place
+36
-9
+36
-9
constellation/src/server/mod.rs
+36
-9
constellation/src/server/mod.rs
···
14
14
use std::collections::{HashMap, HashSet};
15
15
use std::time::{Duration, UNIX_EPOCH};
16
16
use tokio::net::{TcpListener, ToSocketAddrs};
17
-
use tokio::task::block_in_place;
17
+
use tokio::task::spawn_blocking;
18
18
use tokio_util::sync::CancellationToken;
19
19
20
20
use crate::storage::{LinkReader, StorageStats};
···
34
34
35
35
const INDEX_BEGAN_AT_TS: u64 = 1738083600; // TODO: not this
36
36
37
+
fn to500(e: tokio::task::JoinError) -> http::StatusCode {
38
+
eprintln!("handler join error: {e}");
39
+
http::StatusCode::INTERNAL_SERVER_ERROR
40
+
}
41
+
37
42
pub async fn serve<S, A>(store: S, addr: A, stay_alive: CancellationToken) -> anyhow::Result<()>
38
43
where
39
44
S: LinkReader,
···
45
50
"/",
46
51
get({
47
52
let store = store.clone();
48
-
move |accept| async { block_in_place(|| hello(accept, store)) }
53
+
move |accept| async {
54
+
spawn_blocking(|| hello(accept, store))
55
+
.await
56
+
.map_err(to500)?
57
+
}
49
58
}),
50
59
)
51
60
.route(
52
61
"/links/count",
53
62
get({
54
63
let store = store.clone();
55
-
move |accept, query| async { block_in_place(|| count_links(accept, query, store)) }
64
+
move |accept, query| async {
65
+
spawn_blocking(|| count_links(accept, query, store))
66
+
.await
67
+
.map_err(to500)?
68
+
}
56
69
}),
57
70
)
58
71
.route(
···
60
73
get({
61
74
let store = store.clone();
62
75
move |accept, query| async {
63
-
block_in_place(|| count_distinct_dids(accept, query, store))
76
+
spawn_blocking(|| count_distinct_dids(accept, query, store))
77
+
.await
78
+
.map_err(to500)?
64
79
}
65
80
}),
66
81
)
···
69
84
get({
70
85
let store = store.clone();
71
86
move |accept, query| async {
72
-
block_in_place(|| get_backlinks(accept, query, store))
87
+
spawn_blocking(|| get_backlinks(accept, query, store))
88
+
.await
89
+
.map_err(to500)?
73
90
}
74
91
}),
75
92
)
···
77
94
"/links",
78
95
get({
79
96
let store = store.clone();
80
-
move |accept, query| async { block_in_place(|| get_links(accept, query, store)) }
97
+
move |accept, query| async {
98
+
spawn_blocking(|| get_links(accept, query, store))
99
+
.await
100
+
.map_err(to500)?
101
+
}
81
102
}),
82
103
)
83
104
.route(
···
85
106
get({
86
107
let store = store.clone();
87
108
move |accept, query| async {
88
-
block_in_place(|| get_distinct_dids(accept, query, store))
109
+
spawn_blocking(|| get_distinct_dids(accept, query, store))
110
+
.await
111
+
.map_err(to500)?
89
112
}
90
113
}),
91
114
)
···
95
118
get({
96
119
let store = store.clone();
97
120
move |accept, query| async {
98
-
block_in_place(|| count_all_links(accept, query, store))
121
+
spawn_blocking(|| count_all_links(accept, query, store))
122
+
.await
123
+
.map_err(to500)?
99
124
}
100
125
}),
101
126
)
···
104
129
get({
105
130
let store = store.clone();
106
131
move |accept, query| async {
107
-
block_in_place(|| explore_links(accept, query, store))
132
+
spawn_blocking(|| explore_links(accept, query, store))
133
+
.await
134
+
.map_err(to500)?
108
135
}
109
136
}),
110
137
)