Skip to content

Replace custom mutex with shared primitives in IT #639

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Shourya742 merged 2 commits into from
Jul 24, 2026
Merged

Replace custom mutex with shared primitives in IT #639

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1a95a2e
replace custom mutex with shared primitive in message aggregator
Shourya742 Jul 21, 2026
9e90d2d
replace custom mutex with shared primitive in mining device
Shourya742 Jul 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 10 additions & 14 deletions integration-tests/lib/message_aggregator.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,15 +1,15 @@
use std::{collections::VecDeque, sync::Arc};
use std::collections::VecDeque;
use stratum_apps::{
custom_mutex::Mutex,
stratum_core::parsers_sv2::{AnyMessage, Tlv},
sync::SharedLock,
};

use crate::types::MsgType;

#[allow(clippy::type_complexity)]
#[derive(Debug, Clone)]
pub struct MessagesAggregator {
messages: Arc<Mutex<VecDeque<(MsgType, AnyMessage<'static>, Option<Vec<Tlv>>)>>>,
messages: SharedLock<VecDeque<(MsgType, AnyMessage<'static>, Option<Vec<Tlv>>)>>,
}

impl Default for MessagesAggregator {
Expand All @@ -22,7 +22,7 @@ impl MessagesAggregator {
/// Creates a new [`MessagesAggregator`].
pub fn new() -> Self {
Self {
messages: Arc::new(Mutex::new(VecDeque::new())),
messages: SharedLock::new(VecDeque::new()),
}
}

Expand All @@ -39,29 +39,25 @@ impl MessagesAggregator {
tlv_fields: Option<Vec<Tlv>>,
) {
self.messages
.safe_lock(|messages| messages.push_back((msg_type, message, tlv_fields)))
.with(|messages| messages.push_back((msg_type, message, tlv_fields)))
.unwrap();
}

/// Returns false if the queue is empty, true otherwise.
pub fn is_empty(&self) -> bool {
self.messages
.safe_lock(|messages| messages.is_empty())
.unwrap()
self.messages.with(|messages| messages.is_empty()).unwrap()
}

/// Clears all messages from the queue.
pub fn clear(&self) {
self.messages
.safe_lock(|messages| messages.clear())
.unwrap();
self.messages.with(|messages| messages.clear()).unwrap();
}

/// returns true if contains message_type
pub fn has_message_type(&self, message_type: u8) -> bool {
let has_message: bool = self
.messages
.safe_lock(|messages| {
.with(|messages| {
for (t, _, _) in messages.iter() {
if *t == message_type {
return true; // Exit early with `true`
Expand All @@ -77,7 +73,7 @@ impl MessagesAggregator {
/// until the first message of type message_type.
pub fn has_message_type_with_remove(&self, message_type: u8) -> bool {
self.messages
.safe_lock(|messages| {
.with(|messages| {
let mut cloned_messages = messages.clone();
for (pos, (t, _, _)) in cloned_messages.iter().enumerate() {
if *t == message_type {
Expand Down Expand Up @@ -109,7 +105,7 @@ impl MessagesAggregator {
) -> Option<(MsgType, AnyMessage<'static>, Option<Vec<Tlv>>)> {
let is_state = self
.messages
.safe_lock(|messages| {
.with(|messages| {
let mut cloned = messages.clone();
if let Some((msg_type, msg, tlv_fields)) = cloned.pop_front() {
*messages = cloned;
Expand Down
44 changes: 21 additions & 23 deletions integration-tests/lib/mining_device/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@ use std::{
};
pub use stratum_apps::key_utils::Secp256k1PublicKey;
use stratum_apps::{
custom_mutex::Mutex,
network_helpers::noise_connection::Connection,
stratum_core::{
bitcoin::{
Expand All @@ -33,6 +32,7 @@ use stratum_apps::{
noise_sv2::Initiator,
parsers_sv2::{CommonMessages, Mining, MiningDeviceMessages, ParserError},
},
sync::SharedLock,
};
use tokio::net::TcpStream;
use tracing::{debug, error, info};
Expand Down Expand Up @@ -202,7 +202,7 @@ impl SetupConnectionHandler {
}
}
pub async fn setup(
self_: Arc<Mutex<Self>>,
self_: SharedLock<Self>,
receiver: &mut Receiver<EitherFrame>,
sender: &mut Sender<EitherFrame>,
device_id: Option<String>,
Expand All @@ -224,13 +224,13 @@ impl SetupConnectionHandler {
}

fn handle_message_common(
self_: Arc<Mutex<Self>>,
self_: SharedLock<Self>,
message_type: u8,
payload: &mut [u8],
) -> Result<(), ParserError> {
let message: CommonMessages<'_> = (message_type, payload).try_into()?;
self_
.safe_lock(|handler| match message {
.with(|handler| match message {
CommonMessages::SetupConnectionSuccess(m) => {
handler.handle_setup_connection_success(m);
Ok(())
Expand Down Expand Up @@ -289,7 +289,7 @@ pub struct Device {
#[allow(dead_code)]
channel_opened: bool,
channel_id: Option<u32>,
miner: Arc<Mutex<Miner>>,
miner: SharedLock<Miner>,
jobs: Vec<NewMiningJob<'static>>,
prev_hash: Option<SetNewPrevHash<'static>>,
sequence_numbers: Id,
Expand Down Expand Up @@ -338,7 +338,7 @@ impl Device {
nominal_hashrate_multiplier: Option<f32>,
single_submit: bool,
) {
let setup_connection_handler = Arc::new(Mutex::new(SetupConnectionHandler::new()));
let setup_connection_handler = SharedLock::new(SetupConnectionHandler::new());
SetupConnectionHandler::setup(
setup_connection_handler,
&mut receiver,
Expand All @@ -348,7 +348,7 @@ impl Device {
)
.await;
info!("Pool sv2 connection established at {}", addr);
let miner = Arc::new(Mutex::new(Miner::new(handicap)));
let miner = SharedLock::new(Miner::new(handicap));
let (notify_changes_to_mining_thread, update_miners) = async_channel::unbounded();
let self_ = Self {
channel_opened: false,
Expand All @@ -369,7 +369,7 @@ impl Device {
));
let frame: StdFrame = open_channel.try_into().unwrap();
self_.sender.send(frame.into()).await.unwrap();
let self_mutex = std::sync::Arc::new(Mutex::new(self_));
let self_mutex = SharedLock::new(self_);
let cloned = self_mutex.clone();

let (share_send, share_recv) = async_channel::unbounded();
Expand All @@ -392,7 +392,7 @@ impl Device {
let payload = incoming.payload();
Device::handle_message_mining(self_mutex.clone(), message_type, payload).unwrap();
let mut notify_changes_to_mining_thread = self_mutex
.safe_lock(|s| s.notify_changes_to_mining_thread.clone())
.with(|s| s.notify_changes_to_mining_thread.clone())
.unwrap();
if notify_changes_to_mining_thread.should_send
&& (message_type == mining_sv2::MESSAGE_TYPE_NEW_MINING_JOB
Expand All @@ -410,34 +410,34 @@ impl Device {
}

async fn send_share(
self_mutex: Arc<Mutex<Self>>,
self_mutex: SharedLock<Self>,
nonce: u32,
job_id: u32,
version: u32,
ntime: u32,
) {
let share =
MiningDeviceMessages::Mining(Mining::SubmitSharesStandard(SubmitSharesStandard {
channel_id: self_mutex.safe_lock(|s| s.channel_id.unwrap()).unwrap(),
sequence_number: self_mutex.safe_lock(|s| s.sequence_numbers.next()).unwrap(),
channel_id: self_mutex.with(|s| s.channel_id.unwrap()).unwrap(),
sequence_number: self_mutex.with(|s| s.sequence_numbers.next()).unwrap(),
job_id,
nonce,
ntime,
version,
}));
let frame: StdFrame = share.try_into().unwrap();
let sender = self_mutex.safe_lock(|s| s.sender.clone()).unwrap();
let sender = self_mutex.with(|s| s.sender.clone()).unwrap();
sender.send(frame.into()).await.unwrap();
}

fn handle_message_mining(
self_: Arc<Mutex<Self>>,
self_: SharedLock<Self>,
message_type: u8,
payload: &mut [u8],
) -> Result<(), ParserError> {
let message: Mining<'_> = (message_type, payload).try_into()?;
self_
.safe_lock(|device| match message {
.with(|device| match message {
Mining::OpenStandardMiningChannelSuccess(m) => {
device.handle_open_standard_mining_channel_success(m);
Ok(())
Expand Down Expand Up @@ -492,7 +492,7 @@ impl Device {
m.group_channel_id, m.channel_id, req_id
);
self.miner
.safe_lock(|miner| miner.new_target(m.target.to_owned_bytes()))
.with(|miner| miner.new_target(m.target.to_owned_bytes()))
.unwrap();
self.notify_changes_to_mining_thread.should_send = true;
}
Expand Down Expand Up @@ -535,9 +535,7 @@ impl Device {
debug!("NewMiningJob: {}", m);
match (m.is_future(), self.prev_hash.as_ref()) {
(false, Some(p_h)) => {
self.miner
.safe_lock(|miner| miner.new_header(p_h, &m))
.unwrap();
self.miner.with(|miner| miner.new_header(p_h, &m)).unwrap();
self.jobs = vec![m.as_static()];
self.notify_changes_to_mining_thread.should_send = true;
}
Expand Down Expand Up @@ -565,7 +563,7 @@ impl Device {
}
1 => {
self.miner
.safe_lock(|miner| miner.new_header(&m, jobs[0]))
.with(|miner| miner.new_header(&m, jobs[0]))
.unwrap();
self.jobs = vec![jobs[0].clone()];
self.prev_hash = Some(m.as_static());
Expand All @@ -579,7 +577,7 @@ impl Device {
info!("Received SetTarget for channel id: {}", m.channel_id);
debug!("SetTarget: {}", m);
self.miner
.safe_lock(|miner| miner.new_target(m.maximum_target.to_owned_bytes()))
.with(|miner| miner.new_target(m.maximum_target.to_owned_bytes()))
.unwrap();
self.notify_changes_to_mining_thread.should_send = true;
}
Expand Down Expand Up @@ -939,7 +937,7 @@ fn generate_random_32_byte_array() -> [u8; 32] {

fn start_mining_threads(
have_new_job: Receiver<()>,
miner: Arc<Mutex<Miner>>,
miner: SharedLock<Miner>,
share_send: Sender<(u32, u32, u32, u32)>,
) {
tokio::task::spawn(async move {
Expand All @@ -952,7 +950,7 @@ fn start_mining_threads(
while let Some(killer) = killers.pop() {
killer.store(true, Ordering::Relaxed);
}
let miner = miner.safe_lock(|m| m.clone()).unwrap();
let miner = miner.with(|m| m.clone()).unwrap();
for i in 0..p {
let mut miner = miner.clone();
let share_send = share_send.clone();
Expand Down
Loading