gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-05-08 01:05:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Futher simplify TaskSpawner

Browse Source
This commit is contained in:
Paul Hauner
2023-07-04 17:50:06 +10:00
parent a7b259d354
commit 10dc153ea1
1 changed files with 14 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
beacon_node/http_api/src/task_spawner.rs
View File

@@ -104,35 +104,16 @@ impl<E: EthSpec> TaskSpawner<E> {
// send can only fail if the receiver is dropped. // send can only fail if the receiver is dropped.
let _ = tx.send(func_result); let _ = tx.send(func_result);
}; };
let process_fn = BlockingOrAsync::Async(Box::pin(process_fn));
// Send the function to the beacon processor for execution at some arbitrary time. // Send the function to the beacon processor for execution at some arbitrary time.
let error_message = send_to_beacon_processor(
match beacon_processor_send.try_send(priority.work_event(process_fn)) { beacon_processor_send,
Ok(()) => { priority,
match rx.await { BlockingOrAsync::Async(Box::pin(process_fn)),
// The beacon processor executed the task and sent a result. rx,
Ok(func_result) => {
return func_result
}
// The beacon processor dropped the channel without sending a
// result. The beacon processor dropped this task because its
// queues are full or it's shutting down.
Err(_) => "The task did not execute. The server is overloaded or shutting down.",
}
}
Err(TrySendError::Full(_)) => "The task was dropped. The server is overloaded.",
Err(TrySendError::Closed(_)) => {
"The task was dropped. The server is shutting down."
}
};
let error_response = warp::reply::with_status(
warp::reply::json(&error_message),
eth2::StatusCode::INTERNAL_SERVER_ERROR,
) )
.into_response(); .await
Ok(error_response) .unwrap_or_else(Result::Ok)
} else { } else {
// There is no beacon processor so spawn a task directly on the // There is no beacon processor so spawn a task directly on the
// tokio executor. // tokio executor.
@@ -163,34 +144,16 @@ impl<E: EthSpec> TaskSpawner<E> {
// send can only fail if the receiver is dropped. // send can only fail if the receiver is dropped.
let _ = tx.send(func_result); let _ = tx.send(func_result);
}; };
let process_fn = BlockingOrAsync::Async(Box::pin(process_fn));
// Send the function to the beacon processor for execution at some arbitrary time. // Send the function to the beacon processor for execution at some arbitrary time.
let error_message = send_to_beacon_processor(
match beacon_processor_send.try_send(priority.work_event(process_fn)) { beacon_processor_send,
Ok(()) => { priority,
match rx.await { BlockingOrAsync::Async(Box::pin(process_fn)),
// The beacon processor executed the task and sent a result. rx,
Ok(func_result) => {
return func_result
}
// The beacon processor dropped the channel without sending a
// result. The beacon processor dropped this task because its
// queues are full or it's shutting down.
Err(_) => "The task did not execute. The server is overloaded or shutting down.",
}
}
Err(TrySendError::Full(_)) => "The task was dropped. The server is overloaded.",
Err(TrySendError::Closed(_)) => {
"The task was dropped. The server is shutting down."
}
};
warp::reply::with_status(
warp::reply::json(&error_message),
eth2::StatusCode::INTERNAL_SERVER_ERROR,
) )
.into_response() .await
.unwrap_or_else(|error_response| error_response)
} else { } else {
// There is no beacon processor so spawn a task directly on the // There is no beacon processor so spawn a task directly on the
// tokio executor. // tokio executor.