Coverage Report

Created: 2025-05-30 16:37

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
/build/source/nativelink-scheduler/src/cache_lookup_scheduler.rs
Line
Count
Source
1
// Copyright 2024 The NativeLink Authors. All rights reserved.
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//    http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
// See the License for the specific language governing permissions and
13
// limitations under the License.
14
15
use std::collections::HashMap;
16
use std::sync::Arc;
17
18
use async_trait::async_trait;
19
use nativelink_error::{Code, Error, ResultExt, make_err};
20
use nativelink_metric::{MetricsComponent, RootMetricsComponent};
21
use nativelink_proto::build::bazel::remote::execution::v2::{
22
    ActionResult as ProtoActionResult, GetActionResultRequest,
23
};
24
use nativelink_store::ac_utils::get_and_decode_digest;
25
use nativelink_store::grpc_store::GrpcStore;
26
use nativelink_util::action_messages::{
27
    ActionInfo, ActionStage, ActionState, ActionUniqueKey, ActionUniqueQualifier, OperationId,
28
};
29
use nativelink_util::background_spawn;
30
use nativelink_util::common::DigestInfo;
31
use nativelink_util::digest_hasher::DigestHasherFunc;
32
use nativelink_util::known_platform_property_provider::KnownPlatformPropertyProvider;
33
use nativelink_util::operation_state_manager::{
34
    ActionStateResult, ActionStateResultStream, ClientStateManager, OperationFilter,
35
};
36
use nativelink_util::origin_event::OriginMetadata;
37
use nativelink_util::store_trait::Store;
38
use opentelemetry::baggage::BaggageExt;
39
use opentelemetry::context::Context;
40
use opentelemetry_semantic_conventions::attribute::ENDUSER_ID;
41
use parking_lot::{Mutex, MutexGuard};
42
use scopeguard::guard;
43
use tokio::sync::oneshot;
44
use tonic::{Request, Response};
45
use tracing::error;
46
47
/// Actions that are having their cache checked or failed cache lookup and are
48
/// being forwarded upstream.  Missing the `skip_cache_check` actions which are
49
/// forwarded directly.
50
type CheckActions = HashMap<
51
    ActionUniqueKey,
52
    Vec<(
53
        OperationId,
54
        oneshot::Sender<Result<Box<dyn ActionStateResult>, Error>>,
55
    )>,
56
>;
57
58
#[derive(MetricsComponent)]
59
pub struct CacheLookupScheduler {
60
    /// A reference to the AC to find existing actions in.
61
    /// To prevent unintended issues, this store should probably be a `CompletenessCheckingStore`.
62
    #[metric(group = "ac_store")]
63
    ac_store: Store,
64
    /// The "real" scheduler to use to perform actions if they were not found
65
    /// in the action cache.
66
    #[metric(group = "action_scheduler")]
67
    action_scheduler: Arc<dyn ClientStateManager>,
68
    /// Actions that are currently performing a `CacheCheck`.
69
    inflight_cache_checks: Arc<Mutex<CheckActions>>,
70
}
71
72
impl core::fmt::Debug for CacheLookupScheduler {
73
0
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
74
0
        f.debug_struct("CacheLookupScheduler")
75
0
            .field("ac_store", &self.ac_store)
76
0
            .finish_non_exhaustive()
77
0
    }
78
}
79
80
0
async fn get_action_from_store(
81
0
    ac_store: &Store,
82
0
    action_digest: DigestInfo,
83
0
    instance_name: String,
84
0
    digest_function: DigestHasherFunc,
85
0
) -> Result<ProtoActionResult, Error> {
86
    // If we are a GrpcStore we shortcut here, as this is a special store.
87
0
    if let Some(grpc_store) = ac_store.downcast_ref::<GrpcStore>(Some(action_digest.into())) {
  Branch (87:12): [True: 0, False: 0]
  Branch (87:12): [Folded - Ignored]
88
0
        let action_result_request = GetActionResultRequest {
89
0
            instance_name,
90
0
            action_digest: Some(action_digest.into()),
91
0
            inline_stdout: false,
92
0
            inline_stderr: false,
93
0
            inline_output_files: Vec::new(),
94
0
            digest_function: digest_function.proto_digest_func().into(),
95
0
        };
96
0
        grpc_store
97
0
            .get_action_result(Request::new(action_result_request))
98
0
            .await
99
0
            .map(Response::into_inner)
100
    } else {
101
0
        get_and_decode_digest::<ProtoActionResult>(ac_store, action_digest.into()).await
102
    }
103
0
}
104
105
/// Future for when `ActionStateResults` are known.
106
type ActionStateResultOneshot = oneshot::Receiver<Result<Box<dyn ActionStateResult>, Error>>;
107
108
0
fn subscribe_to_existing_action(
109
0
    inflight_cache_checks: &mut MutexGuard<CheckActions>,
110
0
    unique_qualifier: &ActionUniqueKey,
111
0
    client_operation_id: &OperationId,
112
0
) -> Option<ActionStateResultOneshot> {
113
0
    inflight_cache_checks
114
0
        .get_mut(unique_qualifier)
115
0
        .map(|oneshots| {
116
0
            let (tx, rx) = oneshot::channel();
117
0
            oneshots.push((client_operation_id.clone(), tx));
118
0
            rx
119
0
        })
120
0
}
121
122
struct CacheLookupActionStateResult {
123
    action_state: Arc<ActionState>,
124
    maybe_origin_metadata: Option<OriginMetadata>,
125
    change_called: bool,
126
}
127
128
#[async_trait]
129
impl ActionStateResult for CacheLookupActionStateResult {
130
0
    async fn as_state(&self) -> Result<(Arc<ActionState>, Option<OriginMetadata>), Error> {
131
0
        Ok((
132
0
            self.action_state.clone(),
133
0
            self.maybe_origin_metadata.clone(),
134
0
        ))
135
0
    }
136
137
0
    async fn changed(&mut self) -> Result<(Arc<ActionState>, Option<OriginMetadata>), Error> {
138
0
        if self.change_called {
  Branch (138:12): [True: 0, False: 0]
  Branch (138:12): [Folded - Ignored]
139
0
            return Err(make_err!(
140
0
                Code::Internal,
141
0
                "CacheLookupActionStateResult::changed called twice"
142
0
            ));
143
0
        }
144
0
        self.change_called = true;
145
0
        Ok((
146
0
            self.action_state.clone(),
147
0
            self.maybe_origin_metadata.clone(),
148
0
        ))
149
0
    }
150
151
0
    async fn as_action_info(&self) -> Result<(Arc<ActionInfo>, Option<OriginMetadata>), Error> {
152
        // TODO(aaronmondal) We should probably remove as_action_info()
153
        // or implement it properly.
154
0
        return Err(make_err!(
155
0
            Code::Unimplemented,
156
0
            "as_action_info not implemented for CacheLookupActionStateResult::as_action_info"
157
0
        ));
158
0
    }
159
}
160
161
impl CacheLookupScheduler {
162
2
    pub fn new(
163
2
        ac_store: Store,
164
2
        action_scheduler: Arc<dyn ClientStateManager>,
165
2
    ) -> Result<Self, Error> {
166
2
        Ok(Self {
167
2
            ac_store,
168
2
            action_scheduler,
169
2
            inflight_cache_checks: Arc::default(),
170
2
        })
171
2
    }
172
173
1
    async fn inner_add_action(
174
1
        &self,
175
1
        client_operation_id: OperationId,
176
1
        action_info: Arc<ActionInfo>,
177
1
    ) -> Result<Box<dyn ActionStateResult>, Error> {
178
1
        let 
unique_key0
= match &action_info.unique_qualifier {
179
0
            ActionUniqueQualifier::Cacheable(unique_key) => unique_key.clone(),
180
            ActionUniqueQualifier::Uncacheable(_) => {
181
                // Cache lookup skipped, forward to the upstream.
182
1
                return self
183
1
                    .action_scheduler
184
1
                    .add_action(client_operation_id, action_info)
185
1
                    .await;
186
            }
187
        };
188
189
0
        let cache_check_result = {
190
            // Check this isn't a duplicate request first.
191
0
            let mut inflight_cache_checks = self.inflight_cache_checks.lock();
192
0
            subscribe_to_existing_action(
193
0
                &mut inflight_cache_checks,
194
0
                &unique_key,
195
0
                &client_operation_id,
196
            )
197
0
            .ok_or_else(move || {
198
0
                let (action_listener_tx, action_listener_rx) = oneshot::channel();
199
0
                inflight_cache_checks.insert(
200
0
                    unique_key.clone(),
201
0
                    vec![(client_operation_id, action_listener_tx)],
202
0
                );
203
                // In the event we loose the reference to our `scope_guard`, it will remove
204
                // the action from the inflight_cache_checks map.
205
0
                let inflight_cache_checks = self.inflight_cache_checks.clone();
206
                (
207
0
                    action_listener_rx,
208
0
                    guard((), move |()| {
209
0
                        inflight_cache_checks.lock().remove(&unique_key);
210
0
                    }),
211
                )
212
0
            })
213
        };
214
0
        let (action_listener_rx, scope_guard) = match cache_check_result {
215
0
            Ok(action_listener_fut) => {
216
0
                let action_listener = action_listener_fut.await.map_err(|_| {
217
0
                    make_err!(
218
0
                        Code::Internal,
219
                        "ActionStateResult tx hung up in CacheLookupScheduler::add_action"
220
                    )
221
0
                })?;
222
0
                return action_listener;
223
            }
224
0
            Err(client_tx_and_scope_guard) => client_tx_and_scope_guard,
225
        };
226
227
0
        let ac_store = self.ac_store.clone();
228
0
        let action_scheduler = self.action_scheduler.clone();
229
0
        let inflight_cache_checks = self.inflight_cache_checks.clone();
230
        // We need this spawn because we are returning a stream and this spawn will populate the stream's data.
231
0
        background_spawn!("cache_lookup_scheduler_add_action", async move {
232
            // If our spawn ever dies, we will remove the action from the inflight_cache_checks map.
233
0
            let _scope_guard = scope_guard;
234
235
0
            let unique_key = match &action_info.unique_qualifier {
236
0
                ActionUniqueQualifier::Cacheable(unique_key) => unique_key,
237
0
                ActionUniqueQualifier::Uncacheable(unique_key) => {
238
0
                    error!(
239
                        ?action_info,
240
0
                        "ActionInfo::unique_qualifier should be ActionUniqueQualifier::Cacheable()"
241
                    );
242
0
                    unique_key
243
                }
244
            };
245
246
            // Perform cache check.
247
0
            let instance_name = action_info.unique_qualifier.instance_name().clone();
248
0
            let maybe_action_result = get_action_from_store(
249
0
                &ac_store,
250
0
                action_info.unique_qualifier.digest(),
251
0
                instance_name,
252
0
                action_info.unique_qualifier.digest_function(),
253
0
            )
254
0
            .await;
255
0
            match maybe_action_result {
256
0
                Ok(action_result) => {
257
0
                    let maybe_pending_txs = {
258
0
                        let mut inflight_cache_checks = inflight_cache_checks.lock();
259
                        // We are ready to resolve the in-flight actions. We remove the
260
                        // in-flight actions from the map.
261
0
                        inflight_cache_checks.remove(unique_key)
262
                    };
263
0
                    let Some(pending_txs) = maybe_pending_txs else {
  Branch (263:25): [True: 0, False: 0]
  Branch (263:25): [Folded - Ignored]
264
0
                        return; // Nobody is waiting for this action anymore.
265
                    };
266
0
                    let mut action_state = ActionState {
267
0
                        client_operation_id: OperationId::default(),
268
0
                        stage: ActionStage::CompletedFromCache(action_result),
269
0
                        action_digest: action_info.unique_qualifier.digest(),
270
0
                    };
271
272
0
                    let ctx = Context::current();
273
0
                    let baggage = ctx.baggage();
274
275
0
                    let maybe_origin_metadata = if baggage.is_empty() {
  Branch (275:52): [True: 0, False: 0]
  Branch (275:52): [Folded - Ignored]
276
0
                        None
277
                    } else {
278
                        Some(OriginMetadata {
279
0
                            identity: baggage
280
0
                                .get(ENDUSER_ID)
281
0
                                .map(|v| v.as_str().to_string())
282
0
                                .unwrap_or_default(),
283
0
                            bazel_metadata: None, // TODO(aaronmondal): Implement conversion.
284
                        })
285
                    };
286
287
0
                    for (client_operation_id, pending_tx) in pending_txs {
288
0
                        action_state.client_operation_id = client_operation_id;
289
0
                        // Ignore errors here, as the other end may have hung up.
290
0
                        drop(pending_tx.send(Ok(Box::new(CacheLookupActionStateResult {
291
0
                            action_state: Arc::new(action_state.clone()),
292
0
                            maybe_origin_metadata: maybe_origin_metadata.clone(),
293
0
                            change_called: false,
294
0
                        }))));
295
0
                    }
296
0
                    return;
297
                }
298
0
                Err(err) => {
299
                    // NotFound errors just mean we need to execute our action.
300
0
                    if err.code != Code::NotFound {
  Branch (300:24): [True: 0, False: 0]
  Branch (300:24): [Folded - Ignored]
301
0
                        let err = err.append("In CacheLookupScheduler::add_action");
302
0
                        let maybe_pending_txs = {
303
0
                            let mut inflight_cache_checks = inflight_cache_checks.lock();
304
                            // We are ready to resolve the in-flight actions. We remove the
305
                            // in-flight actions from the map.
306
0
                            inflight_cache_checks.remove(unique_key)
307
                        };
308
0
                        let Some(pending_txs) = maybe_pending_txs else {
  Branch (308:29): [True: 0, False: 0]
  Branch (308:29): [Folded - Ignored]
309
0
                            return; // Nobody is waiting for this action anymore.
310
                        };
311
0
                        for (_client_operation_id, pending_tx) in pending_txs {
312
0
                            // Ignore errors here, as the other end may have hung up.
313
0
                            drop(pending_tx.send(Err(err.clone())));
314
0
                        }
315
0
                        return;
316
0
                    }
317
                }
318
            }
319
320
0
            let maybe_pending_txs = {
321
0
                let mut inflight_cache_checks = inflight_cache_checks.lock();
322
0
                inflight_cache_checks.remove(unique_key)
323
            };
324
0
            let Some(pending_txs) = maybe_pending_txs else {
  Branch (324:17): [True: 0, False: 0]
  Branch (324:17): [Folded - Ignored]
325
0
                return; // Noone is waiting for this action anymore.
326
            };
327
328
0
            for (client_operation_id, pending_tx) in pending_txs {
329
                // Ignore errors here, as the other end may have hung up.
330
0
                drop(
331
0
                    pending_tx.send(
332
0
                        action_scheduler
333
0
                            .add_action(client_operation_id, action_info.clone())
334
0
                            .await,
335
                    ),
336
                );
337
            }
338
0
        });
339
0
        action_listener_rx
340
0
            .await
341
0
            .map_err(|_| {
342
0
                make_err!(
343
0
                    Code::Internal,
344
                    "ActionStateResult tx hung up in CacheLookupScheduler::add_action"
345
                )
346
0
            })?
347
0
            .err_tip(|| "In CacheLookupScheduler::add_action")
348
1
    }
349
350
1
    async fn inner_filter_operations(
351
1
        &self,
352
1
        filter: OperationFilter,
353
1
    ) -> Result<ActionStateResultStream, Error> {
354
1
        self.action_scheduler
355
1
            .filter_operations(filter)
356
1
            .await
357
1
            .err_tip(|| "In CacheLookupScheduler::filter_operations")
358
1
    }
359
}
360
361
#[async_trait]
362
impl ClientStateManager for CacheLookupScheduler {
363
    async fn add_action(
364
        &self,
365
        client_operation_id: OperationId,
366
        action_info: Arc<ActionInfo>,
367
2
    ) -> Result<Box<dyn ActionStateResult>, Error> {
368
1
        self.inner_add_action(client_operation_id, action_info)
369
1
            .await
370
2
    }
371
372
    async fn filter_operations(
373
        &self,
374
        filter: OperationFilter,
375
2
    ) -> Result<ActionStateResultStream, Error> {
376
1
        self.inner_filter_operations(filter).await
377
2
    }
378
379
0
    fn as_known_platform_property_provider(&self) -> Option<&dyn KnownPlatformPropertyProvider> {
380
0
        self.action_scheduler.as_known_platform_property_provider()
381
0
    }
382
}
383
384
impl RootMetricsComponent for CacheLookupScheduler {}