io.h

Go to the documentation of this file.

/* Copyright (c) 2011 Wildfire Games
 *
 * Permission is hereby granted, free of charge, to any person obtaining
 * a copy of this software and associated documentation files (the
 * "Software"), to deal in the Software without restriction, including
 * without limitation the rights to use, copy, modify, merge, publish,
 * distribute, sublicense, and/or sell copies of the Software, and to
 * permit persons to whom the Software is furnished to do so, subject to
 * the following conditions:
 * 
 * The above copyright notice and this permission notice shall be included
 * in all copies or substantial portions of the Software.
 * 
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
 * IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
 * CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
 * TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
 * SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */

/*
 * provide asynchronous and synchronous I/O with hooks to allow
 * overlapped processing or progress reporting.
 */

#ifndef INCLUDED_IO
#define INCLUDED_IO

#include "lib/config2.h"
#include "lib/alignment.h"
#include "lib/bits.h"
#include "lib/timer.h"
#include "lib/file/file.h"
#include "lib/sysdep/filesystem.h"  // wtruncate
#include "lib/posix/posix_aio.h"    // LIO_READ, LIO_WRITE

#include "lib/allocators/unique_range.h"

namespace ERR
{
    const Status IO = -110301;
}

namespace io {

// @return memory suitable for use as an I/O buffer (address is a
// multiple of alignment, size is rounded up to a multiple of alignment)
// @param alignment is automatically increased if smaller than the
// UniqueRange requirement.
//
// use this instead of the file cache for write buffers that are
// never reused (avoids displacing other items).
static inline UniqueRange Allocate(size_t size, size_t alignment = maxSectorSize)
{
    return RVALUE(AllocateAligned(size, alignment));
}


#pragma pack(push, 1)

// required information for any I/O (this is basically the same as aiocb,
// but also applies to synchronous I/O and has shorter/nicer names.)
struct Operation
{
    // @param buf can be 0, in which case temporary block buffers are allocated.
    // otherwise, it must be aligned and padded to the I/O alignment, e.g. via
    // io::Allocate.
    Operation(const File& file, void* buf, off_t size, off_t offset = 0)
        : fd(file.Descriptor()), opcode((file.Flags() & O_WRONLY)? LIO_WRITE : LIO_READ)
        , offset(offset), size(size), buf((void*)buf)
    {
    }

    void Validate() const
    {
        ENSURE(fd >= 0);
        ENSURE(opcode == LIO_READ || opcode == LIO_WRITE);

        ENSURE(offset >= 0);
        ENSURE(size >= 0);
        // buf can legitimately be 0 (see above)
    }

    int fd;
    int opcode;

    off_t offset;
    off_t size;
    void* buf;
};


// optional information how an Operation is to be carried out
struct Parameters
{
    // default to single blocking I/Os
    Parameters()
        : alignment(1)  // no alignment requirements
        , blockSize(0)  // do not split into blocks
        , queueDepth(1) // disable aio
    {
    }

    // parameters for asynchronous I/O that maximize throughput on current drives
    struct OverlappedTag {};
    Parameters(OverlappedTag)
        : alignment(maxSectorSize), blockSize(128*KiB), queueDepth(32)
    {
    }

    Parameters(size_t blockSize, size_t queueDepth, off_t alignment = maxSectorSize)
        : alignment(alignment), blockSize(blockSize), queueDepth(queueDepth)
    {
    }

    void Validate(const Operation& op) const
    {
        ENSURE(is_pow2(alignment));
        ENSURE(alignment > 0);

        if(blockSize != 0)
        {
            ENSURE(is_pow2(blockSize));
            ENSURE(pageSize <= blockSize);  // (don't bother checking an upper bound)
        }

        ENSURE(1 <= queueDepth && queueDepth <= maxQueueDepth);

        ENSURE(IsAligned(op.offset, alignment));
        // op.size doesn't need to be aligned
        ENSURE(IsAligned(op.buf, alignment));
    }

    // (ATTO only allows 10, which improves upon 8)
    static const size_t maxQueueDepth = 32;

    off_t alignment;

    size_t blockSize;   // 0 for one big "block"

    size_t queueDepth;
};

#define IO_OVERLAPPED io::Parameters(io::Parameters::OverlappedTag())


struct DefaultCompletedHook
{
    /**
     * called after a block I/O has completed.
     * @return Status (see RETURN_STATUS_FROM_CALLBACK).
     *
     * allows progress notification and processing data while waiting for
     * previous I/Os to complete.
     **/
    Status operator()(const u8* UNUSED(block), size_t UNUSED(blockSize)) const
    {
        return INFO::OK;
    }
};


struct DefaultIssueHook
{
    /**
     * called before a block I/O is issued.
     * @return Status (see RETURN_STATUS_FROM_CALLBACK).
     *
     * allows generating the data to write while waiting for
     * previous I/Os to complete.
     **/
    Status operator()(aiocb& UNUSED(cb)) const
    {
        return INFO::OK;
    }
};


// ring buffer of partially initialized aiocb that can be passed
// directly to aio_write etc. after setting offset and buffer.
class ControlBlockRingBuffer
{
public:
    ControlBlockRingBuffer(const Operation& op, const Parameters& p)
        : controlBlocks()   // zero-initialize
    {
        const size_t blockSize = p.blockSize? p.blockSize : (size_t)op.size;

        const bool temporaryBuffersRequested = (op.buf == 0);
        if(temporaryBuffersRequested)
            buffers = RVALUE(io::Allocate(blockSize * p.queueDepth, p.alignment));

        for(size_t i = 0; i < ARRAY_SIZE(controlBlocks); i++)
        {
            aiocb& cb = operator[](i);
            cb.aio_fildes = op.fd;
            cb.aio_nbytes = blockSize;
            cb.aio_lio_opcode = op.opcode;
            if(temporaryBuffersRequested)
                cb.aio_buf = (volatile void*)(uintptr_t(buffers.get()) + i * blockSize);
        }
    }

    INLINE aiocb& operator[](size_t counter)
    {
        return controlBlocks[counter % ARRAY_SIZE(controlBlocks)];
    }

private:
    UniqueRange buffers;
    aiocb controlBlocks[Parameters::maxQueueDepth];
};

#pragma pack(pop)


LIB_API Status Issue(aiocb& cb, size_t queueDepth);
LIB_API Status WaitUntilComplete(aiocb& cb, size_t queueDepth);


//-----------------------------------------------------------------------------
// Run

#ifndef ENABLE_IO_STATS
#define ENABLE_IO_STATS 0
#endif

// (hooks must be passed by const reference to allow passing rvalues.
// functors with non-const member data can mark them as mutable.)
template<class CompletedHook, class IssueHook>
static inline Status Run(const Operation& op, const Parameters& p = Parameters(), const CompletedHook& completedHook = CompletedHook(), const IssueHook& issueHook = IssueHook())
{
    op.Validate();
    p.Validate(op);

    ControlBlockRingBuffer controlBlockRingBuffer(op, p);

#if ENABLE_IO_STATS
    const double t0 = timer_Time();
    COMPILER_FENCE;
#endif

    const off_t numBlocks = p.blockSize? (off_t)DivideRoundUp((u64)op.size, (u64)p.blockSize) : 1;
    for(off_t blocksIssued = 0, blocksCompleted = 0; blocksCompleted < numBlocks; blocksCompleted++)
    {
        for(; blocksIssued != numBlocks && blocksIssued < blocksCompleted + (off_t)p.queueDepth; blocksIssued++)
        {
            aiocb& cb = controlBlockRingBuffer[blocksIssued];
            cb.aio_offset = op.offset + blocksIssued * p.blockSize;
            if(op.buf)
                cb.aio_buf = (volatile void*)(uintptr_t(op.buf) + blocksIssued * p.blockSize);
            if(blocksIssued == numBlocks-1)
                cb.aio_nbytes = round_up(size_t(op.size - blocksIssued * p.blockSize), size_t(p.alignment));

            RETURN_STATUS_FROM_CALLBACK(issueHook(cb));

            RETURN_STATUS_IF_ERR(Issue(cb, p.queueDepth));
        }

        aiocb& cb = controlBlockRingBuffer[blocksCompleted];
        RETURN_STATUS_IF_ERR(WaitUntilComplete(cb, p.queueDepth));

        RETURN_STATUS_FROM_CALLBACK(completedHook((u8*)cb.aio_buf, cb.aio_nbytes));
    }

#if ENABLE_IO_STATS
    COMPILER_FENCE;
    const double t1 = timer_Time();
    const off_t totalSize = p.blockSize? numBlocks*p.blockSize : op.size;
    debug_printf(L"IO: %.2f MB/s (%.2f)\n", totalSize/(t1-t0)/1e6, (t1-t0)*1e3);
#endif

    return INFO::OK;
}

// (overloads allow omitting parameters without requiring a template argument list)
template<class CompletedHook>
static inline Status Run(const Operation& op, const Parameters& p = Parameters(), const CompletedHook& completedHook = CompletedHook())
{
    return Run(op, p, completedHook, DefaultIssueHook());
}

static inline Status Run(const Operation& op, const Parameters& p = Parameters())
{
    return Run(op, p, DefaultCompletedHook(), DefaultIssueHook());
}


//-----------------------------------------------------------------------------
// Store

// efficient writing requires preallocation; the resulting file is
// padded to the sector size and needs to be truncated afterwards.
// this function takes care of both.
template<class CompletedHook, class IssueHook>
static inline Status Store(const OsPath& pathname, const void* data, size_t size, const Parameters& p = Parameters(), const CompletedHook& completedHook = CompletedHook(), const IssueHook& issueHook = IssueHook())
{
    File file;
    int oflag = O_WRONLY;
    if(p.queueDepth != 1)
        oflag |= O_DIRECT;
    RETURN_STATUS_IF_ERR(file.Open(pathname, oflag));
    io::Operation op(file, (void*)data, size);

#if OS_WIN
    (void)waio_Preallocate(op.fd, (off_t)size);
#endif

    RETURN_STATUS_IF_ERR(io::Run(op, p, completedHook, issueHook));

    file.Close();   // (required by wtruncate)

    RETURN_STATUS_IF_ERR(wtruncate(pathname, size));

    return INFO::OK;
}

template<class CompletedHook>
static inline Status Store(const OsPath& pathname, const void* data, size_t size, const Parameters& p = Parameters(), const CompletedHook& completedHook = CompletedHook())
{
    return Store(pathname, data, size, p, completedHook, DefaultIssueHook());
}

static inline Status Store(const OsPath& pathname, const void* data, size_t size, const Parameters& p = Parameters())
{
    return Store(pathname, data, size, p, DefaultCompletedHook(), DefaultIssueHook());
}


//-----------------------------------------------------------------------------
// Load

// convenience function provided for symmetry with Store.
template<class CompletedHook, class IssueHook>
static inline Status Load(const OsPath& pathname, void* buf, size_t size, const Parameters& p = Parameters(), const CompletedHook& completedHook = CompletedHook(), const IssueHook& issueHook = IssueHook())
{
    File file;
    int oflag = O_RDONLY;
    if(p.queueDepth != 1)
        oflag |= O_DIRECT;
    RETURN_STATUS_IF_ERR(file.Open(pathname, oflag));
    io::Operation op(file, buf, size);
    return io::Run(op, p, completedHook, issueHook);
}

template<class CompletedHook>
static inline Status Load(const OsPath& pathname, void* buf, size_t size, const Parameters& p = Parameters(), const CompletedHook& completedHook = CompletedHook())
{
    return Load(pathname, buf, size, p, completedHook, DefaultIssueHook());
}

static inline Status Load(const OsPath& pathname, void* buf, size_t size, const Parameters& p = Parameters())
{
    return Load(pathname, buf, size, p, DefaultCompletedHook(), DefaultIssueHook());
}

}   // namespace io

#endif  // #ifndef INCLUDED_IO