Seednodes memory leaks fix

docs/HEAPTRACK.md

@@ -0,0 +1,21 @@
+# Memory profiling MM2 with heaptrack
+1. Install dependencies required by heaptrack if they are not already installed on the system
+* extra-cmake-modules
+* Qt 5.2 or higher: Core, Widgets
+* KDE Frameworks 5: CoreAddons, I18n, ItemModels, ThreadWeaver, ConfigWidgets, KIO, IconThemes
+
+2. Install heaptrack on Ubuntu (18.04) or higher:
+```
+sudo apt install heaptrack heaptrack-gui
+```
+
+3. Use heaptrack to run MM2 binary and pass parameters as usual. An example for this would be:
+```
+heaptrack ./mm2 "{\"gui\":\"MM2GUI\",\"netid\":7777, \"userhome\":\"/${HOME#"/"}\", \"passphrase\":\"YOUR_PASSPHRASE_HERE\", \"rpc_password\":\"YOUR_PASSWORD_HERE\",\"i_am_seed\":true}" &
+```
+Running heaptrack like this writes a gzipped result file in the same folder the above command ran from. We can now take a look at using the next step.
+
+4. After running MM2 for sometime we can visualize the memory profiling results using the below command. Note that ```heaptrack.mm2.xxxx.gz``` is the name of the file generated through the above command with numbers instead of xxxx
+```
+heaptrack_gui heaptrack.mm2.xxxx.gz
+```

mm2src/common/Cargo.toml

@@ -22,6 +22,7 @@ cfg-if = "1.0"
 crossbeam = "0.7"
 derive_more = "0.99"
 findshlibs = "0.5"
+fnv = "1.0.6"
 fomat-macros = "0.2"
 futures01 = { version = "0.1", package = "futures" }
 futures = { version = "0.3", package = "futures", features = ["compat", "async-await", "thread-pool"] }
@@ -50,6 +51,7 @@ serde_repr = "0.1.6"
 ser_error = { path = "../derives/ser_error" }
 ser_error_derive = { path = "../derives/ser_error_derive" }
 uuid = { version = "0.7", features = ["serde", "v4"] }
+wasm-timer = "0.2.4"
 winapi = "0.3"
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]

mm2src/common/common.rs

@@ -100,6 +100,7 @@ pub mod mm_number;
 pub mod privkey;
 pub mod seri;
 #[path = "patterns/state_machine.rs"] pub mod state_machine;
+pub mod time_cache;
 
 #[cfg(target_arch = "wasm32")] pub mod wasm_indexed_db;
 #[cfg(target_arch = "wasm32")] pub mod wasm_rpc;

mm2src/gossipsub/src/time_cache.rs → mm2src/common/time_cache.rs

@@ -27,17 +27,17 @@ use std::collections::VecDeque;
 use std::time::Duration;
 use wasm_timer::Instant;
 
-struct ExpiringElement<Element> {
+pub struct ExpiringElement<Element> {
     /// The element that expires
-    element: Element,
+    pub element: Element,
     /// The expire time.
     expires: Instant,
 }
 
 pub struct TimeCache<Key, Value> {
     /// Mapping a key to its value together with its latest expire time (can be updated through
     /// reinserts).
-    map: FnvHashMap<Key, ExpiringElement<Value>>,
+    pub map: FnvHashMap<Key, ExpiringElement<Value>>,
     /// An ordered list of keys by expires time.
     list: VecDeque<ExpiringElement<Key>>,
     /// The time elements remain in the cache.
@@ -178,6 +178,28 @@ where
         }
     }
 
+    // Removes a certain key even if it didn't expire plus removing other expired keys
+    pub fn remove(&mut self, key: Key) -> Option<Value> {
+        let now = Instant::now();
+        let mut value = None;
+        while let Some(element) = self.list.pop_front() {
+            if element.expires > now && element.element != key {
+                self.list.push_front(element);
+                break;
+            }
+            if let Occupied(entry) = self.map.entry(element.element.clone()) {
+                if element.element == key {
+                    value = Some(entry.remove().element);
+                    break;
+                }
+                if entry.get().expires <= now {
+                    entry.remove();
+                }
+            }
+        }
+        value
+    }
+
     /// Empties the entire cache.
     #[allow(dead_code)]
     pub fn clear(&mut self) {
@@ -191,6 +213,8 @@ where
 
     pub fn len(&self) -> usize { self.map.len() }
 
+    pub fn is_empty(&self) -> bool { self.map.is_empty() }
+
     pub fn ttl(&self) -> Duration { self.ttl }
 }
 

mm2src/gossipsub/Cargo.toml

@@ -13,6 +13,7 @@ categories = ["network-programming", "asynchronous"]
 base64 = "0.11.0"
 bytes = "0.5.4"
 byteorder = "1.3.2"
+common = { path = "../common" }
 fnv = "1.0.6"
 futures = "0.3.1"
 futures_codec = "0.4.0"

mm2src/gossipsub/src/behaviour.rs

@@ -23,8 +23,8 @@ use crate::handler::GossipsubHandler;
 use crate::mcache::MessageCache;
 use crate::protocol::{GossipsubControlAction, GossipsubMessage, GossipsubSubscription, GossipsubSubscriptionAction,
                       MessageId};
-use crate::time_cache::{Entry as TimeCacheEntry, TimeCache};
 use crate::topic::{Topic, TopicHash};
+use common::time_cache::{Entry as TimeCacheEntry, TimeCache};
 use futures::prelude::*;
 use libp2p_core::{connection::ConnectionId, ConnectedPoint, Multiaddr, PeerId};
 use libp2p_swarm::{NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, PollParameters, ProtocolsHandler};

mm2src/gossipsub/src/lib.rs

@@ -141,7 +141,6 @@ mod behaviour;
 mod config;
 mod handler;
 mod mcache;
-mod time_cache;
 mod topic;
 
 mod rpc_proto {

mm2src/lp_native_dex.rs

@@ -31,8 +31,8 @@ use std::str;
 #[cfg(not(target_arch = "wasm32"))]
 use crate::mm2::database::init_and_migrate_db;
 use crate::mm2::lp_network::{lp_ports, p2p_event_process_loop, P2PContext};
-use crate::mm2::lp_ordermatch::{broadcast_maker_orders_keep_alive_loop, lp_ordermatch_loop, orders_kick_start,
-                                BalanceUpdateOrdermatchHandler};
+use crate::mm2::lp_ordermatch::{broadcast_maker_orders_keep_alive_loop, clean_memory_loop, lp_ordermatch_loop,
+                                orders_kick_start, BalanceUpdateOrdermatchHandler};
 use crate::mm2::lp_swap::{running_swaps_num, swap_kick_starts};
 use crate::mm2::rpc::spawn_rpc;
 use crate::mm2::{MM_DATETIME, MM_VERSION};
@@ -344,6 +344,8 @@ pub async fn lp_init(ctx: MmArc) -> Result<(), String> {
 
     spawn(broadcast_maker_orders_keep_alive_loop(ctx.clone()));
 
+    spawn(clean_memory_loop(ctx.clone()));
+
     let ctx_id = try_s!(ctx.ffi_handle());
 
     spawn_rpc(ctx_id);