ThreadPoolScheduler Class Reference

ThreadPoolScheduler implements DeviceAccessScheduler by combining a thread pool with synchronous I/O calls. More...

#include <ThreadPoolScheduler.h>

Inheritance diagram for ThreadPoolScheduler:

List of all members.

Public Member Functions
	ThreadPoolScheduler (DeviceAccessSchedulerParams const &)
	Constructor.
virtual	~ThreadPoolScheduler ()
	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	registerDevice (SharedRandomAccessDevice pDevice)
	Registers a device for which this scheduler will process requests.
virtual void	unregisterDevice (SharedRandomAccessDevice pDevice)
	Unregisters a device.
Static Public Member Functions
static DeviceAccessScheduler *	newScheduler (DeviceAccessSchedulerParams const &params)
	Creates a scheduler.
Private Attributes
ThreadPool< RandomAccessRequest >	pool

---

Detailed Description

ThreadPoolScheduler implements DeviceAccessScheduler by combining a thread pool with synchronous I/O calls.

Definition at line 37 of file ThreadPoolScheduler.h.

---

Constructor & Destructor Documentation

ThreadPoolScheduler::ThreadPoolScheduler

(

DeviceAccessSchedulerParams const &

)

[explicit]

Constructor.

Definition at line 34 of file ThreadPoolScheduler.cpp.

References DeviceAccessSchedulerParams::maxRequests, pool, and ThreadPoolBase::start().

{
    // threads and requests are 1-to-1, but threads are expensive,
    // so arbitrarily cap at 10
    uint nThreads = std::min<uint>(10, params.maxRequests);
    pool.start(nThreads);
}

ThreadPoolScheduler::~ThreadPoolScheduler

(

)

[virtual]

Destructor: stop must already have been called.

Definition at line 43 of file ThreadPoolScheduler.cpp.

{
}

---

Member Function Documentation

bool ThreadPoolScheduler::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 47 of file ThreadPoolScheduler.cpp.

References RandomAccessRequest::bindingList, RandomAccessRequest::cbOffset, RandomAccessRequest::cbTransfer, IntrusiveListMutator< T, DerivedListNode >::detach(), RandomAccessRequestBinding::getBufferSize(), RandomAccessRequest::pDevice, pool, ThreadPool< Task >::submitTask(), and RandomAccessRequest::type.

{
    RandomAccessRequest::BindingListMutator bindingMutator(request.bindingList);
    FileSize cbOffset = request.cbOffset;
    // break up the request into one per binding
    // TODO:  don't do this if device supports scatter/gather; and skip
    // breakup if only one binding in the first place
    while (bindingMutator) {
        RandomAccessRequestBinding *pBinding = bindingMutator.detach();
        if (!pBinding) {
            break;
        }
        RandomAccessRequest subRequest;
        subRequest.pDevice = request.pDevice;
        subRequest.cbOffset = cbOffset;
        subRequest.cbTransfer = pBinding->getBufferSize();
        cbOffset += subRequest.cbTransfer;
        subRequest.type = request.type;
        subRequest.bindingList.push_back(*pBinding);
        pool.submitTask(subRequest);
    }
    assert(cbOffset == request.cbOffset + request.cbTransfer);
    return true;
}

void ThreadPoolScheduler::stop

(

)

[virtual]

Shuts down, waiting for all pending requests to complete.

Implements DeviceAccessScheduler.

Definition at line 72 of file ThreadPoolScheduler.cpp.

References pool, and ThreadPoolBase::stop().

{
    pool.stop();
}

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().

{
}

---

Member Data Documentation

ThreadPool<RandomAccessRequest> ThreadPoolScheduler::pool [private]

Definition at line 40 of file ThreadPoolScheduler.h.

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

---

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

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

---