Skip to content

Commit

Permalink
Fix build for macOS, tune to prevent packet re-ordering
Browse files Browse the repository at this point in the history
  • Loading branch information
@joseph-henry
joseph-henry committed Jul 3, 2024
1 parent 2d89f07 commit 7a8eee2
Show file tree
Hide file tree
Showing 7 changed files with 102 additions and 96 deletions.
4 changes: 2 additions & 2 deletions osdep/EthernetTap.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -84,9 +84,9 @@ std::shared_ptr<EthernetTap> EthernetTap::newInstance(
// The "feth" virtual Ethernet device type appeared in Darwin 17.x.x. Older versions
// (Sierra and earlier) must use the a kernel extension.
if (strtol(osrelease,(char **)0,10) < 17) {
return std::shared_ptr<EthernetTap>(new MacKextEthernetTap(homePath,concurrency,mac,mtu,metric,nwid,friendlyName,handler,arg));
return std::shared_ptr<EthernetTap>(new MacKextEthernetTap(homePath,mac,mtu,metric,nwid,friendlyName,handler,arg));
} else {
return std::shared_ptr<EthernetTap>(new MacEthernetTap(homePath,concurrency,mac,mtu,metric,nwid,friendlyName,handler,arg));
return std::shared_ptr<EthernetTap>(new MacEthernetTap(homePath,mac,mtu,metric,nwid,friendlyName,handler,arg));
}
}
}
Expand Down
2 changes: 1 addition & 1 deletion osdep/LinuxEthernetTap.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -223,7 +223,7 @@ LinuxEthernetTap::LinuxEthernetTap(
(void)::pipe(_shutdownSignalPipe);

bool _enablePinning = false;
char* envvar = std::getenv("ZT_CPU_PINNING");
char* envvar = std::getenv("ZT_CORE_PINNING");
if (envvar) {
int tmp = atoi(envvar);
if (tmp > 0) {
Expand Down
22 changes: 4 additions & 18 deletions osdep/MacEthernetTap.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,6 @@ static bool fethMaxMtuAdjusted = false;

MacEthernetTap::MacEthernetTap(
const char *homePath,
unsigned int concurrency,
const MAC &mac,
unsigned int mtu,
unsigned int metric,
Expand All @@ -78,7 +77,6 @@ MacEthernetTap::MacEthernetTap(
void (*handler)(void *,void *,uint64_t,const MAC &,const MAC &,unsigned int,unsigned int,const void *data,unsigned int len),
void *arg) :
_handler(handler),
_concurrency(concurrency),
_arg(arg),
_nwid(nwid),
_homePath(homePath),
Expand Down Expand Up @@ -288,9 +286,6 @@ MacEthernetTap::~MacEthernetTap()
}

Thread::join(_thread);
for (std::thread &t : _rxThreads) {
t.join();
}
}

void MacEthernetTap::setEnabled(bool en) { _enabled = en; }
Expand Down Expand Up @@ -479,25 +474,17 @@ void MacEthernetTap::setMtu(unsigned int mtu)
void MacEthernetTap::threadMain()
throw()
{
Thread::sleep(250);

for (unsigned int i = 0; i < _concurrency; ++i) {
_rxThreads.push_back(std::thread([this, i] {

fprintf(stderr, "starting thread %d\n", i);

char agentReadBuf[ZT_MACETHERNETTAP_AGENT_READ_BUF_SIZE];
char agentStderrBuf[256];
fd_set readfds,nullfds;
MAC to,from;

Thread::sleep(250);

const int nfds = std::max(std::max(_shutdownSignalPipe[0],_agentStdout),_agentStderr) + 1;
long agentReadPtr = 0;

if (i == 0) {
fcntl(_agentStdout,F_SETFL,fcntl(_agentStdout,F_GETFL)|O_NONBLOCK);
fcntl(_agentStderr,F_SETFL,fcntl(_agentStderr,F_GETFL)|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);
Expand Down Expand Up @@ -546,7 +533,6 @@ void MacEthernetTap::threadMain()
*/
}
}
}));}

::close(_agentStdin);
::close(_agentStdout);
Expand Down
4 changes: 0 additions & 4 deletions osdep/MacEthernetTap.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@
#include <stdexcept>
#include <string>
#include <vector>
#include <thread>

namespace ZeroTier {

Expand All @@ -37,7 +36,6 @@ class MacEthernetTap : public EthernetTap
public:
MacEthernetTap(
const char *homePath,
unsigned int concurrency,
const MAC &mac,
unsigned int mtu,
unsigned int metric,
Expand Down Expand Up @@ -69,7 +67,6 @@ class MacEthernetTap : public EthernetTap
uint64_t _nwid;
Thread _thread;
std::string _homePath;
unsigned int _concurrency;
std::string _dev;
std::vector<MulticastGroup> _multicastGroups;
Mutex _putLock;
Expand All @@ -82,7 +79,6 @@ class MacEthernetTap : public EthernetTap
volatile bool _enabled;
mutable std::vector<InetAddress> _ifaddrs;
mutable uint64_t _lastIfAddrsUpdate;
std::vector<std::thread> _rxThreads;

};

Expand Down
2 changes: 0 additions & 2 deletions osdep/MacKextEthernetTap.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -306,7 +306,6 @@ static Mutex globalTapCreateLock;

MacKextEthernetTap::MacKextEthernetTap(
const char *homePath,
unsigned int concurrency,
const MAC &mac,
unsigned int mtu,
unsigned int metric,
Expand All @@ -318,7 +317,6 @@ MacKextEthernetTap::MacKextEthernetTap(
_arg(arg),
_nwid(nwid),
_homePath(homePath),
_concurrency(concurrency),
_mtu(mtu),
_metric(metric),
_fd(0),
Expand Down
2 changes: 0 additions & 2 deletions osdep/MacKextEthernetTap.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,6 @@ class MacKextEthernetTap : public EthernetTap
public:
MacKextEthernetTap(
const char *homePath,
unsigned int concurrency,
const MAC &mac,
unsigned int mtu,
unsigned int metric,
Expand Down Expand Up @@ -72,7 +71,6 @@ class MacKextEthernetTap : public EthernetTap
std::string _homePath;
std::string _dev;
std::vector<MulticastGroup> _multicastGroups;
unsigned int _concurrency;
unsigned int _mtu;
unsigned int _metric;
int _fd;
Expand Down
162 changes: 95 additions & 67 deletions service/OneService.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -801,6 +801,7 @@ class OneServiceImpl : public OneService
std::vector<PacketRecord *> _rxPacketVector;
std::vector<std::thread> _rxPacketThreads;
Mutex _rxPacketVector_m,_rxPacketThreads_m;
bool _enableMulticore;

bool _allowTcpFallbackRelay;
bool _forceTcpRelay;
Expand Down Expand Up @@ -934,74 +935,87 @@ class OneServiceImpl : public OneService
_ports[1] = 0;
_ports[2] = 0;

bool _enablePinning = false;
char* pinningVar = std::getenv("ZT_CPU_PINNING");
if (pinningVar) {
int tmp = atoi(pinningVar);
_enableMulticore = false;
char* multicoreVar = std::getenv("ZT_ENABLE_MULTICORE");
if (multicoreVar) {
int tmp = atoi(multicoreVar);
if (tmp > 0) {
_enablePinning = true;
_enableMulticore = true;
}
}
char* concurrencyVar = std::getenv("ZT_PACKET_PROCESSING_CONCURRENCY");
if (concurrencyVar) {
int tmp = atoi(concurrencyVar);
if (tmp > 0) {
_rxThreadCount = tmp;
if (_enableMulticore) {
bool _enablePinning = false;
char* pinningVar = std::getenv("ZT_CORE_PINNING");
if (pinningVar) {
int tmp = atoi(pinningVar);
if (tmp > 0) {
_enablePinning = true;
}
}
char* concurrencyVar = std::getenv("ZT_CONCURRENCY");
if (concurrencyVar) {
int tmp = atoi(concurrencyVar);
if (tmp > 0) {
_rxThreadCount = tmp;
}
else {
_rxThreadCount = std::thread::hardware_concurrency() >= 4 ? 2 : 1;
}
}
else {
_rxThreadCount = std::thread::hardware_concurrency();
_rxThreadCount = std::thread::hardware_concurrency() >= 4 ? 2 : 1;
}
}
else {
_rxThreadCount = std::thread::hardware_concurrency();
}
for (unsigned int i = 0; i < _rxThreadCount; ++i) {
_rxPacketThreads.push_back(std::thread([this, i]() {
fprintf(stderr, "using %d rx threads\n", _rxThreadCount);
for (unsigned int i = 0; i < _rxThreadCount; ++i) {
_rxPacketThreads.push_back(std::thread([this, i, _enablePinning]() {

if (_enablePinning) {
#if defined(__LINUX__) || defined(__FreeBSD__) /* || defined(__APPLE__) */
int pinCore = i % _rxThreadCount;
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 pinCore = i % _rxThreadCount;
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);
#endif
#ifdef __LINUX__
int rc = pthread_setaffinity_np(self, sizeof(cpu_set_t), &cpuset);
int rc = pthread_setaffinity_np(self, sizeof(cpu_set_t), &cpuset);
#elif __FreeBSD__
int rc = pthread_setaffinity_np(self, sizeof(cpu_set_t), &cpuset);
int rc = pthread_setaffinity_np(self, sizeof(cpu_set_t), &cpuset);
#endif
#if defined(__LINUX__) || defined(__FreeBSD__) /* || defined(__APPLE__) */
if (rc != 0)
{
fprintf(stderr, "failed to pin thread %d to core %d: %s\n", i, pinCore, strerror(errno));
exit(1);
}
#endif
PacketRecord* packet = nullptr;
for (;;) {
if (! _rxPacketQueue.get(packet)) {
break;
}
if (! packet) {
break;
}
const ZT_ResultCode err = _node->processWirePacket(nullptr, packet->now, packet->sock, &(packet->from), packet->data, packet->size, &_nextBackgroundTaskDeadline);
if (rc != 0)
{
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;
fprintf(stderr, "failed to pin thread %d to core %d: %s\n", i, pinCore, strerror(errno));
exit(1);
}
#endif
}
}));
PacketRecord* packet = nullptr;
for (;;) {
if (! _rxPacketQueue.get(packet)) {
break;
}
if (! packet) {
break;
}
const ZT_ResultCode err = _node->processWirePacket(nullptr, packet->now, packet->sock, &(packet->from), packet->data, packet->size, &_nextBackgroundTaskDeadline);
{
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;
}
}
}));
}
}

prometheus::simpleapi::saver.set_registry(prometheus::simpleapi::registry_ptr);
Expand Down Expand Up @@ -2865,25 +2879,39 @@ class OneServiceImpl : public OneService
_lastDirectReceiveFromGlobal = now;
}

PacketRecord* packet;
_rxPacketVector_m.lock();
if (_rxPacketVector.empty()) {
packet = new PacketRecord;
if (_enableMulticore) {
PacketRecord* packet;
_rxPacketVector_m.lock();
if (_rxPacketVector.empty()) {
packet = new PacketRecord;
}
else {
packet = _rxPacketVector.back();
_rxPacketVector.pop_back();
}
_rxPacketVector_m.unlock();

packet->sock = reinterpret_cast<int64_t>(sock);
packet->now = now;
memcpy(&(packet->from), from, sizeof(struct sockaddr_storage));
packet->size = (unsigned int)len;
memcpy(packet->data, data, len);
_rxPacketQueue.postLimit(packet, 256 * _rxThreadCount);
}
else {
packet = _rxPacketVector.back();
_rxPacketVector.pop_back();
const ZT_ResultCode rc = _node->processWirePacket(nullptr,now,reinterpret_cast<int64_t>(sock),reinterpret_cast<const struct sockaddr_storage *>(from),data,len,&_nextBackgroundTaskDeadline);
if (ZT_ResultCode_isFatal(rc)) {
char tmp[256];
OSUtils::ztsnprintf(tmp,sizeof(tmp),"fatal error code from processWirePacket: %d",(int)rc);
Mutex::Lock _l(_termReason_m);
_termReason = ONE_UNRECOVERABLE_ERROR;
_fatalErrorMessage = tmp;
this->terminate();
}
}
_rxPacketVector_m.unlock();

packet->sock = reinterpret_cast<int64_t>(sock);
packet->now = now;
memcpy(&(packet->from), from, sizeof(struct sockaddr_storage));
packet->size = (unsigned int)len;
memcpy(packet->data, data, len);
_rxPacketQueue.postLimit(packet, 256 * _rxThreadCount);
}


inline void phyOnTcpConnect(PhySocket *sock,void **uptr,bool success)
{
if (!success) {
Expand Down Expand Up @@ -3103,7 +3131,7 @@ class OneServiceImpl : public OneService
n.setTap(EthernetTap::newInstance(
nullptr,
_homePath.c_str(),
_rxThreadCount,
_enableMulticore ? _rxThreadCount : 1,
MAC(nwc->mac),
nwc->mtu,
(unsigned int)ZT_IF_METRIC,
Expand Down

0 comments on commit 7a8eee2

Please sign in to comment.