AioPollingScheduler Class Reference

AioPollingScheduler implements DeviceAccessScheduler via Unix aio calls and threads which poll for completion. More...

#include <AioPollingScheduler.h>

Inheritance diagram for AioPollingScheduler:

List of all members.

Public Member Functions
	AioPollingScheduler (DeviceAccessSchedulerParams const &)
	Constructor.
virtual	~AioPollingScheduler ()
	Destructor: stop must already have been called.
virtual bool	schedule (RandomAccessRequest &request)
	Initiates a request, the details of which must already have been defined by the caller.
virtual void	stop ()
	Shuts down, waiting for all pending requests to complete.
virtual void	run ()
virtual void	registerDevice (SharedRandomAccessDevice pDevice)
	Registers a device for which this scheduler will process requests.
virtual void	unregisterDevice (SharedRandomAccessDevice pDevice)
	Unregisters a device.
virtual void	start ()
	Spawns the OS thread.
void	join ()
	Waits for the OS thread to terminate.
bool	isStarted () const
	Returns: true if start has been called (and subsequent join has not completed)
bool	isStopped () const
	Returns: opposite of isStarted()
boost::thread &	getBoostThread ()
	Accesses the underlying boost::thread, e.g.
std::string	getName ()
void	setName (std::string const &s)
Static Public Member Functions
static DeviceAccessScheduler *	newScheduler (DeviceAccessSchedulerParams const &params)
	Creates a scheduler.
Protected Member Functions
void	initAndRun ()
virtual void	beforeRun ()
virtual void	afterRun ()
Protected Attributes
boost::thread *	pBoostThread
bool	bRunning
std::string	name
Private Attributes
StrictMutex	mutex
LocalCondition	newRequestPending
bool	quit
std::vector< aiocb * >	currentRequests
std::vector< aiocb * >	newRequests

---

Detailed Description

AioPollingScheduler implements DeviceAccessScheduler via Unix aio calls and threads which poll for completion.

Definition at line 43 of file AioPollingScheduler.h.

---

Constructor & Destructor Documentation

AioPollingScheduler::AioPollingScheduler

(

DeviceAccessSchedulerParams const &

)

[explicit]

Constructor.

Definition at line 35 of file AioPollingScheduler.cpp.

References quit, and Thread::start().

{
    // TODO:  pass params.maxRequests on to OS, and use
    // params.nThreads
    quit = false;
    Thread::start();
}

AioPollingScheduler::~AioPollingScheduler

(

)

[virtual]

Destructor: stop must already have been called.

Definition at line 44 of file AioPollingScheduler.cpp.

{
}

---

Member Function Documentation

bool AioPollingScheduler::schedule

(

RandomAccessRequest &

request

)

[virtual]

Initiates a request, the details of which must already have been defined by the caller.

When the request completes, this scheduler will call notifyTransferCompletion on each binding associated with the request, and also break up the binding list. The bindings must not be altered by the caller until this notification is received. However, the request parameter itself need not live beyond this call.

Care must be taken to ensure that the schedule/notify sequences cannot deadlock. For example, the caller of schedule may hold a lock on a binding, and the implementation of schedule may acquire a scheduler lock internally. The notification callback may also need to take a lock on the binding. Thus, it is important that no scheduler lock be held while notifyTransferCompletion is called.

Parameters:

request

parameters for the request to be scheduled

Returns:

true if the request was successfully scheduled without any retries

Implements DeviceAccessScheduler.

Definition at line 48 of file AioPollingScheduler.cpp.

References RandomAccessRequest::bindingList, IntrusiveListMutator< T, DerivedListNode >::detach(), mutex, newRequestPending, newRequests, RandomAccessRequest::pDevice, and RandomAccessDevice::prepareTransfer().

{
    StrictMutexGuard guard(mutex);
    uint iFirst = newRequests.size();
    request.pDevice->prepareTransfer(request);
    RandomAccessRequest::BindingListMutator bindingMutator(request.bindingList);
    while (bindingMutator) {
        RandomAccessRequestBinding *pBinding = bindingMutator.detach();
        pBinding->aio_sigevent.sigev_notify = SIGEV_NONE;
        newRequests.push_back(pBinding);
    }
    aiocb **pFirst = &(newRequests.front()) + iFirst;
    int rc = lio_listio(LIO_NOWAIT,pFirst,newRequests.size() - iFirst,NULL);
    // TODO:  handle error cases
    assert(rc == 0);
    newRequestPending.notify_all();
    return true;
}

void AioPollingScheduler::stop

(

)

[virtual]

Shuts down, waiting for all pending requests to complete.

Implements DeviceAccessScheduler.

Definition at line 67 of file AioPollingScheduler.cpp.

References Thread::join(), mutex, newRequestPending, and quit.

{
    StrictMutexGuard guard(mutex);
    quit = true;
    newRequestPending.notify_all();
    guard.unlock();
    Thread::join();
}

void AioPollingScheduler::run

(

)

[virtual]

Implements Thread.

Definition at line 76 of file AioPollingScheduler.cpp.

References currentRequests, mutex, newRequestPending, newRequests, RandomAccessRequestBinding::notifyTransferCompletion(), and quit.

{
    int rc;
    struct timespec ts;
    // poll every tenth of a millisecond
    // TODO:  determine a reasonable default, or adjust this
    // dynamically?
    ts.tv_sec = 0;
    ts.tv_nsec = 100000;
    for (;;) {
        StrictMutexGuard guard(mutex);
        while (!newRequests.size()) {
            if (quit) {
                return;
            }
            newRequestPending.wait(guard);
        }
        currentRequests.resize(newRequests.size());
        std::copy(
            newRequests.begin(),
            newRequests.end(),
            currentRequests.begin());
        newRequests.clear();
        guard.unlock();
        do {
            // REVIEW jvs 4-Aug-2004:  Using &front like this is not portable.
            rc = aio_suspend(
                &(currentRequests.front()),
                currentRequests.size(),
                &ts);
            if (rc) {
                switch (errno) {
                case EAGAIN:
                case EINTR:
                    continue;
                default:
                    std::cerr << rc << std::endl;
                    permAssert(false);
                }
            }
        } while (false);
        // currentRequests does not need a lock, since this thread is the only
        // one which manipulates it.  And the lock cannot be held when
        // notifyTransferCompletion is called, otherwise deadlock is possible.
        // However, access to newRequests does require a lock, since other
        // threads may be calling schedule, so use a fine-grained lock on it.
        for (uint i = 0; i < currentRequests.size(); ++i) {
            aiocb *pcb = currentRequests[i];
            RandomAccessRequestBinding *pBinding =
                static_cast<RandomAccessRequestBinding *>(pcb);
            rc = aio_error(pcb);
            if (rc == EINPROGRESS) {
                guard.lock();
                // cout << "EINPROGRESS " << pcb->aio_offset << std::endl;
                newRequests.push_back(pcb);
                guard.unlock();
            } else {
                rc = aio_return(pcb);
                // guard.lock();
                // cout << "complete " << pcb->aio_offset << " " << rc
                // << std::endl;
                // guard.unlock();
                pBinding->notifyTransferCompletion(rc >= 0);
            }
        }
        currentRequests.clear();
    }
}

DeviceAccessScheduler * DeviceAccessScheduler::newScheduler

(

DeviceAccessSchedulerParams const &

params

)

[static, inherited]

Creates a scheduler.

Parameters:

params

DeviceAccessSchedulerParams to use

Returns:

new scheduler; caller is responsible for deleting it

Definition at line 70 of file DeviceAccessScheduler.cpp.

References DeviceAccessSchedulerParams::AIO_LINUX_SCHEDULER, DeviceAccessSchedulerParams::AIO_POLLING_SCHEDULER, DeviceAccessSchedulerParams::AIO_SIGNAL_SCHEDULER, dlopenAioLinuxScheduler(), DeviceAccessSchedulerParams::IO_COMPLETION_PORT_SCHEDULER, DeviceAccessSchedulerParams::schedulerType, DeviceAccessSchedulerParams::THREAD_POOL_SCHEDULER, and DeviceAccessSchedulerParams::usingDefaultSchedulerType.

Referenced by CacheImpl< PageT, VictimPolicyT >::CacheImpl(), and RandomAccessFileDeviceTest::testAsyncIOImpl().

{
    switch (params.schedulerType) {
    case DeviceAccessSchedulerParams::THREAD_POOL_SCHEDULER:
        return new ThreadPoolScheduler(params);

#ifdef __MSVC__
    case DeviceAccessSchedulerParams::IO_COMPLETION_PORT_SCHEDULER:
        return new IoCompletionPortScheduler(params);
#endif

#ifdef USE_LIBAIO_H
    case DeviceAccessSchedulerParams::AIO_LINUX_SCHEDULER:
        {
            DeviceAccessScheduler *pScheduler = dlopenAioLinuxScheduler(params);
            if (pScheduler) {
                return pScheduler;
            } else {
                // if the aioLinux scheduler was explicitly selected (vs simply
                // using the default type for the OS), then the AIO runtime
                // library must be installed; otherwise, fall through to use
                // ThreadPoolScheduler as fallback
                if (params.usingDefaultSchedulerType) {
                    break;
                }
                throw FennelExcn(FennelResource::instance().libaioRequired());
            }
        }
#endif

#ifdef USE_AIO_H
    case DeviceAccessSchedulerParams::AIO_POLLING_SCHEDULER:
        return new AioPollingScheduler(params);
    case DeviceAccessSchedulerParams::AIO_SIGNAL_SCHEDULER:
        return new AioSignalScheduler(params);
#endif

    default:
        // fall through to use ThreadPoolScheduler as a fallback
        break;
    }
    return new ThreadPoolScheduler(params);
}

void DeviceAccessScheduler::registerDevice

(

SharedRandomAccessDevice

pDevice

)

[virtual, inherited]

Registers a device for which this scheduler will process requests.

The default implementation does nothing.

Parameters:

pDevice

device to be registered

Reimplemented in AioLinuxScheduler, and IoCompletionPortScheduler.

Definition at line 142 of file DeviceAccessScheduler.cpp.

Referenced by RandomAccessFileDeviceTest::testAsyncIOImpl().

{
}

void DeviceAccessScheduler::unregisterDevice

(

SharedRandomAccessDevice

pDevice

)

[virtual, inherited]

Unregisters a device.

The default implementation does nothing.

Parameters:

pDevice

device to be unregistered

Definition at line 147 of file DeviceAccessScheduler.cpp.

Referenced by RandomAccessFileDeviceTest::testAsyncIOImpl().

{
}

void Thread::initAndRun

(

)

[protected, inherited]

Definition at line 66 of file Thread.cpp.

References Thread::afterRun(), Thread::beforeRun(), and Thread::run().

Referenced by Thread::start().

{
    beforeRun();
    run();
    afterRun();
}

void Thread::beforeRun

(

)

[protected, virtual, inherited]

Definition at line 73 of file Thread.cpp.

References Thread::bRunning.

Referenced by Thread::initAndRun().

{
    bRunning = true;
}

void Thread::afterRun

(

)

[protected, virtual, inherited]

Definition at line 78 of file Thread.cpp.

References Thread::bRunning.

Referenced by Thread::initAndRun().

{
    bRunning = false;
}

void Thread::start

(

)

[virtual, inherited]

Spawns the OS thread.

Definition at line 50 of file Thread.cpp.

References Thread::initAndRun(), and Thread::pBoostThread.

Referenced by AioLinuxScheduler::AioLinuxScheduler(), AioPollingScheduler(), StatsTimer::start(), ResourceTest::testConcurrency(), and LocalConditionTest::testNotifyAll().

{
    pBoostThread = new boost::thread(
        boost::bind(&Thread::initAndRun,this));
}

void Thread::join

(

)

[inherited]

Waits for the OS thread to terminate.

Definition at line 56 of file Thread.cpp.

References Thread::pBoostThread.

Referenced by CheckpointThread::closeImpl(), TimerThread::stop(), stop(), AioLinuxScheduler::stop(), ResourceTest::testConcurrency(), and LocalConditionTest::testNotifyAll().

{
    assert(pBoostThread);
    boost::thread t;
    assert(*pBoostThread != t);
    pBoostThread->join();
    delete pBoostThread;
    pBoostThread = NULL;
}

bool Thread::isStarted

(

)

const [inline, inherited]

Returns:

true if start has been called (and subsequent join has not completed)

Reimplemented in AioLinuxScheduler.

Definition at line 71 of file Thread.h.

Referenced by CheckpointThread::closeImpl(), CacheImpl< PageT, VictimPolicyT >::closeImpl(), AioLinuxScheduler::isStarted(), and TimerThread::stop().

    {
        return pBoostThread ? true : false;
    }

bool Thread::isStopped

(

)

const [inline, inherited]

Returns:

opposite of isStarted()

Definition at line 79 of file Thread.h.

    {
        return !isStarted();
    }

boost::thread& Thread::getBoostThread

(

)

[inline, inherited]

Accesses the underlying boost::thread, e.g.

for use in a boost::thread_group. This thread must already be started.

Returns:

the underlying boost::thread

Definition at line 90 of file Thread.h.

    {
        assert(isStarted());
        return *pBoostThread;
    }

std::string Thread::getName

(

)

[inline, inherited]

Definition at line 96 of file Thread.h.

    {
        return name;
    }

void Thread::setName

(

std::string const &

s

)

[inline, inherited]

Definition at line 101 of file Thread.h.

    {
        name = s;
    }

---

Member Data Documentation

StrictMutex AioPollingScheduler::mutex [private]

Definition at line 46 of file AioPollingScheduler.h.

Referenced by run(), schedule(), and stop().

LocalCondition AioPollingScheduler::newRequestPending [private]

Definition at line 47 of file AioPollingScheduler.h.

Referenced by run(), schedule(), and stop().

bool AioPollingScheduler::quit [private]

Definition at line 48 of file AioPollingScheduler.h.

Referenced by AioPollingScheduler(), run(), and stop().

std::vector<aiocb *> AioPollingScheduler::currentRequests [private]

Definition at line 50 of file AioPollingScheduler.h.

Referenced by run().

std::vector<aiocb *> AioPollingScheduler::newRequests [private]

Definition at line 51 of file AioPollingScheduler.h.

Referenced by run(), and schedule().

boost::thread* Thread::pBoostThread [protected, inherited]

Definition at line 44 of file Thread.h.

Referenced by Thread::join(), Thread::start(), Thread::Thread(), and Thread::~Thread().

bool Thread::bRunning [protected, inherited]

Definition at line 45 of file Thread.h.

Referenced by Thread::afterRun(), Thread::beforeRun(), Thread::Thread(), and Thread::~Thread().

std::string Thread::name [protected, inherited]

Definition at line 46 of file Thread.h.

Referenced by Thread::Thread().

---

The documentation for this class was generated from the following files:

• /home/pub/open/dev/fennel/device/AioPollingScheduler.h
• /home/pub/open/dev/fennel/device/AioPollingScheduler.cpp

---