MERGE current "dev" into "netcon" -- should not affect netcon itself but will retest -- brings ZeroTier core up to 1.1.0

service/ClusterDefinition.hpp

@@ -0,0 +1,125 @@
+/*
+ * ZeroTier One - Network Virtualization Everywhere
+ * Copyright (C) 2011-2015  ZeroTier, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * --
+ *
+ * ZeroTier may be used and distributed under the terms of the GPLv3, which
+ * are available at: http://www.gnu.org/licenses/gpl-3.0.html
+ *
+ * If you would like to embed ZeroTier into a commercial application or
+ * redistribute it in a modified binary form, please contact ZeroTier Networks
+ * LLC. Start here: http://www.zerotier.com/
+ */
+
+#ifndef ZT_CLUSTERDEFINITION_HPP
+#define ZT_CLUSTERDEFINITION_HPP
+
+#ifdef ZT_ENABLE_CLUSTER
+
+#include <vector>
+#include <algorithm>
+
+#include "../node/Constants.hpp"
+#include "../node/Utils.hpp"
+#include "../osdep/OSUtils.hpp"
+
+namespace ZeroTier {
+
+/**
+ * Parser for cluster definition file
+ */
+class ClusterDefinition
+{
+public:
+	struct MemberDefinition
+	{
+		MemberDefinition() : id(0),x(0),y(0),z(0) { name[0] = (char)0; }
+
+		unsigned int id;
+		int x,y,z;
+		char name[256];
+		InetAddress clusterEndpoint;
+		std::vector<InetAddress> zeroTierEndpoints;
+	};
+
+	ClusterDefinition(uint64_t myAddress,const char *pathToClusterFile)
+	{
+		std::string cf;
+		if (!OSUtils::readFile(pathToClusterFile,cf))
+			return;
+
+		char myAddressStr[64];
+		Utils::snprintf(myAddressStr,sizeof(myAddressStr),"%.10llx",myAddress);
+
+		std::vector<std::string> lines(Utils::split(cf.c_str(),"\r\n","",""));
+		for(std::vector<std::string>::iterator l(lines.begin());l!=lines.end();++l) {
+			std::vector<std::string> fields(Utils::split(l->c_str()," \t","",""));
+			if ((fields.size() < 5)||(fields[0][0] == '#')||(fields[0] != myAddressStr))
+				continue;
+
+			int id = Utils::strToUInt(fields[1].c_str());
+			if ((id < 0)||(id > ZT_CLUSTER_MAX_MEMBERS))
+				continue;
+			MemberDefinition &md = _md[id];
+
+			md.id = (unsigned int)id;
+			if (fields.size() >= 6) {
+				std::vector<std::string> xyz(Utils::split(fields[5].c_str(),",","",""));
+				md.x = (xyz.size() > 0) ? Utils::strToInt(xyz[0].c_str()) : 0;
+				md.y = (xyz.size() > 1) ? Utils::strToInt(xyz[1].c_str()) : 0;
+				md.z = (xyz.size() > 2) ? Utils::strToInt(xyz[2].c_str()) : 0;
+			}
+			Utils::scopy(md.name,sizeof(md.name),fields[2].c_str());
+			md.clusterEndpoint.fromString(fields[3]);
+			if (!md.clusterEndpoint)
+				continue;
+			std::vector<std::string> zips(Utils::split(fields[4].c_str(),",","",""));
+			for(std::vector<std::string>::iterator zip(zips.begin());zip!=zips.end();++zip) {
+				InetAddress i;
+				i.fromString(*zip);
+				if (i)
+					md.zeroTierEndpoints.push_back(i);
+			}
+
+			_ids.push_back((unsigned int)id);
+		}
+
+		std::sort(_ids.begin(),_ids.end());
+	}
+
+	inline const MemberDefinition &operator[](unsigned int id) const throw() { return _md[id]; }
+	inline unsigned int size() const throw() { return (unsigned int)_ids.size(); }
+	inline const std::vector<unsigned int> &ids() const throw() { return _ids; }
+
+	inline std::vector<MemberDefinition> members() const
+	{
+		std::vector<MemberDefinition> m;
+		for(std::vector<unsigned int>::const_iterator i(_ids.begin());i!=_ids.end();++i)
+			m.push_back(_md[*i]);
+		return m;
+	}
+
+private:
+	MemberDefinition _md[ZT_CLUSTER_MAX_MEMBERS];
+	std::vector<unsigned int> _ids;
+};
+
+} // namespace ZeroTier
+
+#endif // ZT_ENABLE_CLUSTER
+
+#endif

service/ClusterGeoIpService.cpp

@@ -0,0 +1,197 @@
+/*
+ * ZeroTier One - Network Virtualization Everywhere
+ * Copyright (C) 2011-2015  ZeroTier, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * --
+ *
+ * ZeroTier may be used and distributed under the terms of the GPLv3, which
+ * are available at: http://www.gnu.org/licenses/gpl-3.0.html
+ *
+ * If you would like to embed ZeroTier into a commercial application or
+ * redistribute it in a modified binary form, please contact ZeroTier Networks
+ * LLC. Start here: http://www.zerotier.com/
+ */
+
+#ifdef ZT_ENABLE_CLUSTER
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <stdint.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <sys/wait.h>
+#include <signal.h>
+#include <errno.h>
+
+#include <iostream>
+
+#include "ClusterGeoIpService.hpp"
+#include "../node/Utils.hpp"
+#include "../osdep/OSUtils.hpp"
+
+// 120 days
+#define ZT_CLUSTERGEOIPSERVICE_INTERNAL_CACHE_TTL 10368000000ULL
+
+namespace ZeroTier {
+
+ClusterGeoIpService::ClusterGeoIpService(const char *pathToExe) :
+	_pathToExe(pathToExe),
+	_sOutputFd(-1),
+	_sInputFd(-1),
+	_sPid(0),
+	_run(true)
+{
+	_thread = Thread::start(this);
+}
+
+ClusterGeoIpService::~ClusterGeoIpService()
+{
+	_run = false;
+	long p = _sPid;
+	if (p > 0) {
+		::kill(p,SIGTERM);
+		Thread::sleep(500);
+		::kill(p,SIGKILL);
+	}
+	Thread::join(_thread);
+}
+
+bool ClusterGeoIpService::locate(const InetAddress &ip,int &x,int &y,int &z)
+{
+	InetAddress ipNoPort(ip);
+	ipNoPort.setPort(0); // we index cache by IP only
+	const uint64_t now = OSUtils::now();
+
+	bool r = false;
+	{
+		Mutex::Lock _l(_cache_m);
+		std::map< InetAddress,_CE >::iterator c(_cache.find(ipNoPort));
+		if (c != _cache.end()) {
+			x = c->second.x;
+			y = c->second.y;
+			z = c->second.z;
+			if ((now - c->second.ts) < ZT_CLUSTERGEOIPSERVICE_INTERNAL_CACHE_TTL)
+				return true;
+			else r = true; // return true but refresh as well
+		}
+	}
+
+	{
+		Mutex::Lock _l(_sOutputLock);
+		if (_sOutputFd >= 0) {
+			std::string ips(ipNoPort.toIpString());
+			ips.push_back('\n');
+			//fprintf(stderr,"ClusterGeoIpService: << %s",ips.c_str());
+			::write(_sOutputFd,ips.data(),ips.length());
+		}
+	}
+
+	return r;
+}
+
+void ClusterGeoIpService::threadMain()
+	throw()
+{
+	char linebuf[65536];
+	char buf[65536];
+	long n,lineptr;
+
+	while (_run) {
+		{
+			Mutex::Lock _l(_sOutputLock);
+
+			_sOutputFd = -1;
+			_sInputFd = -1;
+			_sPid = 0;
+
+			int stdinfds[2] = { 0,0 };  // sub-process's stdin, our output
+			int stdoutfds[2] = { 0,0 }; // sub-process's stdout, our input
+			::pipe(stdinfds);
+			::pipe(stdoutfds);
+
+			long p = (long)::vfork();
+			if (p < 0) {
+				Thread::sleep(500);
+				continue;
+			} else if (p == 0) {
+				::close(stdinfds[1]);
+				::close(stdoutfds[0]);
+				::dup2(stdinfds[0],STDIN_FILENO);
+				::dup2(stdoutfds[1],STDOUT_FILENO);
+				::execl(_pathToExe.c_str(),_pathToExe.c_str(),(const char *)0);
+				::exit(1);
+			} else {
+				::close(stdinfds[0]);
+				::close(stdoutfds[1]);
+				_sOutputFd = stdinfds[1];
+				_sInputFd = stdoutfds[0];
+				_sPid = p;
+			}
+		}
+
+		lineptr = 0;
+		while (_run) {
+			n = ::read(_sInputFd,buf,sizeof(buf));
+			if (n <= 0) {
+				if (errno == EINTR)
+					continue;
+				else break;
+			}
+			for(long i=0;i<n;++i) {
+				if (lineptr > (long)sizeof(linebuf))
+					lineptr = 0;
+				if ((buf[i] == '\n')||(buf[i] == '\r')) {
+					linebuf[lineptr] = (char)0;
+					if (lineptr > 0) {
+						//fprintf(stderr,"ClusterGeoIpService: >> %s\n",linebuf);
+						try {
+							std::vector<std::string> result(Utils::split(linebuf,",","",""));
+							if ((result.size() >= 7)&&(result[1] == "1")) {
+								InetAddress rip(result[0],0);
+								if ((rip.ss_family == AF_INET)||(rip.ss_family == AF_INET6)) {
+									_CE ce;
+									ce.ts = OSUtils::now();
+									ce.x = (int)::strtol(result[4].c_str(),(char **)0,10);
+									ce.y = (int)::strtol(result[5].c_str(),(char **)0,10);
+									ce.z = (int)::strtol(result[6].c_str(),(char **)0,10);
+									//fprintf(stderr,"ClusterGeoIpService: %s is at %d,%d,%d\n",rip.toIpString().c_str(),ce.x,ce.y,ce.z);
+									{
+										Mutex::Lock _l2(_cache_m);
+										_cache[rip] = ce;
+									}
+								}
+							}
+						} catch ( ... ) {}
+					}
+					lineptr = 0;
+				} else linebuf[lineptr++] = buf[i];
+			}
+		}
+
+		::close(_sOutputFd);
+		::close(_sInputFd);
+		::kill(_sPid,SIGTERM);
+		Thread::sleep(250);
+		::kill(_sPid,SIGKILL);
+		::waitpid(_sPid,(int *)0,0);
+	}
+}
+
+} // namespace ZeroTier
+
+#endif // ZT_ENABLE_CLUSTER

service/ClusterGeoIpService.hpp

@@ -0,0 +1,94 @@
+/*
+ * ZeroTier One - Network Virtualization Everywhere
+ * Copyright (C) 2011-2015  ZeroTier, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * --
+ *
+ * ZeroTier may be used and distributed under the terms of the GPLv3, which
+ * are available at: http://www.gnu.org/licenses/gpl-3.0.html
+ *
+ * If you would like to embed ZeroTier into a commercial application or
+ * redistribute it in a modified binary form, please contact ZeroTier Networks
+ * LLC. Start here: http://www.zerotier.com/
+ */
+
+#ifndef ZT_CLUSTERGEOIPSERVICE_HPP
+#define ZT_CLUSTERGEOIPSERVICE_HPP
+
+#ifdef ZT_ENABLE_CLUSTER
+
+#include <vector>
+#include <map>
+#include <string>
+
+#include "../node/Constants.hpp"
+#include "../node/InetAddress.hpp"
+#include "../node/Mutex.hpp"
+#include "../osdep/Thread.hpp"
+
+namespace ZeroTier {
+
+/**
+ * Runs the Cluster GeoIP service in the background and resolves geoIP queries
+ */
+class ClusterGeoIpService
+{
+public:
+	/**
+	 * @param pathToExe Path to cluster geo-resolution service executable
+	 */
+	ClusterGeoIpService(const char *pathToExe);
+
+	~ClusterGeoIpService();
+
+	/**
+	 * Attempt to locate an IP
+	 *
+	 * This returns true if x, y, and z are set. Otherwise it returns false
+	 * and a geo-locate job is ordered in the background. This usually takes
+	 * 500-1500ms to complete, after which time results will be available.
+	 * If false is returned the supplied coordinate variables are unchanged.
+	 *
+	 * @param ip IPv4 or IPv6 address
+	 * @param x Reference to variable to receive X
+	 * @param y Reference to variable to receive Y
+	 * @param z Reference to variable to receive Z
+	 * @return True if coordinates were set
+	 */
+	bool locate(const InetAddress &ip,int &x,int &y,int &z);
+
+	void threadMain()
+		throw();
+
+private:
+	const std::string _pathToExe;
+	int _sOutputFd;
+	int _sInputFd;
+	volatile long _sPid;
+	volatile bool _run;
+	Thread _thread;
+	Mutex _sOutputLock;
+
+	struct _CE { uint64_t ts; int x,y,z; };
+	std::map< InetAddress,_CE > _cache;
+	Mutex _cache_m;
+};
+
+} // namespace ZeroTier
+
+#endif // ZT_ENABLE_CLUSTER
+
+#endif

service/ControlPlane.cpp

@@ -182,14 +182,12 @@ static std::string _jsonEnumerate(unsigned int depth,const ZT_PeerPhysicalPath *
 			"%s\t\"address\": \"%s\",\n"
 			"%s\t\"lastSend\": %llu,\n"
 			"%s\t\"lastReceive\": %llu,\n"
-			"%s\t\"fixed\": %s,\n"
 			"%s\t\"active\": %s,\n"
 			"%s\t\"preferred\": %s\n"
 			"%s}",
 			prefix,_jsonEscape(reinterpret_cast<const InetAddress *>(&(pp[i].address))->toString()).c_str(),
 			prefix,pp[i].lastSend,
 			prefix,pp[i].lastReceive,
-			prefix,(pp[i].fixed == 0) ? "false" : "true",
 			prefix,(pp[i].active == 0) ? "false" : "true",
 			prefix,(pp[i].preferred == 0) ? "false" : "true",
 			prefix);
@@ -267,7 +265,7 @@ unsigned int ControlPlane::handleRequest(
 	std::string &responseBody,
 	std::string &responseContentType)
 {
-	char json[1024];
+	char json[8194];
 	unsigned int scode = 404;
 	std::vector<std::string> ps(Utils::split(path.c_str(),"/","",""));
 	std::map<std::string,std::string> urlArgs;
@@ -356,29 +354,65 @@ unsigned int ControlPlane::handleRequest(
 
 			if (ps[0] == "status") {
 				responseContentType = "application/json";
+
 				ZT_NodeStatus status;
 				_node->status(&status);
+
+				std::string clusterJson;
+#ifdef ZT_ENABLE_CLUSTER
+				{
+					ZT_ClusterStatus cs;
+					_node->clusterStatus(&cs);
+
+					if (cs.clusterSize >= 1) {
+						char t[1024];
+						Utils::snprintf(t,sizeof(t),"{\n\t\t\"myId\": %u,\n\t\t\"clusterSize\": %u,\n\t\t\"members\": [",cs.myId,cs.clusterSize);
+						clusterJson.append(t);
+						for(unsigned int i=0;i<cs.clusterSize;++i) {
+							Utils::snprintf(t,sizeof(t),"%s\t\t\t{\n\t\t\t\t\"id\": %u,\n\t\t\t\t\"msSinceLastHeartbeat\": %u,\n\t\t\t\t\"alive\": %s,\n\t\t\t\t\"x\": %d,\n\t\t\t\t\"y\": %d,\n\t\t\t\t\"z\": %d,\n\t\t\t\t\"load\": %llu,\n\t\t\t\t\"peers\": %llu\n\t\t\t}",
+								((i == 0) ? "\n" : ",\n"),
+								cs.members[i].id,
+								cs.members[i].msSinceLastHeartbeat,
+								(cs.members[i].alive != 0) ? "true" : "false",
+								cs.members[i].x,
+								cs.members[i].y,
+								cs.members[i].z,
+								cs.members[i].load,
+								cs.members[i].peers);
+							clusterJson.append(t);
+						}
+						clusterJson.append(" ]\n\t\t}");
+					}
+				}
+#endif
+
 				Utils::snprintf(json,sizeof(json),
 					"{\n"
 					"\t\"address\": \"%.10llx\",\n"
 					"\t\"publicIdentity\": \"%s\",\n"
+					"\t\"worldId\": %llu,\n"
+					"\t\"worldTimestamp\": %llu,\n"
 					"\t\"online\": %s,\n"
 					"\t\"tcpFallbackActive\": %s,\n"
 					"\t\"versionMajor\": %d,\n"
 					"\t\"versionMinor\": %d,\n"
 					"\t\"versionRev\": %d,\n"
 					"\t\"version\": \"%d.%d.%d\",\n"
-					"\t\"clock\": %llu\n"
+					"\t\"clock\": %llu,\n"
+					"\t\"cluster\": %s\n"
 					"}\n",
 					status.address,
 					status.publicIdentity,
+					status.worldId,
+					status.worldTimestamp,
 					(status.online) ? "true" : "false",
 					(_svc->tcpFallbackActive()) ? "true" : "false",
 					ZEROTIER_ONE_VERSION_MAJOR,
 					ZEROTIER_ONE_VERSION_MINOR,
 					ZEROTIER_ONE_VERSION_REVISION,
 					ZEROTIER_ONE_VERSION_MAJOR,ZEROTIER_ONE_VERSION_MINOR,ZEROTIER_ONE_VERSION_REVISION,
-					(unsigned long long)OSUtils::now());
+					(unsigned long long)OSUtils::now(),
+					((clusterJson.length() > 0) ? clusterJson.c_str() : "null"));
 				responseBody = json;
 				scode = 200;
 			} else if (ps[0] == "config") {

service/OneService.cpp

@@ -58,6 +58,8 @@
 
 #include "OneService.hpp"
 #include "ControlPlane.hpp"
+#include "ClusterGeoIpService.hpp"
+#include "ClusterDefinition.hpp"
 
 /**
  * Uncomment to enable UDP breakage switch
@@ -378,9 +380,14 @@ static int SnodeVirtualNetworkConfigFunction(ZT_Node *node,void *uptr,uint64_t n
 static void SnodeEventCallback(ZT_Node *node,void *uptr,enum ZT_Event event,const void *metaData);
 static long SnodeDataStoreGetFunction(ZT_Node *node,void *uptr,const char *name,void *buf,unsigned long bufSize,unsigned long readIndex,unsigned long *totalSize);
 static int SnodeDataStorePutFunction(ZT_Node *node,void *uptr,const char *name,const void *data,unsigned long len,int secure);
-static int SnodeWirePacketSendFunction(ZT_Node *node,void *uptr,const struct sockaddr_storage *localAddr,const struct sockaddr_storage *addr,const void *data,unsigned int len);
+static int SnodeWirePacketSendFunction(ZT_Node *node,void *uptr,const struct sockaddr_storage *localAddr,const struct sockaddr_storage *addr,const void *data,unsigned int len,unsigned int ttl);
 static void SnodeVirtualNetworkFrameFunction(ZT_Node *node,void *uptr,uint64_t nwid,uint64_t sourceMac,uint64_t destMac,unsigned int etherType,unsigned int vlanId,const void *data,unsigned int len);
 
+#ifdef ZT_ENABLE_CLUSTER
+static void SclusterSendFunction(void *uptr,unsigned int toMemberId,const void *data,unsigned int len);
+static int SclusterGeoIpFunction(void *uptr,const struct sockaddr_storage *addr,int *x,int *y,int *z);
+#endif
+
 static void StapFrameHandler(void *uptr,uint64_t nwid,const MAC &from,const MAC &to,unsigned int etherType,unsigned int vlanId,const void *data,unsigned int len);
 
 static int ShttpOnMessageBegin(http_parser *parser);
@@ -430,31 +437,45 @@ struct TcpConnection
 	Mutex writeBuf_m;
 };
 
+// Use a bigger buffer on AMD64 since these are likely to be bigger and
+// servers. Otherwise use a smaller buffer. This makes no difference
+// except under very high load.
+#if (defined(__amd64) || defined(__amd64__) || defined(__x86_64) || defined(__x86_64__) || defined(__AMD64) || defined(__AMD64__))
+#define ZT_UDP_DESIRED_BUF_SIZE 1048576
+#else
+#define ZT_UDP_DESIRED_BUF_SIZE 131072
+#endif
+
 class OneServiceImpl : public OneService
 {
 public:
-	OneServiceImpl(const char *hp,unsigned int port,const char *overrideRootTopology) :
-		_homePath((hp) ? hp : "."),
-		_tcpFallbackResolver(ZT_TCP_FALLBACK_RELAY),
+	OneServiceImpl(const char *hp,unsigned int port) :
+		_homePath((hp) ? hp : ".")
+		,_tcpFallbackResolver(ZT_TCP_FALLBACK_RELAY)
 #ifdef ZT_ENABLE_NETWORK_CONTROLLER
-		_controller((SqliteNetworkController *)0),
+		,_controller((SqliteNetworkController *)0)
 #endif
-		_phy(this,false,true),
-		_overrideRootTopology((overrideRootTopology) ? overrideRootTopology : ""),
-		_node((Node *)0),
-		_controlPlane((ControlPlane *)0),
-		_lastDirectReceiveFromGlobal(0),
-		_lastSendToGlobal(0),
-		_lastRestart(0),
-		_nextBackgroundTaskDeadline(0),
-		_tcpFallbackTunnel((TcpConnection *)0),
-		_termReason(ONE_STILL_RUNNING),
-		_port(0),
+		,_phy(this,false,true)
+		,_node((Node *)0)
+		,_controlPlane((ControlPlane *)0)
+		,_lastDirectReceiveFromGlobal(0)
+		,_lastSendToGlobal(0)
+		,_lastRestart(0)
+		,_nextBackgroundTaskDeadline(0)
+		,_tcpFallbackTunnel((TcpConnection *)0)
+		,_termReason(ONE_STILL_RUNNING)
+		,_port(0)
 #ifdef ZT_USE_MINIUPNPC
-		_v4UpnpUdpSocket((PhySocket *)0),
-		_upnpClient((UPNPClient *)0),
+		,_v4UpnpUdpSocket((PhySocket *)0)
+		,_upnpClient((UPNPClient *)0)
 #endif
-		_run(true)
+#ifdef ZT_ENABLE_CLUSTER
+		,_clusterMessageSocket((PhySocket *)0)
+		,_clusterGeoIpService((ClusterGeoIpService *)0)
+		,_clusterDefinition((ClusterDefinition *)0)
+		,_clusterMemberId(0)
+#endif
+		,_run(true)
 	{
 		const int portTrials = (port == 0) ? 256 : 1; // if port is 0, pick random
 		for(int k=0;k<portTrials;++k) {
@@ -465,7 +486,7 @@ public:
 			}
 
 			_v4LocalAddress = InetAddress((uint32_t)0,port);
-			_v4UdpSocket = _phy.udpBind((const struct sockaddr *)&_v4LocalAddress,reinterpret_cast<void *>(&_v4LocalAddress),131072);
+			_v4UdpSocket = _phy.udpBind((const struct sockaddr *)&_v4LocalAddress,reinterpret_cast<void *>(&_v4LocalAddress),ZT_UDP_DESIRED_BUF_SIZE);
 
 			if (_v4UdpSocket) {
 				struct sockaddr_in in4;
@@ -477,7 +498,7 @@ public:
 
 				if (_v4TcpListenSocket) {
 					_v6LocalAddress = InetAddress("\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0",16,port);
-					_v6UdpSocket = _phy.udpBind((const struct sockaddr *)&_v6LocalAddress,reinterpret_cast<void *>(&_v6LocalAddress),131072);
+					_v6UdpSocket = _phy.udpBind((const struct sockaddr *)&_v6LocalAddress,reinterpret_cast<void *>(&_v6LocalAddress),ZT_UDP_DESIRED_BUF_SIZE);
 
 					struct sockaddr_in6 in6;
 					memset((void *)&in6,0,sizeof(in6));
@@ -511,7 +532,7 @@ public:
 		for(int k=0;k<512;++k) {
 			const unsigned int upnport = 40000 + (((port + 1) * (k + 1)) % 25500);
 			_v4UpnpLocalAddress = InetAddress(0,upnport);
-			_v4UpnpUdpSocket = _phy.udpBind((const struct sockaddr *)&_v4UpnpLocalAddress,reinterpret_cast<void *>(&_v4UpnpLocalAddress),131072);
+			_v4UpnpUdpSocket = _phy.udpBind((const struct sockaddr *)&_v4UpnpLocalAddress,reinterpret_cast<void *>(&_v4UpnpLocalAddress),ZT_UDP_DESIRED_BUF_SIZE);
 			if (_v4UpnpUdpSocket) {
 				_upnpClient = new UPNPClient(upnport);
 				break;
@@ -526,12 +547,19 @@ public:
 		_phy.close(_v6UdpSocket);
 		_phy.close(_v4TcpListenSocket);
 		_phy.close(_v6TcpListenSocket);
+#ifdef ZT_ENABLE_CLUSTER
+		_phy.close(_clusterMessageSocket);
+#endif
 #ifdef ZT_USE_MINIUPNPC
 		_phy.close(_v4UpnpUdpSocket);
 		delete _upnpClient;
 #endif
 #ifdef ZT_ENABLE_NETWORK_CONTROLLER
 		delete _controller;
+#endif
+#ifdef ZT_ENABLE_CLUSTER
+		delete _clusterGeoIpService;
+		delete _clusterDefinition;
 #endif
 	}
 
@@ -565,14 +593,77 @@ public:
 				SnodeWirePacketSendFunction,
 				SnodeVirtualNetworkFrameFunction,
 				SnodeVirtualNetworkConfigFunction,
-				SnodeEventCallback,
-				((_overrideRootTopology.length() > 0) ? _overrideRootTopology.c_str() : (const char *)0));
+				SnodeEventCallback);
 
 #ifdef ZT_ENABLE_NETWORK_CONTROLLER
 			_controller = new SqliteNetworkController(_node,(_homePath + ZT_PATH_SEPARATOR_S + ZT_CONTROLLER_DB_PATH).c_str(),(_homePath + ZT_PATH_SEPARATOR_S + "circuitTestResults.d").c_str());
 			_node->setNetconfMaster((void *)_controller);
 #endif
 
+#ifdef ZT_ENABLE_CLUSTER
+		if (OSUtils::fileExists((_homePath + ZT_PATH_SEPARATOR_S + "cluster").c_str())) {
+			_clusterDefinition = new ClusterDefinition(_node->address(),(_homePath + ZT_PATH_SEPARATOR_S + "cluster").c_str());
+			if (_clusterDefinition->size() > 0) {
+				std::vector<ClusterDefinition::MemberDefinition> members(_clusterDefinition->members());
+				for(std::vector<ClusterDefinition::MemberDefinition>::iterator m(members.begin());m!=members.end();++m) {
+					PhySocket *cs = _phy.udpBind(reinterpret_cast<const struct sockaddr *>(&(m->clusterEndpoint)));
+					if (cs) {
+						if (_clusterMessageSocket) {
+							_phy.close(_clusterMessageSocket,false);
+							_phy.close(cs,false);
+
+							Mutex::Lock _l(_termReason_m);
+							_termReason = ONE_UNRECOVERABLE_ERROR;
+							_fatalErrorMessage = "Cluster: can't determine my cluster member ID: able to bind more than one cluster message socket IP/port!";
+							return _termReason;
+						}
+						_clusterMessageSocket = cs;
+						_clusterMemberId = m->id;
+					}
+				}
+
+				if (!_clusterMessageSocket) {
+					Mutex::Lock _l(_termReason_m);
+					_termReason = ONE_UNRECOVERABLE_ERROR;
+					_fatalErrorMessage = "Cluster: can't determine my cluster member ID: unable to bind to any cluster message socket IP/port.";
+					return _termReason;
+				}
+
+				if (OSUtils::fileExists((_homePath + ZT_PATH_SEPARATOR_S + "cluster-geo.exe").c_str()))
+					_clusterGeoIpService = new ClusterGeoIpService((_homePath + ZT_PATH_SEPARATOR_S + "cluster-geo.exe").c_str());
+
+				const ClusterDefinition::MemberDefinition &me = (*_clusterDefinition)[_clusterMemberId];
+				InetAddress endpoints[255];
+				unsigned int numEndpoints = 0;
+				for(std::vector<InetAddress>::const_iterator i(me.zeroTierEndpoints.begin());i!=me.zeroTierEndpoints.end();++i)
+					endpoints[numEndpoints++] = *i;
+
+				if (_node->clusterInit(
+					_clusterMemberId,
+					reinterpret_cast<const struct sockaddr_storage *>(endpoints),
+					numEndpoints,
+					me.x,
+					me.y,
+					me.z,
+					&SclusterSendFunction,
+					this,
+					(_clusterGeoIpService) ? &SclusterGeoIpFunction : 0,
+					this) == ZT_RESULT_OK) {
+
+					std::vector<ClusterDefinition::MemberDefinition> members(_clusterDefinition->members());
+					for(std::vector<ClusterDefinition::MemberDefinition>::iterator m(members.begin());m!=members.end();++m) {
+						if (m->id != _clusterMemberId)
+							_node->clusterAddMember(m->id);
+					}
+
+				}
+			} else {
+				delete _clusterDefinition;
+				_clusterDefinition = (ClusterDefinition *)0;
+			}
+		}
+#endif
+
 			_controlPlane = new ControlPlane(this,_node,(_homePath + ZT_PATH_SEPARATOR_S + "ui").c_str());
 			_controlPlane->addAuthToken(authToken.c_str());
 
@@ -589,6 +680,10 @@ public:
 				}
 			}
 
+			// Start two background threads to handle expensive ops out of line
+			Thread::start(_node);
+			Thread::start(_node);
+
 			_nextBackgroundTaskDeadline = 0;
 			uint64_t clockShouldBe = OSUtils::now();
 			_lastRestart = clockShouldBe;
@@ -664,7 +759,7 @@ public:
 #ifdef ZT_USE_MINIUPNPC
 					std::vector<InetAddress> upnpAddresses(_upnpClient->get());
 					for(std::vector<InetAddress>::const_iterator ext(upnpAddresses.begin());ext!=upnpAddresses.end();++ext)
-						_node->addLocalInterfaceAddress(reinterpret_cast<const struct sockaddr_storage *>(&(*ext)),0,ZT_LOCAL_INTERFACE_ADDRESS_TRUST_NORMAL);
+						_node->addLocalInterfaceAddress(reinterpret_cast<const struct sockaddr_storage *>(&(*ext)));
 #endif
 
 					struct ifaddrs *ifatbl = (struct ifaddrs *)0;
@@ -682,7 +777,7 @@ public:
 								if (!isZT) {
 									InetAddress ip(ifa->ifa_addr);
 									ip.setPort(_port);
-									_node->addLocalInterfaceAddress(reinterpret_cast<const struct sockaddr_storage *>(&ip),0,ZT_LOCAL_INTERFACE_ADDRESS_TRUST_NORMAL);
+									_node->addLocalInterfaceAddress(reinterpret_cast<const struct sockaddr_storage *>(&ip));
 								}
 							}
 							ifa = ifa->ifa_next;
@@ -716,7 +811,7 @@ public:
 								while (ua) {
 									InetAddress ip(ua->Address.lpSockaddr);
 									ip.setPort(_port);
-									_node->addLocalInterfaceAddress(reinterpret_cast<const struct sockaddr_storage *>(&ip),0,ZT_LOCAL_INTERFACE_ADDRESS_TRUST_NORMAL);
+									_node->addLocalInterfaceAddress(reinterpret_cast<const struct sockaddr_storage *>(&ip));
 									ua = ua->Next;
 								}
 							}
@@ -798,10 +893,19 @@ public:
 
 	inline void phyOnDatagram(PhySocket *sock,void **uptr,const struct sockaddr *from,void *data,unsigned long len)
 	{
+#ifdef ZT_ENABLE_CLUSTER
+		if (sock == _clusterMessageSocket) {
+			_lastDirectReceiveFromGlobal = OSUtils::now();
+			_node->clusterHandleIncomingMessage(data,len);
+			return;
+		}
+#endif
+
 #ifdef ZT_BREAK_UDP
 		if (OSUtils::fileExists("/tmp/ZT_BREAK_UDP"))
 			return;
 #endif
+
 		if ((len >= 16)&&(reinterpret_cast<const InetAddress *>(from)->ipScope() == InetAddress::IP_SCOPE_GLOBAL))
 			_lastDirectReceiveFromGlobal = OSUtils::now();
 		ZT_ResultCode rc = _node->processWirePacket(
@@ -955,7 +1059,7 @@ public:
 							if (from) {
 								ZT_ResultCode rc = _node->processWirePacket(
 									OSUtils::now(),
-									0,
+									&ZT_SOCKADDR_NULL,
 									reinterpret_cast<struct sockaddr_storage *>(&from),
 									data,
 									plen,
@@ -1164,16 +1268,23 @@ public:
 		}
 	}
 
-	inline int nodeWirePacketSendFunction(const struct sockaddr_storage *localAddr,const struct sockaddr_storage *addr,const void *data,unsigned int len)
+	inline int nodeWirePacketSendFunction(const struct sockaddr_storage *localAddr,const struct sockaddr_storage *addr,const void *data,unsigned int len,unsigned int ttl)
 	{
 #ifdef ZT_USE_MINIUPNPC
 		if ((localAddr->ss_family == AF_INET)&&(reinterpret_cast<const struct sockaddr_in *>(localAddr)->sin_port == reinterpret_cast<const struct sockaddr_in *>(&_v4UpnpLocalAddress)->sin_port)) {
 #ifdef ZT_BREAK_UDP
 			if (!OSUtils::fileExists("/tmp/ZT_BREAK_UDP")) {
 #endif
-			if (addr->ss_family == AF_INET)
-				return ((_phy.udpSend(_v4UpnpUdpSocket,(const struct sockaddr *)addr,data,len) != 0) ? 0 : -1);
-			else return -1;
+				if (addr->ss_family == AF_INET) {
+					if (ttl)
+						_phy.setIp4UdpTtl(_v4UpnpUdpSocket,ttl);
+					const int result = ((_phy.udpSend(_v4UpnpUdpSocket,(const struct sockaddr *)addr,data,len) != 0) ? 0 : -1);
+					if (ttl)
+						_phy.setIp4UdpTtl(_v4UpnpUdpSocket,255);
+					return result;
+				} else {
+					return -1;
+				}
 #ifdef ZT_BREAK_UDP
 			}
 #endif
@@ -1186,8 +1297,13 @@ public:
 #ifdef ZT_BREAK_UDP
 				if (!OSUtils::fileExists("/tmp/ZT_BREAK_UDP")) {
 #endif
-				if (_v4UdpSocket)
-					result = ((_phy.udpSend(_v4UdpSocket,(const struct sockaddr *)addr,data,len) != 0) ? 0 : -1);
+					if (_v4UdpSocket) {
+						if (ttl)
+							_phy.setIp4UdpTtl(_v4UdpSocket,ttl);
+						result = ((_phy.udpSend(_v4UdpSocket,(const struct sockaddr *)addr,data,len) != 0) ? 0 : -1);
+						if (ttl)
+							_phy.setIp4UdpTtl(_v4UdpSocket,255);
+					}
 #ifdef ZT_BREAK_UDP
 				}
 #endif
@@ -1318,7 +1434,6 @@ public:
 			_phy.close(tc->sock); // will call close handler, which deletes from _tcpConnections
 	}
 
-private:
 	std::string _dataStorePrepPath(const char *name) const
 	{
 		std::string p(_homePath);
@@ -1342,7 +1457,6 @@ private:
 	SqliteNetworkController *_controller;
 #endif
 	Phy<OneServiceImpl *> _phy;
-	std::string _overrideRootTopology;
 	Node *_node;
 	InetAddress _v4LocalAddress,_v6LocalAddress;
 	PhySocket *_v4UdpSocket;
@@ -1374,6 +1488,13 @@ private:
 	UPNPClient *_upnpClient;
 #endif
 
+#ifdef ZT_ENABLE_CLUSTER
+	PhySocket *_clusterMessageSocket;
+	ClusterGeoIpService *_clusterGeoIpService;
+	ClusterDefinition *_clusterDefinition;
+	unsigned int _clusterMemberId;
+#endif
+
 	bool _run;
 	Mutex _run_m;
 };
@@ -1386,11 +1507,26 @@ static long SnodeDataStoreGetFunction(ZT_Node *node,void *uptr,const char *name,
 { return reinterpret_cast<OneServiceImpl *>(uptr)->nodeDataStoreGetFunction(name,buf,bufSize,readIndex,totalSize); }
 static int SnodeDataStorePutFunction(ZT_Node *node,void *uptr,const char *name,const void *data,unsigned long len,int secure)
 { return reinterpret_cast<OneServiceImpl *>(uptr)->nodeDataStorePutFunction(name,data,len,secure); }
-static int SnodeWirePacketSendFunction(ZT_Node *node,void *uptr,const struct sockaddr_storage *localAddr,const struct sockaddr_storage *addr,const void *data,unsigned int len)
-{ return reinterpret_cast<OneServiceImpl *>(uptr)->nodeWirePacketSendFunction(localAddr,addr,data,len); }
+static int SnodeWirePacketSendFunction(ZT_Node *node,void *uptr,const struct sockaddr_storage *localAddr,const struct sockaddr_storage *addr,const void *data,unsigned int len,unsigned int ttl)
+{ return reinterpret_cast<OneServiceImpl *>(uptr)->nodeWirePacketSendFunction(localAddr,addr,data,len,ttl); }
 static void SnodeVirtualNetworkFrameFunction(ZT_Node *node,void *uptr,uint64_t nwid,uint64_t sourceMac,uint64_t destMac,unsigned int etherType,unsigned int vlanId,const void *data,unsigned int len)
 { reinterpret_cast<OneServiceImpl *>(uptr)->nodeVirtualNetworkFrameFunction(nwid,sourceMac,destMac,etherType,vlanId,data,len); }
 
+#ifdef ZT_ENABLE_CLUSTER
+static void SclusterSendFunction(void *uptr,unsigned int toMemberId,const void *data,unsigned int len)
+{
+	OneServiceImpl *const impl = reinterpret_cast<OneServiceImpl *>(uptr);
+	const ClusterDefinition::MemberDefinition &md = (*(impl->_clusterDefinition))[toMemberId];
+	if (md.clusterEndpoint)
+		impl->_phy.udpSend(impl->_clusterMessageSocket,reinterpret_cast<const struct sockaddr *>(&(md.clusterEndpoint)),data,len);
+}
+static int SclusterGeoIpFunction(void *uptr,const struct sockaddr_storage *addr,int *x,int *y,int *z)
+{
+	OneServiceImpl *const impl = reinterpret_cast<OneServiceImpl *>(uptr);
+	return (int)(impl->_clusterGeoIpService->locate(*(reinterpret_cast<const InetAddress *>(addr)),*x,*y,*z));
+}
+#endif
+
 static void StapFrameHandler(void *uptr,uint64_t nwid,const MAC &from,const MAC &to,unsigned int etherType,unsigned int vlanId,const void *data,unsigned int len)
 { reinterpret_cast<OneServiceImpl *>(uptr)->tapFrameHandler(nwid,from,to,etherType,vlanId,data,len); }
 
@@ -1539,7 +1675,7 @@ std::string OneService::autoUpdateUrl()
 	return std::string();
 }
 
-OneService *OneService::newInstance(const char *hp,unsigned int port,const char *overrideRootTopology) { return new OneServiceImpl(hp,port,overrideRootTopology); }
+OneService *OneService::newInstance(const char *hp,unsigned int port) { return new OneServiceImpl(hp,port); }
 OneService::~OneService() {}
 
 } // namespace ZeroTier

service/OneService.hpp

@@ -43,6 +43,9 @@ namespace ZeroTier {
  * periodically checked and updates are automatically downloaded, verified
  * against a built-in list of update signing keys, and installed. This is
  * only supported for certain platforms.
+ *
+ * If built with ZT_ENABLE_CLUSTER, a 'cluster' file is checked and if
+ * present is read to determine the identity of other cluster members.
  */
 class OneService
 {
@@ -95,12 +98,10 @@ public:
 	 *
 	 * @param hp Home path
 	 * @param port TCP and UDP port for packets and HTTP control (if 0, pick random port)
-	 * @param overrideRootTopology String-serialized root topology (for testing, default: NULL)
 	 */
 	static OneService *newInstance(
 		const char *hp,
-		unsigned int port,
-		const char *overrideRootTopology = (const char *)0);
+		unsigned int port);
 
 	virtual ~OneService();