Merge pull request paritytech#91 from subspace/farmer-lib-phase3b

Farmer lib phase3b

Author: ozgunozerk

Cargo.lock

@@ -16,6 +16,7 @@ include = [
 [dependencies]
 anyhow = "1.0.44"
 async-lock = "2.4.0"
+async-oneshot = "0.5.0"
 async-std = "1.9.0"
 clap = { version = "3.0.0-beta.5", features = ["color", "derive"] }
 dirs = "4.0.0"

crates/subspace-farmer/Cargo.toml

@@ -1,24 +1,18 @@
 use crate::commitments::Commitments;
-use crate::common::Salt;
+use crate::farming::Farming;
 use crate::identity::Identity;
 use crate::object_mappings::ObjectMappings;
 use crate::plot::Plot;
-use crate::rpc::{
-    EncodedBlockWithObjectMapping, FarmerMetadata, ProposedProofOfReplicationResponse, RpcClient,
-    SlotInfo, Solution,
-};
+use crate::rpc::{EncodedBlockWithObjectMapping, FarmerMetadata, RpcClient};
 use anyhow::{anyhow, Result};
-use futures::future;
-use futures::future::Either;
-use log::{debug, error, info, trace};
+use log::{debug, error, info};
 use std::path::PathBuf;
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
-use std::time::Instant;
 use subspace_archiving::archiver::{ArchivedSegment, BlockArchiver, ObjectArchiver};
 use subspace_archiving::pre_genesis_data;
 use subspace_core_primitives::objects::{GlobalObject, PieceObject, PieceObjectMapping};
-use subspace_core_primitives::{crypto, Sha256Hash};
+use subspace_core_primitives::Sha256Hash;
 use subspace_solving::SubspaceCodec;
 
 /// Start farming by using plot in specified path and connecting to WebSocket server at specified
@@ -45,32 +39,29 @@ pub async fn farm(base_directory: PathBuf, ws_server: &str) -> Result<()> {
 
     let identity = Identity::open_or_create(&base_directory)?;
 
-    match future::select(
-        {
-            let client = client.clone();
-            let plot = plot.clone();
-            let commitments = commitments.clone();
-            let public_key = identity.public_key();
-
-            Box::pin(async move {
-                background_plotting(client, plot, commitments, object_mappings, &public_key).await
-            })
-        },
-        Box::pin(
-            async move { subscribe_to_slot_info(&client, &plot, &commitments, &identity).await },
-        ),
-    )
-    .await
-    {
-        Either::Left((result, _)) => match result {
-            Ok(()) => {
-                info!("Background plotting finished successfully");
-
-                Ok(())
-            }
-            Err(error) => Err(anyhow!("Background plotting error: {}", error)),
-        },
-        Either::Right((result, _)) => result.map_err(Into::into),
+    // start the farming task
+    // right now the instance is unused, however, if we want to call stop the process
+    // we can just drop the instance, and it will be stopped magically :)
+    let _farming_instance = Farming::start(
+        plot.clone(),
+        commitments.clone(),
+        client.clone(),
+        identity.clone(),
+    );
+
+    // start the background plotting
+    // NOTE: THIS WILL CHANGE IN THE UPCOMING PR
+    let public_key = identity.public_key();
+    let plotting_result =
+        background_plotting(client, plot, commitments, object_mappings, &public_key).await;
+
+    match plotting_result {
+        Ok(()) => {
+            info!("Background plotting shutdown gracefully");
+
+            Ok(())
+        }
+        Err(error) => Err(anyhow!("Background plotting error: {}", error)),
     }
 }
 
@@ -383,144 +374,3 @@ fn create_global_object_mapping(
         })
         .collect()
 }
-
-async fn subscribe_to_slot_info(
-    client: &RpcClient,
-    plot: &Plot,
-    commitments: &Commitments,
-    identity: &Identity,
-) -> Result<()> {
-    let farmer_public_key_hash = crypto::sha256_hash(&identity.public_key());
-
-    info!("Subscribing to slot info");
-    let mut new_slots = client.subscribe_slot_info().await?;
-
-    let mut salts = Salts::default();
-
-    while let Some(slot_info) = new_slots.next().await? {
-        debug!("New slot: {:?}", slot_info);
-
-        update_commitments(plot, commitments, &mut salts, &slot_info);
-
-        let local_challenge =
-            subspace_solving::derive_local_challenge(slot_info.challenge, &farmer_public_key_hash);
-
-        let solution = match commitments
-            .find_by_range(local_challenge, slot_info.solution_range, slot_info.salt)
-            .await
-        {
-            Some((tag, piece_index)) => {
-                let encoding = plot.read(piece_index).await?;
-                let solution = Solution::new(
-                    identity.public_key().to_bytes(),
-                    piece_index,
-                    encoding.to_vec(),
-                    identity.sign(&tag).to_bytes().to_vec(),
-                    tag,
-                );
-                debug!("Solution found");
-                trace!("Solution found: {:?}", solution);
-
-                Some(solution)
-            }
-            None => {
-                debug!("Solution not found");
-                None
-            }
-        };
-
-        client
-            .propose_proof_of_replication(ProposedProofOfReplicationResponse {
-                slot_number: slot_info.slot_number,
-                solution,
-                secret_key: identity.secret_key().to_bytes().into(),
-            })
-            .await?;
-    }
-
-    Ok(())
-}
-
-#[derive(Default)]
-struct Salts {
-    current: Option<Salt>,
-    next: Option<Salt>,
-}
-
-/// Compare salts in `slot_info` to those known from `salts` and start update plot commitments
-/// accordingly if necessary (in background)
-fn update_commitments(
-    plot: &Plot,
-    commitments: &Commitments,
-    salts: &mut Salts,
-    slot_info: &SlotInfo,
-) {
-    // Check if current salt has changed
-    if salts.current != Some(slot_info.salt) {
-        salts.current.replace(slot_info.salt);
-
-        if salts.next != Some(slot_info.salt) {
-            // If previous `salts.next` is the same as current (expected behavior), need to re-commit
-
-            tokio::spawn({
-                let salt = slot_info.salt;
-                let plot = plot.clone();
-                let commitments = commitments.clone();
-
-                async move {
-                    let started = Instant::now();
-                    info!(
-                        "Salt updated to {}, recommitting in background",
-                        hex::encode(salt)
-                    );
-
-                    if let Err(error) = commitments.create(salt, plot).await {
-                        error!(
-                            "Failed to create commitment for {}: {}",
-                            hex::encode(salt),
-                            error
-                        );
-                    } else {
-                        info!(
-                            "Finished recommitment for {} in {} seconds",
-                            hex::encode(salt),
-                            started.elapsed().as_secs_f32()
-                        );
-                    }
-                }
-            });
-        }
-    }
-
-    if let Some(new_next_salt) = slot_info.next_salt {
-        if salts.next != Some(new_next_salt) {
-            salts.next.replace(new_next_salt);
-
-            tokio::spawn({
-                let plot = plot.clone();
-                let commitments = commitments.clone();
-
-                async move {
-                    let started = Instant::now();
-                    info!(
-                        "Salt will update to {} soon, recommitting in background",
-                        hex::encode(new_next_salt)
-                    );
-                    if let Err(error) = commitments.create(new_next_salt, plot).await {
-                        error!(
-                            "Recommitting salt in background failed for {}: {}",
-                            hex::encode(new_next_salt),
-                            error
-                        );
-                        return;
-                    }
-                    info!(
-                        "Finished recommitment in background for {} in {} seconds",
-                        hex::encode(new_next_salt),
-                        started.elapsed().as_secs_f32()
-                    );
-                }
-            });
-        }
-    }
-}