Optimize LLMQs sending of sig shares#2704

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
codablock merged 4 commits into from
Feb 16, 2019
+85 −6
Merged

Optimize LLMQs sending of sig shares #2704

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
cf29320
Allow to disable optimistic send in PushMessage()
codablock Feb 15, 2019
acb8789
Implement WakeupSelect() to allow preliminary wakeup after message push
codablock Feb 14, 2019
c03480d
Disable optimistic sending when pushing sig share related messages
codablock Feb 15, 2019
d7bd095
Use pipe() together with fcntl instead of pipe2()
codablock Feb 15, 2019
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
15 changes: 12 additions & 3 deletions src/llmq/quorums_signing_shares.cpp
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -896,6 +896,8 @@ void CSigSharesManager::SendMessages()
CollectSigSharesToAnnounce(sigSharesToAnnounce);
}

bool didSend = false;

g_connman->ForEachNode([&](CNode* pnode) {
CNetMsgMaker msgMaker(pnode->GetSendVersion());

Expand All@@ -905,7 +907,8 @@ void CSigSharesManager::SendMessages()
assert(p.second.CountSet() != 0);
LogPrint("llmq", "CSigSharesManager::SendMessages -- QGETSIGSHARES inv={%s}, node=%d\n",
p.second.ToString(), pnode->id);
g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QGETSIGSHARES, p.second));
g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QGETSIGSHARES, p.second), false);
didSend = true;
}
}

Expand All@@ -915,7 +918,8 @@ void CSigSharesManager::SendMessages()
assert(!p.second.sigShares.empty());
LogPrint("llmq", "CSigSharesManager::SendMessages -- QBSIGSHARES inv={%s}, node=%d\n",
p.second.ToInv().ToString(), pnode->id);
g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QBSIGSHARES, p.second));
g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QBSIGSHARES, p.second), false);
didSend = true;
}
}

Expand All@@ -925,12 +929,17 @@ void CSigSharesManager::SendMessages()
assert(p.second.CountSet() != 0);
LogPrint("llmq", "CSigSharesManager::SendMessages -- QSIGSHARESINV inv={%s}, node=%d\n",
p.second.ToString(), pnode->id);
g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARESINV, p.second));
g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARESINV, p.second), false);
didSend = true;
}
}

return true;
});

if (didSend) {
g_connman->WakeSelect();
}
}

void CSigSharesManager::Cleanup()
Expand Down
67 changes: 65 additions & 2 deletions src/net.cpp
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -1210,6 +1210,18 @@ void CConnman::ThreadSocketHandler()
SOCKET hSocketMax = 0;
bool have_fds = false;

#ifndef WIN32
// We add a pipe to the read set so that the select() call can be woken up from the outside
// This is done when data is available for sending and at the same time optimistic sending was disabled
// when pushing the data.
// This is currently only implemented for POSIX compliant systems. This means that Windows will fall back to
// timing out after 50ms and then trying to send. This is ok as we assume that heavy-load daemons are usually
// run on Linux and friends.
FD_SET(wakeupPipe[0], &fdsetRecv);
hSocketMax = std::max(hSocketMax, (SOCKET)wakeupPipe[0]);
have_fds = true;
#endif

BOOST_FOREACH(const ListenSocket& hListenSocket, vhListenSocket) {
FD_SET(hListenSocket.socket, &fdsetRecv);
hSocketMax = std::max(hSocketMax, hListenSocket.socket);
Expand DownExpand Up@@ -1276,6 +1288,20 @@ void CConnman::ThreadSocketHandler()
return;
}

#ifndef WIN32
// drain the wakeup pipe
if (FD_ISSET(wakeupPipe[0], &fdsetRecv)) {
LogPrint("net", "woke up select()\n");
char buf[128];
while (true) {
int r = read(wakeupPipe[0], buf, sizeof(buf));
if (r <= 0) {
break;
}
}
}
#endif

//
// Accept new connections
//
Expand DownExpand Up@@ -1426,6 +1452,21 @@ void CConnman::WakeMessageHandler()
condMsgProc.notify_one();
}

void CConnman::WakeSelect()
{
#ifndef WIN32
if (wakeupPipe[1] == -1) {
return;
}

LogPrint("net", "waking up select()\n");

char buf[1];
if (write(wakeupPipe[1], buf, 1) != 1) {
LogPrint("net", "write to wakeupPipe failed\n");
}
#endif
}



Expand DownExpand Up@@ -2387,6 +2428,22 @@ bool CConnman::Start(CScheduler& scheduler, std::string& strNodeError, Options c
fMsgProcWake = false;
}

#ifndef WIN32
if (pipe(wakeupPipe) != 0) {
Comment thread
UdjinM6 marked this conversation as resolved.
wakeupPipe[0] = wakeupPipe[1] = -1;
LogPrint("net", "pipe() for wakeupPipe failed\n");
} else {
int fFlags = fcntl(wakeupPipe[0], F_GETFL, 0);
if (fcntl(wakeupPipe[0], F_SETFL, fFlags | O_NONBLOCK) == -1) {
LogPrint("net", "fcntl for O_NONBLOCK on wakeupPipe failed\n");
}
fFlags = fcntl(wakeupPipe[1], F_GETFL, 0);
if (fcntl(wakeupPipe[1], F_SETFL, fFlags | O_NONBLOCK) == -1) {
LogPrint("net", "fcntl for O_NONBLOCK on wakeupPipe failed\n");
}
}
#endif

// Send and receive from sockets, accept connections
threadSocketHandler = std::thread(&TraceThread<std::function<void()> >, "net", std::function<void()>(std::bind(&CConnman::ThreadSocketHandler, this)));

Expand DownExpand Up@@ -2512,6 +2569,12 @@ void CConnman::Stop()
semAddnode = NULL;
delete semMasternodeOutbound;
semMasternodeOutbound = NULL;

#ifndef WIN32
if (wakeupPipe[0] != -1) close(wakeupPipe[0]);
if (wakeupPipe[1] != -1) close(wakeupPipe[1]);
wakeupPipe[0] = wakeupPipe[1] = -1;
#endif
}

void CConnman::DeleteNode(CNode* pnode)
Expand DownExpand Up@@ -3054,7 +3117,7 @@ bool CConnman::NodeFullyConnected(const CNode* pnode)
return pnode && pnode->fSuccessfullyConnected && !pnode->fDisconnect;
}

void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg, bool allowOptimisticSend)
{
size_t nMessageSize = msg.data.size();
size_t nTotalSize = nMessageSize + CMessageHeader::HEADER_SIZE;
Expand All@@ -3071,7 +3134,7 @@ void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
size_t nBytesSent = 0;
{
LOCK(pnode->cs_vSend);
bool optimisticSend(pnode->vSendMsg.empty());
bool optimisticSend(allowOptimisticSend && pnode->vSendMsg.empty());

//log total amount of bytes per command
pnode->mapSendBytesPerMsgCmd[msg.command] += nTotalSize;
Expand Down
9 changes: 8 additions & 1 deletion src/net.h
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,7 @@ class CConnman

bool IsMasternodeOrDisconnectRequested(const CService& addr);

void PushMessage(CNode* pnode, CSerializedNetMsg&& msg);
void PushMessage(CNode* pnode, CSerializedNetMsg&& msg, bool allowOptimisticSend = true);

template<typename Condition, typename Callable>
bool ForEachNodeContinueIf(const Condition& cond, Callable&& func)
Expand DownExpand Up@@ -408,6 +408,8 @@ class CConnman
unsigned int GetReceiveFloodSize() const;

void WakeMessageHandler();
void WakeSelect();

private:
struct ListenSocket {
SOCKET socket;
Expand DownExpand Up@@ -525,6 +527,11 @@ class CConnman

CThreadInterrupt interruptNet;

#ifndef WIN32
/** a pipe which is added to select() calls to wakeup before the timeout */
int wakeupPipe[2]{-1,-1};
#endif

std::thread threadDNSAddressSeed;
std::thread threadSocketHandler;
std::thread threadOpenAddedConnections;
Expand Down
, 'i'); if (__m === '*' || __re.test(location.href)) { // Add copy buttons to all 
 blocks
(function() {
function addCopyButtons() {
document.querySelectorAll('pre code').forEach(function(codeBlock) {
if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;
codeBlock.parentElement.setAttribute('data-copy-added', 'true');
var btn = document.createElement('button');
btn.textContent = 'Copy';
btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';
btn.onmouseover = function() { this.style.opacity = '1'; };
btn.onmouseout = function() { this.style.opacity = '0.7'; };
btn.onclick = function() {
navigator.clipboard.writeText(codeBlock.textContent).then(function() {
btn.textContent = 'Copied!';
setTimeout(function() { btn.textContent = 'Copy'; }, 1500);
});
};
codeBlock.parentElement.style.position = 'relative';
codeBlock.parentElement.appendChild(btn);
});
}
addCopyButtons();
// Re-run on dynamic content
var observer = new MutationObserver(addCopyButtons);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '

Skip to content
Optimize LLMQs sending of sig shares#2704

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

codablock
merged 4 commits into

from

Feb 16, 2019

+85

−6

 Merged


Optimize LLMQs sending of sig shares
#2704
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
cf29320

Allow to disable optimistic send in PushMessage()


codablock Feb 15, 2019
acb8789

Implement WakeupSelect() to allow preliminary wakeup after message push


codablock Feb 14, 2019
c03480d

Disable optimistic sending when pushing sig share related messages


codablock Feb 15, 2019
d7bd095

Use pipe() together with fcntl instead of pipe2()


codablock Feb 15, 2019

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


15 changes: 12 additions & 3 deletions
src/llmq/quorums_signing_shares.cpp

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -896,6 +896,8 @@ void CSigSharesManager::SendMessages()
 CollectSigSharesToAnnounce(sigSharesToAnnounce);
 }

 bool didSend = false;

 g_connman->ForEachNode([&](CNode* pnode) {
 CNetMsgMaker msgMaker(pnode->GetSendVersion());

Expand All@@ -905,7 +907,8 @@ void CSigSharesManager::SendMessages()
 assert(p.second.CountSet() != 0);
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QGETSIGSHARES inv={%s}, node=%d\n",
 p.second.ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QGETSIGSHARES, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QGETSIGSHARES, p.second), false);
 didSend = true;
 }
 }

Expand All@@ -915,7 +918,8 @@ void CSigSharesManager::SendMessages()
 assert(!p.second.sigShares.empty());
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QBSIGSHARES inv={%s}, node=%d\n",
 p.second.ToInv().ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QBSIGSHARES, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QBSIGSHARES, p.second), false);
 didSend = true;
 }
 }

Expand All@@ -925,12 +929,17 @@ void CSigSharesManager::SendMessages()
 assert(p.second.CountSet() != 0);
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QSIGSHARESINV inv={%s}, node=%d\n",
 p.second.ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARESINV, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARESINV, p.second), false);
 didSend = true;
 }
 }

 return true;
 });

 if (didSend) {
 g_connman->WakeSelect();
 }
}

void CSigSharesManager::Cleanup()
Expand Down

67 changes: 65 additions & 2 deletions
src/net.cpp

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -1210,6 +1210,18 @@ void CConnman::ThreadSocketHandler()
 SOCKET hSocketMax = 0;
 bool have_fds = false;

#ifndef WIN32
 // We add a pipe to the read set so that the select() call can be woken up from the outside
 // This is done when data is available for sending and at the same time optimistic sending was disabled
 // when pushing the data.
 // This is currently only implemented for POSIX compliant systems. This means that Windows will fall back to
 // timing out after 50ms and then trying to send. This is ok as we assume that heavy-load daemons are usually
 // run on Linux and friends.
 FD_SET(wakeupPipe[0], &fdsetRecv);
 hSocketMax = std::max(hSocketMax, (SOCKET)wakeupPipe[0]);
 have_fds = true;
#endif

 BOOST_FOREACH(const ListenSocket& hListenSocket, vhListenSocket) {
 FD_SET(hListenSocket.socket, &fdsetRecv);
 hSocketMax = std::max(hSocketMax, hListenSocket.socket);
Expand DownExpand Up@@ -1276,6 +1288,20 @@ void CConnman::ThreadSocketHandler()
 return;
 }

#ifndef WIN32
 // drain the wakeup pipe
 if (FD_ISSET(wakeupPipe[0], &fdsetRecv)) {
 LogPrint("net", "woke up select()\n");
 char buf[128];
 while (true) {
 int r = read(wakeupPipe[0], buf, sizeof(buf));
 if (r <= 0) {
 break;
 }
 }
 }
#endif

 //
 // Accept new connections
 //
Expand DownExpand Up@@ -1426,6 +1452,21 @@ void CConnman::WakeMessageHandler()
 condMsgProc.notify_one();
}

void CConnman::WakeSelect()
{
#ifndef WIN32
 if (wakeupPipe[1] == -1) {
 return;
 }

 LogPrint("net", "waking up select()\n");

 char buf[1];
 if (write(wakeupPipe[1], buf, 1) != 1) {
 LogPrint("net", "write to wakeupPipe failed\n");
 }
#endif
}



Expand DownExpand Up@@ -2387,6 +2428,22 @@ bool CConnman::Start(CScheduler& scheduler, std::string& strNodeError, Options c
 fMsgProcWake = false;
 }

#ifndef WIN32
 if (pipe(wakeupPipe) != 0) {
Comment thread
UdjinM6 marked this conversation as resolved.

 wakeupPipe[0] = wakeupPipe[1] = -1;
 LogPrint("net", "pipe() for wakeupPipe failed\n");
 } else {
 int fFlags = fcntl(wakeupPipe[0], F_GETFL, 0);
 if (fcntl(wakeupPipe[0], F_SETFL, fFlags | O_NONBLOCK) == -1) {
 LogPrint("net", "fcntl for O_NONBLOCK on wakeupPipe failed\n");
 }
 fFlags = fcntl(wakeupPipe[1], F_GETFL, 0);
 if (fcntl(wakeupPipe[1], F_SETFL, fFlags | O_NONBLOCK) == -1) {
 LogPrint("net", "fcntl for O_NONBLOCK on wakeupPipe failed\n");
 }
 }
#endif

 // Send and receive from sockets, accept connections
 threadSocketHandler = std::thread(&TraceThread<std::function<void()> >, "net", std::function<void()>(std::bind(&CConnman::ThreadSocketHandler, this)));

Expand DownExpand Up@@ -2512,6 +2569,12 @@ void CConnman::Stop()
 semAddnode = NULL;
 delete semMasternodeOutbound;
 semMasternodeOutbound = NULL;

#ifndef WIN32
 if (wakeupPipe[0] != -1) close(wakeupPipe[0]);
 if (wakeupPipe[1] != -1) close(wakeupPipe[1]);
 wakeupPipe[0] = wakeupPipe[1] = -1;
#endif
}

void CConnman::DeleteNode(CNode* pnode)
Expand DownExpand Up@@ -3054,7 +3117,7 @@ bool CConnman::NodeFullyConnected(const CNode* pnode)
 return pnode && pnode->fSuccessfullyConnected && !pnode->fDisconnect;
}

void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg, bool allowOptimisticSend)
{
 size_t nMessageSize = msg.data.size();
 size_t nTotalSize = nMessageSize + CMessageHeader::HEADER_SIZE;
Expand All@@ -3071,7 +3134,7 @@ void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
 size_t nBytesSent = 0;
 {
 LOCK(pnode->cs_vSend);
 bool optimisticSend(pnode->vSendMsg.empty());
 bool optimisticSend(allowOptimisticSend && pnode->vSendMsg.empty());

 //log total amount of bytes per command
 pnode->mapSendBytesPerMsgCmd[msg.command] += nTotalSize;
Expand Down

9 changes: 8 additions & 1 deletion
src/net.h

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,7 @@ class CConnman

 bool IsMasternodeOrDisconnectRequested(const CService& addr);

 void PushMessage(CNode* pnode, CSerializedNetMsg&& msg);
 void PushMessage(CNode* pnode, CSerializedNetMsg&& msg, bool allowOptimisticSend = true);

 template<typename Condition, typename Callable>
 bool ForEachNodeContinueIf(const Condition& cond, Callable&& func)
Expand DownExpand Up@@ -408,6 +408,8 @@ class CConnman
 unsigned int GetReceiveFloodSize() const;

 void WakeMessageHandler();
 void WakeSelect();

private:
 struct ListenSocket {
 SOCKET socket;
Expand DownExpand Up@@ -525,6 +527,11 @@ class CConnman

 CThreadInterrupt interruptNet;

#ifndef WIN32
 /** a pipe which is added to select() calls to wakeup before the timeout */
 int wakeupPipe[2]{-1,-1};
#endif

 std::thread threadDNSAddressSeed;
 std::thread threadSocketHandler;
 std::thread threadOpenAddedConnections;
Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Force GitHub README to respect dark mode
(function() {
var style = document.createElement('style');
style.textContent = '
.markdown-body {
color-scheme: dark light;
}
.markdown-body pre { background: #161b22 !important; }
.markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }
.markdown-body table th, .markdown-body table td { border-color: #30363d !important; }
.markdown-body img { background: #0d1117; }
.markdown-body blockquote { border-left-color: #8b949e; }
.markdown-body hr { border-color: #30363d; }
';
document.head.appendChild(style);
})();
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Optimize LLMQs sending of sig shares#2704

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

codablock
merged 4 commits into

from

Feb 16, 2019

+85

−6

 Merged


Optimize LLMQs sending of sig shares
#2704
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
cf29320

Allow to disable optimistic send in PushMessage()


codablock Feb 15, 2019
acb8789

Implement WakeupSelect() to allow preliminary wakeup after message push


codablock Feb 14, 2019
c03480d

Disable optimistic sending when pushing sig share related messages


codablock Feb 15, 2019
d7bd095

Use pipe() together with fcntl instead of pipe2()


codablock Feb 15, 2019

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


15 changes: 12 additions & 3 deletions
src/llmq/quorums_signing_shares.cpp

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -896,6 +896,8 @@ void CSigSharesManager::SendMessages()
 CollectSigSharesToAnnounce(sigSharesToAnnounce);
 }

 bool didSend = false;

 g_connman->ForEachNode([&](CNode* pnode) {
 CNetMsgMaker msgMaker(pnode->GetSendVersion());

Expand All@@ -905,7 +907,8 @@ void CSigSharesManager::SendMessages()
 assert(p.second.CountSet() != 0);
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QGETSIGSHARES inv={%s}, node=%d\n",
 p.second.ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QGETSIGSHARES, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QGETSIGSHARES, p.second), false);
 didSend = true;
 }
 }

Expand All@@ -915,7 +918,8 @@ void CSigSharesManager::SendMessages()
 assert(!p.second.sigShares.empty());
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QBSIGSHARES inv={%s}, node=%d\n",
 p.second.ToInv().ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QBSIGSHARES, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QBSIGSHARES, p.second), false);
 didSend = true;
 }
 }

Expand All@@ -925,12 +929,17 @@ void CSigSharesManager::SendMessages()
 assert(p.second.CountSet() != 0);
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QSIGSHARESINV inv={%s}, node=%d\n",
 p.second.ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARESINV, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARESINV, p.second), false);
 didSend = true;
 }
 }

 return true;
 });

 if (didSend) {
 g_connman->WakeSelect();
 }
}

void CSigSharesManager::Cleanup()
Expand Down

67 changes: 65 additions & 2 deletions
src/net.cpp

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -1210,6 +1210,18 @@ void CConnman::ThreadSocketHandler()
 SOCKET hSocketMax = 0;
 bool have_fds = false;

#ifndef WIN32
 // We add a pipe to the read set so that the select() call can be woken up from the outside
 // This is done when data is available for sending and at the same time optimistic sending was disabled
 // when pushing the data.
 // This is currently only implemented for POSIX compliant systems. This means that Windows will fall back to
 // timing out after 50ms and then trying to send. This is ok as we assume that heavy-load daemons are usually
 // run on Linux and friends.
 FD_SET(wakeupPipe[0], &fdsetRecv);
 hSocketMax = std::max(hSocketMax, (SOCKET)wakeupPipe[0]);
 have_fds = true;
#endif

 BOOST_FOREACH(const ListenSocket& hListenSocket, vhListenSocket) {
 FD_SET(hListenSocket.socket, &fdsetRecv);
 hSocketMax = std::max(hSocketMax, hListenSocket.socket);
Expand DownExpand Up@@ -1276,6 +1288,20 @@ void CConnman::ThreadSocketHandler()
 return;
 }

#ifndef WIN32
 // drain the wakeup pipe
 if (FD_ISSET(wakeupPipe[0], &fdsetRecv)) {
 LogPrint("net", "woke up select()\n");
 char buf[128];
 while (true) {
 int r = read(wakeupPipe[0], buf, sizeof(buf));
 if (r <= 0) {
 break;
 }
 }
 }
#endif

 //
 // Accept new connections
 //
Expand DownExpand Up@@ -1426,6 +1452,21 @@ void CConnman::WakeMessageHandler()
 condMsgProc.notify_one();
}

void CConnman::WakeSelect()
{
#ifndef WIN32
 if (wakeupPipe[1] == -1) {
 return;
 }

 LogPrint("net", "waking up select()\n");

 char buf[1];
 if (write(wakeupPipe[1], buf, 1) != 1) {
 LogPrint("net", "write to wakeupPipe failed\n");
 }
#endif
}



Expand DownExpand Up@@ -2387,6 +2428,22 @@ bool CConnman::Start(CScheduler& scheduler, std::string& strNodeError, Options c
 fMsgProcWake = false;
 }

#ifndef WIN32
 if (pipe(wakeupPipe) != 0) {
Comment thread
UdjinM6 marked this conversation as resolved.

 wakeupPipe[0] = wakeupPipe[1] = -1;
 LogPrint("net", "pipe() for wakeupPipe failed\n");
 } else {
 int fFlags = fcntl(wakeupPipe[0], F_GETFL, 0);
 if (fcntl(wakeupPipe[0], F_SETFL, fFlags | O_NONBLOCK) == -1) {
 LogPrint("net", "fcntl for O_NONBLOCK on wakeupPipe failed\n");
 }
 fFlags = fcntl(wakeupPipe[1], F_GETFL, 0);
 if (fcntl(wakeupPipe[1], F_SETFL, fFlags | O_NONBLOCK) == -1) {
 LogPrint("net", "fcntl for O_NONBLOCK on wakeupPipe failed\n");
 }
 }
#endif

 // Send and receive from sockets, accept connections
 threadSocketHandler = std::thread(&TraceThread<std::function<void()> >, "net", std::function<void()>(std::bind(&CConnman::ThreadSocketHandler, this)));

Expand DownExpand Up@@ -2512,6 +2569,12 @@ void CConnman::Stop()
 semAddnode = NULL;
 delete semMasternodeOutbound;
 semMasternodeOutbound = NULL;

#ifndef WIN32
 if (wakeupPipe[0] != -1) close(wakeupPipe[0]);
 if (wakeupPipe[1] != -1) close(wakeupPipe[1]);
 wakeupPipe[0] = wakeupPipe[1] = -1;
#endif
}

void CConnman::DeleteNode(CNode* pnode)
Expand DownExpand Up@@ -3054,7 +3117,7 @@ bool CConnman::NodeFullyConnected(const CNode* pnode)
 return pnode && pnode->fSuccessfullyConnected && !pnode->fDisconnect;
}

void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg, bool allowOptimisticSend)
{
 size_t nMessageSize = msg.data.size();
 size_t nTotalSize = nMessageSize + CMessageHeader::HEADER_SIZE;
Expand All@@ -3071,7 +3134,7 @@ void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
 size_t nBytesSent = 0;
 {
 LOCK(pnode->cs_vSend);
 bool optimisticSend(pnode->vSendMsg.empty());
 bool optimisticSend(allowOptimisticSend && pnode->vSendMsg.empty());

 //log total amount of bytes per command
 pnode->mapSendBytesPerMsgCmd[msg.command] += nTotalSize;
Expand Down

9 changes: 8 additions & 1 deletion
src/net.h

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,7 @@ class CConnman

 bool IsMasternodeOrDisconnectRequested(const CService& addr);

 void PushMessage(CNode* pnode, CSerializedNetMsg&& msg);
 void PushMessage(CNode* pnode, CSerializedNetMsg&& msg, bool allowOptimisticSend = true);

 template<typename Condition, typename Callable>
 bool ForEachNodeContinueIf(const Condition& cond, Callable&& func)
Expand DownExpand Up@@ -408,6 +408,8 @@ class CConnman
 unsigned int GetReceiveFloodSize() const;

 void WakeMessageHandler();
 void WakeSelect();

private:
 struct ListenSocket {
 SOCKET socket;
Expand DownExpand Up@@ -525,6 +527,11 @@ class CConnman

 CThreadInterrupt interruptNet;

#ifndef WIN32
 /** a pipe which is added to select() calls to wakeup before the timeout */
 int wakeupPipe[2]{-1,-1};
#endif

 std::thread threadDNSAddressSeed;
 std::thread threadSocketHandler;
 std::thread threadOpenAddedConnections;
Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Highlight search terms from Google/DuckDuckGo/Bing referrer
(function() {
var ref = document.referrer;
var terms = [];
if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {
var url = new URL(ref);
var q = url.searchParams.get('q') || url.searchParams.get('p');
if (q) {
terms = q.split(/\s+/).filter(function(t) { return t.length > 2; });
}
}
if (terms.length === 0) return;
var style = document.createElement('style');
style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';
document.head.appendChild(style);
function highlight(node) {
if (node.nodeType === 3) { // text node
var text = node.textContent;
var found = false;
terms.forEach(function(term) {
var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\]\\]/g, '\\') + ')', 'gi');
if (regex.test(text)) {
found = true;
var frag = document.createDocumentFragment();
var parts = text.split(regex);
parts.forEach(function(part, i) {
if (i % 2 === 0) {
frag.appendChild(document.createTextNode(part));
} else {
var span = document.createElement('span');
span.className = 'userscript-highlight';
span.textContent = part;
frag.appendChild(span);
}
});
node.parentNode.replaceChild(frag, node);
}
});
} else if (node.nodeType === 1 && node.childNodes) { // element
var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];
if (!skipTags.includes(node.tagName)) {
Array.from(node.childNodes).forEach(highlight);
}
}
}
highlight(document.body);
// Re-highlight on dynamic content
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1 || node.nodeType === 3) highlight(node);
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Optimize LLMQs sending of sig shares#2704

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

codablock
merged 4 commits into

from

Feb 16, 2019

+85

−6

 Merged


Optimize LLMQs sending of sig shares
#2704
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
cf29320

Allow to disable optimistic send in PushMessage()


codablock Feb 15, 2019
acb8789

Implement WakeupSelect() to allow preliminary wakeup after message push


codablock Feb 14, 2019
c03480d

Disable optimistic sending when pushing sig share related messages


codablock Feb 15, 2019
d7bd095

Use pipe() together with fcntl instead of pipe2()


codablock Feb 15, 2019

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


15 changes: 12 additions & 3 deletions
src/llmq/quorums_signing_shares.cpp

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -896,6 +896,8 @@ void CSigSharesManager::SendMessages()
 CollectSigSharesToAnnounce(sigSharesToAnnounce);
 }

 bool didSend = false;

 g_connman->ForEachNode([&](CNode* pnode) {
 CNetMsgMaker msgMaker(pnode->GetSendVersion());

Expand All@@ -905,7 +907,8 @@ void CSigSharesManager::SendMessages()
 assert(p.second.CountSet() != 0);
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QGETSIGSHARES inv={%s}, node=%d\n",
 p.second.ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QGETSIGSHARES, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QGETSIGSHARES, p.second), false);
 didSend = true;
 }
 }

Expand All@@ -915,7 +918,8 @@ void CSigSharesManager::SendMessages()
 assert(!p.second.sigShares.empty());
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QBSIGSHARES inv={%s}, node=%d\n",
 p.second.ToInv().ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QBSIGSHARES, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QBSIGSHARES, p.second), false);
 didSend = true;
 }
 }

Expand All@@ -925,12 +929,17 @@ void CSigSharesManager::SendMessages()
 assert(p.second.CountSet() != 0);
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QSIGSHARESINV inv={%s}, node=%d\n",
 p.second.ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARESINV, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARESINV, p.second), false);
 didSend = true;
 }
 }

 return true;
 });

 if (didSend) {
 g_connman->WakeSelect();
 }
}

void CSigSharesManager::Cleanup()
Expand Down

67 changes: 65 additions & 2 deletions
src/net.cpp

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -1210,6 +1210,18 @@ void CConnman::ThreadSocketHandler()
 SOCKET hSocketMax = 0;
 bool have_fds = false;

#ifndef WIN32
 // We add a pipe to the read set so that the select() call can be woken up from the outside
 // This is done when data is available for sending and at the same time optimistic sending was disabled
 // when pushing the data.
 // This is currently only implemented for POSIX compliant systems. This means that Windows will fall back to
 // timing out after 50ms and then trying to send. This is ok as we assume that heavy-load daemons are usually
 // run on Linux and friends.
 FD_SET(wakeupPipe[0], &fdsetRecv);
 hSocketMax = std::max(hSocketMax, (SOCKET)wakeupPipe[0]);
 have_fds = true;
#endif

 BOOST_FOREACH(const ListenSocket& hListenSocket, vhListenSocket) {
 FD_SET(hListenSocket.socket, &fdsetRecv);
 hSocketMax = std::max(hSocketMax, hListenSocket.socket);
Expand DownExpand Up@@ -1276,6 +1288,20 @@ void CConnman::ThreadSocketHandler()
 return;
 }

#ifndef WIN32
 // drain the wakeup pipe
 if (FD_ISSET(wakeupPipe[0], &fdsetRecv)) {
 LogPrint("net", "woke up select()\n");
 char buf[128];
 while (true) {
 int r = read(wakeupPipe[0], buf, sizeof(buf));
 if (r <= 0) {
 break;
 }
 }
 }
#endif

 //
 // Accept new connections
 //
Expand DownExpand Up@@ -1426,6 +1452,21 @@ void CConnman::WakeMessageHandler()
 condMsgProc.notify_one();
}

void CConnman::WakeSelect()
{
#ifndef WIN32
 if (wakeupPipe[1] == -1) {
 return;
 }

 LogPrint("net", "waking up select()\n");

 char buf[1];
 if (write(wakeupPipe[1], buf, 1) != 1) {
 LogPrint("net", "write to wakeupPipe failed\n");
 }
#endif
}



Expand DownExpand Up@@ -2387,6 +2428,22 @@ bool CConnman::Start(CScheduler& scheduler, std::string& strNodeError, Options c
 fMsgProcWake = false;
 }

#ifndef WIN32
 if (pipe(wakeupPipe) != 0) {
Comment thread
UdjinM6 marked this conversation as resolved.

 wakeupPipe[0] = wakeupPipe[1] = -1;
 LogPrint("net", "pipe() for wakeupPipe failed\n");
 } else {
 int fFlags = fcntl(wakeupPipe[0], F_GETFL, 0);
 if (fcntl(wakeupPipe[0], F_SETFL, fFlags | O_NONBLOCK) == -1) {
 LogPrint("net", "fcntl for O_NONBLOCK on wakeupPipe failed\n");
 }
 fFlags = fcntl(wakeupPipe[1], F_GETFL, 0);
 if (fcntl(wakeupPipe[1], F_SETFL, fFlags | O_NONBLOCK) == -1) {
 LogPrint("net", "fcntl for O_NONBLOCK on wakeupPipe failed\n");
 }
 }
#endif

 // Send and receive from sockets, accept connections
 threadSocketHandler = std::thread(&TraceThread<std::function<void()> >, "net", std::function<void()>(std::bind(&CConnman::ThreadSocketHandler, this)));

Expand DownExpand Up@@ -2512,6 +2569,12 @@ void CConnman::Stop()
 semAddnode = NULL;
 delete semMasternodeOutbound;
 semMasternodeOutbound = NULL;

#ifndef WIN32
 if (wakeupPipe[0] != -1) close(wakeupPipe[0]);
 if (wakeupPipe[1] != -1) close(wakeupPipe[1]);
 wakeupPipe[0] = wakeupPipe[1] = -1;
#endif
}

void CConnman::DeleteNode(CNode* pnode)
Expand DownExpand Up@@ -3054,7 +3117,7 @@ bool CConnman::NodeFullyConnected(const CNode* pnode)
 return pnode && pnode->fSuccessfullyConnected && !pnode->fDisconnect;
}

void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg, bool allowOptimisticSend)
{
 size_t nMessageSize = msg.data.size();
 size_t nTotalSize = nMessageSize + CMessageHeader::HEADER_SIZE;
Expand All@@ -3071,7 +3134,7 @@ void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
 size_t nBytesSent = 0;
 {
 LOCK(pnode->cs_vSend);
 bool optimisticSend(pnode->vSendMsg.empty());
 bool optimisticSend(allowOptimisticSend && pnode->vSendMsg.empty());

 //log total amount of bytes per command
 pnode->mapSendBytesPerMsgCmd[msg.command] += nTotalSize;
Expand Down

9 changes: 8 additions & 1 deletion
src/net.h

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,7 @@ class CConnman

 bool IsMasternodeOrDisconnectRequested(const CService& addr);

 void PushMessage(CNode* pnode, CSerializedNetMsg&& msg);
 void PushMessage(CNode* pnode, CSerializedNetMsg&& msg, bool allowOptimisticSend = true);

 template<typename Condition, typename Callable>
 bool ForEachNodeContinueIf(const Condition& cond, Callable&& func)
Expand DownExpand Up@@ -408,6 +408,8 @@ class CConnman
 unsigned int GetReceiveFloodSize() const;

 void WakeMessageHandler();
 void WakeSelect();

private:
 struct ListenSocket {
 SOCKET socket;
Expand DownExpand Up@@ -525,6 +527,11 @@ class CConnman

 CThreadInterrupt interruptNet;

#ifndef WIN32
 /** a pipe which is added to select() calls to wakeup before the timeout */
 int wakeupPipe[2]{-1,-1};
#endif

 std::thread threadDNSAddressSeed;
 std::thread threadSocketHandler;
 std::thread threadOpenAddedConnections;
Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Strip utm_, fbclid, gclid, etc. from all links on page
(function() {
var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',
'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',
'ref', 'ref_src', 'source', 'medium', 'campaign'];
function cleanUrl(url) {
try {
var u = new URL(url, window.location.origin);
var changed = false;
trackingParams.forEach(function(p) {
if (u.searchParams.has(p)) {
u.searchParams.delete(p);
changed = true;
}
});
return changed ? u.toString() : url;
} catch (e) {
return url;
}
}
function cleanLinks() {
document.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
cleanLinks();
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1) {
if (node.tagName === 'A') cleanLinks();
node.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '

Skip to content
Optimize LLMQs sending of sig shares#2704

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

codablock
merged 4 commits into

from

Feb 16, 2019

+85

−6

 Merged


Optimize LLMQs sending of sig shares
#2704
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
cf29320

Allow to disable optimistic send in PushMessage()


codablock Feb 15, 2019
acb8789

Implement WakeupSelect() to allow preliminary wakeup after message push


codablock Feb 14, 2019
c03480d

Disable optimistic sending when pushing sig share related messages


codablock Feb 15, 2019
d7bd095

Use pipe() together with fcntl instead of pipe2()


codablock Feb 15, 2019

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


15 changes: 12 additions & 3 deletions
src/llmq/quorums_signing_shares.cpp

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -896,6 +896,8 @@ void CSigSharesManager::SendMessages()
 CollectSigSharesToAnnounce(sigSharesToAnnounce);
 }

 bool didSend = false;

 g_connman->ForEachNode([&](CNode* pnode) {
 CNetMsgMaker msgMaker(pnode->GetSendVersion());

Expand All@@ -905,7 +907,8 @@ void CSigSharesManager::SendMessages()
 assert(p.second.CountSet() != 0);
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QGETSIGSHARES inv={%s}, node=%d\n",
 p.second.ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QGETSIGSHARES, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QGETSIGSHARES, p.second), false);
 didSend = true;
 }
 }

Expand All@@ -915,7 +918,8 @@ void CSigSharesManager::SendMessages()
 assert(!p.second.sigShares.empty());
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QBSIGSHARES inv={%s}, node=%d\n",
 p.second.ToInv().ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QBSIGSHARES, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QBSIGSHARES, p.second), false);
 didSend = true;
 }
 }

Expand All@@ -925,12 +929,17 @@ void CSigSharesManager::SendMessages()
 assert(p.second.CountSet() != 0);
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QSIGSHARESINV inv={%s}, node=%d\n",
 p.second.ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARESINV, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARESINV, p.second), false);
 didSend = true;
 }
 }

 return true;
 });

 if (didSend) {
 g_connman->WakeSelect();
 }
}

void CSigSharesManager::Cleanup()
Expand Down

67 changes: 65 additions & 2 deletions
src/net.cpp

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -1210,6 +1210,18 @@ void CConnman::ThreadSocketHandler()
 SOCKET hSocketMax = 0;
 bool have_fds = false;

#ifndef WIN32
 // We add a pipe to the read set so that the select() call can be woken up from the outside
 // This is done when data is available for sending and at the same time optimistic sending was disabled
 // when pushing the data.
 // This is currently only implemented for POSIX compliant systems. This means that Windows will fall back to
 // timing out after 50ms and then trying to send. This is ok as we assume that heavy-load daemons are usually
 // run on Linux and friends.
 FD_SET(wakeupPipe[0], &fdsetRecv);
 hSocketMax = std::max(hSocketMax, (SOCKET)wakeupPipe[0]);
 have_fds = true;
#endif

 BOOST_FOREACH(const ListenSocket& hListenSocket, vhListenSocket) {
 FD_SET(hListenSocket.socket, &fdsetRecv);
 hSocketMax = std::max(hSocketMax, hListenSocket.socket);
Expand DownExpand Up@@ -1276,6 +1288,20 @@ void CConnman::ThreadSocketHandler()
 return;
 }

#ifndef WIN32
 // drain the wakeup pipe
 if (FD_ISSET(wakeupPipe[0], &fdsetRecv)) {
 LogPrint("net", "woke up select()\n");
 char buf[128];
 while (true) {
 int r = read(wakeupPipe[0], buf, sizeof(buf));
 if (r <= 0) {
 break;
 }
 }
 }
#endif

 //
 // Accept new connections
 //
Expand DownExpand Up@@ -1426,6 +1452,21 @@ void CConnman::WakeMessageHandler()
 condMsgProc.notify_one();
}

void CConnman::WakeSelect()
{
#ifndef WIN32
 if (wakeupPipe[1] == -1) {
 return;
 }

 LogPrint("net", "waking up select()\n");

 char buf[1];
 if (write(wakeupPipe[1], buf, 1) != 1) {
 LogPrint("net", "write to wakeupPipe failed\n");
 }
#endif
}



Expand DownExpand Up@@ -2387,6 +2428,22 @@ bool CConnman::Start(CScheduler& scheduler, std::string& strNodeError, Options c
 fMsgProcWake = false;
 }

#ifndef WIN32
 if (pipe(wakeupPipe) != 0) {
Comment thread
UdjinM6 marked this conversation as resolved.

 wakeupPipe[0] = wakeupPipe[1] = -1;
 LogPrint("net", "pipe() for wakeupPipe failed\n");
 } else {
 int fFlags = fcntl(wakeupPipe[0], F_GETFL, 0);
 if (fcntl(wakeupPipe[0], F_SETFL, fFlags | O_NONBLOCK) == -1) {
 LogPrint("net", "fcntl for O_NONBLOCK on wakeupPipe failed\n");
 }
 fFlags = fcntl(wakeupPipe[1], F_GETFL, 0);
 if (fcntl(wakeupPipe[1], F_SETFL, fFlags | O_NONBLOCK) == -1) {
 LogPrint("net", "fcntl for O_NONBLOCK on wakeupPipe failed\n");
 }
 }
#endif

 // Send and receive from sockets, accept connections
 threadSocketHandler = std::thread(&TraceThread<std::function<void()> >, "net", std::function<void()>(std::bind(&CConnman::ThreadSocketHandler, this)));

Expand DownExpand Up@@ -2512,6 +2569,12 @@ void CConnman::Stop()
 semAddnode = NULL;
 delete semMasternodeOutbound;
 semMasternodeOutbound = NULL;

#ifndef WIN32
 if (wakeupPipe[0] != -1) close(wakeupPipe[0]);
 if (wakeupPipe[1] != -1) close(wakeupPipe[1]);
 wakeupPipe[0] = wakeupPipe[1] = -1;
#endif
}

void CConnman::DeleteNode(CNode* pnode)
Expand DownExpand Up@@ -3054,7 +3117,7 @@ bool CConnman::NodeFullyConnected(const CNode* pnode)
 return pnode && pnode->fSuccessfullyConnected && !pnode->fDisconnect;
}

void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg, bool allowOptimisticSend)
{
 size_t nMessageSize = msg.data.size();
 size_t nTotalSize = nMessageSize + CMessageHeader::HEADER_SIZE;
Expand All@@ -3071,7 +3134,7 @@ void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
 size_t nBytesSent = 0;
 {
 LOCK(pnode->cs_vSend);
 bool optimisticSend(pnode->vSendMsg.empty());
 bool optimisticSend(allowOptimisticSend && pnode->vSendMsg.empty());

 //log total amount of bytes per command
 pnode->mapSendBytesPerMsgCmd[msg.command] += nTotalSize;
Expand Down

9 changes: 8 additions & 1 deletion
src/net.h

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,7 @@ class CConnman

 bool IsMasternodeOrDisconnectRequested(const CService& addr);

 void PushMessage(CNode* pnode, CSerializedNetMsg&& msg);
 void PushMessage(CNode* pnode, CSerializedNetMsg&& msg, bool allowOptimisticSend = true);

 template<typename Condition, typename Callable>
 bool ForEachNodeContinueIf(const Condition& cond, Callable&& func)
Expand DownExpand Up@@ -408,6 +408,8 @@ class CConnman
 unsigned int GetReceiveFloodSize() const;

 void WakeMessageHandler();
 void WakeSelect();

private:
 struct ListenSocket {
 SOCKET socket;
Expand DownExpand Up@@ -525,6 +527,11 @@ class CConnman

 CThreadInterrupt interruptNet;

#ifndef WIN32
 /** a pipe which is added to select() calls to wakeup before the timeout */
 int wakeupPipe[2]{-1,-1};
#endif

 std::thread threadDNSAddressSeed;
 std::thread threadSocketHandler;
 std::thread threadOpenAddedConnections;
Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Auto-enable theater mode on YouTube
(function() {
function tryTheater() {
var btn = document.querySelector('button[aria-label="Theater mode"], ytd-player #player button[title="Theater mode"]');
if (btn && !btn.classList.contains('activated')) {
btn.click();
}
}
// Try immediately
tryTheater();
// Try after navigation (SPA)
var lastUrl = location.href;
setInterval(function() {
if (location.href !== lastUrl) {
lastUrl = location.href;
setTimeout(tryTheater, 500);
}
}, 1000);
// Also try on player load
var observer = new MutationObserver(tryTheater);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Optimize LLMQs sending of sig shares#2704

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

codablock
merged 4 commits into

from

Feb 16, 2019

+85

−6

 Merged


Optimize LLMQs sending of sig shares
#2704
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
cf29320

Allow to disable optimistic send in PushMessage()


codablock Feb 15, 2019
acb8789

Implement WakeupSelect() to allow preliminary wakeup after message push


codablock Feb 14, 2019
c03480d

Disable optimistic sending when pushing sig share related messages


codablock Feb 15, 2019
d7bd095

Use pipe() together with fcntl instead of pipe2()


codablock Feb 15, 2019

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


15 changes: 12 additions & 3 deletions
src/llmq/quorums_signing_shares.cpp

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -896,6 +896,8 @@ void CSigSharesManager::SendMessages()
 CollectSigSharesToAnnounce(sigSharesToAnnounce);
 }

 bool didSend = false;

 g_connman->ForEachNode([&](CNode* pnode) {
 CNetMsgMaker msgMaker(pnode->GetSendVersion());

Expand All@@ -905,7 +907,8 @@ void CSigSharesManager::SendMessages()
 assert(p.second.CountSet() != 0);
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QGETSIGSHARES inv={%s}, node=%d\n",
 p.second.ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QGETSIGSHARES, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QGETSIGSHARES, p.second), false);
 didSend = true;
 }
 }

Expand All@@ -915,7 +918,8 @@ void CSigSharesManager::SendMessages()
 assert(!p.second.sigShares.empty());
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QBSIGSHARES inv={%s}, node=%d\n",
 p.second.ToInv().ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QBSIGSHARES, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QBSIGSHARES, p.second), false);
 didSend = true;
 }
 }

Expand All@@ -925,12 +929,17 @@ void CSigSharesManager::SendMessages()
 assert(p.second.CountSet() != 0);
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QSIGSHARESINV inv={%s}, node=%d\n",
 p.second.ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARESINV, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARESINV, p.second), false);
 didSend = true;
 }
 }

 return true;
 });

 if (didSend) {
 g_connman->WakeSelect();
 }
}

void CSigSharesManager::Cleanup()
Expand Down

67 changes: 65 additions & 2 deletions
src/net.cpp

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -1210,6 +1210,18 @@ void CConnman::ThreadSocketHandler()
 SOCKET hSocketMax = 0;
 bool have_fds = false;

#ifndef WIN32
 // We add a pipe to the read set so that the select() call can be woken up from the outside
 // This is done when data is available for sending and at the same time optimistic sending was disabled
 // when pushing the data.
 // This is currently only implemented for POSIX compliant systems. This means that Windows will fall back to
 // timing out after 50ms and then trying to send. This is ok as we assume that heavy-load daemons are usually
 // run on Linux and friends.
 FD_SET(wakeupPipe[0], &fdsetRecv);
 hSocketMax = std::max(hSocketMax, (SOCKET)wakeupPipe[0]);
 have_fds = true;
#endif

 BOOST_FOREACH(const ListenSocket& hListenSocket, vhListenSocket) {
 FD_SET(hListenSocket.socket, &fdsetRecv);
 hSocketMax = std::max(hSocketMax, hListenSocket.socket);
Expand DownExpand Up@@ -1276,6 +1288,20 @@ void CConnman::ThreadSocketHandler()
 return;
 }

#ifndef WIN32
 // drain the wakeup pipe
 if (FD_ISSET(wakeupPipe[0], &fdsetRecv)) {
 LogPrint("net", "woke up select()\n");
 char buf[128];
 while (true) {
 int r = read(wakeupPipe[0], buf, sizeof(buf));
 if (r <= 0) {
 break;
 }
 }
 }
#endif

 //
 // Accept new connections
 //
Expand DownExpand Up@@ -1426,6 +1452,21 @@ void CConnman::WakeMessageHandler()
 condMsgProc.notify_one();
}

void CConnman::WakeSelect()
{
#ifndef WIN32
 if (wakeupPipe[1] == -1) {
 return;
 }

 LogPrint("net", "waking up select()\n");

 char buf[1];
 if (write(wakeupPipe[1], buf, 1) != 1) {
 LogPrint("net", "write to wakeupPipe failed\n");
 }
#endif
}



Expand DownExpand Up@@ -2387,6 +2428,22 @@ bool CConnman::Start(CScheduler& scheduler, std::string& strNodeError, Options c
 fMsgProcWake = false;
 }

#ifndef WIN32
 if (pipe(wakeupPipe) != 0) {
Comment thread
UdjinM6 marked this conversation as resolved.

 wakeupPipe[0] = wakeupPipe[1] = -1;
 LogPrint("net", "pipe() for wakeupPipe failed\n");
 } else {
 int fFlags = fcntl(wakeupPipe[0], F_GETFL, 0);
 if (fcntl(wakeupPipe[0], F_SETFL, fFlags | O_NONBLOCK) == -1) {
 LogPrint("net", "fcntl for O_NONBLOCK on wakeupPipe failed\n");
 }
 fFlags = fcntl(wakeupPipe[1], F_GETFL, 0);
 if (fcntl(wakeupPipe[1], F_SETFL, fFlags | O_NONBLOCK) == -1) {
 LogPrint("net", "fcntl for O_NONBLOCK on wakeupPipe failed\n");
 }
 }
#endif

 // Send and receive from sockets, accept connections
 threadSocketHandler = std::thread(&TraceThread<std::function<void()> >, "net", std::function<void()>(std::bind(&CConnman::ThreadSocketHandler, this)));

Expand DownExpand Up@@ -2512,6 +2569,12 @@ void CConnman::Stop()
 semAddnode = NULL;
 delete semMasternodeOutbound;
 semMasternodeOutbound = NULL;

#ifndef WIN32
 if (wakeupPipe[0] != -1) close(wakeupPipe[0]);
 if (wakeupPipe[1] != -1) close(wakeupPipe[1]);
 wakeupPipe[0] = wakeupPipe[1] = -1;
#endif
}

void CConnman::DeleteNode(CNode* pnode)
Expand DownExpand Up@@ -3054,7 +3117,7 @@ bool CConnman::NodeFullyConnected(const CNode* pnode)
 return pnode && pnode->fSuccessfullyConnected && !pnode->fDisconnect;
}

void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg, bool allowOptimisticSend)
{
 size_t nMessageSize = msg.data.size();
 size_t nTotalSize = nMessageSize + CMessageHeader::HEADER_SIZE;
Expand All@@ -3071,7 +3134,7 @@ void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
 size_t nBytesSent = 0;
 {
 LOCK(pnode->cs_vSend);
 bool optimisticSend(pnode->vSendMsg.empty());
 bool optimisticSend(allowOptimisticSend && pnode->vSendMsg.empty());

 //log total amount of bytes per command
 pnode->mapSendBytesPerMsgCmd[msg.command] += nTotalSize;
Expand Down

9 changes: 8 additions & 1 deletion
src/net.h

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,7 @@ class CConnman

 bool IsMasternodeOrDisconnectRequested(const CService& addr);

 void PushMessage(CNode* pnode, CSerializedNetMsg&& msg);
 void PushMessage(CNode* pnode, CSerializedNetMsg&& msg, bool allowOptimisticSend = true);

 template<typename Condition, typename Callable>
 bool ForEachNodeContinueIf(const Condition& cond, Callable&& func)
Expand DownExpand Up@@ -408,6 +408,8 @@ class CConnman
 unsigned int GetReceiveFloodSize() const;

 void WakeMessageHandler();
 void WakeSelect();

private:
 struct ListenSocket {
 SOCKET socket;
Expand DownExpand Up@@ -525,6 +527,11 @@ class CConnman

 CThreadInterrupt interruptNet;

#ifndef WIN32
 /** a pipe which is added to select() calls to wakeup before the timeout */
 int wakeupPipe[2]{-1,-1};
#endif

 std::thread threadDNSAddressSeed;
 std::thread threadSocketHandler;
 std::thread threadOpenAddedConnections;
Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Remove or un-stick sticky/fixed headers that block content
(function() {
function unstick() {
document.querySelectorAll('header, nav, [role="banner"], .header, .navbar, .sticky, .fixed-top, [style*="position: fixed"], [style*="position:sticky"]').forEach(function(el) {
if (el.style.position === 'fixed' || el.style.position === 'sticky' || getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {
el.style.position = 'static';
el.style.top = 'auto';
el.style.zIndex = 'auto';
}
});
}
unstick();
var observer = new MutationObserver(unstick);
observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });
})();
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Optimize LLMQs sending of sig shares#2704

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

codablock
merged 4 commits into

from

Feb 16, 2019

+85

−6

 Merged


Optimize LLMQs sending of sig shares
#2704
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
cf29320

Allow to disable optimistic send in PushMessage()


codablock Feb 15, 2019
acb8789

Implement WakeupSelect() to allow preliminary wakeup after message push


codablock Feb 14, 2019
c03480d

Disable optimistic sending when pushing sig share related messages


codablock Feb 15, 2019
d7bd095

Use pipe() together with fcntl instead of pipe2()


codablock Feb 15, 2019

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


15 changes: 12 additions & 3 deletions
src/llmq/quorums_signing_shares.cpp

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -896,6 +896,8 @@ void CSigSharesManager::SendMessages()
 CollectSigSharesToAnnounce(sigSharesToAnnounce);
 }

 bool didSend = false;

 g_connman->ForEachNode([&](CNode* pnode) {
 CNetMsgMaker msgMaker(pnode->GetSendVersion());

Expand All@@ -905,7 +907,8 @@ void CSigSharesManager::SendMessages()
 assert(p.second.CountSet() != 0);
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QGETSIGSHARES inv={%s}, node=%d\n",
 p.second.ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QGETSIGSHARES, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QGETSIGSHARES, p.second), false);
 didSend = true;
 }
 }

Expand All@@ -915,7 +918,8 @@ void CSigSharesManager::SendMessages()
 assert(!p.second.sigShares.empty());
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QBSIGSHARES inv={%s}, node=%d\n",
 p.second.ToInv().ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QBSIGSHARES, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QBSIGSHARES, p.second), false);
 didSend = true;
 }
 }

Expand All@@ -925,12 +929,17 @@ void CSigSharesManager::SendMessages()
 assert(p.second.CountSet() != 0);
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QSIGSHARESINV inv={%s}, node=%d\n",
 p.second.ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARESINV, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARESINV, p.second), false);
 didSend = true;
 }
 }

 return true;
 });

 if (didSend) {
 g_connman->WakeSelect();
 }
}

void CSigSharesManager::Cleanup()
Expand Down

67 changes: 65 additions & 2 deletions
src/net.cpp

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -1210,6 +1210,18 @@ void CConnman::ThreadSocketHandler()
 SOCKET hSocketMax = 0;
 bool have_fds = false;

#ifndef WIN32
 // We add a pipe to the read set so that the select() call can be woken up from the outside
 // This is done when data is available for sending and at the same time optimistic sending was disabled
 // when pushing the data.
 // This is currently only implemented for POSIX compliant systems. This means that Windows will fall back to
 // timing out after 50ms and then trying to send. This is ok as we assume that heavy-load daemons are usually
 // run on Linux and friends.
 FD_SET(wakeupPipe[0], &fdsetRecv);
 hSocketMax = std::max(hSocketMax, (SOCKET)wakeupPipe[0]);
 have_fds = true;
#endif

 BOOST_FOREACH(const ListenSocket& hListenSocket, vhListenSocket) {
 FD_SET(hListenSocket.socket, &fdsetRecv);
 hSocketMax = std::max(hSocketMax, hListenSocket.socket);
Expand DownExpand Up@@ -1276,6 +1288,20 @@ void CConnman::ThreadSocketHandler()
 return;
 }

#ifndef WIN32
 // drain the wakeup pipe
 if (FD_ISSET(wakeupPipe[0], &fdsetRecv)) {
 LogPrint("net", "woke up select()\n");
 char buf[128];
 while (true) {
 int r = read(wakeupPipe[0], buf, sizeof(buf));
 if (r <= 0) {
 break;
 }
 }
 }
#endif

 //
 // Accept new connections
 //
Expand DownExpand Up@@ -1426,6 +1452,21 @@ void CConnman::WakeMessageHandler()
 condMsgProc.notify_one();
}

void CConnman::WakeSelect()
{
#ifndef WIN32
 if (wakeupPipe[1] == -1) {
 return;
 }

 LogPrint("net", "waking up select()\n");

 char buf[1];
 if (write(wakeupPipe[1], buf, 1) != 1) {
 LogPrint("net", "write to wakeupPipe failed\n");
 }
#endif
}



Expand DownExpand Up@@ -2387,6 +2428,22 @@ bool CConnman::Start(CScheduler& scheduler, std::string& strNodeError, Options c
 fMsgProcWake = false;
 }

#ifndef WIN32
 if (pipe(wakeupPipe) != 0) {
Comment thread
UdjinM6 marked this conversation as resolved.

 wakeupPipe[0] = wakeupPipe[1] = -1;
 LogPrint("net", "pipe() for wakeupPipe failed\n");
 } else {
 int fFlags = fcntl(wakeupPipe[0], F_GETFL, 0);
 if (fcntl(wakeupPipe[0], F_SETFL, fFlags | O_NONBLOCK) == -1) {
 LogPrint("net", "fcntl for O_NONBLOCK on wakeupPipe failed\n");
 }
 fFlags = fcntl(wakeupPipe[1], F_GETFL, 0);
 if (fcntl(wakeupPipe[1], F_SETFL, fFlags | O_NONBLOCK) == -1) {
 LogPrint("net", "fcntl for O_NONBLOCK on wakeupPipe failed\n");
 }
 }
#endif

 // Send and receive from sockets, accept connections
 threadSocketHandler = std::thread(&TraceThread<std::function<void()> >, "net", std::function<void()>(std::bind(&CConnman::ThreadSocketHandler, this)));

Expand DownExpand Up@@ -2512,6 +2569,12 @@ void CConnman::Stop()
 semAddnode = NULL;
 delete semMasternodeOutbound;
 semMasternodeOutbound = NULL;

#ifndef WIN32
 if (wakeupPipe[0] != -1) close(wakeupPipe[0]);
 if (wakeupPipe[1] != -1) close(wakeupPipe[1]);
 wakeupPipe[0] = wakeupPipe[1] = -1;
#endif
}

void CConnman::DeleteNode(CNode* pnode)
Expand DownExpand Up@@ -3054,7 +3117,7 @@ bool CConnman::NodeFullyConnected(const CNode* pnode)
 return pnode && pnode->fSuccessfullyConnected && !pnode->fDisconnect;
}

void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg, bool allowOptimisticSend)
{
 size_t nMessageSize = msg.data.size();
 size_t nTotalSize = nMessageSize + CMessageHeader::HEADER_SIZE;
Expand All@@ -3071,7 +3134,7 @@ void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
 size_t nBytesSent = 0;
 {
 LOCK(pnode->cs_vSend);
 bool optimisticSend(pnode->vSendMsg.empty());
 bool optimisticSend(allowOptimisticSend && pnode->vSendMsg.empty());

 //log total amount of bytes per command
 pnode->mapSendBytesPerMsgCmd[msg.command] += nTotalSize;
Expand Down

9 changes: 8 additions & 1 deletion
src/net.h

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,7 @@ class CConnman

 bool IsMasternodeOrDisconnectRequested(const CService& addr);

 void PushMessage(CNode* pnode, CSerializedNetMsg&& msg);
 void PushMessage(CNode* pnode, CSerializedNetMsg&& msg, bool allowOptimisticSend = true);

 template<typename Condition, typename Callable>
 bool ForEachNodeContinueIf(const Condition& cond, Callable&& func)
Expand DownExpand Up@@ -408,6 +408,8 @@ class CConnman
 unsigned int GetReceiveFloodSize() const;

 void WakeMessageHandler();
 void WakeSelect();

private:
 struct ListenSocket {
 SOCKET socket;
Expand DownExpand Up@@ -525,6 +527,11 @@ class CConnman

 CThreadInterrupt interruptNet;

#ifndef WIN32
 /** a pipe which is added to select() calls to wakeup before the timeout */
 int wakeupPipe[2]{-1,-1};
#endif

 std::thread threadDNSAddressSeed;
 std::thread threadSocketHandler;
 std::thread threadOpenAddedConnections;
Expand Down

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Universal Dark Mode - works on any site
(function() {
var enabled = true;
function applyDarkMode() {
if (!enabled) return;
// Create style element if it doesn't exist
var style = document.getElementById('universal-dark-mode-style');
if (!style) {
style = document.createElement('style');
style.id = 'universal-dark-mode-style';
document.head.appendChild(style);
}
// Dark mode CSS - inverts colors but preserves images/video
style.textContent = '
/* Invert everything except media */
html {
filter: invert(1) hue-rotate(180deg) !important;
background: #1a1a2e !important;
}
/* Restore images, videos, iframes, canvas */
img, video, iframe, canvas, svg, picture, [style*="background-image"] {
filter: invert(1) hue-rotate(180deg) !important;
}
/* Preserve specific elements that should not be inverted */
.no-dark-mode, .no-dark-mode *,
[data-theme="light"], [data-theme="light"],
.ace_editor, .ace_editor *,
.CodeMirror, .CodeMirror *,
.monaco-editor, .monaco-editor *,
.markdown-body pre, .markdown-body pre *,
.highlight, .highlight *,
pre code, pre code * {
filter: none !important;
}
/* Fix common UI elements */
.modal, .popup, .dropdown-menu, .tooltip, .popover {
filter: invert(1) hue-rotate(180deg) !important;
background: #2d2d44 !important;
border-color: #444 !important;
}
/* Scrollbars */
::-webkit-scrollbar { background: #1a1a2e !important; }
::-webkit-scrollbar-thumb { background: #444 !important; }
::-webkit-scrollbar-thumb:hover { background: #555 !important; }
/* Selection */
::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
';
}
function removeDarkMode() {
var style = document.getElementById('universal-dark-mode-style');
if (style) style.remove();
}
// Toggle with Alt+Shift+D
document.addEventListener('keydown', function(e) {
if (e.altKey && e.shiftKey && e.key === 'D') {
e.preventDefault();
enabled = !enabled;
if (enabled) {
applyDarkMode();
console.log('[Universal Dark Mode] Enabled');
} else {
removeDarkMode();
console.log('[Universal Dark Mode] Disabled');
}
}
});
// Apply on load
applyDarkMode();
// Re-apply on dynamic content
var observer = new MutationObserver(function(mutations) {
if (enabled && !document.getElementById('universal-dark-mode-style')) {
applyDarkMode();
}
});
observer.observe(document.head, { childList: true });
console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');
})();
}
} catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); }
})();
})();

Skip to content
Optimize LLMQs sending of sig shares#2704

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

codablock
merged 4 commits into

from

Feb 16, 2019

+85

−6

 Merged


Optimize LLMQs sending of sig shares
#2704
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
cf29320

Allow to disable optimistic send in PushMessage()


codablock Feb 15, 2019
acb8789

Implement WakeupSelect() to allow preliminary wakeup after message push


codablock Feb 14, 2019
c03480d

Disable optimistic sending when pushing sig share related messages


codablock Feb 15, 2019
d7bd095

Use pipe() together with fcntl instead of pipe2()


codablock Feb 15, 2019

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


15 changes: 12 additions & 3 deletions
src/llmq/quorums_signing_shares.cpp

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -896,6 +896,8 @@ void CSigSharesManager::SendMessages()
 CollectSigSharesToAnnounce(sigSharesToAnnounce);
 }

 bool didSend = false;

 g_connman->ForEachNode([&](CNode* pnode) {
 CNetMsgMaker msgMaker(pnode->GetSendVersion());

Expand All@@ -905,7 +907,8 @@ void CSigSharesManager::SendMessages()
 assert(p.second.CountSet() != 0);
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QGETSIGSHARES inv={%s}, node=%d\n",
 p.second.ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QGETSIGSHARES, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QGETSIGSHARES, p.second), false);
 didSend = true;
 }
 }

Expand All@@ -915,7 +918,8 @@ void CSigSharesManager::SendMessages()
 assert(!p.second.sigShares.empty());
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QBSIGSHARES inv={%s}, node=%d\n",
 p.second.ToInv().ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QBSIGSHARES, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QBSIGSHARES, p.second), false);
 didSend = true;
 }
 }

Expand All@@ -925,12 +929,17 @@ void CSigSharesManager::SendMessages()
 assert(p.second.CountSet() != 0);
 LogPrint("llmq", "CSigSharesManager::SendMessages -- QSIGSHARESINV inv={%s}, node=%d\n",
 p.second.ToString(), pnode->id);
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARESINV, p.second));
 g_connman->PushMessage(pnode, msgMaker.Make(NetMsgType::QSIGSHARESINV, p.second), false);
 didSend = true;
 }
 }

 return true;
 });

 if (didSend) {
 g_connman->WakeSelect();
 }
}

void CSigSharesManager::Cleanup()
Expand Down

67 changes: 65 additions & 2 deletions
src/net.cpp

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -1210,6 +1210,18 @@ void CConnman::ThreadSocketHandler()
 SOCKET hSocketMax = 0;
 bool have_fds = false;

#ifndef WIN32
 // We add a pipe to the read set so that the select() call can be woken up from the outside
 // This is done when data is available for sending and at the same time optimistic sending was disabled
 // when pushing the data.
 // This is currently only implemented for POSIX compliant systems. This means that Windows will fall back to
 // timing out after 50ms and then trying to send. This is ok as we assume that heavy-load daemons are usually
 // run on Linux and friends.
 FD_SET(wakeupPipe[0], &fdsetRecv);
 hSocketMax = std::max(hSocketMax, (SOCKET)wakeupPipe[0]);
 have_fds = true;
#endif

 BOOST_FOREACH(const ListenSocket& hListenSocket, vhListenSocket) {
 FD_SET(hListenSocket.socket, &fdsetRecv);
 hSocketMax = std::max(hSocketMax, hListenSocket.socket);
Expand DownExpand Up@@ -1276,6 +1288,20 @@ void CConnman::ThreadSocketHandler()
 return;
 }

#ifndef WIN32
 // drain the wakeup pipe
 if (FD_ISSET(wakeupPipe[0], &fdsetRecv)) {
 LogPrint("net", "woke up select()\n");
 char buf[128];
 while (true) {
 int r = read(wakeupPipe[0], buf, sizeof(buf));
 if (r <= 0) {
 break;
 }
 }
 }
#endif

 //
 // Accept new connections
 //
Expand DownExpand Up@@ -1426,6 +1452,21 @@ void CConnman::WakeMessageHandler()
 condMsgProc.notify_one();
}

void CConnman::WakeSelect()
{
#ifndef WIN32
 if (wakeupPipe[1] == -1) {
 return;
 }

 LogPrint("net", "waking up select()\n");

 char buf[1];
 if (write(wakeupPipe[1], buf, 1) != 1) {
 LogPrint("net", "write to wakeupPipe failed\n");
 }
#endif
}



Expand DownExpand Up@@ -2387,6 +2428,22 @@ bool CConnman::Start(CScheduler& scheduler, std::string& strNodeError, Options c
 fMsgProcWake = false;
 }

#ifndef WIN32
 if (pipe(wakeupPipe) != 0) {
Comment thread
UdjinM6 marked this conversation as resolved.

 wakeupPipe[0] = wakeupPipe[1] = -1;
 LogPrint("net", "pipe() for wakeupPipe failed\n");
 } else {
 int fFlags = fcntl(wakeupPipe[0], F_GETFL, 0);
 if (fcntl(wakeupPipe[0], F_SETFL, fFlags | O_NONBLOCK) == -1) {
 LogPrint("net", "fcntl for O_NONBLOCK on wakeupPipe failed\n");
 }
 fFlags = fcntl(wakeupPipe[1], F_GETFL, 0);
 if (fcntl(wakeupPipe[1], F_SETFL, fFlags | O_NONBLOCK) == -1) {
 LogPrint("net", "fcntl for O_NONBLOCK on wakeupPipe failed\n");
 }
 }
#endif

 // Send and receive from sockets, accept connections
 threadSocketHandler = std::thread(&TraceThread<std::function<void()> >, "net", std::function<void()>(std::bind(&CConnman::ThreadSocketHandler, this)));

Expand DownExpand Up@@ -2512,6 +2569,12 @@ void CConnman::Stop()
 semAddnode = NULL;
 delete semMasternodeOutbound;
 semMasternodeOutbound = NULL;

#ifndef WIN32
 if (wakeupPipe[0] != -1) close(wakeupPipe[0]);
 if (wakeupPipe[1] != -1) close(wakeupPipe[1]);
 wakeupPipe[0] = wakeupPipe[1] = -1;
#endif
}

void CConnman::DeleteNode(CNode* pnode)
Expand DownExpand Up@@ -3054,7 +3117,7 @@ bool CConnman::NodeFullyConnected(const CNode* pnode)
 return pnode && pnode->fSuccessfullyConnected && !pnode->fDisconnect;
}

void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg, bool allowOptimisticSend)
{
 size_t nMessageSize = msg.data.size();
 size_t nTotalSize = nMessageSize + CMessageHeader::HEADER_SIZE;
Expand All@@ -3071,7 +3134,7 @@ void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg)
 size_t nBytesSent = 0;
 {
 LOCK(pnode->cs_vSend);
 bool optimisticSend(pnode->vSendMsg.empty());
 bool optimisticSend(allowOptimisticSend && pnode->vSendMsg.empty());

 //log total amount of bytes per command
 pnode->mapSendBytesPerMsgCmd[msg.command] += nTotalSize;
Expand Down

9 changes: 8 additions & 1 deletion
src/net.h

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -201,7 +201,7 @@ class CConnman

 bool IsMasternodeOrDisconnectRequested(const CService& addr);

 void PushMessage(CNode* pnode, CSerializedNetMsg&& msg);
 void PushMessage(CNode* pnode, CSerializedNetMsg&& msg, bool allowOptimisticSend = true);

 template<typename Condition, typename Callable>
 bool ForEachNodeContinueIf(const Condition& cond, Callable&& func)
Expand DownExpand Up@@ -408,6 +408,8 @@ class CConnman
 unsigned int GetReceiveFloodSize() const;

 void WakeMessageHandler();
 void WakeSelect();

private:
 struct ListenSocket {
 SOCKET socket;
Expand DownExpand Up@@ -525,6 +527,11 @@ class CConnman

 CThreadInterrupt interruptNet;

#ifndef WIN32
 /** a pipe which is added to select() calls to wakeup before the timeout */
 int wakeupPipe[2]{-1,-1};
#endif

 std::thread threadDNSAddressSeed;
 std::thread threadSocketHandler;
 std::thread threadOpenAddedConnections;
Expand Down