MultiDeleteUTest.cxxtest

/*
 * tests/persist/rocks/MultiDeleteUTest.cxxtest
 *
 * Verify that racing atom insertion and deletion does not trigger
 * internal errors.
 *
 * Copyright (C) 2008, 2009, 2019, 2021 Linas Vepstas <linasvepstas@gmail.com>
 *
 * LICENSE:
 * SPDX-License-Identifier: AGPL-3.0-or-later
 *
 * 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 <cstdio>
#include <filesystem>
#include <string>

#include <opencog/atoms/base/Atom.h>
#include <opencog/atoms/base/Link.h>
#include <opencog/atoms/base/Node.h>
#include <opencog/atoms/atom_types/atom_types.h>
#include <opencog/atoms/value/FloatValue.h>
#include <opencog/atoms/value/LinkValue.h>
#include <opencog/atoms/value/VoidValue.h>
#include <opencog/atomspace/AtomSpace.h>

#include <opencog/persist/api/StorageNode.h>
#include <opencog/persist/rocks-types/atom_types.h>

#include <opencog/util/Logger.h>

using namespace opencog;

class MultiDeleteUTest :  public CxxTest::TestSuite
{
	private:
		std::string uri;
		AtomSpacePtr as;
		StorageNodePtr store;

	public:

		MultiDeleteUTest(void)
		{
			logger().set_level(Logger::INFO);
			// logger().set_level(Logger::DEBUG);
			logger().set_print_to_stdout_flag(true);

			uri = "rocks:///tmp/cog-rocks-multi-delete-utest";
		}

		~MultiDeleteUTest()
		{
			// erase the log file if no assertions failed
			if (!CxxTest::TestTracker::tracker().suiteFailed())
			{
				std::remove(logger().get_filename().c_str());
				// Also remove the database directory
				// URI is "rocks:///tmp/..." so path starts at position 8
				std::string dbpath = uri.substr(8);
				std::filesystem::remove_all(dbpath);
			}
		}

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

		void worker(int);
		void test_add_delete(void);
};

/*
 * This is called once before each test, for each test (!!)
 */
void MultiDeleteUTest::setUp(void)
{
}

void MultiDeleteUTest::tearDown(void)
{
}

// ============================================================

// Add and then remove a bunch of links.
// This is designed so that multiple threads will be
// touching the same atoms, adding and removing the same atoms.
// This should not cause any failures or crashes.
void MultiDeleteUTest::worker(int thread_id)
{
	int nloops = 50;
	int nsplice = 50;

	Handle ha(as->add_node(CONCEPT_NODE, "a"));

	for (int j=0; j<nsplice; j++)
	{
		for (int i=0; i<nloops; i++)
		{
			int id = (thread_id + i) % (thread_id + 7);
			Handle hb(as->add_node(CONCEPT_NODE, std::to_string(id)));
			Handle li(as->add_link(LIST_LINK, {ha, hb, ha, hb}));

			// Store the link.
			store->store_atom(li);
		}

		for (int i=0; i<nloops; i++)
		{
			int id = (thread_id + i) % (thread_id + 7);
			Handle hb(as->get_node(CONCEPT_NODE, std::to_string(id)));
			Handle li(as->get_link(LIST_LINK, {ha, hb, ha, hb}));

			// If it's already gone, do nothing.
			if (nullptr == li) continue;

			// Remove the link.
			store->remove_atom(as.get(), li);
		}
	}

	// Just one barrier, at the end.
	store->barrier();
}

// ============================================================

void MultiDeleteUTest::test_add_delete(void)
{
	logger().debug("BEGIN TEST: %s", __FUNCTION__);

	// Configuration. Number of actual running threads may be less,
	// depending on the CPU architecture.
	int n_threads = 10;

	as = createAtomSpace();

	Handle hsn = as->add_node(ROCKS_STORAGE_NODE, std::string(uri));
	store = StorageNodeCast(hsn);

	store->setValue(as->add_node(PREDICATE_NODE, "*-open-*"), createVoidValue());
	TS_ASSERT(store->connected())

	// Clear out left-over junk, just in case.
	store->erase();

	printf("Start creating %d threads\n", n_threads);

	std::vector<std::thread> thread_pool;
	for (int i=0; i < n_threads; i++) {
		thread_pool.push_back(
			std::thread(&MultiDeleteUTest::worker, this, i));
	}

	for (std::thread& t : thread_pool) t.join();
	printf("Done joining threads\n");

	printf("Post-update AtomSpace size=%lu\n", as->get_size());
	// printf("AtomSpace contents =\n%s\n", as->to_string().c_str());

	// Verify the local AtomSpace size. Should be 35:
	// The StorageNode itself, the (Concept "aaaa"), *-open-* predicate, and
	// 16 Concept and Lists from thread-id modulo 16
	// Note: *-IsKeyFlag-* is NOT added by store_atom, only by load operations
	TS_ASSERT_EQUALS(as->get_size(), 3 + 2*(n_threads+6));

	// Verify the DB contents: pull everything from the DB,
	// back to here. Do this "independently", by closing and
	// reopening; this forces the Storagenode through assorted
	// flush and close code, potentially exposing bugs.
	store->setValue(as->add_node(PREDICATE_NODE, "*-close-*"), createVoidValue());

	HandleSeq hs;
	as->get_handles_by_type(hs, CONCEPT_NODE);
	for (const Handle& h : hs)
		as->extract_atom(h, true);
	hs.clear();
	printf("AtomSpace contents =\n%s\n", as->to_string().c_str());
	// Storage node + *-open-* + *-close-*
	TS_ASSERT_EQUALS(as->get_size(), 3);

	store->setValue(as->add_node(PREDICATE_NODE, "*-open-*"), createVoidValue());
	TS_ASSERT(store->connected())

	store->load_atomspace();
	printf("Final AtomSpace size=%lu\n", as->get_size());

	// Size should be 21: the 18 from before plus
	// (Predicate "*-TruthValueKey-*"), (Predicate "*-IsKeyFlag-*"), and *-open-*
	// Well, actually, there is no guarantee that the removes will
	// happen after the stores. Best we can do is check lower and
	// upper bounds of 21 and 37:
	size_t min_size = 5 + n_threads + 6;
	size_t max_size = 5 + 2*(n_threads + 6);
	if (as->get_size() < min_size or max_size < as->get_size())
		printf("Error: Bad size! AtomSpace contents =\n%s\n",
			as->to_string().c_str());

	TS_ASSERT(as->get_size() >= min_size);
	TS_ASSERT(as->get_size() <= max_size);

	store->erase();
	store->setValue(as->add_node(PREDICATE_NODE, "*-close-*"), createVoidValue());
	store = nullptr;

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

/* ============================= END OF FILE ================= */
opencog-atomspace-rocks / tests / persist / rocks / MultiDeleteUTest.cxxtest
