text:
/*
  Copyright (c) <2013-2016>, <BenHJ>
  All rights reserved.

  Redistribution and use in source and binary forms, with or without
  modification, are permitted provided that the following conditions are met:

  1. Redistributions of source code must retain the above copyright notice, this
  list of conditions and the following disclaimer.
  2. Redistributions in binary form must reproduce the above copyright notice,
  this list of conditions and the following disclaimer in the documentation
  and/or other materials provided with the distribution.
  3. Neither the name of the copyright holder nor the names of its contributors
  may be used to endorse or promote products derived from this software without
  specific prior written permission.

  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
  ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
  ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
  (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
  LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
  ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
  SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/

#pragma once

#include "knoxcrypt/CoreIO.hpp"
#include "knoxcrypt/FileBlock.hpp"
#include "knoxcrypt/OpenDisposition.hpp"
#include "knoxcrypt/ContainerImageStream.hpp"

#include <functional>
#include <boost/optional.hpp>

#include <memory>
#include <vector>
#include <iosfwd>                           // streamsize, seekdir
#include <string>
#include <deque>

namespace knoxcrypt
{

    class File;
    using SharedFile = std::shared_ptr<File>;

    class File
    {

        using SetEntryInfoSizeCallback = std::function<void(uint64_t)>;
        using OptionalSizeCallback = boost::optional<SetEntryInfoSizeCallback>;
        using SharedFileBlock = std::shared_ptr<FileBlock>;

      public:
        File() = delete;
        /**
         * @brief when creating a new file this constructor should be used
         * @param io the core knoxcrypt io (path, blocks, password)
         * @param name the name of the file entry
         * @param enforceStartBlock true if start block should be set
         */
        File(SharedCoreIO const &io, std::string const &name, bool const enforceStartBlock = false);

        /**
         * @brief when reading or appending or overwriting this constructor should be used
         * @param io the core knoxcrypt io (path, blocks, password)
         * @param name the name of the file entry
         * @param block the starting block of the file entry
         * @param openDisposition open mode
         */
        File(SharedCoreIO const &io,
                    std::string const &name,
                    uint64_t const startBlock,
                    OpenDisposition const &openDisposition);

        typedef char                                   char_type;
        typedef boost::iostreams::seekable_device_tag  category;

        /**
         * @brief  simply acceses the name of the knoxcrypt file
         * @return the knoxcrypt file's name
         */
        std::string filename() const;

        /**
         * @brief  accesses file size
         * @return file size in bytes
         */
        uint64_t fileSize() const;

        /**
         * @brief  retrieves the first file block making up this knoxcrypt file
         * @return the start block index of this file
         */
        uint64_t getStartVolumeBlockIndex() const;

        /**
         * @brief truncates a file to new size
         * @param newSize the new fileSize
         */
        void truncate(std::ios_base::streamoff newSize);

        /**
         * @brief  for reading bytes from the knoxcrypt file
         * @param  s buffer to store the read bytes
         * @param  n number of bytes to read
         * @return number of bytes read
         */
        std::streamsize read(char* s, std::streamsize n);

        /**
         * @brief  for writing bytes to the knoxcrypt file
         * @param  s buffer that stores the bytes to write
         * @param  n the number of bytes to write
         * @return the number of bytes written
         */
        std::streamsize write(const char* s, std::streamsize n);

        /**
         * @brief  allows seeking to a given position in the knoxcrypt file
         * @param  off the offset to seek to
         * @param  way the position of where to offset from (begin, current, or end)
         * @return returns the offset (NOTE: should this be returning the actual
         * 'tell' position instead?)
         */
        boost::iostreams::stream_offset seek(boost::iostreams::stream_offset off,
                                             std::ios_base::seekdir way = std::ios_base::beg);

        /**
         * @brief  indicates the stream position
         * @return the current stream position
         */
        boost::iostreams::stream_offset tell() const;

        /**
         * @brief flushes any remaining data
         */
        void flush();

        /**
         * @brief deallocates blocks associated with this file entry; used
         * in conjunction with deleting the file
         */
        void unlink();

        /**
         * @brief sets the callback that will be used to updated the reported
         * file size as stored in the entry info metadata of the parent
         * @param callback the setSize function (probably) of EntryInfo
         */
        void setOptionalSizeUpdateCallback(SetEntryInfoSizeCallback callback);

        /**
         * @brief  queries what the current open mode is
         * @return the open disposition
         */
        OpenDisposition getOpenDisposition() const;

        SharedImageStream getStream() const;

        /// calls in to doReset (see comment therein)
        void reset();

      private:

        // the core knoxcrypt io (path, blocks, rootBlock, password)
        SharedCoreIO m_io;

        // the name of the file entry
        std::string m_name;

        // whether to enforce the starting file block (potentially dangerous!)
        mutable bool m_enforceStartBlock;

        // the size of the file
        uint64_t m_fileSize;

        // the current block being read from / written to
        mutable SharedFileBlock m_workingBlock;

        // a buffer used for storing chunks of data
        std::vector<uint8_t> m_buffer;

        // the start file block index
        mutable uint64_t m_startVolumeBlock;

        // the index of the block in the actual blocks container;
        // in comparison to m_currentBlock, this is where the block
        // exists in m_fileBlocks
        mutable int64_t m_blockIndex;

        // open mode
        mutable OpenDisposition m_openDisposition;

        // the current 'stream position' of file entry
        std::streamoff m_pos;

        // how many blocks make up the file?
        mutable uint64_t m_blockCount;

        // an optional size update callback to be used in setting the reported
        // size in the entry info held in the parent folder entry info cache
        OptionalSizeCallback m_optionalSizeCallback;

        // hold on to the image stream globally to reduce seek time when
        // instantiating a new FileBlock
        mutable SharedImageStream m_stream;

        /**
         * @brief  for keeping track of what the current file block as indicated
         *         by the current working file block
         * @return the current volume block
         */
        uint64_t getCurrentVolumeBlockIndex();

        /**
         * @brief creates a new file block for writing and updates the working block
         */
        void newWritableFileBlock() const;

        /**
         * @brief counts the number of blocks and sets file size
         */
        void enumerateBlockStats();

        /**
         * @brief  buffers as many bytes as permitted by the working block
         * @param  s the data to buffer
         * @param  n the number of bytes to try and buffer
         * @param  offset where abouts in 's' we are
         * @return the actual number of bytes buffered
         */
        uint32_t bufferBytesForWorkingBlock(const char* s, std::streamsize n, uint32_t offset);

        /**
         * @brief writes data to working block
         * @param bytes the number of bytes to write
         */
        void writeBufferedDataToWorkingBlock(uint32_t const bytes);

        /**
         * @brief  reads bytes from the working block in to buffer
         * @return the number of bytes read
         */
        std::streamsize readWorkingBlockBytes(uint32_t const bytes);

        /**
         * @brief will build a new file block for writing to if there are
         * no file blocks or if there are file blocks and it is determined
         * that we're not in append mode
         */
        void checkAndUpdateWorkingBlockWithNew() const;

        /**
         * @brief used in the context of discovering if currently set block
         * has enough space to write more data to
         * @return true if space available, false otherwise
         */
        bool workingBlockHasAvailableSpace() const;

        /**
         * @brief  computes number of bytes available for writing in block
         * @return number of bytes left for writing
         */
        uint32_t getBytesLeftInWorkingBlock();

        /**
         * @brief  retrieves a specific block making up the file
         * @param  n the block to retrieve
         * @return a file block
         */
        FileBlock getBlockWithIndex(uint64_t n) const;

        /// to be called during unlinking and when the data
        /// represents a folder and there are no more entries
        void doReset();
    };

}

max_stars_repo_name: benhj/KnoxCrypt

max_issues_repo_name: benhj/KnoxCrypt

max_forks_repo_name: benhj/KnoxCrypt


text:
/*
 * Copyright (c) 2018 Atmosphère-NX
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms and conditions of the GNU General Public License,
 * version 2, as published by the Free Software Foundation.
 *
 * This program is distributed in the hope it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
 * more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#pragma once
#include <switch.h>
#include <cstdlib>
#include <cstring>
#include <tuple>
#include <boost/callable_traits.hpp>
#include <type_traits>
#include <memory>

#include "ipc_out.hpp"
#include "ipc_buffers.hpp"
#include "ipc_special.hpp"

#include "ipc_domain_object.hpp"

#include "ipc_response_context.hpp"

#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wunused-but-set-variable"

template<typename Tuple>
struct PopFront;

template<typename Head, typename... Tail>
struct PopFront<std::tuple<Head, Tail...>> {
    using type = std::tuple<Tail...>;
};

template <typename ...> struct WhichType;

template <typename...>
struct TypeList{};

template <typename... T1s, typename... T2s>
constexpr auto Concatenate(TypeList<T1s...>, TypeList<T2s...>) {
    return TypeList<T1s..., T2s...>{};
}

template <template <typename> typename Condition, typename R>
constexpr auto FilterTypes(R result, TypeList<>) {
    return result;
}

template <template <typename> typename Condition, typename R, typename T, typename... Ts>
constexpr auto FilterTypes(R result, TypeList<T, Ts...>) {
    if constexpr (Condition<T>{})
        return FilterTypes<Condition>(Concatenate(result, TypeList<T>{}), TypeList<Ts...>{});
    else
        return FilterTypes<Condition>(result, TypeList<Ts...>{});
}

template<typename Types> struct TypeListToTuple;

template<typename... Types>
struct TypeListToTuple<TypeList<Types...>> {
    using type = std::tuple<Types...>;
};

template <template <typename> typename Condition, typename... Types>
using FilteredTypes = typename TypeListToTuple<std::decay_t<decltype(FilterTypes<Condition>(TypeList<>{}, TypeList<Types...>{}))>>::type;

enum class ArgType {
    InData,
    OutData,
    InHandle,
    OutHandle,
    InSession,
    OutSession,
    PidDesc,
    InBuffer,
    OutBuffer,
    InPointer,
    OutPointerClientSize,
    OutPointerServerSize,
};

template<typename X>
constexpr ArgType GetArgType() {
    if constexpr (std::is_base_of_v<OutDataTag, X>) {
        return ArgType::OutData;
    } else if constexpr (std::is_base_of_v<OutSessionTag, X>) {
        return ArgType::OutSession;
    } else if constexpr (std::is_base_of_v<OutHandleTag, X>) {
        return ArgType::OutHandle;
    } else if constexpr (std::is_base_of_v<InBufferBase, X>) {
        return ArgType::InBuffer;
    } else if constexpr (std::is_base_of_v<OutBufferBase, X>) {
        return ArgType::OutBuffer;
    } else if constexpr (std::is_base_of_v<InPointerBase, X>) {
        return ArgType::InPointer;
    } else if constexpr (std::is_base_of_v<OutPointerWithClientSizeBase, X>) {
        return ArgType::OutPointerClientSize;
    } else if constexpr (std::is_base_of_v<OutPointerWithServerSizeBase, X>) {
        return ArgType::OutPointerServerSize;
    } else if constexpr (std::is_base_of_v<PidDescriptorTag, X>) {
        return ArgType::PidDesc;
    } else if constexpr (std::is_base_of_v<IpcHandleTag, X>) {
        return ArgType::InHandle;
    } else if constexpr (std::is_trivial_v<X> && !std::is_pointer_v<X>) {
        return ArgType::InData;
    } else {
        static_assert(std::is_pod_v<X> && !std::is_pod_v<X>, "Unhandled InSession!");
        return ArgType::InSession;
    }
}

template<ArgType ArgT>
struct ArgTypeFilter {
    template<typename X>
    using type = std::conditional_t<GetArgType<X>() == ArgT, std::true_type, std::false_type>;
};

template<ArgType ArgT>
struct IsArgTypeBuffer {
    static constexpr bool value = ArgT == ArgType::InBuffer || ArgT == ArgType::OutBuffer || ArgT == ArgType::InPointer || ArgT == ArgType::OutPointerClientSize || ArgT == ArgType::OutPointerServerSize;
};

struct ArgTypeBufferFilter {
    template<typename X>
    using type = std::conditional_t<IsArgTypeBuffer<GetArgType<X>()>::value, std::true_type, std::false_type>;
};

template<ArgType ArgT>
struct IsArgTypeInData {
    static constexpr bool value = ArgT == ArgType::InData || ArgT == ArgType::PidDesc;
};

struct ArgTypeInDataFilter {
    template<typename X>
    using type = std::conditional_t<IsArgTypeInData<GetArgType<X>()>::value, std::true_type, std::false_type>;
};

template<typename T>
struct RawDataHelper {
    static_assert(GetArgType<T>() == ArgType::InData || GetArgType<T>() == ArgType::PidDesc);
    static constexpr size_t align = (GetArgType<T>() == ArgType::InData) ? __alignof__(T) : __alignof__(u64);
    static constexpr size_t size = (GetArgType<T>() == ArgType::InData) ? sizeof(T) : sizeof(u64);
};

template<typename T>
struct RawDataHelper<Out<T>> {
    static_assert(GetArgType<T>() == ArgType::InData);
    static constexpr size_t align = __alignof(T);
    static constexpr size_t size = sizeof(T);
};

template<typename Ts>
struct RawDataComputer;

template<typename... Ts>
struct RawDataComputer<std::tuple<Ts...>> {
    /* https://referencesource.microsoft.com/#System.Core/System/Linq/Enumerable.cs,2604 */
    static constexpr void QuickSort(std::array<size_t, sizeof...(Ts)> &map, std::array<size_t, sizeof...(Ts)> &values, int left, int right) {
        do {
            int i = left;
            int j = right;
            int x = map[i + ((j - i) >> 1)];
            do {
                while (i < static_cast<int>(sizeof...(Ts)) && values[x] > values[map[i]]) i++;
                while (j >= 0 && values[x] < values[map[j]]) j--;
                if (i > j) break;
                if (i < j) {
                    const size_t temp = map[i];
                    map[i] = map[j];
                    map[j] = temp;
                }
                i++;
                j--;
            } while (i <= j);
            if (j - left <= right - i) {
                if (left < j) QuickSort(map, values, left, j);
                left = i;
            } else {
                if (i < right) QuickSort(map, values, i, right);
                right = j;
            }
        } while (left < right);
    }

    static constexpr void StableSort(std::array<size_t, sizeof...(Ts)> &map, std::array<size_t, sizeof...(Ts)> &values) {
        /* First, quicksort a copy of the map. */
        std::array<size_t, sizeof...(Ts)> map_unstable(map);
        QuickSort(map_unstable, values, 0, sizeof...(Ts)-1);

        /* Now, create stable sorted map from unstably quicksorted indices (via repeated insertion sort on element runs). */
        for (size_t i = 0; i < sizeof...(Ts); i++) {
            map[i] = map_unstable[i];
            for (ssize_t j = i-1; j >= 0 && values[map[j]] == values[map[j+1]] && map[j] > map[j+1]; j--) {
                const size_t temp = map[j];
                map[j] = map[j+1];
                map[j+1] = temp;
            }
        }
    }

    static constexpr std::array<size_t, sizeof...(Ts)+1> GetOffsets() {
        std::array<size_t, sizeof...(Ts)+1> offsets = {};
        offsets[0] = 0;
        if constexpr (sizeof...(Ts) > 0) {
            /* Get size, alignment for each type. */
            std::array<size_t, sizeof...(Ts)> sizes = { RawDataHelper<Ts>::size... };
            std::array<size_t, sizeof...(Ts)> aligns = { RawDataHelper<Ts>::align... };

            /* We want to sort...by alignment. */
            std::array<size_t, sizeof...(Ts)> map = {};
            for (size_t i = 0; i < sizeof...(Ts); i++) { map[i] = i; }
            StableSort(map, aligns);

            /* Iterate over sorted types. */
            size_t cur_offset = 0;
            for (size_t i = 0; i < sizeof...(Ts); i++) {
                const size_t align = aligns[map[i]];
                if (cur_offset % align != 0) {
                    cur_offset += align - (cur_offset % align);
                }
                offsets[map[i]] = cur_offset;
                cur_offset += sizes[map[i]];
            }
            offsets[sizeof...(Ts)] = cur_offset;
        }
        return offsets;
    }

    static constexpr std::array<size_t, sizeof...(Ts)+1> offsets = GetOffsets();
};

template <typename _Args, typename _ReturnType>
struct CommandMetaInfo;

template<typename... _Args, typename _ReturnType>
struct CommandMetaInfo<std::tuple<_Args...>, _ReturnType> {
    using Args = std::tuple<_Args...>;
    using ReturnType = _ReturnType;

    static constexpr bool ReturnsResult = std::is_same_v<ReturnType, Result>;
    static constexpr bool ReturnsVoid = std::is_same_v<ReturnType, void>;

    using InDatas = FilteredTypes<ArgTypeInDataFilter::type, _Args...>;
    using OutDatas = FilteredTypes<ArgTypeFilter<ArgType::OutData>::type, _Args...>;
    using InHandles = FilteredTypes<ArgTypeFilter<ArgType::InHandle>::type, _Args...>;
    using OutHandles = FilteredTypes<ArgTypeFilter<ArgType::OutHandle>::type, _Args...>;
    using InSessions = FilteredTypes<ArgTypeFilter<ArgType::InSession>::type, _Args...>;
    using OutSessions = FilteredTypes<ArgTypeFilter<ArgType::OutSession>::type, _Args...>;
    using PidDescs = FilteredTypes<ArgTypeFilter<ArgType::PidDesc>::type, _Args...>;

    using InBuffers = FilteredTypes<ArgTypeFilter<ArgType::InBuffer>::type, _Args...>;
    using OutBuffers = FilteredTypes<ArgTypeFilter<ArgType::OutBuffer>::type, _Args...>;
    using InPointers = FilteredTypes<ArgTypeFilter<ArgType::InPointer>::type, _Args...>;
    using ClientSizeOutPointers = FilteredTypes<ArgTypeFilter<ArgType::OutPointerClientSize>::type, _Args...>;
    using ServerSizeOutPointers = FilteredTypes<ArgTypeFilter<ArgType::OutPointerServerSize>::type, _Args...>;
    using Buffers = FilteredTypes<ArgTypeBufferFilter::type, _Args...>;

    static constexpr size_t NumInDatas = std::tuple_size_v<InDatas>;
    static constexpr size_t NumOutDatas = std::tuple_size_v<OutDatas>;
    static constexpr size_t NumInHandles = std::tuple_size_v<InHandles>;
    static constexpr size_t NumOutHandles = std::tuple_size_v<OutHandles>;
    static constexpr size_t NumInSessions = std::tuple_size_v<InSessions>;
    static constexpr size_t NumOutSessions = std::tuple_size_v<OutSessions>;
    static constexpr size_t NumPidDescs = std::tuple_size_v<PidDescs>;

    static constexpr size_t NumInBuffers = std::tuple_size_v<InBuffers>;
    static constexpr size_t NumOutBuffers = std::tuple_size_v<OutBuffers>;
    static constexpr size_t NumInPointers = std::tuple_size_v<InPointers>;
    static constexpr size_t NumClientSizeOutPointers = std::tuple_size_v<ClientSizeOutPointers>;
    static constexpr size_t NumServerSizeOutPointers = std::tuple_size_v<ServerSizeOutPointers>;
    static constexpr size_t NumBuffers = std::tuple_size_v<Buffers>;

    static_assert(NumInSessions == 0, "InSessions not yet supported!");
    static_assert(NumPidDescs == 0 || NumPidDescs == 1, "Methods can only take in 0 or 1 PIDs!");
    static_assert(NumBuffers <= 8, "Methods can only take in <= 8 Buffers!");
    static_assert(NumInHandles <= 8, "Methods can take in <= 8 Handles!");
    static_assert(NumOutHandles + NumOutSessions <= 8, "Methods can only return <= 8 Handles+Sessions!");

    static constexpr std::array<size_t, NumInDatas+1> InDataOffsets = RawDataComputer<InDatas>::offsets;
    static constexpr size_t InRawArgSize = InDataOffsets[NumInDatas];
    static constexpr size_t InRawArgSizeWithOutPointers = ((InRawArgSize + NumClientSizeOutPointers * sizeof(u16)) + 3) & ~3;

    static constexpr std::array<size_t, NumOutDatas+1> OutDataOffsets = RawDataComputer<OutDatas>::offsets;
    static constexpr size_t OutRawArgSize = OutDataOffsets[NumOutDatas];
};

/* ================================================================================= */
/* Actual wrapping implementation goes here. */

/* Validator. */
struct Validator {

    template <typename T>
    static constexpr bool ValidateCommandArgument(IpcResponseContext *ctx, size_t& a_index, size_t& b_index, size_t& x_index, size_t& h_index, size_t& cur_c_size_offset, size_t& total_c_size) {
        constexpr ArgType argT = GetArgType<T>();
        if constexpr (argT == ArgType::InBuffer) {
            return (ctx->request.Buffers[a_index] != nullptr || ctx->request.BufferSizes[a_index] == 0) && ctx->request.BufferDirections[a_index] == BufferDirection_Send && ctx->request.BufferTypes[a_index++] == T::expected_type;
        } else if constexpr (argT == ArgType::OutBuffer) {
            return (ctx->request.Buffers[b_index] != nullptr || ctx->request.BufferSizes[b_index] == 0) && ctx->request.BufferDirections[b_index] == BufferDirection_Recv && ctx->request.BufferTypes[b_index++] == T::expected_type;
        } else if constexpr (argT == ArgType::InPointer) {
            return ctx->request.Statics[x_index++] != nullptr;
        } else if constexpr (argT == ArgType::InHandle) {
            if constexpr (std::is_same_v<T, MovedHandle>) {
                return !ctx->request.WasHandleCopied[h_index++];
            } else if constexpr (std::is_same_v<T, CopiedHandle>) {
                return ctx->request.WasHandleCopied[h_index++];
            }
        } else {
            if constexpr (argT == ArgType::OutPointerServerSize) {
                total_c_size += T::num_elements * sizeof(T);
            } else if constexpr (argT == ArgType::OutPointerServerSize) {
                total_c_size += *((u16 *)((uintptr_t)(ctx->request.Raw) + 0x10 + cur_c_size_offset));
                cur_c_size_offset += sizeof(u16);
            }
            return true;
        }
    }

    template <typename Ts>
    struct ValidateCommandTuple;

    template <typename ...Ts>
    struct ValidateCommandTuple<std::tuple<Ts...>> {
        static constexpr bool IsValid(IpcResponseContext *ctx, size_t& a_index, size_t& b_index, size_t& x_index, size_t& h_index, size_t& cur_c_size_offset, size_t& total_c_size) {
            return (ValidateCommandArgument<Ts>(ctx, a_index, b_index, x_index, h_index, cur_c_size_offset, total_c_size) && ...);
        }
    };

    template<typename MetaInfo>
	static constexpr Result Validate(IpcResponseContext *ctx) {
        if (ctx->request.RawSize < MetaInfo::InRawArgSizeWithOutPointers) {
            return 0xF601;
        }

        if (ctx->request.NumBuffers != MetaInfo::NumInBuffers + MetaInfo::NumOutBuffers) {
            return 0xF601;
        }

        if (ctx->request.NumStatics != MetaInfo::NumInPointers) {
            return 0xF601;
        }

        if (ctx->request.NumStaticsOut != MetaInfo::NumClientSizeOutPointers + MetaInfo::NumServerSizeOutPointers) {
            return 0xF601;
        }

        if (ctx->request.NumHandles != MetaInfo::NumInHandles) {
            return 0xF601;
        }

        if ((ctx->request.HasPid && MetaInfo::NumPidDescs == 0) || (!ctx->request.HasPid && MetaInfo::NumPidDescs != 0)) {
            return 0xF601;
        }

        if (((u32 *)ctx->request.Raw)[0] != SFCI_MAGIC) {
            return 0xF601;
        }

        size_t a_index = 0, b_index = MetaInfo::NumInBuffers, x_index = 0, h_index = 0;
        size_t cur_c_size_offset = MetaInfo::InRawArgSize + (0x10 - ((uintptr_t)ctx->request.Raw - (ctx->request.IsDomainRequest ? sizeof(DomainMessageHeader) : 0) - (uintptr_t)ctx->request.RawWithoutPadding));
        size_t total_c_size = 0;

        if (!ValidateCommandTuple<typename MetaInfo::Args>::IsValid(ctx, a_index, b_index, x_index, h_index, cur_c_size_offset, total_c_size)) {
            return 0xF601;
        }

        if (total_c_size > ctx->pb_size) {
            return 0xF601;
        }

        return 0;
	}
};

/* ================================================================================= */

/* Decoder. */
template<typename MetaInfo>
struct Decoder {

    template<typename T>
    static constexpr T DecodeCommandArgument(IpcResponseContext *ctx, size_t& a_index, size_t& b_index, size_t& x_index, size_t& c_index, size_t& in_h_index, size_t& out_h_index, size_t& out_obj_index, size_t& in_data_index, size_t& out_data_index, size_t& pb_offset, size_t& c_sz_offset) {
        constexpr ArgType argT = GetArgType<T>();
        if constexpr (argT == ArgType::InBuffer) {
            const T& value = T(ctx->request.Buffers[a_index], ctx->request.BufferSizes[a_index], ctx->request.BufferTypes[a_index]);
            ++a_index;
            return value;
        } else if constexpr (argT == ArgType::OutBuffer) {
            const T& value = T(ctx->request.Buffers[b_index], ctx->request.BufferSizes[b_index], ctx->request.BufferTypes[b_index]);
            ++b_index;
            return value;
        } else if constexpr (argT == ArgType::InPointer) {
            const T& value = T(ctx->request.Statics[x_index], ctx->request.StaticSizes[x_index]);
            ++x_index;
            return value;
        } else if constexpr (argT == ArgType::InHandle) {
            return T(ctx->request.Handles[in_h_index++]);
        } else if constexpr (argT == ArgType::OutHandle) {
            return T(&ctx->out_handles[out_h_index++]);
        } else if constexpr (argT == ArgType::PidDesc) {
            uintptr_t ptr = ((uintptr_t)ctx->request.Raw + 0x10 + MetaInfo::InDataOffsets[in_data_index++]);
            *(u64 *)ptr = ctx->request.Pid;
            return T(ctx->request.Pid);
        } else if constexpr (argT == ArgType::InData) {
            uintptr_t ptr = ((uintptr_t)ctx->request.Raw + 0x10 + MetaInfo::InDataOffsets[in_data_index++]);
            if constexpr (std::is_same_v<bool, T>) {
                return *((u8 *)ptr) & 1;
            } else {
                return *((T *)ptr);
            }
        } else if constexpr (argT == ArgType::OutData) {
            uintptr_t ptr = ((uintptr_t)ctx->out_data + MetaInfo::OutDataOffsets[out_data_index++]);
            return T(reinterpret_cast<typename OutHelper<T>::type *>(ptr));
        } else if constexpr (argT == ArgType::OutPointerClientSize || argT == ArgType::OutPointerServerSize) {
            u16 sz;
            if constexpr(argT == ArgType::OutPointerServerSize) {
                sz = T::element_size;
            } else {
                sz = *(const u16 *)((uintptr_t)ctx->request.Raw + 0x10 + c_sz_offset);
            }
            u8* buf = ctx->pb + pb_offset;
            c_sz_offset += sizeof(u16);
            pb_offset += sz;
            ipcAddSendStatic(&ctx->reply, buf, sz, c_index++);
            return T(buf, sz);
        } else if constexpr (argT == ArgType::OutSession) {
            if (IsDomainObject(ctx->obj_holder)) {
               const T& value = T(ctx->out_objs[out_obj_index], ctx->obj_holder->GetServiceObject<IDomainObject>(), &ctx->out_object_ids[out_obj_index]);
               out_obj_index++;
               return value;
            } else {
               const T& value = T(ctx->out_objs[out_obj_index], nullptr, 0);
               out_obj_index++;
               return value;
            }
        }
    }

    template <typename Ts>
    struct DecodeTuple;

    template <typename ...Ts>
    struct DecodeTuple<std::tuple<Ts...>> {
        static constexpr std::tuple<Ts...> GetArgs(IpcResponseContext *ctx, size_t& a_index, size_t& b_index, size_t& x_index, size_t& c_index, size_t& in_h_index, size_t& out_h_index, size_t& out_obj_index, size_t& in_data_index, size_t& out_data_index, size_t& pb_offset, size_t& c_sz_offset) {
            return std::tuple<Ts... > {
                    DecodeCommandArgument<Ts>(ctx, a_index, b_index, x_index, c_index, in_h_index, out_h_index, out_obj_index, in_data_index, out_data_index, pb_offset, c_sz_offset)
                    ...
            };
        }
    };

	static constexpr typename MetaInfo::Args Decode(IpcResponseContext *ctx) {
        size_t a_index = 0, b_index = MetaInfo::NumInBuffers, x_index = 0, c_index = 0, in_h_index = 0, out_h_index = 0, out_obj_index = 0;
        size_t in_data_index = 0x0, out_data_index = 0, pb_offset = 0;
        size_t c_sz_offset = MetaInfo::InRawArgSize + (0x10 - ((uintptr_t)ctx->request.Raw  - (ctx->request.IsDomainRequest ? sizeof(DomainMessageHeader) : 0) - (uintptr_t)ctx->request.RawWithoutPadding));
        return DecodeTuple<typename MetaInfo::Args>::GetArgs(ctx, a_index, b_index, x_index, c_index,  in_h_index, out_h_index, out_obj_index, in_data_index, out_data_index, pb_offset, c_sz_offset);
    }
};

/* ================================================================================= */

template<typename MetaInfo, typename T>
static constexpr void EncodeArgument(IpcResponseContext *ctx, size_t&out_obj_index, T& arg) {
    constexpr ArgType argT = GetArgType<T>();
    if constexpr (argT == ArgType::OutHandle) {
        if constexpr (std::is_same_v<MovedHandle, typename OutHelper<T>::type>) {
            ipcSendHandleMove(&ctx->reply, arg.GetValue().handle);
        } else {
            ipcSendHandleCopy(&ctx->reply, arg.GetValue().handle);
        }
    } else if constexpr (argT == ArgType::OutSession) {
        if (IsDomainObject(ctx->obj_holder)) {
            auto domain = ctx->obj_holder->GetServiceObject<IDomainObject>();
            domain->SetObject(arg.GetObjectId(), std::move(arg.GetHolder()));
        } else {
            ctx->manager->AddSession(ctx->out_object_server_handles[out_obj_index++], std::move(arg.GetHolder()));
        }
    }
}

template<typename MetaInfo, typename ArgsTuple>
struct Encoder;

template <typename MetaInfo, typename ...Args>
struct Encoder<MetaInfo, std::tuple<Args...>> {

    static constexpr void EncodeFailure(IpcResponseContext *ctx, Result rc) {
        memset(armGetTls(), 0, 0x100);
        ipcInitialize(&ctx->reply);
        struct {
            u64 magic;
            u64 result;
        } *raw;

        if (IsDomainObject(ctx->obj_holder)) {
            raw = (decltype(raw))ipcPrepareHeaderForDomain(&ctx->reply, sizeof(*raw), 0);
            auto resp_header = (DomainResponseHeader *)((uintptr_t)raw - sizeof(DomainResponseHeader));
            *resp_header = {0};
        } else {
            raw = (decltype(raw))ipcPrepareHeader(&ctx->reply, sizeof(*raw));
        }
        raw->magic = SFCO_MAGIC;
        raw->result = rc;
    }

	static constexpr void EncodeSuccess(IpcResponseContext *ctx, Args... args) {
        size_t out_obj_index = 0;

        ((EncodeArgument<MetaInfo, Args>(ctx, out_obj_index, args)), ...);

        const bool is_domain = IsDomainObject(ctx->obj_holder);

        if (!is_domain) {
            for (unsigned int i = 0; i < MetaInfo::NumOutSessions; i++) {
                ipcSendHandleMove(&ctx->reply, ctx->out_handles[MetaInfo::NumOutHandles + i].handle);
            }
        }

        struct {
            u64 magic;
            u64 result;
        } *raw;
        if (is_domain) {
            raw = (decltype(raw))ipcPrepareHeaderForDomain(&ctx->reply, sizeof(*raw) + MetaInfo::OutRawArgSize + sizeof(*ctx->out_object_ids) * MetaInfo::NumOutSessions, 0);
            auto resp_header = (DomainResponseHeader *)((uintptr_t)raw - sizeof(DomainResponseHeader));
            *resp_header = {0};
            resp_header->NumObjectIds = MetaInfo::NumOutSessions;
        } else {
            raw = (decltype(raw))ipcPrepareHeader(&ctx->reply, sizeof(*raw)+ MetaInfo::OutRawArgSize);
        }

        raw->magic = SFCO_MAGIC;
        raw->result = 0;

        memcpy((void *)((uintptr_t)raw + sizeof(*raw)), ctx->out_data, MetaInfo::OutRawArgSize);
        if (is_domain) {
            memcpy((void *)((uintptr_t)raw + sizeof(*raw) + MetaInfo::OutRawArgSize), ctx->out_object_ids, sizeof(*ctx->out_object_ids) * MetaInfo::NumOutSessions);
        }
    }

};

/* ================================================================================= */

template<auto IpcCommandImpl>
constexpr Result WrapIpcCommandImpl(IpcResponseContext *ctx) {
    using InArgs = typename PopFront<typename boost::callable_traits::args_t<decltype(IpcCommandImpl)>>::type;
    using OutArgs = typename boost::callable_traits::return_type_t<decltype(IpcCommandImpl)>;
    using ClassType = typename boost::callable_traits::class_of_t<decltype(IpcCommandImpl)>;

    using CommandMetaData = CommandMetaInfo<InArgs, OutArgs>;

    static_assert(CommandMetaData::ReturnsResult || CommandMetaData::ReturnsVoid, "IpcCommandImpls must return Result or void");

    ipcInitialize(&ctx->reply);
    memset(ctx->out_data, 0, CommandMetaData::OutRawArgSize);

    Result rc = Validator::Validate<CommandMetaData>(ctx);

    if (R_FAILED(rc)) {
        return 0xAAEE;
    }

    ClassType *this_ptr = nullptr;
    if (IsDomainObject(ctx->obj_holder)) {
        this_ptr = ctx->obj_holder->GetServiceObject<IDomainObject>()->GetObject(ctx->request.InThisObjectId)->GetServiceObject<ClassType>();
    } else {
        this_ptr = ctx->obj_holder->GetServiceObject<ClassType>();
    }
    if (this_ptr == nullptr) {
        return 0xBBEE;
    }

    std::shared_ptr<IServiceObject> out_objects[CommandMetaData::NumOutSessions];

    /* Allocate out object IDs. */
    size_t num_out_objects;
    if (IsDomainObject(ctx->obj_holder)) {
        for (num_out_objects = 0; num_out_objects < CommandMetaData::NumOutSessions; num_out_objects++) {
            if (R_FAILED((rc = ctx->obj_holder->GetServiceObject<IDomainObject>()->ReserveObject(&ctx->out_object_ids[num_out_objects])))) {
                break;
            }
            ctx->out_objs[num_out_objects] = &out_objects[num_out_objects];
        }
    } else {
        for (num_out_objects = 0; num_out_objects < CommandMetaData::NumOutSessions; num_out_objects++) {
            Handle server_h, client_h;
            if (R_FAILED((rc = SessionManagerBase::CreateSessionHandles(&server_h, &client_h)))) {
                break;
            }
            ctx->out_object_server_handles[num_out_objects] = server_h;
            ctx->out_handles[CommandMetaData::NumOutHandles + num_out_objects].handle = client_h;
            ctx->out_objs[num_out_objects] = &out_objects[num_out_objects];
        }
    }

    ON_SCOPE_EXIT {
        /* Clean up objects as necessary. */
        if (R_FAILED(rc)) {
            if (IsDomainObject(ctx->obj_holder)) {
                for (unsigned int i = 0; i < num_out_objects; i++) {
                    ctx->obj_holder->GetServiceObject<IDomainObject>()->FreeObject(ctx->out_object_ids[i]);
                }
            } else {
                for (unsigned int i = 0; i < num_out_objects; i++) {
                    svcCloseHandle(ctx->out_object_server_handles[i]);
                    svcCloseHandle(ctx->out_handles[CommandMetaData::NumOutHandles + i].handle);
                }
            }
        }

        for (unsigned int i = 0; i < num_out_objects; i++) {
            ctx->out_objs[i] = nullptr;
        }
    };

    if (R_SUCCEEDED(rc)) {
        auto args = Decoder<CommandMetaData>::Decode(ctx);

        if constexpr (CommandMetaData::ReturnsResult) {
            rc = std::apply( [=](auto&&... args) { return (this_ptr->*IpcCommandImpl)(args...); }, args);
        } else {
            std::apply( [=](auto&&... args) { (this_ptr->*IpcCommandImpl)(args...); }, args);
        }

        if (R_SUCCEEDED(rc)) {
            std::apply(Encoder<CommandMetaData, decltype(args)>::EncodeSuccess, std::tuple_cat(std::make_tuple(ctx), args));
        } else {
            std::apply(Encoder<CommandMetaData, decltype(args)>::EncodeFailure, std::tuple_cat(std::make_tuple(ctx), std::make_tuple(rc)));
        }
    } else {
        std::apply(Encoder<CommandMetaData, typename CommandMetaData::Args>::EncodeFailure, std::tuple_cat(std::make_tuple(ctx), std::make_tuple(rc)));
    }

    return rc;
}

template <u32 c, auto CommandImpl, FirmwareVersion l = FirmwareVersion_Min, FirmwareVersion h = FirmwareVersion_Max>
inline static constexpr ServiceCommandMeta MakeServiceCommandMeta() {
    return {
        .fw_low = l,
        .fw_high = h,
        .cmd_id = c,
        .handler = WrapIpcCommandImpl<CommandImpl>,
    };
};

#pragma GCC diagnostic pop

ext: hpp


text:
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you 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.

#include <boost/thread.hpp>
#include <boost/thread/mutex.hpp>
#include <glog/logging.h>
#include <unistd.h>

#include "gutil/strings/substitute.h"
#include "common/logging.h"
#include "test/gtest-util.h"
#include "util/thread-pool.h"

#include "common/names.h"

namespace pegasus {

const int NUM_THREADS = 5;
int thread_counters[NUM_THREADS];

// Per-thread mutex to ensure visibility of counters after thread pool terminates
mutex thread_mutexes[NUM_THREADS];

void Count(int thread_id, const int& i) {
  lock_guard<mutex> l(thread_mutexes[thread_id]);
  thread_counters[thread_id] += i;
}

TEST(ThreadPoolTest, BasicTest) {
  const int OFFERED_RANGE = 10000;
  for (int i = 0; i < NUM_THREADS; ++i) {
    thread_counters[i] = 0;
  }

  ThreadPool<int> thread_pool("thread-pool", "worker", 5, 250, Count);
  ASSERT_OK(thread_pool.Init());
  for (int i = 0; i <= OFFERED_RANGE; ++i) {
    ASSERT_TRUE(thread_pool.Offer(i));
  }

  thread_pool.DrainAndShutdown();

  // Check that Offer() after Shutdown() will return false
  ASSERT_FALSE(thread_pool.Offer(-1));
  EXPECT_EQ(0, thread_pool.GetQueueSize());

  int expected_count = (OFFERED_RANGE * (OFFERED_RANGE + 1)) / 2;
  int count = 0;
  for (int i = 0; i < NUM_THREADS; ++i) {
    lock_guard<mutex> l(thread_mutexes[i]);
    LOG(INFO) << "Counter " << i << ": " << thread_counters[i];
    count += thread_counters[i];
  }

  EXPECT_EQ(expected_count, count);
}

class SleepWorkItem : public SynchronousWorkItem {
public:
  SleepWorkItem(int64_t timeout_ms, bool* destructor_called)
    : timeout_ms_(timeout_ms), destructor_called_ptr_(destructor_called) {
    *destructor_called_ptr_ = false;
  }

  ~SleepWorkItem() {
    *destructor_called_ptr_ = true;
  }

  virtual Status Execute() override {
    if (timeout_ms_ > 0) SleepForMs(timeout_ms_);
    message_ = "Done";
    return Status::OK();
  }
  virtual std::string GetDescription() override {
    return Substitute("Simple task with $0 millisecond timeout", timeout_ms_);
  }

  std::string GetMessage() { return message_; }
private:
  std::string message_ = "Not done";
  int64_t timeout_ms_;
  bool* destructor_called_ptr_;
};

TEST(ThreadPoolTest, SynchronousThreadPoolNoSleep) {
  // Create a synchronous pool with one thread and a queue size of one.
  SynchronousThreadPool pool("sync-thread-pool", "worker", 1, 1);
  ASSERT_OK(pool.Init());

  // Base case: work item takes no time, run it with a timeout of 5 milliseconds
  unique_ptr<bool> no_sleep_destroyed(new bool);
  std::shared_ptr<SleepWorkItem> no_sleep(new SleepWorkItem(0, no_sleep_destroyed.get()));
  ASSERT_OK(pool.SynchronousOffer(no_sleep, 5));
  ASSERT_EQ(no_sleep->GetMessage(), "Done");
  // If the SynchronousOffer() completed successfully, the thread pool does not have any
  // shared_ptr to the work item. The caller is the only holder, so when it calls
  // reset, the destructor must be called.
  no_sleep.reset();
  // The work item should be destroyed even if we are not shutting down the pool.
  // IMPALA-8371: There is a race condition with the worker thread, as the worker thread
  // may not have released its shared_ptr to the work item. Wait for a limited period of
  // time for the work thread to release the shared_ptr.
  for (int i = 0; i < 10; i++) {
    if (*no_sleep_destroyed) break;
    SleepForMs(5);
  }
  ASSERT_TRUE(*no_sleep_destroyed);
  pool.DrainAndShutdown();
}

TEST(ThreadPoolTest, SynchronousThreadPoolTimeouts) {
  // Create a synchronous pool with one thread and a queue size of one.
  SynchronousThreadPool pool("sync-thread-pool", "worker", 1, 1);
  ASSERT_OK(pool.Init());

  // Timeout case #1: Submit one task that takes 100 milliseconds. Offer it with a timeout
  // of 1 millisecond so that the caller immediately times out.
  unique_ptr<bool> long_sleep_destroyed(new bool);
  std::shared_ptr<SleepWorkItem> long_sleep(
      new SleepWorkItem(100, long_sleep_destroyed.get()));
  Status timeout_status = pool.SynchronousOffer(long_sleep, 1);
  ASSERT_EQ(timeout_status.code(), StatusCode::ThreadPoolTaskTimeoutError);
  // The work item is still running, and even if the caller releases its shared_ptr,
  // the work item is not destroyed.
  long_sleep.reset();
  ASSERT_FALSE(*long_sleep_destroyed);

  // The single thread in the thread pool is still running. Submit another task
  // that will queue. The task doesn't matter.
  unique_ptr<bool> queued_task_destroyed(new bool);
  std::shared_ptr<SleepWorkItem> queued_task(
      new SleepWorkItem(0, queued_task_destroyed.get()));
  Status queued_task_status = pool.SynchronousOffer(queued_task, 1);
  ASSERT_EQ(queued_task_status.code(), StatusCode::ThreadPoolTaskTimeoutError);
  // The work item is queued, and even if the caller releases its shared_ptr, the work
  // item is not destroyed.
  queued_task.reset();
  ASSERT_FALSE(*queued_task_destroyed);

  // Now, the queue is full. Any new task will fail to submit.
  unique_ptr<bool> fail_to_submit_destroyed(new bool);
  std::shared_ptr<SleepWorkItem> fail_to_submit(
      new SleepWorkItem(0, fail_to_submit_destroyed.get()));
  Status fail_to_submit_status = pool.SynchronousOffer(fail_to_submit, 1);
  ASSERT_EQ(fail_to_submit_status.code(), StatusCode::ThreadPoolSubmitFailed);
  // When the submit fails, the thread pool does not keep any shared_ptr to the work
  // item. When the caller releases its shared_ptr, the work item is immediately
  // destroyed.
  fail_to_submit.reset();
  ASSERT_TRUE(*fail_to_submit_destroyed);

  // The tasks will still complete
  pool.DrainAndShutdown();
  // The work items that the thread pool had running and in the queue are destroyed
  // when they complete (even though the caller long since released its shared_ptr).
  ASSERT_TRUE(*long_sleep_destroyed);
  ASSERT_TRUE(*queued_task_destroyed);
}

}

PEGASUS_TEST_MAIN();

ext: cc

max_stars_repo_name: jerrychenhf/pegasus

max_issues_repo_name: jerrychenhf/pegasus

max_forks_repo_name: jerrychenhf/pegasus


text:
/**
 * @file radon.cpp
 *
 */

#include "radon.h"
#include "logger.h"
#include "plugin_factory.h"
#include "point_list.h"
#include "util.h"
#include <boost/filesystem.hpp>
#include <sstream>
#include <thread>

using namespace std;
using namespace himan::plugin;

const int MAX_WORKERS = 32;
static once_flag oflag;
static map<string, string> tableNameCache;
static mutex tableNameMutex;

void radon::Init()
{
	if (!itsInit)
	{
		try
		{
			call_once(oflag, [&]() {

				string radonHost = "vorlon";

				try
				{
					radonHost = util::GetEnv("RADON_HOSTNAME");
				}
				catch (...)
				{
				}

				string radonName = "radon";

				try
				{
					radonName = util::GetEnv("RADON_DATABASENAME");
				}
				catch (...)
				{
				}

				int radonPort = 5432;

				try
				{
					radonPort = stoi(util::GetEnv("RADON_PORT"));
				}
				catch (...)
				{
				}

				NFmiRadonDBPool::Instance()->Username("wetodb");
				NFmiRadonDBPool::Instance()->Password(util::GetEnv("RADON_WETODB_PASSWORD"));
				NFmiRadonDBPool::Instance()->Database(radonName);
				NFmiRadonDBPool::Instance()->Hostname(radonHost);
				NFmiRadonDBPool::Instance()->Port(radonPort);

				if (NFmiRadonDBPool::Instance()->MaxWorkers() < MAX_WORKERS)
				{
					NFmiRadonDBPool::Instance()->MaxWorkers(MAX_WORKERS);
				}
				itsLogger.Info("Connected to radon (db=" + radonName + ", host=" + radonHost + ":" +
				               std::to_string(radonPort) + ")");
			});

			itsRadonDB = std::unique_ptr<NFmiRadonDB>(NFmiRadonDBPool::Instance()->GetConnection());
		}
		catch (int e)
		{
			itsLogger.Fatal("Failed to get connection");
			himan::Abort();
		}
		catch (const std::exception& e)
		{
			itsLogger.Fatal(e.what());
			himan::Abort();
		}
		itsInit = true;
	}
}

radon::radon() : itsInit(false), itsRadonDB()
{
	itsLogger = logger("radon");
}
void radon::PoolMaxWorkers(int maxWorkers)
{
	NFmiRadonDBPool::Instance()->MaxWorkers(maxWorkers);
}
vector<std::string> radon::CSV(search_options& options)
{
	Init();

	vector<string> csv;

	if (options.prod.Class() != kPreviClass)
	{
		itsLogger.Error("Grid producer does not have csv based data");
		return csv;
	}

	stringstream query;

	const auto analtime = options.time.OriginDateTime().String();
	const auto key = to_string(options.prod.Id()) + "_" + analtime;
	string tableName;

	const auto mapValue = tableNameCache.find(key);

	if (mapValue == tableNameCache.end())
	{
		query << "SELECT table_name FROM as_previ WHERE producer_id = " << options.prod.Id()
		      << " AND (min_analysis_time, max_analysis_time) OVERLAPS ('" << analtime << "', '" << analtime << "')";

		itsRadonDB->Query(query.str());

		const auto row = itsRadonDB->FetchRow();

		if (row.empty())
		{
			itsLogger.Error("No tables found from as_previ for producer " + options.prod.Name());
			return csv;
		}

		tableName = row[0];

		lock_guard<mutex> lock(tableNameMutex);
		tableNameCache[key] = tableName;
	}
	else
	{
		tableName = (*mapValue).second;
	}

	string levelValue2 = "-1";

	if (!IsKHPMissingValue(options.level.Value2()))
	{
		levelValue2 = boost::lexical_cast<string>(options.level.Value2());
	}

	const string period = util::MakeSQLInterval(options.time);
	string forecastTypeValue = "-1";  // default, deterministic/analysis

	if (options.ftype.Type() >= 3 && options.ftype.Type() <= 4)
	{
		forecastTypeValue = boost::lexical_cast<string>(options.ftype.Value());
	}

	query.str("");

	query << "SELECT "
	      << "t.station_id,"
	      << "s.name AS station_name,"
	      << "st_x(s.position) AS longitude,"
	      << "st_y(s.position) AS latitude,"
	      << "t.value "
	      << "FROM " << tableName << " t, station s, param p "
	      << "WHERE "
	      << "t.station_id = s.id "
	      << "AND t.analysis_time = '" << analtime << "' "
	      << "AND t.param_id = p.id "
	      << "AND p.name = '" << options.param.Name() << "' "
	      << "AND t.level_id = " << options.level.Type() << " "
	      << "AND t.level_value = " << options.level.Value() << " "
	      << "AND (t.level_value2 = " << options.level.Value2() << " OR t.level_value2 = -1) "
	      << "AND t.forecast_period = '" << period << "' "
	      << "AND t.forecast_type_id = " << options.ftype.Type() << " "
	      << "AND t.forecast_type_value = " << forecastTypeValue << " "
	      << "AND t.station_id IN (";

	const point_list* list = dynamic_cast<const point_list*>(options.configuration->BaseGrid());

	if (!list)
	{
		throw std::runtime_error("Input grid type is not point_list");
	}

	const auto stations = list->Stations();

	for (const auto& station : stations)
	{
		query << station.Id() << ",";
	}

	query.seekp(-1, std::ios_base::end);
	query << ")";

	itsRadonDB->Query(query.str());

	while (true)
	{
		query.str("");

		const auto row = itsRadonDB->FetchRow();

		if (row.empty())
		{
			break;
		}

		// producer_id,origintime,station_id,station_name,longitude,latitude,param_name,level_name,level_value,level_value2,forecast_period,forecast_type_id,forecast_type_value,value

		query << options.prod.Id() << "," << analtime << "," << row[0] << "," << row[1] << "," << row[2] << ","
		      << row[3] << "," << options.param.Name() << "," << HPLevelTypeToString.at(options.level.Type()) << ","
		      << options.level.Value() << "," << options.level.Value2() << "," << period << "," << options.ftype.Type()
		      << "," << options.ftype.Value() << "," << row[4] << endl;

		csv.push_back(query.str());
	}

	return csv;
}

vector<vector<string>> GetGridGeoms(himan::plugin::search_options& options, unique_ptr<NFmiRadonDB>& itsRadonDB)
{
	vector<vector<string>> gridgeoms;
	vector<string> sourceGeoms = options.configuration->SourceGeomNames();

	const string ref_prod = options.prod.Name();
	const string analtime = options.time.OriginDateTime().String("%Y-%m-%d %H:%M:%S+00");

	if (sourceGeoms.empty())
	{
		// Get all geometries
		gridgeoms = itsRadonDB->GetGridGeoms(ref_prod, analtime);
	}
	else
	{
		// TODO: This set of queries could be coalesced into just one query, but in that case we need
		// to make sure that the order of the returned geometries (if there are more than one) is the
		// same as what was given to us (i.e. what was specified in the configuration file).
		// See CreateFileSQLQuery() for guide how to do it.

		for (size_t i = 0; i < sourceGeoms.size(); i++)
		{
			vector<vector<string>> geoms = itsRadonDB->GetGridGeoms(ref_prod, analtime, sourceGeoms[i]);
			gridgeoms.insert(gridgeoms.end(), geoms.begin(), geoms.end());
		}
	}

	return gridgeoms;
}

string CreateFileSQLQuery(himan::plugin::search_options& options, const vector<vector<string>>& gridgeoms)
{
	const string analtime = options.time.OriginDateTime().String("%Y-%m-%d %H:%M:%S+00");
	string levelValue = boost::lexical_cast<string>(options.level.Value());
	string levelValue2 = "-1";

	if (options.level.Value2() != himan::kHPMissingValue)
	{
		levelValue2 = boost::lexical_cast<string>(options.level.Value2());
	}

	if (options.prod.Class() != himan::kGridClass)
	{
		throw runtime_error("Previ producer does not have file data");
	}

	const string level_name = himan::HPLevelTypeToString.at(options.level.Type());

	string forecastTypeValue = "-1";  // default, deterministic/analysis

	if (options.ftype.Type() >= 3 && options.ftype.Type() <= 4)
	{
		forecastTypeValue = boost::lexical_cast<string>(options.ftype.Value());
	}

	string forecastTypeId = boost::lexical_cast<string>(options.ftype.Type());

	if (options.time.Step().Hours() == 0 && options.ftype.Type() == 1)
	{
		// ECMWF (and maybe others) use forecast type id == 2 for analysis hour
		forecastTypeId += ",2";
	}

	const string parm_name = options.param.Name();

	// HIMAN-172: Reducing radon query count
	// Coalesce multiple-geometry queries into one if all are
	// using the same source table.
	// If the source table is not the same, make a union query.

	bool sameTableForAllGeometries = true;

	const string firstTable = gridgeoms[0][1];

	for (size_t i = 1; i < gridgeoms.size(); i++)
	{
		if (firstTable != gridgeoms[i][1])
		{
			sameTableForAllGeometries = false;
		}
	}

	stringstream query;

	if (sameTableForAllGeometries)
	{
		const std::string partition = gridgeoms[0][5];
		const std::string schema = gridgeoms[0][4];

		// clang-format off

		query << "SELECT t.file_location, g.name, byte_offset, byte_length, file_format_id, file_protocol_id, message_no, t.file_server "
		      << "FROM " << schema << "." << partition << " t, geom g, param p, level l"
		      << " WHERE t.geometry_id = g.id"
		      << " AND t.producer_id = " << options.prod.Id()
		      << " AND t.param_id = p.id"
		      << " AND l.id = t.level_id"
		      << " AND t.analysis_time = '" << analtime << "'"
		      << " AND p.name = '" << parm_name << "'"
		      << " AND l.name = upper('" << level_name << "')"
		      << " AND t.level_value = " << levelValue << " AND t.level_value2 = " << levelValue2
		      << " AND t.forecast_period = '" << himan::util::MakeSQLInterval(options.time) << "'"
		      << " AND t.file_format_id NOT IN (3,4)" // no netcdf
		      << " AND forecast_type_id IN (" << forecastTypeId << ")"
		      << " AND forecast_type_value = " << forecastTypeValue << " AND g.id IN (";

		// clang-format on

		for (const auto& geom : gridgeoms)
		{
			query << geom[0] << ",";
		}

		query.seekp(-1, ios_base::end);

		// Add custom sort order, as we want to preserve to order from conf file

		query << ") ORDER BY forecast_period, level_id, level_value"
		      << ", array_position(array[";

		for (const auto& geom : gridgeoms)
		{
			query << "'" << geom[0] << "',";
		}

		query.seekp(-1, ios_base::end);
		query << "], geometry_id::text)";
	}
	else
	{
		for (size_t i = 0; i < gridgeoms.size(); i++)
		{
			string tablename = gridgeoms[i][1];
			string geomid = gridgeoms[i][0];

			query << "SELECT file_location, geometry_name, byte_offset, byte_length, file_format_id, file_protocol_id, "
			         "message_no "
			      << "FROM " << tablename << "_v "
			      << "WHERE analysis_time = '" << analtime << "'"
			      << " AND param_name = '" << parm_name << "'"
			      << " AND level_name = upper('" << level_name << "') "
			      << " AND level_value = " << levelValue << " AND level_value2 = " << levelValue2
			      << " AND forecast_period = '" << himan::util::MakeSQLInterval(options.time) << "'"
			      << " AND t.file_format_id NOT IN (3,4)"  // no netcdf
			      << " AND geometry_id = " << geomid << " AND forecast_type_id IN (" << forecastTypeId << ")"
			      << " AND forecast_type_value = " << forecastTypeValue << " UNION ALL";
		}

		query.seekp(-9, ios_base::end);
		query << " ORDER BY forecast_period, level_id, level_value";
	}

	return query.str();
}

vector<himan::file_information> radon::Files(search_options& options)
{
	Init();

	vector<string> values;
	vector<file_information> ret;

	const auto gridgeoms = GetGridGeoms(options, itsRadonDB);

	if (gridgeoms.empty())
	{
		return ret;
	}

	const auto query = CreateFileSQLQuery(options, gridgeoms);

	if (query.empty())
	{
		return ret;
	}

	try
	{
		itsRadonDB->Query(query);
		values = itsRadonDB->FetchRow();
	}
	catch (const pqxx::sql_error& e)
	{
		// Sometimes we get errors like:
		// ERROR:  deadlock detected
		// DETAIL:  Process 23465 waits for AccessShareLock on relation 35841462 of database 32027825; blocked by
		// process 23477.
		//
		// This is caused when table partitions are dropped while himan is trying to query the table.
		// As a workaround, re-execute the query.

		itsLogger.Warning("Caught database error: " + string(e.what()));
		sleep(1);
		itsRadonDB->Query(query);
		values = itsRadonDB->FetchRow();
	}

	if (values.empty())
	{
		return ret;
	}

	itsLogger.Trace(fmt::format("Found data for parameter {} from radon geometry {}, file name '{}' position {}/{}:{}",
	                            options.param.Name(), values[1], values[0], values[6], values[2], values[3]));

	file_information finfo;
	finfo.file_location = values[0];
	finfo.file_server = values[7];
	switch (stoi(values[4]))
	{
		case 1:
		case 2:
			finfo.file_type = static_cast<HPFileType>(stoi(values[4]));
			break;
		case 3:
		case 4:
			finfo.file_type = kNetCDF;
			break;
		case 5:
			finfo.file_type = kGeoTIFF;
			break;
		default:
			itsLogger.Error("Unknown file type: " + values[4]);
			return ret;
	}

	finfo.storage_type = static_cast<HPFileStorageType>(stoi(values[5]));

	try
	{
		finfo.offset = static_cast<unsigned long>(stoul(values[2]));
		finfo.length = static_cast<unsigned long>(stoul(values[3]));
	}
	catch (const invalid_argument& e)
	{
		finfo.offset = boost::none;
		finfo.length = boost::none;
	}

	try
	{
		finfo.message_no = static_cast<unsigned long>(stoul(values[6]));
	}
	catch (const invalid_argument& e)
	{
		finfo.message_no = boost::none;
	}

	return {finfo};
}

bool radon::Save(const info<double>& resultInfo, const file_information& finfo, const string& targetGeomName)
{
	return Save<double>(resultInfo, finfo, targetGeomName);
}

template <typename T>
bool radon::Save(const info<T>& resultInfo, const file_information& finfo, const string& targetGeomName)
{
	Init();

	if (resultInfo.Producer().Class() == kGridClass)
	{
		return SaveGrid(resultInfo, finfo, targetGeomName);
	}
	else if (resultInfo.Producer().Class() == kPreviClass)
	{
		return SavePrevi(resultInfo);
	}

	return false;
}

template bool radon::Save<double>(const info<double>&, const file_information&, const string&);
template bool radon::Save<float>(const info<float>&, const file_information&, const string&);

template <typename T>
bool radon::SavePrevi(const info<T>& resultInfo)
{
	stringstream query;

	auto analysisTime = resultInfo.Time().OriginDateTime().String("%Y-%m-%d %H:%M:%S+00");

	query << "SELECT id,table_name FROM as_previ WHERE producer_id = " << resultInfo.Producer().Id()
	      << " AND (min_analysis_time, max_analysis_time) OVERLAPS ('" << analysisTime << "', '" << analysisTime
	      << "')";

	itsRadonDB->Query(query.str());

	auto row = itsRadonDB->FetchRow();

	if (row.empty())
	{
		itsLogger.Warning("Data set definition not found from radon");
		return false;
	}

	string table_name = row[1];

	auto levelinfo = itsRadonDB->GetLevelFromDatabaseName(HPLevelTypeToString.at(resultInfo.Level().Type()));

	if (levelinfo.empty())
	{
		itsLogger.Error("Level information not found from radon for level " +
		                HPLevelTypeToString.at(resultInfo.Level().Type()) + ", producer " +
		                to_string(resultInfo.Producer().Id()));
		return false;
	}

	auto paraminfo = itsRadonDB->GetParameterFromDatabaseName(resultInfo.Producer().Id(), resultInfo.Param().Name(),
	                                                          stoi(levelinfo["id"]), resultInfo.Level().Value());

	if (paraminfo.empty())
	{
		itsLogger.Error("Parameter information not found from radon for parameter " + resultInfo.Param().Name() +
		                ", producer " + to_string(resultInfo.Producer().Id()));
		return false;
	}

	int forecastTypeValue = -1;  // default, deterministic/analysis

	if (resultInfo.ForecastType().Type() >= 3 && resultInfo.ForecastType().Type() <= 4)
	{
		forecastTypeValue = static_cast<int>(resultInfo.ForecastType().Value());
	}

	double levelValue2 = IsKHPMissingValue(resultInfo.Level().Value2()) ? -1 : resultInfo.Level().Value2();

	auto localInfo = resultInfo;

	for (localInfo.ResetLocation(); localInfo.NextLocation();)
	{
		query.str("");

		query << "INSERT INTO data." << table_name
		      << " (producer_id, station_id, analysis_time, param_id, level_id, "
		         "level_value, level_value2, forecast_period, "
		         "forecast_type_id, forecast_type_value, value) VALUES ("
		      << localInfo.Producer().Id() << ", " << localInfo.Station().Id() << ", "
		      << "'" << analysisTime << "', " << paraminfo["id"] << ", " << levelinfo["id"] << ", "
		      << localInfo.Level().Value() << ", " << levelValue2 << ", "
		      << "'" << util::MakeSQLInterval(localInfo.Time()) << "', "
		      << static_cast<int>(localInfo.ForecastType().Type()) << ", " << forecastTypeValue << ","
		      << localInfo.Value() << ")";

		try
		{
			itsRadonDB->Execute(query.str());
			itsRadonDB->Commit();
		}
		catch (const pqxx::unique_violation& e)
		{
			itsRadonDB->Rollback();

			query.str("");
			query << "UPDATE data." << table_name << " SET "
			      << "value = " << localInfo.Value() << " WHERE "
			      << "producer_id = " << resultInfo.Producer().Id() << " AND "
			      << "station_id = " << localInfo.Station().Id() << " AND "
			      << "analysis_time = '" << analysisTime << "' AND "
			      << "param_id = " << paraminfo["id"] << " AND "
			      << "level_id = " << levelinfo["id"] << " AND "
			      << "level_value = " << resultInfo.Level().Value() << " AND "
			      << "level_value2 = " << levelValue2 << " AND "
			      << "forecast_period = "
			      << "'" << util::MakeSQLInterval(resultInfo.Time()) << "' AND "
			      << "forecast_type_id = " << static_cast<int>(resultInfo.ForecastType().Type()) << " AND "
			      << "forecast_type_value = " << forecastTypeValue;

			itsRadonDB->Execute(query.str());
			itsRadonDB->Commit();
		}
	}

	return true;
}

template bool radon::SavePrevi<double>(const info<double>&);
template bool radon::SavePrevi<float>(const info<float>&);

template <typename T>
bool radon::SaveGrid(const info<T>& resultInfo, const file_information& finfo, const string& targetGeomName)
{
	stringstream query;

	if (resultInfo.Grid()->Class() != kRegularGrid)
	{
		itsLogger.Error("Only regular grid data can be stored to radon for now");
		return false;
	}

	// Start by trying to search with the geometry name. If there are duplicates, the 'wrong' geometry maybe returned if
	// we search with the geometry definition first.
	map<string, string> geominfo;

	if (!targetGeomName.empty())
	{
		geominfo = itsRadonDB->GetGeometryDefinition(targetGeomName);
	}

	if (geominfo.empty())
	{
		if (resultInfo.Grid()->Class() == kRegularGrid)
		{
			auto gr = dynamic_pointer_cast<regular_grid>(resultInfo.Grid());

			// rotated coordinates as the first point for rotated_latitude_longitude

			const himan::point fp =
			    (gr->Type() == himan::kRotatedLatitudeLongitude)
			        ? dynamic_pointer_cast<himan::rotated_latitude_longitude_grid>(gr)->Rotate(gr->FirstPoint())
			        : gr->FirstPoint();

			geominfo =
			    itsRadonDB->GetGeometryDefinition(gr->Ni(), gr->Nj(), fp.Y(), fp.X(), gr->Di(), gr->Dj(), gr->Type());
		}
	}

	if (geominfo.empty())
	{
		itsLogger.Warning("Grid geometry not found from radon");
		return false;
	}

	const string geom_id = geominfo["id"];
	const string geom_name = geominfo["name"];
	auto analysisTime = resultInfo.Time().OriginDateTime().String("%Y-%m-%d %H:%M:%S+00");

	auto tableinfo = itsRadonDB->GetTableName(resultInfo.Producer().Id(), analysisTime, geom_name);

	if (tableinfo.empty())
	{
		itsLogger.Warning("Data set definition not found from radon");
		return false;
	}

	const string schema_name = tableinfo["schema_name"];
	const string table_name = tableinfo["partition_name"];
	const string record_count = tableinfo["record_count"];

	query.str("");

	string host;

	switch (finfo.storage_type)
	{
		case kLocalFileSystem:
		{
			char host_[128];
			gethostname(host_, 128);
			host = string(host_);
		}
		break;
		case kS3ObjectStorageSystem:
		{
			const char* host_ = getenv("S3_HOSTNAME");
			host = string(host_);
		}
		break;
		default:
			break;
	}

	if (host.empty())
	{
		itsLogger.Error("Hostname could not be determined");
		himan::Abort();
	}

	auto levelinfo = itsRadonDB->GetLevelFromDatabaseName(HPLevelTypeToString.at(resultInfo.Level().Type()));

	if (levelinfo.empty())
	{
		itsLogger.Error("Level information not found from radon for level " +
		                HPLevelTypeToString.at(resultInfo.Level().Type()) + ", producer " +
		                to_string(resultInfo.Producer().Id()));
		return false;
	}

	if (resultInfo.Param().Id() == kHPMissingInt)
	{
		itsLogger.Error("Parameter information not found from radon for parameter " + resultInfo.Param().Name() +
		                ", producer " + to_string(resultInfo.Producer().Id()));
		return false;
	}

	/*
	 * We have our own error logging for unique key violations
	 */

	// itsRadonDB->Verbose(false);
	int forecastTypeValue = -1;  // default, deterministic/analysis

	if (resultInfo.ForecastType().Type() >= 3 && resultInfo.ForecastType().Type() <= 4)
	{
		forecastTypeValue = static_cast<int>(resultInfo.ForecastType().Value());
	}

	double levelValue2 = IsKHPMissingValue(resultInfo.Level().Value2()) ? -1 : resultInfo.Level().Value2();
	const string fullTableName = schema_name + "." + table_name;

	auto FormatToSQL = [](const boost::optional<unsigned long>& opt) -> string {
		if (opt)
		{
			return to_string(opt.get());
		}

		return "NULL";
	};

	int radonFileFormat = 0;

	switch (finfo.file_type)
	{
		case kGRIB1:
			radonFileFormat = 1;
			break;
		case kGRIB2:
			radonFileFormat = 2;
			break;
		case kNetCDF:
			radonFileFormat = 4;
			break;
		case kGeoTIFF:
			radonFileFormat = 5;
			break;
		default:
			itsLogger.Error("Unknown file type: " + to_string(finfo.file_type));
			himan::Abort();
	}

	query
	    << "INSERT INTO " << fullTableName
	    << " (producer_id, analysis_time, geometry_id, param_id, level_id, level_value, level_value2, forecast_period, "
	    << "forecast_type_id, forecast_type_value, file_location, file_server, file_format_id, file_protocol_id, "
	    << "message_no, byte_offset, byte_length) VALUES (" << resultInfo.Producer().Id() << ", "
	    << "'" << analysisTime << "', " << geom_id << ", " << resultInfo.Param().Id() << ", " << levelinfo["id"] << ", "
	    << resultInfo.Level().Value() << ", " << levelValue2 << ", "
	    << "'" << util::MakeSQLInterval(resultInfo.Time()) << "', "
	    << static_cast<int>(resultInfo.ForecastType().Type()) << ", " << forecastTypeValue << ","
	    << "'" << finfo.file_location << "', "
	    << "'" << host << "', " << radonFileFormat << ", " << finfo.storage_type << ", "
	    << FormatToSQL(finfo.message_no) << ", " << FormatToSQL(finfo.offset) << ", " << FormatToSQL(finfo.length)
	    << ")";

	try
	{
		itsRadonDB->Execute(query.str());
		itsRadonDB->Commit();

		// After first insert set record_count to 1, to mark that this partition has data

		if (record_count == "0")
		{
			itsLogger.Trace("Updating as_grid record_count column for " + fullTableName);

			query.str("");
			query << "UPDATE as_grid SET record_count = 1 WHERE schema_name = '" << schema_name
			      << "' AND partition_name = '" << table_name << "' AND analysis_time = '" << analysisTime << "'";

			itsRadonDB->Execute(query.str());
		}
	}
	catch (const pqxx::unique_violation& e)
	{
		itsRadonDB->Rollback();

		query.str("");
		query << "UPDATE " << fullTableName << " SET "
		      << "file_location = '" << finfo.file_location << "', "
		      << "file_server = '" << host << "', "
		      << "file_format_id = " << radonFileFormat << ", "
		      << "file_protocol_id = " << finfo.storage_type << ", "
		      << "message_no = " << FormatToSQL(finfo.message_no) << ", "
		      << "byte_offset = " << FormatToSQL(finfo.offset) << ", "
		      << "byte_length = " << FormatToSQL(finfo.length) << " WHERE "
		      << "producer_id = " << resultInfo.Producer().Id() << " AND "
		      << "analysis_time = '" << analysisTime << "' AND "
		      << "geometry_id = " << geom_id << " AND "
		      << "param_id = " << resultInfo.Param().Id() << " AND "
		      << "level_id = " << levelinfo["id"] << " AND "
		      << "level_value = " << resultInfo.Level().Value() << " AND "
		      << "level_value2 = " << levelValue2 << " AND "
		      << "forecast_period = "
		      << "'" << util::MakeSQLInterval(resultInfo.Time()) << "' AND "
		      << "forecast_type_id = " << static_cast<int>(resultInfo.ForecastType().Type()) << " AND "
		      << "forecast_type_value = " << forecastTypeValue;

		itsRadonDB->Execute(query.str());
		itsRadonDB->Commit();
	}

	query.str("");
	query << "Saved information on file '" << finfo.file_location << "'";

	if (finfo.message_no)
	{
		query << " message no " << finfo.message_no.get();
	}

	itsLogger.Trace(query.str());

	return true;
}

template bool radon::SaveGrid<double>(const info<double>&, const file_information&, const string&);
template bool radon::SaveGrid<float>(const info<float>&, const file_information&, const string&);

ext: cpp


text:
// ------------------------------------------------------------
// Copyright (c) Microsoft Corporation.  All rights reserved.
// Licensed under the MIT License (MIT). See License.txt in the repo root for license information.
// ------------------------------------------------------------

#include "stdafx.h"

#include <boost/test/unit_test.hpp>
#include "Common/boost-taef.h"

#include "FabricRuntime.h"
#include "Common/Common.h"
#include "Hosting2/FabricNodeHost.Test.h"

using namespace std;
using namespace Common;
using namespace Hosting2;
using namespace Transport;
using namespace Fabric;
using namespace ServiceModel;
using namespace Management;
using namespace ImageModel;

const StringLiteral TraceType("ProcessActivatorTest");

class ProcessActivatorTest
{
protected:
    // This test should be run with elevated Admin access
    // Setup checks that the process runs elevated; otherwise, it fails;
	ProcessActivatorTest()
		: fabricNodeHost_(make_shared<TestFabricNodeHost>())
		, maxOpenRetryCount_(3)
		, openRetryDelayInMilliSec_(2000)
	{

		BOOST_REQUIRE(Setup());
	}

    TEST_CLASS_SETUP(Setup);
    ~ProcessActivatorTest() { BOOST_REQUIRE(Cleanup()); }
    TEST_CLASS_CLEANUP(Cleanup);

    //Disabling this test till we resolve 1226706
    //TEST_METHOD(ProcessRedirectionFileSizeLarge);
    static wstring GetUnitTestFolder();
    static bool FileStartsWith(File & file, wstring const & searchString);
    static void DeactivateProcessTest(shared_ptr<TestFabricNodeHost> const & fabricNodeHost, bool forceTerminateProcess);
    static ErrorCode CreateProcessSyncEvent(__out HandleUPtr & eventHandle);
    static void OpenFileWithRetry(int retryMaxCount, std::wstring const & fileName, File & fileHandle);
    static void ProcessRedirectionFileSizeTest(shared_ptr<TestFabricNodeHost> const & fabricNodeHost, int maxSize);
    static wstring ProcessSyncEventName;

	int maxOpenRetryCount_;
	int openRetryDelayInMilliSec_;
	shared_ptr<TestFabricNodeHost> fabricNodeHost_;
};

wstring ProcessActivatorTest::ProcessSyncEventName = L"HostingProcessSync";

BOOST_FIXTURE_TEST_SUITE(ProcessActivatorTestSuite,ProcessActivatorTest)

BOOST_AUTO_TEST_CASE(ProcessRedirectionFileRetention)
{
	Trace.WriteInfo(TraceType, "BEGIN TEST: ProcessRedirectionFileRetention");

    int retentionCount = 3;
    LONG maxSize = 20480;
    HostingConfig::GetConfig().EnableActivateNoWindow = true;

    wstring unitTestsFolder = ProcessActivatorTest::GetUnitTestFolder();
    wstring processName = Path::Combine(unitTestsFolder, L"Hosting2.TestProcess.exe");
    wstring redirectionFileName = L"Hosting2TestProcessLog";

    Hosting2::ProcessActivator pa(
        *fabricNodeHost_->GetFabricNode()->CreateComponentRoot());

    // Process Description
    wstring logFolder = Directory::GetCurrentDirectoryW();

    Common::EnvironmentMap emap;
    Hosting2::ProcessDescription pd(
        processName,
        L"consoleWrite",
        unitTestsFolder,
        emap,
        logFolder,
        logFolder,
        logFolder,
        logFolder,
        true,
        true,
        false,
        false,
        redirectionFileName,
        retentionCount,
        maxSize);

    // Activate multiple processes
    for (int i = 0 ; i<5; i++)
    {
        AsyncOperationWaiterSPtr activateWaiter = make_shared<AsyncOperationWaiter>();
        IProcessActivationContextSPtr processActivationContext;
        auto async = pa.BeginActivate(
            nullptr,
            pd,
            fabricNodeHost_->GetHosting().FabricBinFolder,
            [](pid_t, Common::ErrorCode const &, DWORD) {},
            TimeSpan::MaxValue,
            [activateWaiter, &pa, &processActivationContext](AsyncOperationSPtr const & operation)
            { activateWaiter->SetError(pa.EndActivate(operation, processActivationContext)); activateWaiter->Set(); },
            fabricNodeHost_->GetFabricNode()->CreateAsyncOperationRoot());

        activateWaiter->WaitOne();
        VERIFY_IS_TRUE(activateWaiter->GetError().IsSuccess());
        processActivationContext->TerminateAndCleanup(false, 0);
    }

    // Look for files
    wstring pattern = redirectionFileName;
    pattern.append(L"*.out");

    // Verify that the files generated are not more than the retention count.
    //
    auto files = Common::Directory::Find(logFolder, pattern, 5, false);
    Trace.WriteInfo(TraceType, "Found {0} files. Expected {1}", files.size(), retentionCount);
    for (auto const & file : files)
    {
        Trace.WriteInfo(TraceType, "Found {0}", file);
    }
    VERIFY_IS_TRUE((files.size() == static_cast<size_t>(retentionCount)), L"Found different number files than expected");

	Trace.WriteInfo(TraceType, "END TEST: ProcessRedirectionFileRetention");
}

BOOST_AUTO_TEST_CASE(ProcessRedirectionFileSize)
{
	Trace.WriteInfo(TraceType, "BEGIN TEST: ProcessRedirectionFileSize");

    ProcessRedirectionFileSizeTest(fabricNodeHost_, 5);

	Trace.WriteInfo(TraceType, "END TEST: ProcessRedirectionFileSize");
}

//Disabling this test till we resolve 1226706
//void ProcessActivatorTest::ProcessRedirectionFileSizeLarge()
//{
//    ProcessRedirectionFileSizeTest(1024);
//}

BOOST_AUTO_TEST_CASE(ProcessDeactivationCtrlC)
{
	Trace.WriteInfo(TraceType, "BEGIN TEST: ProcessDeactivationCtrlC");

    DeactivateProcessTest(fabricNodeHost_, false);

	Trace.WriteInfo(TraceType, "END TEST: ProcessDeactivationCtrlC");
}

BOOST_AUTO_TEST_CASE(ProcessDeactivationForce)
{
	Trace.WriteInfo(TraceType, "BEGIN TEST: ProcessDeactivationForce");

    DeactivateProcessTest(fabricNodeHost_, true);

	Trace.WriteInfo(TraceType, "END TEST: ProcessDeactivationForce");
}

BOOST_AUTO_TEST_SUITE_END()

bool ProcessActivatorTest::Setup()
{
	for (int retryCount = 1; retryCount <= maxOpenRetryCount_; retryCount++)
	{
		if (fabricNodeHost_->Open())
		{
			return true;
		}

		Sleep(openRetryDelayInMilliSec_);
	}

	return false;
}

bool ProcessActivatorTest::Cleanup()
{
   return fabricNodeHost_->Close();
}

wstring ProcessActivatorTest::GetUnitTestFolder()
{
    wstring unitTestsFolder;
#if !defined(PLATFORM_UNIX)
    VERIFY_IS_TRUE(Environment::ExpandEnvironmentStringsW(L"%_NTTREE%\\FabricUnitTests", unitTestsFolder));
#else
    VERIFY_IS_TRUE(Environment::ExpandEnvironmentStringsW(L"%_NTTREE%/test", unitTestsFolder));
#endif
	return unitTestsFolder;
}

void ProcessActivatorTest::ProcessRedirectionFileSizeTest(shared_ptr<TestFabricNodeHost> const & fabricNodeHost, int maxSize)
{
    int retentionCount = 1;
    HostingConfig::GetConfig().EnableActivateNoWindow = true;

    wstring unitTestsFolder = ProcessActivatorTest::GetUnitTestFolder();
    wstring processName = Path::Combine(unitTestsFolder, L"Hosting2.TestProcess.exe");
    wstring redirectionFileName = L"Hosting2TestProcessLogFileSize";

    Hosting2::ProcessActivator pa(
        *fabricNodeHost->GetFabricNode()->CreateComponentRoot());

    // ProcessDescription
    wstring logFolder = Directory::GetCurrentDirectoryW();
    Common::EnvironmentMap emap;
    Hosting2::ProcessDescription pd(
        processName,
        L"consoleWrite",
        unitTestsFolder,
        emap,
        logFolder,
        logFolder,
        logFolder,
        logFolder,
        true,
        true,
        false,
        false,
        redirectionFileName,
        retentionCount,
        maxSize);

    HandleUPtr eventHandle;
    {
        // Create event to sync the host process.
        VERIFY_IS_TRUE(ProcessActivatorTest::CreateProcessSyncEvent(eventHandle).IsSuccess());

        // Activate process
        AsyncOperationWaiterSPtr activateWaiter = make_shared<AsyncOperationWaiter>();
        IProcessActivationContextSPtr processActivationContext;
        auto async = pa.BeginActivate(
            nullptr,
            pd,
            fabricNodeHost->GetHosting().FabricBinFolder,
            [](pid_t, Common::ErrorCode const &, DWORD) {},
            TimeSpan::MaxValue,
            [activateWaiter, &pa, &processActivationContext](AsyncOperationSPtr const & operation)
            { activateWaiter->SetError(pa.EndActivate(operation, processActivationContext)); activateWaiter->Set(); },
            fabricNodeHost->GetFabricNode()->CreateAsyncOperationRoot());
        activateWaiter->WaitOne();
        VERIFY_IS_TRUE(activateWaiter->GetError().IsSuccess());

        // Wait for the process to start-up and write to console
        auto waitResult = ::WaitForSingleObject(eventHandle->Value, 10000);
        VERIFY_IS_TRUE(
            WAIT_OBJECT_0 == waitResult,
            wformatString("Actual wait result = {0}", waitResult).c_str());
    }

    // Look for the out files
    wstring pattern = redirectionFileName;
    pattern.append(L"*.out");

    vector<wstring> files = Common::Directory::Find(logFolder, pattern, 1, false);
    Trace.WriteInfo(TraceType, "Found {0} files. ", files.size());

    // for small size, the file will be overwritten. Hence there is a possiblity that
    // we have the file deleted or 0 bytes
    if (maxSize < 50)
    {
        // There is a possiblity that we closed just after deleting. In that case this 0 is acceptable
        VERIFY_IS_TRUE((files.size() <= 1), wformatString("Found {0} files", files.size()).c_str());
        if (files.size() == 0) { return; }

        File file;
        OpenFileWithRetry(5, files[0], file);

        // Verify that the file size is within the limit
        // It can be zero if we closed just after creating a new file
        DWORD size = ::GetFileSize(file.GetHandle(), nullptr);
        VERIFY_IS_TRUE(
            (size < (DWORD)(maxSize * 1024)),
            wformatString("File size is = {0}", size).c_str());
    }
    else
    {
        VERIFY_IS_TRUE((files.size() == 1), wformatString("Found {0} files", files.size()).c_str());

        DWORD size = 0;
        int retryCount = 0;
        File file;
        OpenFileWithRetry(5, files[0], file);

        while (++retryCount < 5)
        {
            // At times the file is not flushed. So the size is 0. Retry in that case.
            size = ::GetFileSize(file.GetHandle(), nullptr);
            if (size > 0) break;
            Sleep(500);
        }

        VERIFY_IS_TRUE(
            (size < (DWORD)(maxSize * 1024) && size > 0),
            wformatString("File size is = {0}", size).c_str());

        // Verify the content
        VERIFY_IS_TRUE(ProcessActivatorTest::FileStartsWith(file, L"Writing test data to console"), L"Does not contain string");
    }

    VERIFY_IS_TRUE(::ResetEvent(eventHandle->Value) == TRUE ? true : false);
}

void ProcessActivatorTest::OpenFileWithRetry(int retryMaxCount, std::wstring const & fileName, File & fileHandle)
{
    int retryCount = 0;
    while (++retryCount < retryMaxCount)
    {
        auto error = fileHandle.TryOpen(fileName, FileMode::Open, FileAccess::Read, FileShare::Read);
        if (error.IsSuccess())
        {
            Trace.WriteInfo(TraceType, "Opened {0}: {1}", fileName, error);
            break;
        }
        else
        {
            Trace.WriteWarning(TraceType, "Failed to open {0}: {1}", fileName, error);
            Sleep(1000);
            continue;
        }
    }
    VERIFY_IS_TRUE(fileHandle.GetHandle() != INVALID_HANDLE_VALUE);
}

bool ProcessActivatorTest::FileStartsWith(File & file, wstring const & searchString)
{
    char buffer[1000];
    DWORD bytesRead;
    wchar_t wideBuffer[1000];
    file.TryRead2(buffer, 1000, bytesRead);
    ::MultiByteToWideChar(CP_ACP, 0, buffer, bytesRead, wideBuffer, 1000);
    return StringUtility::Contains<wstring>(wstring(wideBuffer), searchString);
}

void ProcessActivatorTest::DeactivateProcessTest(shared_ptr<TestFabricNodeHost> const & fabricNodeHost, bool forceTerminateProcess)
{
    HostingConfig::GetConfig().EnableActivateNoWindow = true;

    wstring unitTestsFolder = ProcessActivatorTest::GetUnitTestFolder();
    wstring processName = Path::Combine(unitTestsFolder, L"Hosting2.TestProcess.exe");
    wstring argName = forceTerminateProcess ? L"CtrlCBlock" : L"CtrlCExit";
    wstring redirectionFileName = wformatString("{0}_{1}", L"Hosting2TestCtrlC", argName);

    // Create process description
    Hosting2::ProcessActivator pa(
        *fabricNodeHost->GetFabricNode()->CreateComponentRoot());

    wstring logFolder = Directory::GetCurrentDirectoryW();

    Common::EnvironmentMap emap;
    Hosting2::ProcessDescription pd(
        processName,
        argName,
        unitTestsFolder,
        emap,
        logFolder,
        logFolder,
        logFolder,
        logFolder,
        false,
        false,
        false,
        false);

    // Create Event for syncing the process.
    // This is needed to ensure that we do not terminate the process before we set the
    // CtrlC handler in the process.
    HandleUPtr eventHandle;
    VERIFY_IS_TRUE(ProcessActivatorTest::CreateProcessSyncEvent(eventHandle).IsSuccess());

    // Create process
    AsyncOperationWaiterSPtr activateWaiter = make_shared<AsyncOperationWaiter>();
    IProcessActivationContextSPtr processActivationContext;
    auto async = pa.BeginActivate(
        nullptr,
        pd,
        fabricNodeHost->GetHosting().FabricBinFolder,
        [](pid_t, Common::ErrorCode const &, DWORD) {},
        TimeSpan::MaxValue,
        [activateWaiter, &pa, &processActivationContext](AsyncOperationSPtr const & operation)
        { activateWaiter->SetError(pa.EndActivate(operation, processActivationContext)); activateWaiter->Set(); },
        fabricNodeHost->GetFabricNode()->CreateAsyncOperationRoot());
    activateWaiter->WaitOne();
    VERIFY_IS_TRUE(activateWaiter->GetError().IsSuccess());

    // Wait for the event that was created previously. Hosting2.TestProcess.exe will set
    // the event to indicate that it has registerd the ctrl C handler
    auto waitResult = ::WaitForSingleObject(eventHandle->Value, 10000);
    VERIFY_IS_TRUE(
        WAIT_OBJECT_0 == waitResult,
        wformatString("Actual wait result = {0}", waitResult).c_str());

    // Deactivate the process
    AsyncOperationWaiterSPtr deactivateWaiter = make_shared<AsyncOperationWaiter>();
    auto asyncOp = pa.BeginDeactivate(
        processActivationContext,
        TimeSpan::FromSeconds(10),
        [deactivateWaiter, &pa](AsyncOperationSPtr const & operation)
        { deactivateWaiter->SetError(pa.EndDeactivate(operation)); deactivateWaiter->Set(); },
        fabricNodeHost->GetFabricNode()->CreateAsyncOperationRoot());
    deactivateWaiter->WaitOne();
    VERIFY_IS_TRUE(deactivateWaiter->GetError().IsSuccess());

    // Wait for the process to shutdown
    waitResult = ::WaitForSingleObject(static_pointer_cast<ProcessActivationContext>(processActivationContext)->ProcessHandle->Value, 10000);
    VERIFY_IS_TRUE(
        WAIT_OBJECT_0 == waitResult,
        wformatString("Process did not shut down. Actual wait result = {0}", waitResult).c_str());

    // Verify that the deactivation code is expected
    DWORD processExitCode = 0;
    VERIFY_IS_TRUE(
        ::GetExitCodeProcess(static_pointer_cast<ProcessActivationContext>(processActivationContext)->ProcessHandle->Value, &processExitCode) == TRUE ? true : false,
        Common::wformatString("Could not retrieve the exit code for process. Error = {0}", ErrorCode::FromWin32Error()).c_str());

    DWORD expectedExitCode = (forceTerminateProcess ?  ProcessActivator::ProcessDeactivateExitCode : STATUS_CONTROL_C_EXIT);
    VERIFY_IS_TRUE(
        processExitCode == expectedExitCode,
        Common::wformatString("Process did not exit with expected exit code. Expected = {0}, Actual = {1}", expectedExitCode, processExitCode).c_str());

    VERIFY_IS_TRUE(deactivateWaiter->WaitOne(15000), L"Deactivate did not finsih in 15 sec");
    VERIFY_IS_TRUE(deactivateWaiter->GetError().IsSuccess(), L"Process did not shutdown");
}

// Create an event here. It will be opened by the child process.
// Using CreateEvent instead of Common::ManualResetEvent as ManualResetEvent creates
// event with SecurityAttribute as null restricting the event to be inherited.
ErrorCode ProcessActivatorTest::CreateProcessSyncEvent(__out HandleUPtr & eventHandle)
{
    SECURITY_ATTRIBUTES saAttr;
    saAttr.nLength = sizeof(SECURITY_ATTRIBUTES);
    saAttr.bInheritHandle = true;
    saAttr.lpSecurityDescriptor = NULL;

    eventHandle = make_unique<Handle>(
        ::CreateEventW(
        &saAttr,
        true,
        false,
        ProcessSyncEventName.c_str()));
    return (eventHandle->Value == INVALID_HANDLE_VALUE) ? ErrorCode::FromWin32Error() : ErrorCode::Success();
}

max_stars_repo_name: gridgentoo/ServiceFabricAzure

max_issues_repo_name: gridgentoo/ServiceFabricAzure

max_forks_repo_name: gridgentoo/ServiceFabricAzure