For this test we use the MutexSemaphoreGroup and related type to emulate a double exchange example:
Thread 1 (main) : Performs the housekeeping and data source Thread 2 (child) : Is the data sink
Both threads have a control semaphore which the other thread uses to control access to a resource , in this case the the data counter gvalue. They are also used as a synchronization mechanism,
A third semaphore is used as a flag/signal to indicate that the threads should drop out of the source->sink loop.
/*
CoreLinux++
Copyright (C) 2000 CoreLinux Consortium
The CoreLinux++ Library is free software; you can redistribute it and/or
modify it under the terms of the GNU Library General Public License as
published by the Free Software Foundation; either version 2 of the
License, or (at your option) any later version.
The CoreLinux++ Library Library 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
Library General Public License for more details.
You should have received a copy of the GNU Library General Public
License along with the GNU C Library; see the file COPYING.LIB. If not,
write to the Free Software Foundation, Inc., 59 Temple Place - Suite 330,
Boston, MA 02111-1307, USA.
*/
//
// For clone()
//
extern "C"
{
#include <signal.h>
#include <sys/wait.h>
#include <sched.h>
}
#include <corelinux/Common.hpp>
#include <corelinux/AbstractSemaphore.hpp>
#include <corelinux/MutexSemaphoreGroup.hpp>
using namespace corelinux;
#include <iostream>
#include <exception>
//
// In module function prototypes
//
int main( void );
//
// General Functions
//
void handleAssertion( AssertionCref );
void handleException( ExceptionCref );
void handleStatus( SemaphoreOperationStatus );
//
// Thread entry point declarations
//
Int threadTest( VoidPtr );
//
// Global data
//
const int STACKSIZE(8192); // Thread stack size
SemaphoreIdentifier parentSemId(0); // Identifier for control
SemaphoreIdentifier childSemId(1); // Identifier for control
SemaphoreIdentifier completeSemId(2); // Identifier for control
ThreadIdentifier badThread(-1); // Use for testing
Dword gvalue(0); // Resource being protected
MutexSemaphoreGroupPtr agGroup( NULLPTR );
//
// Main entry point
//
int main( void )
{
cout << endl;
AbstractSemaphorePtr amutexParent( NULLPTR );
AbstractSemaphorePtr amutexChild( NULLPTR );
AbstractSemaphorePtr amutexFlag( NULLPTR );
//
// Practice graceful exception management
//
try
{
//
// Setup our SemaphoreGroup and control Semaphores
//
agGroup = new MutexSemaphoreGroup
(
3, OWNER_ALL | GROUP_ALL | PUBLIC_ALL
);
//
// Start with two (2) locks enabled, and one not
// the flag.
//
amutexParent = agGroup->createLockedSemaphore
(
parentSemId, FAIL_IF_EXISTS
);
amutexChild = agGroup->createLockedSemaphore
(
childSemId, FAIL_IF_EXISTS
);
amutexFlag = agGroup->createSemaphore
(
completeSemId, FAIL_IF_EXISTS
);
//
// Setup thread stack and point to bottom
//
VoidPtr *newstack((VoidPtr*) new Byte[STACKSIZE]);
newstack = (void **)(STACKSIZE + (char *) newstack);
//
// Spawn thread
// threadTest - entry point int (fn*func)(void*arg)
// newstack - bos
// flags - Share everything
// agGroup - SemaphoreGroupPtr (arg)
//
ThreadIdentifier aChildId =
clone
(
threadTest,
newstack,
CLONE_VM | CLONE_FS | CLONE_FILES | CLONE_SIGHAND | SIGCHLD,
agGroup
);
//
// Check for errors
//
if( aChildId != badThread )
{
//
// Signal handling would be a better mechanism, but for now
// loop a few
//
bool childError( false );
ThreadIdentifier myId(Thread::getThreadIdentifier());
for( int x = 0; x < 10 ; ++x )
{
if( amutexFlag->isLocked() == false )
{
//
// Do some work
//
++gvalue;
cout << "[" << IntRef(myId) <<
"]\tWriting resource = " << gvalue << endl;
// Prepare the last resource read
if( gvalue == 10 )
{
amutexFlag->lockWithWait();
amutexParent->release();
}
else
{
amutexParent->release();
amutexChild->lockWithWait();
}
}
else
{
childError = true;
cout << "Child aborted run." << endl;
x = 10;
}
}
Int aStatus(0);
waitpid(aChildId.getScalar(),&aStatus,0);
//
// Clear semaphores
//
amutexParent->release();
amutexChild->release();
amutexFlag->release();
}
else
{
cerr << "Error number from thread" << Thread::getKernelError() << endl;
handleStatus(amutexParent->release());
}
//
// Housekeeping
//
agGroup->destroySemaphore(amutexChild);
amutexChild = NULLPTR;
agGroup->destroySemaphore(amutexParent);
amutexParent = NULLPTR;
agGroup->destroySemaphore(amutexFlag);
amutexParent = NULLPTR;
delete agGroup;
agGroup = NULLPTR;
}
catch( SemaphoreExceptionRef aRef )
{
handleException(aRef);
cerr << "Error number " << aRef.getErrNum() << endl;
}
catch( AssertionRef aAssert )
{
handleAssertion(aAssert);
}
catch( ExceptionRef aException )
{
handleException(aException);
}
catch( std::exception & e )
{
cerr << e.what() << endl;
}
catch( ... )
{
cerr << "Unknown exception." << endl;
}
//
// If we have not gone through the housekeeping
//
if( agGroup != NULLPTR )
{
if( amutexParent != NULLPTR )
{
agGroup->destroySemaphore(amutexParent);
}
else
{
; // do nothing
}
if( amutexChild != NULLPTR )
{
agGroup->destroySemaphore(amutexChild);
}
else
{
; // do nothing
}
if( amutexFlag != NULLPTR )
{
agGroup->destroySemaphore(amutexFlag);
}
else
{
; // do nothing
}
delete agGroup;
}
return 0;
}
//
// Child thread implementation
//
Int threadTest( void *aV )
{
AbstractSemaphorePtr aParent( NULLPTR );
AbstractSemaphorePtr aChild( NULLPTR );
AbstractSemaphorePtr aFlag( NULLPTR );
ThreadIdentifier myId(Thread::getThreadIdentifier());
try
{
SemaphoreGroupPtr aSG( (SemaphoreGroupPtr)aV );
aParent = aSG->createSemaphore( parentSemId, FAIL_IF_NOTEXISTS );
aChild = aSG->createSemaphore( childSemId, FAIL_IF_NOTEXISTS );
aFlag = aSG->createSemaphore( completeSemId, FAIL_IF_NOTEXISTS );
while( aFlag->isLocked() == false )
{
aParent->lockWithWait();
cout << "\t[" << IntRef(myId) <<
"]\tReading resource = " << gvalue << endl;
aChild->release();
}
}
catch(SemaphoreExceptionRef aSemExcp )
{
cerr << "Thread bailing out with exception" << endl;
handleException( aSemExcp );
aFlag->lockWithWait();
}
catch( ... )
{
cerr << "Thread bailing out with exception" << endl;
aFlag->lockWithWait();
}
return 0;
}
//
// Peform defaults (just show it)
//
void handleStatus( SemaphoreOperationStatus aStatus )
{
switch( aStatus )
{
case SUCCESS:
cerr << "Success" << endl;
break;
case BALKED:
cerr << "Request balked" << endl;
break;
case TIMEDOUT:
cerr << "Request timed out" << endl;
break;
case UNAVAILABLE:
cerr << "Request not satisfied" << endl;
break;
case KERNELERROR:
cerr << "Kernel reported error : " << Thread::getKernelError() << endl;
break;
default:
NEVER_GET_HERE;
break;
}
return;
}
void handleAssertion( AssertionCref aAssert )
{
cerr << aAssert.getFile() << ":" << aAssert.getLine() << ":" <<
"Assertion: ";
if( aAssert.getType() == Assertion::NEVERGETHERE )
{
cerr << "NEVER_GET_HERE";
}
else
{
if( aAssert.getType() == Assertion::REQUIRE )
{
cerr << "REQUIRE";
}
else if( aAssert.getType() == Assertion::ENSURE )
{
cerr << "ENSURE";
}
else if( aAssert.getType() == Assertion::CHECK )
{
cerr << "CHECK";
}
else
{
cerr << "ASSERT";
}
cerr << "( " << aAssert.getWhy() << " )";
}
cerr << endl;
}
void handleException( ExceptionCref aExcp )
{
cerr << aExcp.getFile() << ":" << aExcp.getLine() << ":" <<
"Exception: " << aExcp.getWhy() << endl;
}
/*
Common rcs information do not modify
$Author: frankc $
$Revision: 1.9 $
$Date: 2000/09/07 12:54:44 $
$Locker: $
*/