Merge pull request #2363 from zerotier/dev

1.14.1 merge dev to main
This commit is contained in:
Adam Ierymenko 2024-09-12 18:51:36 -04:00 committed by GitHub
commit 8140264cb6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
46 changed files with 1636 additions and 835 deletions

6
.clangd Normal file
View File

@ -0,0 +1,6 @@
CompileFlags:
Add:
- "-std=c++17"
- "-I../ext"
- "-I../ext/prometheus-cpp-lite-1.0/core/include"
- "-I../ext/prometheus-cpp-lite-1.0/simpleapi/include"

View File

@ -9,13 +9,12 @@ jobs:
git config --global core.autocrlf input git config --global core.autocrlf input
# git config --global core.eol lf # git config --global core.eol lf
- name: checkout - name: checkout
uses: actions/checkout@v3 uses: actions/checkout@v4
- name: Install Rust - name: Install Rust
uses: actions-rs/toolchain@v1 uses: dtolnay/rust-toolchain@stable
with: with:
toolchain: stable toolchain: stable
target: x86_64-unknown-linux-gnu targets: x86_64-unknown-linux-gnu
override: true
components: rustfmt, clippy components: rustfmt, clippy
- name: Set up cargo cache - name: Set up cargo cache
@ -33,6 +32,14 @@ jobs:
run: | run: |
make selftest make selftest
./zerotier-selftest ./zerotier-selftest
- name: 'Tar files' # keeps permissions (execute)
run: tar -cvf zerotier-one.tar zerotier-one
- name: Archive production artifacts
uses: actions/upload-artifact@v4
with:
name: zerotier-one-ubuntu-x64
path: zerotier-one.tar
retention-days: 7
build_macos: build_macos:
runs-on: macos-latest runs-on: macos-latest
@ -42,13 +49,18 @@ jobs:
git config --global core.autocrlf input git config --global core.autocrlf input
# git config --global core.eol lf # git config --global core.eol lf
- name: checkout - name: checkout
uses: actions/checkout@v3 uses: actions/checkout@v4
- name: Install Rust - name: Install Rust aarch64
uses: actions-rs/toolchain@v1 uses: dtolnay/rust-toolchain@stable
with: with:
toolchain: stable toolchain: stable
target: aarch64-apple-darwin target: aarch64-apple-darwin
override: true components: rustfmt, clippy
- name: Install Rust x86_64
uses: dtolnay/rust-toolchain@stable
with:
toolchain: stable
target: x86_64-apple-darwin
components: rustfmt, clippy components: rustfmt, clippy
- name: Set up cargo cache - name: Set up cargo cache
uses: Swatinem/rust-cache@v2 uses: Swatinem/rust-cache@v2
@ -58,13 +70,21 @@ jobs:
shared-key: ${{ runner.os }}-cargo- shared-key: ${{ runner.os }}-cargo-
workspaces: | workspaces: |
rustybits/ rustybits/
- name: make - name: make
run: make run: make
- name: selftest - name: selftest
run: | run: |
make selftest make selftest
./zerotier-selftest ./zerotier-selftest
- name: 'Tar files' # keeps permissions (execute)
run: tar -cvf zerotier-one.tar zerotier-one
- name: Archive production artifacts
uses: actions/upload-artifact@v4
with:
name: zerotier-one-mac
path: zerotier-one.tar
retention-days: 7
build_windows: build_windows:
runs-on: windows-latest runs-on: windows-latest
@ -74,13 +94,12 @@ jobs:
git config --global core.autocrlf true git config --global core.autocrlf true
# git config --global core.eol lf # git config --global core.eol lf
- name: checkout - name: checkout
uses: actions/checkout@v3 uses: actions/checkout@v4
- name: Install Rust - name: Install Rust
uses: actions-rs/toolchain@v1 uses: dtolnay/rust-toolchain@stable
with: with:
toolchain: stable toolchain: stable
target: aarch64-apple-darwin target: aarch64-apple-darwin
override: true
components: rustfmt, clippy components: rustfmt, clippy
- name: Set up cargo cache - name: Set up cargo cache
uses: Swatinem/rust-cache@v2 uses: Swatinem/rust-cache@v2
@ -92,7 +111,13 @@ jobs:
rustybits/ rustybits/
- name: setup msbuild - name: setup msbuild
uses: microsoft/setup-msbuild@v1.1.3 uses: microsoft/setup-msbuild@v2
- name: msbuild - name: msbuild
run: | run: |
msbuild windows\ZeroTierOne.sln /m /p:Configuration=Release /property:Platform=x64 /t:ZeroTierOne msbuild windows\ZeroTierOne.sln /m /p:Configuration=Release /property:Platform=x64 /t:ZeroTierOne
- name: Archive production artifacts
uses: actions/upload-artifact@v4
with:
name: zerotier-one-windows
path: windows/Build
retention-days: 7

1
.gitignore vendored
View File

@ -124,6 +124,7 @@ attic/world/mkworld
workspace/ workspace/
workspace2/ workspace2/
zeroidc/target/ zeroidc/target/
tcp-proxy/target
#snapcraft specifics #snapcraft specifics
/parts/ /parts/

View File

@ -14,6 +14,7 @@ The version must be incremented in all of the following files:
/debian/changelog /debian/changelog
/ext/installfiles/mac/ZeroTier One.pkgproj /ext/installfiles/mac/ZeroTier One.pkgproj
/ext/installfiles/windows/ZeroTier One.aip /ext/installfiles/windows/ZeroTier One.aip
../DesktopUI/mac-app-template/ZeroTier.app/Contents/Info.plist
The final .AIP file can only be edited on Windows with [Advanced Installer Enterprise](http://www.advancedinstaller.com/). In addition to incrementing the version be sure that a new product code is generated. (The "upgrade code" GUID on the other hand must never change.) The final .AIP file can only be edited on Windows with [Advanced Installer Enterprise](http://www.advancedinstaller.com/). In addition to incrementing the version be sure that a new product code is generated. (The "upgrade code" GUID on the other hand must never change.)

View File

@ -58,7 +58,7 @@ To build on Mac and Linux just type `make`. On FreeBSD and OpenBSD `gmake` (GNU
- Xcode command line tools for macOS 10.13 or newer are required. - Xcode command line tools for macOS 10.13 or newer are required.
- Rust for x86_64 and ARM64 targets *if SSO is enabled in the build*. - Rust for x86_64 and ARM64 targets *if SSO is enabled in the build*.
- **Linux** - **Linux**
- The minimum compiler versions required are GCC/G++ 4.9.3 or CLANG/CLANG++ 3.4.2. (Install `clang` on CentOS 7 as G++ is too old.) - The minimum compiler versions required are GCC/G++ 8.x or CLANG/CLANG++ 5.x.
- Linux makefiles automatically detect and prefer clang/clang++ if present as it produces smaller and slightly faster binaries in most cases. You can override by supplying CC and CXX variables on the make command line. - Linux makefiles automatically detect and prefer clang/clang++ if present as it produces smaller and slightly faster binaries in most cases. You can override by supplying CC and CXX variables on the make command line.
- Rust for x86_64 and ARM64 targets *if SSO is enabled in the build*. - Rust for x86_64 and ARM64 targets *if SSO is enabled in the build*.
- **Windows** - **Windows**

View File

@ -1,6 +1,22 @@
ZeroTier Release Notes ZeroTier Release Notes
====== ======
# 2024-09-12 -- Version 1.14.1
* Multithreaded packet I/O support! Currently this is just for Linux and must
be enabled in local.conf. It will likely make the largest difference on small
multi-core devices where CPU is a bottleneck and high throughput is desired.
It may be enabled by default in the future but we want it to be thoroughly
tested. It's a little harder than it seems at first glance due to the need
to keep packets in sequence and balance load.
* Several multipath bug fixes.
* Updated the versions on a number of libraries related to OIDC support and HTTP.
* MacOS .app now shows the correct version in its Info.plist manifest.
* Sanitize MAC addresses in JSON format rules parser.
* Some basic information about the platform (OS, CPU architecture) is now reported
to network controllers when networks are joined so it can be displayed to
network admins and in the future used in policy checking and inventory operations.
# 2024-05-02 -- Version 1.14.0 # 2024-05-02 -- Version 1.14.0
* Linux I/O performance improvements under heavy load * Linux I/O performance improvements under heavy load

View File

@ -382,6 +382,24 @@ void DB::_networkChanged(nlohmann::json &old,nlohmann::json &networkConfig,bool
const std::string ids = old["id"]; const std::string ids = old["id"];
const uint64_t networkId = Utils::hexStrToU64(ids.c_str()); const uint64_t networkId = Utils::hexStrToU64(ids.c_str());
if (networkId) { if (networkId) {
try {
// deauth all members on the network
nlohmann::json network;
std::vector<nlohmann::json> members;
this->get(networkId, network, members);
for(auto i=members.begin();i!=members.end();++i) {
const std::string nodeID = (*i)["id"];
const uint64_t memberId = Utils::hexStrToU64(nodeID.c_str());
std::unique_lock<std::shared_mutex> ll(_changeListeners_l);
for(auto j=_changeListeners.begin();j!=_changeListeners.end();++j) {
(*j)->onNetworkMemberDeauthorize(this,networkId,memberId);
}
}
} catch (std::exception &e) {
std::cerr << "Error deauthorizing members on network delete: " << e.what() << std::endl;
}
// delete the network
std::unique_lock<std::shared_mutex> l(_networks_l); std::unique_lock<std::shared_mutex> l(_networks_l);
_networks.erase(networkId); _networks.erase(networkId);
} }

View File

@ -315,12 +315,14 @@ static bool _parseRule(json &r,ZT_VirtualNetworkRule &rule)
return true; return true;
} else if (t == "MATCH_MAC_SOURCE") { } else if (t == "MATCH_MAC_SOURCE") {
rule.t |= ZT_NETWORK_RULE_MATCH_MAC_SOURCE; rule.t |= ZT_NETWORK_RULE_MATCH_MAC_SOURCE;
const std::string mac(OSUtils::jsonString(r["mac"],"0")); std::string mac(OSUtils::jsonString(r["mac"],"0"));
Utils::cleanMac(mac);
Utils::unhex(mac.c_str(),(unsigned int)mac.length(),rule.v.mac,6); Utils::unhex(mac.c_str(),(unsigned int)mac.length(),rule.v.mac,6);
return true; return true;
} else if (t == "MATCH_MAC_DEST") { } else if (t == "MATCH_MAC_DEST") {
rule.t |= ZT_NETWORK_RULE_MATCH_MAC_DEST; rule.t |= ZT_NETWORK_RULE_MATCH_MAC_DEST;
const std::string mac(OSUtils::jsonString(r["mac"],"0")); std::string mac(OSUtils::jsonString(r["mac"],"0"));
Utils::cleanMac(mac);
Utils::unhex(mac.c_str(),(unsigned int)mac.length(),rule.v.mac,6); Utils::unhex(mac.c_str(),(unsigned int)mac.length(),rule.v.mac,6);
return true; return true;
} else if (t == "MATCH_IPV4_SOURCE") { } else if (t == "MATCH_IPV4_SOURCE") {

View File

@ -780,11 +780,25 @@ void PostgreSQL::initializeNetworks()
fprintf(stderr, "adding networks to redis...\n"); fprintf(stderr, "adding networks to redis...\n");
if (_rc->clusterMode) { if (_rc->clusterMode) {
auto tx = _cluster->transaction(_myAddressStr, true, false); auto tx = _cluster->transaction(_myAddressStr, true, false);
tx.sadd(setKey, networkSet.begin(), networkSet.end()); uint64_t count = 0;
for (std::string nwid : networkSet) {
tx.sadd(setKey, nwid);
if (++count % 30000 == 0) {
tx.exec();
tx = _cluster->transaction(_myAddressStr, true, false);
}
}
tx.exec(); tx.exec();
} else { } else {
auto tx = _redis->transaction(true, false); auto tx = _redis->transaction(true, false);
tx.sadd(setKey, networkSet.begin(), networkSet.end()); uint64_t count = 0;
for (std::string nwid : networkSet) {
tx.sadd(setKey, nwid);
if (++count % 30000 == 0) {
tx.exec();
tx = _redis->transaction(true, false);
}
}
tx.exec(); tx.exec();
} }
fprintf(stderr, "done.\n"); fprintf(stderr, "done.\n");
@ -1005,14 +1019,24 @@ void PostgreSQL::initializeMembers()
fprintf(stderr, "Load member data into redis...\n"); fprintf(stderr, "Load member data into redis...\n");
if (_rc->clusterMode) { if (_rc->clusterMode) {
auto tx = _cluster->transaction(_myAddressStr, true, false); auto tx = _cluster->transaction(_myAddressStr, true, false);
uint64_t count = 0;
for (auto it : networkMembers) { for (auto it : networkMembers) {
tx.sadd(it.first, it.second); tx.sadd(it.first, it.second);
if (++count % 30000 == 0) {
tx.exec();
tx = _cluster->transaction(_myAddressStr, true, false);
}
} }
tx.exec(); tx.exec();
} else { } else {
auto tx = _redis->transaction(true, false); auto tx = _redis->transaction(true, false);
uint64_t count = 0;
for (auto it : networkMembers) { for (auto it : networkMembers) {
tx.sadd(it.first, it.second); tx.sadd(it.first, it.second);
if (++count % 30000 == 0) {
tx.exec();
tx = _redis->transaction(true, false);
}
} }
tx.exec(); tx.exec();
} }
@ -1180,7 +1204,7 @@ void PostgreSQL::_membersWatcher_Redis() {
_memberChanged(oldConfig,newConfig,(this->_ready >= 2)); _memberChanged(oldConfig,newConfig,(this->_ready >= 2));
} }
} catch (...) { } catch (...) {
fprintf(stderr, "json parse error in networkWatcher_Redis\n"); fprintf(stderr, "json parse error in _membersWatcher_Redis: %s\n", a.second.c_str());
} }
} }
if (_rc->clusterMode) { if (_rc->clusterMode) {
@ -1269,8 +1293,8 @@ void PostgreSQL::_networksWatcher_Redis() {
if (oldConfig.is_object()||newConfig.is_object()) { if (oldConfig.is_object()||newConfig.is_object()) {
_networkChanged(oldConfig,newConfig,(this->_ready >= 2)); _networkChanged(oldConfig,newConfig,(this->_ready >= 2));
} }
} catch (...) { } catch (std::exception &e) {
fprintf(stderr, "json parse error in networkWatcher_Redis\n"); fprintf(stderr, "json parse error in networkWatcher_Redis: what: %s json: %s\n", e.what(), a.second.c_str());
} }
} }
if (_rc->clusterMode) { if (_rc->clusterMode) {

6
debian/changelog vendored
View File

@ -1,3 +1,9 @@
zerotier-one (1.14.1) unstable; urgency=medium
* See RELEASE-NOTES.md for release notes.
-- Adam Ierymenko <adam.ierymenko@zerotier.com> Wed, 11 Sep 2024 01:00:00 -0700
zerotier-one (1.14.0) unstable; urgency=medium zerotier-one (1.14.0) unstable; urgency=medium
* See RELEASE-NOTES.md for release notes. * See RELEASE-NOTES.md for release notes.

View File

@ -701,7 +701,7 @@
<key>USE_HFS+_COMPRESSION</key> <key>USE_HFS+_COMPRESSION</key>
<false/> <false/>
<key>VERSION</key> <key>VERSION</key>
<string>1.14.0</string> <string>1.14.1</string>
</dict> </dict>
<key>TYPE</key> <key>TYPE</key>
<integer>0</integer> <integer>0</integer>

View File

@ -111,6 +111,44 @@ namespace {
bool finishInitializing(); bool finishInitializing();
}; };
//
// RAII construct for calling AttachCurrentThread and DetachCurrent automatically
//
struct ScopedJNIThreadAttacher {
JavaVM *jvm;
JNIEnv **env_p;
jint getEnvRet;
ScopedJNIThreadAttacher(JavaVM *jvmIn, JNIEnv **env_pIn, jint getEnvRetIn) :
jvm(jvmIn),
env_p(env_pIn),
getEnvRet(getEnvRetIn) {
if (getEnvRet != JNI_EDETACHED) {
return;
}
jint attachCurrentThreadRet;
if ((attachCurrentThreadRet = jvm->AttachCurrentThread(env_p, NULL)) != JNI_OK) {
LOGE("Error calling AttachCurrentThread: %d", attachCurrentThreadRet);
assert(false && "Error calling AttachCurrentThread");
}
}
~ScopedJNIThreadAttacher() {
if (getEnvRet != JNI_EDETACHED) {
return;
}
jint detachCurrentThreadRet;
if ((detachCurrentThreadRet = jvm->DetachCurrentThread()) != JNI_OK) {
LOGE("Error calling DetachCurrentThread: %d", detachCurrentThreadRet);
assert(false && "Error calling DetachCurrentThread");
}
}
};
/* /*
* This must return 0 on success. It can return any OS-dependent error code * This must return 0 on success. It can return any OS-dependent error code
@ -194,7 +232,25 @@ namespace {
assert(ref); assert(ref);
assert(ref->node == node); assert(ref->node == node);
JNIEnv *env; JNIEnv *env;
GETENV(env, ref->jvm);
jint getEnvRet;
assert(ref->jvm);
getEnvRet = ref->jvm->GetEnv(reinterpret_cast<void**>(&env), JNI_VERSION_1_6);
if (!(getEnvRet == JNI_OK || getEnvRet == JNI_EDETACHED)) {
LOGE("Error calling GetEnv: %d", getEnvRet);
assert(false && "Error calling GetEnv");
}
//
// Thread might actually be detached.
//
// e.g:
// https://github.com/zerotier/ZeroTierOne/blob/91e7ce87f09ac1cfdeaf6ff22c3cedcd93574c86/node/Switch.cpp#L519
//
// Make sure to attach if needed
//
ScopedJNIThreadAttacher attacher{ref->jvm, &env, getEnvRet};
if (env->ExceptionCheck()) { if (env->ExceptionCheck()) {
LOGE("Unhandled pending exception"); LOGE("Unhandled pending exception");

View File

@ -71,7 +71,7 @@ else
override CFLAGS+=-Wall -Wno-deprecated -pthread $(INCLUDES) -DNDEBUG $(DEFS) override CFLAGS+=-Wall -Wno-deprecated -pthread $(INCLUDES) -DNDEBUG $(DEFS)
CXXFLAGS?=-O3 -fstack-protector CXXFLAGS?=-O3 -fstack-protector
override CXXFLAGS+=-Wall -Wno-deprecated -std=c++17 -pthread $(INCLUDES) -DNDEBUG $(DEFS) override CXXFLAGS+=-Wall -Wno-deprecated -std=c++17 -pthread $(INCLUDES) -DNDEBUG $(DEFS)
LDFLAGS=-pie -Wl,-z,relro,-z,now LDFLAGS?=-pie -Wl,-z,relro,-z,now
ZT_CARGO_FLAGS=--release ZT_CARGO_FLAGS=--release
endif endif
@ -364,7 +364,7 @@ override CFLAGS+=-fPIC -fPIE
override CXXFLAGS+=-fPIC -fPIE override CXXFLAGS+=-fPIC -fPIE
# Non-executable stack # Non-executable stack
override ASFLAGS+=--noexecstack override LDFLAGS+=-Wl,-z,noexecstack
.PHONY: all .PHONY: all
all: one all: one

View File

@ -1,8 +1,8 @@
CC=clang CC=clang
CXX=clang++ CXX=clang++
TOPDIR=$(shell PWD) TOPDIR=$(shell pwd)
INCLUDES=-I$(shell PWD)/rustybits/target -isystem $(TOPDIR)/ext -I$(TOPDIR)/ext/prometheus-cpp-lite-1.0/core/include -I$(TOPDIR)/ext-prometheus-cpp-lite-1.0/3rdparty/http-client-lite/include -I$(TOPDIR)/ext/prometheus-cpp-lite-1.0/simpleapi/include INCLUDES=-I$(shell pwd)/rustybits/target -isystem $(TOPDIR)/ext -I$(TOPDIR)/ext/prometheus-cpp-lite-1.0/core/include -I$(TOPDIR)/ext-prometheus-cpp-lite-1.0/3rdparty/http-client-lite/include -I$(TOPDIR)/ext/prometheus-cpp-lite-1.0/simpleapi/include
DEFS= DEFS=
LIBS= LIBS=
ARCH_FLAGS=-arch x86_64 -arch arm64 ARCH_FLAGS=-arch x86_64 -arch arm64

View File

@ -373,6 +373,7 @@ SharedPtr<Path> Bond::getAppropriatePath(int64_t now, int32_t flowId)
*/ */
if (_policy == ZT_BOND_POLICY_ACTIVE_BACKUP) { if (_policy == ZT_BOND_POLICY_ACTIVE_BACKUP) {
if (_abPathIdx != ZT_MAX_PEER_NETWORK_PATHS && _paths[_abPathIdx].p) { if (_abPathIdx != ZT_MAX_PEER_NETWORK_PATHS && _paths[_abPathIdx].p) {
//fprintf(stderr, "trying to send via (_abPathIdx=%d) %s\n", _abPathIdx, pathToStr(_paths[_abPathIdx].p).c_str());
return _paths[_abPathIdx].p; return _paths[_abPathIdx].p;
} }
} }
@ -1032,6 +1033,13 @@ void Bond::curateBond(int64_t now, bool rebuildBond)
bool satisfiedUpDelay = (now - _paths[i].lastAliveToggle) >= _upDelay; bool satisfiedUpDelay = (now - _paths[i].lastAliveToggle) >= _upDelay;
// How long since the last QoS was received (Must be less than ZT_PEER_PATH_EXPIRATION since the remote peer's _qosSendInterval isn't known) // How long since the last QoS was received (Must be less than ZT_PEER_PATH_EXPIRATION since the remote peer's _qosSendInterval isn't known)
bool acceptableQoSAge = (_paths[i].lastQoSReceived == 0 && inTrial) || ((now - _paths[i].lastQoSReceived) < ZT_PEER_EXPIRED_PATH_TRIAL_PERIOD); bool acceptableQoSAge = (_paths[i].lastQoSReceived == 0 && inTrial) || ((now - _paths[i].lastQoSReceived) < ZT_PEER_EXPIRED_PATH_TRIAL_PERIOD);
// Allow active-backup to operate without the receipt of QoS records
// This may be expanded to the other modes as an option
if (_policy == ZT_BOND_POLICY_ACTIVE_BACKUP) {
acceptableQoSAge = true;
}
currEligibility = _paths[i].allowed() && ((acceptableAge && satisfiedUpDelay && acceptableQoSAge) || inTrial); currEligibility = _paths[i].allowed() && ((acceptableAge && satisfiedUpDelay && acceptableQoSAge) || inTrial);
if (currEligibility) { if (currEligibility) {
@ -1043,12 +1051,11 @@ void Bond::curateBond(int64_t now, bool rebuildBond)
*/ */
if (currEligibility != _paths[i].eligible) { if (currEligibility != _paths[i].eligible) {
if (currEligibility == 0) { if (currEligibility == 0) {
log("link %s is no longer eligible", pathToStr(_paths[i].p).c_str()); log("link %s is no longer eligible (reason: allowed=%d, age=%d, ud=%d, qos=%d, trial=%d)", pathToStr(_paths[i].p).c_str(), _paths[i].allowed(), acceptableAge, satisfiedUpDelay, acceptableQoSAge, inTrial);
} }
if (currEligibility == 1) { if (currEligibility == 1) {
log("link %s is eligible", pathToStr(_paths[i].p).c_str()); log("link %s is eligible", pathToStr(_paths[i].p).c_str());
} }
debug("\t[%d] allowed=%d, age=%d, qa=%d, ud=%d, trial=%d", i, _paths[i].allowed(), acceptableAge, acceptableQoSAge, satisfiedUpDelay, inTrial);
dumpPathStatus(now, i); dumpPathStatus(now, i);
if (currEligibility) { if (currEligibility) {
rebuildBond = true; rebuildBond = true;
@ -1496,7 +1503,8 @@ void Bond::processActiveBackupTasks(void* tPtr, int64_t now)
{ {
int prevActiveBackupPathIdx = _abPathIdx; int prevActiveBackupPathIdx = _abPathIdx;
int nonPreferredPathIdx = ZT_MAX_PEER_NETWORK_PATHS; int nonPreferredPathIdx = ZT_MAX_PEER_NETWORK_PATHS;
bool bFoundPrimaryLink = false; bool foundPathOnPrimaryLink = false;
bool foundPreferredPath = false;
if (_abPathIdx != ZT_MAX_PEER_NETWORK_PATHS && ! _paths[_abPathIdx].p) { if (_abPathIdx != ZT_MAX_PEER_NETWORK_PATHS && ! _paths[_abPathIdx].p) {
_abPathIdx = ZT_MAX_PEER_NETWORK_PATHS; _abPathIdx = ZT_MAX_PEER_NETWORK_PATHS;
@ -1559,15 +1567,16 @@ void Bond::processActiveBackupTasks(void* tPtr, int64_t now)
if (! _paths[i].preferred()) { if (! _paths[i].preferred()) {
// Found path on primary link, take note in case we don't find a preferred path // Found path on primary link, take note in case we don't find a preferred path
nonPreferredPathIdx = i; nonPreferredPathIdx = i;
bFoundPrimaryLink = true; foundPathOnPrimaryLink = true;
} }
if (_paths[i].preferred()) { if (_paths[i].preferred()) {
_abPathIdx = i; _abPathIdx = i;
bFoundPrimaryLink = true; foundPathOnPrimaryLink = true;
if (_paths[_abPathIdx].p) { if (_paths[_abPathIdx].p) {
SharedPtr<Link> abLink = RR->bc->getLinkBySocket(_policyAlias, _paths[_abPathIdx].p->localSocket()); SharedPtr<Link> abLink = RR->bc->getLinkBySocket(_policyAlias, _paths[_abPathIdx].p->localSocket());
if (abLink) { if (abLink) {
log("found preferred primary link %s", pathToStr(_paths[_abPathIdx].p).c_str()); log("found preferred primary link (_abPathIdx=%d), %s", _abPathIdx, pathToStr(_paths[_abPathIdx].p).c_str());
foundPreferredPath = true;
} }
break; // Found preferred path on primary link break; // Found preferred path on primary link
} }
@ -1575,8 +1584,8 @@ void Bond::processActiveBackupTasks(void* tPtr, int64_t now)
} }
} }
} }
if (bFoundPrimaryLink && (nonPreferredPathIdx != ZT_MAX_PEER_NETWORK_PATHS)) { if (!foundPreferredPath && foundPathOnPrimaryLink && (nonPreferredPathIdx != ZT_MAX_PEER_NETWORK_PATHS)) {
log("found non-preferred primary link"); log("found non-preferred primary link (_abPathIdx=%d)", _abPathIdx);
_abPathIdx = nonPreferredPathIdx; _abPathIdx = nonPreferredPathIdx;
} }
} }
@ -1614,10 +1623,10 @@ void Bond::processActiveBackupTasks(void* tPtr, int64_t now)
} }
if (_paths[(*it)].p && ! _paths[(*it)].eligible) { if (_paths[(*it)].p && ! _paths[(*it)].eligible) {
SharedPtr<Link> link = RR->bc->getLinkBySocket(_policyAlias, _paths[(*it)].p->localSocket()); SharedPtr<Link> link = RR->bc->getLinkBySocket(_policyAlias, _paths[(*it)].p->localSocket());
it = _abFailoverQueue.erase(it);
if (link) { if (link) {
log("link %s is ineligible, removing from failover queue (%zu links remain in queue)", pathToStr(_paths[_abPathIdx].p).c_str(), _abFailoverQueue.size()); log("link %s is ineligible, removing from failover queue (%zu links remain in queue)", pathToStr(_paths[(*it)].p).c_str(), _abFailoverQueue.size());
} }
it = _abFailoverQueue.erase(it);
continue; continue;
} }
else { else {
@ -1684,7 +1693,7 @@ void Bond::processActiveBackupTasks(void* tPtr, int64_t now)
} }
} }
if (! bFoundPathInQueue) { if (! bFoundPathInQueue) {
_abFailoverQueue.push_front(i); _abFailoverQueue.push_back(i);
log("add link %s to failover queue (%zu links in queue)", pathToStr(_paths[i].p).c_str(), _abFailoverQueue.size()); log("add link %s to failover queue (%zu links in queue)", pathToStr(_paths[i].p).c_str(), _abFailoverQueue.size());
addPathToBond(i, 0); addPathToBond(i, 0);
} }
@ -1734,13 +1743,14 @@ void Bond::processActiveBackupTasks(void* tPtr, int64_t now)
} }
} }
if (! bFoundPathInQueue) { if (! bFoundPathInQueue) {
_abFailoverQueue.push_front(i); _abFailoverQueue.push_back(i);
log("add link %s to failover queue (%zu links in queue)", pathToStr(_paths[i].p).c_str(), _abFailoverQueue.size()); log("add link %s to failover queue (%zu links in queue)", pathToStr(_paths[i].p).c_str(), _abFailoverQueue.size());
addPathToBond(i, 0); addPathToBond(i, 0);
} }
} }
} }
} }
/*
// Sort queue based on performance // Sort queue based on performance
if (! _abFailoverQueue.empty()) { if (! _abFailoverQueue.empty()) {
for (int i = 0; i < _abFailoverQueue.size(); i++) { for (int i = 0; i < _abFailoverQueue.size(); i++) {
@ -1752,7 +1762,7 @@ void Bond::processActiveBackupTasks(void* tPtr, int64_t now)
} }
_abFailoverQueue[hole_position] = value_to_insert; _abFailoverQueue[hole_position] = value_to_insert;
} }
} }*/
/** /**
* Short-circuit if we have no queued paths * Short-circuit if we have no queued paths
@ -1902,7 +1912,7 @@ void Bond::setBondParameters(int policy, SharedPtr<Bond> templateBond, bool useT
* Policy defaults * Policy defaults
*/ */
_abPathIdx = ZT_MAX_PEER_NETWORK_PATHS; _abPathIdx = ZT_MAX_PEER_NETWORK_PATHS;
_abLinkSelectMethod = ZT_BOND_RESELECTION_POLICY_OPTIMIZE; _abLinkSelectMethod = ZT_BOND_RESELECTION_POLICY_ALWAYS;
_rrPacketsSentOnCurrLink = 0; _rrPacketsSentOnCurrLink = 0;
_rrIdx = 0; _rrIdx = 0;
_packetsPerLink = 64; _packetsPerLink = 64;
@ -2021,7 +2031,8 @@ void Bond::dumpInfo(int64_t now, bool force)
_lastSummaryDump = now; _lastSummaryDump = now;
float overhead = (_overheadBytes / (timeSinceLastDump / 1000.0f) / 1000.0f); float overhead = (_overheadBytes / (timeSinceLastDump / 1000.0f) / 1000.0f);
_overheadBytes = 0; _overheadBytes = 0;
log("bond: bp=%d, fi=%" PRIu64 ", mi=%d, ud=%d, dd=%d, flows=%zu, leaf=%d, overhead=%f KB/s, links=(%d/%d)", log("bond: ready=%d, bp=%d, fi=%" PRIu64 ", mi=%d, ud=%d, dd=%d, flows=%zu, leaf=%d, overhead=%f KB/s, links=(%d/%d)",
isReady(),
_policy, _policy,
_failoverInterval, _failoverInterval,
_monitorInterval, _monitorInterval,

View File

@ -1144,6 +1144,7 @@ class Bond {
__attribute__((format(printf, 2, 3))) __attribute__((format(printf, 2, 3)))
#endif #endif
{ {
//if (_peerId != 0x0 && _peerId != 0x0) { return; }
#ifdef ZT_TRACE #ifdef ZT_TRACE
time_t rawtime; time_t rawtime;
struct tm* timeinfo; struct tm* timeinfo;
@ -1175,6 +1176,7 @@ class Bond {
__attribute__((format(printf, 2, 3))) __attribute__((format(printf, 2, 3)))
#endif #endif
{ {
//if (_peerId != 0x0 && _peerId != 0x0) { return; }
#ifdef ZT_DEBUG #ifdef ZT_DEBUG
time_t rawtime; time_t rawtime;
struct tm* timeinfo; struct tm* timeinfo;

View File

@ -202,6 +202,72 @@
#define ZT_PACKED_STRUCT(D) D __attribute__((packed)) #define ZT_PACKED_STRUCT(D) D __attribute__((packed))
#endif #endif
#if defined(_WIN32)
#define ZT_PLATFORM_NAME "windows" // Windows
#elif defined(_WIN64)
#define ZT_PLATFORM_NAME "windows" // Windows
#elif defined(__CYGWIN__)
#define ZT_PLATFORM_NAME "windows" // Windows (Cygwin POSIX under Microsoft Window)
#elif defined(__ANDROID__)
#define ZT_PLATFORM_NAME "android" // Android (implies Linux, so it must come first)
#elif defined(__linux__)
#define ZT_PLATFORM_NAME "linux" // Debian, Ubuntu, Gentoo, Fedora, openSUSE, RedHat, Centos and other
#elif defined(__unix__) || !defined(__APPLE__) && defined(__MACH__)
#include <sys/param.h>
#if defined(BSD)
#define ZT_PLATFORM_NAME "bsd" // FreeBSD, NetBSD, OpenBSD, DragonFly BSD
#endif
#elif defined(__hpux)
#define ZT_PLATFORM_NAME "hp-ux" // HP-UX
#elif defined(_AIX)
#define ZT_PLATFORM_NAME "aix" // IBM AIX
#elif defined(__APPLE__) && defined(__MACH__) // Apple OSX and iOS (Darwin)
#include <TargetConditionals.h>
#if defined(TARGET_IPHONE_SIMULATOR) && TARGET_IPHONE_SIMULATOR == 1
#define ZT_PLATFORM_NAME "ios_sim" // Apple iOS
#elif defined(TARGET_OS_IPAD) && TARGET_OS_IPAD == 1
#define ZT_PLATFORM_NAME "ios_ipad"
#elif defined(TARGET_OS_IPHONE) && TARGET_OS_IPHONE == 1
#define ZT_PLATFORM_NAME "ios_iphone" // Apple iOS
#elif defined(TARGET_OS_MAC) && TARGET_OS_MAC == 1
#define ZT_PLATFORM_NAME "macos" // Apple OSX
#endif
#elif defined(__sun) && defined(__SVR4)
#define ZT_PLATFORM_NAME "solaris" // Oracle Solaris, Open Indiana
#else
#define ZT_PLATFORM_NAME "unknown"
#endif
#ifndef ZT_PLATFORM_NAME
#define ZT_PLATFORM_NAME "unknown"
#endif
#if defined(__amd64) || defined(__amd64__) || defined(__x86_64) || defined(__x86_64__) || defined(__AMD64) || defined(__AMD64__) || defined(_M_X64) || defined(_M_AMD64)
#define ZT_ARCH_NAME "x86_64"
#elif defined(__i386__) || defined(__i486__) || defined(__i586__) || defined(__i686__) || defined(_X86_) || defined(_M_IX86) || defined(__X86__) || defined(__I86__) || defined(_M_I86)
#define ZT_ARCH_NAME "x86"
#elif defined(__aarch64__) || defined(__AARCH64EL__) || defined(_M_ARM64)
#define ZT_ARCH_NAME "arm64"
#elif defined(__arm__) || defined(__TARGET_ARCH_ARM) || defined(_ARM) || defined(_M_ARM) || defined(_M_ARMT) || defined(__arm) || defined(__thumb__)
#define ZT_ARCH_NAME "arm"
#elif defined(__loongarch__) || defined(_LOONGARCH_ARCH)
#define ZT_ARCH_NAME "loongarch"
#elif defined(__mips__) || defined(__MIPS__)
#define ZT_ARCH_NAME "mips"
#elif defined(__riscv) || defined(__riscv_xlen)
#define ZT_ARCH_NAME "riscv"
#elif defined(__powerpc__) || defined(__powerpc64__) || defined(__ppc__) || defined(__ppc64__) || defined (_M_PPC)
#define ZT_ARCH_NAME "powerpc"
#elif defined(__s390__) || defined(__s390x__) || defined(__zarch__)
#define ZT_ARCH_NAME "s390"
#else
#define ZT_ARCH_NAME "unknown"
#endif
#ifndef ZT_ARCH_NAME
#define ZT_ARCH_NAME "unknown"
#endif
#define ZT_TARGET_NAME (ZT_PLATFORM_NAME "/" ZT_ARCH_NAME)
/** /**
* Length of a ZeroTier address in bytes * Length of a ZeroTier address in bytes
*/ */

View File

@ -38,6 +38,7 @@
#include "Path.hpp" #include "Path.hpp"
#include "Bond.hpp" #include "Bond.hpp"
#include "Metrics.hpp" #include "Metrics.hpp"
#include "PacketMultiplexer.hpp"
namespace ZeroTier { namespace ZeroTier {
@ -334,7 +335,6 @@ bool IncomingPacket::_doACK(const RuntimeEnvironment* RR, void* tPtr, const Shar
bool IncomingPacket::_doQOS_MEASUREMENT(const RuntimeEnvironment* RR, void* tPtr, const SharedPtr<Peer>& peer) bool IncomingPacket::_doQOS_MEASUREMENT(const RuntimeEnvironment* RR, void* tPtr, const SharedPtr<Peer>& peer)
{ {
Metrics::pkt_qos_in++; Metrics::pkt_qos_in++;
SharedPtr<Bond> bond = peer->bond();
if (! peer->rateGateQoS(RR->node->now(), _path)) { if (! peer->rateGateQoS(RR->node->now(), _path)) {
return true; return true;
} }
@ -793,7 +793,7 @@ bool IncomingPacket::_doFRAME(const RuntimeEnvironment *RR,void *tPtr,const Shar
{ {
Metrics::pkt_frame_in++; Metrics::pkt_frame_in++;
int32_t _flowId = ZT_QOS_NO_FLOW; int32_t _flowId = ZT_QOS_NO_FLOW;
if (peer->flowHashingSupported()) {
if (size() > ZT_PROTO_VERB_EXT_FRAME_IDX_PAYLOAD) { if (size() > ZT_PROTO_VERB_EXT_FRAME_IDX_PAYLOAD) {
const unsigned int etherType = at<uint16_t>(ZT_PROTO_VERB_FRAME_IDX_ETHERTYPE); const unsigned int etherType = at<uint16_t>(ZT_PROTO_VERB_FRAME_IDX_ETHERTYPE);
const unsigned int frameLen = size() - ZT_PROTO_VERB_FRAME_IDX_PAYLOAD; const unsigned int frameLen = size() - ZT_PROTO_VERB_FRAME_IDX_PAYLOAD;
@ -855,7 +855,6 @@ bool IncomingPacket::_doFRAME(const RuntimeEnvironment *RR,void *tPtr,const Shar
} }
} }
} }
}
const uint64_t nwid = at<uint64_t>(ZT_PROTO_VERB_FRAME_IDX_NETWORK_ID); const uint64_t nwid = at<uint64_t>(ZT_PROTO_VERB_FRAME_IDX_NETWORK_ID);
const SharedPtr<Network> network(RR->node->network(nwid)); const SharedPtr<Network> network(RR->node->network(nwid));
@ -869,7 +868,7 @@ bool IncomingPacket::_doFRAME(const RuntimeEnvironment *RR,void *tPtr,const Shar
const unsigned int frameLen = size() - ZT_PROTO_VERB_FRAME_IDX_PAYLOAD; const unsigned int frameLen = size() - ZT_PROTO_VERB_FRAME_IDX_PAYLOAD;
const uint8_t *const frameData = reinterpret_cast<const uint8_t *>(data()) + ZT_PROTO_VERB_FRAME_IDX_PAYLOAD; const uint8_t *const frameData = reinterpret_cast<const uint8_t *>(data()) + ZT_PROTO_VERB_FRAME_IDX_PAYLOAD;
if (network->filterIncomingPacket(tPtr,peer,RR->identity.address(),sourceMac,network->mac(),frameData,frameLen,etherType,0) > 0) { if (network->filterIncomingPacket(tPtr,peer,RR->identity.address(),sourceMac,network->mac(),frameData,frameLen,etherType,0) > 0) {
RR->node->putFrame(tPtr,nwid,network->userPtr(),sourceMac,network->mac(),etherType,0,(const void *)frameData,frameLen); RR->pm->putFrame(tPtr,nwid,network->userPtr(),sourceMac,network->mac(),etherType,0,(const void *)frameData,frameLen, _flowId);
} }
} }
} else { } else {
@ -942,7 +941,7 @@ bool IncomingPacket::_doEXT_FRAME(const RuntimeEnvironment *RR,void *tPtr,const
} }
// fall through -- 2 means accept regardless of bridging checks or other restrictions // fall through -- 2 means accept regardless of bridging checks or other restrictions
case 2: case 2:
RR->node->putFrame(tPtr,nwid,network->userPtr(),from,to,etherType,0,(const void *)frameData,frameLen); RR->pm->putFrame(tPtr,nwid,network->userPtr(),from,to,etherType,0,(const void *)frameData,frameLen, flowId);
break; break;
} }
} }

View File

@ -1313,6 +1313,7 @@ void Network::requestConfiguration(void *tPtr)
rmd.add(ZT_NETWORKCONFIG_REQUEST_METADATA_KEY_MAX_NETWORK_TAGS,(uint64_t)ZT_MAX_NETWORK_TAGS); rmd.add(ZT_NETWORKCONFIG_REQUEST_METADATA_KEY_MAX_NETWORK_TAGS,(uint64_t)ZT_MAX_NETWORK_TAGS);
rmd.add(ZT_NETWORKCONFIG_REQUEST_METADATA_KEY_FLAGS,(uint64_t)0); rmd.add(ZT_NETWORKCONFIG_REQUEST_METADATA_KEY_FLAGS,(uint64_t)0);
rmd.add(ZT_NETWORKCONFIG_REQUEST_METADATA_KEY_RULES_ENGINE_REV,(uint64_t)ZT_RULES_ENGINE_REVISION); rmd.add(ZT_NETWORKCONFIG_REQUEST_METADATA_KEY_RULES_ENGINE_REV,(uint64_t)ZT_RULES_ENGINE_REVISION);
rmd.add(ZT_NETWORKCONFIG_REQUEST_METADATA_KEY_OS_ARCH,ZT_TARGET_NAME);
RR->t->networkConfigRequestSent(tPtr,*this,ctrl); RR->t->networkConfigRequestSent(tPtr,*this,ctrl);

View File

@ -105,6 +105,8 @@ namespace ZeroTier {
// Network config version // Network config version
#define ZT_NETWORKCONFIG_REQUEST_METADATA_KEY_VERSION "v" #define ZT_NETWORKCONFIG_REQUEST_METADATA_KEY_VERSION "v"
// Network config version
#define ZT_NETWORKCONFIG_REQUEST_METADATA_KEY_OS_ARCH "o"
// Protocol version (see Packet.hpp) // Protocol version (see Packet.hpp)
#define ZT_NETWORKCONFIG_REQUEST_METADATA_KEY_PROTOCOL_VERSION "pv" #define ZT_NETWORKCONFIG_REQUEST_METADATA_KEY_PROTOCOL_VERSION "pv"
// Software vendor // Software vendor

View File

@ -35,6 +35,7 @@
#include "Network.hpp" #include "Network.hpp"
#include "Trace.hpp" #include "Trace.hpp"
#include "Metrics.hpp" #include "Metrics.hpp"
#include "PacketMultiplexer.hpp"
// FIXME: remove this suppression and actually fix warnings // FIXME: remove this suppression and actually fix warnings
#ifdef __GNUC__ #ifdef __GNUC__
@ -119,9 +120,10 @@ Node::Node(void *uptr,void *tptr,const struct ZT_Node_Callbacks *callbacks,int64
const unsigned long mcs = sizeof(Multicaster) + (((sizeof(Multicaster) & 0xf) != 0) ? (16 - (sizeof(Multicaster) & 0xf)) : 0); const unsigned long mcs = sizeof(Multicaster) + (((sizeof(Multicaster) & 0xf) != 0) ? (16 - (sizeof(Multicaster) & 0xf)) : 0);
const unsigned long topologys = sizeof(Topology) + (((sizeof(Topology) & 0xf) != 0) ? (16 - (sizeof(Topology) & 0xf)) : 0); const unsigned long topologys = sizeof(Topology) + (((sizeof(Topology) & 0xf) != 0) ? (16 - (sizeof(Topology) & 0xf)) : 0);
const unsigned long sas = sizeof(SelfAwareness) + (((sizeof(SelfAwareness) & 0xf) != 0) ? (16 - (sizeof(SelfAwareness) & 0xf)) : 0); const unsigned long sas = sizeof(SelfAwareness) + (((sizeof(SelfAwareness) & 0xf) != 0) ? (16 - (sizeof(SelfAwareness) & 0xf)) : 0);
const unsigned long bc = sizeof(Bond) + (((sizeof(Bond) & 0xf) != 0) ? (16 - (sizeof(Bond) & 0xf)) : 0); const unsigned long bcs = sizeof(Bond) + (((sizeof(Bond) & 0xf) != 0) ? (16 - (sizeof(Bond) & 0xf)) : 0);
const unsigned long pms = sizeof(PacketMultiplexer) + (((sizeof(PacketMultiplexer) & 0xf) != 0) ? (16 - (sizeof(PacketMultiplexer) & 0xf)) : 0);
m = reinterpret_cast<char *>(::malloc(16 + ts + sws + mcs + topologys + sas + bc)); m = reinterpret_cast<char *>(::malloc(16 + ts + sws + mcs + topologys + sas + bcs + pms));
if (!m) { if (!m) {
throw std::bad_alloc(); throw std::bad_alloc();
} }
@ -141,6 +143,8 @@ Node::Node(void *uptr,void *tptr,const struct ZT_Node_Callbacks *callbacks,int64
RR->sa = new (m) SelfAwareness(RR); RR->sa = new (m) SelfAwareness(RR);
m += sas; m += sas;
RR->bc = new (m) Bond(RR); RR->bc = new (m) Bond(RR);
m += bcs;
RR->pm = new (m) PacketMultiplexer(RR);
} catch ( ... ) { } catch ( ... ) {
if (RR->sa) { if (RR->sa) {
RR->sa->~SelfAwareness(); RR->sa->~SelfAwareness();
@ -160,6 +164,9 @@ Node::Node(void *uptr,void *tptr,const struct ZT_Node_Callbacks *callbacks,int64
if (RR->bc) { if (RR->bc) {
RR->bc->~Bond(); RR->bc->~Bond();
} }
if (RR->pm) {
RR->pm->~PacketMultiplexer();
}
::free(m); ::free(m);
throw; throw;
} }
@ -191,6 +198,9 @@ Node::~Node()
if (RR->bc) { if (RR->bc) {
RR->bc->~Bond(); RR->bc->~Bond();
} }
if (RR->pm) {
RR->pm->~PacketMultiplexer();
}
::free(RR->rtmem); ::free(RR->rtmem);
} }
@ -230,6 +240,11 @@ ZT_ResultCode Node::processVirtualNetworkFrame(
} }
} }
void Node::initMultithreading(unsigned int concurrency, bool cpuPinningEnabled)
{
RR->pm->setUpPostDecodeReceiveThreads(concurrency, cpuPinningEnabled);
}
// Closure used to ping upstream and active/online peers // Closure used to ping upstream and active/online peers
class _PingPeersThatNeedPing class _PingPeersThatNeedPing
{ {

View File

@ -283,7 +283,10 @@ public:
return _lowBandwidthMode; return _lowBandwidthMode;
} }
private: void initMultithreading(unsigned int concurrency, bool cpuPinningEnabled);
public:
RuntimeEnvironment _RR; RuntimeEnvironment _RR;
RuntimeEnvironment *RR; RuntimeEnvironment *RR;
void *_uPtr; // _uptr (lower case) is reserved in Visual Studio :P void *_uPtr; // _uptr (lower case) is reserved in Visual Studio :P

122
node/PacketMultiplexer.cpp Normal file
View File

@ -0,0 +1,122 @@
/*
* Copyright (c)2013-2021 ZeroTier, Inc.
*
* Use of this software is governed by the Business Source License included
* in the LICENSE.TXT file in the project's root directory.
*
* Change Date: 2026-01-01
*
* On the date above, in accordance with the Business Source License, use
* of this software will be governed by version 2.0 of the Apache License.
*/
/****/
#include "PacketMultiplexer.hpp"
#include "Node.hpp"
#include "RuntimeEnvironment.hpp"
#include "Constants.hpp"
#include <stdio.h>
#include <stdlib.h>
namespace ZeroTier {
PacketMultiplexer::PacketMultiplexer(const RuntimeEnvironment* renv)
{
RR = renv;
};
void PacketMultiplexer::putFrame(void* tPtr, uint64_t nwid, void** nuptr, const MAC& source, const MAC& dest, unsigned int etherType, unsigned int vlanId, const void* data, unsigned int len, unsigned int flowId)
{
#if defined(__APPLE__) || defined(__OpenBSD__) || defined(__NetBSD__) || defined(__WINDOWS__)
RR->node->putFrame(tPtr,nwid,nuptr,source,dest,etherType,vlanId,(const void *)data,len);
return;
#endif
if (!_enabled) {
RR->node->putFrame(tPtr,nwid,nuptr,source,dest,etherType,vlanId,(const void *)data,len);
return;
}
PacketRecord* packet;
_rxPacketVector_m.lock();
if (_rxPacketVector.empty()) {
packet = new PacketRecord;
}
else {
packet = _rxPacketVector.back();
_rxPacketVector.pop_back();
}
_rxPacketVector_m.unlock();
packet->tPtr = tPtr;
packet->nwid = nwid;
packet->nuptr = nuptr;
packet->source = source.toInt();
packet->dest = dest.toInt();
packet->etherType = etherType;
packet->vlanId = vlanId;
packet->len = len;
packet->flowId = flowId;
memcpy(packet->data, data, len);
int bucket = flowId % _concurrency;
_rxPacketQueues[bucket]->postLimit(packet, 2048);
}
void PacketMultiplexer::setUpPostDecodeReceiveThreads(unsigned int concurrency, bool cpuPinningEnabled)
{
#if defined(__APPLE__) || defined(__OpenBSD__) || defined(__NetBSD__) || defined(__WINDOWS__)
return;
#endif
_enabled = true;
_concurrency = concurrency;
bool _enablePinning = cpuPinningEnabled;
for (unsigned int i = 0; i < _concurrency; ++i) {
fprintf(stderr, "Reserved queue for thread %d\n", i);
_rxPacketQueues.push_back(new BlockingQueue<PacketRecord*>());
}
// Each thread picks from its own queue to feed into the core
for (unsigned int i = 0; i < _concurrency; ++i) {
_rxThreads.push_back(std::thread([this, i, _enablePinning]() {
fprintf(stderr, "Created post-decode packet ingestion thread %d\n", i);
PacketRecord* packet = nullptr;
for (;;) {
if (! _rxPacketQueues[i]->get(packet)) {
break;
}
if (! packet) {
break;
}
// fprintf(stderr, "popped packet from queue %d\n", i);
MAC sourceMac = MAC(packet->source);
MAC destMac = MAC(packet->dest);
RR->node->putFrame(packet->tPtr, packet->nwid, packet->nuptr, sourceMac, destMac, packet->etherType, 0, (const void*)packet->data, packet->len);
{
Mutex::Lock l(_rxPacketVector_m);
_rxPacketVector.push_back(packet);
}
/*
if (ZT_ResultCode_isFatal(err)) {
char tmp[256];
OSUtils::ztsnprintf(tmp, sizeof(tmp), "error processing packet: %d", (int)err);
Mutex::Lock _l(_termReason_m);
_termReason = ONE_UNRECOVERABLE_ERROR;
_fatalErrorMessage = tmp;
this->terminate();
break;
}
*/
}
}));
}
}
} // namespace ZeroTier

View File

@ -0,0 +1,65 @@
/*
* Copyright (c)2013-2021 ZeroTier, Inc.
*
* Use of this software is governed by the Business Source License included
* in the LICENSE.TXT file in the project's root directory.
*
* Change Date: 2026-01-01
*
* On the date above, in accordance with the Business Source License, use
* of this software will be governed by version 2.0 of the Apache License.
*/
/****/
#ifndef ZT_PACKET_MULTIPLEXER_HPP
#define ZT_PACKET_MULTIPLEXER_HPP
#include "../osdep/BlockingQueue.hpp"
#include "MAC.hpp"
#include "Mutex.hpp"
#include "RuntimeEnvironment.hpp"
#include <thread>
#include <vector>
namespace ZeroTier {
struct PacketRecord {
void* tPtr;
uint64_t nwid;
void** nuptr;
uint64_t source;
uint64_t dest;
unsigned int etherType;
unsigned int vlanId;
uint8_t data[ZT_MAX_MTU];
unsigned int len;
unsigned int flowId;
};
class PacketMultiplexer {
public:
const RuntimeEnvironment* RR;
PacketMultiplexer(const RuntimeEnvironment* renv);
void setUpPostDecodeReceiveThreads(unsigned int concurrency, bool cpuPinningEnabled);
void putFrame(void* tPtr, uint64_t nwid, void** nuptr, const MAC& source, const MAC& dest, unsigned int etherType, unsigned int vlanId, const void* data, unsigned int len, unsigned int flowId);
std::vector<BlockingQueue<PacketRecord*>*> _rxPacketQueues;
unsigned int _concurrency;
// pool
std::vector<PacketRecord*> _rxPacketVector;
std::vector<std::thread> _rxPacketThreads;
Mutex _rxPacketVector_m, _rxPacketThreads_m;
std::vector<std::thread> _rxThreads;
unsigned int _rxThreadCount;
bool _enabled;
};
} // namespace ZeroTier
#endif // ZT_PACKET_MULTIPLEXER_HPP

View File

@ -31,6 +31,7 @@ class NetworkController;
class SelfAwareness; class SelfAwareness;
class Trace; class Trace;
class Bond; class Bond;
class PacketMultiplexer;
/** /**
* Holds global state for an instance of ZeroTier::Node * Holds global state for an instance of ZeroTier::Node
@ -77,6 +78,7 @@ public:
Topology *topology; Topology *topology;
SelfAwareness *sa; SelfAwareness *sa;
Bond *bc; Bond *bc;
PacketMultiplexer *pm;
// This node's identity and string representations thereof // This node's identity and string representations thereof
Identity identity; Identity identity;

View File

@ -24,6 +24,7 @@
#include <stdexcept> #include <stdexcept>
#include <vector> #include <vector>
#include <map> #include <map>
#include <algorithm>
#if defined(__FreeBSD__) #if defined(__FreeBSD__)
#include <sys/endian.h> #include <sys/endian.h>
@ -849,6 +850,19 @@ public:
* Hexadecimal characters 0-f * Hexadecimal characters 0-f
*/ */
static const char HEXCHARS[16]; static const char HEXCHARS[16];
/*
* Remove `-` and `:` from a MAC address (in-place).
*
* @param mac The MAC address
*/
static inline void cleanMac(std::string& mac)
{
auto start = mac.begin();
auto end = mac.end();
auto new_end = std::remove_if(start, end, [](char c) { return c == 45 || c == 58; });
mac.erase(new_end, end);
}
}; };
} // namespace ZeroTier } // namespace ZeroTier

View File

@ -29,7 +29,8 @@ CORE_OBJS=\
node/Topology.o \ node/Topology.o \
node/Trace.o \ node/Trace.o \
node/Utils.o \ node/Utils.o \
node/Bond.o node/Bond.o \
node/PacketMultiplexer.o
ONE_OBJS=\ ONE_OBJS=\
controller/EmbeddedNetworkController.o \ controller/EmbeddedNetworkController.o \

View File

@ -39,7 +39,9 @@
#include <net/if_dl.h> #include <net/if_dl.h>
#include <net/if_media.h> #include <net/if_media.h>
#include <net/route.h> #include <net/route.h>
#include <pthread_np.h>
#include <sched.h>
#include <string> #include <string>
#include <map> #include <map>
#include <set> #include <set>
@ -53,6 +55,7 @@
#include "BSDEthernetTap.hpp" #include "BSDEthernetTap.hpp"
#define ZT_BASE32_CHARS "0123456789abcdefghijklmnopqrstuv" #define ZT_BASE32_CHARS "0123456789abcdefghijklmnopqrstuv"
#define ZT_TAP_BUF_SIZE (1024 * 16)
// ff:ff:ff:ff:ff:ff with no ADI // ff:ff:ff:ff:ff:ff with no ADI
static const ZeroTier::MulticastGroup _blindWildcardMulticastGroup(ZeroTier::MAC(0xff),0); static const ZeroTier::MulticastGroup _blindWildcardMulticastGroup(ZeroTier::MAC(0xff),0);
@ -61,6 +64,8 @@ namespace ZeroTier {
BSDEthernetTap::BSDEthernetTap( BSDEthernetTap::BSDEthernetTap(
const char *homePath, const char *homePath,
unsigned int concurrency,
bool pinning,
const MAC &mac, const MAC &mac,
unsigned int mtu, unsigned int mtu,
unsigned int metric, unsigned int metric,
@ -69,6 +74,8 @@ BSDEthernetTap::BSDEthernetTap(
void (*handler)(void *,void *,uint64_t,const MAC &,const MAC &,unsigned int,unsigned int,const void *,unsigned int), void (*handler)(void *,void *,uint64_t,const MAC &,const MAC &,unsigned int,unsigned int,const void *,unsigned int),
void *arg) : void *arg) :
_handler(handler), _handler(handler),
_concurrency(concurrency),
_pinning(pinning),
_arg(arg), _arg(arg),
_nwid(nwid), _nwid(nwid),
_mtu(mtu), _mtu(mtu),
@ -195,11 +202,9 @@ BSDEthernetTap::BSDEthernetTap(
BSDEthernetTap::~BSDEthernetTap() BSDEthernetTap::~BSDEthernetTap()
{ {
::write(_shutdownSignalPipe[1],"\0",1); // causes thread to exit ::write(_shutdownSignalPipe[1],"\0",1); // causes thread to exit
Thread::join(_thread);
::close(_fd); ::close(_fd);
::close(_shutdownSignalPipe[0]); ::close(_shutdownSignalPipe[0]);
::close(_shutdownSignalPipe[1]); ::close(_shutdownSignalPipe[1]);
long cpid = (long)vfork(); long cpid = (long)vfork();
if (cpid == 0) { if (cpid == 0) {
#ifdef ZT_TRACE #ifdef ZT_TRACE
@ -211,6 +216,10 @@ BSDEthernetTap::~BSDEthernetTap()
int exitcode = -1; int exitcode = -1;
::waitpid(cpid,&exitcode,0); ::waitpid(cpid,&exitcode,0);
} }
Thread::join(_thread);
for (std::thread &t : _rxThreads) {
t.join();
}
} }
void BSDEthernetTap::setEnabled(bool en) void BSDEthernetTap::setEnabled(bool en)
@ -418,20 +427,40 @@ void BSDEthernetTap::setMtu(unsigned int mtu)
void BSDEthernetTap::threadMain() void BSDEthernetTap::threadMain()
throw() throw()
{ {
fd_set readfds,nullfds;
MAC to,from;
int n,nfds,r;
char getBuf[ZT_MAX_MTU + 64];
// Wait for a moment after startup -- wait for Network to finish // Wait for a moment after startup -- wait for Network to finish
// constructing itself. // constructing itself.
Thread::sleep(500); Thread::sleep(500);
for (unsigned int i = 0; i < _concurrency; ++i) {
_rxThreads.push_back(std::thread([this, i, _pinning] {
if (_pinning) {
int pinCore = i % _concurrency;
fprintf(stderr, "Pinning thread %d to core %d\n", i, pinCore);
pthread_t self = pthread_self();
cpu_set_t cpuset;
CPU_ZERO(&cpuset);
CPU_SET(pinCore, &cpuset);
//int rc = sched_setaffinity(self, sizeof(cpu_set_t), &cpuset);
int rc = pthread_setaffinity_np(self, sizeof(cpu_set_t), &cpuset);
if (rc != 0)
{
fprintf(stderr, "Failed to pin thread %d to core %d: %s\n", i, pinCore, strerror(errno));
exit(1);
}
}
uint8_t b[ZT_TAP_BUF_SIZE];
MAC to, from;
fd_set readfds, nullfds;
int n, nfds, r;
FD_ZERO(&readfds); FD_ZERO(&readfds);
FD_ZERO(&nullfds); FD_ZERO(&nullfds);
nfds = (int)std::max(_shutdownSignalPipe[0],_fd) + 1; nfds = (int)std::max(_shutdownSignalPipe[0],_fd) + 1;
r = 0; r = 0;
for(;;) { for(;;) {
FD_SET(_shutdownSignalPipe[0],&readfds); FD_SET(_shutdownSignalPipe[0],&readfds);
FD_SET(_fd,&readfds); FD_SET(_fd,&readfds);
@ -441,7 +470,7 @@ void BSDEthernetTap::threadMain()
break; break;
if (FD_ISSET(_fd,&readfds)) { if (FD_ISSET(_fd,&readfds)) {
n = (int)::read(_fd,getBuf + r,sizeof(getBuf) - r); n = (int)::read(_fd,b + r,sizeof(b) - r);
if (n < 0) { if (n < 0) {
if ((errno != EINTR)&&(errno != ETIMEDOUT)) if ((errno != EINTR)&&(errno != ETIMEDOUT))
break; break;
@ -455,10 +484,10 @@ void BSDEthernetTap::threadMain()
r = _mtu + 14; r = _mtu + 14;
if (_enabled) { if (_enabled) {
to.setTo(getBuf,6); to.setTo(b,6);
from.setTo(getBuf + 6,6); from.setTo(b + 6,6);
unsigned int etherType = ntohs(((const uint16_t *)getBuf)[6]); unsigned int etherType = ntohs(((const uint16_t *)b)[6]);
_handler(_arg,(void *)0,_nwid,from,to,etherType,0,(const void *)(getBuf + 14),r - 14); _handler(_arg,(void *)0,_nwid,from,to,etherType,0,(const void *)(b + 14),r - 14);
} }
r = 0; r = 0;
@ -466,6 +495,8 @@ void BSDEthernetTap::threadMain()
} }
} }
} }
}));
}
} }
} // namespace ZeroTier } // namespace ZeroTier

View File

@ -20,6 +20,7 @@
#include <string> #include <string>
#include <vector> #include <vector>
#include <stdexcept> #include <stdexcept>
#include <thread>
#include "../node/Constants.hpp" #include "../node/Constants.hpp"
#include "../node/MulticastGroup.hpp" #include "../node/MulticastGroup.hpp"
@ -34,6 +35,8 @@ class BSDEthernetTap : public EthernetTap
public: public:
BSDEthernetTap( BSDEthernetTap(
const char *homePath, const char *homePath,
unsigned int concurrency,
bool pinning,
const MAC &mac, const MAC &mac,
unsigned int mtu, unsigned int mtu,
unsigned int metric, unsigned int metric,
@ -62,6 +65,8 @@ public:
private: private:
void (*_handler)(void *,void *,uint64_t,const MAC &,const MAC &,unsigned int,unsigned int,const void *,unsigned int); void (*_handler)(void *,void *,uint64_t,const MAC &,const MAC &,unsigned int,unsigned int,const void *,unsigned int);
void *_arg; void *_arg;
unsigned int _concurrency;
bool _pinning;
uint64_t _nwid; uint64_t _nwid;
Thread _thread; Thread _thread;
std::string _dev; std::string _dev;
@ -73,6 +78,7 @@ private:
volatile bool _enabled; volatile bool _enabled;
mutable std::vector<InetAddress> _ifaddrs; mutable std::vector<InetAddress> _ifaddrs;
mutable uint64_t _lastIfAddrsUpdate; mutable uint64_t _lastIfAddrsUpdate;
std::vector<std::thread> _rxThreads;
}; };
} // namespace ZeroTier } // namespace ZeroTier

View File

@ -57,6 +57,8 @@ namespace ZeroTier {
std::shared_ptr<EthernetTap> EthernetTap::newInstance( std::shared_ptr<EthernetTap> EthernetTap::newInstance(
const char *tapDeviceType, // OS-specific, NULL for default const char *tapDeviceType, // OS-specific, NULL for default
unsigned int concurrency,
bool pinning,
const char *homePath, const char *homePath,
const MAC &mac, const MAC &mac,
unsigned int mtu, unsigned int mtu,
@ -92,7 +94,7 @@ std::shared_ptr<EthernetTap> EthernetTap::newInstance(
#endif // __APPLE__ #endif // __APPLE__
#ifdef __LINUX__ #ifdef __LINUX__
return std::shared_ptr<EthernetTap>(new LinuxEthernetTap(homePath,mac,mtu,metric,nwid,friendlyName,handler,arg)); return std::shared_ptr<EthernetTap>(new LinuxEthernetTap(homePath,concurrency,pinning,mac,mtu,metric,nwid,friendlyName,handler,arg));
#endif // __LINUX__ #endif // __LINUX__
#ifdef __WINDOWS__ #ifdef __WINDOWS__
@ -130,7 +132,7 @@ std::shared_ptr<EthernetTap> EthernetTap::newInstance(
#endif // __WINDOWS__ #endif // __WINDOWS__
#ifdef __FreeBSD__ #ifdef __FreeBSD__
return std::shared_ptr<EthernetTap>(new BSDEthernetTap(homePath,mac,mtu,metric,nwid,friendlyName,handler,arg)); return std::shared_ptr<EthernetTap>(new BSDEthernetTap(homePath,concurrency,pinning,mac,mtu,metric,nwid,friendlyName,handler,arg));
#endif // __FreeBSD__ #endif // __FreeBSD__
#ifdef __NetBSD__ #ifdef __NetBSD__

View File

@ -32,6 +32,8 @@ class EthernetTap
public: public:
static std::shared_ptr<EthernetTap> newInstance( static std::shared_ptr<EthernetTap> newInstance(
const char *tapDeviceType, // OS-specific, NULL for default const char *tapDeviceType, // OS-specific, NULL for default
unsigned int concurrency,
bool pinning,
const char *homePath, const char *homePath,
const MAC &mac, const MAC &mac,
unsigned int mtu, unsigned int mtu,

View File

@ -60,7 +60,7 @@
#define IFNAMSIZ 16 #define IFNAMSIZ 16
#endif #endif
#define ZT_TAP_BUF_SIZE 16384 #define ZT_TAP_BUF_SIZE (1024 * 16)
// ff:ff:ff:ff:ff:ff with no ADI // ff:ff:ff:ff:ff:ff with no ADI
static const ZeroTier::MulticastGroup _blindWildcardMulticastGroup(ZeroTier::MAC(0xff),0); static const ZeroTier::MulticastGroup _blindWildcardMulticastGroup(ZeroTier::MAC(0xff),0);
@ -111,6 +111,8 @@ static void _base32_5_to_8(const uint8_t *in,char *out)
LinuxEthernetTap::LinuxEthernetTap( LinuxEthernetTap::LinuxEthernetTap(
const char *homePath, const char *homePath,
unsigned int concurrency,
bool pinning,
const MAC &mac, const MAC &mac,
unsigned int mtu, unsigned int mtu,
unsigned int metric, unsigned int metric,
@ -220,29 +222,45 @@ LinuxEthernetTap::LinuxEthernetTap(
(void)::pipe(_shutdownSignalPipe); (void)::pipe(_shutdownSignalPipe);
_tapReaderThread = std::thread([this]{ for (unsigned int i = 0; i < concurrency; ++i) {
_rxThreads.push_back(std::thread([this, i, concurrency, pinning] {
if (pinning) {
int pinCore = i % concurrency;
fprintf(stderr, "Pinning tap thread %d to core %d\n", i, pinCore);
pthread_t self = pthread_self();
cpu_set_t cpuset;
CPU_ZERO(&cpuset);
CPU_SET(pinCore, &cpuset);
int rc = pthread_setaffinity_np(self, sizeof(cpu_set_t), &cpuset);
if (rc != 0)
{
fprintf(stderr, "Failed to pin tap thread %d to core %d: %s\n", i, pinCore, strerror(errno));
exit(1);
}
}
uint8_t b[ZT_TAP_BUF_SIZE]; uint8_t b[ZT_TAP_BUF_SIZE];
fd_set readfds,nullfds; fd_set readfds, nullfds;
int n,nfds,r; int n, nfds, r;
std::vector<void *> buffers; if (i == 0) {
struct ifreq ifr; struct ifreq ifr;
memset(&ifr, 0, sizeof(ifr));
strcpy(ifr.ifr_name, _dev.c_str());
memset(&ifr,0,sizeof(ifr)); const int sock = socket(AF_INET, SOCK_DGRAM, 0);
strcpy(ifr.ifr_name,_dev.c_str());
const int sock = socket(AF_INET,SOCK_DGRAM,0);
if (sock <= 0) if (sock <= 0)
return; return;
if (ioctl(sock,SIOCGIFFLAGS,(void *)&ifr) < 0) { if (ioctl(sock, SIOCGIFFLAGS, (void*)&ifr) < 0) {
::close(sock); ::close(sock);
printf("WARNING: ioctl() failed setting up Linux tap device (bring interface up)\n"); printf("WARNING: ioctl() failed setting up Linux tap device (bring interface up)\n");
return; return;
} }
ifr.ifr_ifru.ifru_hwaddr.sa_family = ARPHRD_ETHER; ifr.ifr_ifru.ifru_hwaddr.sa_family = ARPHRD_ETHER;
_mac.copyTo(ifr.ifr_ifru.ifru_hwaddr.sa_data,6); _mac.copyTo(ifr.ifr_ifru.ifru_hwaddr.sa_data, 6);
if (ioctl(sock,SIOCSIFHWADDR,(void *)&ifr) < 0) { if (ioctl(sock, SIOCSIFHWADDR, (void*)&ifr) < 0) {
::close(sock); ::close(sock);
printf("WARNING: ioctl() failed setting up Linux tap device (set MAC)\n"); printf("WARNING: ioctl() failed setting up Linux tap device (set MAC)\n");
return; return;
@ -252,7 +270,7 @@ LinuxEthernetTap::LinuxEthernetTap(
if (isOldLinuxKernel()) { if (isOldLinuxKernel()) {
ifr.ifr_ifru.ifru_mtu = (int)_mtu; ifr.ifr_ifru.ifru_mtu = (int)_mtu;
if (ioctl(sock,SIOCSIFMTU,(void *)&ifr) < 0) { if (ioctl(sock, SIOCSIFMTU, (void*)&ifr) < 0) {
::close(sock); ::close(sock);
printf("WARNING: ioctl() failed setting up Linux tap device (set MTU)\n"); printf("WARNING: ioctl() failed setting up Linux tap device (set MTU)\n");
return; return;
@ -261,10 +279,9 @@ LinuxEthernetTap::LinuxEthernetTap(
usleep(100000); usleep(100000);
} }
ifr.ifr_flags |= IFF_MULTICAST; ifr.ifr_flags |= IFF_MULTICAST;
ifr.ifr_flags |= IFF_UP; ifr.ifr_flags |= IFF_UP;
if (ioctl(sock,SIOCSIFFLAGS,(void *)&ifr) < 0) { if (ioctl(sock, SIOCSIFFLAGS, (void*)&ifr) < 0) {
::close(sock); ::close(sock);
printf("WARNING: ioctl() failed setting up Linux tap device (bring interface up)\n"); printf("WARNING: ioctl() failed setting up Linux tap device (bring interface up)\n");
return; return;
@ -272,46 +289,49 @@ LinuxEthernetTap::LinuxEthernetTap(
usleep(100000); usleep(100000);
if (!isOldLinuxKernel()) { if (! isOldLinuxKernel()) {
ifr.ifr_ifru.ifru_hwaddr.sa_family = ARPHRD_ETHER; ifr.ifr_ifru.ifru_hwaddr.sa_family = ARPHRD_ETHER;
_mac.copyTo(ifr.ifr_ifru.ifru_hwaddr.sa_data,6); _mac.copyTo(ifr.ifr_ifru.ifru_hwaddr.sa_data, 6);
if (ioctl(sock,SIOCSIFHWADDR,(void *)&ifr) < 0) { if (ioctl(sock, SIOCSIFHWADDR, (void*)&ifr) < 0) {
::close(sock); ::close(sock);
printf("WARNING: ioctl() failed setting up Linux tap device (set MAC)\n"); printf("WARNING: ioctl() failed setting up Linux tap device (set MAC)\n");
return; return;
} }
ifr.ifr_ifru.ifru_mtu = (int)_mtu; ifr.ifr_ifru.ifru_mtu = (int)_mtu;
if (ioctl(sock,SIOCSIFMTU,(void *)&ifr) < 0) { if (ioctl(sock, SIOCSIFMTU, (void*)&ifr) < 0) {
::close(sock); ::close(sock);
printf("WARNING: ioctl() failed setting up Linux tap device (set MTU)\n"); printf("WARNING: ioctl() failed setting up Linux tap device (set MTU)\n");
return; return;
} }
} }
fcntl(_fd,F_SETFL,O_NONBLOCK); fcntl(_fd, F_SETFL, O_NONBLOCK);
::close(sock); ::close(sock);
}
if (!_run) if (! _run) {
return; return;
}
FD_ZERO(&readfds); FD_ZERO(&readfds);
FD_ZERO(&nullfds); FD_ZERO(&nullfds);
nfds = (int)std::max(_shutdownSignalPipe[0],_fd) + 1; nfds = (int)std::max(_shutdownSignalPipe[0], _fd) + 1;
r = 0; r = 0;
for(;;) { for (;;) {
FD_SET(_shutdownSignalPipe[0],&readfds); FD_SET(_shutdownSignalPipe[0], &readfds);
FD_SET(_fd,&readfds); FD_SET(_fd, &readfds);
select(nfds,&readfds,&nullfds,&nullfds,(struct timeval *)0); select(nfds, &readfds, &nullfds, &nullfds, (struct timeval*)0);
if (FD_ISSET(_shutdownSignalPipe[0],&readfds)) if (FD_ISSET(_shutdownSignalPipe[0], &readfds)) {
break; break;
}
if (FD_ISSET(_fd,&readfds)) { if (FD_ISSET(_fd, &readfds)) {
for(;;) { // read until there are no more packets, then return to outer select() loop for (;;) {
n = (int)::read(_fd,b + r,ZT_TAP_BUF_SIZE - r); // read until there are no more packets, then return to outer select() loop
n = (int)::read(_fd, b + r, ZT_TAP_BUF_SIZE - r);
if (n > 0) { if (n > 0) {
// Some tap drivers like to send the ethernet frame and the // Some tap drivers like to send the ethernet frame and the
// payload in two chunks, so handle that by accumulating // payload in two chunks, so handle that by accumulating
@ -322,33 +342,35 @@ LinuxEthernetTap::LinuxEthernetTap(
r = _mtu + 14; r = _mtu + 14;
if (_enabled) { if (_enabled) {
//_tapq.post(std::pair<void *,int>(buf,r)); MAC to(b, 6), from(b + 6, 6);
//buf = nullptr; unsigned int etherType = Utils::ntoh(((const uint16_t*)b)[6]);
MAC to(b, 6),from(b + 6, 6); _handler(_arg, nullptr, _nwid, from, to, etherType, 0, (const void*)(b + 14), (unsigned int)(r - 14));
unsigned int etherType = Utils::ntoh(((const uint16_t *)b)[6]);
_handler(_arg, nullptr, _nwid, from, to, etherType, 0, (const void *)(b + 14),(unsigned int)(r - 14));
} }
r = 0; r = 0;
} }
} else { }
else {
r = 0; r = 0;
break; break;
} }
} }
} }
} }
}); }));
}
} }
LinuxEthernetTap::~LinuxEthernetTap() LinuxEthernetTap::~LinuxEthernetTap()
{ {
_run = false; _run = false;
(void)::write(_shutdownSignalPipe[1],"\0",1); (void)::write(_shutdownSignalPipe[1],"\0",1);
_tapReaderThread.join();
::close(_fd); ::close(_fd);
::close(_shutdownSignalPipe[0]); ::close(_shutdownSignalPipe[0]);
::close(_shutdownSignalPipe[1]); ::close(_shutdownSignalPipe[1]);
for (std::thread &t : _rxThreads) {
t.join();
}
} }
void LinuxEthernetTap::setEnabled(bool en) void LinuxEthernetTap::setEnabled(bool en)

View File

@ -26,6 +26,7 @@
#include <mutex> #include <mutex>
#include "../node/MulticastGroup.hpp" #include "../node/MulticastGroup.hpp"
#include "EthernetTap.hpp" #include "EthernetTap.hpp"
#include "BlockingQueue.hpp"
namespace ZeroTier { namespace ZeroTier {
@ -34,6 +35,8 @@ class LinuxEthernetTap : public EthernetTap
public: public:
LinuxEthernetTap( LinuxEthernetTap(
const char *homePath, const char *homePath,
unsigned int concurrency,
bool pinning,
const MAC &mac, const MAC &mac,
unsigned int mtu, unsigned int mtu,
unsigned int metric, unsigned int metric,
@ -57,9 +60,6 @@ public:
virtual void setMtu(unsigned int mtu); virtual void setMtu(unsigned int mtu);
virtual void setDns(const char *domain, const std::vector<InetAddress> &servers) {} virtual void setDns(const char *domain, const std::vector<InetAddress> &servers) {}
private: private:
void (*_handler)(void *,void *,uint64_t,const MAC &,const MAC &,unsigned int,unsigned int,const void *,unsigned int); void (*_handler)(void *,void *,uint64_t,const MAC &,const MAC &,unsigned int,unsigned int,const void *,unsigned int);
void *_arg; void *_arg;
@ -73,9 +73,9 @@ private:
int _shutdownSignalPipe[2]; int _shutdownSignalPipe[2];
std::atomic_bool _enabled; std::atomic_bool _enabled;
std::atomic_bool _run; std::atomic_bool _run;
std::thread _tapReaderThread;
mutable std::vector<InetAddress> _ifaddrs; mutable std::vector<InetAddress> _ifaddrs;
mutable uint64_t _lastIfAddrsUpdate; mutable uint64_t _lastIfAddrsUpdate;
std::vector<std::thread> _rxThreads;
}; };
} // namespace ZeroTier } // namespace ZeroTier

View File

@ -447,7 +447,9 @@ MacKextEthernetTap::~MacKextEthernetTap()
::write(_shutdownSignalPipe[1],"\0",1); // causes thread to exit ::write(_shutdownSignalPipe[1],"\0",1); // causes thread to exit
Thread::join(_thread); Thread::join(_thread);
for (std::thread &t : _rxThreads) {
t.join();
}
::close(_fd); ::close(_fd);
::close(_shutdownSignalPipe[0]); ::close(_shutdownSignalPipe[0]);
::close(_shutdownSignalPipe[1]); ::close(_shutdownSignalPipe[1]);

View File

@ -20,6 +20,7 @@
#include <stdexcept> #include <stdexcept>
#include <string> #include <string>
#include <vector> #include <vector>
#include <thread>
#include "../node/Constants.hpp" #include "../node/Constants.hpp"
#include "../node/MAC.hpp" #include "../node/MAC.hpp"
@ -75,6 +76,7 @@ private:
int _fd; int _fd;
int _shutdownSignalPipe[2]; int _shutdownSignalPipe[2];
volatile bool _enabled; volatile bool _enabled;
std::vector<std::thread> _rxThreads;
}; };
} // namespace ZeroTier } // namespace ZeroTier

View File

@ -1,3 +1,8 @@
## Docker image for Synology's DSM7 ## Docker image for Synology's DSM7
Documentation: [docs.zerotier.com/devices/synology](https://docs.zerotier.com/devices/synology) Documentation: [docs.zerotier.com/devices/synology](https://docs.zerotier.com/devices/synology)
### Build & Push changes to DockerHub
```shell
./build.sh build
```

View File

@ -3,19 +3,17 @@
ZTO_VER=$(git describe --tags $(git rev-list --tags --max-count=1)) ZTO_VER=$(git describe --tags $(git rev-list --tags --max-count=1))
ZTO_COMMIT=$(git rev-parse HEAD) ZTO_COMMIT=$(git rev-parse HEAD)
build() build() {
{
sudo docker build --load --rm -t zerotier-synology . --build-arg ZTO_COMMIT=${ZTO_COMMIT} --build-arg ZTO_VER=${ZTO_VER}
LATEST_DOCKER_IMAGE_HASH=$(sudo docker images -q zerotier-synology)
sudo docker tag ${LATEST_DOCKER_IMAGE_HASH} zerotier/zerotier-synology:${ZTO_VER}
sudo docker tag ${LATEST_DOCKER_IMAGE_HASH} zerotier/zerotier-synology:latest
}
push()
{
sudo docker login --username=${DOCKERHUB_USERNAME} sudo docker login --username=${DOCKERHUB_USERNAME}
sudo docker push zerotier/zerotier-synology:${ZTO_VER}
sudo docker push zerotier/zerotier-synology:latest sudo docker buildx build \
--push \
--platform linux/arm/v7,linux/arm64/v8,linux/amd64 \
--tag zerotier/zerotier-synology:${ZTO_VER} \
--tag zerotier/zerotier-synology:latest \
--build-arg ZTO_COMMIT=${ZTO_COMMIT} \
--build-arg ZTO_VER=${ZTO_VER} \
.
} }
"$@" "$@"

1277
rustybits/Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@ -91,7 +91,7 @@
<NMakeOutput> <NMakeOutput>
</NMakeOutput> </NMakeOutput>
<NMakeCleanCommandLine>cargo clean</NMakeCleanCommandLine> <NMakeCleanCommandLine>cargo clean</NMakeCleanCommandLine>
<NMakeReBuildCommandLine>cargo clean &amp; cargo build --release --target=x86_64-pc-windows-msvc</NMakeReBuildCommandLine> <NMakeReBuildCommandLine>cargo clean &amp; cargo build -p zeroidc --release --target=x86_64-pc-windows-msvc</NMakeReBuildCommandLine>
<NMakePreprocessorDefinitions>NDEBUG;$(NMakePreprocessorDefinitions)</NMakePreprocessorDefinitions> <NMakePreprocessorDefinitions>NDEBUG;$(NMakePreprocessorDefinitions)</NMakePreprocessorDefinitions>
</PropertyGroup> </PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|ARM64'"> <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|ARM64'">

View File

@ -16,7 +16,6 @@
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
#include <stdint.h> #include <stdint.h>
#include <string> #include <string>
#include <map> #include <map>
#include <vector> #include <vector>
@ -26,6 +25,11 @@
#include <mutex> #include <mutex>
#include <condition_variable> #include <condition_variable>
#ifdef __FreeBSD__
#include <sched.h>
#include <pthread_np.h>
#endif
#include "../version.h" #include "../version.h"
#include "../include/ZeroTierOne.h" #include "../include/ZeroTierOne.h"
@ -42,6 +46,7 @@
#include "../node/SHA512.hpp" #include "../node/SHA512.hpp"
#include "../node/Bond.hpp" #include "../node/Bond.hpp"
#include "../node/Peer.hpp" #include "../node/Peer.hpp"
#include "../node/PacketMultiplexer.hpp"
#include "../osdep/Phy.hpp" #include "../osdep/Phy.hpp"
#include "../osdep/OSUtils.hpp" #include "../osdep/OSUtils.hpp"
@ -759,7 +764,7 @@ struct TcpConnection
Mutex writeq_m; Mutex writeq_m;
}; };
struct OneServiceIncomingPacket struct PacketRecord
{ {
uint64_t now; uint64_t now;
int64_t sock; int64_t sock;
@ -793,6 +798,14 @@ public:
bool _serverThreadRunning; bool _serverThreadRunning;
bool _serverThreadRunningV6; bool _serverThreadRunningV6;
BlockingQueue<PacketRecord *> _rxPacketQueue;
std::vector<PacketRecord *> _rxPacketVector;
std::vector<std::thread> _rxPacketThreads;
Mutex _rxPacketVector_m,_rxPacketThreads_m;
bool _multicoreEnabled;
bool _cpuPinningEnabled;
unsigned int _concurrency;
bool _allowTcpFallbackRelay; bool _allowTcpFallbackRelay;
bool _forceTcpRelay; bool _forceTcpRelay;
bool _allowSecondaryPort; bool _allowSecondaryPort;
@ -844,8 +857,6 @@ public:
// Deadline for the next background task service function // Deadline for the next background task service function
volatile int64_t _nextBackgroundTaskDeadline; volatile int64_t _nextBackgroundTaskDeadline;
std::map<uint64_t,NetworkState> _nets; std::map<uint64_t,NetworkState> _nets;
Mutex _nets_m; Mutex _nets_m;
@ -942,6 +953,13 @@ public:
#ifdef __WINDOWS__ #ifdef __WINDOWS__
WinFWHelper::removeICMPRules(); WinFWHelper::removeICMPRules();
#endif #endif
_rxPacketQueue.stop();
_rxPacketThreads_m.lock();
for(auto t=_rxPacketThreads.begin();t!=_rxPacketThreads.end();++t) {
t->join();
}
_rxPacketThreads_m.unlock();
_binder.closeAll(_phy); _binder.closeAll(_phy);
#if ZT_VAULT_SUPPORT #if ZT_VAULT_SUPPORT
@ -956,6 +974,13 @@ public:
if (_serverThreadRunningV6) { if (_serverThreadRunningV6) {
_serverThreadV6.join(); _serverThreadV6.join();
} }
_rxPacketVector_m.lock();
while (!_rxPacketVector.empty()) {
delete _rxPacketVector.back();
_rxPacketVector.pop_back();
}
_rxPacketVector_m.unlock();
#ifdef ZT_USE_MINIUPNPC #ifdef ZT_USE_MINIUPNPC
delete _portMapper; delete _portMapper;
@ -964,6 +989,15 @@ public:
delete _rc; delete _rc;
} }
void setUpMultithreading()
{
#if defined(__APPLE__) || defined(__OpenBSD__) || defined(__NetBSD__) || defined(__WINDOWS__)
return;
#endif
_node->initMultithreading(_concurrency, _cpuPinningEnabled);
bool pinning = _cpuPinningEnabled;
}
virtual ReasonForTermination run() virtual ReasonForTermination run()
{ {
try { try {
@ -1272,6 +1306,9 @@ public:
const unsigned long delay = (dl > now) ? (unsigned long)(dl - now) : 500; const unsigned long delay = (dl > now) ? (unsigned long)(dl - now) : 500;
clockShouldBe = now + (int64_t)delay; clockShouldBe = now + (int64_t)delay;
_phy.poll(delay); _phy.poll(delay);
} }
} catch (std::exception &e) { } catch (std::exception &e) {
Mutex::Lock _l(_termReason_m); Mutex::Lock _l(_termReason_m);
@ -2510,7 +2547,7 @@ public:
} }
_node->bondController()->addCustomLink(customPolicyStr, new Link(linkNameStr,ipvPref,mtu,capacity,enabled,linkMode,failoverToStr)); _node->bondController()->addCustomLink(customPolicyStr, new Link(linkNameStr,ipvPref,mtu,capacity,enabled,linkMode,failoverToStr));
} }
std::string linkSelectMethodStr(OSUtils::jsonString(customPolicy["activeReselect"],"optimize")); std::string linkSelectMethodStr(OSUtils::jsonString(customPolicy["activeReselect"],"always"));
if (linkSelectMethodStr == "always") { if (linkSelectMethodStr == "always") {
newTemplateBond->setLinkSelectMethod(ZT_BOND_RESELECTION_POLICY_ALWAYS); newTemplateBond->setLinkSelectMethod(ZT_BOND_RESELECTION_POLICY_ALWAYS);
} }
@ -2562,7 +2599,29 @@ public:
fprintf(stderr,"WARNING: using manually-specified secondary and/or tertiary ports. This can cause NAT issues." ZT_EOL_S); fprintf(stderr,"WARNING: using manually-specified secondary and/or tertiary ports. This can cause NAT issues." ZT_EOL_S);
} }
_portMappingEnabled = OSUtils::jsonBool(settings["portMappingEnabled"],true); _portMappingEnabled = OSUtils::jsonBool(settings["portMappingEnabled"],true);
_node->setLowBandwidthMode(OSUtils::jsonBool(settings["lowBandwidthMode"],false)); #if defined(__LINUX__) || defined(__FreeBSD__)
_multicoreEnabled = OSUtils::jsonBool(settings["multicoreEnabled"],false);
_concurrency = OSUtils::jsonInt(settings["concurrency"],1);
_cpuPinningEnabled = OSUtils::jsonBool(settings["cpuPinningEnabled"],false);
if (_multicoreEnabled) {
unsigned int maxConcurrency = std::thread::hardware_concurrency();
if (_concurrency <= 1 || _concurrency >= maxConcurrency) {
unsigned int conservativeDefault = (std::thread::hardware_concurrency() >= 4 ? 2 : 1);
fprintf(stderr, "Concurrency level provided (%d) is invalid, assigning conservative default value of (%d)\n", _concurrency, conservativeDefault);
_concurrency = conservativeDefault;
}
setUpMultithreading();
}
else {
// Force values in case the user accidentally defined them with multicore disabled
_concurrency = 1;
_cpuPinningEnabled = false;
}
#else
_multicoreEnabled = false;
_concurrency = 1;
_cpuPinningEnabled = false;
#endif
#ifndef ZT_SDK #ifndef ZT_SDK
const std::string up(OSUtils::jsonString(settings["softwareUpdate"],ZT_SOFTWARE_UPDATE_DEFAULT)); const std::string up(OSUtils::jsonString(settings["softwareUpdate"],ZT_SOFTWARE_UPDATE_DEFAULT));
@ -2877,14 +2936,17 @@ public:
// Handlers for Node and Phy<> callbacks // Handlers for Node and Phy<> callbacks
// ========================================================================= // =========================================================================
inline void phyOnDatagram(PhySocket *sock,void **uptr,const struct sockaddr *localAddr,const struct sockaddr *from,void *data,unsigned long len)
inline void phyOnDatagram(PhySocket* sock, void** uptr, const struct sockaddr* localAddr, const struct sockaddr* from, void* data, unsigned long len)
{ {
if (_forceTcpRelay) { if (_forceTcpRelay) {
return; return;
} }
Metrics::udp_recv += len; Metrics::udp_recv += len;
const uint64_t now = OSUtils::now(); const uint64_t now = OSUtils::now();
if ((len >= 16)&&(reinterpret_cast<const InetAddress *>(from)->ipScope() == InetAddress::IP_SCOPE_GLOBAL)) { if ((len >= 16) && (reinterpret_cast<const InetAddress*>(from)->ipScope() == InetAddress::IP_SCOPE_GLOBAL)) {
_lastDirectReceiveFromGlobal = now; _lastDirectReceiveFromGlobal = now;
} }
const ZT_ResultCode rc = _node->processWirePacket(nullptr,now,reinterpret_cast<int64_t>(sock),reinterpret_cast<const struct sockaddr_storage *>(from),data,len,&_nextBackgroundTaskDeadline); const ZT_ResultCode rc = _node->processWirePacket(nullptr,now,reinterpret_cast<int64_t>(sock),reinterpret_cast<const struct sockaddr_storage *>(from),data,len,&_nextBackgroundTaskDeadline);
@ -2898,6 +2960,7 @@ public:
} }
} }
inline void phyOnTcpConnect(PhySocket *sock,void **uptr,bool success) inline void phyOnTcpConnect(PhySocket *sock,void **uptr,bool success)
{ {
if (!success) { if (!success) {
@ -3116,6 +3179,8 @@ public:
n.setTap(EthernetTap::newInstance( n.setTap(EthernetTap::newInstance(
nullptr, nullptr,
_concurrency,
_cpuPinningEnabled,
_homePath.c_str(), _homePath.c_str(),
MAC(nwc->mac), MAC(nwc->mac),
nwc->mtu, nwc->mtu,
@ -3630,8 +3695,9 @@ public:
inline void nodeVirtualNetworkFrameFunction(uint64_t nwid,void **nuptr,uint64_t sourceMac,uint64_t destMac,unsigned int etherType,unsigned int vlanId,const void *data,unsigned int len) inline void nodeVirtualNetworkFrameFunction(uint64_t nwid,void **nuptr,uint64_t sourceMac,uint64_t destMac,unsigned int etherType,unsigned int vlanId,const void *data,unsigned int len)
{ {
NetworkState *n = reinterpret_cast<NetworkState *>(*nuptr); NetworkState *n = reinterpret_cast<NetworkState *>(*nuptr);
if ((!n)||(!n->tap())) if ((!n)||(!n->tap())) {
return; return;
}
n->tap()->put(MAC(sourceMac),MAC(destMac),etherType,data,len); n->tap()->put(MAC(sourceMac),MAC(destMac),etherType,data,len);
} }

View File

@ -27,7 +27,7 @@
/** /**
* Revision * Revision
*/ */
#define ZEROTIER_ONE_VERSION_REVISION 0 #define ZEROTIER_ONE_VERSION_REVISION 1
/** /**
* Build version * Build version

View File

@ -88,6 +88,7 @@
<ClCompile Include="..\..\node\Node.cpp" /> <ClCompile Include="..\..\node\Node.cpp" />
<ClCompile Include="..\..\node\OutboundMulticast.cpp" /> <ClCompile Include="..\..\node\OutboundMulticast.cpp" />
<ClCompile Include="..\..\node\Packet.cpp" /> <ClCompile Include="..\..\node\Packet.cpp" />
<ClCompile Include="..\..\node\PacketMultiplexer.cpp" />
<ClCompile Include="..\..\node\Path.cpp" /> <ClCompile Include="..\..\node\Path.cpp" />
<ClCompile Include="..\..\node\Peer.cpp" /> <ClCompile Include="..\..\node\Peer.cpp" />
<ClCompile Include="..\..\node\Poly1305.cpp"> <ClCompile Include="..\..\node\Poly1305.cpp">

View File

@ -294,6 +294,9 @@
<ClCompile Include="..\..\node\Metrics.cpp"> <ClCompile Include="..\..\node\Metrics.cpp">
<Filter>Source Files\node</Filter> <Filter>Source Files\node</Filter>
</ClCompile> </ClCompile>
<ClCompile Include="..\..\node\PacketMultiplexer.cpp">
<Filter>Source Files\node</Filter>
</ClCompile>
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>
<ClInclude Include="resource.h"> <ClInclude Include="resource.h">

View File

@ -1,5 +1,5 @@
Name: zerotier-one Name: zerotier-one
Version: 1.14.0 Version: 1.14.1
Release: 1%{?dist} Release: 1%{?dist}
Summary: ZeroTier network virtualization service Summary: ZeroTier network virtualization service