valib/bhshmq.git

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

			using namespace robust;

			@@ -8,60 +9,79 @@
			eLockerMask = MaskBits(sizeof(int) * 8),
			};

			typedef CircularBuffer<int64_t, Allocator<int64_t>> Rcb;
			Rcb *GetRCB(SharedMemory &shm, const int nelem)
			{
			int cap = nelem + 1;
			typedef uint64_t Data;
			auto size = sizeof(Rcb) + sizeof(Data) * cap;
			void *p = shm.Alloc(size);
			if (p) {
			return new (p) Rcb(cap, shm.get_segment_manager());
			}
			return nullptr;
			}

			void MySleep()
			{
			std::this_thread::sleep_for(2us);
			}

			/////////////////////////////////////////////////////////////////////////////////////////

			BOOST_AUTO_TEST_CASE(InitTest)
			{
			AtomicReqRep rr;
			auto client = [&]() {
			for (int i = 0; i < 5; ++i) {
			int64_t reply = 0;
			bool r = rr.ClientRequest(i, reply);
			printf("init request %d, %s, reply %d\n", i, (r ? "ok" : "failed"), reply);
			}
			};

			bool run = true;
			auto server = [&]() {
			auto onReq = [](int64_t req) { return req + 100; };
			while (run) {
			rr.ServerProcess(onReq);
			}
			};

			ThreadManager clients, servers;
			servers.Launch(server);
			for (int i = 0; i < 2; ++i) {
			clients.Launch(client);
			}
			clients.WaitAll();
			run = false;
			servers.WaitAll();
			}

			BOOST_AUTO_TEST_CASE(QueueTest)
			{
			const int nthread = 100;
			const uint64_t nmsg = 1000 * 1000 * 10;

			SharedMemory &shm = TestShm();
			shm.Remove();
			pid_t pid = getpid();
			printf("pid : %d\n", pid);
			auto Access = [](pid_t pid) {
			char buf[100] = {0};
			sprintf(buf, "/proc/%d/stat", pid);
			int r = access(buf, F_OK);
			printf("access %d\n", r);
			};
			Access(pid);
			Access(pid + 1);
			// Sleep(10s);
			// return;
			// return; /////////////////////////////////////////////////
			int64_t i64 = 0;
			char c = 0;
			for (int i = 0; i < 256; ++i) {
			c = i;
			i64 = int64_t(c) << 1;
			BOOST_CHECK_EQUAL(c, (i64 >> 1));
			uint64_t u64 = i;
			BOOST_CHECK_EQUAL((u64 & 255), i);
			}

			int nelement = 640;
			auto rcb = GetRCB(shm, nelement);
			BOOST_CHECK(rcb != nullptr);
			BOOST_CHECK(rcb->empty());
			BOOST_CHECK(rcb->push_back(1));
			BOOST_CHECK(rcb->size() == 1);
			int64_t d;
			BOOST_CHECK(rcb->pop_front(d));
			BOOST_CHECK(rcb->empty());

			const uint64_t nmsg = 1000 * 1000 * 1;
			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);

			typedef AtomicQ63 Rcb;

			Rcb tmp;
			BOOST_CHECK(tmp.push(1));
			int64_t d;
			BOOST_CHECK(tmp.pop(d));

			NamedShmObject<Rcb> rcb(shm, "test_rcb", eOpenOrCreate);
			bool try_more = true;

			auto Writer = [&]() {
			uint64_t n = 0;
			while ((n = nwrite++) < nmsg) {
			while (!rcb->push_back(n)) {
			while (!rcb->push(n, try_more)) {
			// MySleep();
			}
			++writedone;
			@@ -71,11 +91,11 @@
			auto Reader = [&]() {
			while (nread.load() < nmsg) {
			int64_t d;
			if (rcb->pop_front(d)) {
			if (rcb->pop(d, try_more)) {
			++nread;
			total += d;
			} else {
			MySleep();
			// MySleep();
			}
			}
			};
			@@ -89,105 +109,26 @@
			next += 1s;
			auto w = writedone.load();
			auto r = nread.load();
			printf("write: %6ld, spd: %6ld, read: %6ld, spd: %6ld , queue size: %d\n", w, w - lw, r, r - lr, rcb->size());
			printf("write: %6ld, spd: %6ld, read: %6ld, spd: %6ld\n",
			w, w - lw, r, r - lr);
			lw = w;
			lr = r;
			} while (nread.load() < nmsg);
			};

			ThreadManager threads;
			boost::timer::auto_cpu_timer timer;
			printf("Testing Robust Buffer, msgs %ld, queue size: %d \n", nmsg, nelement);
			threads.Launch(status);
			for (int i = 0; i < 10; ++i) {
			threads.Launch(Reader);
			threads.Launch(Writer);
			std::thread st(status);
			{
			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, 16, nthread);
			for (int i = 0; i < nthread; ++i) {
			threads.Launch(Reader);
			threads.Launch(Writer);
			}
			threads.WaitAll();
			}
			threads.WaitAll();
			st.join();
			printf("total: %ld, expected: %ld\n", total.load(), correct_total);
			BOOST_CHECK_EQUAL(total.load(), correct_total);
			}

			BOOST_AUTO_TEST_CASE(MutexTest)
			{
			typedef robust::Mutex RobustMutex;

			for (int i = 0; i < 20; ++i) {
			int size = i;
			int left = size & 7;
			int rsize = size + ((8 - left) & 7);
			printf("size: %3d, rsize: %3d\n", size, rsize);
			}
			SharedMemory &shm = TestShm();
			// shm.Remove();
			// return;
			GlobalInit(shm);

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

			std::mutex m;
			typedef std::chrono::steady_clock Clock;
			auto Now = []() { return Clock::now().time_since_epoch(); };
			if (pi) {
			auto old = *pi;
			printf("int : %d, add1: %d\n", old, ++*pi);
			}

			{
			boost::timer::auto_cpu_timer timer;
			const int ntimes = 1000 * 1000;
			printf("test lock/unlock %d times: ", ntimes);
			RobustMutex mutex;
			auto Lock = [&]() {
			for (int i = 0; i < ntimes; ++i) {
			mutex.lock();
			mutex.unlock();
			}
			};
			std::thread t1(Lock), t2(Lock);
			t1.join();
			t2.join();
			}

			auto MSFromNow = [](const int ms) {
			using namespace boost::posix_time;
			ptime cur = boost::posix_time::microsec_clock::universal_time();
			return cur + millisec(ms);
			};

			auto TryLock = [&]() {
			if (mtx->try_lock()) {
			printf("try_lock ok\n");
			return true;
			} else {
			printf("try_lock failed\n");
			return false;
			}
			};
			auto Unlock = [&]() {
			mtx->unlock();
			printf("unlocked\n");
			};

			if (mtx) {
			printf("mtx exists\n");
			if (TryLock()) {
			auto op = [&]() {
			if (TryLock()) {
			Unlock();
			}
			};
			op();
			std::thread t(op);
			t.join();
			// Unlock();
			} else {
			// mtx->unlock();
			}
			} else {
			printf("mtx not exists\n");
			}
			}