From 90631adb9beda188bd00494f1a84446c0b94d0f9 Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 13 Nov 2018 12:07:58 -0800 Subject: Improve multithreading support for OneService (faster, dynamic adjustment of thread count based on HW concurrency). --- osdep/BlockingQueue.hpp | 23 +++++++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) (limited to 'osdep') diff --git a/osdep/BlockingQueue.hpp b/osdep/BlockingQueue.hpp index 351a095a..03986efe 100644 --- a/osdep/BlockingQueue.hpp +++ b/osdep/BlockingQueue.hpp @@ -32,6 +32,8 @@ #include #include +#include "Thread.hpp" + namespace ZeroTier { /** @@ -52,6 +54,23 @@ public: c.notify_one(); } + inline void postWait(T t,unsigned long maxQueueSize) + { + for(;;) { + { + std::lock_guard lock(m); + if (q.size() < maxQueueSize) { + q.push(t); + c.notify_one(); + return; + } + } + if (!r) + break; + Thread::sleep(1); + } + } + inline void stop(void) { std::lock_guard lock(m); @@ -98,8 +117,8 @@ public: private: volatile bool r; std::queue q; - std::mutex m; - std::condition_variable c; + mutable std::mutex m; + mutable std::condition_variable c; }; } // namespace ZeroTier -- cgit v1.2.3 From 350116c513a8d6ea48842245b1d85287b834fbf5 Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 13 Nov 2018 12:42:03 -0800 Subject: Stability improvements for Mac virtual Ethernet tap driver. --- osdep/MacEthernetTap.cpp | 11 +++++++---- osdep/MacEthernetTapAgent.c | 11 ++++++----- 2 files changed, 13 insertions(+), 9 deletions(-) (limited to 'osdep') diff --git a/osdep/MacEthernetTap.cpp b/osdep/MacEthernetTap.cpp index 92df2f31..1cfb9375 100644 --- a/osdep/MacEthernetTap.cpp +++ b/osdep/MacEthernetTap.cpp @@ -147,7 +147,7 @@ MacEthernetTap::MacEthernetTap( _agentStdin2 = agentStdin[0]; _agentStdout2 = agentStdout[1]; _agentStderr2 = agentStderr[1]; - long apid = (long)vfork(); + long apid = (long)fork(); if (apid < 0) { throw std::runtime_error("fork failed"); } else if (apid == 0) { @@ -155,10 +155,13 @@ MacEthernetTap::MacEthernetTap( ::dup2(agentStdout[1],STDOUT_FILENO); ::dup2(agentStderr[1],STDERR_FILENO); ::close(agentStdin[0]); + ::close(agentStdin[1]); + ::close(agentStdout[0]); ::close(agentStdout[1]); + ::close(agentStderr[0]); ::close(agentStderr[1]); ::execl(agentPath.c_str(),agentPath.c_str(),devnostr,ethaddr,mtustr,metricstr,(char *)0); - ::exit(-1); + ::_exit(-1); } else { _agentPid = apid; } @@ -356,8 +359,8 @@ void MacEthernetTap::threadMain() const int nfds = std::max(std::max(_shutdownSignalPipe[0],_agentStdout),_agentStderr) + 1; long agentReadPtr = 0; - fcntl(_agentStdout,F_SETFL,O_NONBLOCK); - fcntl(_agentStderr,F_SETFL,O_NONBLOCK); + fcntl(_agentStdout,F_SETFL,fcntl(_agentStdout,F_GETFL)|O_NONBLOCK); + fcntl(_agentStderr,F_SETFL,fcntl(_agentStderr,F_GETFL)|O_NONBLOCK); FD_ZERO(&readfds); FD_ZERO(&nullfds); diff --git a/osdep/MacEthernetTapAgent.c b/osdep/MacEthernetTapAgent.c index ca1f7a4e..7eaac871 100644 --- a/osdep/MacEthernetTapAgent.c +++ b/osdep/MacEthernetTapAgent.c @@ -104,8 +104,8 @@ #define P_IFCONFIG "/sbin/ifconfig" -static unsigned char s_pktReadBuf[262144] __attribute__ ((__aligned__(16))); -static unsigned char s_stdinReadBuf[262144] __attribute__ ((__aligned__(16))); +static unsigned char s_pktReadBuf[524288] __attribute__ ((__aligned__(16))); +static unsigned char s_stdinReadBuf[524288] __attribute__ ((__aligned__(16))); static char s_deviceName[IFNAMSIZ]; static char s_peerDeviceName[IFNAMSIZ]; static int s_bpffd = -1; @@ -175,7 +175,7 @@ static int run(const char *path,...) } else if (pid == 0) { dup2(STDERR_FILENO,STDOUT_FILENO); execv(args[0],args); - exit(-1); + _exit(-1); } int rv = 0; waitpid(pid,&rv,0); @@ -322,7 +322,6 @@ int main(int argc,char **argv) return ZT_MACETHERNETTAPAGENT_EXIT_CODE_UNABLE_TO_CREATE; } - fcntl(STDIN_FILENO,F_SETFL,fcntl(STDIN_FILENO,F_GETFL)|O_NONBLOCK); fcntl(s_ndrvfd,F_SETFL,fcntl(s_ndrvfd,F_GETFL)|O_NONBLOCK); fcntl(s_bpffd,F_SETFL,fcntl(s_bpffd,F_GETFL)|O_NONBLOCK); @@ -381,6 +380,7 @@ int main(int argc,char **argv) } } break; + case ZT_MACETHERNETTAPAGENT_STDIN_CMD_IFCONFIG: { char *args[16]; args[0] = P_IFCONFIG; @@ -410,12 +410,13 @@ int main(int argc,char **argv) } else if (pid == 0) { dup2(STDERR_FILENO,STDOUT_FILENO); execv(args[0],args); - exit(-1); + _exit(-1); } int rv = 0; waitpid(pid,&rv,0); } } break; + case ZT_MACETHERNETTAPAGENT_STDIN_CMD_EXIT: return ZT_MACETHERNETTAPAGENT_EXIT_CODE_SUCCESS; } -- cgit v1.2.3 From 4ed7d20a4880a168a447eb0b5d02b3c643ab0cff Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 13 Nov 2018 12:46:36 -0800 Subject: Another stability fix... turns out vfork is problematic here. --- osdep/MacEthernetTapAgent.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'osdep') diff --git a/osdep/MacEthernetTapAgent.c b/osdep/MacEthernetTapAgent.c index 7eaac871..91eff25c 100644 --- a/osdep/MacEthernetTapAgent.c +++ b/osdep/MacEthernetTapAgent.c @@ -404,7 +404,7 @@ int main(int argc,char **argv) } args[argNo] = (char *)0; if (argNo > 2) { - pid_t pid = vfork(); + pid_t pid = fork(); if (pid < 0) { return -1; } else if (pid == 0) { -- cgit v1.2.3 From 6684559cd9b8a96da29c7a59c8d787918a4171f9 Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 13 Nov 2018 13:51:25 -0800 Subject: More Mac tap improvements and threading efficiency improvements. --- osdep/BlockingQueue.hpp | 26 +++++++++++++------------- osdep/MacEthernetTap.cpp | 4 ++-- osdep/MacEthernetTap.hpp | 2 ++ osdep/MacEthernetTapAgent.c | 11 ++--------- service/OneService.cpp | 6 ++++-- 5 files changed, 23 insertions(+), 26 deletions(-) (limited to 'osdep') diff --git a/osdep/BlockingQueue.hpp b/osdep/BlockingQueue.hpp index 03986efe..6f81ddf2 100644 --- a/osdep/BlockingQueue.hpp +++ b/osdep/BlockingQueue.hpp @@ -54,20 +54,16 @@ public: c.notify_one(); } - inline void postWait(T t,unsigned long maxQueueSize) + inline void postLimit(T t,const unsigned long limit) { + std::unique_lock lock(m); for(;;) { - { - std::lock_guard lock(m); - if (q.size() < maxQueueSize) { - q.push(t); - c.notify_one(); - return; - } - } - if (!r) + if (q.size() < limit) { + q.push(t); + c.notify_one(); break; - Thread::sleep(1); + } + gc.wait(lock); } } @@ -84,10 +80,14 @@ public: if (!r) return false; while (q.empty()) { c.wait(lock); - if (!r) return false; + if (!r) { + gc.notify_all(); + return false; + } } value = q.front(); q.pop(); + gc.notify_all(); return true; } @@ -118,7 +118,7 @@ private: volatile bool r; std::queue q; mutable std::mutex m; - mutable std::condition_variable c; + mutable std::condition_variable c,gc; }; } // namespace ZeroTier diff --git a/osdep/MacEthernetTap.cpp b/osdep/MacEthernetTap.cpp index 1cfb9375..fb3e3a75 100644 --- a/osdep/MacEthernetTap.cpp +++ b/osdep/MacEthernetTap.cpp @@ -287,7 +287,9 @@ void MacEthernetTap::put(const MAC &from,const MAC &to,unsigned int etherType,co iov[1].iov_len = 15; iov[2].iov_base = const_cast(data); iov[2].iov_len = len; + _putLock.lock(); writev(_agentStdin,iov,3); + _putLock.unlock(); } } @@ -396,8 +398,6 @@ void MacEthernetTap::threadMain() break; } } - } else { - break; } } if (FD_ISSET(_agentStderr,&readfds)) { diff --git a/osdep/MacEthernetTap.hpp b/osdep/MacEthernetTap.hpp index 4b3ac019..eaf131a7 100644 --- a/osdep/MacEthernetTap.hpp +++ b/osdep/MacEthernetTap.hpp @@ -38,6 +38,7 @@ #include "../node/MAC.hpp" #include "../node/InetAddress.hpp" #include "../node/MulticastGroup.hpp" +#include "../node/Mutex.hpp" #include "Thread.hpp" @@ -80,6 +81,7 @@ private: std::string _homePath; std::string _dev; std::vector _multicastGroups; + Mutex _putLock; unsigned int _mtu; unsigned int _metric; int _shutdownSignalPipe[2]; diff --git a/osdep/MacEthernetTapAgent.c b/osdep/MacEthernetTapAgent.c index 91eff25c..a595e154 100644 --- a/osdep/MacEthernetTapAgent.c +++ b/osdep/MacEthernetTapAgent.c @@ -104,8 +104,8 @@ #define P_IFCONFIG "/sbin/ifconfig" -static unsigned char s_pktReadBuf[524288] __attribute__ ((__aligned__(16))); -static unsigned char s_stdinReadBuf[524288] __attribute__ ((__aligned__(16))); +static unsigned char s_pktReadBuf[262144] __attribute__ ((__aligned__(16))); +static unsigned char s_stdinReadBuf[262144] __attribute__ ((__aligned__(16))); static char s_deviceName[IFNAMSIZ]; static char s_peerDeviceName[IFNAMSIZ]; static int s_bpffd = -1; @@ -322,9 +322,6 @@ int main(int argc,char **argv) return ZT_MACETHERNETTAPAGENT_EXIT_CODE_UNABLE_TO_CREATE; } - fcntl(s_ndrvfd,F_SETFL,fcntl(s_ndrvfd,F_GETFL)|O_NONBLOCK); - fcntl(s_bpffd,F_SETFL,fcntl(s_bpffd,F_GETFL)|O_NONBLOCK); - fprintf(stderr,"I %s %s %d.%d.%d.%d\n",s_deviceName,s_peerDeviceName,ZEROTIER_ONE_VERSION_MAJOR,ZEROTIER_ONE_VERSION_MINOR,ZEROTIER_ONE_VERSION_REVISION,ZEROTIER_ONE_VERSION_BUILD); FD_ZERO(&rfds); @@ -357,8 +354,6 @@ int main(int argc,char **argv) } p += BPF_WORDALIGN(h->bh_hdrlen + h->bh_caplen); } - } else { - return ZT_MACETHERNETTAPAGENT_EXIT_CODE_READ_ERROR; } } @@ -431,8 +426,6 @@ int main(int argc,char **argv) break; } } - } else { - return ZT_MACETHERNETTAPAGENT_EXIT_CODE_READ_ERROR; } } } diff --git a/service/OneService.cpp b/service/OneService.cpp index bf24466d..1351cbfb 100644 --- a/service/OneService.cpp +++ b/service/OneService.cpp @@ -472,6 +472,7 @@ public: unsigned int _tertiaryPort; volatile unsigned int _udpPortPickerCounter; + unsigned long _incomingPacketConcurrency; std::vector _incomingPacketMemoryPool; BlockingQueue _incomingPacketQueue; std::vector _incomingPacketThreads; @@ -606,7 +607,8 @@ public: _ports[1] = 0; _ports[2] = 0; - for(long t=0;tsize = (unsigned int)len; ZT_FAST_MEMCPY(pkt->data,data,len); - _incomingPacketQueue.postWait(pkt,64); + _incomingPacketQueue.postLimit(pkt,16 * _incomingPacketConcurrency); } inline void phyOnTcpConnect(PhySocket *sock,void **uptr,bool success) -- cgit v1.2.3 From 690bd933d52c7dbbcddde7c0aff30f7fee91a6d9 Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 13 Nov 2018 13:52:45 -0800 Subject: Support shutdown with postLimit in BlockingQueue --- osdep/BlockingQueue.hpp | 3 +++ 1 file changed, 3 insertions(+) (limited to 'osdep') diff --git a/osdep/BlockingQueue.hpp b/osdep/BlockingQueue.hpp index 6f81ddf2..9e2f73cb 100644 --- a/osdep/BlockingQueue.hpp +++ b/osdep/BlockingQueue.hpp @@ -63,6 +63,8 @@ public: c.notify_one(); break; } + if (!r) + break; gc.wait(lock); } } @@ -72,6 +74,7 @@ public: std::lock_guard lock(m); r = false; c.notify_all(); + gc.notify_all(); } inline bool get(T &value) -- cgit v1.2.3