path: root/src/third_party/wiredtiger/test/csuite/wt2323_join_visibility/main.c

/*-
 * Public Domain 2014-2016 MongoDB, Inc.
 * Public Domain 2008-2014 WiredTiger, Inc.
 *
 * This is free and unencumbered software released into the public domain.
 *
 * Anyone is free to copy, modify, publish, use, compile, sell, or
 * distribute this software, either in source code form or as a compiled
 * binary, for any purpose, commercial or non-commercial, and by any
 * means.
 *
 * In jurisdictions that recognize copyright laws, the author or authors
 * of this software dedicate any and all copyright interest in the
 * software to the public domain. We make this dedication for the benefit
 * of the public at large and to the detriment of our heirs and
 * successors. We intend this dedication to be an overt act of
 * relinquishment in perpetuity of all present and future rights to this
 * software under copyright law.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
 * IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
 * OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
 * ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
 * OTHER DEALINGS IN THE SOFTWARE.
 */
#include "test_util.h"

/*
 * JIRA ticket reference: WT-2323
 *
 * Test case description: We create two kinds of threads that race: One kind
 * is populating/updating records in a table with a few indices, the other
 * is reading from a join of that table. The hope in constructing this test
 * was to have the updates interleaved between reads of multiple indices by
 * the join, yielding an inconsistent view of the data.  In the main table,
 * we insert account records, with a positive or negative balance.  The
 * negative balance accounts always have a flag set to non-zero, positive
 * balances have the flag set to zero.  The join we do is:
 *
 *   select (*) from account where account.postal_code = '54321' and
 *                      account.balance < 0 and account.flags == 0
 *
 * which should always yield no results.
 *
 * Failure mode: This test never actually failed with any combination of
 * parameters, with N_INSERT up to 50000000.  It seems that a snapshot is
 * implicitly allocated in the session used by a join by the set_key calls
 * that occur before the first 'next' of the join cursor is done.  Despite
 * that, the test seems interesting enough to keep around, with the number
 * of inserts set low as a default.
 */

void (*custom_die)(void) = NULL;

#define	N_RECORDS	10000
#define	N_INSERT	500000
#define	N_INSERT_THREAD	2
#define	N_JOIN_THREAD	2
#define	S64 "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789::"
#define	S1024 (S64 S64 S64 S64 S64 S64 S64 S64 S64 S64 S64 S64 S64 S64 S64 S64)

typedef struct {
	char posturi[256];
	char baluri[256];
	char flaguri[256];
	char joinuri[256];
	bool bloom;
	bool remove;
} SHARED_OPTS;

typedef struct {
	TEST_OPTS *testopts;
	SHARED_OPTS *sharedopts;
	int threadnum;
	int nthread;
	int done;
	int joins;
	int removes;
	int inserts;
	int notfounds;
	int rollbacks;
} THREAD_ARGS;

static void *thread_insert(void *);
static void *thread_join(void *);
static void test_join(TEST_OPTS *, SHARED_OPTS *, bool, bool);

int
main(int argc, char *argv[])
{
	SHARED_OPTS *sharedopts, _sharedopts;
	TEST_OPTS *opts, _opts;
	const char *tablename;

	opts = &_opts;
	sharedopts = &_sharedopts;
	memset(opts, 0, sizeof(*opts));
	memset(sharedopts, 0, sizeof(*sharedopts));

	testutil_check(testutil_parse_opts(argc, argv, opts));
	testutil_make_work_dir(opts->home);

	tablename = strchr(opts->uri, ':');
	testutil_assert(tablename != NULL);
	tablename++;
	snprintf(sharedopts->posturi, sizeof(sharedopts->posturi),
	    "index:%s:post", tablename);
	snprintf(sharedopts->baluri, sizeof(sharedopts->baluri),
	    "index:%s:bal", tablename);
	snprintf(sharedopts->flaguri, sizeof(sharedopts->flaguri),
	    "index:%s:flag", tablename);
	snprintf(sharedopts->joinuri, sizeof(sharedopts->joinuri),
	    "join:%s", opts->uri);

	testutil_check(wiredtiger_open(opts->home, NULL,
	    "create,cache_size=1G", &opts->conn));

	test_join(opts, sharedopts, true, true);
	test_join(opts, sharedopts, true, false);
	test_join(opts, sharedopts, false, true);
	test_join(opts, sharedopts, false, false);

	testutil_cleanup(opts);

	return (0);
}

static void
test_join(TEST_OPTS *opts, SHARED_OPTS *sharedopts, bool bloom,
    bool sometimes_remove)
{
	THREAD_ARGS insert_args[N_INSERT_THREAD], join_args[N_JOIN_THREAD];
	WT_CURSOR *maincur;
	WT_SESSION *session;
	pthread_t insert_tid[N_INSERT_THREAD], join_tid[N_JOIN_THREAD];
	int i;

	memset(insert_args, 0, sizeof(insert_args));
	memset(join_args, 0, sizeof(join_args));

	sharedopts->bloom = bloom;
	sharedopts->remove = sometimes_remove;

	fprintf(stderr, "Running with bloom=%d, remove=%d\n",
	    (int)bloom, (int)sometimes_remove);

	testutil_check(
	    opts->conn->open_session(opts->conn, NULL, NULL, &session));

	/*
	 * Note: id is repeated as id2.  This makes it easier to
	 * identify the primary key in dumps of the index files.
	 */
	testutil_check(session->create(session, opts->uri,
	    "key_format=i,value_format=iiSii,"
	    "columns=(id,post,bal,extra,flag,id2)"));

	testutil_check(session->create(session, sharedopts->posturi,
	    "columns=(post)"));
	testutil_check(session->create(session, sharedopts->baluri,
	    "columns=(bal)"));
	testutil_check(session->create(session, sharedopts->flaguri,
	    "columns=(flag)"));

	/*
	 * Insert a single record with all items we need to
	 * call search() on, this makes our join logic easier.
	 */
	testutil_check(session->open_cursor(session, opts->uri, NULL, NULL,
	    &maincur));
	maincur->set_key(maincur, N_RECORDS);
	maincur->set_value(maincur, 54321, 0, "", 0, N_RECORDS);
	testutil_check(maincur->insert(maincur));
	testutil_check(maincur->close(maincur));

	for (i = 0; i < N_INSERT_THREAD; ++i) {
		insert_args[i].threadnum = i;
		insert_args[i].nthread = N_INSERT_THREAD;
		insert_args[i].testopts = opts;
		insert_args[i].sharedopts = sharedopts;
		testutil_check(pthread_create(&insert_tid[i], NULL,
		    thread_insert, (void *)&insert_args[i]));
	}

	for (i = 0; i < N_JOIN_THREAD; ++i) {
		join_args[i].threadnum = i;
		join_args[i].nthread = N_JOIN_THREAD;
		join_args[i].testopts = opts;
		join_args[i].sharedopts = sharedopts;
		testutil_check(pthread_create(&join_tid[i], NULL,
		    thread_join, (void *)&join_args[i]));
	}

	/*
	 * Wait for insert threads to finish.  When they
	 * are done, signal join threads to complete.
	 */
	for (i = 0; i < N_INSERT_THREAD; ++i)
		testutil_check(pthread_join(insert_tid[i], NULL));

	for (i = 0; i < N_JOIN_THREAD; ++i)
		join_args[i].done = 1;

	for (i = 0; i < N_JOIN_THREAD; ++i)
		testutil_check(pthread_join(join_tid[i], NULL));

	fprintf(stderr, "\n");
	for (i = 0; i < N_JOIN_THREAD; ++i) {
		fprintf(stderr, "  join thread %d did %d joins\n",
		    i, join_args[i].joins);
	}
	for (i = 0; i < N_INSERT_THREAD; ++i)
		fprintf(stderr,
		    "  insert thread %d did "
		    "%d inserts, %d removes, %d notfound, %d rollbacks\n",
		    i, insert_args[i].inserts, insert_args[i].removes,
		    insert_args[i].notfounds, insert_args[i].rollbacks);

	testutil_check(session->drop(session, sharedopts->posturi, NULL));
	testutil_check(session->drop(session, sharedopts->baluri, NULL));
	testutil_check(session->drop(session, sharedopts->flaguri, NULL));
	testutil_check(session->drop(session, opts->uri, NULL));
	testutil_check(session->close(session, NULL));
}

static void *thread_insert(void *arg)
{
	SHARED_OPTS *sharedopts;
	TEST_OPTS *opts;
	THREAD_ARGS *threadargs;
	WT_CURSOR *maincur;
	WT_RAND_STATE rnd;
	WT_SESSION *session;
	int bal, i, flag, key, post, ret;
	const char *extra = S1024;

	threadargs = (THREAD_ARGS *)arg;
	opts = threadargs->testopts;
	sharedopts = threadargs->sharedopts;
	testutil_check(__wt_random_init_seed(NULL, &rnd));

	testutil_check(opts->conn->open_session(
	    opts->conn, NULL, NULL, &session));

	testutil_check(session->open_cursor(session, opts->uri, NULL, NULL,
	    &maincur));

	for (i = 0; i < N_INSERT; i++) {
		/*
		 * Insert threads may stomp on each other's records;
		 * that's okay.
		 */
		key = (int)(__wt_random(&rnd) % N_RECORDS);
		maincur->set_key(maincur, key);
		if (sharedopts->remove)
			testutil_check(session->begin_transaction(session,
			    "isolation=snapshot"));
		if (sharedopts->remove && __wt_random(&rnd) % 5 == 0 &&
		    maincur->search(maincur) == 0) {
			/*
			 * Another thread can be removing at the
			 * same time.
			 */
			ret = maincur->remove(maincur);
			testutil_assert(ret == 0 ||
			    (N_INSERT_THREAD > 1 &&
			    (ret == WT_NOTFOUND || ret == WT_ROLLBACK)));
			if (ret == 0)
				threadargs->removes++;
			else if (ret == WT_NOTFOUND)
				threadargs->notfounds++;
			else if (ret == WT_ROLLBACK)
				threadargs->rollbacks++;
		} else {
			if (__wt_random(&rnd) % 2 == 0)
				post = 54321;
			else
				post = i % 100000;
			if (__wt_random(&rnd) % 2 == 0) {
				bal = -100;
				flag = 1;
			} else {
				bal = 1 + (i % 1000) * 100;
				flag = 0;
			}
			maincur->set_value(maincur, post, bal, extra, flag,
			    key);
			ret = maincur->insert(maincur);
			testutil_assert(ret == 0 ||
			    (N_INSERT_THREAD > 1 && ret == WT_ROLLBACK));
			testutil_check(maincur->reset(maincur));
			if (ret == 0)
				threadargs->inserts++;
			else if (ret == WT_ROLLBACK)
				threadargs->rollbacks++;
		}
		if (sharedopts->remove)
			testutil_check(session->commit_transaction(session,
			    NULL));
		if (i % 1000 == 0 && i != 0) {
			if (i % 10000 == 0)
				fprintf(stderr, "*");
			else
				fprintf(stderr, ".");
		}
	}
	testutil_check(maincur->close(maincur));
	testutil_check(session->close(session, NULL));
	return (NULL);
}

static void *thread_join(void *arg)
{
	SHARED_OPTS *sharedopts;
	TEST_OPTS *opts;
	THREAD_ARGS *threadargs;
	WT_CURSOR *postcur, *balcur, *flagcur, *joincur;
	WT_SESSION *session;
	int bal, flag, key, key2, post, ret;
	char cfg[128];
	char *extra;

	threadargs = (THREAD_ARGS *)arg;
	opts = threadargs->testopts;
	sharedopts = threadargs->sharedopts;

	testutil_check(opts->conn->open_session(
	    opts->conn, NULL, NULL, &session));

	testutil_check(session->open_cursor(
	    session, sharedopts->posturi, NULL, NULL, &postcur));
	testutil_check(session->open_cursor(
	    session, sharedopts->baluri, NULL, NULL, &balcur));
	testutil_check(session->open_cursor(
	    session, sharedopts->flaguri, NULL, NULL, &flagcur));

	for (threadargs->joins = 0; threadargs->done == 0;
	     threadargs->joins++) {
		testutil_check(session->open_cursor(
		    session, sharedopts->joinuri, NULL, NULL, &joincur));
		postcur->set_key(postcur, 54321);
		testutil_check(postcur->search(postcur));
		testutil_check(session->join(session, joincur, postcur,
		    "compare=eq"));

		balcur->set_key(balcur, 0);
		testutil_check(balcur->search(balcur));
		if (sharedopts->bloom)
			sprintf(cfg, "compare=lt,strategy=bloom,count=%d",
			    N_RECORDS);
		else
			sprintf(cfg, "compare=lt");
		testutil_check(session->join(session, joincur, balcur, cfg));

		flagcur->set_key(flagcur, 0);
		testutil_check(flagcur->search(flagcur));
		if (sharedopts->bloom)
			sprintf(cfg, "compare=eq,strategy=bloom,count=%d",
			    N_RECORDS);
		else
			sprintf(cfg, "compare=eq");
		testutil_check(session->join(session, joincur, flagcur, cfg));

		/* Expect no values returned */
		ret = joincur->next(joincur);
		if (ret == 0) {
			/*
			 * The values may already have been changed, but
			 * print them for informational purposes.
			 */
			testutil_check(joincur->get_key(joincur, &key));
			testutil_check(joincur->get_value(joincur, &post,
			    &bal, &extra, &flag, &key2));
			fprintf(stderr, "FAIL: iteration %d: "
			    "key=%d/%d, postal_code=%d, balance=%d, flag=%d\n",
			    threadargs->joins, key, key2, post, bal, flag);
			/* Save the results. */
			testutil_check(opts->conn->close(opts->conn, NULL));
			opts->conn = NULL;
			return (NULL);
		}
		testutil_assert(ret == WT_NOTFOUND);
		testutil_check(joincur->close(joincur));

		/*
		 * Reset the cursors, potentially allowing the insert
		 * threads to proceed.
		 */
		testutil_check(postcur->reset(postcur));
		testutil_check(balcur->reset(balcur));
		testutil_check(flagcur->reset(flagcur));
		if (threadargs->joins % 100 == 0)
			fprintf(stderr, "J");
	}
	testutil_check(postcur->close(postcur));
	testutil_check(balcur->close(balcur));
	testutil_check(flagcur->close(flagcur));
	testutil_check(session->close(session, NULL));
	return (NULL);
}