Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 9 additions & 12 deletions components/spider-client/src/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2,20 +2,17 @@

use std::num::NonZeroUsize;

use spider_core::{
job::JobState,
task::TaskGraph,
types::{
id::{JobId, ResourceGroupId},
io::{TaskInput, TaskOutput},
},
};
use spider_core::job::JobState;
use spider_core::task::TaskGraph;
use spider_core::types::id::JobId;
use spider_core::types::id::ResourceGroupId;
use spider_core::types::io::TaskInput;
use spider_core::types::io::TaskOutput;
use tonic::transport::Endpoint;

use crate::{
error::ClientError,
grpc::{job::JobOrchestrationClient, resource_group::ResourceGroupManagementClient},
};
use crate::error::ClientError;
use crate::grpc::job::JobOrchestrationClient;
use crate::grpc::resource_group::ResourceGroupManagementClient;

/// User-facing client for the Spider storage gRPC services.
#[derive(Debug, Clone)]
Expand Down
37 changes: 18 additions & 19 deletions components/spider-client/src/grpc/job.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2,27 +2,26 @@

use std::num::NonZeroUsize;

use spider_core::{
compression::encode_zstd_bytes,
job::JobState,
task::TaskGraph,
types::{
id::{JobId, ResourceGroupId},
io::{SerializedTaskOutputs, TaskInput, TaskInputsSerializer, TaskOutput},
},
};
use spider_proto_rust::{
error::Error as ProtoError,
storage::{self, job_orchestration_service_client::JobOrchestrationServiceClient},
};
use spider_core::compression::encode_zstd_bytes;
use spider_core::job::JobState;
use spider_core::task::TaskGraph;
use spider_core::types::id::JobId;
use spider_core::types::id::ResourceGroupId;
use spider_core::types::io::SerializedTaskOutputs;
use spider_core::types::io::TaskInput;
use spider_core::types::io::TaskInputsSerializer;
use spider_core::types::io::TaskOutput;
use spider_proto_rust::error::Error as ProtoError;
use spider_proto_rust::storage::JobOrchestrationServiceClient;
use spider_proto_rust::storage::{self};
use spider_utils::grpc::client::ConnectionPool;
use tonic::{
Code,
Status,
transport::{Channel, Endpoint},
};
use tonic::Code;
use tonic::Status;
use tonic::transport::Channel;
use tonic::transport::Endpoint;

use crate::error::{ClientError, to_transport_error};
use crate::error::ClientError;
use crate::error::to_transport_error;

/// gRPC client for the storage server's job-orchestration service.
#[derive(Debug, Clone)]
Expand Down
18 changes: 8 additions & 10 deletions components/spider-client/src/grpc/resource_group.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3,18 +3,16 @@
use std::num::NonZeroUsize;

use spider_core::types::id::ResourceGroupId;
use spider_proto_rust::storage::{
self,
resource_group_management_service_client::ResourceGroupManagementServiceClient,
};
use spider_proto_rust::storage::ResourceGroupManagementServiceClient;
use spider_proto_rust::storage::{self};
use spider_utils::grpc::client::ConnectionPool;
use tonic::{
Code,
Status,
transport::{Channel, Endpoint},
};
use tonic::Code;
use tonic::Status;
use tonic::transport::Channel;
use tonic::transport::Endpoint;

use crate::error::{ClientError, to_transport_error};
use crate::error::ClientError;
use crate::error::to_transport_error;

/// gRPC client for the storage server's resource-group-management service.
#[derive(Debug, Clone)]
Expand Down
3 changes: 2 additions & 1 deletion components/spider-core/src/compression.rs
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,8 @@ pub fn decode_zstd_bytes(compressed: &[u8]) -> Result<Vec<u8>, Error> {

#[cfg(test)]
mod tests {
use super::{decode_zstd_bytes, encode_zstd_bytes};
use super::decode_zstd_bytes;
use super::encode_zstd_bytes;

#[test]
fn zstd_bytes_round_trip() {
Expand Down
7 changes: 3 additions & 4 deletions components/spider-core/src/session.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,9 @@
//! execution manager and the scheduler) can observe and advance a shared view of storage's current
//! session id.

use std::sync::{
Arc,
atomic::{AtomicU64, Ordering},
};
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;

use crate::types::id::SessionId;

Expand Down
42 changes: 23 additions & 19 deletions components/spider-core/src/task/task_graph.rs
Original file line number Diff line number Diff line change
@@ -1,16 +1,22 @@
use semver::{Version, VersionReq};
use serde::{
Deserialize,
Serialize,
de::{self, MapAccess, Visitor},
ser::{SerializeMap, SerializeSeq, Serializer},
};
use strum::{AsRefStr, EnumCount, EnumIter, IntoEnumIterator};

use crate::{
compression::{decode_zstd_bytes, encode_zstd_bytes},
task::{DataTypeDescriptor, Error},
};
use semver::Version;
use semver::VersionReq;
use serde::Deserialize;
use serde::Serialize;
use serde::de::MapAccess;
use serde::de::Visitor;
use serde::de::{self};
use serde::ser::SerializeMap;
use serde::ser::SerializeSeq;
use serde::ser::Serializer;
use strum::AsRefStr;
use strum::EnumCount;
use strum::EnumIter;
use strum::IntoEnumIterator;

use crate::compression::decode_zstd_bytes;
use crate::compression::encode_zstd_bytes;
use crate::task::DataTypeDescriptor;
use crate::task::Error;

/// A unique identifier for a task within a task graph, assigned based on insertion order.
///
Expand Down Expand Up @@ -1078,12 +1084,10 @@ mod tests {
use std::fmt::Debug;

use super::*;
use crate::task::{
DataTypeDescriptor,
IntTypeDescriptor,
MapKeyTypeDescriptor,
ValueTypeDescriptor,
};
use crate::task::DataTypeDescriptor;
use crate::task::IntTypeDescriptor;
use crate::task::MapKeyTypeDescriptor;
use crate::task::ValueTypeDescriptor;

const TEST_PACKAGE: &str = "test_pkg";

Expand Down
3 changes: 2 additions & 1 deletion components/spider-core/src/task/type_descriptor.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
use non_empty_string::NonEmptyString;
use serde::{Deserialize, Serialize};
use serde::Deserialize;
use serde::Serialize;

use crate::task::Error;

Expand Down
18 changes: 11 additions & 7 deletions components/spider-core/src/types/id.rs
Original file line number Diff line number Diff line change
@@ -1,10 +1,13 @@
use std::{
fmt::{Debug, Display},
marker::PhantomData,
};
use std::fmt::Debug;
use std::fmt::Display;
use std::marker::PhantomData;

use serde::{Deserialize, Deserializer, Serialize, Serializer};
use sqlx::{Database, encode::IsNull};
use serde::Deserialize;
use serde::Deserializer;
use serde::Serialize;
use serde::Serializer;
use sqlx::Database;
use sqlx::encode::IsNull;

use crate::task::TaskIndex;

Expand Down Expand Up @@ -224,7 +227,8 @@ pub type SignedJobId = SignedId<JobIdMarker>;

#[cfg(test)]
mod tests {
use super::{JobId, ResourceGroupId};
use super::JobId;
use super::ResourceGroupId;

#[test]
fn id_serializes_as_u64() {
Expand Down
21 changes: 13 additions & 8 deletions components/spider-core/src/types/io.rs
Original file line number Diff line number Diff line change
@@ -1,11 +1,16 @@
use serde::{Deserialize, Serialize};
use spider_utils::wire::{self, StreamDeserializer, WireError, WireFrameBuilder};

use crate::{
compression::{self, decode_zstd_bytes, encode_zstd_bytes},
task::{TdlContext, TimeoutPolicy},
types::id::TaskInstanceId,
};
use serde::Deserialize;
use serde::Serialize;
use spider_utils::wire::StreamDeserializer;
use spider_utils::wire::WireError;
use spider_utils::wire::WireFrameBuilder;
use spider_utils::wire::{self};

use crate::compression::decode_zstd_bytes;
use crate::compression::encode_zstd_bytes;
use crate::compression::{self};
use crate::task::TdlContext;
use crate::task::TimeoutPolicy;
use crate::types::id::TaskInstanceId;

/// Represents an input of a task.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
Expand Down
7 changes: 6 additions & 1 deletion components/spider-core/src/types/scheduler.rs
Original file line number Diff line number Diff line change
@@ -1,6 +1,11 @@
use std::net::IpAddr;

use crate::types::id::{JobId, ResourceGroupId, SchedulerId, SessionId, TaskAssignmentId, TaskId};
use crate::types::id::JobId;
use crate::types::id::ResourceGroupId;
use crate::types::id::SchedulerId;
use crate::types::id::SessionId;
use crate::types::id::TaskAssignmentId;
use crate::types::id::TaskId;

/// The currently registered scheduler endpoint.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
Expand Down
3 changes: 2 additions & 1 deletion components/spider-derive/src/lib.rs
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
mod mysql;

use proc_macro::TokenStream;
use syn::{DeriveInput, parse_macro_input};
use syn::DeriveInput;
use syn::parse_macro_input;

/// Derives MySQL-compatible enum support for use with [`sqlx`].
///
Expand Down
4 changes: 3 additions & 1 deletion components/spider-derive/src/mysql.rs
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
use proc_macro2::TokenStream;
use quote::quote;
use syn::{Data, DataEnum, DeriveInput};
use syn::Data;
use syn::DataEnum;
use syn::DeriveInput;

pub fn derive_mysql_enum(input: &DeriveInput) -> syn::Result<TokenStream> {
let enum_type_name = &input.ident;
Expand Down
Original file line number Diff line number Diff line change
@@ -1,14 +1,16 @@
//! Command-line entrypoint for the execution manager.

use std::{error::Error, path::PathBuf};
use std::error::Error;
use std::path::PathBuf;

use clap::Parser;
use spider_execution_manager::{
Config,
client::grpc::{GrpcLivenessClient, GrpcSchedulerClient, GrpcStorageClient},
runtime::Runtime,
};
use spider_utils::{config::YamlConfig, logging::set_up_logging};
use spider_execution_manager::Config;
use spider_execution_manager::client::grpc::GrpcLivenessClient;
use spider_execution_manager::client::grpc::GrpcSchedulerClient;
use spider_execution_manager::client::grpc::GrpcStorageClient;
use spider_execution_manager::runtime::Runtime;
use spider_utils::config::YamlConfig;
use spider_utils::logging::set_up_logging;

/// Command-line arguments for the execution manager.
#[derive(Debug, Parser)]
Expand Down
28 changes: 15 additions & 13 deletions components/spider-execution-manager/src/client/grpc/liveness.rs
Original file line number Diff line number Diff line change
@@ -1,21 +1,22 @@
//! gRPC-backed [`LivenessClient`] implementation.

use std::{net::IpAddr, num::NonZeroUsize};
use std::net::IpAddr;
use std::num::NonZeroUsize;

use async_trait::async_trait;
use spider_core::types::id::{ExecutionManagerId, SessionId};
use spider_proto_rust::storage::{
self,
execution_manager_liveness_service_client::ExecutionManagerLivenessServiceClient,
};
use spider_core::types::id::ExecutionManagerId;
use spider_core::types::id::SessionId;
use spider_proto_rust::storage::ExecutionManagerLivenessServiceClient;
use spider_proto_rust::storage::{self};
use spider_utils::grpc::client::ConnectionPool;
use tonic::{
Code,
Status,
transport::{Channel, Endpoint},
};
use tonic::Code;
use tonic::Status;
use tonic::transport::Channel;
use tonic::transport::Endpoint;

use crate::client::liveness::{LivenessClient, LivenessResponseError, RegistrationResponse};
use crate::client::liveness::LivenessClient;
use crate::client::liveness::LivenessResponseError;
use crate::client::liveness::RegistrationResponse;

/// gRPC-backed [`LivenessClient`] implementation.
#[derive(Debug, Clone)]
Expand Down Expand Up @@ -143,7 +144,8 @@ mod tests {
use spider_core::types::id::ExecutionManagerId;

use super::*;
use crate::client::{LivenessResponseError, RegistrationResponse};
use crate::client::LivenessResponseError;
use crate::client::RegistrationResponse;

#[test]
fn register_response_to_result_returns_registration() {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,16 +3,19 @@
use std::num::NonZeroUsize;

use async_trait::async_trait;
use spider_core::types::{id::ExecutionManagerId, scheduler::TaskAssignmentRecord};
use spider_proto_rust::scheduler::{self, scheduler_service_client::SchedulerServiceClient};
use spider_core::types::id::ExecutionManagerId;
use spider_core::types::scheduler::TaskAssignmentRecord;
use spider_proto_rust::scheduler::SchedulerServiceClient;
use spider_proto_rust::scheduler::{self};
use spider_utils::grpc::client::ConnectionPool;
use tonic::{
Code,
Status,
transport::{Channel, Endpoint},
};

use crate::client::{SchedulerClient, SchedulerError, SchedulerResponse};
use tonic::Code;
use tonic::Status;
use tonic::transport::Channel;
use tonic::transport::Endpoint;

use crate::client::SchedulerClient;
use crate::client::SchedulerError;
use crate::client::SchedulerResponse;

/// gRPC-backed [`SchedulerClient`] implementation.
#[derive(Debug, Clone)]
Expand Down
Loading
Loading