rust/cfdp
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

continue

Browse Source
This commit is contained in:
Robin Müller
2025-09-18 17:05:20 +02:00
parent c0a7dfd9f8
commit 8c373551dc
3 changed files with 148 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

190
src/dest.rs
View File

@@ -30,36 +30,36 @@
//! 4. A Finished PDU has been sent back to the remote side.
//! 5. A Finished PDU ACK was received.
use crate::{
DummyPduProvider, GenericSendError, IndicationConfig, PduProvider, PositiveAckParams,
lost_segments::{LostSegmentError, LostSegmentStore},
user::TransactionFinishedParams,
DummyPduProvider, GenericSendError, IndicationConfig, PduProvider, PositiveAckParams,
};
use core::{
cell::{Cell, RefCell},
str::{Utf8Error, from_utf8, from_utf8_unchecked},
str::{from_utf8, from_utf8_unchecked, Utf8Error},
};
use super::{
Countdown, EntityType, LocalEntityConfig, PacketTarget, PduSendProvider, RemoteConfigStore,
RemoteEntityConfig, State, TimerContext, TimerCreator, TransactionId, UserFaultHook,
filestore::{FilestoreError, VirtualFilestore},
user::{CfdpUser, FileSegmentRecvdParams, MetadataReceivedParams},
Countdown, EntityType, LocalEntityConfig, PacketTarget, PduSender, RemoteConfigStore,
RemoteEntityConfig, State, TimerContext, TimerCreator, TransactionId, UserFaultHook,
};
use smallvec::SmallVec;
use spacepackets::{
cfdp::{
ChecksumType, ConditionCode, FaultHandlerCode, LargeFileFlag, PduType, TransactionStatus,
TransmissionMode,
pdu::{
CfdpPdu, CommonPduConfig, FileDirectiveType, PduError, PduHeader,
ack::AckPdu,
eof::EofPdu,
file_data::FileDataPdu,
finished::{DeliveryCode, FileStatus, FinishedPduCreator},
metadata::{MetadataGenericParams, MetadataPduReader},
nak::NakPduCreatorWithReservedSeqReqsBuf,
CfdpPdu, CommonPduConfig, FileDirectiveType, PduError, PduHeader,
},
tlv::{EntityIdTlv, GenericTlv, ReadableTlv, TlvType, msg_to_user::MsgToUserTlv},
tlv::{msg_to_user::MsgToUserTlv, EntityIdTlv, GenericTlv, ReadableTlv, TlvType},
ChecksumType, ConditionCode, FaultHandlerCode, LargeFileFlag, PduType, TransactionStatus,
TransmissionMode,
},
util::{UnsignedByteField, UnsignedEnum},
};
@@ -334,9 +334,9 @@ pub enum DestError {
/// in different concurrent contexts. For example, you can dynamically create new handlers and
/// run them inside a thread pool, or move the newly created handler to a new thread."""
pub struct DestinationHandler<
PduSender: PduSendProvider,
PduSenderInstance: PduSender,
UserFaultHookInstance: UserFaultHook,
Vfs: VirtualFilestore,
VirtualFileStoreInstance: VirtualFilestore,
RemoteConfigStoreInstance: RemoteConfigStore,
TimerCreatorInstance: TimerCreator<Countdown = CountdownInstance>,
CountdownInstance: Countdown,
@@ -347,8 +347,8 @@ pub struct DestinationHandler<
state: State,
transaction_params: TransactionParams<CountdownInstance>,
pdu_and_cksum_buffer: RefCell<alloc::vec::Vec<u8>>,
pub pdu_sender: PduSender,
pub vfs: Vfs,
pub pdu_sender: PduSenderInstance,
pub vfs: VirtualFileStoreInstance,
pub remote_cfg_table: RemoteConfigStoreInstance,
pub check_timer_creator: TimerCreatorInstance,
lost_segment_tracker: LostSegmentTracker,
@@ -366,14 +366,14 @@ pub type StdDestinationHandler<PduSender, UserFaultHook> = DestinationHandler<
>;
#[cfg(feature = "std")]
impl<PduSender: PduSendProvider, UserFaultHookInstance: UserFaultHook>
StdDestinationHandler<PduSender, UserFaultHookInstance>
impl<PduSenderInstance: PduSender, UserFaultHookInstance: UserFaultHook>
StdDestinationHandler<PduSenderInstance, UserFaultHookInstance>
{
#[cfg(feature = "std")]
pub fn new_std(
local_cfg: LocalEntityConfig<UserFaultHookInstance>,
pdu_and_cksum_buf_size: usize,
pdu_sender: PduSender,
pdu_sender: PduSenderInstance,
) -> Self {
Self::new(
local_cfg,
@@ -388,18 +388,18 @@ impl<PduSender: PduSendProvider, UserFaultHookInstance: UserFaultHook>
}
impl<
PduSender: PduSendProvider,
UserFaultHookInstance: UserFaultHook,
Vfs: VirtualFilestore,
RemoteConfigStoreInstance: RemoteConfigStore,
TimerCreatorInstance: TimerCreator<Countdown = CountdownInstance>,
CountdownInstance: Countdown,
LostSegmentTracker: LostSegmentStore,
>
PduSenderInstance: PduSender,
UserFaultHookInstance: UserFaultHook,
VirtualFilestoreInstance: VirtualFilestore,
RemoteConfigStoreInstance: RemoteConfigStore,
TimerCreatorInstance: TimerCreator<Countdown = CountdownInstance>,
CountdownInstance: Countdown,
LostSegmentTracker: LostSegmentStore,
>
DestinationHandler<
PduSender,
PduSenderInstance,
UserFaultHookInstance,
Vfs,
VirtualFilestoreInstance,
RemoteConfigStoreInstance,
TimerCreatorInstance,
CountdownInstance,
@@ -429,8 +429,8 @@ impl<
pub fn new(
local_cfg: LocalEntityConfig<UserFaultHookInstance>,
pdu_and_cksum_buf_size: usize,
pdu_sender: PduSender,
vfs: Vfs,
pdu_sender: PduSenderInstance,
vfs: VirtualFilestoreInstance,
remote_cfg_table: RemoteConfigStoreInstance,
timer_creator: TimerCreatorInstance,
lost_segment_tracker: LostSegmentTracker,
@@ -987,20 +987,110 @@ impl<
})
}
}
self.transaction_params.deferred_procedure_timer = Some(
self.check_timer_creator
.create_countdown(TimerContext::NakActivity {
expiry_time: self
.transaction_params
.remote_cfg
.as_ref()
.unwrap()
.nak_timer_interval,
}),
);
}
fn deferred_lost_segment_handling(&mut self) {
assert!(
self.transaction_params.acked_params.is_some(),
"acknowledged parameters unexpectedly None"
);
let acked_params = self.transaction_params.acked_params.as_mut().unwrap();
if self.lost_segment_tracker.is_empty() && !acked_params.metadata_missing {
// We are done and have received everything.
match self.checksum_verify(
self.transaction_params.progress,
self.transaction_params.checksum,
) {
true => {
self.transaction_params
.finished_params
.condition_code
.set(ConditionCode::NoError);
self.transaction_params
.finished_params
.delivery_code
.set(DeliveryCode::Complete);
}
false => {
self.transaction_params
.finished_params
.condition_code
.set(ConditionCode::FileChecksumFailure);
self.transaction_params
.finished_params
.delivery_code
.set(DeliveryCode::Incomplete);
}
}
self.set_step(TransactionStep::TransferCompletion);
acked_params.deferred_procedure_active = false;
return;
}
let mut first_nak_issuance = self.transaction_params.deferred_procedure_timer.is_none();
if first_nak_issuance {
self.transaction_params.deferred_procedure_timer = Some(
self.check_timer_creator
.create_countdown(TimerContext::NakActivity {
expiry_time: self
.transaction_params
.remote_cfg
.as_ref()
.unwrap()
.nak_timer_interval,
}),
);
} else if !self
.transaction_params
.deferred_procedure_timer
.as_ref()
.unwrap()
.has_expired()
{
return;
}
if !first_nak_issuance
&& acked_params.nak_activity_counter + 1
== self
.transaction_params
.remote_cfg
.as_ref()
.unwrap()
.nak_timer_expiration_limit
{
self.transaction_params.finished_params.delivery_code = DeliveryCode::Incomplete;
if self.declare_fault(ConditionCode::NakLimitReached)
== FaultHandlerCode::AbandonTransaction
{
self.abandon_transaction();
}
return;
}
let pdu_header = PduHeader::new_no_file_data(self.transaction_params.pdu_conf, 0);
let max_segment_reqs = NakPduCreatorWithReservedSeqReqsBuf::calculate_max_segment_requests(
self.transaction_params
.remote_cfg
.as_ref()
.unwrap()
.max_packet_len,
&pdu_header,
);
let mut segments_to_send = self.lost_segment_tracker.number_of_segments();
if acked_params.metadata_missing {
segments_to_send += 1;
}
if segments_to_send <= max_segment_reqs {
}
let mut nak_pdu_creator = NakPduCreatorWithReservedSeqReqsBuf::new(
self.pdu_and_cksum_buffer.get_mut(),
pdu_header,
core::cmp::min(segments_to_send as usize, max_segment_reqs),
)
for index in 0..segments_to_send {
}
// TODO.
}
@@ -1511,21 +1601,21 @@ mod tests {
use rand::Rng;
use spacepackets::{
cfdp::{
ChecksumType, TransmissionMode,
lv::Lv,
pdu::{WritablePduPacket, finished::FinishedPduReader, metadata::MetadataPduCreator},
pdu::{finished::FinishedPduReader, metadata::MetadataPduCreator, WritablePduPacket},
ChecksumType, TransmissionMode,
},
util::{UbfU16, UnsignedByteFieldU8},
};
use crate::{
CRC_32, FaultHandler, IndicationConfig, PduRawWithInfo, RemoteConfigStoreStd,
filestore::NativeFilestore,
lost_segments::LostSegmentsList,
tests::{
LOCAL_ID, REMOTE_ID, SentPdu, TestCfdpSender, TestCfdpUser, TestCheckTimer,
TestCheckTimerCreator, TestFaultHandler, TimerExpiryControl, basic_remote_cfg_table,
basic_remote_cfg_table, SentPdu, TestCfdpSender, TestCfdpUser, TestCheckTimer,
TestCheckTimerCreator, TestFaultHandler, TimerExpiryControl, LOCAL_ID, REMOTE_ID,
},
FaultHandler, IndicationConfig, PduRawWithInfo, RemoteConfigStoreStd, CRC_32,
};
use super::*;
@@ -1842,14 +1932,12 @@ mod tests {
let dest_handler =
default_dest_handler(fault_handler, test_sender, &TimerExpiryControl::default());
assert!(dest_handler.transmission_mode().is_none());
assert!(
dest_handler
.local_cfg
.fault_handler
.user_hook
.borrow()
.all_queues_empty()
);
assert!(dest_handler
.local_cfg
.fault_handler
.user_hook
.borrow()
.all_queues_empty());
assert!(dest_handler.pdu_sender.queue_empty());
assert_eq!(dest_handler.state(), State::Idle);
assert_eq!(dest_handler.step(), TransactionStep::Idle);

6
src/lib.rs
View File

@@ -693,7 +693,7 @@ pub enum GenericSendError {
Other,
}
pub trait PduSendProvider {
pub trait PduSender {
fn send_pdu(
&self,
pdu_type: PduType,
@@ -716,7 +716,7 @@ pub mod std_mod {
use super::*;
impl PduSendProvider for mpsc::Sender<PduOwnedWithInfo> {
impl PduSender for mpsc::Sender<PduOwnedWithInfo> {
fn send_pdu(
&self,
pdu_type: PduType,
@@ -1471,7 +1471,7 @@ pub(crate) mod tests {
pub packet_queue: RefCell<VecDeque<SentPdu>>,
}
impl PduSendProvider for TestCfdpSender {
impl PduSender for TestCfdpSender {
fn send_pdu(
&self,
pdu_type: PduType,

12
src/source.rs
View File

@@ -72,7 +72,7 @@ use crate::{
};
use super::{
LocalEntityConfig, PacketTarget, PduSendProvider, RemoteConfigStore, RemoteEntityConfig, State,
LocalEntityConfig, PacketTarget, PduSender, RemoteConfigStore, RemoteEntityConfig, State,
TransactionId, UserFaultHook,
filestore::{FilestoreError, VirtualFilestore},
request::{ReadablePutRequest, StaticPutRequestCacher},
@@ -242,7 +242,7 @@ pub enum FsmContext {
/// is required, it is recommended to create a new handler and run all active handlers inside a
/// thread pool, or move the newly created handler to a new thread.
pub struct SourceHandler<
PduSender: PduSendProvider,
PduSenderInstance: PduSender,
UserFaultHookInstance: UserFaultHook,
Vfs: VirtualFilestore,
RemoteConfigStoreInstance: RemoteConfigStore,
@@ -251,7 +251,7 @@ pub struct SourceHandler<
SequenceCounterInstance: SequenceCounter,
> {
local_cfg: LocalEntityConfig<UserFaultHookInstance>,
pdu_sender: PduSender,
pdu_sender: PduSenderInstance,
pdu_and_cksum_buffer: RefCell<alloc::vec::Vec<u8>>,
put_request_cacher: StaticPutRequestCacher,
remote_cfg_table: RemoteConfigStoreInstance,
@@ -271,7 +271,7 @@ pub struct SourceHandler<
}
impl<
PduSender: PduSendProvider,
PduSenderInstance: PduSender,
UserFaultHookInstance: UserFaultHook,
Vfs: VirtualFilestore,
RemoteConfigStoreInstance: RemoteConfigStore,
@@ -280,7 +280,7 @@ impl<
SequenceCounterInstance: SequenceCounter,
>
SourceHandler<
PduSender,
PduSenderInstance,
UserFaultHookInstance,
Vfs,
RemoteConfigStoreInstance,
@@ -314,7 +314,7 @@ impl<
#[allow(clippy::too_many_arguments)]
pub fn new(
cfg: LocalEntityConfig<UserFaultHookInstance>,
pdu_sender: PduSender,
pdu_sender: PduSenderInstance,
vfs: Vfs,
put_request_cacher: StaticPutRequestCacher,
pdu_and_cksum_buf_size: usize,