file_processor.h

Go to the documentation of this file.

// Copyright (c) 2018, Smart Projects Holdings Ltd
// All rights reserved.
// See LICENSE file for license details.

#ifndef _UGCS_VSM_FILE_PROCESSOR_H_
#define _UGCS_VSM_FILE_PROCESSOR_H_

#include <ugcs/vsm/io_request.h>
#include <ugcs/vsm/request_worker.h>
#include <unistd.h>
#include <thread>

namespace ugcs {
namespace vsm {

class File_processor: public Request_processor {
    DEFINE_COMMON_CLASS(File_processor, Request_container)

public:
    VSM_DEFINE_EXCEPTION(Exception);
    VSM_DEFINE_DERIVED_EXCEPTION(Exception, Not_found_exception);
    VSM_DEFINE_DERIVED_EXCEPTION(Exception, Permission_denied_exception);
    VSM_DEFINE_DERIVED_EXCEPTION(Exception, Already_exists_exception);

    class Native_controller;

    class Stream: public Io_stream {
        DEFINE_COMMON_CLASS(Stream, Io_stream)

    public:
        typedef Reference_guard<Stream::Ptr> Ref;

        enum class Lock_result {
            OK,
            BLOCKED,
            ERROR
        };

        class Mode {
        public:
            bool read:1,

                 write:1,

                 extended:1,

             should_not_exist:1;

            Mode(const std::string &mode_str);
        };

        class Native_handle {
        public:
            typedef std::unique_ptr<Native_handle> Unique_ptr;

            class Stream_ref_holder {
            public:
                Stream_ref_holder(Stream::Ptr stream,
                                  std::unique_lock<std::mutex> &&lock):
                    stream(stream), lock(std::move(lock))
                {}

                Stream_ref_holder(Stream_ref_holder &&) = default;

                ~Stream_ref_holder()
                {
                    /* Prevent access to the mutex after stream is released. */
                    if (lock) {
                        lock.unlock();
                        lock.release();
                    }
                    stream = nullptr;
                }
            private:
                Stream::Ptr stream;
                std::unique_lock<std::mutex> lock;
            };

            virtual
            ~Native_handle()
            {}

            void
            Set_stream(Stream::Ptr stream);

            virtual void
            Write() = 0;

            virtual void
            Read() = 0;

            virtual Lock_result
            Try_lock() = 0;

            virtual bool
            Lock() = 0;

            virtual bool
            Unlock() = 0;

            virtual bool
            Cancel_write() = 0;

            virtual bool
            Cancel_read() = 0;

            virtual void
            Close() = 0;

            void
            Handle_write_abort();

            void
            Handle_read_abort();

            Read_request::Ptr cur_read_request;
            Write_request::Ptr cur_write_request;
            Io_request::Ptr cur_lock_request;
            Io_request::Ptr cur_unlock_request;
            bool is_closed = false;
            Stream::Ptr stream;
            Stream::Ptr write_active,
                        read_active;

        protected:
            Stream_ref_holder
            Set_write_activity(bool is_active,
                               std::unique_lock<std::mutex> &&lock =
                                   std::unique_lock<std::mutex>());

            Stream_ref_holder
            Set_read_activity(bool is_active,
                              std::unique_lock<std::mutex> &&lock =
                                  std::unique_lock<std::mutex>());

            Native_controller &
            Get_native_controller() const
            {
                return stream->Get_native_controller();
            }
        };

        Stream(File_processor::Ptr processor, const std::string &path, Mode mode,
               bool maintain_pos, Native_handle::Unique_ptr&& native_handle);

        Offset
        Get_current_pos() const
        {
            return cur_pos;
        }

        Offset
        Seek(Offset pos, bool is_relative = false);

        typedef Callback_proxy<void, Io_result> Lock_handler;

        Operation_waiter
        Lock(
            Lock_handler completion_handler,
            Request_completion_context::Ptr comp_ctx = Request_temp_completion_context::Create(),
            bool lock = true);

        Operation_waiter
        Unlock(
            Lock_handler completion_handler = Make_dummy_callback<void, Io_result>(),
            Request_completion_context::Ptr comp_ctx = Request_temp_completion_context::Create())
        {
            return Lock(completion_handler, comp_ctx, false);
        }

    private:
        friend class File_processor;
        friend class Native_handle;

        typedef struct {
            bool flock_thread_active = false;
            bool flock_acquire_requested = false;
            bool flock_acquired = false;
            std::condition_variable flock_notifier;
        } Lock_cb;

        Lock_cb lock_cb;

        File_processor::Weak_ptr processor;
        std::mutex op_mutex;
        std::list<Read_request::Ptr> read_queue;
        std::list<Write_request::Ptr> write_queue;

        Mode mode;
        bool maintain_pos;
        Offset cur_pos = 0;
        Native_handle::Unique_ptr native_handle;

        virtual Operation_waiter
        Write_impl(Io_buffer::Ptr buffer,
                   Offset offset,
                   Write_handler completion_handler,
                   Request_completion_context::Ptr comp_ctx) override;

        virtual Operation_waiter
        Read_impl(size_t max_to_read, size_t min_to_read, Offset offset,
                  Read_handler completion_handler,
                  Request_completion_context::Ptr comp_ctx) override;

        virtual Operation_waiter
        Close_impl(Close_handler completion_handler,
                   Request_completion_context::Ptr comp_ctx) override;

        void
        Queue_write(Write_request::Ptr request);

        void
        Push_write_queue();

        void
        Queue_read(Read_request::Ptr request);

        void
        Push_read_queue();

        void
        Handle_write();

        void
        Handle_read();

        void
        Handle_lock();

        void
        Handle_unlock();

        void
        Handle_close(Request::Ptr request);

        void
        Handle_write_cancel(Write_request::Ptr request);

        void
        Handle_read_cancel(Read_request::Ptr request);

        void
        Handle_lock_cancel(Io_request::Ptr request);

        void
        Cancel_lock_operation(bool stop_locker_thread);

        void
        Handle_write_completion(Request::Handler completion_handler);

        void
        Handle_read_completion(Request::Handler completion_handler);

        void
        Handle_write_abort();

        void
        Handle_read_abort();

        Native_controller &
        Get_native_controller() const;

        void
        Complete_request(Io_request::Ptr& request, Io_result result = Io_result::OTHER_FAILURE);

        void
        Locker_thread();
    };

    class Native_controller {
    public:
        virtual
        ~Native_controller()
        {}

        virtual void
        Enable() = 0;

        virtual void
        Disable() = 0;

        virtual void
        Register_handle(Stream::Native_handle &handle) = 0;

        virtual void
        Unregister_handle(Stream::Native_handle &handle) = 0;

        static std::unique_ptr<Native_controller>
        Create();
    };

    template <typename... Args>
    static Ptr
    Get_instance(Args &&... args)
    {
        return singleton.Get_instance(std::forward<Args>(args)...);
    }

    File_processor();

    Stream::Ref
    Open(const std::string &name, const std::string &mode, bool maintain_pos = true);

    static FILE*
    Fopen_utf8(const std::string &name, const std::string & mode);

    static bool
    Rename_utf8(const std::string &old_name, const std::string &new_name);

    static bool
    Remove_utf8(const std::string &name);

    static int
    Access_utf8(const std::string &name, int mode);

private:
    friend class Stream;

    static Singleton<File_processor> singleton;
    Request_worker::Ptr worker;
    Request_completion_context::Ptr comp_ctx;
    std::unique_ptr<Native_controller> native_controller;

    virtual void
    On_enable() override;

    virtual void
    On_disable() override;

    void
    Unregister_handle(Stream::Native_handle &handle);

    Stream::Native_handle::Unique_ptr
    Open_native_handle(const std::string &name, const std::string &mode);

protected:
    void
    Register_stream(File_processor::Stream::Ptr);
};

} /* namespace vsm */
} /* namespace ugcs */

#endif /* _UGCS_VSM_FILE_PROCESSOR_H_ */