/* 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 #include #include #include #include #include #include #include #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(arg)); while (!threadStop) { op = rand_r(reinterpret_cast(&tv.tv_usec)) % 10; if (op < 8) // read { interval = rand_r(reinterpret_cast(&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(&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(&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(&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(arg)); } } delete rwlock; pthread_exit(0); } static void* RWRunner_local(void* arg) { struct timeval tv; int op, op2, interval; RWLock_local* rwlock = reinterpret_cast(arg); gettimeofday(&tv, NULL); while (!threadStop) { op = rand_r(reinterpret_cast(&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(&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(&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(&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(&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(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(&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 #include 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); }