/***
* ==++==
*
* Copyright (c) Microsoft Corporation. All rights reserved.
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
* ==--==
* =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
*
* Asynchronous I/O: stream buffer implementation details
*
* We're going to some lengths to avoid exporting C++ class member functions and implementation details across
* module boundaries, and the factoring requires that we keep the implementation details away from the main header
* files. The supporting functions, which are in this file, use C-like signatures to avoid as many issues as
* possible.
*
* For the latest on this and related APIs, please see http://casablanca.codeplex.com.
*
* =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
****/
#include "stdafx.h"
#include "cpprest/details/fileio.h"

using namespace boost::asio;
using namespace Concurrency::streams::details;

namespace Concurrency { namespace streams { namespace details {

/***
* ==++==
*
* Implementation details of the file stream buffer
*
* =-=-=-
****/

/// <summary>
/// The public parts of the file information record contain only what is implementation-
/// independent. The actual allocated record is larger and has details that the implementation
/// require in order to function.
/// </summary>
struct _file_info_impl : _file_info
{
    _file_info_impl(int handle, std::ios_base::openmode mode, bool buffer_reads) :
        _file_info(mode, 512),
        m_handle(handle),
        m_buffer_reads(buffer_reads),
        m_outstanding_writes(0)
    {
    }

    /// <summary>
    /// The file handle of the file
    /// </summary>
    int  m_handle;

    bool m_buffer_reads;

    /// <summary>
    /// A list of callback waiting to be signalled that there are no outstanding writes.
    /// </summary>
    std::vector<_filestream_callback *> m_sync_waiters;

    std::atomic<long> m_outstanding_writes;
};

}}}

/// <summary>
/// Perform post-CreateFile processing.
/// </summary>
/// <param name="fh">The Win32 file handle</param>
/// <param name="callback">The callback interface pointer</param>
/// <param name="mode">The C++ file open mode</param>
/// <returns>The error code if there was an error in file creation.</returns>
bool _finish_create(int fh, _filestream_callback *callback, std::ios_base::openmode mode, int /* prot */)
{
    if (fh != -1)
    {
        // Buffer reads internally if and only if we're just reading (not also writing) and
        // if the file is opened exclusively. If either is false, we're better off just
        // letting the OS do its buffering, even if it means that prompt reads won't
        // happen.
        bool buffer = (mode == std::ios_base::in);

        // seek to end if requested
        if (mode & std::ios_base::ate)
        {
            lseek(fh, 0, SEEK_END);
        }

        auto info = new _file_info_impl(fh, mode, buffer);

        if (mode & std::ios_base::app || mode & std::ios_base::ate)
        {
            info->m_wrpos = static_cast<size_t>(-1); // Start at the end of the file.
        }

        callback->on_opened(info);
        return true;
    }
    else
    {
        callback->on_error(std::make_exception_ptr(utility::details::create_system_error(errno)));
        return false;
    }
}

int get_open_flags(std::ios_base::openmode mode)
{
    int result = 0;
    if (mode & std::ios_base::in)
    {
        if (mode & std::ios_base::out)
        {
            result = O_RDWR;
        }
        else
        {
            result = O_RDONLY;
        }
    }
    else if (mode & std::ios_base::out)
    {
        result = O_WRONLY|O_CREAT;
    }

    if (mode & std::ios_base::app)
    {
        result |= O_APPEND;
    }

    if (mode & std::ios_base::trunc)
    {
        result |= O_TRUNC|O_CREAT;
    }

    return result;
}

/// <summary>
/// Open a file and create a streambuf instance to represent it.
/// </summary>
/// <param name="callback">A pointer to the callback interface to invoke when the file has been opened.</param>
/// <param name="filename">The name of the file to open</param>
/// <param name="mode">A creation mode for the stream buffer</param>
/// <param name="prot">A file protection mode to use for the file stream</param>
/// <returns>True if the opening operation could be initiated, false otherwise.</returns>
/// <remarks>
/// True does not signal that the file will eventually be successfully opened, just that the process was started.
/// </remarks>
bool _open_fsb_str(_filestream_callback *callback, const char *filename, std::ios_base::openmode mode, int prot)
{
    if ( callback == nullptr || filename == nullptr) return false;

    std::string name(filename);

    pplx::create_task([=]() -> void
    {
        int cmode = get_open_flags(mode);
        if(cmode==O_RDWR)
        {
            cmode |= O_CREAT;
        }

        int f = open(name.c_str(), cmode, 0600);

        _finish_create(f, callback, mode, prot);
    });

    return true;
}

/// <summary>
/// Close a file stream buffer.
/// </summary>
/// <param name="info">The file info record of the file</param>
/// <param name="callback">A pointer to the callback interface to invoke when the file has been opened.</param>
/// <returns>True if the closing operation could be initiated, false otherwise.</returns>
/// <remarks>
/// True does not signal that the file will eventually be successfully closed, just that the process was started.
/// </remarks>
bool _close_fsb_nolock(_file_info **info, Concurrency::streams::details::_filestream_callback *callback)
{
    if ( callback == nullptr ) return false;
    if ( info == nullptr || *info == nullptr ) return false;

    _file_info_impl *fInfo = static_cast<_file_info_impl *>(*info);

    if ( fInfo->m_handle == -1 ) return false;

    // Since closing a file may involve waiting for outstanding writes which can take some time
    // if the file is on a network share, the close action is done in a separate task, as
    // CloseHandle doesn't have I/O completion events.
    pplx::create_task([=] () -> void
        {
            bool result = false;

            {
                pplx::extensibility::scoped_recursive_lock_t lock(fInfo->m_lock);

                if ( fInfo->m_handle != -1 )
                {
                    result = close(fInfo->m_handle) != -1;
                }

                if ( fInfo->m_buffer != nullptr )
                {
                    delete[] fInfo->m_buffer;
                }
            }

            delete fInfo;
            if (result)
            {
                callback->on_closed();
            }
            else
            {
                callback->on_error(std::make_exception_ptr(utility::details::create_system_error(errno)));
            }
        });

    *info = nullptr;

    return true;
}

bool _close_fsb(_file_info **info, Concurrency::streams::details::_filestream_callback *callback)
{
    if ( callback == nullptr ) return false;
    if ( info == nullptr || *info == nullptr ) return false;

    pplx::extensibility::scoped_recursive_lock_t lock((*info)->m_lock);

    return _close_fsb_nolock(info, callback);
}

/// <summary>
/// Initiate an asynchronous (overlapped) write to the file stream.
/// </summary>
/// <param name="info">The file info record of the file</param>
/// <param name="callback">A pointer to the callback interface to invoke when the write request is completed.</param>
/// <param name="ptr">A pointer to the data to write</param>
/// <param name="count">The size (in bytes) of the data</param>
/// <returns>0 if the write request is still outstanding, -1 if the request failed, otherwise the size of the data written</returns>
size_t _write_file_async(Concurrency::streams::details::_file_info_impl *fInfo, Concurrency::streams::details::_filestream_callback *callback, const void *ptr, size_t count, size_t position)
{
    ++fInfo->m_outstanding_writes;
 
    pplx::create_task([=]() -> void
    {
        off_t abs_position;
        bool must_restore_pos;
        off_t orig_pos;
        if( position == static_cast<size_t>(-1) )
        {
            orig_pos = lseek(fInfo->m_handle, 0, SEEK_CUR);
            abs_position = lseek(fInfo->m_handle, 0, SEEK_END);
            must_restore_pos = true;
        }
        else
        {
            abs_position = position;
            orig_pos = 0;
            must_restore_pos = false;
        }

        auto bytes_written = pwrite(fInfo->m_handle, ptr, count, abs_position);
        if (bytes_written == -1)
        {
            callback->on_error(std::make_exception_ptr(utility::details::create_system_error(errno)));
        }

        if(must_restore_pos)
        {
            lseek(fInfo->m_handle, orig_pos, SEEK_SET);
        }

        callback->on_completed(bytes_written);

        {
            pplx::extensibility::scoped_recursive_lock_t lock(fInfo->m_lock);

            // Decrement the counter of outstanding write events.
            if ( --fInfo->m_outstanding_writes == 0 )
            {
                // If this was the last one, signal all objects waiting for it to complete.

                for (auto iter = fInfo->m_sync_waiters.begin(); iter != fInfo->m_sync_waiters.end(); iter++)
                {
                    (*iter)->on_completed(0);
                }
                fInfo->m_sync_waiters.clear();
            }
        }
    });

    return 0;
}

/// <summary>
/// Initiate an asynchronous (overlapped) read from the file stream.
/// </summary>
/// <param name="info">The file info record of the file</param>
/// <param name="callback">A pointer to the callback interface to invoke when the write request is completed.</param>
/// <param name="ptr">A pointer to a buffer where the data should be placed</param>
/// <param name="count">The size (in bytes) of the buffer</param>
/// <param name="offset">The offset in the file to read from</param>
/// <returns>0 if the read request is still outstanding, -1 if the request failed, otherwise the size of the data read into the buffer</returns>
size_t _read_file_async(Concurrency::streams::details::_file_info_impl *fInfo, Concurrency::streams::details::_filestream_callback *callback, void *ptr, size_t count, size_t offset)
{
    pplx::create_task([=]() -> void
    {
        auto bytes_read = pread(fInfo->m_handle, ptr, count, offset);
        if (bytes_read < 0)
        {
            callback->on_error(std::make_exception_ptr(utility::details::create_system_error(errno)));
        }
        else
        {
            callback->on_completed(bytes_read);
        }
    });

    return 0;
}

template<typename Func>
class _filestream_callback_fill_buffer : public _filestream_callback
{
public:
    _filestream_callback_fill_buffer(_file_info *info, _filestream_callback *callback, const Func &func) : m_info(info), m_func(func), m_callback(callback) { }

    virtual void on_completed(size_t result) override
    {
        m_func(result);
        delete this;
    }
    virtual void on_error(const std::exception_ptr &e) override
    {
        auto exptr = std::make_exception_ptr(e);
        m_callback->on_error(exptr);
        delete this;
    }
private:
    _file_info *m_info;
    Func        m_func;
    _filestream_callback *m_callback;
};

template<typename Func>
_filestream_callback_fill_buffer<Func> *create_callback(_file_info *info, _filestream_callback *callback, const Func &func)
{
    return new _filestream_callback_fill_buffer<Func>(info, callback, func);
}

static const size_t PageSize = 512;

size_t _fill_buffer_fsb(_file_info_impl *fInfo, _filestream_callback *callback, size_t count, size_t charSize)
{
    size_t byteCount = count * charSize;
    if ( fInfo->m_buffer == nullptr )
    {
        fInfo->m_bufsize = std::max(PageSize, byteCount);
        fInfo->m_buffer = new char[static_cast<size_t>(fInfo->m_bufsize)];
        fInfo->m_bufoff = fInfo->m_rdpos;

        auto cb = create_callback(fInfo, callback,
            [=] (size_t result)
            {
                pplx::extensibility::scoped_recursive_lock_t lock(fInfo->m_lock);
                fInfo->m_buffill = result / charSize;
                callback->on_completed(result);
            });

        return _read_file_async(fInfo, cb, (uint8_t *)fInfo->m_buffer, fInfo->m_bufsize, fInfo->m_rdpos * charSize);
    }

    // First, we need to understand how far into the buffer we have already read
    // and how much remains.

    size_t bufpos = fInfo->m_rdpos - fInfo->m_bufoff;
    size_t bufrem = fInfo->m_buffill - bufpos;

    if ( bufrem < count )
    {
        fInfo->m_bufsize = std::max(PageSize, byteCount);

        // Then, we allocate a new buffer.

        char *newbuf = new char[static_cast<size_t>(fInfo->m_bufsize)];

        // Then, we copy the unread part to the new buffer and delete the old buffer

        if ( bufrem > 0 )
            memcpy(newbuf, fInfo->m_buffer + bufpos * charSize, bufrem * charSize);

        delete[] fInfo->m_buffer;
        fInfo->m_buffer = newbuf;

        // Then, we read the remainder of the count into the new buffer
        fInfo->m_bufoff = fInfo->m_rdpos;

        auto cb = create_callback(fInfo, callback,
            [=] (size_t result)
            {
                pplx::extensibility::scoped_recursive_lock_t lock(fInfo->m_lock);
                fInfo->m_buffill = result / charSize;
                callback->on_completed(result + bufrem * charSize);
            });

        return _read_file_async(fInfo, cb, (uint8_t*)fInfo->m_buffer + bufrem * charSize, fInfo->m_bufsize - bufrem * charSize, (fInfo->m_rdpos+bufrem)*charSize);
    }
    else
        return byteCount;
}


/// <summary>
/// Read data from a file stream into a buffer
/// </summary>
/// <param name="info">The file info record of the file</param>
/// <param name="callback">A pointer to the callback interface to invoke when the write request is completed.</param>
/// <param name="ptr">A pointer to a buffer where the data should be placed</param>
/// <param name="count">The size (in bytes) of the buffer</param>
/// <returns>0 if the read request is still outstanding, -1 if the request failed, otherwise the size of the data read into the buffer</returns>
size_t _getn_fsb(Concurrency::streams::details::_file_info *info, Concurrency::streams::details::_filestream_callback *callback, void *ptr, size_t count, size_t charSize)
{
    if ( callback == nullptr || info == nullptr ) return static_cast<size_t>(-1);

    _file_info_impl *fInfo = (_file_info_impl *)info;

    pplx::extensibility::scoped_recursive_lock_t lock(info->m_lock);

    if ( fInfo->m_handle == -1 ) return static_cast<size_t>(-1);

    size_t byteCount = count * charSize;

    if ( fInfo->m_buffer_reads )
    {
        auto cb = create_callback(fInfo, callback,
            [=] (size_t read)
            {
                auto copy = std::min(read, byteCount);
                auto bufoff = fInfo->m_rdpos - fInfo->m_bufoff;
                memcpy(ptr, fInfo->m_buffer + bufoff * charSize, copy);
                fInfo->m_atend = copy < byteCount;
                callback->on_completed(copy);
            });

        size_t read = _fill_buffer_fsb(fInfo, cb, count, charSize);

        if ( static_cast<int>(read) > 0 )
        {
            auto copy = std::min(read, byteCount);
            auto bufoff = fInfo->m_rdpos - fInfo->m_bufoff;
            memcpy(ptr, fInfo->m_buffer + bufoff * charSize, copy);
            fInfo->m_atend = copy < byteCount;
            return copy;
        }

        return read;
    }
    else
    {
        return _read_file_async(fInfo, callback, ptr, count, fInfo->m_rdpos * charSize);
    }
}

/// <summary>
/// Write data from a buffer into the file stream.
/// </summary>
/// <param name="info">The file info record of the file</param>
/// <param name="callback">A pointer to the callback interface to invoke when the write request is completed.</param>
/// <param name="ptr">A pointer to a buffer where the data should be placed</param>
/// <param name="count">The size (in bytes) of the buffer</param>
/// <returns>0 if the read request is still outstanding, -1 if the request failed, otherwise the size of the data read into the buffer</returns>
size_t _putn_fsb(Concurrency::streams::details::_file_info *info, Concurrency::streams::details::_filestream_callback *callback, const void *ptr, size_t count, size_t charSize)
{
    if (callback == nullptr || info == nullptr) return static_cast<size_t>(-1);

    _file_info_impl *fInfo = static_cast<_file_info_impl *>(info);

    pplx::extensibility::scoped_recursive_lock_t lock(fInfo->m_lock);

    if ( fInfo->m_handle == -1 ) return static_cast<size_t>(-1);

    size_t byteSize = count * charSize;

    // To preserve the async write order, we have to move the write head before read.
    auto lastPos = fInfo->m_wrpos;
    if (fInfo->m_wrpos != static_cast<size_t>(-1))
    {
        fInfo->m_wrpos += count;
        lastPos *= charSize;
    }

    return _write_file_async(fInfo, callback, ptr, byteSize, lastPos);
}

/// <summary>
/// Flush all buffered data to the underlying file.
/// </summary>
/// <param name="info">The file info record of the file</param>
/// <param name="callback">A pointer to the callback interface to invoke when the write request is completed.</param>
/// <returns>True if the request was initiated</returns>
bool _sync_fsb(Concurrency::streams::details::_file_info *info, Concurrency::streams::details::_filestream_callback *callback)
{
    if ( callback == nullptr ) return false;
    if ( info == nullptr ) return false;

    _file_info_impl *fInfo = static_cast<_file_info_impl *>(info);

    pplx::extensibility::scoped_recursive_lock_t lock(fInfo->m_lock);

    if ( fInfo->m_handle == -1 ) return false;

    if ( fInfo->m_outstanding_writes > 0 )
        fInfo->m_sync_waiters.push_back(callback);
    else
        callback->on_completed(0);

    return true;
}

/// <summary>
/// Adjust the internal buffers and pointers when the application seeks to a new read location in the stream.
/// </summary>
/// <param name="info">The file info record of the file</param>
/// <param name="pos">The new position (offset from the start) in the file stream</param>
/// <returns>New file position or -1 if error</returns>
size_t _seekrdtoend_fsb(Concurrency::streams::details::_file_info *info, int64_t offset, size_t char_size)
{
    if ( info == nullptr ) return static_cast<size_t>(-1);

    _file_info_impl *fInfo = static_cast<_file_info_impl *>(info);

    pplx::extensibility::scoped_recursive_lock_t lock(info->m_lock);

    if ( fInfo->m_handle == -1 ) return static_cast<size_t>(-1);

    if ( fInfo->m_buffer != nullptr )
    {
        delete[] fInfo->m_buffer;
        fInfo->m_buffer = nullptr;
        fInfo->m_bufoff = fInfo->m_buffill = fInfo->m_bufsize = 0;
    }

    auto newpos = lseek(fInfo->m_handle, static_cast<off_t>(offset * char_size), SEEK_END);

    if ( newpos == -1 ) return static_cast<size_t>(-1);

    fInfo->m_rdpos = newpos / char_size;
    return fInfo->m_rdpos;
}

utility::size64_t _get_size(_In_ concurrency::streams::details::_file_info *info, size_t char_size)
{
    if ( info == nullptr ) return static_cast<size_t>(-1);

    _file_info_impl *fInfo = static_cast<_file_info_impl *>(info);

    pplx::extensibility::scoped_recursive_lock_t lock(info->m_lock);

    if ( fInfo->m_handle == -1 ) return static_cast<size_t>(-1);

    if ( fInfo->m_buffer != nullptr )
    {
        delete[] fInfo->m_buffer;
        fInfo->m_buffer = nullptr;
        fInfo->m_bufoff = fInfo->m_buffill = fInfo->m_bufsize = 0;
    }

    auto oldpos = lseek(fInfo->m_handle, 0, SEEK_CUR);

    if ( oldpos == -1 ) return utility::size64_t(-1);

    auto newpos = lseek(fInfo->m_handle, 0, SEEK_END);

    if ( newpos == -1 ) return utility::size64_t(-1);

    lseek(fInfo->m_handle, oldpos, SEEK_SET);

    return utility::size64_t(newpos / char_size);
}

/// <summary>
/// Adjust the internal buffers and pointers when the application seeks to a new read location in the stream.
/// </summary>
/// <param name="info">The file info record of the file</param>
/// <param name="pos">The new position (offset from the start) in the file stream</param>
/// <returns>New file position or -1 if error</returns>
size_t _seekrdpos_fsb(Concurrency::streams::details::_file_info *info, size_t pos, size_t)
{
    if ( info == nullptr ) return static_cast<size_t>(-1);

    _file_info_impl *fInfo = static_cast<_file_info_impl *>(info);

    pplx::extensibility::scoped_recursive_lock_t lock(info->m_lock);

    if ( fInfo->m_handle == -1 ) return static_cast<size_t>(-1);

    if ( pos < fInfo->m_bufoff || pos > (fInfo->m_bufoff+fInfo->m_buffill) )
    {
        delete[] fInfo->m_buffer;
        fInfo->m_buffer = nullptr;
        fInfo->m_bufoff = fInfo->m_buffill = fInfo->m_bufsize = 0;
    }

    fInfo->m_rdpos = pos;
    return fInfo->m_rdpos;
}

/// <summary>
/// Adjust the internal buffers and pointers when the application seeks to a new write location in the stream.
/// </summary>
/// <param name="info">The file info record of the file</param>
/// <param name="pos">The new position (offset from the start) in the file stream</param>
/// <returns>New file position or -1 if error</returns>
size_t _seekwrpos_fsb(Concurrency::streams::details::_file_info *info, size_t pos, size_t)
{
    if ( info == nullptr ) return static_cast<size_t>(-1);

    _file_info_impl *fInfo = static_cast<_file_info_impl *>(info);

    pplx::extensibility::scoped_recursive_lock_t lock(info->m_lock);

    if ( fInfo->m_handle == -1 ) return static_cast<size_t>(-1);

    fInfo->m_wrpos = pos;
    return fInfo->m_wrpos;
}