1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
use std::collections::BTreeMap;
use std::fs;
use anyhow::{Context, Result};
use filecoin_proofs_api::StorageProofsError;
use tracing::debug;
use super::tasks::{
AddPieces, SnapEncode, SnapProve, Transfer, TransferRoute, TreeD, WindowPoSt, WindowPoStOutput, WinningPoSt, WinningPoStOutput, C2,
PC1, PC2,
};
use crate::core::{Processor, Task};
use crate::fil_proofs::{
create_tree_d, generate_window_post, generate_winning_post, seal_commit_phase2, seal_pre_commit_phase1, seal_pre_commit_phase2,
snap_encode_into, snap_generate_sector_update_proof, to_prover_id, write_and_preprocess, PartitionProofBytes, PrivateReplicaInfo,
};
pub mod piece;
mod transfer;
#[derive(Copy, Clone, Default, Debug)]
pub struct BuiltinProcessor;
impl Processor<AddPieces> for BuiltinProcessor {
fn process(&self, task: AddPieces) -> Result<<AddPieces as Task>::Output> {
let staged_file = fs::OpenOptions::new()
.create(true)
.read(true)
.write(true)
.truncate(true)
.open(&task.staged_filepath)
.with_context(|| format!("open staged file: {}", task.staged_filepath.display()))?;
let mut piece_infos = Vec::with_capacity(task.pieces.len().min(1));
for piece in task.pieces {
debug!(piece_file = ?piece.piece_file, "trying to add piece");
let source = piece::fetcher::open(piece.piece_file, piece.payload_size, piece.piece_size.0).context("open piece file")?;
let (piece_info, _) =
write_and_preprocess(task.seal_proof_type, source, &staged_file, piece.piece_size).context("add piece")?;
piece_infos.push(piece_info);
}
if piece_infos.is_empty() {
let sector_size: u64 = task.seal_proof_type.sector_size().into();
let pi = piece::add_piece_for_cc_sector(&staged_file, sector_size).context("add piece for cc secrtor")?;
piece_infos.push(pi);
}
Ok(piece_infos)
}
}
impl Processor<TreeD> for BuiltinProcessor {
fn process(&self, task: TreeD) -> Result<<TreeD as Task>::Output> {
create_tree_d(task.registered_proof, Some(task.staged_file), task.cache_dir).map(|_| true)
}
}
impl Processor<PC1> for BuiltinProcessor {
fn process(&self, task: PC1) -> Result<<PC1 as Task>::Output> {
seal_pre_commit_phase1(
task.registered_proof,
task.cache_path,
task.in_path,
task.out_path,
task.prover_id,
task.sector_id,
task.ticket,
&task.piece_infos[..],
)
}
}
impl Processor<PC2> for BuiltinProcessor {
fn process(&self, task: PC2) -> Result<<PC2 as Task>::Output> {
seal_pre_commit_phase2(task.pc1out, task.cache_dir, task.sealed_file)
}
}
impl Processor<C2> for BuiltinProcessor {
fn process(&self, task: C2) -> Result<<C2 as Task>::Output> {
seal_commit_phase2(task.c1out, task.prover_id, task.sector_id)
}
}
impl Processor<SnapEncode> for BuiltinProcessor {
fn process(&self, task: SnapEncode) -> Result<<SnapEncode as Task>::Output> {
snap_encode_into(
task.registered_proof,
task.new_replica_path,
task.new_cache_path,
task.sector_path,
task.sector_cache_path,
task.staged_data_path,
&task.piece_infos[..],
)
}
}
impl Processor<SnapProve> for BuiltinProcessor {
fn process(&self, task: SnapProve) -> Result<<SnapProve as Task>::Output> {
snap_generate_sector_update_proof(
task.registered_proof,
task.vannilla_proofs.into_iter().map(PartitionProofBytes).collect(),
task.comm_r_old,
task.comm_r_new,
task.comm_d_new,
)
}
}
impl Processor<Transfer> for BuiltinProcessor {
fn process(&self, task: Transfer) -> Result<<Transfer as Task>::Output> {
task.routes.into_iter().try_for_each(|route| transfer::do_transfer(&route))?;
Ok(true)
}
}
impl Processor<WindowPoSt> for BuiltinProcessor {
fn process(&self, task: WindowPoSt) -> Result<<WindowPoSt as Task>::Output> {
let replicas = BTreeMap::from_iter(task.replicas.into_iter().map(|rep| {
(
rep.sector_id,
PrivateReplicaInfo::new(task.proof_type, rep.comm_r, rep.cache_dir, rep.sealed_file),
)
}));
generate_window_post(&task.seed, &replicas, to_prover_id(task.miner_id))
.map(|proofs| WindowPoStOutput {
proofs: proofs.into_iter().map(|r| r.1).collect(),
faults: vec![],
})
.or_else(|e| {
if let Some(StorageProofsError::FaultySectors(sectors)) = e.downcast_ref::<StorageProofsError>() {
return Ok(WindowPoStOutput {
proofs: vec![],
faults: sectors.iter().map(|id| (*id).into()).collect(),
});
}
Err(e)
})
}
}
impl Processor<WinningPoSt> for BuiltinProcessor {
fn process(&self, task: WinningPoSt) -> Result<<WinningPoSt as Task>::Output> {
let replicas = BTreeMap::from_iter(task.replicas.into_iter().map(|rep| {
(
rep.sector_id,
PrivateReplicaInfo::new(task.proof_type, rep.comm_r, rep.cache_dir, rep.sealed_file),
)
}));
generate_winning_post(&task.seed, &replicas, to_prover_id(task.miner_id)).map(|proofs| WinningPoStOutput {
proofs: proofs.into_iter().map(|r| r.1).collect(),
})
}
}
