linuxaio_queue.cpp

Go to the documentation of this file.

/***************************************************************************
 * lib/io/linuxaio_queue.cpp
 *
 * Part of the STXXL. See http://stxxl.sourceforge.net
 *
 * Copyright (C) 2011 Johannes Singler <[email protected]>
 * Copyright (C) 2014 Timo Bingmann <[email protected]>
 *
 * Distributed under the Boost Software License, Version 1.0.
 * (See accompanying file LICENSE_1_0.txt or copy at
 * http://www.boost.org/LICENSE_1_0.txt)
 **************************************************************************/

#include <stxxl/bits/io/linuxaio_queue.h>

#if STXXL_HAVE_LINUXAIO_FILE

#include <unistd.h>
#include <sys/syscall.h>

#include <stxxl/bits/verbose.h>
#include <stxxl/bits/mng/block_manager.h>
#include <stxxl/bits/common/error_handling.h>
#include <stxxl/bits/io/linuxaio_request.h>
#include <stxxl/bits/io/linuxaio_queue.h>

#include <algorithm>

#ifndef STXXL_CHECK_FOR_PENDING_REQUESTS_ON_SUBMISSION
#define STXXL_CHECK_FOR_PENDING_REQUESTS_ON_SUBMISSION 1
#endif

STXXL_BEGIN_NAMESPACE

linuxaio_queue::linuxaio_queue(int desired_queue_length)
 : num_waiting_requests(0), num_free_events(0), num_posted_requests(0),
 post_thread_state(NOT_RUNNING), wait_thread_state(NOT_RUNNING)
{
 if (desired_queue_length == 0) {
 // default value, 64 entries per queue (i.e. usually per disk) should
 // be enough
 max_events = 64;
 }
 else
 max_events = desired_queue_length;

 // negotiate maximum number of simultaneous events with the OS
 context = 0;
 long result;
 while ((result = syscall(SYS_io_setup, max_events, &context)) == -1 &&
 errno == EAGAIN && max_events > 1)
 {
 max_events <<= 1; // try with half as many events
 }
 if (result != 0) {
 STXXL_THROW_ERRNO(io_error, "linuxaio_queue::linuxaio_queue"
 " io_setup() nr_events=" << max_events);
 }

 for (int e = 0; e < max_events; ++e)
 num_free_events++; // cannot set semaphore to value directly

 STXXL_MSG("Set up an linuxaio queue with " << max_events << " entries.");

 start_thread(post_async, static_cast<void*>(this), post_thread, post_thread_state);
 start_thread(wait_async, static_cast<void*>(this), wait_thread, wait_thread_state);
}

linuxaio_queue::~linuxaio_queue()
{
 stop_thread(post_thread, post_thread_state, num_waiting_requests);
 stop_thread(wait_thread, wait_thread_state, num_posted_requests);
 syscall(SYS_io_destroy, context);
}

void linuxaio_queue::add_request(request_ptr& req)
{
 if (req.empty())
 STXXL_THROW_INVALID_ARGUMENT("Empty request submitted to disk_queue.");
 if (post_thread_state() != RUNNING)
 STXXL_ERRMSG("Request submitted to stopped queue.");
 if (!dynamic_cast<linuxaio_request*>(req.get()))
 STXXL_ERRMSG("Non-LinuxAIO request submitted to LinuxAIO queue.");

 scoped_mutex_lock lock(waiting_mtx);

 waiting_requests.push_back(req);
 num_waiting_requests++;
}

bool linuxaio_queue::cancel_request(request_ptr& req)
{
 if (req.empty())
 STXXL_THROW_INVALID_ARGUMENT("Empty request canceled disk_queue.");
 if (post_thread_state() != RUNNING)
 STXXL_ERRMSG("Request canceled in stopped queue.");
 if (!dynamic_cast<linuxaio_request*>(req.get()))
 STXXL_ERRMSG("Non-LinuxAIO request submitted to LinuxAIO queue.");

 queue_type::iterator pos;
 {
 scoped_mutex_lock lock(waiting_mtx);

 pos = std::find(waiting_requests.begin(), waiting_requests.end(),
 req _STXXL_FORCE_SEQUENTIAL);
 if (pos != waiting_requests.end())
 {
 waiting_requests.erase(pos);

 // polymorphic_downcast to linuxaio_request,
 // request is canceled, but was not yet posted.
 dynamic_cast<linuxaio_request*>(pos->get())->completed(false, true);

 num_waiting_requests--; // will never block
 return true;
 }
 }

 scoped_mutex_lock lock(posted_mtx);

 pos = std::find(posted_requests.begin(), posted_requests.end(),
 req _STXXL_FORCE_SEQUENTIAL);
 if (pos != posted_requests.end())
 {
 // polymorphic_downcast to linuxaio_request,
 bool canceled_io_operation = (dynamic_cast<linuxaio_request*>(req.get()))->cancel_aio();

 if (canceled_io_operation)
 {
 posted_requests.erase(pos);

 // polymorphic_downcast to linuxaio_request,

 // request is canceled, already posted
 dynamic_cast<linuxaio_request*>(pos->get())->completed(true, true);

 num_free_events++;
 num_posted_requests--; // will never block
 return true;
 }
 }

 return false;
}

// internal routines, run by the posting thread
void linuxaio_queue::post_requests()
{
 request_ptr req;
 io_event* events = new io_event[max_events];

 for ( ; ; ) // as long as thread is running
 {
 // might block until next request or message comes in
 int num_currently_waiting_requests = num_waiting_requests--;

 // terminate if termination has been requested
 if (post_thread_state() == TERMINATING && num_currently_waiting_requests == 0)
 break;

 scoped_mutex_lock lock(waiting_mtx);
 if (!waiting_requests.empty())
 {
 req = waiting_requests.front();
 waiting_requests.pop_front();
 lock.unlock();

 num_free_events--; // might block because too many requests are posted

 // polymorphic_downcast
 while (!dynamic_cast<linuxaio_request*>(req.get())->post())
 {
 // post failed, so first handle events to make queues (more)
 // empty, then try again.

 // wait for at least one event to complete, no time limit
 long num_events = syscall(SYS_io_getevents, context, 1, max_events, events, NULL);
 if (num_events < 0) {
 STXXL_THROW_ERRNO(io_error, "linuxaio_queue::post_requests"
 " io_getevents() nr_events=" << num_events);
 }

 handle_events(events, num_events, false);
 }

 // request is finally posted

 {
 scoped_mutex_lock lock(posted_mtx);
 posted_requests.push_back(req);
 num_posted_requests++;
 }
 }
 else
 {
 lock.unlock();

 // num_waiting_requests-- was premature, compensate for that
 num_waiting_requests++;
 }
 }

 delete[] events;
}

void linuxaio_queue::handle_events(io_event* events, long num_events, bool canceled)
{
 for (int e = 0; e < num_events; ++e)
 {
 // unsigned_type is as long as a pointer, and like this, we avoid an icpc warning
 request_ptr* r = reinterpret_cast<request_ptr*>(static_cast<unsigned_type>(events[e].data));
 r->get()->completed(canceled);
 delete r; // release auto_ptr reference
 num_free_events++;
 num_posted_requests--; // will never block
 }
}

// internal routines, run by the waiting thread
void linuxaio_queue::wait_requests()
{
 request_ptr req;
 io_event* events = new io_event[max_events];

 for ( ; ; ) // as long as thread is running
 {
 // might block until next request is posted or message comes in
 int num_currently_posted_requests = num_posted_requests--;

 // terminate if termination has been requested
 if (wait_thread_state() == TERMINATING && num_currently_posted_requests == 0)
 break;

 // wait for at least one of them to finish
 long num_events = syscall(SYS_io_getevents, context, 1, max_events, events, NULL);
 if (num_events < 0) {
 STXXL_THROW_ERRNO(io_error, "linuxaio_queue::wait_requests"
 " io_getevents() nr_events=" << max_events);
 }

 num_posted_requests++; // compensate for the one eaten prematurely above

 handle_events(events, num_events, false);
 }

 delete[] events;
}

void* linuxaio_queue::post_async(void* arg)
{
 (static_cast<linuxaio_queue*>(arg))->post_requests();

 self_type* pthis = static_cast<self_type*>(arg);
 pthis->post_thread_state.set_to(TERMINATED);

#if STXXL_STD_THREADS && STXXL_MSVC >= 1700
 // Workaround for deadlock bug in Visual C++ Runtime 2012 and 2013, see
 // request_queue_impl_worker.cpp. -tb
 ExitThread(NULL);
#else
 return NULL;
#endif
}

void* linuxaio_queue::wait_async(void* arg)
{
 (static_cast<linuxaio_queue*>(arg))->wait_requests();

 self_type* pthis = static_cast<self_type*>(arg);
 pthis->wait_thread_state.set_to(TERMINATED);

#if STXXL_STD_THREADS && STXXL_MSVC >= 1700
 // Workaround for deadlock bug in Visual C++ Runtime 2012 and 2013, see
 // request_queue_impl_worker.cpp. -tb
 ExitThread(NULL);
#else
 return NULL;
#endif
}

STXXL_END_NAMESPACE

#endif // #if STXXL_HAVE_LINUXAIO_FILE
// vim: et:ts=4:sw=4