/*
* tests/persist/rocks/MultiPersistUTest.cxxtest
*
* Test multi-threaded persistence. Assumes PersistUTest is passing.
* Ported over from original unit test at
* tests/persist/sql/multi-driver/MultiPersistUTest.cxxtest
*
* Copyright (C) 2008, 2009, 2013, 2017, 2019 Linas Vepstas
* 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 <atomic>
#include <cstdio>
#include <filesystem>
#include <thread>
#include <opencog/atoms/atom_types/atom_types.h>
#include <opencog/atoms/base/Atom.h>
#include <opencog/atoms/base/Link.h>
#include <opencog/atoms/base/Node.h>
#include <opencog/atomspace/AtomSpace.h>
#include <opencog/persist/rocks-types/atom_types.h>
#include <opencog/persist/rocks/RocksStorage.h>
#include <opencog/atoms/value/FloatValue.h>
#include <opencog/atoms/value/VoidValue.h>
#include <opencog/util/Logger.h>
using namespace opencog;
class MultiPersistUTest : public CxxTest::TestSuite
{
private:
AtomSpacePtr _as;
StorageNodePtr store;
#define NTHREADS 6
#define NHACK 10
#define NATOMS (NTHREADS * NHACK)
NodePtr *n1;
NodePtr *n2;
NodePtr *n3;
NodePtr *n4;
LinkPtr *l;
LinkPtr *l2;
LinkPtr *l3;
Handle *h1;
Handle *h2;
Handle *h3;
Handle *h4;
Handle *hl;
Handle *hl2;
Handle *hl3;
int n_threads;
bool spinwait;
public:
MultiPersistUTest(void);
~MultiPersistUTest()
{
// 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
std::filesystem::remove_all("/tmp/cog-rocks-multi-persist-utest");
}
}
void setUp(void);
void tearDown(void);
void add_to_space(int, AtomSpace *, StorageNodePtr, std::string id);
void fetch_space(int, StorageNodePtr);
void check_space(int, AtomSpace *, std::string dbgmsg);
void check_empty(int, AtomSpace *);
void atomCompare(Handle, Handle, std::string);
void threaded_add(int);
void test_atomspace(void);
};
MultiPersistUTest:: MultiPersistUTest(void)
{
// logger().set_level(Logger::DEBUG);
logger().set_level(Logger::INFO);
logger().set_print_to_stdout_flag(true);
n_threads = NTHREADS;
n1 = new NodePtr[NATOMS];
n2 = new NodePtr[NATOMS];
n3 = new NodePtr[NATOMS];
n4 = new NodePtr[NATOMS];
l = new LinkPtr[NATOMS];
l2 = new LinkPtr[NATOMS];
l3 = new LinkPtr[NATOMS];
h1 = new Handle[NATOMS];
h2 = new Handle[NATOMS];
h3 = new Handle[NATOMS];
h4 = new Handle[NATOMS];
hl = new Handle[NATOMS];
hl2 = new Handle[NATOMS];
hl3 = new Handle[NATOMS];
}
/*
* This is called once before each test, for each test (!!)
*/
void MultiPersistUTest::setUp(void)
{
_as = createAtomSpace();
Handle hsn = _as->add_node(ROCKS_STORAGE_NODE, "rocks:///tmp/cog-rocks-multi-persist-utest");
store = StorageNodeCast(hsn);
store->setValue(_as->add_node(PREDICATE_NODE, "*-open-*"), createVoidValue());
TS_ASSERT(store->connected())
}
void MultiPersistUTest::tearDown(void)
{
store->setValue(_as->add_node(PREDICATE_NODE, "*-close-*"), createVoidValue());
_as = nullptr;
}
// ============================================================
void MultiPersistUTest::atomCompare(Handle a, Handle b, std::string where)
{
logger().debug("Check %s expect %s", where.c_str(), a->to_string().c_str());
TSM_ASSERT("No atom found", b != nullptr);
if (nullptr == b)
{
printf("Check %s expect %s\n", where.c_str(), a->to_string().c_str());
return;
}
TSM_ASSERT_EQUALS("Type mismatch", a->get_type(), b->get_type());
NodePtr na(NodeCast(a));
NodePtr nb(NodeCast(b));
if (na and nb)
{
TSM_ASSERT_EQUALS("Name mismatch", na->get_name(), nb->get_name());
}
LinkPtr la(LinkCast(a));
LinkPtr lb(LinkCast(b));
if (la and lb)
{
TSM_ASSERT_EQUALS("Arity mismatch", la->get_arity(), lb->get_arity());
if (*la != *lb)
{
printf("Mismatching la = %s\n", la->to_string().c_str());
printf("Mismatching lb = %s\n", lb->to_string().c_str());
TSM_ASSERT_EQUALS("Link mismatch", la, lb);
}
}
ValuePtr ta = a->getValue(truth_key());
ValuePtr tb = b->getValue(truth_key());
if (ta or tb)
{
TSM_ASSERT("Missing truth value", ta);
TSM_ASSERT("Missing truth value", tb);
if (ta and tb)
{
// Only compare mean. Different threads set the
// confidence in a racey manner, and so anything
// might show up there. We don't know what order
// the stores hit the server, or whether data is
// changing in-flight.
FloatValuePtr fva = FloatValueCast(ta);
FloatValuePtr fvb = FloatValueCast(tb);
bool okay = fabs(fva->value()[0] - fvb->value()[0]) < 1e-6;
TSM_ASSERT("Truth value miscompare", okay);
if (not okay)
{
printf("Error, truth value miscompare at %s, "
"ma=%f mb=%f ca=%f cb=%f\n",
where.c_str(),
fva->value()[0], fvb->value()[0],
fva->value()[1], fvb->value()[1]);
printf("Expect %s\n", a->to_string().c_str());
}
}
}
}
// ============================================================
void MultiPersistUTest::add_to_space(int idx, AtomSpace *as,
StorageNodePtr store, std::string id)
{
// Create an atom ...
ValuePtr stv(createFloatValue(std::vector<double>({0.11, 100.0+idx})));
h1[idx] = as->add_node(CONCEPT_NODE, id + "fromNode");
h1[idx]->setValue(truth_key(), stv);
n1[idx] = NodeCast(h1[idx]);
ValuePtr stv2(createFloatValue(std::vector<double>({0.22, 200.0+idx})));
h2[idx] = as->add_node(CONCEPT_NODE, id + "toNode");
h2[idx]->setValue(truth_key(), stv2);
n2[idx] = NodeCast(h2[idx]);
ValuePtr stv3(createFloatValue(std::vector<double>({0.33, 300.0+idx})));
h3[idx] = as->add_node(CONCEPT_NODE, id + "third wheel");
h3[idx]->setValue(truth_key(), stv3);
n3[idx] = NodeCast(h3[idx]);
// The NumberNode will go through the AtomTable clone factory
// and should thus elicit any errors in clone uuid handling.
char buf[40]; sprintf(buf, "%f", idx+0.14159265358979);
h4[idx] = as->add_node(NUMBER_NODE, buf);
ValuePtr stv4(createFloatValue(std::vector<double>({0.44, 400.0+idx})));
h4[idx]->setValue(truth_key(), stv4);
n4[idx] = NodeCast(h4[idx]);
// We need to store these, as we will test the fetch of their
// values just a little while later.
store->store_atom(h1[idx]);
store->store_atom(h2[idx]);
store->store_atom(h3[idx]);
store->store_atom(h4[idx]);
HandleSeq hvec;
hvec.push_back(h1[idx]);
hvec.push_back(h2[idx]);
hvec.push_back(h3[idx]);
hvec.push_back(h4[idx]);
// Note that SetLink is an unordered link.
hl[idx] = as->add_link(SET_LINK, std::move(hvec));
store->store_atom(hl[idx]);
l[idx] = LinkCast(hl[idx]);
hl2[idx] = as->add_link(LIST_LINK, hl[idx], h2[idx]);
store->store_atom(hl2[idx]);
l2[idx] = LinkCast(hl2[idx]);
hl3[idx] = as->add_link(EVALUATION_LINK, h1[idx], hl2[idx], h3[idx]);
store->store_atom(hl3[idx]);
l3[idx] = LinkCast(hl3[idx]);
// Do it again, just for good measure.
store->store_atom(hl[idx]);
store->store_atom(hl2[idx]);
store->store_atom(hl3[idx]);
// printf("Added for idx=%d\n", idx);
fflush(stdout);
}
void MultiPersistUTest::fetch_space(int idx, StorageNodePtr space)
{
Handle ab1 = createNode(n1[idx]->get_type(), n1[idx]->get_name());
Handle hb1(ab1->get_handle());
space->fetch_atom(hb1);
Handle ab2 = createNode(n2[idx]->get_type(), n2[idx]->get_name());
Handle hb2(ab2->get_handle());
space->fetch_atom(hb2);
Handle ab3 = createNode(n3[idx]->get_type(), n3[idx]->get_name());
Handle hb3(ab3->get_handle());
space->fetch_atom(hb3);
Handle ab4 = createNode(n4[idx]->get_type(), n4[idx]->get_name());
Handle hb4(ab4->get_handle());
space->fetch_atom(hb4);
HandleSeq hvec;
hvec.push_back(hb1);
hvec.push_back(hb2);
hvec.push_back(hb3);
hvec.push_back(hb4);
Handle alb = createLink(hvec, hl[idx]->get_type());
Handle hlb(alb->get_handle());
space->fetch_atom(hlb);
HandleSeq hv2({hlb, hb2});
Handle alb2 = createLink(hv2, hl2[idx]->get_type());
Handle hlb2(alb2->get_handle());
space->fetch_atom(hlb2);
HandleSeq hv3({hb1, hlb2, hb3});
Handle alb3 = createLink(hv3, hl3[idx]->get_type());
Handle hlb3(alb3->get_handle());
space->fetch_atom(hlb3);
}
void MultiPersistUTest::check_space(int idx, AtomSpace *space, std::string dbgmsg)
{
// printf("Checking %d\n", idx);
fflush(stdout);
std::string nam1 = n1[idx]->get_name();
Handle hb1 = space->get_node(n1[idx]->get_type(), std::move(nam1));
atomCompare(h1[idx], hb1, dbgmsg + "-hb1");
std::string nam2 = n2[idx]->get_name();
Handle hb2 = space->get_node(n2[idx]->get_type(), std::move(nam2));
atomCompare(h2[idx], hb2, dbgmsg + "-hb2");
std::string nam3 = n3[idx]->get_name();
Handle hb3 = space->get_node(n3[idx]->get_type(), std::move(nam3));
atomCompare(h3[idx], hb3, dbgmsg + "-hb3");
std::string nam4 = n4[idx]->get_name();
Handle hb4 = space->get_node(n4[idx]->get_type(), std::move(nam4));
atomCompare(h4[idx], hb4, dbgmsg + "-hb4");
HandleSeq hvec;
hvec.push_back(hb1);
hvec.push_back(hb2);
hvec.push_back(hb3);
hvec.push_back(hb4);
Handle hlb = space->get_link(hl[idx]->get_type(), std::move(hvec));
atomCompare(hl[idx], hlb, dbgmsg + "-hlinkb");
HandleSeq hv2({hlb, hb2});
Handle hlb2 = space->get_link(hl2[idx]->get_type(), std::move(hv2));
atomCompare(hl2[idx], hlb2, dbgmsg + "-hlinkb2");
HandleSeq hv3({hb1, hlb2, hb3});
Handle hlb3 = space->get_link(hl3[idx]->get_type(), std::move(hv3));
atomCompare(hl3[idx], hlb3, dbgmsg + "-hlinkb3");
}
void MultiPersistUTest::check_empty(int idx, AtomSpace *space)
{
std::string nam1 = n1[idx]->get_name();
Handle hb1 = space->get_node(n1[idx]->get_type(), std::move(nam1));
TSM_ASSERT("Should not find this atom", hb1 == Handle::UNDEFINED);
std::string nam2 = n2[idx]->get_name();
Handle hb2 = space->get_node(n2[idx]->get_type(), std::move(nam2));
TSM_ASSERT("Should not find this atom", hb2 == Handle::UNDEFINED);
std::string nam3 = n3[idx]->get_name();
Handle hb3 = space->get_node(n3[idx]->get_type(), std::move(nam3));
TSM_ASSERT("Should not find this atom", hb3 == Handle::UNDEFINED);
std::string nam4 = n4[idx]->get_name();
Handle hb4 = space->get_node(n4[idx]->get_type(), std::move(nam4));
TSM_ASSERT("Should not find this atom", hb4 == Handle::UNDEFINED);
HandleSeq hvec;
hvec.push_back(h1[idx]);
hvec.push_back(h2[idx]);
hvec.push_back(h3[idx]);
hvec.push_back(h4[idx]);
Handle hlb = space->get_link(hl[idx]->get_type(), std::move(hvec));
TSM_ASSERT("Should not find this atom", hlb == Handle::UNDEFINED);
}
// ============================================================
void MultiPersistUTest::threaded_add(int thread_id)
{
int idx = 0;
int off = NHACK*thread_id;
add_to_space(off + idx++, _as.get(), store, "AA-aa-wow ");
add_to_space(off + idx++, _as.get(), store, "BB-bb-wow ");
add_to_space(off + idx++, _as.get(), store, "CC-cc-wow ");
add_to_space(off + idx++, _as.get(), store, "DD-dd-wow ");
add_to_space(off + idx++, _as.get(), store, "EE-ee-wow ");
/* Make sure UTF-8 works fine. */
add_to_space(off + idx++, _as.get(), store, "Попытка выбраться вызвала слабый стон ");
add_to_space(off + idx++, _as.get(), store, "はにがうりだそうであってるのかはち ");
add_to_space(off + idx++, _as.get(), store, "係拉丁字母");
/* Verify that the atoms were added */
for (int i=0; i<idx; i++)
check_space(off+i, _as.get(), "verify-add");
}
// ============================================================
// Perform one repetition of the unit test.
void MultiPersistUTest::test_atomspace(void)
{
logger().debug("BEGIN TEST: %s", __FUNCTION__);
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(&MultiPersistUTest::threaded_add, this, i));
}
for (std::thread& t : thread_pool) t.join();
printf("Done joining threads\n");
/* Extract atoms from the AtomSpace. This does not delete them from
* the cogserver storage, though; to do that, they must be deleted,
* not extracted.
*/
store->barrier();
_as->clear();
TSM_ASSERT("Non-empty atomspace", 0 == _as->get_size());
printf("===============================================\n");
/* The clear above nuked the storage, so restart */
Handle hsn = _as->add_node(ROCKS_STORAGE_NODE, "rocks:///tmp/cog-rocks-multi-persist-utest");
store = StorageNodeCast(hsn);
store->setValue(_as->add_node(PREDICATE_NODE, "*-open-*"), createVoidValue());
TS_ASSERT(store->connected())
int i, off;
#define NATS 8
/* Verify that the atoms can still be fetched from storage. */
for (off=0; off<n_threads; off++)
for (i=0; i<NATS; i++) {
fetch_space(NHACK*off+i, store);
check_space(NHACK*off+i, _as.get(), "verify-fetch-idx=" + std::to_string(i));
}
/* Do it again, for good luck. */
_as->clear();
TSM_ASSERT("Non-empty atomspace", 0 == _as->get_size());
/* The clear above nuked the storage, so restart */
hsn = _as->add_node(ROCKS_STORAGE_NODE, "rocks:///tmp/cog-rocks-multi-persist-utest");
store = StorageNodeCast(hsn);
store->setValue(_as->add_node(PREDICATE_NODE, "*-open-*"), createVoidValue());
TS_ASSERT(store->connected())
/* Verify that the atoms can still be fetched from storage. */
for (off=0; off<n_threads; off++)
for (i=0; i<NATS; i++) {
fetch_space(NHACK*off+i, store);
check_space(NHACK*off+i, _as.get(), "verify-again");
}
/* Kill data for good */
_as->clear();
TSM_ASSERT("Non-empty atomspace", 0 == _as->get_size());
/* Verify that the atoms are no longer being found. */
for (off=0; off<n_threads; off++)
for (i=0; i<NATS; i++)
check_empty(NHACK*off+i, _as.get());
logger().debug("END TEST: %s", __FUNCTION__);
}
/* ============================= END OF FILE ================= */