[r2r] refactor p2p message processing flow

Dockerfile

@@ -48,6 +48,9 @@ RUN cd /mm2 && cargo fetch
 # Only needed when we're developing or changing something locally.
 #COPY . /mm2
 
+# Important for x86_64 builds
+ENV JEMALLOC_SYS_WITH_MALLOC_CONF="background_thread:true,narenas:1,tcache:false,dirty_decay_ms:0,muzzy_decay_ms:0,metadata_thp:auto"
+
 # Build MM1 and MM2.
 # Increased verbosity here allows us to see the MM1 CMake logs.
 RUN cd /mm2 &&\

README.md

@@ -87,7 +87,13 @@ If you want to build from source, the following prerequisites are required:
     rustup default nightly-2022-02-01
     rustup component add rustfmt-preview
     ```
-
+
+**Note for x86_64 UNIX systems**:
+To have more efficient memory consumption please execute the following command before building mm2. (It's also good to have before launching mm2.)
+```sh
+export JEMALLOC_SYS_WITH_MALLOC_CONF="background_thread:true,narenas:1,tcache:false,dirty_decay_ms:0,muzzy_decay_ms:0,metadata_thp:auto"
+```
+
 To build, run `cargo build` (or `cargo build -vv` to get verbose build output).
 
 For more detailed instructions, please refer to the [Installation Guide](https://developers.komodoplatform.com/basic-docs/atomicdex/atomicdex-setup/get-started-atomicdex.html). 

azure-pipelines-build-stage-job.yml

@@ -50,6 +50,7 @@ jobs:
         condition: ne ( variables['Build.Reason'], 'PullRequest' )
         env:
           MANUAL_MM_VERSION: true
+          JEMALLOC_SYS_WITH_MALLOC_CONF: "background_thread:true,narenas:1,tcache:false,dirty_decay_ms:0,muzzy_decay_ms:0,metadata_thp:auto"
       - task: Docker@2
         displayName: Build & Push container of dev branch
         condition: and( eq( variables['Agent.OS'], 'Linux' ), eq( variables['Build.SourceBranchName'], 'dev' ) )

azure-pipelines-release-stage-job.yml

@@ -67,6 +67,7 @@ jobs:
         condition: eq( variables['DEBUG_UPLOADED'], '' )
         env:
           MANUAL_MM_VERSION: true
+          JEMALLOC_SYS_WITH_MALLOC_CONF: "background_thread:true,narenas:1,tcache:false,dirty_decay_ms:0,muzzy_decay_ms:0,metadata_thp:auto"
       - bash: |
           zip upload/mm2-$(COMMIT_HASH)-$(Agent.OS)-Debug target-xenial/debug/mm2 target-xenial/debug/libmm2.a -j
         displayName: 'Prepare debug build upload Linux'

mm2src/coins/eth.rs

@@ -1377,8 +1377,10 @@ impl MarketCoinOps for EthCoin {
         Box::new(fut.boxed().compat())
     }
 
-    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, String> {
-        Ok(try_s!(signed_eth_tx_from_bytes(bytes)).into())
+    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, MmError<TransactionErr>> {
+        signed_eth_tx_from_bytes(bytes)
+            .map(TransactionEnum::from)
+            .map_to_mm(TransactionErr::InvalidTx)
     }
 
     fn current_block(&self) -> Box<dyn Future<Item = u64, Error = String> + Send> {

mm2src/coins/lightning.rs

@@ -19,9 +19,9 @@ use crate::utxo::{sat_from_big_decimal, BlockchainNetwork, FeePolicy, GetUtxoLis
 use crate::{BalanceFut, CoinBalance, FeeApproxStage, FoundSwapTxSpend, HistorySyncState, MarketCoinOps, MmCoin,
             NegotiateSwapContractAddrErr, RawTransactionFut, RawTransactionRequest, SearchForSwapTxSpendInput,
             SignatureError, SignatureResult, SwapOps, TradeFee, TradePreimageFut, TradePreimageResult,
-            TradePreimageValue, TransactionEnum, TransactionFut, UnexpectedDerivationMethod, UtxoStandardCoin,
-            ValidateAddressResult, ValidatePaymentInput, VerificationError, VerificationResult, WithdrawError,
-            WithdrawFut, WithdrawRequest};
+            TradePreimageValue, TransactionEnum, TransactionErr, TransactionFut, UnexpectedDerivationMethod,
+            UtxoStandardCoin, ValidateAddressResult, ValidatePaymentInput, VerificationError, VerificationResult,
+            WithdrawError, WithdrawFut, WithdrawRequest};
 use async_trait::async_trait;
 use bitcoin::hashes::Hash;
 use bitcoin_hashes::sha256::Hash as Sha256;
@@ -482,7 +482,11 @@ impl MarketCoinOps for LightningCoin {
     }
 
     // Todo: Implement this when implementing swaps for lightning as it's is used mainly for swaps
-    fn tx_enum_from_bytes(&self, _bytes: &[u8]) -> Result<TransactionEnum, String> { unimplemented!() }
+    fn tx_enum_from_bytes(&self, _bytes: &[u8]) -> Result<TransactionEnum, MmError<TransactionErr>> {
+        MmError::err(TransactionErr::Plain(
+            "tx_enum_from_bytes is not supported for Lightning yet.".to_string(),
+        ))
+    }
 
     fn current_block(&self) -> Box<dyn Future<Item = u64, Error = String> + Send> { Box::new(futures01::future::ok(0)) }
 

mm2src/coins/lp_coins.rs

@@ -367,6 +367,7 @@ pub enum TransactionErr {
     TxRecoverable(TransactionEnum, String),
     /// Simply for plain error messages.
     Plain(String),
+    InvalidTx(String),
 }
 
 impl TransactionErr {
@@ -385,6 +386,7 @@ impl TransactionErr {
         match self {
             TransactionErr::TxRecoverable(_, err) => err.to_string(),
             TransactionErr::Plain(err) => err.to_string(),
+            TransactionErr::InvalidTx(err) => err.to_string(),
         }
     }
 }
@@ -615,7 +617,7 @@ pub trait MarketCoinOps {
         swap_contract_address: &Option<BytesJson>,
     ) -> TransactionFut;
 
-    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, String>;
+    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, MmError<TransactionErr>>;
 
     fn current_block(&self) -> Box<dyn Future<Item = u64, Error = String> + Send>;
 

mm2src/coins/qrc20.rs

@@ -1111,7 +1111,7 @@ impl MarketCoinOps for Qrc20Coin {
         Box::new(fut.boxed().compat())
     }
 
-    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, String> {
+    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, MmError<TransactionErr>> {
         utxo_common::tx_enum_from_bytes(self.as_ref(), bytes)
     }
 

mm2src/coins/solana.rs

@@ -3,9 +3,9 @@ use crate::solana::solana_common::{lamports_to_sol, PrepareTransferData, Suffici
 use crate::solana::spl::SplTokenInfo;
 use crate::{BalanceError, BalanceFut, FeeApproxStage, FoundSwapTxSpend, NegotiateSwapContractAddrErr,
             RawTransactionFut, RawTransactionRequest, SearchForSwapTxSpendInput, SignatureResult, TradePreimageFut,
-            TradePreimageResult, TradePreimageValue, TransactionDetails, TransactionFut, TransactionType,
-            UnexpectedDerivationMethod, ValidateAddressResult, ValidatePaymentInput, VerificationResult,
-            WithdrawError, WithdrawFut, WithdrawRequest, WithdrawResult};
+            TradePreimageResult, TradePreimageValue, TransactionDetails, TransactionErr, TransactionFut,
+            TransactionType, UnexpectedDerivationMethod, ValidateAddressResult, ValidatePaymentInput,
+            VerificationResult, WithdrawError, WithdrawFut, WithdrawRequest, WithdrawResult};
 use async_trait::async_trait;
 use base58::ToBase58;
 use bincode::{deserialize, serialize};
@@ -436,7 +436,11 @@ impl MarketCoinOps for SolanaCoin {
         unimplemented!()
     }
 
-    fn tx_enum_from_bytes(&self, _bytes: &[u8]) -> Result<TransactionEnum, String> { unimplemented!() }
+    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, MmError<TransactionErr>> {
+        MmError::err(TransactionErr::Plain(
+            "tx_enum_from_bytes is not supported for Solana yet.".to_string(),
+        ))
+    }
 
     fn current_block(&self) -> Box<dyn Future<Item = u64, Error = String> + Send> {
         let coin = self.clone();

mm2src/coins/solana/spl.rs

@@ -3,9 +3,9 @@ use crate::solana::solana_common::{ui_amount_to_amount, PrepareTransferData, Suf
 use crate::solana::{solana_common, AccountError, SolanaCommonOps, SolanaFeeDetails};
 use crate::{BalanceFut, FeeApproxStage, FoundSwapTxSpend, NegotiateSwapContractAddrErr, RawTransactionFut,
             RawTransactionRequest, SearchForSwapTxSpendInput, SignatureResult, SolanaCoin, TradePreimageFut,
-            TradePreimageResult, TradePreimageValue, TransactionDetails, TransactionFut, TransactionType,
-            UnexpectedDerivationMethod, ValidateAddressResult, ValidatePaymentInput, VerificationResult,
-            WithdrawError, WithdrawFut, WithdrawRequest, WithdrawResult};
+            TradePreimageResult, TradePreimageValue, TransactionDetails, TransactionErr, TransactionFut,
+            TransactionType, UnexpectedDerivationMethod, ValidateAddressResult, ValidatePaymentInput,
+            VerificationResult, WithdrawError, WithdrawFut, WithdrawRequest, WithdrawResult};
 use async_trait::async_trait;
 use bincode::serialize;
 use common::{async_blocking, now_ms};
@@ -267,7 +267,11 @@ impl MarketCoinOps for SplToken {
         unimplemented!()
     }
 
-    fn tx_enum_from_bytes(&self, _bytes: &[u8]) -> Result<TransactionEnum, String> { unimplemented!() }
+    fn tx_enum_from_bytes(&self, _bytes: &[u8]) -> Result<TransactionEnum, MmError<TransactionErr>> {
+        MmError::err(TransactionErr::Plain(
+            "tx_enum_from_bytes is not supported for Spl yet.".to_string(),
+        ))
+    }
 
     fn current_block(&self) -> Box<dyn Future<Item = u64, Error = String> + Send> { self.platform_coin.current_block() }
 

mm2src/coins/test_coin.rs

@@ -2,8 +2,8 @@ use super::{CoinBalance, HistorySyncState, MarketCoinOps, MmCoin, RawTransaction
             TradeFee, TransactionEnum, TransactionFut};
 use crate::{BalanceFut, CanRefundHtlc, FeeApproxStage, FoundSwapTxSpend, NegotiateSwapContractAddrErr,
             SearchForSwapTxSpendInput, SignatureResult, TradePreimageFut, TradePreimageResult, TradePreimageValue,
-            UnexpectedDerivationMethod, ValidateAddressResult, ValidatePaymentInput, VerificationResult, WithdrawFut,
-            WithdrawRequest};
+            TransactionErr, UnexpectedDerivationMethod, ValidateAddressResult, ValidatePaymentInput,
+            VerificationResult, WithdrawFut, WithdrawRequest};
 use async_trait::async_trait;
 use futures01::Future;
 use keys::KeyPair;
@@ -78,7 +78,11 @@ impl MarketCoinOps for TestCoin {
         unimplemented!()
     }
 
-    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, String> { unimplemented!() }
+    fn tx_enum_from_bytes(&self, _bytes: &[u8]) -> Result<TransactionEnum, MmError<TransactionErr>> {
+        MmError::err(TransactionErr::Plain(
+            "tx_enum_from_bytes is not supported for Test coin yet.".to_string(),
+        ))
+    }
 
     fn current_block(&self) -> Box<dyn Future<Item = u64, Error = String> + Send> { unimplemented!() }
 

mm2src/coins/utxo/bch.rs

@@ -1130,7 +1130,7 @@ impl MarketCoinOps for BchCoin {
         )
     }
 
-    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, String> {
+    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, MmError<TransactionErr>> {
         utxo_common::tx_enum_from_bytes(self.as_ref(), bytes)
     }
 

mm2src/coins/utxo/qtum.rs

@@ -782,7 +782,7 @@ impl MarketCoinOps for QtumCoin {
         )
     }
 
-    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, String> {
+    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, MmError<TransactionErr>> {
         utxo_common::tx_enum_from_bytes(self.as_ref(), bytes)
     }
 

mm2src/coins/utxo/slp.rs

@@ -1160,7 +1160,7 @@ impl MarketCoinOps for SlpToken {
         )
     }
 
-    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, String> {
+    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, MmError<TransactionErr>> {
         self.platform_coin.tx_enum_from_bytes(bytes)
     }
 
@@ -2016,6 +2016,7 @@ mod slp_tests {
         let err = match tx_err.clone() {
             TransactionErr::TxRecoverable(_tx, err) => err,
             TransactionErr::Plain(err) => err,
+            TransactionErr::InvalidTx(err) => err,
         };
 
         println!("{:?}", err);

mm2src/coins/utxo/utxo_common.rs

@@ -1823,8 +1823,9 @@ pub fn wait_for_output_spend(
     Box::new(fut.boxed().compat())
 }
 
-pub fn tx_enum_from_bytes(coin: &UtxoCoinFields, bytes: &[u8]) -> Result<TransactionEnum, String> {
-    let mut transaction: UtxoTx = try_s!(deserialize(bytes).map_err(|err| format!("{:?}", err)));
+pub fn tx_enum_from_bytes(coin: &UtxoCoinFields, bytes: &[u8]) -> Result<TransactionEnum, MmError<TransactionErr>> {
+    let mut transaction: UtxoTx = deserialize(bytes).map_to_mm(|e| TransactionErr::InvalidTx(e.to_string()))?;
+
     transaction.tx_hash_algo = coin.tx_hash_algo;
     Ok(transaction.into())
 }

mm2src/coins/utxo/utxo_standard.rs

@@ -539,7 +539,7 @@ impl MarketCoinOps for UtxoStandardCoin {
         )
     }
 
-    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, String> {
+    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, MmError<TransactionErr>> {
         utxo_common::tx_enum_from_bytes(self.as_ref(), bytes)
     }
 

mm2src/coins/z_coin.rs

@@ -15,8 +15,8 @@ use crate::{BalanceError, BalanceFut, CoinBalance, FeeApproxStage, FoundSwapTxSp
             MmCoin, NegotiateSwapContractAddrErr, NumConversError, PrivKeyActivationPolicy, RawTransactionFut,
             RawTransactionRequest, SearchForSwapTxSpendInput, SignatureError, SignatureResult, SwapOps, TradeFee,
             TradePreimageFut, TradePreimageResult, TradePreimageValue, TransactionDetails, TransactionEnum,
-            TransactionFut, TxFeeDetails, UnexpectedDerivationMethod, ValidateAddressResult, ValidatePaymentInput,
-            VerificationError, VerificationResult, WithdrawFut, WithdrawRequest};
+            TransactionErr, TransactionFut, TxFeeDetails, UnexpectedDerivationMethod, ValidateAddressResult,
+            ValidatePaymentInput, VerificationError, VerificationResult, WithdrawFut, WithdrawRequest};
 use crate::{Transaction, WithdrawError};
 use async_trait::async_trait;
 use bitcrypto::{dhash160, dhash256};
@@ -975,8 +975,10 @@ impl MarketCoinOps for ZCoin {
         )
     }
 
-    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, String> {
-        ZTransaction::read(bytes).map(|tx| tx.into()).map_err(|e| e.to_string())
+    fn tx_enum_from_bytes(&self, bytes: &[u8]) -> Result<TransactionEnum, MmError<TransactionErr>> {
+        ZTransaction::read(bytes)
+            .map(TransactionEnum::from)
+            .map_to_mm(|e| TransactionErr::InvalidTx(e.to_string()))
     }
 
     fn current_block(&self) -> Box<dyn Future<Item = u64, Error = String> + Send> {

mm2src/mm2_main/Cargo.toml

@@ -110,6 +110,9 @@ trie-root = "0.16.0"
 uuid = { version = "0.7", features = ["serde", "v4"] }
 wasm-timer = "0.2.4"
 
+[target.x86_64-unknown-linux-gnu.dependencies]
+jemallocator = "0.5.0"
+
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 instant = { version = "0.1.12", features = ["wasm-bindgen"] }
 js-sys = { version = "0.3.27" }

mm2src/mm2_main/src/lp_network.rs

@@ -133,12 +133,15 @@ async fn process_p2p_message(
     ctx: MmArc,
     peer_id: PeerId,
     message_id: MessageId,
-    message: GossipsubMessage,
+    mut message: GossipsubMessage,
     i_am_relay: bool,
 ) {
     let mut to_propagate = false;
     let mut orderbook_pairs = vec![];
 
+    message.topics.dedup();
+    drop_mutability!(message);
+
     for topic in message.topics {
         let mut split = topic.as_str().split(TOPIC_SEPARATOR);
         match split.next() {
@@ -154,10 +157,18 @@ async fn process_p2p_message(
             Some(lp_swap::TX_HELPER_PREFIX) => {
                 if let Some(pair) = split.next() {
                     if let Ok(Some(coin)) = lp_coinfind(&ctx, pair).await {
-                        match coin.send_raw_tx_bytes(&message.data).compat().await {
-                            Ok(id) => log::debug!("Transaction broadcasted successfully: {:?} ", id),
-                            Err(e) => log::error!("Broadcast transaction failed. {}", e),
-                        }
+                        if let Err(e) = coin.tx_enum_from_bytes(&message.data) {
+                            log::error!("Message cannot continue the process due to: {:?}", e);
+                            continue;
+                        };
+
+                        let fut = coin.send_raw_tx_bytes(&message.data);
+                        spawn(async {
+                            match fut.compat().await {
+                                Ok(id) => log::debug!("Transaction broadcasted successfully: {:?} ", id),
+                                Err(e) => log::error!("Broadcast transaction failed. {}", e),
+                            };
+                        })
                     }
                 }
             },

mm2src/mm2_main/src/lp_ordermatch.rs

@@ -891,7 +891,7 @@ fn test_alb_ordered_pair() {
 }
 
 #[allow(dead_code)]
-fn parse_orderbook_pair_from_topic(topic: &str) -> Option<(&str, &str)> {
+pub fn parse_orderbook_pair_from_topic(topic: &str) -> Option<(&str, &str)> {
     let mut split = topic.split(|maybe_sep| maybe_sep == TOPIC_SEPARATOR);
     match split.next() {
         Some(ORDERBOOK_PREFIX) => match split.next() {

mm2src/mm2_main/src/lp_swap/maker_swap.rs

@@ -620,7 +620,7 @@ impl MakerSwap {
             Ok(tx) => tx,
             Err(e) => {
                 return Ok((Some(MakerSwapCommand::Finish), vec![
-                    MakerSwapEvent::TakerFeeValidateFailed(ERRL!("{}", e).into()),
+                    MakerSwapEvent::TakerFeeValidateFailed(ERRL!("{:?}", e).into()),
                 ]))
             },
         };
@@ -790,7 +790,9 @@ impl MakerSwap {
             Ok(tx) => tx,
             Err(err) => {
                 return Ok((Some(MakerSwapCommand::RefundMakerPayment), vec![
-                    MakerSwapEvent::TakerPaymentValidateFailed(ERRL!("!taker_coin.tx_enum_from_bytes: {}", err).into()),
+                    MakerSwapEvent::TakerPaymentValidateFailed(
+                        ERRL!("!taker_coin.tx_enum_from_bytes: {:?}", err).into(),
+                    ),
                     MakerSwapEvent::MakerPaymentWaitRefundStarted {
                         wait_until: self.wait_refund_until(),
                     },

mm2src/mm2_main/src/lp_swap/taker_swap.rs

@@ -1116,7 +1116,7 @@ impl TakerSwap {
             Err(e) => {
                 return Ok((Some(TakerSwapCommand::Finish), vec![
                     TakerSwapEvent::MakerPaymentValidateFailed(
-                        ERRL!("Error parsing the 'maker-payment': {}", e).into(),
+                        ERRL!("Error parsing the 'maker-payment': {:?}", e).into(),
                     ),
                 ]))
             },