Update with the most recent storage contract

This commit is contained in:
Bruno Valente 2024-07-22 21:17:27 +08:00
parent f0c3f2cfd0
commit 8a726f2825
23 changed files with 1639 additions and 646 deletions

View File

@ -9,6 +9,7 @@ mod loader;
mod metrics;
mod mine;
mod miner_id;
mod monitor;
pub mod pora;
mod recall_range;
mod sealer;

View File

@ -9,7 +9,6 @@ use tokio::time::{sleep, Duration, Instant};
use storage::config::ShardConfig;
use zgs_spec::{SECTORS_PER_LOAD, SECTORS_PER_MAX_MINING_RANGE, SECTORS_PER_PRICING};
use super::metrics;
use crate::recall_range::RecallRange;
use crate::{
pora::{AnswerWithoutProof, Miner},
@ -20,7 +19,7 @@ use crate::{
use std::sync::Arc;
pub struct PoraService {
mine_context_receiver: mpsc::UnboundedReceiver<MineContextMessage>,
mine_context_receiver: broadcast::Receiver<MineContextMessage>,
mine_answer_sender: mpsc::UnboundedSender<AnswerWithoutProof>,
msg_recv: broadcast::Receiver<MinerMessage>,
loader: Arc<dyn PoraLoader>,
@ -33,9 +32,29 @@ pub struct PoraService {
iter_batch: usize,
}
struct PoraPuzzle {
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct PoraPuzzle {
context: MineContext,
target_quality: U256,
max_shards: u64,
}
impl PoraPuzzle {
pub fn new(context: MineContext, target_quality: U256, max_shards: u64) -> Self {
Self {
context,
target_quality,
max_shards,
}
}
pub fn max_shards(&self) -> u64 {
self.max_shards
}
pub fn context_digest(&self) -> H256 {
H256(self.context.digest)
}
}
#[derive(Clone, Debug, Default)]
pub struct MineRangeConfig {
@ -89,7 +108,7 @@ impl PoraService {
pub fn spawn(
executor: TaskExecutor,
msg_recv: broadcast::Receiver<MinerMessage>,
mine_context_receiver: mpsc::UnboundedReceiver<MineContextMessage>,
mine_context_receiver: broadcast::Receiver<MineContextMessage>,
loader: Arc<dyn PoraLoader>,
config: &MinerConfig,
miner_id: H256,
@ -138,15 +157,19 @@ impl PoraService {
Ok(MinerMessage::SetStartPosition(pos)) => {
info!("Change start position to: {:?}", pos);
self.mine_range.start_position = pos;
self.report_reason_if_mine_stop("update mine range");
}
Ok(MinerMessage::SetEndPosition(pos)) => {
info!("Change end position to: {:?}", pos);
self.mine_range.end_position = pos;
self.report_reason_if_mine_stop("update mine range");
}
Ok(MinerMessage::SetShardConfig(shard_config)) => {
self.mine_range.shard_config = shard_config;
self.report_reason_if_mine_stop("update shard");
}
Err(broadcast::error::RecvError::Closed)=>{
Err(broadcast::error::RecvError::Closed) => {
warn!("Unexpected: Mine service config channel closed.");
channel_opened = false;
}
@ -157,21 +180,26 @@ impl PoraService {
}
maybe_msg = self.mine_context_receiver.recv() => {
if let Some(msg) = maybe_msg {
info!("Update mine service: {:?}", msg);
info!("Mine iterations statistics: {}", metrics::report());
self.puzzle = msg.map(|(context, target_quality)| PoraPuzzle {
context, target_quality
});
} else {
warn!("Mine context channel closed.");
match maybe_msg {
Ok(msg) => {
info!("Update mine service: {:?}", msg);
self.puzzle = msg;
self.report_reason_if_mine_stop("update mine context");
},
Err(broadcast::error::RecvError::Closed) => {
warn!("Mine context channel closed.");
},
Err(_) => {}
}
}
() = &mut diastole, if !diastole.is_elapsed() => {
}
_ = async {}, if mining_enabled && cpu_percent > 0 && self.as_miner().map_or(false, |miner| miner.range.mining_length > 0) && diastole.is_elapsed() => {
_ = async {}, if mining_enabled
&& cpu_percent > 0
&& self.as_miner().is_ok()
&& diastole.is_elapsed() => {
let nonce = H256(rand::thread_rng().gen());
let miner = self.as_miner().unwrap();
@ -194,13 +222,27 @@ impl PoraService {
}
#[inline]
fn as_miner(&self) -> Option<Miner> {
let puzzle = self.puzzle.as_ref()?;
fn as_miner(&self) -> Result<Miner, &'static str> {
let puzzle = self.puzzle.as_ref().ok_or("no mine context")?;
let range = self.mine_range.to_valid_range(&puzzle.context)?;
(range.mining_length > 0).then_some(())?;
let range = self
.mine_range
.to_valid_range(&puzzle.context)
.ok_or("no mine range")?;
Some(Miner {
if range.mining_length == 0 {
return Err("mine range is zero");
}
if puzzle.max_shards() < self.mine_range.shard_config.num_shard as u64 {
return Err("too many mine range");
}
if self.mine_range.shard_config.num_shard as u64 > puzzle.context.flow_length.as_u64() {
return Err("Not enough flow length to shard");
}
Ok(Miner {
range,
miner_id: &self.miner_id,
mine_range_config: &self.mine_range,
@ -209,4 +251,10 @@ impl PoraService {
loader: &*self.loader,
})
}
fn report_reason_if_mine_stop(&self, event: &'static str) {
if let Err(reason) = self.as_miner() {
info!(reason, "Mine stopped on {}", event);
}
}
}

27
node/miner/src/monitor.rs Normal file
View File

@ -0,0 +1,27 @@
use std::time::Duration;
use task_executor::TaskExecutor;
use tokio::time::sleep;
use super::metrics;
pub struct Monitor {
period: Duration,
}
impl Monitor {
pub fn spawn(executor: TaskExecutor, period: Duration) {
let monitor = Monitor { period };
executor.spawn(
async move { Box::pin(monitor.start()).await },
"pora_master",
);
}
async fn start(&self) {
loop {
info!("Mine iterations statistics: {}", metrics::report());
let _ = sleep(self.period).await;
}
}
}

View File

@ -102,15 +102,16 @@ impl<'a> Miner<'a> {
}
let quality = self.pora(idx, &sealed_data, pad_seed);
let quality_scale = self.range.shard_mask.count_zeros();
if quality <= U256::MAX >> quality_scale
&& quality << quality_scale <= *self.target_quality
{
let difficulty_scale_x64 = self
.range
.difficulty_scale_x64(self.context.flow_length.as_u64());
if quality <= (self.target_quality / difficulty_scale_x64) << 64 {
debug!(
"Find a PoRA valid answer, quality: {}, target_quality {}, scale {}",
"Find a PoRA valid answer, quality: {}, target_quality {}, scale {:.3}",
U256::MAX / quality,
U256::MAX / self.target_quality,
quality_scale
difficulty_scale_x64.as_u128() as f64 / (u64::MAX as f64 + 1.0)
);
inc_counter(&HIT_COUNT);
// Undo mix data when find a valid solition

View File

@ -1,6 +1,6 @@
use ethereum_types::U256;
use tiny_keccak::{Hasher, Keccak};
use zgs_spec::SECTORS_PER_LOAD;
use zgs_spec::{SECTORS_PER_LOAD, SECTORS_PER_MAX_MINING_RANGE};
#[derive(PartialEq, Eq, Clone, Copy, Debug)]
pub struct RecallRange {
@ -38,6 +38,16 @@ impl RecallRange {
Some(self.start_position + recall_offset * SECTORS_PER_LOAD as u64)
}
pub fn difficulty_scale_x64(&self, flow_length: u64) -> U256 {
let no_shard_mine_length = std::cmp::min(flow_length, SECTORS_PER_MAX_MINING_RANGE as u64);
let sharded_mine_length = std::cmp::min(
flow_length >> self.shard_mask.count_zeros(),
SECTORS_PER_MAX_MINING_RANGE as u64,
);
(U256::from(no_shard_mine_length) << 64) / sharded_mine_length
}
}
impl From<RecallRange> for contract_interface::RecallRange {

View File

@ -1,9 +1,11 @@
use crate::miner_id::check_and_request_miner_id;
use crate::monitor::Monitor;
use crate::sealer::Sealer;
use crate::submitter::Submitter;
use crate::{config::MinerConfig, mine::PoraService, watcher::MineContextWatcher};
use network::NetworkMessage;
use std::sync::Arc;
use std::time::Duration;
use storage::config::ShardConfig;
use storage_async::Store;
use tokio::sync::broadcast;
@ -48,7 +50,7 @@ impl MineService {
let mine_answer_receiver = PoraService::spawn(
executor.clone(),
msg_recv.resubscribe(),
mine_context_receiver,
mine_context_receiver.resubscribe(),
store.clone(),
&config,
miner_id,
@ -57,12 +59,15 @@ impl MineService {
Submitter::spawn(
executor.clone(),
mine_answer_receiver,
mine_context_receiver,
provider.clone(),
store.clone(),
&config,
);
Sealer::spawn(executor, provider, store, &config, miner_id);
Sealer::spawn(executor.clone(), provider, store, &config, miner_id);
Monitor::spawn(executor, Duration::from_secs(5));
debug!("Starting miner service");

View File

@ -6,19 +6,23 @@ use ethers::providers::PendingTransaction;
use hex::ToHex;
use shared_types::FlowRangeProof;
use std::sync::Arc;
use std::time::Duration;
use storage::H256;
use storage_async::Store;
use task_executor::TaskExecutor;
use tokio::sync::mpsc;
use tokio::sync::{broadcast, mpsc};
use crate::config::{MineServiceMiddleware, MinerConfig};
use crate::pora::AnswerWithoutProof;
use crate::watcher::MineContextMessage;
use zgs_spec::{BYTES_PER_SEAL, SECTORS_PER_SEAL};
const SUBMISSION_RETIES: usize = 3;
const SUBMISSION_RETIES: usize = 15;
pub struct Submitter {
mine_answer_receiver: mpsc::UnboundedReceiver<AnswerWithoutProof>,
mine_context_receiver: broadcast::Receiver<MineContextMessage>,
mine_contract: PoraMine<MineServiceMiddleware>,
flow_contract: ZgsFlow<MineServiceMiddleware>,
default_gas_limit: Option<U256>,
@ -29,6 +33,7 @@ impl Submitter {
pub fn spawn(
executor: TaskExecutor,
mine_answer_receiver: mpsc::UnboundedReceiver<AnswerWithoutProof>,
mine_context_receiver: broadcast::Receiver<MineContextMessage>,
provider: Arc<MineServiceMiddleware>,
store: Arc<Store>,
config: &MinerConfig,
@ -39,6 +44,7 @@ impl Submitter {
let submitter = Submitter {
mine_answer_receiver,
mine_context_receiver,
mine_contract,
flow_contract,
store,
@ -51,18 +57,39 @@ impl Submitter {
}
async fn start(mut self) {
let mut current_context_digest: Option<H256> = None;
loop {
match self.mine_answer_receiver.recv().await {
Some(answer) => {
if let Err(e) = self.submit_answer(answer).await {
warn!(e)
tokio::select! {
answer_msg = self.mine_answer_receiver.recv() => {
match answer_msg {
Some(answer) => {
if Some(answer.context_digest) != current_context_digest {
info!("Skip submission because of inconsistent context digest");
continue;
}
if let Err(e) = self.submit_answer(answer).await {
warn!(e);
}
}
None => {
warn!("Mine submitter stopped because mine answer channel is closed.");
return;
}
}
}
None => {
warn!("Mine submitter stopped because mine answer channel is closed.");
break;
context_msg = self.mine_context_receiver.recv() => {
match context_msg {
Ok(puzzle) => {
current_context_digest = puzzle.map(|p| p.context_digest());
}
Err(broadcast::error::RecvError::Closed) => {
warn!("Mine context channel closed.");
},
Err(_) => {}
}
}
};
}
}
}
@ -136,6 +163,7 @@ impl Submitter {
let receipt = pending_transaction
.retries(SUBMISSION_RETIES)
.interval(Duration::from_secs(2))
.await
.map_err(|e| format!("Fail to execute mine answer transaction: {:?}", e))?
.ok_or(format!(

View File

@ -19,9 +19,9 @@ use std::sync::Arc;
use std::time::Duration;
use std::{ops::DerefMut, str::FromStr};
use crate::{config::MineServiceMiddleware, MinerConfig, MinerMessage};
use crate::{config::MineServiceMiddleware, mine::PoraPuzzle, MinerConfig, MinerMessage};
pub type MineContextMessage = Option<(MineContext, U256)>;
pub type MineContextMessage = Option<PoraPuzzle>;
lazy_static! {
pub static ref EMPTY_HASH: H256 =
@ -33,7 +33,7 @@ pub struct MineContextWatcher {
flow_contract: ZgsFlow<MineServiceMiddleware>,
mine_contract: PoraMine<MineServiceMiddleware>,
mine_context_sender: mpsc::UnboundedSender<MineContextMessage>,
mine_context_sender: broadcast::Sender<MineContextMessage>,
last_report: MineContextMessage,
msg_recv: broadcast::Receiver<MinerMessage>,
@ -45,14 +45,14 @@ impl MineContextWatcher {
msg_recv: broadcast::Receiver<MinerMessage>,
provider: Arc<MineServiceMiddleware>,
config: &MinerConfig,
) -> mpsc::UnboundedReceiver<MineContextMessage> {
) -> broadcast::Receiver<MineContextMessage> {
let provider = provider;
let mine_contract = PoraMine::new(config.mine_address, provider.clone());
let flow_contract = ZgsFlow::new(config.flow_address, provider.clone());
let (mine_context_sender, mine_context_receiver) =
mpsc::unbounded_channel::<MineContextMessage>();
broadcast::channel::<MineContextMessage>(4096);
let watcher = MineContextWatcher {
provider,
flow_contract,
@ -108,12 +108,17 @@ impl MineContextWatcher {
let context_call = self.flow_contract.make_context_with_result();
let valid_call = self.mine_contract.can_submit();
let quality_call = self.mine_contract.pora_target();
let shards_call = self.mine_contract.max_shards();
let (context, can_submit, quality) =
try_join!(context_call.call(), valid_call.call(), quality_call.call())
.map_err(|e| format!("Failed to query mining context: {:?}", e))?;
let (context, can_submit, quality, max_shards) = try_join!(
context_call.call(),
valid_call.call(),
quality_call.call(),
shards_call.call()
)
.map_err(|e| format!("Failed to query mining context: {:?}", e))?;
let report = if can_submit && context.digest != EMPTY_HASH.0 {
Some((context, quality))
Some(PoraPuzzle::new(context, quality, max_shards))
} else {
None
};

View File

@ -8,5 +8,5 @@ check_abis() {
diff $(./scripts/search_abi.sh "$artifacts_path" "$contract_name.json") "storage-contracts-abis/$contract_name.json"
done
}
check_abis DummyMarket DummyReward Flow PoraMine PoraMineTest FixedPrice OnePoolReward FixedPriceFlow
check_abis DummyMarket DummyReward Flow PoraMine PoraMineTest FixedPrice ChunkLinearReward FixedPriceFlow

View File

@ -54,7 +54,7 @@ copy_abis() {
done
}
copy_abis DummyMarket DummyReward Flow PoraMine PoraMineTest FixedPrice OnePoolReward FixedPriceFlow
copy_abis DummyMarket DummyReward Flow PoraMine PoraMineTest FixedPrice ChunkLinearReward FixedPriceFlow
# Step 4: Get the current Git revision and write it to a specified file

View File

@ -1 +1 @@
6ec57942cba3068d3373bcb2c88f8fd51d2322b1
dbeff538b949599c203e43be6ecc05e9e997d09d

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

View File

@ -45,14 +45,14 @@ class MineTest(TestFramework):
self.log.info("Submission done, current epoch is %d", start_epoch)
self.log.info("Wait for the first mine context release")
wait_until(lambda: int(blockchain.eth_blockNumber(), 16) >= start_epoch + 1, timeout=180)
wait_until(lambda: self.contract.epoch() >= start_epoch + 1, timeout=180)
self.contract.update_context()
self.log.info("Wait for the first mine answer")
wait_until(lambda: self.mine_contract.last_mined_epoch() == start_epoch + 1 and not self.mine_contract.can_submit(), timeout=180)
self.log.info("Wait for the second mine context release")
wait_until(lambda: int(blockchain.eth_blockNumber(), 16) >= start_epoch + 2, timeout=180)
wait_until(lambda: self.contract.epoch() >= start_epoch + 2, timeout=180)
self.contract.update_context()
self.log.info("Wait for the second mine answer")
@ -60,7 +60,7 @@ class MineTest(TestFramework):
self.nodes[0].miner_stop()
self.log.info("Wait for the third mine context release")
wait_until(lambda: int(blockchain.eth_blockNumber(), 16) >= start_epoch + 3, timeout=180)
wait_until(lambda: self.contract.epoch() >= start_epoch + 3, timeout=180)
self.contract.update_context()
self.log.info("Submit the second data chunk")

View File

@ -50,6 +50,7 @@ class MineTest(TestFramework):
SECTORS_PER_PRICING = int(8 * ( 2 ** 30 ) / 256)
self.log.info("Submit the actual data chunk (256 MB)")
self.submit_data(b"\x11", int(SECTORS_PER_PRICING / 32))
@ -74,7 +75,7 @@ class MineTest(TestFramework):
firstReward = rewards[0].args.amount
self.log.info("Received reward %d Gwei", firstReward / (10**9))
self.reward_contract.transfer(10000 * 10 ** 18)
self.reward_contract.donate(10000 * 10 ** 18)
self.log.info("Donation Done")
self.log.info("Submit the data hash only (8 GB)")
self.submit_data(b"\x11", int(SECTORS_PER_PRICING), no_submit=True)

View File

@ -303,7 +303,7 @@ class BlockchainNode(TestNode):
mine_contract.functions.setTargetSubmissions(2).transact(TX_PARAMS)
self.log.debug("Mine Initialized")
flow_initialize_hash = (flow_contract.get_function_by_signature('initialize(address)'))(dummy_market_contract.address).transact(TX_PARAMS)
flow_initialize_hash = flow_contract.functions.initialize(dummy_market_contract.address).transact(TX_PARAMS)
self.log.debug("Flow Initialized")
self.wait_for_transaction_receipt(w3, flow_initialize_hash)
@ -325,7 +325,7 @@ class BlockchainNode(TestNode):
market_contract, _ = deploy_contract("FixedPrice", [])
self.log.debug("Market deployed")
reward_contract, _ =deploy_contract("OnePoolReward", [LIFETIME_MONTH])
reward_contract, _ =deploy_contract("ChunkLinearReward", [LIFETIME_MONTH * 31 * 86400])
self.log.debug("Reward deployed")
flow_contract, _ = deploy_contract("FixedPriceFlow", [mine_period, 0])
@ -336,13 +336,15 @@ class BlockchainNode(TestNode):
mine_contract.functions.setTargetSubmissions(2).transact(TX_PARAMS)
self.log.debug("Mine Initialized")
market_contract.functions.initialize(LIFETIME_MONTH, flow_contract.address, reward_contract.address).transact(TX_PARAMS)
price_per_sector = int(LIFETIME_MONTH * 256 * 10 * 1_000_000_000_000_000_000 / 1024 / 1024 / 1024 / 12)
market_contract.functions.initialize(price_per_sector, flow_contract.address, reward_contract.address).transact(TX_PARAMS)
self.log.debug("Market Initialized")
reward_contract.functions.initialize(market_contract.address, mine_contract.address).transact(TX_PARAMS)
reward_contract.functions.setBaseReward(10 ** 18).transact(TX_PARAMS)
self.log.debug("Reward Initialized")
flow_initialize_hash = (flow_contract.get_function_by_signature('initialize(address)'))(market_contract.address).transact(TX_PARAMS)
flow_initialize_hash = flow_contract.functions.initialize(market_contract.address).transact(TX_PARAMS)
self.log.debug("Flow Initialized")
self.wait_for_transaction_receipt(w3, flow_initialize_hash)

View File

@ -28,6 +28,14 @@ class ContractProxy:
contract = self._get_contract(node_idx)
return getattr(contract.functions, fn_name)(**args).transact(copy(TX_PARAMS))
def _send_payable(self, fn_name, node_idx, value, **args):
assert node_idx < len(self.blockchain_nodes)
contract = self._get_contract(node_idx)
tx_params = copy(TX_PARAMS)
tx_params["value"] = value
return getattr(contract.functions, fn_name)(**args).transact(tx_params)
def _logs(self, event_name, node_idx, **args):
assert node_idx < len(self.blockchain_nodes)
@ -66,7 +74,6 @@ class FlowContractProxy(ContractProxy):
contract.w3, tx_hash, parent_hash=parent_hash
)
if receipt["status"] != 1:
print(receipt)
assert_equal(receipt["status"], 1)
return tx_hash
@ -98,6 +105,12 @@ class MineContractProxy(ContractProxy):
class IRewardContractProxy(ContractProxy):
class RewardContractProxy(ContractProxy):
def reward_distributes(self, node_idx=0):
return self._logs("DistributeReward", node_idx)
return self._logs("DistributeReward", node_idx)
def donate(self, value, node_idx = 0):
return self._send_payable("donate", node_idx, value)
def base_reward(self, node_idx = 0):
return self._call("baseReward", node_idx)

View File

@ -15,7 +15,7 @@ from pathlib import Path
from eth_utils import encode_hex
from test_framework.bsc_node import BSCNode
from test_framework.contract_proxy import FlowContractProxy, MineContractProxy, IRewardContractProxy
from test_framework.contract_proxy import FlowContractProxy, MineContractProxy, RewardContractProxy
from test_framework.zgs_node import ZgsNode
from test_framework.blockchain_node import BlockChainNodeType
from test_framework.conflux_node import ConfluxNode, connect_sample_nodes
@ -174,7 +174,7 @@ class TestFramework:
contract, tx_hash, mine_contract, reward_contract = self.blockchain_nodes[0].setup_contract(self.enable_market, self.mine_period)
self.contract = FlowContractProxy(contract, self.blockchain_nodes)
self.mine_contract = MineContractProxy(mine_contract, self.blockchain_nodes)
self.reward_contract = IRewardContractProxy(reward_contract, self.blockchain_nodes)
self.reward_contract = RewardContractProxy(reward_contract, self.blockchain_nodes)
for node in self.blockchain_nodes[1:]: