<commit_before>/*
 * This file is open source software, licensed to you under the terms
 * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
 * distributed with this work for additional information regarding copyright
 * ownership.  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.
 */
/*
 * Copyright (C) 2015 Cloudius Systems, Ltd.
 */

#include <iostream>
#include "core/function_traits.hh"
#include "core/apply.hh"
#include "core/shared_ptr.hh"
#include "core/sstring.hh"
#include "core/future-util.hh"
#include "util/is_smart_ptr.hh"

namespace rpc {

template <bool IsSmartPtr>
struct serialize_helper;

template <>
struct serialize_helper<false> {
    template <typename Serializer, typename T>
    static inline future<> serialize(Serializer& serialize, output_stream<char>& out, T&& t) {
        return serialize(out, std::forward<T>(t));

    }
};

template <>
struct serialize_helper<true> {
    template <typename Serializer, typename T>
    static inline future<> serialize(Serializer& serialize, output_stream<char>& out, T&& t) {
        return serialize(out, *t);
    }
};

template<std::size_t N, typename Serializer, typename... T>
inline std::enable_if_t<N == sizeof...(T), future<>> marshall(Serializer&, output_stream<char>& out, std::tuple<T...>&&) {
    return out.flush();
}

template<std::size_t N = 0, typename Serializer, typename... T>
inline std::enable_if_t<N != sizeof...(T), future<>> marshall(Serializer& serialize, output_stream<char>& out, std::tuple<T...>&& args) {
    using tuple_type = std::tuple<T...>;
    using t_type = typename std::tuple_element<N, tuple_type>::type;
    using serialize_helper_type = serialize_helper<is_smart_ptr<typename std::remove_reference<t_type>::type>::value>;
    return serialize_helper_type::serialize(serialize, out, std::forward<t_type>(std::get<N>(args))).then([&serialize, &out, args = std::move(args)] () mutable {
        return marshall<N + 1>(serialize, out, std::move(args));
    });
}

// ArgsReady is a functor that will be called after each element is deserialized.
// It gets element's position in a tuple as a parameter. It is used by argument
// desererialization to mark already deserialized element as containing valid values
// that needs to be destroyed by a destructor.
template<std::size_t N, typename Serializer, typename ArgReady, typename... T>
inline std::enable_if_t<N == sizeof...(T), future<>> unmarshall(Serializer&, input_stream<char>&, std::tuple<T&...>&&, ArgReady&& argready) {
    return make_ready_future<>();
}

template<std::size_t N = 0, typename Serializer, typename ArgReady, typename... T>
inline std::enable_if_t<N != sizeof...(T), future<>> unmarshall(Serializer& deserialize, input_stream<char>& in, std::tuple<T&...>&& args, ArgReady&& argready) {
    // And you may ask yourself "What is that beautiful house?"^H^H^H^H "Why
    // make_ready_future() here?". And there answer would be to convert
    // exception thrown by deserialize info a future
    return make_ready_future().then([&deserialize, &in, args = std::move(args), argready = std::forward<ArgReady>(argready)] () mutable {
            return deserialize(in, std::get<N>(args)).then([&deserialize, &in, args = std::move(args), argready = std::forward<ArgReady>(argready)] () mutable {
                argready(N);
                return unmarshall<N + 1>(deserialize, in, std::move(args), std::forward<ArgReady>(argready));
            });
    });
}

template<typename Serializer, typename... T>
inline future<> unmarshall(Serializer& deserializer, input_stream<char>& in, std::tuple<T&...>&& args) {
    return unmarshall(deserializer, in, std::move(args), [](std::size_t n){});
}

// ref_tuple gets tuple and returns another tuple with references to members of received tuple
template<typename... T, std::size_t... I>
inline std::tuple<T&...> ref_tuple_impl(std::tuple<T...>& t, std::index_sequence<I...>) {
    return std::tie(std::get<I>(t)...);
}

template<typename... T>
inline std::tuple<T&...> ref_tuple(std::tuple<T...>& t) {
    return ref_tuple_impl(t, std::make_index_sequence<sizeof...(T)>());
}

struct reply_payload_base {
    sstring ex;
};

template <typename T>
struct reply_payload : reply_payload_base {
    void value_set() {
        v_set = true;
    }
    void value_set(T&& v) {
        new (&u.v) T(std::move(v));
        value_set();
    }
    union U {
        U() {}
        ~U() {}
        typename std::aligned_storage<sizeof(T), alignof(T)>::type pad;
        T v;
    } u;
    ~reply_payload() {
        if (v_set) {
            u.v.~T();
        }
    }
private:
    bool v_set = false; // set it to true when U::v is valid object
};

template<>
struct reply_payload<void> : reply_payload_base {
    void value_set() {}
};

template<typename Payload, typename... T>
struct rcv_reply_base : reply_payload<Payload> {
    bool done = false;
    promise<T...> p;
    template<typename... V>
    void set_value(V&&... v) {
        this->value_set();
        done = true;
        p.set_value(std::forward<V>(v)...);
    }
    ~rcv_reply_base() {
        if (!done) {
            p.set_exception(closed_error());
        }
    }
};

template<typename Serializer, typename MsgType, typename T>
struct rcv_reply : rcv_reply_base<T, T> {
    inline future<> get_reply(typename protocol<Serializer, MsgType>::client& dst) {
        return unmarshall(dst.serializer(), dst.in(), std::tie(this->u.v)).then([this] {
            this->set_value(this->u.v);
        });
    }
};

template<typename Serializer, typename MsgType, typename... T>
struct rcv_reply<Serializer, MsgType, future<T...>> : rcv_reply_base<std::tuple<T...>, T...> {
    inline future<> get_reply(typename protocol<Serializer, MsgType>::client& dst) {
        return unmarshall(dst.serializer(), dst.in(), ref_tuple(this->u.v)).then([this] {
            this->set_value(this->u.v);
        });
    }
};

template<typename Serializer, typename MsgType>
struct rcv_reply<Serializer, MsgType, void> : rcv_reply_base<void, void> {
    inline future<> get_reply(typename protocol<Serializer, MsgType>::client& dst) {
        this->set_value();
        return make_ready_future<>();<commit_msg>rpc: add missing move()s in get_reply()<commit_after>/*
 * This file is open source software, licensed to you under the terms
 * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
 * distributed with this work for additional information regarding copyright
 * ownership.  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.
 */
/*
 * Copyright (C) 2015 Cloudius Systems, Ltd.
 */

#include <iostream>
#include "core/function_traits.hh"
#include "core/apply.hh"
#include "core/shared_ptr.hh"
#include "core/sstring.hh"
#include "core/future-util.hh"
#include "util/is_smart_ptr.hh"

namespace rpc {

template <bool IsSmartPtr>
struct serialize_helper;

template <>
struct serialize_helper<false> {
    template <typename Serializer, typename T>
    static inline future<> serialize(Serializer& serialize, output_stream<char>& out, T&& t) {
        return serialize(out, std::forward<T>(t));

    }
};

template <>
struct serialize_helper<true> {
    template <typename Serializer, typename T>
    static inline future<> serialize(Serializer& serialize, output_stream<char>& out, T&& t) {
        return serialize(out, *t);
    }
};

template<std::size_t N, typename Serializer, typename... T>
inline std::enable_if_t<N == sizeof...(T), future<>> marshall(Serializer&, output_stream<char>& out, std::tuple<T...>&&) {
    return out.flush();
}

template<std::size_t N = 0, typename Serializer, typename... T>
inline std::enable_if_t<N != sizeof...(T), future<>> marshall(Serializer& serialize, output_stream<char>& out, std::tuple<T...>&& args) {
    using tuple_type = std::tuple<T...>;
    using t_type = typename std::tuple_element<N, tuple_type>::type;
    using serialize_helper_type = serialize_helper<is_smart_ptr<typename std::remove_reference<t_type>::type>::value>;
    return serialize_helper_type::serialize(serialize, out, std::forward<t_type>(std::get<N>(args))).then([&serialize, &out, args = std::move(args)] () mutable {
        return marshall<N + 1>(serialize, out, std::move(args));
    });
}

// ArgsReady is a functor that will be called after each element is deserialized.
// It gets element's position in a tuple as a parameter. It is used by argument
// desererialization to mark already deserialized element as containing valid values
// that needs to be destroyed by a destructor.
template<std::size_t N, typename Serializer, typename ArgReady, typename... T>
inline std::enable_if_t<N == sizeof...(T), future<>> unmarshall(Serializer&, input_stream<char>&, std::tuple<T&...>&&, ArgReady&& argready) {
    return make_ready_future<>();
}

template<std::size_t N = 0, typename Serializer, typename ArgReady, typename... T>
inline std::enable_if_t<N != sizeof...(T), future<>> unmarshall(Serializer& deserialize, input_stream<char>& in, std::tuple<T&...>&& args, ArgReady&& argready) {
    // And you may ask yourself "What is that beautiful house?"^H^H^H^H "Why
    // make_ready_future() here?". And there answer would be to convert
    // exception thrown by deserialize info a future
    return make_ready_future().then([&deserialize, &in, args = std::move(args), argready = std::forward<ArgReady>(argready)] () mutable {
            return deserialize(in, std::get<N>(args)).then([&deserialize, &in, args = std::move(args), argready = std::forward<ArgReady>(argready)] () mutable {
                argready(N);
                return unmarshall<N + 1>(deserialize, in, std::move(args), std::forward<ArgReady>(argready));
            });
    });
}

template<typename Serializer, typename... T>
inline future<> unmarshall(Serializer& deserializer, input_stream<char>& in, std::tuple<T&...>&& args) {
    return unmarshall(deserializer, in, std::move(args), [](std::size_t n){});
}

// ref_tuple gets tuple and returns another tuple with references to members of received tuple
template<typename... T, std::size_t... I>
inline std::tuple<T&...> ref_tuple_impl(std::tuple<T...>& t, std::index_sequence<I...>) {
    return std::tie(std::get<I>(t)...);
}

template<typename... T>
inline std::tuple<T&...> ref_tuple(std::tuple<T...>& t) {
    return ref_tuple_impl(t, std::make_index_sequence<sizeof...(T)>());
}

struct reply_payload_base {
    sstring ex;
};

template <typename T>
struct reply_payload : reply_payload_base {
    void value_set() {
        v_set = true;
    }
    void value_set(T&& v) {
        new (&u.v) T(std::move(v));
        value_set();
    }
    union U {
        U() {}
        ~U() {}
        typename std::aligned_storage<sizeof(T), alignof(T)>::type pad;
        T v;
    } u;
    ~reply_payload() {
        if (v_set) {
            u.v.~T();
        }
    }
private:
    bool v_set = false; // set it to true when U::v is valid object
};

template<>
struct reply_payload<void> : reply_payload_base {
    void value_set() {}
};

template<typename Payload, typename... T>
struct rcv_reply_base : reply_payload<Payload> {
    bool done = false;
    promise<T...> p;
    template<typename... V>
    void set_value(V&&... v) {
        this->value_set();
        done = true;
        p.set_value(std::forward<V>(v)...);
    }
    ~rcv_reply_base() {
        if (!done) {
            p.set_exception(closed_error());
        }
    }
};

template<typename Serializer, typename MsgType, typename T>
struct rcv_reply : rcv_reply_base<T, T> {
    inline future<> get_reply(typename protocol<Serializer, MsgType>::client& dst) {
        return unmarshall(dst.serializer(), dst.in(), std::tie(this->u.v)).then([this] {
            this->set_value(std::move(this->u.v));
        });
    }
};

template<typename Serializer, typename MsgType, typename... T>
struct rcv_reply<Serializer, MsgType, future<T...>> : rcv_reply_base<std::tuple<T...>, T...> {
    inline future<> get_reply(typename protocol<Serializer, MsgType>::client& dst) {
        return unmarshall(dst.serializer(), dst.in(), ref_tuple(this->u.v)).then([this] {
            this->set_value(std::move(this->u.v));
        });
    }
};

template<typename Serializer, typename MsgType>
struct rcv_reply<Serializer, MsgType, void> : rcv_reply_base<void, void> {
    inline future<> get_reply(typename protocol<Serializer, MsgType>::client& dst) {
        this->set_value();
        return make_ready_future<>();
{'fraction_non_alphanumeric': 0.15400240746313573, 'fraction_numerical': 0.0018055973517905506, 'mean_word_length': 3.8093342981186686, 'pattern_counts': {'":': 0, '<': 175, '<?xml version=': 0, '>': 191, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7342414', 'n_tokens_mistral': 4224, 'n_tokens_neox': 4133, 'n_words': 1126}