text:
#include "SystemCommonPch.h"
#include "CopyEx.h"
#include "RuntimeFastPath.h"
#include "common/simd/SIMD.hpp"
#include "common/simd/SIMD128.hpp"
#include "common/simd/SIMD256.hpp"
#include <boost/predef/other/endian.h>
#define HALF_ENABLE_F16C_INTRINSICS 0 // only use half as fallback
#include "3rdParty/half/half.hpp"
#if !BOOST_ENDIAN_LITTLE_BYTE
#   error("unsupported std::byte order (non little endian)")
#endif
#if COMMON_ARCH_ARM && COMMON_COMPILER_MSVC
using float16_t = uint16_t;
#endif

using namespace std::string_view_literals;
using namespace common::simd;
using common::CheckCPUFeature;
using common::CopyManager;

#define Broadcast2Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
#define Broadcast4Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
DEFINE_FASTPATH(CopyManager, Broadcast2);
DEFINE_FASTPATH(CopyManager, Broadcast4);
#define ZExtCopy12Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
#define ZExtCopy14Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
#define ZExtCopy24Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
#define ZExtCopy28Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
#define ZExtCopy48Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
DEFINE_FASTPATH(CopyManager, ZExtCopy12);
DEFINE_FASTPATH(CopyManager, ZExtCopy14);
DEFINE_FASTPATH(CopyManager, ZExtCopy24);
DEFINE_FASTPATH(CopyManager, ZExtCopy28);
DEFINE_FASTPATH(CopyManager, ZExtCopy48);
#define SExtCopy12Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
#define SExtCopy14Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
#define SExtCopy24Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
#define SExtCopy28Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
#define SExtCopy48Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
DEFINE_FASTPATH(CopyManager, SExtCopy12);
DEFINE_FASTPATH(CopyManager, SExtCopy14);
DEFINE_FASTPATH(CopyManager, SExtCopy24);
DEFINE_FASTPATH(CopyManager, SExtCopy28);
DEFINE_FASTPATH(CopyManager, SExtCopy48);
#define TruncCopy21Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
#define TruncCopy41Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
#define TruncCopy42Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
#define TruncCopy81Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
#define TruncCopy82Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
#define TruncCopy84Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
DEFINE_FASTPATH(CopyManager, TruncCopy21);
DEFINE_FASTPATH(CopyManager, TruncCopy41);
DEFINE_FASTPATH(CopyManager, TruncCopy42);
DEFINE_FASTPATH(CopyManager, TruncCopy81);
DEFINE_FASTPATH(CopyManager, TruncCopy82);
DEFINE_FASTPATH(CopyManager, TruncCopy84);
#define CvtI32F32Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count, mulVal)
#define CvtI16F32Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count, mulVal)
#define CvtI8F32Args  BOOST_PP_VARIADIC_TO_SEQ(dest, src, count, mulVal)
#define CvtU32F32Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count, mulVal)
#define CvtU16F32Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count, mulVal)
#define CvtU8F32Args  BOOST_PP_VARIADIC_TO_SEQ(dest, src, count, mulVal)
DEFINE_FASTPATH(CopyManager, CvtI32F32);
DEFINE_FASTPATH(CopyManager, CvtI16F32);
DEFINE_FASTPATH(CopyManager, CvtI8F32 );
DEFINE_FASTPATH(CopyManager, CvtU32F32);
DEFINE_FASTPATH(CopyManager, CvtU16F32);
DEFINE_FASTPATH(CopyManager, CvtU8F32 );
#define CvtF32I32Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count, mulVal, sat)
#define CvtF32I16Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count, mulVal, sat)
#define CvtF32I8Args  BOOST_PP_VARIADIC_TO_SEQ(dest, src, count, mulVal, sat)
#define CvtF32U16Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count, mulVal, sat)
#define CvtF32U8Args  BOOST_PP_VARIADIC_TO_SEQ(dest, src, count, mulVal, sat)
DEFINE_FASTPATH(CopyManager, CvtF32I32);
DEFINE_FASTPATH(CopyManager, CvtF32I16);
DEFINE_FASTPATH(CopyManager, CvtF32I8 );
DEFINE_FASTPATH(CopyManager, CvtF32U16);
DEFINE_FASTPATH(CopyManager, CvtF32U8 );
#define CvtF32F16Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
#define CvtF16F32Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
#define CvtF32F64Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
#define CvtF64F32Args BOOST_PP_VARIADIC_TO_SEQ(dest, src, count)
DEFINE_FASTPATH(CopyManager, CvtF32F16);
DEFINE_FASTPATH(CopyManager, CvtF16F32);
DEFINE_FASTPATH(CopyManager, CvtF32F64);
DEFINE_FASTPATH(CopyManager, CvtF64F32);

namespace
{
using common::fastpath::FuncVarBase;

struct LOOP : FuncVarBase {};
struct SIMD128
{
    static bool RuntimeCheck() noexcept
    {
#if COMMON_ARCH_X86
        return CheckCPUFeature("sse2"sv);
#else
        return CheckCPUFeature("asimd"sv);
#endif
    }
};
struct SIMDSSSE3
{
    static bool RuntimeCheck() noexcept
    {
#if COMMON_ARCH_X86
        return CheckCPUFeature("ssse3"sv);
#else
        return false;
#endif
    }
};
struct SIMDSSE41
{
    static bool RuntimeCheck() noexcept
    {
#if COMMON_ARCH_X86
        return CheckCPUFeature("sse4_1"sv);
#else
        return false;
#endif
    }
};
struct SIMD256
{
    static bool RuntimeCheck() noexcept
    {
#if COMMON_ARCH_X86
        return CheckCPUFeature("avx"sv);
#else
        return false;
#endif
    }
};
struct SIMDAVX2
{
    static bool RuntimeCheck() noexcept
    {
#if COMMON_ARCH_X86
        return CheckCPUFeature("avx2"sv);
#else
        return false;
#endif
    }
};
struct F16C
{
    static bool RuntimeCheck() noexcept
    {
#if COMMON_ARCH_X86
        return CheckCPUFeature("f16c"sv);
#else
        return false;
#endif
    }
};
struct AVX512F
{
    static bool RuntimeCheck() noexcept
    {
#if COMMON_ARCH_X86
        return CheckCPUFeature("avx512f"sv);
#else
        return false;
#endif
    }
};
}

DEFINE_FASTPATH_METHOD(Broadcast2, LOOP)
{
    const uint64_t data = src * 0x0001000100010001u;
    while (count >= 4)
    {
        *reinterpret_cast<uint64_t*>(dest) = data;
        dest += 4; count -= 4;
    }
    switch (count)
    {
    case 3: *dest++ = src;
        [[fallthrough]];
    case 2: *dest++ = src;
        [[fallthrough]];
    case 1: *dest++ = src;
        [[fallthrough]];
    default: break;
    }
}
DEFINE_FASTPATH_METHOD(Broadcast4, LOOP)
{
    const uint64_t dat = src * 0x0000000100000001u;
    while (count >= 2)
    {
        *reinterpret_cast<uint64_t*>(dest) = dat;
        dest += 2; count -= 2;
    }
    if (count)
    {
        *dest++ = src;
    }
}

template<typename Src, typename Dst, bool Sat = false>
struct ScalarCast
{
    template<typename... Args>
    constexpr ScalarCast(Args&&...) noexcept {}
    Dst operator()(const Src src) const noexcept
    {
        if constexpr (Sat)
        {
            if (src >= static_cast<Src>(std::numeric_limits<Dst>::max()))
                return std::numeric_limits<Dst>::max();
            if (src <= static_cast<Src>(std::numeric_limits<Dst>::min()))
                return std::numeric_limits<Dst>::min();
        }
        return static_cast<Dst>(src);
    }
};
template<typename Src, typename Dst, bool Sat = false>
struct ScalarI2FCast
{
    Dst Muler;
    ScalarI2FCast(Dst mulVal) noexcept : Muler(mulVal) {}
    Dst operator()(const Src src) const noexcept
    {
        return static_cast<Dst>(src) * Muler;
    }
};
template<typename Src, typename Dst, bool Sat = false>
struct ScalarF2ICast
{
    Src Muler;
    ScalarF2ICast(Src mulVal) noexcept : Muler(mulVal) {}
    Dst operator()(Src src) const noexcept
    {
        src *= Muler;
        if constexpr (Sat)
        {
            if (src >= static_cast<Src>(std::numeric_limits<Dst>::max()))
                return std::numeric_limits<Dst>::max();
            if (src <= static_cast<Src>(std::numeric_limits<Dst>::min()))
                return std::numeric_limits<Dst>::min();
        }
        return static_cast<Dst>(src);
    }
};
template<template<typename, typename, bool> class Cast, bool Sat = false, typename Src, typename Dst, typename... Args>
static void CastLoop(Dst* dest, const Src* src, size_t count, Args&&... args) noexcept
{
    Cast<Src, Dst, Sat> cast(std::forward<Args>(args)...);
    for (size_t iter = count / 8; iter > 0; iter--)
    {
        dest[0] = cast(src[0]);
        dest[1] = cast(src[1]);
        dest[2] = cast(src[2]);
        dest[3] = cast(src[3]);
        dest[4] = cast(src[4]);
        dest[5] = cast(src[5]);
        dest[6] = cast(src[6]);
        dest[7] = cast(src[7]);
        dest += 8; src += 8;
    }
    switch (count % 8)
    {
    case 7: *dest++ = cast(*src++);
        [[fallthrough]];
    case 6: *dest++ = cast(*src++);
        [[fallthrough]];
    case 5: *dest++ = cast(*src++);
        [[fallthrough]];
    case 4: *dest++ = cast(*src++);
        [[fallthrough]];
    case 3: *dest++ = cast(*src++);
        [[fallthrough]];
    case 2: *dest++ = cast(*src++);
        [[fallthrough]];
    case 1: *dest++ = cast(*src++);
        break;
    case 0:
        return;
    };
}
DEFINE_FASTPATH_METHOD(ZExtCopy12, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(ZExtCopy14, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(ZExtCopy24, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(ZExtCopy28, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(ZExtCopy48, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy12, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy14, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy24, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy28, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy48, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy21, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy41, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy42, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy81, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy82, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy84, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}
#define DEFINE_CVTFP2I_LOOP(func) DEFINE_FASTPATH_METHOD(func, LOOP)    \
{                                                                       \
    if (mulVal == 0)                                                    \
        CastLoop<ScalarCast>(dest, src, count);                         \
    else                                                                \
        CastLoop<ScalarI2FCast>(dest, src, count, mulVal);              \
}
DEFINE_CVTFP2I_LOOP(CvtI32F32)
DEFINE_CVTFP2I_LOOP(CvtI16F32)
DEFINE_CVTFP2I_LOOP(CvtI8F32)
DEFINE_CVTFP2I_LOOP(CvtU32F32)
DEFINE_CVTFP2I_LOOP(CvtU16F32)
DEFINE_CVTFP2I_LOOP(CvtU8F32)
#define DEFINE_CVTI2FP_LOOP(func) DEFINE_FASTPATH_METHOD(func, LOOP)    \
{                                                                       \
    if (mulVal == 0)                                                    \
    {                                                                   \
        if (sat)                                                        \
            CastLoop<ScalarCast, true>(dest, src, count);               \
        else                                                            \
            CastLoop<ScalarCast>(dest, src, count);                     \
    }                                                                   \
    else                                                                \
    {                                                                   \
        if (sat)                                                        \
            CastLoop<ScalarF2ICast, true>(dest, src, count, mulVal);    \
        else                                                            \
            CastLoop<ScalarF2ICast>(dest, src, count, mulVal);          \
    }                                                                   \
}
DEFINE_CVTI2FP_LOOP(CvtF32I32)
DEFINE_CVTI2FP_LOOP(CvtF32I16)
DEFINE_CVTI2FP_LOOP(CvtF32I8)
DEFINE_CVTI2FP_LOOP(CvtF32U16)
DEFINE_CVTI2FP_LOOP(CvtF32U8)
DEFINE_FASTPATH_METHOD(CvtF16F32, LOOP)
{
    const auto src_ = reinterpret_cast<const half_float::half*>(src);
    CastLoop<ScalarCast>(dest, src_, count);
}
DEFINE_FASTPATH_METHOD(CvtF32F16, LOOP)
{
    const auto dst = reinterpret_cast<half_float::half*>(dest);
    CastLoop<ScalarCast>(dst, src, count);
}
DEFINE_FASTPATH_METHOD(CvtF32F64, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(CvtF64F32, LOOP)
{
    CastLoop<ScalarCast>(dest, src, count);
}

template<typename T, auto F, typename Src, typename Dst>
forceinline void BroadcastSIMD4(Dst* dest, const Src src, size_t count) noexcept
{
    constexpr auto N = T::Count;
    T data(src);
    for (auto iter = count / (N * 4); iter > 0; --iter)
    {
        data.Save(dest + N * 0);
        data.Save(dest + N * 1);
        data.Save(dest + N * 2);
        data.Save(dest + N * 3);
        dest += N * 4;
    }
    count = count % (N * 4);
    switch (count / N)
    {
    case 3: data.Save(dest); dest += N;
        [[fallthrough]];
    case 2: data.Save(dest); dest += N;
        [[fallthrough]];
    case 1: data.Save(dest); dest += N;
        [[fallthrough]];
    default: break;
    }
    count = count % N;
    F(dest, src, count);
}

template<typename TIn, typename TCast, typename T, CastMode Mode>
struct LoadCastSave
{
    using Src = std::decay_t<decltype(std::declval<TIn>().Val[0])>;
    using Dst = std::decay_t<decltype(std::declval<TCast>().Val[0])>;
    static constexpr size_t N = TIn::Count, M = TCast::Count;
    static constexpr size_t K = (N >= M) ? (N / M) : (M / N - 1);
    static constexpr auto Idxes = std::make_index_sequence<K>{};
    void operator()(Dst* dst, const Src* src) const noexcept
    {
        Proc(dst, src, Idxes);
    }
    template<size_t... I>
    void Proc(Dst* dst, const Src* src, std::index_sequence<I...>) const noexcept
    {
        const T& self = *static_cast<const T*>(this);
        if constexpr (N == M)
        {
            const auto val = self.Pre(TIn(src));
            self.Post(val.template Cast<TCast, Mode>()).Save(dst);
        }
        else if constexpr (N < M)
        {
            const TIn val[] = { self.Pre(TIn(src)), self.Pre(TIn(src + N + N * I))... };
            self.Post(val[0].template Cast<TCast, Mode>(val[1 + I]...)).Save(dst);
        }
        else
        {
            const auto mid = self.Pre(TIn(src)).template Cast<TCast, Mode>();
            (..., self.Post(mid[I]).Save(dst + M * I));
        }
    }
};
template<typename TIn, typename TCast, CastMode Mode = detail::CstMode<TIn, TCast>()>
struct DefaultCast : public LoadCastSave<TIn, TCast, DefaultCast<TIn, TCast, Mode>, Mode>
{
    template<typename... Args>
    constexpr DefaultCast(Args&&...) noexcept {}
    TIn Pre(TIn val) const noexcept { return val; }
    TCast Post(TCast val) const noexcept { return val; }
};
template<typename TIn, typename TCast, CastMode Mode = detail::CstMode<TIn, TCast>()>
struct I2FCast : public LoadCastSave<TIn, TCast, I2FCast<TIn, TCast, Mode>, Mode>
{
    TCast Muler;
    template<typename... Args>
    I2FCast(TCast mulVal, Args&&...) noexcept : Muler(mulVal) {}
    TIn Pre(TIn val) const noexcept { return val; }
    TCast Post(TCast val) const noexcept { return val.Mul(Muler); }
};
template<typename TIn, typename TCast, CastMode Mode = detail::CstMode<TIn, TCast>()>
struct F2ICast : public LoadCastSave<TIn, TCast, F2ICast<TIn, TCast, Mode>, Mode>
{
    TIn Muler;
    template<typename... Args>
    F2ICast(TIn mulVal, Args&&...) noexcept : Muler(mulVal) {}
    TIn Pre(TIn val) const noexcept { return val.Mul(Muler); }
    TCast Post(TCast val) const noexcept { return val; }
};
template<typename Cast, auto F, typename Src, typename Dst, typename... Args>
forceinline void CastSIMD4(Dst* dest, const Src* src, size_t count, Args&&... args) noexcept
{
    Cast cast(std::forward<Args>(args)...);
    constexpr auto K = std::max(Cast::N, Cast::M);
    for (size_t iter = count / (K * 4); iter > 0; iter--)
    {
        cast(dest + K * 0, src + K * 0);
        cast(dest + K * 1, src + K * 1);
        cast(dest + K * 2, src + K * 2);
        cast(dest + K * 3, src + K * 3);
        src += K * 4; dest += K * 4;
    }
    count = count % (K * 4);
    switch (count / K)
    {
    case 3: cast(dest, src); src += K; dest += K;
          [[fallthrough]];
    case 2: cast(dest, src); src += K; dest += K;
          [[fallthrough]];
    case 1: cast(dest, src); src += K; dest += K;
          [[fallthrough]];
    default: break;
    }
    count = count % K;
    F(dest, src, count, std::forward<Args>(args)...);
}

#define DEFINE_CVTI2FP_SIMD4(func, from, to, algo, prev)                        \
DEFINE_FASTPATH_METHOD(func, algo)                                              \
{                                                                               \
    if (mulVal == 0)                                                            \
        CastSIMD4<DefaultCast<from, to>, &Func<prev>>(dest, src, count, mulVal);\
    else                                                                        \
        CastSIMD4<I2FCast<from, to>, &Func<prev>>(dest, src, count, mulVal);    \
}
#define DEFINE_CVTFP2I_SIMD4(func, from, to, algo, prev)                \
DEFINE_FASTPATH_METHOD(func, algo)                                      \
{                                                                       \
    if (mulVal == 0)                                                    \
    {                                                                   \
        if (sat)                                                        \
            CastSIMD4<DefaultCast<from, to, CastMode::RangeSaturate>,   \
                &Func<prev>>(dest, src, count, mulVal, sat);            \
        else                                                            \
            CastSIMD4<DefaultCast<from, to>,                            \
                &Func<prev>>(dest, src, count, mulVal, sat);            \
    }                                                                   \
    else                                                                \
    {                                                                   \
        if (sat)                                                        \
            CastSIMD4<F2ICast<from, to, CastMode::RangeSaturate>,       \
                &Func<prev>>(dest, src, count, mulVal, sat);            \
        else                                                            \
            CastSIMD4<F2ICast<from, to>,                                \
                &Func<prev>>(dest, src, count, mulVal, sat);            \
    }                                                                   \
}

#if (COMMON_ARCH_X86 && COMMON_SIMD_LV >= 20) || (COMMON_ARCH_ARM && COMMON_SIMD_LV >= 10)

DEFINE_FASTPATH_METHOD(Broadcast2, SIMD128)
{
    BroadcastSIMD4<U16x8, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(Broadcast4, SIMD128)
{
    BroadcastSIMD4<U32x4, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(ZExtCopy12, SIMD128)
{
    CastSIMD4<DefaultCast<U8x16, U16x8>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(ZExtCopy24, SIMD128)
{
    CastSIMD4<DefaultCast<U16x8, U32x4>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(ZExtCopy28, SIMD128)
{
    CastSIMD4<DefaultCast<U16x8, U64x2>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(ZExtCopy48, SIMD128)
{
    CastSIMD4<DefaultCast<U32x4, U64x2>, &Func<LOOP>>(dest, src, count);
}
DEFINE_CVTI2FP_SIMD4(CvtI32F32, I32x4, F32x4, SIMD128, LOOP)
DEFINE_CVTI2FP_SIMD4(CvtI16F32, I16x8, F32x4, SIMD128, LOOP)
DEFINE_CVTI2FP_SIMD4(CvtI8F32,  I8x16, F32x4, SIMD128, LOOP)
DEFINE_CVTI2FP_SIMD4(CvtU32F32, U32x4, F32x4, SIMD128, LOOP)
DEFINE_CVTI2FP_SIMD4(CvtU16F32, U16x8, F32x4, SIMD128, LOOP)
DEFINE_CVTI2FP_SIMD4(CvtU8F32,  U8x16, F32x4, SIMD128, LOOP)

DEFINE_CVTFP2I_SIMD4(CvtF32I32, F32x4, I32x4, SIMD128, LOOP)
DEFINE_CVTFP2I_SIMD4(CvtF32I16, F32x4, I16x8, SIMD128, LOOP)
DEFINE_CVTFP2I_SIMD4(CvtF32I8,  F32x4, I8x16, SIMD128, LOOP)
DEFINE_CVTFP2I_SIMD4(CvtF32U16, F32x4, U16x8, SIMD128, LOOP)
DEFINE_CVTFP2I_SIMD4(CvtF32U8,  F32x4, U8x16, SIMD128, LOOP)

#endif

#if (COMMON_ARCH_X86 && COMMON_SIMD_LV >= 20) || (COMMON_ARCH_ARM && COMMON_SIMD_LV >= 200)
DEFINE_FASTPATH_METHOD(CvtF32F64, SIMD128)
{
    CastSIMD4<DefaultCast<F32x4, F64x2>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(CvtF64F32, SIMD128)
{
    CastSIMD4<DefaultCast<F64x2, F32x4>, &Func<LOOP>>(dest, src, count);
}
#endif

#if COMMON_ARCH_ARM && COMMON_SIMD_LV >= 10
DEFINE_FASTPATH_METHOD(ZExtCopy14, SIMD128)
{
    CastSIMD4<DefaultCast<U8x16, U32x4>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy12, SIMD128)
{
    CastSIMD4<DefaultCast<I8x16, I16x8>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy14, SIMD128)
{
    CastSIMD4<DefaultCast<I8x16, I32x4>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy24, SIMD128)
{
    CastSIMD4<DefaultCast<I16x8, I32x4>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy28, SIMD128)
{
    CastSIMD4<DefaultCast<I16x8, I64x2>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy48, SIMD128)
{
    CastSIMD4<DefaultCast<I32x4, I64x2>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy21, SIMD128)
{
    CastSIMD4<DefaultCast<U16x8, U8x16>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy41, SIMD128)
{
    CastSIMD4<DefaultCast<U32x4, U8x16>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy42, SIMD128)
{
    CastSIMD4<DefaultCast<U32x4, U16x8>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy81, SIMD128)
{
    CastSIMD4<DefaultCast<U64x2, U8x16>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy82, SIMD128)
{
    CastSIMD4<DefaultCast<U64x2, U16x8>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy84, SIMD128)
{
    CastSIMD4<DefaultCast<U64x2, U32x4>, &Func<LOOP>>(dest, src, count);
}
#endif

#if COMMON_ARCH_X86 && COMMON_SIMD_LV >= 31
DEFINE_FASTPATH_METHOD(ZExtCopy14, SIMDSSSE3)
{
    CastSIMD4<DefaultCast<U8x16, U32x4>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy21, SIMDSSSE3)
{
    CastSIMD4<DefaultCast<U16x8, U8x16>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy41, SIMDSSSE3)
{
    CastSIMD4<DefaultCast<U32x4, U8x16>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy42, SIMDSSSE3)
{
    CastSIMD4<DefaultCast<U32x4, U16x8>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy81, SIMDSSSE3)
{
    CastSIMD4<DefaultCast<U64x2, U8x16>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy82, SIMDSSSE3)
{
    CastSIMD4<DefaultCast<U64x2, U16x8>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy84, SIMDSSSE3)
{
    CastSIMD4<DefaultCast<U64x2, U32x4>, &Func<LOOP>>(dest, src, count);
}
#endif

#if COMMON_ARCH_X86 && COMMON_SIMD_LV >= 41
DEFINE_FASTPATH_METHOD(SExtCopy12, SIMDSSE41)
{
    CastSIMD4<DefaultCast<I8x16, I16x8>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy14, SIMDSSE41)
{
    CastSIMD4<DefaultCast<I8x16, I32x4>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy24, SIMDSSE41)
{
    CastSIMD4<DefaultCast<I16x8, I32x4>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy28, SIMDSSE41)
{
    CastSIMD4<DefaultCast<I16x8, I64x2>, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy48, SIMDSSE41)
{
    CastSIMD4<DefaultCast<I32x4, I64x2>, &Func<LOOP>>(dest, src, count);
}
#endif

#if COMMON_ARCH_X86 && COMMON_SIMD_LV >= 100
DEFINE_FASTPATH_METHOD(Broadcast2, SIMD256)
{
    BroadcastSIMD4<U16x16, &Func<SIMD128>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(Broadcast4, SIMD256)
{
    BroadcastSIMD4<U32x8, &Func<SIMD128>>(dest, src, count);
}

DEFINE_CVTI2FP_SIMD4(CvtI32F32, I32x8, F32x8, SIMD256, SIMD128)

DEFINE_CVTFP2I_SIMD4(CvtF32I32, F32x8, I32x8, SIMD256, SIMD128)

DEFINE_FASTPATH_METHOD(CvtF32F64, SIMD256)
{
    CastSIMD4<DefaultCast<F32x8, F64x4>, &Func<SIMD128>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(CvtF64F32, SIMD256)
{
    CastSIMD4<DefaultCast<F64x4, F32x8>, &Func<SIMD128>>(dest, src, count);
}
#endif

#if COMMON_ARCH_X86 && COMMON_SIMD_LV >= 200
DEFINE_FASTPATH_METHOD(ZExtCopy12, SIMDAVX2)
{
    CastSIMD4<DefaultCast<U8x32, U16x16>, &Func<SIMD128>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(ZExtCopy14, SIMDAVX2)
{
    CastSIMD4<DefaultCast<U8x32, U32x8>, &Func<SIMDSSSE3>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(ZExtCopy24, SIMDAVX2)
{
    CastSIMD4<DefaultCast<U16x16, U32x8>, &Func<SIMD128>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(ZExtCopy28, SIMDAVX2)
{
    CastSIMD4<DefaultCast<U16x16, U64x4>, &Func<SIMD128>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(ZExtCopy48, SIMDAVX2)
{
    CastSIMD4<DefaultCast<U32x8, U64x4>, &Func<SIMD128>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy12, SIMDAVX2)
{
    CastSIMD4<DefaultCast<I8x32, I16x16>, &Func<SIMDSSE41>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy14, SIMDAVX2)
{
    CastSIMD4<DefaultCast<I8x32, I32x8>, &Func<SIMDSSE41>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy24, SIMDAVX2)
{
    CastSIMD4<DefaultCast<I16x16, I32x8>, &Func<SIMDSSE41>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy28, SIMDAVX2)
{
    CastSIMD4<DefaultCast<I16x16, I64x4>, &Func<SIMDSSE41>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(SExtCopy48, SIMDAVX2)
{
    CastSIMD4<DefaultCast<I32x8, I64x4>, &Func<SIMDSSE41>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy21, SIMDAVX2)
{
    CastSIMD4<DefaultCast<U16x16, U8x32>, &Func<SIMDSSSE3>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy41, SIMDAVX2)
{
    CastSIMD4<DefaultCast<U32x8, U8x32>, &Func<SIMDSSSE3>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy42, SIMDAVX2)
{
    CastSIMD4<DefaultCast<U32x8, U16x16>, &Func<SIMDSSSE3>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy81, SIMDAVX2)
{
    CastSIMD4<DefaultCast<U64x4, U8x32>, &Func<SIMDSSSE3>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy82, SIMDAVX2)
{
    CastSIMD4<DefaultCast<U64x4, U16x16>, &Func<SIMDSSSE3>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(TruncCopy84, SIMDAVX2)
{
    CastSIMD4<DefaultCast<U64x4, U32x8>, &Func<SIMDSSSE3>>(dest, src, count);
}

DEFINE_CVTI2FP_SIMD4(CvtI16F32, I16x16, F32x8, SIMDAVX2, SIMD128)
DEFINE_CVTI2FP_SIMD4(CvtI8F32,  I8x32,  F32x8, SIMDAVX2, SIMD128)
DEFINE_CVTI2FP_SIMD4(CvtU32F32, U32x8,  F32x8, SIMDAVX2, SIMD128)
DEFINE_CVTI2FP_SIMD4(CvtU16F32, U16x16, F32x8, SIMDAVX2, SIMD128)
DEFINE_CVTI2FP_SIMD4(CvtU8F32,  U8x32,  F32x8, SIMDAVX2, SIMD128)

DEFINE_CVTFP2I_SIMD4(CvtF32I16, F32x8, I16x16, SIMDAVX2, SIMD128)
DEFINE_CVTFP2I_SIMD4(CvtF32I8,  F32x8, I8x32,  SIMDAVX2, SIMD128)
DEFINE_CVTFP2I_SIMD4(CvtF32U16, F32x8, U16x16, SIMDAVX2, SIMD128)
DEFINE_CVTFP2I_SIMD4(CvtF32U8,  F32x8, U8x32,  SIMDAVX2, SIMD128)
#endif

#if COMMON_ARCH_X86 && COMMON_SIMD_LV >= 100 && (defined(__F16C__) || COMMON_COMPILER_MSVC)
struct F1632Cast
{
    using Src = uint16_t;
    using Dst = float;
    static constexpr size_t N = 8, M = 8;
    void operator()(float* dst, const uint16_t* src) const noexcept
    {
        F32x8(_mm256_cvtph_ps(U16x8(src))).Save(dst);
    }
};
struct F3216Cast
{
    static constexpr size_t N = 8, M = 8;
    void operator()(uint16_t* dst, const float* src) const noexcept
    {
        U16x8(_mm256_cvtps_ph(F32x8(src), _MM_FROUND_TO_NEAREST_INT | _MM_FROUND_NO_EXC)).Save(dst);
    }
};
DEFINE_FASTPATH_METHOD(CvtF16F32, F16C)
{
    CastSIMD4<F1632Cast, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(CvtF32F16, F16C)
{
    CastSIMD4<F3216Cast, &Func<LOOP>>(dest, src, count);
}
#endif

#if COMMON_ARCH_ARM && COMMON_SIMD_LV >= 10 && (__ARM_FP & 2)
struct F1632Cast1
{
    using Src = uint16_t;
    using Dst = float;
    static constexpr size_t N = 8, M = 8;
    void operator()(float* dst, const uint16_t* src) const noexcept
    {
# if COMMON_SIMD_LV >= 200
#   if COMMON_COMPILER_MSVC
        const auto in = vld1q_u16(src); // msvc only typedef __n128 & __n64
#   else
        const auto in = vld1q_f16(reinterpret_cast<const float16_t*>(src));
#   endif
        F32x4(vcvt_f32_f16(vget_low_f16(in))).Save(dst + 0);
        F32x4(vcvt_high_f32_f16(in))         .Save(dst + 4);
# else
#   if COMMON_COMPILER_MSVC
        const auto in = vld1_u16_x2(src); // msvc only typedef __n128 & __n64
#   else
        const auto in = vld1_f16_x2(reinterpret_cast<const float16_t*>(src));
#   endif
        F32x4(vcvt_f32_f16(in.val[0])).Save(dst + 0);
        F32x4(vcvt_f32_f16(in.val[1])).Save(dst + 4);
# endif
    }
};
struct F3216Cast1
{
    static constexpr size_t N = 8, M = 8;
    void operator()(uint16_t* dst, const float* src) const noexcept
    {
        const auto dst_ = reinterpret_cast<float16_t*>(dst);
        const auto in = vld1q_f32_x2(src);
# if COMMON_SIMD_LV >= 200
        const auto out = vcvt_high_f16_f32(vcvt_f16_f32(in.val[0]), in.val[1]);
#   if COMMON_COMPILER_MSVC
        vst1q_u16(dst_, out); // msvc only typedef __n128 & __n64
#   else
        vst1q_f16(dst_, out);
#   endif
# else
        vst1_f16(dst_ + 0, vcvt_f16_f32(in.val[0]));
        vst1_f16(dst_ + 4, vcvt_f16_f32(in.val[1]));
# endif
    }
};
DEFINE_FASTPATH_METHOD(CvtF16F32, SIMD128)
{
    CastSIMD4<F1632Cast1, &Func<LOOP>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(CvtF32F16, SIMD128)
{
    CastSIMD4<F3216Cast1, &Func<LOOP>>(dest, src, count);
}
#endif

#if COMMON_ARCH_X86 && (COMMON_SIMD_LV >= 310 || (COMMON_COMPILER_MSVC && COMMON_MSVC_VER >= 191000))
struct F1632CastAVX512
{
    using Src = uint16_t;
    using Dst = float;
    static constexpr size_t N = 16, M = 16;
    void operator()(float* dst, const uint16_t* src) const noexcept
    {
        _mm512_storeu_ps(dst, _mm512_cvtph_ps(U16x16(src)));
    }
};
struct F3216CastAVX512
{
    static constexpr size_t N = 16, M = 16;
    void operator()(uint16_t* dst, const float* src) const noexcept
    {
        U16x16(_mm512_cvtps_ph(_mm512_loadu_ps(src), _MM_FROUND_TO_NEAREST_INT | _MM_FROUND_NO_EXC)).Save(dst);
    }
};
DEFINE_FASTPATH_METHOD(CvtF16F32, AVX512F)
{
    CastSIMD4<F1632CastAVX512, &Func<F16C>>(dest, src, count);
}
DEFINE_FASTPATH_METHOD(CvtF32F16, AVX512F)
{
    CastSIMD4<F3216CastAVX512, &Func<F16C>>(dest, src, count);
}
#endif

namespace common
{

common::span<const CopyManager::PathInfo> CopyManager::GetSupportMap() noexcept
{
    static auto list = []()
    {
        std::vector<CopyManager::PathInfo> ret;
        RegistFuncVars(CopyManager, Broadcast2, SIMD256, SIMD128, LOOP);
        RegistFuncVars(CopyManager, Broadcast4, SIMD256, SIMD128, LOOP);
        RegistFuncVars(CopyManager, ZExtCopy12, SIMDAVX2, SIMD128, LOOP);
        RegistFuncVars(CopyManager, ZExtCopy14, SIMDAVX2, SIMDSSSE3, SIMD128, LOOP);
        RegistFuncVars(CopyManager, ZExtCopy24, SIMDAVX2, SIMD128, LOOP);
        RegistFuncVars(CopyManager, ZExtCopy28, SIMDAVX2, SIMD128, LOOP);
        RegistFuncVars(CopyManager, ZExtCopy48, SIMDAVX2, SIMD128, LOOP);
        RegistFuncVars(CopyManager, SExtCopy12, SIMDAVX2, SIMDSSE41, SIMD128, LOOP);
        RegistFuncVars(CopyManager, SExtCopy14, SIMDAVX2, SIMDSSE41, SIMD128, LOOP);
        RegistFuncVars(CopyManager, SExtCopy24, SIMDAVX2, SIMDSSE41, SIMD128, LOOP);
        RegistFuncVars(CopyManager, SExtCopy28, SIMDAVX2, SIMDSSE41, SIMD128, LOOP);
        RegistFuncVars(CopyManager, SExtCopy48, SIMDAVX2, SIMDSSE41, SIMD128, LOOP);
        RegistFuncVars(CopyManager, TruncCopy21, SIMDAVX2, SIMDSSSE3, SIMD128, LOOP);
        RegistFuncVars(CopyManager, TruncCopy41, SIMDAVX2, SIMDSSSE3, SIMD128, LOOP);
        RegistFuncVars(CopyManager, TruncCopy42, SIMDAVX2, SIMDSSSE3, SIMD128, LOOP);
        RegistFuncVars(CopyManager, TruncCopy81, SIMDAVX2, SIMDSSSE3, SIMD128, LOOP);
        RegistFuncVars(CopyManager, TruncCopy82, SIMDAVX2, SIMDSSSE3, SIMD128, LOOP);
        RegistFuncVars(CopyManager, TruncCopy84, SIMDAVX2, SIMDSSSE3, SIMD128, LOOP);
        RegistFuncVars(CopyManager, CvtI32F32, SIMD256,  SIMD128, LOOP);
        RegistFuncVars(CopyManager, CvtI16F32, SIMDAVX2, SIMD128, LOOP);
        RegistFuncVars(CopyManager, CvtI8F32,  SIMDAVX2, SIMD128, LOOP);
        RegistFuncVars(CopyManager, CvtU32F32, SIMDAVX2, SIMD128, LOOP);
        RegistFuncVars(CopyManager, CvtU16F32, SIMDAVX2, SIMD128, LOOP);
        RegistFuncVars(CopyManager, CvtU8F32,  SIMDAVX2, SIMD128, LOOP);
        RegistFuncVars(CopyManager, CvtF32I32, SIMD256,  SIMD128, LOOP);
        RegistFuncVars(CopyManager, CvtF32I16, SIMDAVX2, SIMD128, LOOP);
        RegistFuncVars(CopyManager, CvtF32I8,  SIMDAVX2, SIMD128, LOOP);
        RegistFuncVars(CopyManager, CvtF32U16, SIMDAVX2, SIMD128, LOOP);
        RegistFuncVars(CopyManager, CvtF32U8,  SIMDAVX2, SIMD128, LOOP);
        RegistFuncVars(CopyManager, CvtF16F32, AVX512F, F16C, SIMD128, LOOP);
        RegistFuncVars(CopyManager, CvtF32F16, AVX512F, F16C, SIMD128, LOOP);
        RegistFuncVars(CopyManager, CvtF32F64, SIMD256, SIMD128, LOOP);
        RegistFuncVars(CopyManager, CvtF64F32, SIMD256, SIMD128, LOOP);
        return ret;
    }();
    return list;
}
CopyManager::CopyManager(common::span<const VarItem> requests) noexcept { Init(requests); }
CopyManager::~CopyManager() {}
bool CopyManager::IsComplete() const noexcept
{
    return Broadcast2 && Broadcast4 &&
        ZExtCopy12 && ZExtCopy14 && ZExtCopy24 && ZExtCopy28 && ZExtCopy48 &&
        SExtCopy12 && SExtCopy14 && SExtCopy24 && SExtCopy28 && SExtCopy48 &&
        TruncCopy21 && TruncCopy41 && TruncCopy42 && TruncCopy82 && TruncCopy84 &&
        CvtI32F32 && CvtI16F32 && CvtI8F32 && CvtU32F32 && CvtU16F32 && CvtU8F32 &&
        CvtF32I32 && CvtF32I16 && CvtF32I8 && CvtF32U16 && CvtF32U8 &&
        CvtF16F32 && CvtF32F16 && CvtF32F64 && CvtF64F32;
}
const CopyManager CopyEx;

}


text:
#include <boost/test/unit_test.hpp>
#include <boost/test/data/monomorphic.hpp>
#include <boost/test/data/test_case.hpp>
#include "pch.h"
#include "fixtures.h"

using namespace boost::unit_test;

BOOST_AUTO_TEST_SUITE(test_shortcut)

BOOST_TEST_DECORATOR(*fixture<fixture_tmp_dir>())
BOOST_TEST_DECORATOR(*fixture<fixture_com>())
BOOST_DATA_TEST_CASE(test_create_shortcut, data::make({ L"bar", L"" }), icon_path) {
	create_shortcut(L"foo", L"test.lnk", icon_path);
	IShellLink *psl;
	BOOST_TEST_REQUIRE(SUCCEEDED(CoCreateInstance(CLSID_ShellLink, nullptr, CLSCTX_INPROC_SERVER, IID_PPV_ARGS(&psl))));
	IPersistFile *ppf;
	BOOST_TEST_REQUIRE(SUCCEEDED(psl->QueryInterface(IID_PPV_ARGS(&ppf))));
	BOOST_TEST_REQUIRE(SUCCEEDED(ppf->Load(L"test.lnk", 0)));
	const auto buf = std::make_unique<wchar_t[]>(INFOTIPSIZE);
	BOOST_TEST_REQUIRE(SUCCEEDED(psl->GetPath(buf.get(), INFOTIPSIZE, nullptr, SLGP_RAWPATH)));
	wchar_t mod_path[MAX_PATH];
	BOOST_TEST_REQUIRE(GetModuleFileName(nullptr, mod_path, MAX_PATH));
	BOOST_TEST(_wcsicmp(buf.get(), mod_path) == 0);
	BOOST_TEST_REQUIRE(SUCCEEDED(psl->GetDescription(buf.get(), INFOTIPSIZE)));
	BOOST_TEST(buf.get() == L"Launch the WSL distribution foo.");
	BOOST_TEST_REQUIRE(SUCCEEDED(psl->GetArguments(buf.get(), INFOTIPSIZE)));
	BOOST_TEST(buf.get() == L"run -w -n \"foo\"");
	int idx;
	BOOST_TEST_REQUIRE(SUCCEEDED(psl->GetIconLocation(buf.get(), INFOTIPSIZE, &idx)));
	if (icon_path[0] == 0) {
		BOOST_TEST(buf.get()[0] == 0);
	} else {
		BOOST_TEST(_wcsicmp(buf.get(), (std::filesystem::current_path() / icon_path).wstring().c_str()) == 0);
	}
	BOOST_TEST(idx == 0);
}

BOOST_TEST_DECORATOR(*fixture<fixture_tmp_dir>())
BOOST_AUTO_TEST_CASE(test_create_shortcut_unintialized) {
	create_shortcut(L"foo", L"test.lnk", L"bar");
}

BOOST_AUTO_TEST_SUITE_END()

ext: cpp


text:
#ifndef SHIFT_LIVEDEBUG_DEBUG_SERVER_HPP
#define SHIFT_LIVEDEBUG_DEBUG_SERVER_HPP

#include <vector>
#include <thread>
#include <shared_mutex>
#include <filesystem>
#include <shift/core/boost_disable_warnings.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/beast/core.hpp>
#include <boost/beast/http.hpp>
#include <shift/core/boost_restore_warnings.hpp>
#include <shift/core/singleton.hpp>
#include "shift/livedebug/error_code.hpp"
#include "shift/livedebug/request_handler.hpp"

namespace shift::livedebug
{
class listener;

///
class debug_server final
: public core::singleton<debug_server, core::create::using_new>
{
public:
  /// Constructor.
  debug_server(std::size_t worker_thread_count,
               std::filesystem::path&& root_path);

  /// Destructor.
  ~debug_server();

  /// Runs the I/O context on the requested number of threads.
  void start();

  /// Closes all listening ports, stops the I/O context and ends all worker
  /// threads.
  void stop();

  /// Opens a new listening socket that accepts connections.
  /// @pre
  ///   The endpoint must not already be used, neither from prior calls to
  ///   listen nor by other applications.
  [[nodiscard]] std::error_code listen(
    const boost::asio::ip::tcp::endpoint& endpoint);

  /// Closes a listening socket on the specified endpoint.
  /// @pre
  ///   The server must listen on the specified endpoint.
  void close(const boost::asio::ip::tcp::endpoint& endpoint);

  /// ToDo: Add URL filter parameter.
  void add_request_handler(std::unique_ptr<request_handler>&& new_handler);

  /// This function produces an HTTP response for the given request.
  void handle_request(
    boost::beast::http::request<boost::beast::http::string_body>&& request,
    livedebug::session& session);

private:
  /// ToDo: Hide implementation details.
  boost::asio::io_context _io_context;
  std::vector<std::thread> _workers;
  std::vector<std::shared_ptr<listener>> _listeners;

  std::shared_mutex _request_handler_mutex;
  std::vector<std::unique_ptr<request_handler>> _request_handlers;
};
}

#endif

ext: hpp

max_stars_repo_name: cspanier/shift

max_issues_repo_name: cspanier/shift

max_forks_repo_name: cspanier/shift


text:
/*
 * Copyright (C) 2018 Codership Oy <info@codership.com>
 *
 * This file is part of wsrep-lib.
 *
 * Wsrep-lib is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 2 of the License, or
 * (at your option) any later version.
 *
 * Wsrep-lib is distributed in the hope that 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 wsrep-lib.  If not, see <https://www.gnu.org/licenses/>.
 */

#include "db_simulator.hpp"
#include "db_client.hpp"
#include "db_threads.hpp"
#include "db_tls.hpp"

#include "wsrep/logger.hpp"

#include <boost/filesystem.hpp>
#include <sstream>

static db::ti thread_instrumentation;
static db::tls tls_service;

void db::simulator::run()
{
    start();
    stop();
    std::flush(std::cerr);
    std::cout << "Results:\n";
    std::cout << stats() << std::endl;
    std::cout << db::ti::stats() << std::endl;
    std::cout << db::tls::stats() << std::endl;
}

void db::simulator::sst(db::server& server,
                        const std::string& request,
                        const wsrep::gtid& gtid,
                        bool bypass)
{
    // The request may contain extra trailing '\0' after it goes
    // through the provider, strip it first.
    std::string name(request);
    name.erase(std::find(name.begin(), name.end(), '\0'), name.end());

    wsrep::unique_lock<wsrep::mutex> lock(mutex_);
    auto i(servers_.find(name));
    wsrep::log_info() << "SST request '" << name << "'";
    if (i == servers_.end())
    {
        wsrep::log_error() << "Server " << request << " not found";
        wsrep::log_info() << "servers:";
        for (const auto& s : servers_)
        {
            wsrep::log_info() << "server: " << s.first;
        }
        throw wsrep::runtime_error("Server " + request + " not found");
    }
    if (bypass == false)
    {
        wsrep::log_info() << "SST "
                          << server.server_state().name()
                          << " -> " << request;
        i->second->storage_engine().store_position(gtid);
        i->second->storage_engine().store_view(
            server.storage_engine().get_view());
    }

    db::client dummy(*(i->second), wsrep::client_id(-1),
                     wsrep::client_state::m_local, params());

    i->second->server_state().sst_received(dummy.client_service(), 0);
    server.server_state().sst_sent(gtid, 0);
}

std::string db::simulator::stats() const
{
    auto duration(std::chrono::duration<double>(
                      clients_stop_ - clients_start_).count());
    long long transactions(stats_.commits + stats_.rollbacks);
    long long bf_aborts(0);
    for (const auto& s : servers_)
    {
        bf_aborts += s.second->storage_engine().bf_aborts();
    }
    std::ostringstream os;
    os << "Number of transactions: " << transactions
       << "\n"
       << "Seconds: " << duration
       << " \n"
       << "Transactions per second: " << double(transactions)/double(duration)
       << "\n"
       << "BF aborts: "
       << bf_aborts
       << "\n"
       << "Client commits: " << stats_.commits
       << "\n"
       << "Client rollbacks: " << stats_.rollbacks
       << "\n"
       << "Client replays: " << stats_.replays;
    return os.str();
}

////////////////////////////////////////////////////////////////////////////////
//                              Private                                       //
////////////////////////////////////////////////////////////////////////////////

void db::simulator::start()
{
    thread_instrumentation.level(params_.thread_instrumentation);
    thread_instrumentation.cond_checks(params_.cond_checks);
    tls_service.init(params_.tls_service);
    wsrep::log_info() << "Provider: " << params_.wsrep_provider;

    std::string cluster_address(build_cluster_address());
    wsrep::log_info() << "Cluster address: " << cluster_address;
    for (size_t i(0); i < params_.n_servers; ++i)
    {
        std::ostringstream name_os;
        name_os << (i + 1);
        std::ostringstream id_os;
        id_os << (i + 1);
        std::ostringstream address_os;
        address_os << "127.0.0.1:" << server_port(i);
        wsrep::id server_id(id_os.str());
        auto it(servers_.insert(
                    std::make_pair(
                        name_os.str(),
                        std::make_unique<db::server>(
                            *this,
                            name_os.str(),
                            address_os.str()))));
        if (it.second == false)
        {
            throw wsrep::runtime_error("Failed to add server");
        }
        boost::filesystem::path dir("dbsim_" + id_os.str() + "_data");
        boost::filesystem::create_directory(dir);

        db::server& server(*it.first->second);
        server.server_state().debug_log_level(params_.debug_log_level);
        std::string server_options(params_.wsrep_provider_options);

        wsrep::provider::services services;
        services.thread_service = params_.thread_instrumentation
                                      ? &thread_instrumentation
                                      : nullptr;
        services.tls_service = params_.tls_service
            ? &tls_service
            : nullptr;
        if (server.server_state().load_provider(params_.wsrep_provider,
                                                server_options, services))
        {
            throw wsrep::runtime_error("Failed to load provider");
        }
        if (server.server_state().connect("sim_cluster", cluster_address, "",
                                          i == 0))
        {
            throw wsrep::runtime_error("Failed to connect");
        }
        wsrep::log_debug() << "main: Starting applier";
        server.start_applier();
        wsrep::log_debug() << "main: Waiting initializing state";
        server.server_state().wait_until_state(wsrep::server_state::s_initializing);
        wsrep::log_debug() << "main: Calling initialized";
        server.server_state().initialized();
        wsrep::log_debug() << "main: Waiting for synced state";
        server.server_state().wait_until_state(
            wsrep::server_state::s_synced);
        wsrep::log_debug() << "main: Server synced";
    }

    // Start client threads
    wsrep::log_info() << "####################### Starting client load";
    clients_start_ = std::chrono::steady_clock::now();
    size_t index(0);
    for (auto& i : servers_)
    {
        if (params_.topology.size() == 0 || params_.topology[index]  == 'm')
        {
            i.second->start_clients();
        }
        ++index;
    }
}

void db::simulator::stop()
{
    for (auto& i : servers_)
    {
        db::server& server(*i.second);
        server.stop_clients();
    }
    clients_stop_ = std::chrono::steady_clock::now();
    wsrep::log_info() << "######## Stats ############";
    wsrep::log_info()  << stats();
    std::cout << db::ti::stats() << std::endl;
    wsrep::log_info() << "######## Stats ############";
    if (params_.fast_exit)
    {
        exit(0);
    }
    for (auto& i : servers_)
    {
        db::server& server(*i.second);
        wsrep::log_info() << "Status for server: "
                          << server.server_state().id();
        auto status(server.server_state().provider().status());
        for_each(status.begin(), status.end(),
                 [](const wsrep::provider::status_variable& sv)
                 {
                     wsrep::log_info() << sv.name() << " = " << sv.value();
                 });
        server.server_state().disconnect();
        server.server_state().wait_until_state(
            wsrep::server_state::s_disconnected);
        server.stop_applier();
        server.server_state().unload_provider();
    }
}

std::string db::simulator::server_port(size_t i) const
{
    std::ostringstream os;
    os << (10000 + (i + 1)*10);
    return os.str();
}

std::string db::simulator::build_cluster_address() const
{
    std::string ret;
    if (params_.wsrep_provider.find("galera_smm") != std::string::npos)
    {
        ret += "gcomm://";
    }

    for (size_t i(0); i < params_.n_servers; ++i)
    {
        std::ostringstream sa_os;
        sa_os << "127.0.0.1:";
        sa_os << server_port(i);
        ret += sa_os.str();
        if (i < params_.n_servers - 1) ret += ",";
    }
    return ret;
}

max_stars_repo_name: zettadb/zettalib

max_issues_repo_name: zettadb/zettalib

max_forks_repo_name: zettadb/zettalib


text:
#pragma once

#include <vector>
#include "GNAObject.hh"
#include "GNAObjectBind1N.hh"
#include <boost/optional.hpp>

namespace GNA{
    namespace GNAObjectTemplates{
        template<typename FloatType>
        class ViewHistBasedT: public GNAObjectBind1N<FloatType>,
                              public TransformationBind<ViewHistBasedT<FloatType>,FloatType,FloatType> {
        private:
            using BaseClass = GNAObjectT<FloatType,FloatType>;
            using typename BaseClass::TypesFunctionArgs;
            using typename BaseClass::FunctionArgs;
        public:
            using ViewHistBasedType = ViewHistBasedT<FloatType>;
            using typename BaseClass::SingleOutput;
            using TransformationDescriptor = typename BaseClass::TransformationDescriptorType;
            using OutputDescriptor = typename BaseClass::OutputDescriptor;

            ViewHistBasedT(FloatType threshold, FloatType ceiling);
            ViewHistBasedT(SingleOutput& output, FloatType threshold, FloatType ceiling);

            void set(SingleOutput& hist){
                hist.single() >> this->transformations.front().inputs.front();
            }

            TransformationDescriptor add_transformation(const std::string& name="");

        protected:
            void histTypes(TypesFunctionArgs& fargs);
            void types(TypesFunctionArgs& fargs);
            void init();

            boost::optional<FloatType> m_threshold;
            boost::optional<FloatType> m_ceiling;

            boost::optional<size_t> m_start;
            boost::optional<size_t> m_len;
            size_t m_full_length;
        };
    }
}

ext: hh

max_stars_repo_name: gnafit/gna

max_issues_repo_name: gnafit/gna

max_forks_repo_name: gnafit/gna


text:
#ifndef SERIALIZATION_H
#define SERIALIZATION_H

/**
 * Headers for all boost::serialization functionalities used in project.
 */

#include <boost/archive/text_oarchive.hpp>
#include <boost/archive/text_iarchive.hpp>
#include <boost/serialization/export.hpp>

#include <boost/serialization/base_object.hpp>
#include <boost/serialization/binary_object.hpp>
#include <boost/serialization/vector.hpp>
#include <boost/serialization/shared_ptr.hpp>
#include <boost/serialization/weak_ptr.hpp>

#endif // SERIALIZATION_H

max_stars_repo_name: Kamilcuk/Anthill

max_issues_repo_name: Kamilcuk/Anthill

max_forks_repo_name: Kamilcuk/Anthill


text:
#include <boost/filesystem.hpp>
#include <gdbplz/connection.hpp>
#include "../Catch/include/catch.hpp"

TEST_CASE("guessing gdb path", "[connection]")
{
	boost::optional<boost::filesystem::path> path = gdbplz::guess_gdb_path();
	REQUIRE((!path || boost::filesystem::exists(path.value())) == true);
}

TEST_CASE("launching with non-existing gdb", "[connection]")
{
	REQUIRE_THROWS_AS([](){
		gdbplz::connection conn("/sfsfsdfwdsfwerfwef/xcvxvfsdfsdfwerwefrsgfdfgdfgergtergdfg");
	}(), gdbplz::gdb_not_found);
}

max_stars_repo_name: milleniumbug/gdbplz

max_issues_repo_name: milleniumbug/gdbplz

max_forks_repo_name: milleniumbug/gdbplz


text:
//---------------------------------------------------------------------------//
// Copyright (c) 2013 Kyle Lutz <kyle.r.lutz@gmail.com>
//
// Distributed under the Boost Software License, Version 1.0
// See accompanying file LICENSE_1_0.txt or copy at
// http://www.boost.org/LICENSE_1_0.txt
//
// See http://boostorg.github.com/compute for more information.
//---------------------------------------------------------------------------//

#ifndef BOOST_COMPUTE_FUNCTIONAL_COMMON_HPP
#define BOOST_COMPUTE_FUNCTIONAL_COMMON_HPP

#include <boost/compute/functional/detail/macros.hpp>

namespace boost {
namespace compute {

BOOST_COMPUTE_DECLARE_BUILTIN_FUNCTION(clamp, T (T, T, T), class T)
BOOST_COMPUTE_DECLARE_BUILTIN_FUNCTION(degrees, T (T), class T)
BOOST_COMPUTE_DECLARE_BUILTIN_FUNCTION(radians, T (T), class T)
BOOST_COMPUTE_DECLARE_BUILTIN_FUNCTION(sign, T (T), class T)
BOOST_COMPUTE_DECLARE_BUILTIN_FUNCTION(smoothstep, T (T, T, T), class T)
BOOST_COMPUTE_DECLARE_BUILTIN_FUNCTION(step, T (T, T), class T)

} // end compute namespace
} // end boost namespace

#endif // BOOST_COMPUTE_FUNCTIONAL_COMMON_HPP

max_stars_repo_name: rudylee/expo

max_issues_repo_name: rudylee/expo

max_forks_repo_name: rudylee/expo


text:
// -*-C++-*-
// bddmain.cpp
// CBdd class implementation

#include "core/utils.h"
#include "core/bddmain.h"

#include <boost/dynamic_bitset.hpp>
#include <deque>

bdd_manager CBdd::manager=0;
bdd* CBdd::vars=0;
int CBdd::length=0;

void CBdd::Init(int len, int props, ...)
{
	va_list ap;
	va_start(ap, props);
	va_end(ap);

	length = len;

	vars = new bdd[len+1];
	manager = bdd_init();

	bdd_node_limit(manager, 10000000);

	if(mtbdd_one_data(manager, -1l, -1l)!=0)    // mtbdd one
	{
		throw new CDesertException("mtbdd_one_data fails: other terminal nodes already exist");
	}

	vars[0]=bdd_new_var_first(manager);
	for(int i=1; i<=length; i++)
		vars[i]=bdd_new_var_after(manager, vars[i-1]);
}

void CBdd::Finit(void)
{
	for (int i=0; i<length; i++)
		bdd_free(manager, vars[i]);
	bdd_quit(manager);
	delete[] vars;

	manager=0;
	length=0;
	vars=0;
}

CBdd CBdd::Encode(int encVal, int startVar, int encLen)
{
	MANAGER_CHECK("CBdd::Encode()");
	bdd ret=bdd_one(manager);
	int i, n;
	for(i=startVar+encLen-1, n=0; n < encLen; i--, n++)
	{
		int bit = encVal & 0x1;
		encVal = encVal >> 1;
		ret = bit ? bdd_and(manager, ret, vars[i] ) : bdd_and(manager, ret, bdd_not(manager, vars[i]) );
	}
	return CBdd(ret);
}

CBdd CBdd::Encode(const boost::dynamic_bitset<>& enc, int begin_var, int num_vars)
{
	MANAGER_CHECK("CBdd::Encode()");
	bdd ret = bdd_one(manager);
	int i, n;
	for (i = begin_var, n = 0; n < num_vars; i++, n++)
	{
		if (enc[n] == true) ret = bdd_and(manager, ret, vars[i]);
		if (enc[n] == false) ret = bdd_and(manager, ret, bdd_not(manager, vars[i]));
	}
	return CBdd(ret);
}

CBdd CBdd::Encode(int *enc, int begin_var, int num_vars)
{
	MANAGER_CHECK("CBdd::Encode()");
	bdd ret=bdd_one(manager);
	int i, n;
	for(i=begin_var, n=0; n < num_vars; i++, n++)
	{
		if(enc[n] == 1) ret = bdd_and(manager, ret, vars[i] );
		if(enc[n] == 0) ret = bdd_and(manager, ret, bdd_not(manager, vars[i]) );
	}
	return CBdd(ret);
}

CBdd CBdd::Encode(CVIndex enc[], int len)
{
  MANAGER_CHECK("CBdd::Encode()");
  bdd ret=bdd_one(manager);
  for(int i=0; i < len; i++)
  {
    if (enc[i].index >= length)
    {
      Error(_T("CBddManager::m_encode"), _T("index (%d) >= length (%d)"), enc[i].index, length);
      break;
    }
    if(enc[i].val == 1) ret = bdd_and(manager, ret, vars[enc[i].index] );
    if(enc[i].val == 0) ret = bdd_and(manager, ret, bdd_not(manager, vars[enc[i].index]));
  }
  return CBdd(ret);
}

int CBdd::Satisfy(CBdd& b, std::deque<boost::dynamic_bitset<>>& encVectors, int (*callback)(void*, const boost::dynamic_bitset<>&), void* arg)
{
	/*
	The function Satisfy fills up and returns a list of bitvectors.
	The size of the list is the number of configurations.
	This size of bitvector is the number of bits in the encoding. For ex.
	for a 2-bit encoding Satisy may return something like 00, 10, 11. Thus,
	in effect it enumerates and returns all encodings that satisfy the BDD. This
	internalyy uses the bdd library function bdd_satisfy.
	*/
  bdd f = b.core;

  if (f == bdd_one(manager))
  {
    int *encVec = new int[length];
    memset(encVec, 0xff, sizeof(int)*length);
    ExpandDontCare(encVec, 0, encVectors, callback, arg);
    delete[] encVec;
  }
  else
  {
    int rows=0;
    int **mat = new int*[BDD_MAX_PATHS];
    bdd_sat_f_mat(manager, f, 0, length-1, mat, &rows);
	if (rows > BDD_MAX_PATHS) {
		for (int i = 0; i < BDD_MAX_PATHS; i++)
		{
			free(mat[i]);
		}
		delete[] mat;
		return DESIGN_SPACE_TOO_LARGE;
	}
    if (rows > 0)
    {
      for (int i=0; i<rows; i++)
      {
        ExpandDontCare(mat[i], 0, encVectors, callback, arg);
        // mat[i] is allocated in bdd_sat_f_mat
        free(mat[i]);
      }
    }

    delete[] mat;
  }

  return encVectors.size();
}

void CBdd::ExpandDontCare(int *enc, int cur, std::deque<boost::dynamic_bitset<>>& encVectors, int(*callback)(void*, const boost::dynamic_bitset<>&), void* arg)
{
	/*
	enumerate the  dont care's
	*/
  if (cur == length)
  {
	  boost::dynamic_bitset<> bs(length);
	  for (int i = 0; i < length; i++)
	  {
#ifdef _DEBUG
		  if (enc[i] != 0 && enc[i] != 1)
			  throw std::runtime_error("Internal error 162");
#endif
		  bs.set(i, (bool)enc[i]);
	  }
	  if (callback == NULL || (*callback)(arg, bs)) {
		encVectors.emplace_back(std::move(bs));
	  }
  } else if (enc[cur] == -1) {
    enc[cur] = 0;
    ExpandDontCare(enc, cur + 1, encVectors, callback, arg);
    enc[cur] = 1;
    ExpandDontCare(enc, cur + 1, encVectors, callback, arg);
    enc[cur] = -1;
  } else {
    ExpandDontCare(enc, cur + 1, encVectors, callback, arg);
  }
}

max_stars_repo_name: lefevre-fraser/openmeta-mms

max_issues_repo_name: lefevre-fraser/openmeta-mms

max_forks_repo_name: lefevre-fraser/openmeta-mms


text:
#include "IndexTaskService.h"

#include <common/JobScheduler.h>
#include <aggregator-manager/IndexWorker.h>
#include <node-manager/NodeManagerBase.h>
#include <node-manager/MasterManagerBase.h>
#include <node-manager/sharding/ScdSharder.h>
#include <node-manager/sharding/ScdDispatcher.h>
#include <node-manager/DistributeRequestHooker.h>
#include <node-manager/DistributeFileSyncMgr.h>
#include <node-manager/DistributeFileSys.h>
#include <util/driver/Request.h>
#include <common/Utilities.h>

#include <glog/logging.h>
#include <boost/filesystem.hpp>

namespace bfs = boost::filesystem;
using namespace izenelib::driver;

namespace sf1r
{
static const char* SCD_BACKUP_DIR = "backup";
const static std::string DISPATCH_TEMP_DIR = "dispatch-temp-dir/";

IndexTaskService::IndexTaskService(IndexBundleConfiguration* bundleConfig)
: bundleConfig_(bundleConfig)
{
    service_ = Sf1rTopology::getServiceName(Sf1rTopology::SearchService);

    //ShardingConfig::RangeListT ranges;
    //ranges.push_back(100);
    //ranges.push_back(1000);
    //shard_cfg_.addRangeShardKey("UnchangableRangeProperty", ranges);
    //ShardingConfig::AttrListT strranges;
    //strranges.push_back("abc");
    //shard_cfg_.addAttributeShardKey("UnchangableAttributeProperty", strranges);

    if (DistributeFileSys::get()->isEnabled() && !bundleConfig_->indexShardKeys_.empty())
    {
        const std::string& coll = bundleConfig_->collectionName_;
        sharding_map_dir_ = DistributeFileSys::get()->getFixedCopyPath("/sharding_map/");
        sharding_map_dir_ = DistributeFileSys::get()->getDFSPathForLocal(sharding_map_dir_);
        if (!bfs::exists(sharding_map_dir_))
        {
            bfs::create_directories(sharding_map_dir_);
        }
        sharding_strategy_.reset(new MapShardingStrategy(sharding_map_dir_ + coll));
        sharding_strategy_->shard_cfg_.shardidList_ = bundleConfig_->col_shard_info_.shardList_;
        if (sharding_strategy_->shard_cfg_.shardidList_.empty())
            throw "sharding config is error!";
        sharding_strategy_->shard_cfg_.setUniqueShardKey(bundleConfig_->indexShardKeys_[0]);
        sharding_strategy_->init();
    }
}

IndexTaskService::~IndexTaskService()
{
}

std::string IndexTaskService::getShardingMapDir()
{
    return sharding_map_dir_;
}

bool IndexTaskService::SendRequestToSharding(uint32_t shardid)
{
    Request::kCallType hooktype = (Request::kCallType)DistributeRequestHooker::get()->getHookType();
    if (hooktype == Request::FromAPI)
    {
        return true;
    }
    const std::string& reqdata = DistributeRequestHooker::get()->getAdditionData();
    bool ret = false;
    if (hooktype == Request::FromDistribute)
    {
        indexAggregator_->singleRequest(bundleConfig_->collectionName_, 0,
            "HookDistributeRequestForIndex", (int)hooktype, reqdata, ret, shardid);
    }
    else
    {
        ret = true;
    }
    if (!ret)
    {
        LOG(WARNING) << "Send Request to shard node failed.";
    }
    return ret;
}

bool IndexTaskService::HookDistributeRequestForIndex()
{
    Request::kCallType hooktype = (Request::kCallType)DistributeRequestHooker::get()->getHookType();
    if (hooktype == Request::FromAPI)
    {
        // from api do not need hook, just process as usually.
        return true;
    }
    const std::string& reqdata = DistributeRequestHooker::get()->getAdditionData();
    bool ret = false;
    if (hooktype == Request::FromDistribute)
    {
        indexAggregator_->distributeRequestWithoutLocal(bundleConfig_->collectionName_, 0, "HookDistributeRequestForIndex", (int)hooktype, reqdata, ret);
    }
    else
    {
        // local hook has been moved to the request controller.
        ret = true;
    }
    if (!ret)
    {
        LOG(WARNING) << "Request failed, HookDistributeRequestForIndex failed.";
    }
    return ret;
}

bool IndexTaskService::index(unsigned int numdoc, std::string scd_path, int disable_sharding_type)
{
    bool result = true;

    // disable_sharding_type , 0 no disable, 1 disable sending request to sharding node,
    // 2 disable sending to sharding node and disable sharding SCD on local. (This means
    //  local node will index all of the docs in the given SCD files.)
    bool disable_sharding = (disable_sharding_type != (int)DefaultShard);

    indexWorker_->disableSharding(disable_sharding_type == (int)LocalOnly);

    if (DistributeFileSys::get()->isEnabled())
    {
        if (scd_path.empty())
        {
            LOG(ERROR) << "scd path should be specified while dfs is enabled.";
            return false;
        }
        scd_path = DistributeFileSys::get()->getDFSPathForLocal(scd_path);
    }
    else
    {
        scd_path = bundleConfig_->indexSCDPath();
    }

    if (bundleConfig_->isMasterAggregator() && indexAggregator_->isNeedDistribute() &&
        DistributeRequestHooker::get()->isRunningPrimary())
    {
        if (DistributeRequestHooker::get()->isHooked())
        {
            if (DistributeRequestHooker::get()->getHookType() == Request::FromDistribute &&
                !disable_sharding)
            {
                result = distributedIndex_(numdoc, scd_path);
            }
            else
            {
                if (disable_sharding)
                    LOG(INFO) << "==== The sharding is disabled! =====";
                //if (DistributeFileSys::get()->isEnabled())
                //{
                //    // while dfs enabled, the master will shard the scd file under the main scd_path
                //    //  to sub-directory directly on dfs.
                //    // and the shard worker only index the sub-directory belong to it.
                //    //
                //    std::string myshard;
                //    myshard = boost::lexical_cast<std::string>(MasterManagerBase::get()->getMyShardId());
                //    scd_path = (bfs::path(scd_path)/bfs::path(DISPATCH_TEMP_DIR + myshard)).string();
                //}

                if (!isNeedDoLocal())
                    return false;
                indexWorker_->index(scd_path, numdoc, result);
            }
        }
        else
        {
            task_type task = boost::bind(&IndexTaskService::distributedIndex_, this, numdoc, scd_path);
            JobScheduler::get()->addTask(task, bundleConfig_->collectionName_);
        }
    }
    else
    {
        if (bundleConfig_->isMasterAggregator() &&
            DistributeRequestHooker::get()->isRunningPrimary() && !DistributeFileSys::get()->isEnabled())
        {
            LOG(INFO) << "only local worker available, copy master scd files and indexing local.";
            // search the directory for files
            static const bfs::directory_iterator kItrEnd;
            std::string masterScdPath = bundleConfig_->masterIndexSCDPath();
            ScdParser parser(bundleConfig_->encoding_);
            bfs::path bkDir = bfs::path(masterScdPath) / SCD_BACKUP_DIR;
            LOG(INFO) << "creating directory : " << bkDir;
            bfs::create_directories(bkDir);
            for (bfs::directory_iterator itr(masterScdPath); itr != kItrEnd; ++itr)
            {
                if (bfs::is_regular_file(itr->status()))
                {
                    std::string fileName = itr->path().filename().string();
                    if (parser.checkSCDFormat(fileName))
                    {
                        try {
                        bfs::copy_file(itr->path().string(), bundleConfig_->indexSCDPath() + "/" + fileName);
                        LOG(INFO) << "SCD File copy to local index path:" << fileName;
                        LOG(INFO) << "moving SCD files to directory " << bkDir;
                            bfs::rename(itr->path().string(), bkDir / itr->path().filename());
                        }
                        catch(const std::exception& e) {
                            LOG(WARNING) << "failed to move file: " << std::endl << fileName << std::endl << e.what();
                        }
                    }
                    else
                    {
                        LOG(WARNING) << "SCD File not valid " << fileName;
                    }
                }
            }
        }

        if (!isNeedDoLocal())
            return false;
        indexWorker_->index(scd_path, numdoc, result);
    }

    return result;
}

bool IndexTaskService::isNeedSharding()
{
    if (bundleConfig_->isMasterAggregator() && indexAggregator_->isNeedDistribute() &&
        DistributeRequestHooker::get()->isRunningPrimary())
    {
        return DistributeRequestHooker::get()->getHookType() == Request::FromDistribute;
    }
    return false;
}

bool IndexTaskService::isNeedDoLocal()
{
    if (NodeManagerBase::get()->isDistributed() && !bundleConfig_->isWorkerNode())
    {
        LOG(INFO) << "local worker is disabled, no need do local work.";
        return false;
    }
    return true;
}

bool IndexTaskService::reindex_from_scd(const std::vector<std::string>& scd_list, int64_t timestamp)
{
    if (!isNeedDoLocal())
        return false;
    indexWorker_->disableSharding(false);
    return indexWorker_->buildCollection(0, scd_list, timestamp);
}

bool IndexTaskService::index(boost::shared_ptr<DocumentManager>& documentManager, int64_t timestamp)
{
    if (!isNeedDoLocal())
        return false;
    return indexWorker_->reindex(documentManager, timestamp);
}

bool IndexTaskService::optimizeIndex()
{
    if (!isNeedDoLocal())
        return false;
    return indexWorker_->optimizeIndex();
}

bool IndexTaskService::createDocument(const Value& documentValue)
{
    if (isNeedSharding())
    {
        if (!scdSharder_)
            createScdSharder(scdSharder_);
        SCDDoc scddoc;
        IndexWorker::value2SCDDoc(documentValue, scddoc);
        shardid_t shardid = scdSharder_->sharding(scddoc);
        if (shardid != MasterManagerBase::get()->getMyShardId())
        {
            // need send to the shard.
            return SendRequestToSharding(shardid);
        }
    }
    return indexWorker_->createDocument(documentValue);
}

bool IndexTaskService::updateDocument(const Value& documentValue)
{
    if (isNeedSharding())
    {
        if (!scdSharder_)
            createScdSharder(scdSharder_);
        SCDDoc scddoc;
        IndexWorker::value2SCDDoc(documentValue, scddoc);
        shardid_t shardid = scdSharder_->sharding(scddoc);
        if (shardid != MasterManagerBase::get()->getMyShardId())
        {
            // need send to the shard.
            return SendRequestToSharding(shardid);
        }
    }

    return indexWorker_->updateDocument(documentValue);
}
bool IndexTaskService::updateDocumentInplace(const Value& request)
{
    if (isNeedSharding())
    {
        if (!scdSharder_)
            createScdSharder(scdSharder_);
        SCDDoc scddoc;
        IndexWorker::value2SCDDoc(request, scddoc);
        shardid_t shardid = scdSharder_->sharding(scddoc);
        if (shardid != MasterManagerBase::get()->getMyShardId())
        {
            // need send to the shard.
            return SendRequestToSharding(shardid);
        }
    }

    return indexWorker_->updateDocumentInplace(request);
}

bool IndexTaskService::destroyDocument(const Value& documentValue)
{
    if (isNeedSharding())
    {
        if (!scdSharder_)
            createScdSharder(scdSharder_);
        SCDDoc scddoc;
        IndexWorker::value2SCDDoc(documentValue, scddoc);
        shardid_t shardid = scdSharder_->sharding(scddoc);
        if (shardid != MasterManagerBase::get()->getMyShardId())
        {
            // need send to the shard.
            return SendRequestToSharding(shardid);
        }
    }

    return indexWorker_->destroyDocument(documentValue);
}

void IndexTaskService::flush()
{
    indexWorker_->flush(true);
}

bool IndexTaskService::getIndexStatus(Status& status)
{
    return indexWorker_->getIndexStatus(status);
}

bool IndexTaskService::isAutoRebuild()
{
    return bundleConfig_->isAutoRebuild_;
}

std::string IndexTaskService::getScdDir(bool rebuild) const
{
    if (rebuild)
        return bundleConfig_->rebuildIndexSCDPath();
    if( bundleConfig_->isMasterAggregator() )
        return bundleConfig_->masterIndexSCDPath();
    return bundleConfig_->indexSCDPath();
}

CollectionPath& IndexTaskService::getCollectionPath() const
{
    return bundleConfig_->collPath_;
}

boost::shared_ptr<DocumentManager> IndexTaskService::getDocumentManager() const
{
    return indexWorker_->getDocumentManager();
}

bool IndexTaskService::distributedIndex_(unsigned int numdoc, std::string scd_dir)
{
    // notify that current master is indexing for the specified collection,
    // we may need to check that whether other Master it's indexing this collection in some cases,
    // or it's depends on Nginx router strategy.
    MasterManagerBase::get()->registerIndexStatus(bundleConfig_->collectionName_, true);

    if (!DistributeFileSys::get()->isEnabled())
    {
        scd_dir = bundleConfig_->masterIndexSCDPath();
    }

    bool ret = distributedIndexImpl_(
                    numdoc,
                    bundleConfig_->collectionName_,
                    scd_dir);

    MasterManagerBase::get()->registerIndexStatus(bundleConfig_->collectionName_, false);

    return ret;
}

bool IndexTaskService::distributedIndexImpl_(
    unsigned int numdoc,
    const std::string& collectionName,
    const std::string& masterScdPath)
{
    if (!scdSharder_)
    {
        if (!createScdSharder(scdSharder_))
        {
            LOG(ERROR) << "create scd sharder failed.";
            return false;
        }
        if (!scdSharder_)
        {
            LOG(INFO) << "no scd sharder!";
            return false;
        }
    }

    if (!MasterManagerBase::get()->isAllShardNodeOK(bundleConfig_->col_shard_info_.shardList_))
    {
        LOG(ERROR) << "some of sharding node is not ready for index.";
        return false;
    }

    std::string scd_dir = masterScdPath;
    std::vector<std::string> outScdFileList;
    //
    // 1. dispatching scd to multiple nodes
    if (!DistributeFileSys::get()->isEnabled())
    {
        boost::shared_ptr<ScdDispatcher> scdDispatcher(new BatchScdDispatcher(scdSharder_,
                collectionName, DistributeFileSys::get()->isEnabled()));
        if(!scdDispatcher->dispatch(outScdFileList, masterScdPath, bundleConfig_->indexSCDPath(), numdoc))
            return false;
    }
    // 2. send index request to multiple nodes
    LOG(INFO) << "start distributed indexing";
    HookDistributeRequestForIndex();

    if (!DistributeFileSys::get()->isEnabled())
        scd_dir = bundleConfig_->indexSCDPath();

    bool ret = true;
    if (isNeedDoLocal())
    {
        // starting local index.
        indexWorker_->index(scd_dir, numdoc, ret);
    }
    if (ret && !DistributeFileSys::get()->isEnabled())
    {
        bfs::path bkDir = bfs::path(masterScdPath) / SCD_BACKUP_DIR;
        bfs::create_directories(bkDir);
        LOG(INFO) << "moving " << outScdFileList.size() << " SCD files to directory " << bkDir;
        for (size_t i = 0; i < outScdFileList.size(); i++)
        {
            try {
                bfs::rename(outScdFileList[i], bkDir / bfs::path(outScdFileList[i]).filename());
            }
            catch(const std::exception& e) {
                LOG(WARNING) << "failed to move file: " << std::endl << outScdFileList[i] << std::endl << e.what();
            }
        }
    }

    if (!isNeedDoLocal())
        return false;
    return ret;
}

bool IndexTaskService::createScdSharder(
    boost::shared_ptr<ScdSharder>& scdSharder)
{
    return indexWorker_->createScdSharder(scdSharder);
}

izenelib::util::UString::EncodingType IndexTaskService::getEncode() const
{
    return bundleConfig_->encoding_;
}

const std::vector<shardid_t>& IndexTaskService::getShardidListForSearch()
{
    return bundleConfig_->col_shard_info_.shardList_;
}

typedef std::map<shardid_t, std::vector<vnodeid_t> > ShardingTopologyT;
static void printSharding(const ShardingTopologyT& sharding_topology)
{
    for(ShardingTopologyT::const_iterator cit = sharding_topology.begin();
        cit != sharding_topology.end(); ++cit)
    {
        std::cout << "sharding : " << (uint32_t)cit->first << " is holding : ";
        for (size_t i = 0; i < cit->second.size(); ++i)
        {
            std::cout << cit->second[i] << ", ";
        }
        std::cout << std::endl;
    }
}

static bool removeSharding(const std::vector<shardid_t>& remove_sharding_nodes,
    std::vector<shardid_t>& left_sharding_nodes,
    size_t new_vnode_for_sharding,
    ShardingTopologyT& current_sharding_topology,
    std::map<vnodeid_t, std::pair<shardid_t, shardid_t> >& migrate_data_list,
    std::vector<shardid_t>& sharding_map)
{
    ShardingTopologyT remove_sharding_topo;
    for (size_t i = 0; i < remove_sharding_nodes.size(); ++i)
    {
        remove_sharding_topo[remove_sharding_nodes[i]] = current_sharding_topology[remove_sharding_nodes[i]];
        current_sharding_topology.erase(remove_sharding_nodes[i]);
    }

    if (current_sharding_topology.empty() || remove_sharding_topo.size() != remove_sharding_nodes.size())
    {
        return false;
    }
    ShardingTopologyT::iterator migrate_to_it = current_sharding_topology.begin();
    for(ShardingTopologyT::iterator remove_it = remove_sharding_topo.begin();
        remove_it != remove_sharding_topo.end(); ++remove_it)
    {
        size_t vnode_index = 0;
        for (; vnode_index < remove_it->second.size();
            ++vnode_index)
        {
            vnodeid_t vid = remove_it->second[vnode_index];
            migrate_data_list[vid] = std::pair<shardid_t, shardid_t>(remove_it->first, migrate_to_it->first);
            LOG(INFO) << "vnode : " << vid << " will be moved from "
                << (uint32_t)remove_it->first << " to " << (uint32_t)migrate_to_it->first;
            sharding_map[vid] = migrate_to_it->first;
            migrate_to_it->second.push_back(vid);
            if (migrate_to_it->second.size() > new_vnode_for_sharding)
            {
                // the new sharding node got enough data. move to next.
                left_sharding_nodes.push_back(migrate_to_it->first);
                ++migrate_to_it;
                if (migrate_to_it == current_sharding_topology.end())
                {
                    ShardingTopologyT::iterator tmpit = remove_it;
                    if (vnode_index != (remove_it->second.size() - 1) ||
                        ++tmpit != remove_sharding_topo.end() )
                    {
                        LOG(INFO) << "the remove sharding vnode can not totally migrate to others.";
                        return false;
                    }
                    break;
                }
            }
        }
        remove_it->second.clear();
    }

    while(migrate_to_it != current_sharding_topology.end())
    {
        left_sharding_nodes.push_back(migrate_to_it->first);
        ++migrate_to_it;
    }
    return true;
}

static void migrateSharding(const std::vector<shardid_t>& new_sharding_nodes,
    size_t new_vnode_for_sharding,
    ShardingTopologyT& current_sharding_topology,
    ShardingTopologyT& new_sharding_topology,
    std::map<vnodeid_t, std::pair<shardid_t, shardid_t> >& migrate_data_list,
    std::vector<shardid_t>& sharding_map)
{
    // move the vnode from src sharding node to dest sharding node.
    size_t migrate_to = 0;

    for(ShardingTopologyT::iterator it = current_sharding_topology.begin();
        it != current_sharding_topology.end(); ++it)
    {
        size_t migrate_start = it->second.size() - 1;
        while (migrate_to < new_sharding_nodes.size())
        {
            // this old sharding has not enough data so do not migrate from this node.
            if (it->second.size() <= new_vnode_for_sharding)
                break;
            size_t old_start = migrate_start;
            for (size_t vnode_index = old_start;
                vnode_index > new_vnode_for_sharding;
                --vnode_index)
            {
                vnodeid_t vid = it->second[vnode_index];
                migrate_data_list[vid] = std::pair<shardid_t, shardid_t>(it->first, new_sharding_nodes[migrate_to]);
                LOG(INFO) << "vnode : " << vid << " will be moved from " << (uint32_t)it->first << " to " << (uint32_t)new_sharding_nodes[migrate_to];
                sharding_map[vid] = new_sharding_nodes[migrate_to];
                new_sharding_topology[new_sharding_nodes[migrate_to]].push_back(vid);
                --migrate_start;
                if (new_sharding_topology[new_sharding_nodes[migrate_to]].size() >= new_vnode_for_sharding)
                {
                    // the new sharding node got enough data. move to next.
                    ++migrate_to;
                    if (migrate_to >= new_sharding_nodes.size())
                        break;
                }
            }
            it->second.erase(it->second.begin() + migrate_start, it->second.end());
        }
        if (migrate_to >= new_sharding_nodes.size())
            break;
    }
}

bool IndexTaskService::doMigrateWork(bool removing,
    const std::map<vnodeid_t, std::pair<shardid_t, shardid_t> >& migrate_data_list,
    const std::vector<shardid_t>& migrate_nodes,
    const std::string& map_file,
    const std::vector<shardid_t>& current_sharding_map)
{
    std::map<std::string, std::map<shardid_t, std::vector<vnodeid_t> > > migrate_from_to_list;
    for(std::map<vnodeid_t, std::pair<shardid_t, shardid_t> >::const_iterator cit = migrate_data_list.begin();
        cit != migrate_data_list.end(); ++cit)
    {
        std::string shardip = MasterManagerBase::get()->getShardNodeIP(cit->second.first);
        if (shardip.empty())
        {
            LOG(ERROR) << "get source shard node ip error. " << cit->second.first;
            return false;
        }
        migrate_from_to_list[shardip][cit->second.second].push_back(cit->first);
    }

    std::map<shardid_t, std::vector<std::string> > generated_insert_scds;
    // the scds used for remove on the src node.
    std::map<shardid_t, std::vector<std::string> > generated_del_scds;
    bool ret = true;
    ret = DistributeFileSyncMgr::get()->generateMigrateScds(bundleConfig_->collectionName_,
        migrate_from_to_list,
        generated_insert_scds,
        generated_del_scds);

    if (!ret)
    {
        LOG(ERROR) << "generate the migrate SCD files failed.";
        return false;
    }

    // wait for all sharding nodes to finish their write queue.
    if(!MasterManagerBase::get()->waitForMigrateReady(getShardidListForSearch()))
    {
        LOG(INFO) << " wait for migrate get ready failed.";
        return false;
    }

    if (!removing)
    {
        // wait new sharding nodes to started.
        if (!MasterManagerBase::get()->waitForNewShardingNodes(migrate_nodes))
        {
            LOG(INFO) << "wait for new sharding nodes to startup failed.";
            return false;
        }
    }

    if (!indexShardingNodes(generated_insert_scds))
    {
        return false;
    }

    if (!removing)
    {
        // wait for the new sharding nodes to finish indexing.
        MasterManagerBase::get()->waitForMigrateIndexing(migrate_nodes);
    }

    indexShardingNodes(generated_del_scds);
    MasterManagerBase::get()->waitForMigrateIndexing(getShardidListForSearch());

    MapShardingStrategy::saveShardingMapToFile(map_file, current_sharding_map);
    // update config will cause the collection to restart, so
    // the IndexTaskService will be destructed.
    updateShardingConfig(migrate_nodes, removing);
    return true;
}

bool IndexTaskService::addNewShardingNodes(const std::vector<shardid_t>& new_sharding_nodes)
{
    if (!bundleConfig_->isMasterAggregator())
    {
        LOG(INFO) << "change sharding node must be send to the master node.";
        return false;
    }
    if (!sharding_strategy_ || sharding_strategy_->shard_cfg_.shardidList_.size() == 0)
    {
        LOG(ERROR) << "no sharding config.";
        return false;
    }
    if (new_sharding_nodes.empty())
    {
        LOG(INFO) << "empty new sharding nodes.";
        return false;
    }
    for (size_t i = 0; i < new_sharding_nodes.size(); ++i)
    {
        if (std::find(bundleConfig_->col_shard_info_.shardList_.begin(),
                bundleConfig_->col_shard_info_.shardList_.end(),
                new_sharding_nodes[i]) != bundleConfig_->col_shard_info_.shardList_.end())
        {
            LOG(ERROR) << "new sharding nodes exists in the old sharding config.";
            return false;
        }
    }

    size_t current_sharding_num = sharding_strategy_->shard_cfg_.shardidList_.size();
    // reading current sharding config(include the load on these nodes),
    // and determine which nodes need
    // migrate which part of their data.
    std::vector<shardid_t> current_sharding_map;
    std::string map_file = sharding_map_dir_ + bundleConfig_->collectionName_;
    MapShardingStrategy::readShardingMapFile(map_file, current_sharding_map);

    if (bundleConfig_->col_shard_info_.shardList_.size() == 1)
    {
        if (current_sharding_map.empty())
        {
            current_sharding_map.resize(MapShardingStrategy::MAX_MAP_SIZE, bundleConfig_->col_shard_info_.shardList_[0]);
        }
    }

    if (current_sharding_map.empty())
    {
        LOG(ERROR) << "sharding map is empty!";
        return false;
    }
    if (current_sharding_map.size() < current_sharding_num + new_sharding_nodes.size())
    {
        LOG(ERROR) << "the actual sharding num is larger than virtual nodes." << current_sharding_map.size();
        return false;
    }
    size_t current_vnode_for_sharding = current_sharding_map.size()/current_sharding_num;
    size_t new_vnode_for_sharding = current_sharding_map.size()/(current_sharding_num + new_sharding_nodes.size());
    ShardingTopologyT current_sharding_topology;
    for (size_t i = 0; i < current_sharding_map.size(); ++i)
    {
        current_sharding_topology[current_sharding_map[i]].push_back(i);
    }

    LOG(INFO) << "Before migrate, the average vnodes for each sharding is: " << current_vnode_for_sharding
       << " and sharding topology is : ";
    printSharding(current_sharding_topology);

    // move the vnode from src sharding node to dest sharding node.
    std::map<vnodeid_t, std::pair<shardid_t, shardid_t> > migrate_data_list;
    ShardingTopologyT new_sharding_topology;

    migrateSharding(new_sharding_nodes, new_vnode_for_sharding,
        current_sharding_topology,
        new_sharding_topology, migrate_data_list,
        current_sharding_map);

    LOG(INFO) << "After migrate, the average vnodes for each sharding will be: " << new_vnode_for_sharding
       << " and sharding topology will be : ";
    printSharding(current_sharding_topology);
    printSharding(new_sharding_topology);

    // this will disallow any new write. This may fail if other migrate
    // running or not all sharding nodes alive.
    if(!MasterManagerBase::get()->notifyAllShardingBeginMigrate(getShardidListForSearch()))
    {
        return false;
    }

    bool ret = doMigrateWork(false, migrate_data_list, new_sharding_nodes,
        map_file, current_sharding_map);

    // allow new write running.
    MasterManagerBase::get()->notifyAllShardingEndMigrate();

    return ret;
}

bool IndexTaskService::indexShardingNodes(const std::map<shardid_t, std::vector<std::string> >& generated_migrate_scds)
{
    std::string tmp_migrate_scd_dir = DistributeFileSys::get()->getFixedCopyPath("/migrate_scds/"
        + bundleConfig_->collectionName_ + "/" + boost::lexical_cast<std::string>(Utilities::createTimeStamp()));

    bfs::create_directories(DistributeFileSys::get()->getDFSPathForLocal(tmp_migrate_scd_dir));

    std::map<shardid_t, std::vector<std::string> >::const_iterator cit = generated_migrate_scds.begin();
    for (; cit != generated_migrate_scds.end(); ++cit)
    {
        LOG(INFO) << "prepare scd files for sharding node : " << (uint32_t)cit->first;
        std::string shard_scd_dir = tmp_migrate_scd_dir + "/shard" + getShardidStr(cit->first) + "/";
        bfs::create_directories(DistributeFileSys::get()->getDFSPathForLocal(shard_scd_dir));
        for (size_t i = 0; i < cit->second.size(); ++i)
        {
            LOG(INFO) << "add migrate scd : " << cit->second[i];
            bfs::rename(DistributeFileSys::get()->getDFSPathForLocal(cit->second[i]),
                bfs::path(DistributeFileSys::get()->getDFSPathForLocal(shard_scd_dir))/(bfs::path(cit->second[i]).filename()));
        }

        // send index command.
        std::string json_req = "{\"collection\":\"" + bundleConfig_->collectionName_
            + "\",\"index_scd_path\":\"" + shard_scd_dir
            + "\",\"disable_sharding\":2"
            + ",\"header\":{\"action\":\"index\",\"controller\":\"commands\"},\"uri\":\"commands/index\"}";

        LOG(INFO) << "send request : " << json_req;
        std::vector<shardid_t> shardid;
        shardid.push_back(cit->first);
        if (!MasterManagerBase::get()->pushWriteReqToShard(json_req, shardid, true, true))
        {
            LOG(WARNING) << "push write failed.";
            return false;
        }
    }
    return true;
}

void IndexTaskService::updateShardingConfig(const std::vector<shardid_t>& new_sharding_nodes, bool removing)
{
    std::string sharding_cfg;
    std::vector<shardid_t> curr_shard_nodes = getShardidListForSearch();
    if (!removing)
    {
        for (size_t i = 0; i < curr_shard_nodes.size(); ++i)
        {
            if (sharding_cfg.empty())
                sharding_cfg = getShardidStr(curr_shard_nodes[i]);
            else
                sharding_cfg += "," + getShardidStr(curr_shard_nodes[i]);
        }
    }
    for (size_t i = 0; i < new_sharding_nodes.size(); ++i)
    {
        if (sharding_cfg.empty())
            sharding_cfg = getShardidStr(new_sharding_nodes[i]);
        else
            sharding_cfg += "," + getShardidStr(new_sharding_nodes[i]);
    }
    LOG(INFO) << "new sharding cfg is : " << sharding_cfg;
    //
    // send index command.
    std::string json_req = "{\"collection\":\"" + bundleConfig_->collectionName_
        + "\",\"new_sharding_cfg\":\"" + sharding_cfg
        + "\",\"header\":{\"action\":\"update_sharding_conf\",\"controller\":\"collection\"},\"uri\":\"collection/update_sharding_conf\"}";

    LOG(INFO) << "send request : " << json_req;
    if (!removing)
        MasterManagerBase::get()->pushWriteReqToShard(json_req, new_sharding_nodes, true, true);
    if (bundleConfig_->isMasterAggregator() && !bundleConfig_->isWorkerNode())
    {
        curr_shard_nodes.push_back(MasterManagerBase::get()->getMyShardId());
    }
    MasterManagerBase::get()->pushWriteReqToShard(json_req, curr_shard_nodes, true, true);
}

bool IndexTaskService::generateMigrateSCD(const std::map<shardid_t, std::vector<vnodeid_t> >& vnode_list,
    std::map<shardid_t, std::string>& generated_insert_scds,
    std::map<shardid_t, std::string>& generated_del_scds)
{
    return indexWorker_->generateMigrateSCD(vnode_list, generated_insert_scds, generated_del_scds);
}

bool IndexTaskService::removeShardingNodes(const std::vector<shardid_t>& remove_sharding_nodes)
{
    if (!bundleConfig_->isMasterAggregator())
    {
        LOG(INFO) << "change sharding node must be send to the master node.";
        return false;
    }

    if (!sharding_strategy_ || sharding_strategy_->shard_cfg_.shardidList_.size() == 0)
    {
        LOG(ERROR) << "no sharding config.";
        return false;
    }
    if (remove_sharding_nodes.empty())
    {
        LOG(INFO) << "empty sharding nodes.";
        return false;
    }

    size_t current_sharding_num = sharding_strategy_->shard_cfg_.shardidList_.size();
    if (current_sharding_num == 1)
    {
        LOG(INFO) << "The last one sharding node can not be removed.";
        return false;
    }

    if (remove_sharding_nodes.size() >= current_sharding_num)
    {
        LOG(INFO) << "You can not remove all sharding nodes.";
        return false;
    }

    for (size_t i = 0; i < remove_sharding_nodes.size(); ++i)
    {
        if (std::find(bundleConfig_->col_shard_info_.shardList_.begin(),
                bundleConfig_->col_shard_info_.shardList_.end(),
                remove_sharding_nodes[i]) == bundleConfig_->col_shard_info_.shardList_.end())
        {
            LOG(ERROR) << "removing sharding node does not exist in the old sharding config.";
            return false;
        }
    }

    std::vector<shardid_t> current_sharding_map;
    std::string map_file = sharding_map_dir_ + bundleConfig_->collectionName_;
    MapShardingStrategy::readShardingMapFile(map_file, current_sharding_map);

    if (current_sharding_map.empty())
    {
        LOG(ERROR) << "sharding map is empty!";
        return false;
    }

    size_t current_vnode_for_sharding = current_sharding_map.size()/current_sharding_num;
    size_t new_vnode_for_sharding = current_sharding_map.size()/(current_sharding_num - remove_sharding_nodes.size());

    ShardingTopologyT current_sharding_topology;
    for (size_t i = 0; i < current_sharding_map.size(); ++i)
    {
        current_sharding_topology[current_sharding_map[i]].push_back(i);
    }

    LOG(INFO) << "Before migrate, the average vnodes for each sharding is: " << current_vnode_for_sharding
       << " and sharding topology is : ";
    printSharding(current_sharding_topology);

    std::map<vnodeid_t, std::pair<shardid_t, shardid_t> > migrate_data_list;
    ShardingTopologyT new_sharding_topology;

    std::vector<shardid_t> left_sharding_nodes;
    bool ret = removeSharding(remove_sharding_nodes, left_sharding_nodes,
        new_vnode_for_sharding,
        current_sharding_topology,
        migrate_data_list,
        current_sharding_map);

    if (!ret || left_sharding_nodes.empty())
    {
        LOG(INFO) << "removing nodes are wrong.";
        return false;
    }
    LOG(INFO) << "After migrate, the average vnodes for each sharding will be: " << new_vnode_for_sharding
       << " and sharding topology will be : ";
    printSharding(current_sharding_topology);

    if(!MasterManagerBase::get()->notifyAllShardingBeginMigrate(getShardidListForSearch()))
    {
        return false;
    }

    ret = doMigrateWork(true, migrate_data_list, left_sharding_nodes,
        map_file, current_sharding_map);

    // allow new write running.
    MasterManagerBase::get()->notifyAllShardingEndMigrate();

    return ret;

}

}