...

testnet/Condition.hpp

@@ -0,0 +1,142 @@
+/*
+ * ZeroTier One - Global Peer to Peer Ethernet
+ * Copyright (C) 2012-2014  ZeroTier Networks LLC
+ *
+ * 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_CONDITION_HPP
+#define ZT_CONDITION_HPP
+
+#include "../node/Constants.hpp"
+#include "../node/NonCopyable.hpp"
+
+#ifdef __WINDOWS__
+
+#include <Windows.h>
+#include <stdlib.h>
+
+namespace ZeroTier {
+
+class Condition : NonCopyable
+{
+public:
+	Condition()
+		throw()
+	{
+		_sem = CreateSemaphore(NULL,0,1,NULL);
+	}
+
+	~Condition()
+	{
+		CloseHandle(_sem);
+	}
+
+	inline void wait() const
+		throw()
+	{
+		WaitForSingleObject(_sem,INFINITE);
+	}
+
+	inline void wait(unsigned long ms) const
+		throw()
+	{
+		if (ms)
+			WaitForSingleObject(_sem,(DWORD)ms);
+		else WaitForSingleObject(_sem,INFINITE);
+	}
+
+	inline void signal() const
+		throw()
+	{
+		ReleaseSemaphore(_sem,1,NULL);
+	}
+
+private:
+	HANDLE _sem;
+};
+
+} // namespace ZeroTier
+
+#else // !__WINDOWS__
+
+#include <time.h>
+#include <stdlib.h>
+#include <pthread.h>
+
+#include "../node/Utils.hpp"
+
+namespace ZeroTier {
+
+class Condition : NonCopyable
+{
+public:
+	Condition()
+		throw()
+	{
+		pthread_mutex_init(&_mh,(const pthread_mutexattr_t *)0);
+		pthread_cond_init(&_cond,(const pthread_condattr_t *)0);
+	}
+
+	~Condition()
+	{
+		pthread_cond_destroy(&_cond);
+		pthread_mutex_destroy(&_mh);
+	}
+
+	inline void wait() const
+		throw()
+	{
+		pthread_mutex_lock(const_cast <pthread_mutex_t *>(&_mh));
+		pthread_cond_wait(const_cast <pthread_cond_t *>(&_cond),const_cast <pthread_mutex_t *>(&_mh));
+		pthread_mutex_unlock(const_cast <pthread_mutex_t *>(&_mh));
+	}
+
+	inline void wait(unsigned long ms) const
+		throw()
+	{
+		uint64_t when = Utils::now() + (uint64_t)ms;
+		struct timespec ts;
+		ts.tv_sec = (unsigned long)(when / 1000);
+		ts.tv_nsec = (unsigned long)(when % 1000) * 1000000;
+		pthread_mutex_lock(const_cast <pthread_mutex_t *>(&_mh));
+		pthread_cond_timedwait(const_cast <pthread_cond_t *>(&_cond),const_cast <pthread_mutex_t *>(&_mh),&ts);
+		pthread_mutex_unlock(const_cast <pthread_mutex_t *>(&_mh));
+	}
+
+	inline void signal() const
+		throw()
+	{
+		pthread_cond_signal(const_cast <pthread_cond_t *>(&_cond));
+	}
+
+private:
+	pthread_cond_t _cond;
+	pthread_mutex_t _mh;
+};
+
+} // namespace ZeroTier
+
+#endif // !__WINDOWS__
+
+#endif

testnet/SimNetSocketManager.hpp

@@ -35,7 +35,7 @@
 #include "../node/Constants.hpp"
 #include "../node/SocketManager.hpp"
 #include "../node/Mutex.hpp"
-#include "../node/Condition.hpp"
+#include "Condition.hpp"
 
 namespace ZeroTier {
 
@@ -96,10 +96,8 @@ public:
 	 */
 	inline void enqueue(const InetAddress &from,const void *data,unsigned int len)
 	{
-		{
-			Mutex::Lock _l(_inbox_m);
-			_inbox.push_back(std::pair< InetAddress,Buffer<ZT_SOCKET_MAX_MESSAGE_LEN> >(from,Buffer<ZT_SOCKET_MAX_MESSAGE_LEN>(data,len)));
-		}
+		Mutex::Lock _l(_inbox_m);
+		_inbox.push_back(std::pair< InetAddress,Buffer<ZT_SOCKET_MAX_MESSAGE_LEN> >(from,Buffer<ZT_SOCKET_MAX_MESSAGE_LEN>(data,len)));
 		_waitCond.signal();
 	}
 

testnet/TestEthernetTap.cpp

@@ -80,9 +80,9 @@ TestEthernetTap::~TestEthernetTap()
 	static const TestFrame zf;
 	{
 		Mutex::Lock _l(_pq_m);
-		_pq.push(zf); // 0 length frame = exit
+		_pq.push_back(zf); // 0 length frame = exit
+		_pq_c.signal();
 	}
-	_pq_c.signal();
 	Thread::join(_thread);
 }
 
@@ -138,9 +138,9 @@ bool TestEthernetTap::injectPacketFromHost(const MAC &from,const MAC &to,unsigne
 
 	{
 		Mutex::Lock _l(_pq_m);
-		_pq.push(TestFrame(from,to,data,etherType & 0xffff,len));
+		_pq.push_back(TestFrame(from,to,data,etherType & 0xffff,len));
+		_pq_c.signal();
 	}
-	_pq_c.signal();
 
 	return true;
 }
@@ -148,23 +148,25 @@ bool TestEthernetTap::injectPacketFromHost(const MAC &from,const MAC &to,unsigne
 void TestEthernetTap::threadMain()
 	throw()
 {
-	TestFrame tf;
+	std::vector<TestFrame> q;
 	for(;;) {
-		tf.len = 0;
 		{
 			Mutex::Lock _l(_pq_m);
-			if (!_pq.empty()) {
-				if (_pq.front().len == 0)
-					break;
-				memcpy(&tf,&(_pq.front()),sizeof(tf));
-				_pq.pop();
+			q = _pq;
+			_pq.clear();
+		}
+
+		for(std::vector<TestFrame>::iterator f(q.begin());f!=q.end();++f) {
+			if (!f->len)
+				return; // empty frame signals thread to die
+			else if (_enabled) {
+				try {
+					_handler(_arg,f->from,f->to,f->etherType,Buffer<4096>(f->data,f->len));
+				} catch ( ... ) {} // handlers should not throw
 			}
 		}
 
-		if ((tf.len > 0)&&(_enabled))
-			_handler(_arg,tf.from,tf.to,tf.etherType,Buffer<4096>(tf.data,tf.len));
-
-		_pq_c.wait();
+		_pq_c.wait(1000);
 	}
 }
 

testnet/TestEthernetTap.hpp

@@ -32,8 +32,6 @@
 #include <stdlib.h>
 #include <string.h>
 
-#include <stdexcept>
-#include <queue>
 #include <string>
 
 #include "../node/Constants.hpp"
@@ -42,7 +40,7 @@
 #include "../node/SharedPtr.hpp"
 #include "../node/Thread.hpp"
 #include "../node/Mutex.hpp"
-#include "../node/Condition.hpp"
+#include "Condition.hpp"
 
 namespace ZeroTier {
 
@@ -109,12 +107,11 @@ public:
 	inline uint64_t nwid() const { return _nwid; }
 
 	// Get things that have been put() and empty queue
-	inline void get(std::vector<TestFrame> &v,bool clearQueue = true)
+	inline void get(std::vector<TestFrame> &v)
 	{
 		Mutex::Lock _l(_gq_m);
 		v = _gq;
-		if (clearQueue)
-			_gq.clear();
+		_gq.clear();
 	}
 
 	void threadMain()
@@ -130,7 +127,7 @@ private:
 	std::string _dev;
 	volatile bool _enabled;
 
-	std::queue< TestFrame > _pq;
+	std::vector< TestFrame > _pq;
 	Mutex _pq_m;
 	Condition _pq_c;