valib/bhshmq.git

			@@ -1,5 +1,6 @@
			#include "robust.h"
			#include "util.h"
			#include <boost/circular_buffer.hpp>

			using namespace robust;

			@@ -33,37 +34,32 @@
			BOOST_CHECK_EQUAL((u64 & 255), i);
			}

			#if 1
			typedef AtomicQueue<3> Rcb;
			uint64_t correct_total = nmsg * (nmsg - 1) / 2;
			std::atomic<uint64_t> total(0);
			std::atomic<uint64_t> nwrite(0);
			std::atomic<uint64_t> writedone(0);

			#if 0
			typedef AtomicQueue<4> Rcb;

			Rcb tmp;
			BOOST_CHECK(tmp.like_empty());
			BOOST_CHECK(tmp.push_back(1));
			BOOST_CHECK(tmp.push(1));
			BOOST_CHECK(tmp.tail() == 1);
			BOOST_CHECK(tmp.head() == 0);
			int64_t d;
			BOOST_CHECK(tmp.pop_front(d));
			BOOST_CHECK(tmp.pop(d));
			BOOST_CHECK(tmp.like_empty());
			BOOST_CHECK(tmp.head() == 1);
			BOOST_CHECK(tmp.tail() == 1);

			ShmObject<Rcb> rcb(shm, "test_rcb");
			#else
			typedef Circular<int64_t> Rcb;
			ShmObject<Rcb> rcb(shm, "test_rcb", 64, shm.get_segment_manager());
			#endif
			bool try_more = true;

			const int nsize = sizeof(Rcb);

			bool try_more = false;
			uint64_t correct_total = nmsg * (nmsg - 1) / 2;
			std::atomic<uint64_t> total(0);
			std::atomic<uint64_t> nwrite(0);
			std::atomic<uint64_t> writedone(0);
			auto Writer = [&]() {
			uint64_t n = 0;
			while ((n = nwrite++) < nmsg) {
			while (!rcb->push_back(n, try_more)) {
			while (!rcb->push(n, try_more)) {
			// MySleep();
			}
			++writedone;
			@@ -73,7 +69,7 @@
			auto Reader = [&]() {
			while (nread.load() < nmsg) {
			int64_t d;
			if (rcb->pop_front(d, try_more)) {
			if (rcb->pop(d, try_more)) {
			++nread;
			total += d;
			} else {
			@@ -81,6 +77,58 @@
			}
			}
			};

			#else
			typedef Circular<int64_t> Rcb;
			ShmObject<Rcb> rcb(shm, "test_rcb", 16, shm.get_segment_manager());

			typedef FMutex Mutex;
			// typedef SemMutex Mutex;
			Mutex mtx(123);
			auto Writer = [&]() {
			uint64_t n = 0;
			while ((n = nwrite++) < nmsg) {
			auto Write = [&]() {
			robust::Guard<Mutex> lk(mtx);
			if (rcb->full()) {
			return false;
			} else {
			rcb->push_back(n);
			return true;
			}
			// return rcb->push_back(n);
			};
			while (!Write()) {
			// MySleep();
			}
			++writedone;
			}
			};
			std::atomic<uint64_t> nread(0);
			auto Reader = [&]() {
			while (nread.load() < nmsg) {
			int64_t d;
			auto Read = [&]() {
			robust::Guard<Mutex> lk(mtx);
			if (rcb->empty()) {
			return false;
			} else {
			d = rcb->front();
			rcb->pop_front();
			return true;
			}
			// return rcb->pop_front(d);
			};
			if (Read()) {
			++nread;
			total += d;
			} else {
			// MySleep();
			}
			}
			};

			#endif

			auto status = [&]() {
			auto next = steady_clock::now();
			@@ -102,7 +150,8 @@
			{
			ThreadManager threads;
			boost::timer::auto_cpu_timer timer;
			printf("Testing Robust Buffer, msgs %ld, queue size: %d, threads: %d \n", nmsg, Rcb::capacity, nthread);
			// printf("Testing Robust Buffer, msgs %ld, queue size: %d, threads: %d \n", nmsg, Rcb::capacity, nthread);
			printf("Testing Robust Buffer, msgs %ld, queue size: %d, threads: %d \n", nmsg, 16, nthread);
			for (int i = 0; i < nthread; ++i) {
			threads.Launch(Reader);
			threads.Launch(Writer);
			@@ -116,7 +165,24 @@

			BOOST_AUTO_TEST_CASE(MutexTest)
			{
			typedef robust::Mutex RobustMutex;
			{
			int fd = open("/tmp/test_fmutex", O_CREAT \| O_RDONLY, 0666);
			flock(fd, LOCK_EX);
			printf("lock 1");
			Sleep(10s);
			flock(fd, LOCK_EX);
			printf("lock 2");
			Sleep(10s);
			flock(fd, LOCK_UN);
			printf("un lock 2");
			Sleep(10s);
			flock(fd, LOCK_UN);
			printf("un lock 1");
			return;
			}

			// typedef robust::MFMutex RobustMutex;
			typedef robust::SemMutex RobustMutex;

			for (int i = 0; i < 20; ++i) {
			int size = i;
			@@ -131,7 +197,9 @@

			const std::string mtx_name("test_mutex");
			const std::string int_name("test_int");
			auto mtx = shm.FindOrCreate<RobustMutex>(mtx_name);
			// auto mtx = shm.FindOrCreate<RobustMutex>(mtx_name, 12345);
			RobustMutex rmtx(12345);
			auto mtx = &rmtx;
			auto pi = shm.FindOrCreate<int>(int_name, 100);

			std::mutex m;
			@@ -142,29 +210,48 @@
			printf("int : %d, add1: %d\n", old, ++*pi);
			}

			{
			const int ntimes = 1000 * 1000;
			RobustMutex mutex;
			auto LockSpeed = [](auto &mutex, const std::string &name) {
			const int ntimes = 1000 * 1;
			auto Lock = [&]() {
			for (int i = 0; i < ntimes; ++i) {
			mutex.lock();
			mutex.unlock();
			}
			};

			printf("\nTesting %s lock/unlock %d times\n", name.c_str(), ntimes);
			{
			boost::timer::auto_cpu_timer timer;
			printf("test lock/unlock %d times: ", ntimes);
			printf("1 thread: ");
			Lock();
			}
			{
			auto InThread = [&](int nthread) {
			boost::timer::auto_cpu_timer timer;
			printf("test lock/unlock %d times, 2 thread: ", ntimes);
			std::thread t1(Lock), t2(Lock);
			t1.join();
			t2.join();
			}
			}
			printf("%d threads: ", nthread);
			std::vector<std::thread> vt;
			for (int i = 0; i < nthread; ++i) {
			vt.emplace_back(Lock);
			}
			for (auto &t : vt) {
			t.join();
			}
			};
			InThread(4);
			InThread(16);
			InThread(100);
			InThread(1000);
			};
			NullMutex null_mtx;
			std::mutex std_mtx;
			CasMutex cas_mtx;
			FMutex mfmtx(3);
			boost::interprocess::interprocess_mutex ipc_mutex;
			SemMutex sem_mtx(3);
			LockSpeed(null_mtx, "null mutex");
			LockSpeed(std_mtx, "std::mutex");
			// LockSpeed(cas_mtx, "CAS mutex");
			LockSpeed(ipc_mutex, "boost ipc mutex");
			LockSpeed(mfmtx, "mutex+flock");
			LockSpeed(sem_mtx, "sem mutex");

			auto TryLock = [&]() {
			if (mtx->try_lock()) {
			@@ -183,6 +270,7 @@
			if (mtx) {
			printf("mtx exists\n");
			if (TryLock()) {
			// Sleep(10s);
			auto op = [&]() {
			if (TryLock()) {
			Unlock();