AioLinuxScheduler Class Reference

AioLinuxScheduler implements DeviceAccessScheduler via Linux-specific kernel-mode libaio calls. More...

#include <AioLinuxScheduler.h>

Inheritance diagram for AioLinuxScheduler:

List of all members.

Public Member Functions
	AioLinuxScheduler (DeviceAccessSchedulerParams const &)
	Constructor.
virtual	~AioLinuxScheduler ()
	Destructor: stop must already have been called.
virtual void	registerDevice (SharedRandomAccessDevice pDevice)
	Registers a device for which this scheduler will process requests.
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	unregisterDevice (SharedRandomAccessDevice pDevice)
	Unregisters a device.
virtual void	start ()
	Spawns the OS thread.
void	join ()
	Waits for the OS thread to terminate.
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 Member Functions
bool	isStarted () const
	Returns: true if start has been called (and subsequent join has not completed)
bool	submitRequests (RandomAccessRequest::BindingList &bindingList)
	Submits a list of requests which have already been fully prepared.
void	deferLeftoverRequests (iocb **ppLeftovers, uint nLeftovers)
	Saves failed requests to the deferred queue for retry.
bool	retryDeferredRequests ()
	Retries submission of requests from the deferred queue; continues until either a submission attempt fails or the queue is exhausted.
Private Attributes
io_context_t	context
	Context for calling libaio.
AtomicCounter	nRequestsOutstanding
	Number of requests for which io_submit has been called and a corresponding io_getevents notification has not yet been processed.
bool	quit
	Flag for passively asking the scheduler to shut down.
std::deque< RandomAccessRequest::BindingList >	deferredQueue
	FIFO queue of requests which have been deferred due to not-fully-successful io_submit calls.
StrictMutex	deferredQueueMutex
	Mutex used to protect deferredQueue.

---

Detailed Description

AioLinuxScheduler implements DeviceAccessScheduler via Linux-specific kernel-mode libaio calls.

Author:

John V. Sichi

Version:

Id

//open/dev/fennel/device/AioLinuxScheduler.h#7

Definition at line 46 of file AioLinuxScheduler.h.

---

Constructor & Destructor Documentation

AioLinuxScheduler::AioLinuxScheduler

(

DeviceAccessSchedulerParams const &

)

[explicit]

Constructor.

Definition at line 42 of file AioLinuxScheduler.cpp.

References AtomicCounter::clear(), context, DeviceAccessSchedulerParams::maxRequests, nRequestsOutstanding, quit, and Thread::start().

{
    quit = false;
    nRequestsOutstanding.clear();
    context = NULL;
    int rc = io_queue_init(params.maxRequests, &context);
    if (rc) {
        throw SysCallExcn("io_queue_init failed");
    }

    // NOTE jvs 29-Oct-2005:  we ignore params.nThreads because
    // no more than one thread is needed for io_getevents
    Thread::start();
}

AioLinuxScheduler::~AioLinuxScheduler

(

)

[virtual]

Destructor: stop must already have been called.

Definition at line 63 of file AioLinuxScheduler.cpp.

References context, isStarted(), and nRequestsOutstanding.

{
    assert(!isStarted());
    assert(!nRequestsOutstanding);
    int rc = io_queue_release(context);
    if (rc) {
        throw SysCallExcn("io_queue_release failed");
    }
}

---

Member Function Documentation

bool AioLinuxScheduler::isStarted

(

)

const [inline, private]

Returns:

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

Reimplemented from Thread.

Definition at line 58 of file AioLinuxScheduler.cpp.

References Thread::isStarted().

Referenced by schedule(), stop(), and ~AioLinuxScheduler().

{
    return Thread::isStarted();
}

bool AioLinuxScheduler::submitRequests

(

RandomAccessRequest::BindingList &

bindingList

)

[private]

Submits a list of requests which have already been fully prepared.

Parameters:

bindingList

list of requests

Returns:

whether all requests were successfully submitted without requiring retry

Definition at line 91 of file AioLinuxScheduler.cpp.

References context, deferLeftoverRequests(), IntrusiveListMutator< T, DerivedListNode >::detach(), nRequestsOutstanding, and RawIntrusiveList::size().

Referenced by retryDeferredRequests(), and schedule().

{
    iocb *requestsArray[bindingList.size()];
    iocb **requests = requestsArray;

    // convert list to array
    int n = 0;
    RandomAccessRequest::BindingListMutator bindingMutator(bindingList);
    for (; bindingMutator; ++n) {
        RandomAccessRequestBinding *pBinding = bindingMutator.detach();
        requests[n] = pBinding;
    }

    if (n == 0) {
        // just in case someone asks for a nop
        return true;
    }

    // submit array
    int rc = io_submit(context, n, requests);
    if (rc == -EAGAIN) {
        rc = 0;
    }

    if (rc < 0) {
        // hard error
        throw SysCallExcn("io_submit failed");
    }

    // keep track of the number successfully submitted
    // (can't use += because nRequestsOutstanding is
    // an AtomicCounter)
    for (int i = 0; i < rc; ++i) {
        ++nRequestsOutstanding;
    }

    if (rc == n) {
        // we're done
        return true;
    } else {
        // io_submit is allowed to do less than we asked for, so
        // we need to resubmit some leftovers
        requests += rc;
        n -= rc;
        deferLeftoverRequests(requests, n);
        return false;
    }
}

void AioLinuxScheduler::deferLeftoverRequests	(	iocb **	ppLeftovers,
		uint	nLeftovers
	)			[private]

Saves failed requests to the deferred queue for retry.

Parameters:

	ppLeftovers	failed requests
	nLeftovers	number of failed requests

Definition at line 141 of file AioLinuxScheduler.cpp.

References deferredQueue, deferredQueueMutex, and IntrusiveList< T, DerivedListNode >::push_back().

Referenced by submitRequests().

{
    assert(nLeftovers > 0);

    // convert array back to list
    RandomAccessRequest::BindingList bindingList;

    for (uint i = 0; i < nLeftovers; ++i) {
        RandomAccessRequestBinding *pBinding =
            static_cast<RandomAccessRequestBinding *>(ppLeftovers[i]);
        bindingList.push_back(*pBinding);
    }

    StrictMutexGuard deferredQueueGuard(deferredQueueMutex);
    deferredQueue.push_back(bindingList);
}

bool AioLinuxScheduler::retryDeferredRequests

(

)

[private]

Retries submission of requests from the deferred queue; continues until either a submission attempt fails or the queue is exhausted.

Returns:

whether all deferred requests were successfully submitted (returns true if there were no deferred requests to begin with)

Definition at line 160 of file AioLinuxScheduler.cpp.

References deferredQueue, deferredQueueMutex, and submitRequests().

Referenced by run().

{
    for (;;) {
        StrictMutexGuard deferredQueueGuard(deferredQueueMutex);
        if (deferredQueue.empty()) {
            // all resubmitted successfully (or none to begin with)
            return true;
        }
        RandomAccessRequest::BindingList bindingList = deferredQueue.front();
        deferredQueue.pop_front();
        // release mutex now to avoid potential deadlocks
        deferredQueueGuard.unlock();

        bool success = submitRequests(bindingList);
        if (!success) {
            // at least one failed
            return false;
        }
    }
}

void AioLinuxScheduler::registerDevice

(

SharedRandomAccessDevice

pDevice

)

[virtual]

Registers a device for which this scheduler will process requests.

The default implementation does nothing.

Parameters:

pDevice

device to be registered

Reimplemented from DeviceAccessScheduler.

Definition at line 73 of file AioLinuxScheduler.cpp.

{
    int hFile = pDevice->getHandle();

    // Linux requires files accessed via libaio to be opened with O_DIRECT,
    // so force that now.
    int flags = fcntl(hFile, F_GETFL);
    fcntl(hFile, F_SETFL, flags | O_DIRECT);
}

bool AioLinuxScheduler::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 84 of file AioLinuxScheduler.cpp.

References RandomAccessRequest::bindingList, isStarted(), RandomAccessRequest::pDevice, RandomAccessDevice::prepareTransfer(), and submitRequests().

{
    assert(isStarted());
    request.pDevice->prepareTransfer(request);
    return submitRequests(request.bindingList);
}

void AioLinuxScheduler::stop

(

)

[virtual]

Shuts down, waiting for all pending requests to complete.

Implements DeviceAccessScheduler.

Definition at line 181 of file AioLinuxScheduler.cpp.

References isStarted(), Thread::join(), and quit.

{
    assert(isStarted());
    quit = true;

    Thread::join();
}

void AioLinuxScheduler::run

(

)

[virtual]

Implements Thread.

Definition at line 189 of file AioLinuxScheduler.cpp.

References context, RandomAccessRequestBinding::getBufferSize(), RandomAccessRequestBinding::notifyTransferCompletion(), nRequestsOutstanding, quit, and retryDeferredRequests().

{
    while (nRequestsOutstanding || !quit) {
        io_event event;
        timespec ts;

        // Check the deferred request queue before entering wait state.
        if (retryDeferredRequests()) {
            // If we retried any requests, they all succeeded, so we're in our
            // normal wait state: timeout every second to check the quit flag.
            ts.tv_sec = 1;
            ts.tv_nsec = 0;
        } else {
            // At least one retry just failed, so during wait, timeout in a
            // millisecond so we can retry the failed requests.
            ts.tv_sec = 0;
            ts.tv_nsec = 1000000;
        }

        long rc = io_getevents(context, 1, 1, &event, &ts);

        // NOTE jvs 20-Jan-2008:  Docs don't mention the possibility of
        // spurious interrupts, but they can occur, at least while
        // debugging with gdb, so treat them as timeout.
        if ((rc == 0) || (rc == -EINTR)) {
            // timed out
            continue;
        }

        if (rc != 1) {
            throw SysCallExcn("io_getevents failed");
        }
        RandomAccessRequestBinding *pBinding =
            static_cast<RandomAccessRequestBinding *>(event.obj);
        bool success = (pBinding->getBufferSize() == event.res)
            && !event.res2;
        pBinding->notifyTransferCompletion(success);
        --nRequestsOutstanding;
    }
}

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::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(), AioPollingScheduler::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(), AioPollingScheduler::stop(), stop(), ResourceTest::testConcurrency(), and LocalConditionTest::testNotifyAll().

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

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

io_context_t AioLinuxScheduler::context [private]

Context for calling libaio.

Definition at line 52 of file AioLinuxScheduler.h.

Referenced by AioLinuxScheduler(), run(), submitRequests(), and ~AioLinuxScheduler().

AtomicCounter AioLinuxScheduler::nRequestsOutstanding [private]

Number of requests for which io_submit has been called and a corresponding io_getevents notification has not yet been processed.

We track these since it is illegal to attempt to close the context while requests are still outstanding.

Definition at line 60 of file AioLinuxScheduler.h.

Referenced by AioLinuxScheduler(), run(), submitRequests(), and ~AioLinuxScheduler().

bool AioLinuxScheduler::quit [private]

Flag for passively asking the scheduler to shut down.

Definition at line 65 of file AioLinuxScheduler.h.

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

std::deque<RandomAccessRequest::BindingList> AioLinuxScheduler::deferredQueue [private]

FIFO queue of requests which have been deferred due to not-fully-successful io_submit calls.

Note that these do not yet count as outstanding. Front of deque is first-in; back of deque is last-in.

Definition at line 73 of file AioLinuxScheduler.h.

Referenced by deferLeftoverRequests(), and retryDeferredRequests().

StrictMutex AioLinuxScheduler::deferredQueueMutex [private]

Mutex used to protect deferredQueue.

Definition at line 78 of file AioLinuxScheduler.h.

Referenced by deferLeftoverRequests(), and retryDeferredRequests().

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/AioLinuxScheduler.h
• /home/pub/open/dev/fennel/device/AioLinuxScheduler.cpp

---