opencog-atomspace / tests / scm / MultiThreadUTest.cxxtest

/*
 * tests/scm/MultiThreadUTest.cxxtest
 *
 * Copyright (C) 2009, 2011, 2014 Linas Vepstas <linasvepstas@gmail.com>
 * All Rights Reserved
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License v3 as
 * published by the Free Software Foundation and including the exceptions
 * at http://opencog.org/wiki/Licenses
 *
 * 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 Affero General Public License
 * along with this program; if not, write to:
 * Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */
#include <sys/syscall.h>

#include <opencog/atomspace/AtomSpace.h>
#include <opencog/guile/SchemeEval.h>
#include <opencog/util/Logger.h>

using namespace opencog;

class MultiThreadUTest :  public CxxTest::TestSuite
{
	private:
		AtomSpacePtr as;

	public:

		MultiThreadUTest(void)
		{
			logger().set_level(Logger::DEBUG);
			logger().set_print_to_stdout_flag(true);
		}

		~MultiThreadUTest()
		{
			// erase the log file if no assertions failed
			if (!CxxTest::TestTracker::tracker().suiteFailed())
				std::remove(logger().get_filename().c_str());
		}

		void setUp(void);
		void tearDown(void);

		void test_init_race(void);
		void threadedInit(int thread_id);
		void test_multi_threads(void);
		void threadedAdd(int thread_id, int N);
};

#define an as->add_node
#define al as->add_link
#define CHKEV(ev) \
	TSM_ASSERT("Caught scm error during eval", \
		(false == ev->eval_error()));

// static pid_t gettid() { return syscall(SYS_gettid); }

void MultiThreadUTest::setUp(void)
{
}

void MultiThreadUTest::tearDown(void)
{
}

/*
 * Test multiple threads, all performing initialization at the same
 * time.  Initialization should happen once, and there should be no
 * corruption.
 */
static volatile bool hold = true;
static std::atomic_int start_cnt(0);

void MultiThreadUTest::test_init_race(void)
{
	logger().debug("BEGIN TEST: %s", __FUNCTION__);
	as = createAtomSpace();
	std::vector<std::thread> thread_pool;

	int n_threads = 120;
	start_cnt = 0;
	hold = true;
	for (int i=0; i < n_threads; i++) {
		thread_pool.push_back(
			std::thread(&MultiThreadUTest::threadedInit, this, i));
	}
	while (start_cnt != n_threads) {}  // spin
	printf("Done creating %d threads\n", n_threads);
	hold = false;

	for (std::thread& t : thread_pool) t.join();

	printf("Done joining %d threads\n", n_threads);
	logger().debug("END TEST: %s", __FUNCTION__);
}

void MultiThreadUTest::threadedInit(int thread_id)
{
	start_cnt ++;
	while (hold) {} // spin

	SchemeEval* ev = SchemeEval::get_scheme_evaluator(as);

	Handle h1 = ev->eval_h("(cog-new-node 'ConceptNode \"stuff\")");
	CHKEV(ev);
	TSM_ASSERT("Failed to create atom", Handle::UNDEFINED != h1);
}


// In this thread, create a bunch of atoms.
// Use same atomsapce as all the other threads.
// Use a unique evaluator for this thread.
//
// Use (let ((thing))) instead of (define thing) to avoid having
// races with the defintion of 'thing'. The problem is that the
// (define thing) will define the symbol in the 'guile-user module',
// which will race across threads. The (let) will do thread-local
// creation and binding of the symbol.
void MultiThreadUTest::threadedAdd(int thread_id, int N)
{
	SchemeEval* ev = SchemeEval::get_scheme_evaluator(as);
	int counter = 0;
	for (int i = 0; i < N; i++) {
		double strength = 0.7 * ((double) i) / ((double) N);
		double confidence = 0.8 * ((double) N-i) / ((double) N);

		std::ostringstream oss;
		oss << "(let ((per-thread-node (cog-new-node 'ConceptNode \""
		    << "thread= " << thread_id << " node= " << counter
		    << "\"))) "
		    << "(cog-set-value! per-thread-node (Predicate \"*-TruthValueKey-*\") (FloatValue " << strength << " " << confidence << ")) "
		    << "per-thread-node)";
		counter ++;

		Handle h = ev->eval_h(oss.str());
		CHKEV(ev);
		TSM_ASSERT("Failed to create atom", Handle::UNDEFINED != h);
	}
}

/*
 * Test multiple evaluators writing to a single atomspace, in multiple threads.
 */
void MultiThreadUTest::test_multi_threads(void)
{
	logger().debug("BEGIN TEST: %s", __FUNCTION__);
	as = createAtomSpace();
	as->clear();

	std::vector<std::thread> thread_pool;

	// Twelve threads -- 15K atoms -- should be enough to run maybe 15-30
	// seconds on modern CPUs.
	int n_threads = 12;
	int num_atoms = 15225;
	for (int i=0; i < n_threads; i++) {
		thread_pool.push_back(
			std::thread(&MultiThreadUTest::threadedAdd, this, i, num_atoms));
	}
	for (std::thread& t : thread_pool) t.join();
	size_t size = as->get_size();
	std::cout << "atomspace size:" << size << std::endl;

	// We should get num_atoms * n_threads distinct atoms
	// Plus two: (Predicate "*-TruthValueKey-*") and (Predicate "*-IsKeyFlag-*")
	TS_ASSERT_EQUALS(size, num_atoms * n_threads + 2);

	logger().debug("END TEST: %s", __FUNCTION__);
}