/*
* tests/persist/cog/MultiUserUTest.cxxtest
* Ported over from
* tests/persist/sql/multi-driver/MultiUserUTest.cxxtest
*
* Test multi-user persistance. Assumes MultiPersistUTest is passing.
* Emulates multiple parallel users, one per thread.
*
* 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 <time.h>
#include <atomic>
#include <cstdio>
#include <thread>
#include <opencog/atoms/atom_types/atom_types.h>
#include <opencog/persist/cog-types/atom_types.h>
#include <opencog/atoms/base/Atom.h>
#include <opencog/atoms/base/Link.h>
#include <opencog/atoms/base/Node.h>
#include <opencog/persist/cog-storage/CogStorage.h>
#include "../TestCogServer.h"
#include <opencog/atoms/value/FloatValue.h>
#include <opencog/atoms/value/VoidValue.h>
#include <opencog/util/Logger.h>
using namespace opencog;
// MultiUserUTest -- emulates multiple users.
//
// This emulates multiple users, by creating multiple atomspaces,
// and multiple connections to the database, and then making sure
// that the contents of the database, and of the atomspaces, do not
// get scrambled.
//
// In principle, this "should" be done by creating multiple processes,
// with one atomspace per process, and having each atomspace attach to
// the same database. But this is kind-of annoying to do in this
// particular test framework, so instead, we do this multi-threaded,
// pretending each thread is a different user. Since we create multiple
// atomspaces, this should be a srong-enough test.
//
// This is similar to, but different from the MultiPersistUTest. That
// test is making sure that locking is working correctly: it uses only
// one atomspace, with multiple threads accessing that one atomspace.
// So that test is a single-user, multi-threaded test.
class MultiUserUTest : public CxxTest::TestSuite
{
private:
#define NUSERS 6
AtomSpacePtr _as[NUSERS];
StorageNodePtr _store[NUSERS];
std::atomic<int> nstarted;
DECLARE_TEST_COGSERVER
#define NCLAUSE 8
#define NATOMS (NUSERS * NCLAUSE)
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 clobber;
public:
MultiUserUTest(void);
~MultiUserUTest()
{
STOP_TEST_COGSERVER
// Erase the log file if no assertions failed
logger().flush();
if (!CxxTest::TestTracker::tracker().suiteFailed())
std::remove(logger().get_filename().c_str());
}
void setUp(void);
void tearDown(void);
void add_to_space(int, AtomSpacePtr, StorageNodePtr, std::string id);
void fetch_space(int, StorageNodePtr);
void check_space(int, AtomSpacePtr, std::string dbgmsg);
void atomCompare(Handle, AtomPtr, std::string);
void run_user_test(int);
void connect_to_db(int);
void add_atoms(int);
void do_test_multiuser(void);
void test_multiuser(void);
void test_clobber(void);
};
MultiUserUTest:: MultiUserUTest(void)
{
logger().set_level(Logger::INFO);
// logger().set_level(Logger::DEBUG);
logger().set_print_to_stdout_flag(true);
INIT_TEST_COGSERVER(16007);
printf("Started CogServer\n");
n_threads = NUSERS;
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 MultiUserUTest::setUp(void)
{
}
void MultiUserUTest::connect_to_db(int i)
{
_as[i] = createAtomSpace();
Handle hsn = _as[i]->add_node(COG_STORAGE_NODE, "cog://localhost:16007");
_store[i] = StorageNodeCast(hsn);
_store[i]->setValue(_as[i]->add_node(PREDICATE_NODE, "*-open-*"), createVoidValue());
TS_ASSERT(_store[i]->connected())
}
void MultiUserUTest::tearDown(void)
{
}
// ============================================================
void MultiUserUTest::atomCompare(Handle a, AtomPtr b, std::string where)
{
// printf("Check %s expect %s\n", 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 mis-match", 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)
{
// Different thrads will set different confidence
// values, which will clobber one-another. So these
// cannot be meainingfully compared.
if (clobber)
{
FloatValuePtr fva = FloatValueCast(ta);
FloatValuePtr fvb = FloatValueCast(tb);
TSM_ASSERT("Truth value miscompare",
fva->value()[0] == fvb->value()[0]);
if (not (fva->value()[0] == fvb->value()[0]))
{
fprintf(stderr, "Error, truth value miscompare, "
"ma=%f mb=%f\n",
fva->value()[0], fvb->value()[0]);
}
}
else
{
TSM_ASSERT("Truth value miscompare", (*ta)==(*tb));
if (not ((*ta) == (*tb)))
{
FloatValuePtr fva = FloatValueCast(ta);
FloatValuePtr fvb = FloatValueCast(tb);
fprintf(stderr, "Error, truth value miscompare, "
"ma=%f mb=%f ca=%f cb=%f\n",
fva->value()[0], fvb->value()[0],
fva->value()[2], fvb->value()[2]);
}
}
}
}
}
// ============================================================
void MultiUserUTest::add_to_space(int idx, AtomSpacePtr as,
StorageNodePtr store, std::string id)
{
if (not clobber) id += std::to_string(idx);
// Create an atom ...
ValuePtr stv(createFloatValue(std::vector<double>({0.11, 0.0, 100.0+idx})));
h1[idx] = as->add_node(CONCEPT_NODE, id + "fromNode");
h1[idx]->setValue(truth_key(), stv);
n1[idx] = NodeCast(h1[idx]);
store->store_atom(h1[idx]);
ValuePtr stv2(createFloatValue(std::vector<double>({0.22, 0.0, 200.0+idx})));
h2[idx] = as->add_node(CONCEPT_NODE, id + "toNode");
h2[idx]->setValue(truth_key(), stv2);
n2[idx] = NodeCast(h2[idx]);
store->store_atom(h2[idx]);
ValuePtr stv3(createFloatValue(std::vector<double>({0.33, 0.0, 300.0+idx})));
h3[idx] = as->add_node(CONCEPT_NODE, id + "third wheel");
h3[idx]->setValue(truth_key(), stv3);
n3[idx] = NodeCast(h3[idx]);
store->store_atom(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, 0.0, 400.0+idx})));
h4[idx]->setValue(truth_key(), stv4);
n4[idx] = NodeCast(h4[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]);
store->barrier();
// printf("Added for idx=%d\n", idx);
fflush(stdout);
}
void MultiUserUTest::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);
space->barrier();
}
void MultiUserUTest::check_space(int idx, AtomSpacePtr space, std::string dbgmsg)
{
// printf("Checking %d\n", idx);
fflush(stdout);
std::string name1 = n1[idx]->get_name();
Handle hb1 = space->get_node(n1[idx]->get_type(), std::move(name1));
atomCompare(h1[idx], hb1, dbgmsg + "-hb1");
std::string name2 = n2[idx]->get_name();
Handle hb2 = space->get_node(n2[idx]->get_type(), std::move(name2));
atomCompare(h2[idx], hb2, dbgmsg + "-hb2");
std::string name3 = n3[idx]->get_name();
Handle hb3 = space->get_node(n3[idx]->get_type(), std::move(name3));
atomCompare(h3[idx], hb3, dbgmsg + "-hb3");
std::string name4 = n4[idx]->get_name();
Handle hb4 = space->get_node(n4[idx]->get_type(), std::move(name4));
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 MultiUserUTest::add_atoms(int thread_id)
{
int idx = 0;
int off = NCLAUSE*thread_id;
AtomSpacePtr as = _as[thread_id];
StorageNodePtr store = _store[thread_id];
add_to_space(off + idx++, as, store, "AA-aa-wow ");
add_to_space(off + idx++, as, store, "BB-bb-wow ");
add_to_space(off + idx++, as, store, "CC-cc-wow ");
add_to_space(off + idx++, as, store, "DD-dd-wow ");
add_to_space(off + idx++, as, store, "EE-ee-wow ");
/* Make sure UTF-8 works fine. */
add_to_space(off + idx++, as, store, "Попытка выбраться вызвала слабый стон ");
add_to_space(off + idx++, as, store, "はにがうりだそうであってるのかはち ");
add_to_space(off + idx++, as, store, "係拉丁字母");
TSM_ASSERT("Bad number of clauses", idx == NCLAUSE);
/* Verify that the atoms were added */
for (int i=0; i<idx; i++)
check_space(off+i, as, "verify-add");
}
// ============================================================
void MultiUserUTest::run_user_test(int thread_id)
{
if (0 < thread_id) sleep(1);
connect_to_db(thread_id);
nstarted.fetch_add(1);
time_t start = time(0);
// Spin, until they are all started.
while (n_threads != nstarted.fetch_add(0))
{
usleep (50000);
time_t end = time(0);
if (10 < end - start) break;
std::this_thread::yield();
}
time_t end = time(0);
TSM_ASSERT("Took too long to connect", end - start < 10);
add_atoms(thread_id);
_store[thread_id]->setValue(_as[thread_id]->add_node(PREDICATE_NODE, "*-close-*"), createVoidValue());
_as[thread_id] = nullptr;
}
// ============================================================
// Perform one repetition of the unit test.
void MultiUserUTest::do_test_multiuser(void)
{
printf("Start creating %d user sessions\n", n_threads);
nstarted = 0;
std::vector<std::thread> thread_pool;
for (int i=0; i < n_threads; i++) {
thread_pool.push_back(
std::thread(&MultiUserUTest::run_user_test, this, i));
}
for (std::thread& t : thread_pool) t.join();
printf("Done joining threads\n");
AtomSpacePtr aspace = createAtomSpace();
Handle hsn = aspace->add_node(COG_STORAGE_NODE, "cog://localhost:16007");
StorageNodePtr store = StorageNodeCast(hsn);
store->setValue(aspace->add_node(PREDICATE_NODE, "*-open-*"), createVoidValue());
TS_ASSERT(store->connected())
for (int off=0; off<n_threads; off++)
for (int i=0; i<NCLAUSE; i++) {
fetch_space(NCLAUSE*off+i, store);
check_space(NCLAUSE*off+i, aspace, "verify-again");
}
store->barrier();
store->setValue(aspace->add_node(PREDICATE_NODE, "*-close-*"), createVoidValue());
aspace = nullptr;
}
void MultiUserUTest::test_multiuser(void)
{
logger().debug("BEGIN TEST: %s", __FUNCTION__);
clobber = false;
do_test_multiuser();
logger().debug("END TEST: %s", __FUNCTION__);
}
void MultiUserUTest::test_clobber(void)
{
logger().debug("BEGIN TEST: %s", __FUNCTION__);
clobber = true;
do_test_multiuser();
logger().debug("END TEST: %s", __FUNCTION__);
}
/* ============================= END OF FILE ================= */