mariadb-columnstore-engine/utils/rwlock/tdriver-rw.cpp

/* Copyright (C) 2014 InfiniDB, Inc.

   This program is free software; you can redistribute it and/or
   modify it under the terms of the GNU General Public License
   as published by the Free Software Foundation; version 2 of
   the License.

   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 General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
   MA 02110-1301, USA. */

/*****************************************************************************
 * $Id$
 *
 ****************************************************************************/

/** @file
 * Brief description of the file contents
 *
 * More detailed description
 */

#include <iostream>
#include <sys/types.h>
#include <stdexcept>
#include <stdlib.h>
#include <sys/time.h>
#include <unistd.h>
#include <pthread.h>

#include <cppunit/extensions/HelperMacros.h>

#include "rwlock.h"
#include "rwlock_local.h"

using namespace std;
using namespace rwlock;

int threadStop;

static void *RWRunner(void *arg)
{
	struct timeval tv;
	int op, op2, interval;
	RWLock* rwlock;
	
	gettimeofday(&tv, NULL);
	rwlock = new RWLock(reinterpret_cast<int64_t>(arg));
	
	while (!threadStop) {
		op = rand_r(reinterpret_cast<uint32_t *>(&tv.tv_usec)) % 10;
		if (op < 8) {   // read
				interval = rand_r(reinterpret_cast<uint32_t *>(&tv.tv_usec)) % 100000;
				rwlock->read_lock();
				rwlock->lock();
				CPPUNIT_ASSERT(rwlock->getReading() > 0);
				CPPUNIT_ASSERT(rwlock->getWriting() == 0);
				rwlock->unlock();
				usleep(interval);

				op2 = rand_r(reinterpret_cast<uint32_t *>(&tv.tv_usec)) % 2;
				if (op2) {
					rwlock->upgrade_to_write();
					rwlock->lock();
					CPPUNIT_ASSERT(rwlock->getReading() == 0);
					CPPUNIT_ASSERT(rwlock->getWriting() == 1);
					rwlock->unlock();
					usleep(interval);
					rwlock->write_unlock();
				}
				else {
					/* For testing the lock recovery code in the BRM workernodes */
					/*
					int crash = rand_r((uint32_t *) &tv.tv_usec) % 100;
					if (crash > 0)   // 1% chance of crashing
						rwlock->read_unlock();
					*/
				}
		}
		else if (op < 9) {    // write
				interval = rand_r(reinterpret_cast<uint32_t *>(&tv.tv_usec)) % 100000;
				rwlock->write_lock();
				rwlock->lock();
				CPPUNIT_ASSERT(rwlock->getReading() == 0);
				CPPUNIT_ASSERT(rwlock->getWriting() == 1);
				rwlock->unlock();
				usleep(interval);

				op2 = rand_r(reinterpret_cast<uint32_t *>(&tv.tv_usec)) % 2;
				if (op2) {
					rwlock->downgrade_to_read();
					rwlock->lock();
					CPPUNIT_ASSERT(rwlock->getReading() > 0);
					CPPUNIT_ASSERT(rwlock->getWriting() == 0);
					rwlock->unlock();
					usleep(interval);
					rwlock->read_unlock();
				}
				else		

					rwlock->write_unlock();
		}
		else if (op == 9) {   // delete
				delete rwlock;
				rwlock = new RWLock(reinterpret_cast<int64_t>(arg));
		}
	}
	delete rwlock;
	pthread_exit(0);
}
				
static void *RWRunner_local(void *arg)
{
	struct timeval tv;
	int op, op2, interval;
	RWLock_local* rwlock = reinterpret_cast<RWLock_local *>(arg);
	
	gettimeofday(&tv, NULL);
	
	while (!threadStop) {
		op = rand_r(reinterpret_cast<uint32_t *>(&tv.tv_usec)) % 10;
//    		cout << "doing op " << op << endl;
		switch(op) {
			case 0:    //read
			case 1:
			case 2:
			case 3:
			case 4:
			case 5:
			case 6:
			case 7:
			case 8:
			{
				interval = rand_r(reinterpret_cast<uint32_t *>(&tv.tv_usec)) % 100000;
				rwlock->read_lock();
				rwlock->lock();
				CPPUNIT_ASSERT(rwlock->getReading() > 0);
				CPPUNIT_ASSERT(rwlock->getWriting() == 0);
				rwlock->unlock();
 				usleep(interval);
				op2 = rand_r(reinterpret_cast<uint32_t *>(&tv.tv_usec)) % 2;
				if (op2) {
					rwlock->upgrade_to_write();
// 					rwlock->lock();
					CPPUNIT_ASSERT(rwlock->getReading() == 0);
					CPPUNIT_ASSERT(rwlock->getWriting() == 1);
// 					rwlock->unlock();
 					usleep(interval);
					rwlock->write_unlock();
				}
				break;
			}
			case 9:		//write
			{
				interval = rand_r(reinterpret_cast<uint32_t *>(&tv.tv_usec)) % 100000;
				rwlock->write_lock();
// 				rwlock->lock();
				CPPUNIT_ASSERT(rwlock->getReading() == 0);
				CPPUNIT_ASSERT(rwlock->getWriting() == 1);
// 				rwlock->unlock();
 				usleep(interval);
				op2 = rand_r(reinterpret_cast<uint32_t *>(&tv.tv_usec)) % 2;
				if (op2) {
					rwlock->downgrade_to_read();
					rwlock->lock();
					CPPUNIT_ASSERT(rwlock->getReading() > 0);
					CPPUNIT_ASSERT(rwlock->getWriting() == 0);
					rwlock->unlock();
 					usleep(interval);
					rwlock->read_unlock();
				}
				else		
					rwlock->write_unlock();
				break;
			}
			default:
				break;
		}
	}
	pthread_exit(0);
}


class RWLockTest : public CppUnit::TestFixture {

CPPUNIT_TEST_SUITE(RWLockTest);

CPPUNIT_TEST(LongRWTest_1);
//CPPUNIT_TEST(LongRWLocalTest_1);

CPPUNIT_TEST_SUITE_END();

private:
public:
	void LongRWTest_1() {
		int key = 0x20000;  // the extentmap key
		
		const int threadCount = 30;
		int i;
		pthread_t threads[threadCount];
	
		cerr << endl << "Multithreaded RWLock test.  "
				"This runs for 60 minutes." << endl;
	
		threadStop = 0;
		
		for (i = 0; i < threadCount; i++) {
			if (pthread_create(&threads[i], NULL, RWRunner, 
				reinterpret_cast<void *>(key)) < 0)
				throw logic_error("Error creating threads for the ipc test");
		}
		
 		sleep(3600);
		threadStop = 1;
		for (i = 0; i < threadCount; i++) {
			cerr << "Waiting for thread #" << i << endl;
			pthread_join(threads[i], NULL);	
		}
	}

	void LongRWLocalTest_1() {
		const int threadCount = 40;
		int i;
		pthread_t threads[threadCount];
		RWLock_local rwlock;
	
		cerr << endl << "Multithreaded RWLock_local test.  "
				"This runs for 30-60 seconds." << endl;
	
		threadStop = 0;
		
		for (i = 0; i < threadCount; i++) {
			if (pthread_create(&threads[i], NULL, RWRunner_local, 
				reinterpret_cast<void *>(&rwlock)) < 0)
				throw logic_error("Error creating threads for the local test");
		}
		
		sleep(30);
		threadStop = 1;
		for (i = 0; i < threadCount; i++) {
			cerr << "Waiting for thread #" << i << endl;
			pthread_join(threads[i], NULL);	
		}
	}

};

CPPUNIT_TEST_SUITE_REGISTRATION( RWLockTest );

#include <cppunit/extensions/TestFactoryRegistry.h>
#include <cppunit/ui/text/TestRunner.h>

int main( int argc, char **argv)
{
  CppUnit::TextUi::TestRunner runner;
  CppUnit::TestFactoryRegistry &registry = CppUnit::TestFactoryRegistry::getRegistry();
  runner.addTest( registry.makeTest() );
  bool wasSuccessful = runner.run( "", false );
  return (wasSuccessful ? 0 : 1);
}