/*
* tests/persist/cog-storage/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 <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/persist/cog-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 "../TestCogServer.h"
#include <opencog/persist/api/StorageNode.h>
#include <opencog/util/Logger.h>
using namespace opencog;
class MultiDeleteUTest : public CxxTest::TestSuite
{
private:
std::string uri;
DECLARE_TEST_COGSERVER
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 = "cog://localhost:16012";
INIT_TEST_COGSERVER(16012);
printf("Started CogServer\n");
}
~MultiDeleteUTest()
{
STOP_TEST_COGSERVER
// 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 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 = 500;
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, li);
}
}
// Just one barrier, at the end.
store->barrier();
}
// ============================================================
void MultiDeleteUTest::test_add_delete(void)
{
logger().debug("BEGIN TEST: %s", __FUNCTION__);
// Configuration. On my machine, the locks allow about 2.5 to
// 3 threads run in parallel, and of that time, a third is spent
// in the kernel (I assume that's kernel time for file-writes.)
int n_threads = 10;
as = createAtomSpace();
Handle hsn = as->add_node(COG_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.
::kill_data(store.get(), _test_asp.get());
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 CogStorageNode itself, the (Concept "aaaa") and
// 16 from thread-id modulo 16, plus *-open-* predicate
TS_ASSERT_EQUALS(as->get_size(), 3 + 2*(n_threads+6));
// Verify the remote AtomSpace: pull everything from the
// cogserver, back to here. Do this "independently", by
// closing the socket; this forces the cogserver through
// its socket-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());
// CogStorageNode + *-close-* predicate + *-open-* predicate
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 20: the 18 from before plus that annoying
// (Predicate "*-TruthValueKey-*") And also *-IsKeyFlag-*
// And 9 more from CogServerNode gorp
// Plus *-open-* and *-close-* predicates
// 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 22 and 38:
size_t min_size = 9 + 4 + 2 + n_threads + 6;
size_t max_size = 9 + 4 + 2 + 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);
::kill_data(store.get(), _test_asp.get());
store->setValue(as->add_node(PREDICATE_NODE, "*-close-*"), createVoidValue());
store = nullptr;
logger().debug("END TEST: %s", __FUNCTION__);
}
/* ============================= END OF FILE ================= */