Merge lp:~gpr/dcplusplus/sync into lp:dcplusplus

[Gennady Proskurin]

+ includes all patches submitted in corresponding bug reports
+ Thread::safeDec/Inc/Exchange functions are replaced by more fine-grained/lightweight/portable implementations

[Jacek Sieka (arnetheduck)]

I'm not quite sure I understand why one of the counters uses interprocess::detail and the other shared_ptr::detail - if we're relying on internals, I'd minimise their use.

Also, I do not understand the comment on weak vs strong ordering, could you elaborate please.

Regarding fastcriticalsection, it's a spin lock which is quite appropriate if the sections locked are (very) short (small risk of contention) as OS mutexes add a lot of overhead...

review: Needs Information

[Gennady Proskurin (gpr)]

About weak/strong ordering, much information is available in internet, for example:
http://en.wikipedia.org/wiki/Memory_ordering
http://en.wikipedia.org/wiki/Memory_barrier
http://msdn.microsoft.com/en-us/library/ms686355(VS.85).aspx

In short, "strong" ordering garantees that all threads in all processors see identical synchronized view of memory, because synchronization primitives (mutex in case of FastCriticalSection in Atomic<strong>) do all necessary memory barriers (synchronization of memory between all processors).

In "weak" ordering (which is used for thread-safe counters in my code), you cannot do any action, based on counter's value. For example, you cannot write code like "if (counter==0) do_some_multithread_action()" because only counter value is synchronized (during atomic access to it), and other memory content may be stale. This is suitable for accounting of some statistics for example.

For smart pointer counters, it is necessary to synchronize memory only in case when counter reaches zero, when destructor is called. This is exactly what shared_ptr counters do. So I used shared_ptr/atomic for intrusive_ptr refcount, which was designed exactly for this purpose - thread-safe refcounting in smart pointers.

But shared_ptr/atomic does not have assignment semantics, only construction. So for Atomic<weak> I used another atomic from boost - interprocess::detail. My Atomic template was designed to be general-purpose thread-safe counter, so it should have assignment.

[Gennady Proskurin (gpr)]

My motivation for writing all this patches (besides bug in using pthread_cond) was bad (coarse) locking for unix. It uses one static mutex in Thread class for almost all synchronization in the whole code.

While I agree that spinning is suitable for FastCriticalSection, it's not easy to implement it properly and effective. At first glance, current implementation may be subject to priority inversion, and therefore, deadlocks.

I'm not sure, what solution will be acceptable for both unix and windows. For intrusive_ptr, I'd just use boost shared_ptr/atomic (as in my patch). For critical sections, I think correctness is priority, and then optimization. And portability is also good.

Unmerged revisions

2216. By Gennady Proskurin on 2010-08-17

Change BufferedSocket::sockets to Atomic (strong variant).

2215. By Gennady Proskurin on 2010-08-17

Use Atomic<bool> for "refreshing" variable.

2214. By Gennady Proskurin on 2010-08-17

Implement Atomic template for atomic counters.

2213. By Gennady Proskurin on 2010-08-17

Portable FastCriticalSection implementation.

=== added file 'dcpp/Atomic.h'
--- dcpp/Atomic.h	1970-01-01 00:00:00 +0000
+++ dcpp/Atomic.h	2010-08-17 18:57:44 +0000
@@ -0,0 +1,122 @@
+#if !defined(DCPP_ATOMIC_H)
+#define DCPP_ATOMIC_H
+
+#include "CriticalSection.h"
+
+#include <boost/interprocess/detail/atomic.hpp>
+#include <boost/cstdint.hpp>
+
+namespace dcpp {
+
+// Ordering arguments:
+
+// memory_ordering_weak
+// Suitable only for thread-safe accounting of some statistics.
+// Value can not be used as "flag" (you cannot do any multi-thread action, based
+// on this value) since it does not garantees necessary memory barriers.
+class memory_ordering_weak {};
+
+// memory_ordering_strong
+// Suitable for any multi-thread purpose
+class memory_ordering_strong {};
+
+template <typename DataType, class Ordering = memory_ordering_strong>
+class Atomic;
+
+
+// uint32_t
+template <>
+class Atomic<boost::uint32_t, memory_ordering_weak> {
+	typedef boost::uint32_t value_type;
+public:
+	Atomic(value_type val) { assign(val); }
+	Atomic(const Atomic& other) { assign(static_cast<value_type>(other)); }
+
+	// operator=
+	// return void to be safe
+	void operator=(value_type val) { assign(val); }
+	void operator=(const Atomic& other) {
+		return operator=(static_cast<value_type>(other));
+	}
+
+	// type cast
+	operator value_type() const {
+		return boost::interprocess::detail::atomic_read32(&m_value);
+	}
+
+	// increment
+	void inc() { boost::interprocess::detail::atomic_inc32(&m_value); }
+
+	// decrement
+	void dec() { boost::interprocess::detail::atomic_dec32(&m_value); }
+
+private:
+	mutable value_type m_value;
+	void assign(value_type val) { boost::interprocess::detail::atomic_write32(&m_value, val); }
+};
+
+// int32_t
+// just forward all operations to underlying Atomic<uint32_t, ...> variable
+template <>
+class Atomic<boost::int32_t, memory_ordering_weak> {
+	typedef boost::int32_t value_type;
+public:
+	Atomic(value_type val) : m_value(val) {}
+	Atomic(const Atomic& other) : m_value(other) {}
+
+	void operator=(value_type val)		{ m_value=val; }
+	void operator=(const Atomic& other)	{ m_value=other; }
+	operator value_type() const		{ return static_cast<value_type>(m_value); }
+
+	void inc() { m_value.inc(); }
+	void dec() { m_value.dec(); }
+private:
+	Atomic<boost::uint32_t,memory_ordering_weak> m_value;
+};
+
+// memory_ordering_strong
+template <typename DataType>
+class Atomic<DataType, memory_ordering_strong> {
+	typedef DataType value_type;
+public:
+	Atomic(value_type new_value) : m_value(new_value) {}
+	Atomic(const Atomic& other) : m_value(static_cast<value_type>(other)) {}
+
+	void operator=(value_type new_value) {
+		FastLock Lock(cs);
+		m_value = new_value;
+	}
+	void operator=(const Atomic& other) {
+		FastLock Lock(cs);
+		m_value = other;
+	}
+	operator value_type() const {
+		FastLock Lock(cs); // shared (read-only) lock would be sufficient here
+		return m_value;
+	}
+
+	void inc() {
+		FastLock Lock(cs);
+		++m_value;
+	}
+	void dec() {
+		FastLock Lock(cs);
+		--m_value;
+	}
+
+	// assign new value, return old value
+	value_type exchange(value_type new_val) {
+		FastLock Lock(cs);
+		value_type old_val = m_value;
+		m_value = new_val;
+		return old_val;
+	}
+private:
+	value_type m_value;
+	mutable FastCriticalSection cs;
+};
+
+
+} // namespace dcpp
+
+#endif // !defined(DCPP_ATOMIC_H)

=== modified file 'dcpp/BufferedSocket.cpp'
--- dcpp/BufferedSocket.cpp	2010-07-10 14:36:48 +0000
+++ dcpp/BufferedSocket.cpp	2010-08-17 18:57:44 +0000
@@ -42,13 +42,13 @@
 {
 	start();
 
-	Thread::safeInc(sockets);
+	sockets.inc();
 }
 
-volatile long BufferedSocket::sockets = 0;
+Atomic<long,memory_ordering_strong> BufferedSocket::sockets(0);
 
 BufferedSocket::~BufferedSocket() throw() {
-	Thread::safeDec(sockets);
+	sockets.dec();
 }
 
 void BufferedSocket::setMode (Modes aMode, size_t aRollback) {

=== modified file 'dcpp/BufferedSocket.h'
--- dcpp/BufferedSocket.h	2010-07-10 14:36:48 +0000
+++ dcpp/BufferedSocket.h	2010-08-17 18:57:44 +0000
@@ -26,6 +26,7 @@
 #include "Speaker.h"
 #include "Util.h"
 #include "Socket.h"
+#include "Atomic.h"
 
 namespace dcpp {
 
@@ -163,7 +164,10 @@
 	void threadSendData() throw(Exception);
 
 	void fail(const string& aError);
-	static volatile long sockets;
+
+	// For this counter we must use "strong" variant of Atomic
+	// We do some actions after checking this value, while it changes in other threads
+	static Atomic<long,memory_ordering_strong> sockets;
 
 	bool checkEvents() throw(Exception);
 	void checkSocket() throw(Exception);

=== modified file 'dcpp/Client.cpp'
--- dcpp/Client.cpp	2010-02-11 21:44:13 +0000
+++ dcpp/Client.cpp	2010-08-17 18:57:44 +0000
@@ -158,24 +158,24 @@
 void Client::updateCounts(bool aRemove) {
 	// We always remove the count and then add the correct one if requested...
 	if(countType == COUNT_NORMAL) {
-		Thread::safeDec(counts.normal);
+		counts.normal.dec();
 	} else if(countType == COUNT_REGISTERED) {
-		Thread::safeDec(counts.registered);
+		counts.registered.dec();
 	} else if(countType == COUNT_OP) {
-		Thread::safeDec(counts.op);
+		counts.op.dec();
 	}
 
 	countType = COUNT_UNCOUNTED;
 
 	if(!aRemove) {
 		if(getMyIdentity().isOp()) {
-			Thread::safeInc(counts.op);
+			counts.op.inc();
 			countType = COUNT_OP;
 		} else if(getMyIdentity().isRegistered()) {
-			Thread::safeInc(counts.registered);
+			counts.registered.inc();
 			countType = COUNT_REGISTERED;
 		} else {
-			Thread::safeInc(counts.normal);
+			counts.normal.inc();
 			countType = COUNT_NORMAL;
 		}
 	}

=== modified file 'dcpp/Client.h'
--- dcpp/Client.h	2010-06-30 14:40:00 +0000
+++ dcpp/Client.h	2010-08-17 18:57:44 +0000
@@ -26,6 +26,7 @@
 #include "BufferedSocketListener.h"
 #include "TimerManager.h"
 #include "ClientListener.h"
+#include "Atomic.h"
 
 namespace dcpp {
 
@@ -72,7 +73,10 @@
 
 	static string getCounts() {
 		char buf[128];
-		return string(buf, snprintf(buf, sizeof(buf), "%ld/%ld/%ld", counts.normal, counts.registered, counts.op));
+		return string(buf, snprintf(buf, sizeof(buf), "%ld/%ld/%ld",
+			static_cast<long>(counts.normal),
+			static_cast<long>(counts.registered),
+			static_cast<long>(counts.op)));
 	}
 
 	StringMap& escapeParams(StringMap& sm) {
@@ -112,12 +116,16 @@
 	friend class ClientManager;
 	Client(const string& hubURL, char separator, bool secure_);
 	virtual ~Client() throw();
+
 	struct Counts {
-		Counts(long n = 0, long r = 0, long o = 0) : normal(n), registered(r), op(o) { }
-		volatile long normal;
-		volatile long registered;
-		volatile long op;
-		bool operator !=(const Counts& rhs) { return normal != rhs.normal || registered != rhs.registered || op != rhs.op; }
+	private:
+		typedef Atomic<boost::int32_t,memory_ordering_weak> atomic_counter_t;
+	public:
+		typedef boost::int32_t value_type;
+		Counts(value_type n = 0, value_type r = 0, value_type o = 0) : normal(n), registered(r), op(o) { }
+		atomic_counter_t normal;
+		atomic_counter_t registered;
+		atomic_counter_t op;
 	};
 
 	enum States {

=== modified file 'dcpp/CriticalSection.h'
--- dcpp/CriticalSection.h	2010-04-23 18:32:11 +0000
+++ dcpp/CriticalSection.h	2010-08-17 18:57:44 +0000
@@ -19,7 +19,8 @@
 #if !defined(CRITICAL_SECTION_H)
 #define CRITICAL_SECTION_H
 
-#include "Thread.h"
+// header-only implementation of mutex
+#include <boost/signals2/mutex.hpp>
 
 namespace dcpp {
 
@@ -78,32 +79,11 @@
  */
 class FastCriticalSection {
 public:
-#ifdef _WIN32
-	FastCriticalSection() : state(0) { }
-
-	void enter() {
-		while(Thread::safeExchange(state, 1) == 1) {
-			Thread::sleep(1);
-		}
-	}
-	void leave() {
-		Thread::safeDec(state);
-	}
-private:
-	volatile long state;
-
-#else
-	// We have to use a pthread (nonrecursive) mutex, didn't find any test_and_set on linux...
-	FastCriticalSection() {
-		static pthread_mutex_t fastmtx = PTHREAD_MUTEX_INITIALIZER;
-		mtx = fastmtx;
-	}
-	~FastCriticalSection() { pthread_mutex_destroy(&mtx); }
-	void enter() { pthread_mutex_lock(&mtx); }
-	void leave() { pthread_mutex_unlock(&mtx); }
-private:
-	pthread_mutex_t mtx;
-#endif
+	void enter() { mtx.lock(); }
+	void leave() { mtx.unlock(); }
+private:
+	typedef boost::signals2::mutex mutex_t;
+	mutex_t mtx;
 };
 
 template<class T>

=== modified file 'dcpp/Pointer.h'
--- dcpp/Pointer.h	2010-02-11 21:44:13 +0000
+++ dcpp/Pointer.h	2010-08-17 18:57:44 +0000
@@ -20,7 +20,7 @@
 #define DCPLUSPLUS_DCPP_POINTER_H
 
 #include <boost/intrusive_ptr.hpp>
-#include "Thread.h"
+#include <boost/smart_ptr/detail/atomic_count.hpp>
 
 namespace dcpp {
 
@@ -36,10 +36,10 @@
 	intrusive_ptr_base() throw() : ref(0) { }
 
 private:
-	friend void intrusive_ptr_add_ref(intrusive_ptr_base* p) { Thread::safeInc(p->ref); }
-	friend void intrusive_ptr_release(intrusive_ptr_base* p) { if(Thread::safeDec(p->ref) == 0) { delete static_cast<T*>(p); } }
+	friend void intrusive_ptr_add_ref(intrusive_ptr_base* p) { ++p->ref; }
+	friend void intrusive_ptr_release(intrusive_ptr_base* p) { if(--p->ref == 0) { delete static_cast<T*>(p); } }
 
-	volatile long ref;
+	boost::detail::atomic_count ref;
 };
 
 

=== modified file 'dcpp/Semaphore.h'
--- dcpp/Semaphore.h	2010-02-11 21:44:13 +0000
+++ dcpp/Semaphore.h	2010-08-17 18:57:44 +0000
@@ -59,7 +59,7 @@
 
 	bool wait() throw() {
 		Lock l(cs);
-		if(count == 0) {
+		while (count == 0) {
 			pthread_cond_wait(&cond, &cs.getMutex());
 		}
 		count--;
@@ -74,7 +74,10 @@
 			millis+=timev.tv_usec/1000;
 			t.tv_sec = timev.tv_sec + (millis/1000);
 			t.tv_nsec = (millis%1000)*1000*1000;
-			int ret = pthread_cond_timedwait(&cond, &cs.getMutex(), &t);
+			int ret;
+			do {
+				ret = pthread_cond_timedwait(&cond, &cs.getMutex(), &t);
+			} while (ret==0 && count==0);
 			if(ret != 0) {
 				return false;
 			}

=== modified file 'dcpp/ShareManager.cpp'
--- dcpp/ShareManager.cpp	2010-07-10 14:36:48 +0000
+++ dcpp/ShareManager.cpp	2010-08-17 18:57:44 +0000
@@ -52,7 +52,7 @@
 namespace dcpp {
 
 ShareManager::ShareManager() : hits(0), xmlListLen(0), bzXmlListLen(0),
-	xmlDirty(true), forceXmlRefresh(false), refreshDirs(false), update(false), initial(true), listN(0), refreshing(0),
+	xmlDirty(true), forceXmlRefresh(false), refreshDirs(false), update(false), initial(true), listN(0), refreshing(false),
 	lastXmlUpdate(0), lastFullUpdate(GET_TICK()), bloom(1<<20)
 {
 	SettingsManager::getInstance()->addListener(this);
@@ -723,7 +723,7 @@
 }
 
 void ShareManager::refresh(bool dirs /* = false */, bool aUpdate /* = true */, bool block /* = false */) throw() {
-	if(Thread::safeExchange(refreshing, 1) == 1) {
+	if(refreshing.exchange(true) == true) {
 		LogManager::getInstance()->message(_("File list refresh in progress, please wait for it to finish before trying to refresh again"));
 		return;
 	}
@@ -798,7 +798,7 @@
 	if(update) {
 		ClientManager::getInstance()->infoUpdated();
 	}
-	refreshing = 0;
+	refreshing = false;
 	return 0;
 }
 

=== modified file 'dcpp/ShareManager.h'
--- dcpp/ShareManager.h	2010-07-10 14:36:48 +0000
+++ dcpp/ShareManager.h	2010-08-17 18:57:44 +0000
@@ -33,6 +33,7 @@
 #include "FastAlloc.h"
 #include "MerkleTree.h"
 #include "Pointer.h"
+#include "Atomic.h"
 
 namespace dcpp {
 
@@ -251,7 +252,7 @@
 
 	int listN;
 
-	volatile long refreshing;
+	Atomic<bool,memory_ordering_strong> refreshing;
 
 	uint64_t lastXmlUpdate;
 	uint64_t lastFullUpdate;